coulomb/the-custodian
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(mcp): resolve repo paths with existence check before trusting hostname match

Browse Source 
Stale host_paths entries (wrong username, old machine) were silently overriding
the correct local_path, causing FileNotFoundError on tools like list_kaizen_agents.

Extracts _resolve_repo_path(repo) helper that tries host_paths[hostname] first
but validates the path exists on disk before trusting it, then falls back to
local_path. Both candidates support ~ expansion. Applied to all 4 call sites:
_kaizen_agents_dir, validate_repo_adr, check_repo_consistency, ingest_sbom_tool.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2026-03-19 21:38:35 +01:00
parent f85c5e4d49
commit 118c5628e9
1 changed files with 52 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
state-hub/mcp_server/server.py
View File

@@ -1032,19 +1032,54 @@ def update_repo_path(repo_slug: str, path: str, host: str | None = None) -> str:
return json.dumps(repo, indent=2)
# ---------------------------------------------------------------------------
# Shared path resolution helper
# ---------------------------------------------------------------------------
def _resolve_repo_path(repo: dict) -> str:
"""Return the best local filesystem path for *repo* on this host.
Resolution order — each candidate is expanded (supports ``~``) and
verified to exist before being accepted:
1. ``host_paths[hostname]`` — host-specific override
2. ``local_path`` — default fallback
Returns the resolved path string, or ``""`` if no valid path is found.
"""
import socket as _socket
hostname = _socket.gethostname()
host_paths = repo.get("host_paths") or {}
candidates = []
if host_paths.get(hostname):
candidates.append(host_paths[hostname])
if repo.get("local_path"):
candidates.append(repo["local_path"])
for raw in candidates:
resolved = str(Path(raw).expanduser())
if Path(resolved).is_dir():
return resolved
return ""
# ---------------------------------------------------------------------------
# Kaizen Agents
# ---------------------------------------------------------------------------
def _kaizen_agents_dir() -> Path:
"""Resolve the kaizen-agentic agents/ directory via host_paths → local_path fallback."""
import socket as _socket
"""Resolve the kaizen-agentic agents/ directory."""
repo = _get("/repos/kaizen-agentic")
hostname = _socket.gethostname()
host_paths = repo.get("host_paths") or {}
base = host_paths.get(hostname) or repo.get("local_path") or ""
base = _resolve_repo_path(repo)
if not base:
raise FileNotFoundError("kaizen-agentic path not found for this host. Register it with update_repo_path().")
import socket as _socket
hostname = _socket.gethostname()
raise FileNotFoundError(
f"kaizen-agentic path not found on host '{hostname}'. "
"Register it with update_repo_path('kaizen-agentic', '/path/to/repo')."
)
agents_dir = Path(base) / "agents"
if not agents_dir.is_dir():
raise FileNotFoundError(f"agents/ directory not found at {agents_dir}")
@@ -1128,8 +1163,8 @@ def validate_repo_adr(repo_slug: str, domain_slug: str | None = None) -> str:
no active state-hub workstreams for the domain lack a backing file (orphan
detection — DB-only records are an ADR-001 violation).
The repo path is resolved from the DB using the current machine's hostname
(host_paths[hostname] → local_path fallback). This tool always runs against
The repo path is resolved from the DB: host_paths[hostname] is tried first
(with existence check), then local_path — both support ~ expansion. This tool always runs against
the server's copy of the repo. Remote agents on a different branch should
sync first, or run validate_repo_adr.py locally with
--api-base http://127.0.0.1:18000.
@@ -1146,22 +1181,15 @@ def validate_repo_adr(repo_slug: str, domain_slug: str | None = None) -> str:
if isinstance(repo, dict) and repo.get("error"):
return f"Repo '{repo_slug}' not found: {repo['error']}"
hostname = _socket.gethostname()
host_paths = repo.get("host_paths") or {}
repo_path = host_paths.get(hostname) or repo.get("local_path") or ""
repo_path = _resolve_repo_path(repo)
if not repo_path:
hostname = _socket.gethostname()
return (
f"⚠ No path registered for repo '{repo_slug}' on this host ({hostname}).\n"
f"⚠ No accessible path found for repo '{repo_slug}' on host '{hostname}'.\n"
f"Register with: update_repo_path('{repo_slug}', '/path/to/repo')\n"
f"Remote agents: run validate_repo_adr.py locally with "
f"--api-base {API_BASE}"
)
if not Path(repo_path).is_dir():
return (
f"⚠ Registered path for '{repo_slug}' on {hostname} does not exist: {repo_path}\n"
f"Update with: update_repo_path('{repo_slug}', '/correct/path')"
)
script = Path(__file__).parent.parent / "scripts" / "validate_repo_adr.py"
cmd = [sys.executable, str(script), repo_path, "--json",
@@ -1238,21 +1266,15 @@ def check_repo_consistency(repo_slug: str, fix: bool = False) -> str:
repo = _get(f"/repos/{repo_slug}")
if isinstance(repo, dict) and repo.get("error"):
return f"Repo '{repo_slug}' not found: {repo['error']}"
hostname = _socket.gethostname()
host_paths = repo.get("host_paths") or {}
repo_path = host_paths.get(hostname) or repo.get("local_path") or ""
repo_path = _resolve_repo_path(repo)
if not repo_path:
hostname = _socket.gethostname()
return (
f"⚠ No path registered for repo '{repo_slug}' on this host ({hostname}).\n"
f"⚠ No accessible path found for repo '{repo_slug}' on host '{hostname}'.\n"
f"Register with: update_repo_path('{repo_slug}', '/path/to/repo')\n"
f"Remote agents: run consistency_check.py locally with "
f"--api-base {API_BASE}"
)
if not Path(repo_path).is_dir():
return (
f"⚠ Registered path for '{repo_slug}' on {hostname} does not exist: {repo_path}\n"
f"Update with: update_repo_path('{repo_slug}', '/correct/path')"
)
script = Path(__file__).parent.parent / "scripts" / "consistency_check.py"
cmd = [sys.executable, str(script), "--repo", repo_slug, "--json",
@@ -1448,20 +1470,13 @@ def ingest_sbom_tool(repo_slug: str, lockfile_path: str | None = None) -> str:
if isinstance(repo, dict) and repo.get("error"):
return f"Repo '{repo_slug}' not found: {repo['error']}"
hostname = _socket.gethostname()
host_paths = repo.get("host_paths") or {}
repo_root = host_paths.get(hostname) or repo.get("local_path") or ""
repo_root = _resolve_repo_path(repo)
if not repo_root:
hostname = _socket.gethostname()
return (
f"⚠ No path registered for repo '{repo_slug}' on this host ({hostname}).\n"
f"⚠ No accessible path found for repo '{repo_slug}' on host '{hostname}'.\n"
f"Register with: update_repo_path('{repo_slug}', '/path/to/repo')"
)
if not Path(repo_root).is_dir():
return (
f"⚠ Registered path for '{repo_slug}' on {hostname} does not exist: {repo_root}\n"
f"Update with: update_repo_path('{repo_slug}', '/correct/path')"
)
script = Path(__file__).parent.parent / "scripts" / "ingest_sbom.py"
cmd = [sys.executable, str(script), "--repo", repo_slug,