coulomb/state-hub
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(CUST-WP-0021): multi-host repo path hardening — all 5 tasks complete

Browse Source 
- T01 (done prior): registered host_paths for bnt-lap001 (14 repos) and
  COULOMBCORE (6 repos) via POST /repos/{slug}/paths/
- T02: validate_repo_adr now accepts repo_slug (not raw path); resolves
  local path via host_paths[hostname] → local_path; clear error for
  unregistered/missing paths
- T03: ingest_sbom_tool lockfile_path is now optional and relative to
  resolved repo root; absolute paths accepted with deprecation warning
- T04: check_repo_consistency pre-flight guard — fetches repo, resolves
  path, returns clear error before spawning subprocess if path missing
- T05: TOOLS.md — updated validate_repo_adr row (slug not path);
  added Multi-Host & Remote Agent Usage section documenting design
  boundary, remote agent workflow, and update_repo_path usage

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2026-03-18 22:53:25 +01:00
parent 75d25e9d3b
commit 4feba3e8d2
2 changed files with 140 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
mcp_server/server.py
View File

@@ -1105,7 +1105,7 @@ def get_kaizen_agent(name: str) -> str:
# ---------------------------------------------------------------------------
@mcp.tool()
def validate_repo_adr(repo_path: str, domain_slug: str | None = None) -> str:
def validate_repo_adr(repo_slug: str, domain_slug: str | None = None) -> str:
"""Check whether a repository is consistent with ADR-001.
Validates that workplan files exist in workplans/ with correct frontmatter,
@@ -1113,12 +1113,41 @@ def validate_repo_adr(repo_path: 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 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.
Args:
repo_path: Absolute path to the repository root.
repo_slug: Registered repo slug (e.g. 'the-custodian', 'ops-bridge').
domain_slug: Domain slug for orphan detection (e.g. 'custodian').
If omitted, inferred from workplan frontmatter.
"""
import socket as _socket
import subprocess
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 ""
if not repo_path:
return (
f"⚠ No path registered for repo '{repo_slug}' on this 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",
"--api-base", API_BASE]
@@ -1138,7 +1167,7 @@ def validate_repo_adr(repo_path: str, domain_slug: str | None = None) -> str:
failures = [f for f in findings if f["level"] == "FAIL"]
warnings = [f for f in findings if f["level"] == "WARN"]
lines = [f"ADR-001 Compliance: {repo_path}", ""]
lines = [f"ADR-001 Compliance: {repo_slug} ({repo_path})", ""]
if failures:
lines.append(f"FAILURES ({len(failures)}):")
@@ -1187,7 +1216,29 @@ def check_repo_consistency(repo_slug: str, fix: bool = False) -> str:
(C-05), create missing DB workstreams (C-06), repo mismatch (C-09),
task status drift (C-10), create unlinked tasks (C-11).
"""
import socket as _socket
import subprocess
# Pre-flight: verify this host has the repo path registered and accessible.
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 ""
if not repo_path:
return (
f"⚠ No path registered for repo '{repo_slug}' on this 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",
"--api-base", API_BASE]
@@ -1359,23 +1410,55 @@ def get_contributions(
@mcp.tool()
def ingest_sbom_tool(repo_slug: str, lockfile_path: str) -> str:
def ingest_sbom_tool(repo_slug: str, lockfile_path: str | None = None) -> str:
"""Ingest a lockfile into the State Hub SBOM store for a repo.
Parses the lockfile and POSTs entries to /sbom/ingest/. Each call creates
a new SBOMSnapshot; previous snapshots are retained as history.
The repo root is resolved from the DB using the current machine's hostname
(host_paths[hostname] → local_path fallback). lockfile_path, when given,
is treated as relative to the repo root. Omit it to auto-detect the lockfile.
Args:
repo_slug: Managed-repo slug (must be registered via register_repo)
lockfile_path: Absolute path to the lockfile (uv.lock, package-lock.json, Cargo.lock, etc.)
lockfile_path: Path to the lockfile, relative to repo root
(e.g. "uv.lock", "frontend/package-lock.json").
Omit to auto-detect from the repo root.
"""
import socket as _socket
import subprocess
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_root = host_paths.get(hostname) or repo.get("local_path") or ""
if not repo_root:
return (
f"⚠ No path registered for repo '{repo_slug}' on this 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"
result = subprocess.run(
[sys.executable, str(script), "--repo", repo_slug,
"--lockfile", lockfile_path, "--api-base", API_BASE],
capture_output=True, text=True,
)
cmd = [sys.executable, str(script), "--repo", repo_slug,
"--repo-path", repo_root, "--api-base", API_BASE]
if lockfile_path:
resolved = Path(repo_root) / lockfile_path
if not resolved.exists():
return f"⚠ Lockfile not found: {resolved}"
cmd += ["--lockfile", str(resolved)]
result = subprocess.run(cmd, capture_output=True, text=True)
output = (result.stdout + result.stderr).strip()
if result.returncode != 0:
return f"ingest_sbom failed (exit {result.returncode}):\n{output}"