diff --git a/Makefile b/Makefile index 7108951..c613638 100644 --- a/Makefile +++ b/Makefile @@ -1,13 +1,15 @@ # Repo Manager — local developer targets (RMGR-ADR-001) UV ?= $(shell command -v uv 2>/dev/null || if [ -x "$$HOME/.local/bin/uv" ]; then printf "%s" "$$HOME/.local/bin/uv"; else printf "%s" "uv"; fi) -.PHONY: help install test lint cli-version +.PHONY: help install test lint cli-version publisher-run publisher-snapshot help: @echo "Repo Manager targets:" @echo " make install # uv sync / editable install" @echo " make test # pytest" @echo " make cli-version # rmgr --version" + @echo " make publisher-run # serve the read-only port.repo publisher" + @echo " make publisher-snapshot REGISTRY=... CURSOR_SECRET=..." @echo " (api/db/migrate land with first runtime slice)" install: @@ -21,3 +23,10 @@ lint: cli-version: $(UV) run rmgr --version + +publisher-run: + $(UV) run rmgr publisher api --host $${HOST:-127.0.0.1} --port $${PORT:-8020} + +publisher-snapshot: + @test -n "$(REGISTRY)" && test -n "$(CURSOR_SECRET)" + $(UV) run rmgr publisher snapshot --registry "$(REGISTRY)" --cursor-secret "$(CURSOR_SECRET)" diff --git a/docs/evidence/RMGR-WP-0013-live-conformance-2026-09-01.md b/docs/evidence/RMGR-WP-0013-live-conformance-2026-09-01.md new file mode 100644 index 0000000..544bad1 --- /dev/null +++ b/docs/evidence/RMGR-WP-0013-live-conformance-2026-09-01.md @@ -0,0 +1,64 @@ +# RMGR-WP-0013 live publisher conformance — 2026-09-01 + +## Scope + +This evidence covers the Repo Manager classification publisher and hub-core's +frozen `helixforge.repository-classification-projection` 1.0.0 consumer. It is +a live source/conformance proof from the workstation checkout fleet, not a +production deployment claim. + +## Registrar bootstrap + +- State Hub source: `primary` / `railiance01` +- Discovery root: `/home/worsch` (operator-local, not published) +- Registered classified repositories: **123** +- Locally classified but unregistered checkouts: **2** + (`snuggles-inventor`, `wise-validator`), reported as warnings and omitted +- Stable IDs: existing State Hub registrar UUIDs; none invented +- Bulk `/repos/` was not used because it timed out after 180 seconds +- Successful bounded import: four workers, 15-second request timeout, two + transient retries + +The generated private registry was written only to `/tmp` and was not committed. +It contained host paths for local observation; those paths did not enter the +published projection. + +## End-to-end transfer + +The publisher served the live registry with a page size of 25. Hub-core fetched +five HMAC-bound pages through its new HTTP `port.repo` client and accepted the +complete transfer atomically. + +```json +{ + "status": "accepted", + "snapshot_id": "c22faf46794ee9288102d6f0f98b04372ebfaa2c41017edf11c6101f157ec654", + "repository_count": 123, + "projection_status": "current", + "content_hash": "a3a1d584c8245b6c339bd672551d0960bf2b0b133e259bec7e884fe302cfe14c", + "source_revision": "b28a45e7c9592f6aecdc4537b463333545ba6f1c68746d6a5e904cff4d654008" +} +``` + +Repo Manager read classification from each repository's authoritative file and +published only contract fields. Hub-core derived its own navigation generation; +no database or private persistence model was shared. + +## Verification + +- Repo Manager: **156 passed**; Ruff clean +- Hub-core: **107 passed** +- Focused publisher tests cover duplicate identities, deterministic UUID order, + Git/source provenance, multi-page transfer, cursor tampering, invalid source + diagnostics, bearer authentication, readiness, and primary registrar import +- Hub-core focused tests cover cursor/token HTTP transport plus all frozen + repository-navigation contract and ingestion behaviors + +## Remaining production gate + +The code and cross-repository transport are ready. Production still needs an +explicit placement/network decision because hub-core runs in the public Core +Hub cluster while the authoritative checkout registry is host-local to the +Repo Manager worker. No broad host-path mount or implicit State Hub dependency +was introduced to hide that boundary. `RMGR-WP-0013-T05` remains waiting for +that deployment handoff; `HUB-WP-0006-T06` must not switch traffic before it. diff --git a/docs/repository-classification-publisher.md b/docs/repository-classification-publisher.md new file mode 100644 index 0000000..4761e45 --- /dev/null +++ b/docs/repository-classification-publisher.md @@ -0,0 +1,113 @@ +# Repository classification publisher + +Repo Manager publishes the frozen +`helixforge.repository-classification-projection` 1.0.0 envelope at: + +```text +GET /ports/repositories/classifications?cursor= +``` + +The route is read-only. It contains registrar UUID, slug, lifecycle, +classification, Git revision, source fingerprint, and observation time. Local +paths, remotes, work records, source bodies, and credentials never enter the +envelope. + +## Private operator registry + +Publication joins repository-owned `.repo-classification.yaml` files to a +private operator registry. The registry is local runtime state because checkout +paths are host-specific; do not commit it. Its schema is: + +```yaml +schema: repo-manager.repository-registry.v1 +repositories: + - repository_id: a9d105c6-10fa-4cf9-8bc1-e248476698d3 + slug: repo-manager + lifecycle: active + path: /home/operator/repo-manager +``` + +Bootstrap current stable identities from the retiring primary State Hub: + +```bash +rmgr publisher import-registry \ + --root /home/operator \ + --api-base http://127.0.0.1:8000 \ + --output ~/.repo-manager/repository-registry.yaml +``` + +The import verifies `instance_role=primary`, discovers only classified Git +working copies below the selected root, and uses bounded direct identity reads. +It does not use State Hub's expensive bulk repository projection and does not +trust State Hub host paths. Transient reads retry; locally classified but +unregistered checkouts are retained as warnings and omitted rather than being +assigned invented identities. Any other failure prevents replacement of the +registry file. + +After repository-registry authority moves fully into Repo Manager, normal +governed registration maintains this same private schema and the bootstrap +command can be retired. The publisher contract does not change. + +## Run + +Set runtime configuration through the environment: + +```text +REPO_MANAGER_REGISTRY_PATH=~/.repo-manager/repository-registry.yaml +REPO_MANAGER_CURSOR_SECRET= +REPO_MANAGER_API_TOKEN= +REPO_MANAGER_PROJECTION_PAGE_SIZE=100 +REPO_MANAGER_PROJECTION_MAX_SNAPSHOTS=4 +``` + +Secret values belong in the platform credential path and must not be committed +or logged. Start the API with: + +```bash +rmgr publisher api --host 0.0.0.0 --port 8020 +``` + +`GET /healthz` is process liveness. `GET /readyz` verifies that configuration +and the registry are usable. When `REPO_MANAGER_API_TOKEN` is set, the +projection route requires the matching bearer token. + +## Snapshot and failure semantics + +A first-page request observes the complete registry into an immutable in-memory +snapshot. Repositories are ordered by registrar UUID. Subsequent page cursors +carry snapshot ID, offset, and page size protected by HMAC-SHA256. Only a small +bounded set of in-flight snapshots is retained; an invalid, tampered, or +expired cursor returns `409`. + +Missing checkouts, invalid classifications, or unavailable Git revisions are +emitted as bounded error diagnostics. Hub-core rejects any such transfer and +keeps the last accepted generation, so a partial observation never becomes the +active navigation projection. + +## Hub-core client + +Configure the hub-core runtime with: + +```text +HUB_CORE_REPO_MANAGER_BASE_URL=http://repo-manager:8020 +HUB_CORE_REPO_MANAGER_API_TOKEN= +HUB_CORE_REPO_MANAGER_TIMEOUT_SECONDS=10 +HUB_CORE_REPO_PROJECTION_REFRESH_SECONDS=300 +``` + +Hub-core refreshes once at startup and then on the configured interval. A +failed refresh marks the dependency stale while preserving the last accepted +generation. Setting the refresh interval to `0` disables the background loop. + +## Verification + +```bash +make test +make publisher-snapshot \ + REGISTRY=~/.repo-manager/repository-registry.yaml \ + CURSOR_SECRET='' +``` + +RMGR-WP-0013's live conformance used five pages of 25 rows. Hub-core accepted +all 123 registered classifications atomically and produced one current +generation; no State Hub classification table was read by the publisher. diff --git a/src/repo_manager/cli.py b/src/repo_manager/cli.py index 1ebcb9d..a87bfc1 100644 --- a/src/repo_manager/cli.py +++ b/src/repo_manager/cli.py @@ -536,6 +536,39 @@ def main(argv: list[str] | None = None) -> int: p_workload_resolve.add_argument("--name", required=True) p_workload_resolve.add_argument("--deployable", default=None) + p_publisher = sub.add_parser( + "publisher", + help="Publish repository classifications through the read-only port.repo contract", + ) + publisher_sub = p_publisher.add_subparsers(dest="publisher_command") + p_publisher_snapshot = publisher_sub.add_parser( + "snapshot", help="Render the first page of a classification projection snapshot" + ) + p_publisher_snapshot.add_argument("--registry", required=True) + p_publisher_snapshot.add_argument( + "--cursor-secret", default=os.environ.get("REPO_MANAGER_CURSOR_SECRET") + ) + p_publisher_snapshot.add_argument("--page-size", type=int, default=100) + p_publisher_import = publisher_sub.add_parser( + "import-registry", + help="Bootstrap the private local registry from primary State Hub identities", + ) + p_publisher_import.add_argument("--root", required=True, help="Local fleet checkout root") + p_publisher_import.add_argument( + "--api-base", + default=os.environ.get("STATE_HUB_API_BASE", "http://127.0.0.1:8000"), + ) + p_publisher_import.add_argument("--output", required=True) + p_publisher_import.add_argument("--timeout", type=float, default=10.0) + p_publisher_import.add_argument("--workers", type=int, default=4) + p_publisher_import.add_argument("--retries", type=int, default=2) + p_publisher_import.add_argument("--force", action="store_true") + p_publisher_api = publisher_sub.add_parser( + "api", help="Run the read-only classification publisher API" + ) + p_publisher_api.add_argument("--host", default="127.0.0.1") + p_publisher_api.add_argument("--port", type=int, default=8020) + p_owner_interface = sub.add_parser( "owner-interface", help="Inspect validated owner-consumable task interfaces", @@ -1224,6 +1257,83 @@ def main(argv: list[str] | None = None) -> int: print(json.dumps(result, indent=2)) return 0 if result.get("ok") else 1 + if args.command == "publisher": + if not args.publisher_command: + p_publisher.print_help() + return 2 + if args.publisher_command == "api": + import uvicorn + + uvicorn.run("repo_manager.runtime:app", host=args.host, port=args.port) + return 0 + if args.publisher_command == "import-registry": + import yaml + + from repo_manager.repository_publisher import import_state_hub_registry + + output = Path(args.output).expanduser() + if output.exists() and not args.force: + print( + json.dumps( + {"ok": False, "error": f"output exists: {output}; use --force to replace"}, + indent=2, + ) + ) + return 1 + try: + result = import_state_hub_registry( + Path(args.root), + api_base=args.api_base, + timeout_seconds=args.timeout, + workers=args.workers, + retries=args.retries, + ) + except (OSError, ValueError) as exc: + print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) + return 1 + if not result["ok"]: + print(json.dumps(result, indent=2)) + return 1 + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(yaml.safe_dump(result, sort_keys=False), encoding="utf-8") + print( + json.dumps( + { + "ok": True, + "output": str(output), + "repository_count": len(result["repositories"]), + "source": result["source"], + }, + indent=2, + ) + ) + return 0 + if not args.cursor_secret: + print( + json.dumps( + { + "ok": False, + "error": "--cursor-secret or REPO_MANAGER_CURSOR_SECRET is required", + }, + indent=2, + ) + ) + return 2 + from repo_manager.repository_publisher import ClassificationPublisher + + try: + publisher = ClassificationPublisher( + Path(args.registry), + cursor_secret=args.cursor_secret, + page_size=args.page_size, + ) + result = publisher.fetch_page() + except ValueError as exc: + print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) + return 1 + print(json.dumps(result, indent=2)) + return 0 + if args.command == "owner-interface": if not args.owner_interface_command: p_owner_interface.print_help() diff --git a/src/repo_manager/repository_publisher.py b/src/repo_manager/repository_publisher.py new file mode 100644 index 0000000..9e3add6 --- /dev/null +++ b/src/repo_manager/repository_publisher.py @@ -0,0 +1,521 @@ +"""Publish repository classifications through the frozen ``port.repo`` contract.""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import re +import subprocess +import threading +from collections import OrderedDict +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any +from urllib.parse import urlparse +from uuid import UUID + +import httpx +import yaml + +from repo_manager import __version__ +from repo_manager.classification import ClassificationError +from repo_manager.observe import load_classification +from repo_manager.time import format_utc, utc_now + +CONTRACT_ID = "helixforge.repository-classification-projection" +CONTRACT_VERSION = "1.0.0" +CLASSIFICATION_CONTRACT_ID = "RMGR-CONTRACT-CLASSIFICATION-0001" +CLASSIFICATION_CONTRACT_VERSION = "1.0" +REGISTRY_SCHEMA = "repo-manager.repository-registry.v1" +LIFECYCLES = frozenset({"active", "archived", "retired"}) +_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") + + +class RegistryError(ValueError): + """The operator-owned repository registry cannot be used safely.""" + + +class ProjectionCursorError(ValueError): + """A projection page cursor is malformed, stale, or has been tampered with.""" + + +@dataclass(frozen=True, slots=True) +class RepositoryRegistration: + repository_id: str + slug: str + lifecycle: str + path: Path + + +@dataclass(frozen=True, slots=True) +class ClassificationSnapshot: + snapshot_id: str + generated_at: str + source_revision: str + repositories: tuple[dict[str, Any], ...] + diagnostics: tuple[dict[str, Any], ...] + + +def load_repository_registry(path: Path) -> tuple[RepositoryRegistration, ...]: + """Load the private operator registry used to join UUIDs to local checkouts.""" + try: + raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + except (OSError, yaml.YAMLError) as exc: + raise RegistryError(f"cannot read repository registry: {exc}") from exc + if not isinstance(raw, dict) or raw.get("schema") != REGISTRY_SCHEMA: + raise RegistryError(f"repository registry must declare schema {REGISTRY_SCHEMA!r}") + rows = raw.get("repositories") + if not isinstance(rows, list): + raise RegistryError("repository registry `repositories` must be a list") + + registrations: list[RepositoryRegistration] = [] + ids: set[str] = set() + slugs: set[str] = set() + for number, row in enumerate(rows, start=1): + if not isinstance(row, dict): + raise RegistryError(f"repository registry row {number} must be an object") + try: + repository_id = str(UUID(str(row["repository_id"]))) + slug = str(row["slug"]) + lifecycle = str(row["lifecycle"]) + repo_path = Path(str(row["path"])) + except (KeyError, TypeError, ValueError) as exc: + raise RegistryError(f"invalid repository registry row {number}: {exc}") from exc + if repository_id in ids: + raise RegistryError(f"duplicate repository_id {repository_id}") + if slug in slugs: + raise RegistryError(f"duplicate repository slug {slug!r}") + if not _SLUG_RE.fullmatch(slug) or len(slug) > 120: + raise RegistryError(f"invalid repository slug {slug!r}") + if lifecycle not in LIFECYCLES: + raise RegistryError( + f"repository {slug!r} lifecycle must be one of {', '.join(sorted(LIFECYCLES))}" + ) + if not repo_path.is_absolute(): + raise RegistryError(f"repository {slug!r} path must be absolute") + ids.add(repository_id) + slugs.add(slug) + registrations.append( + RepositoryRegistration( + repository_id=repository_id, + slug=slug, + lifecycle=lifecycle, + path=repo_path, + ) + ) + return tuple(sorted(registrations, key=lambda item: item.repository_id)) + + +def import_state_hub_registry( + fleet_root: Path, + *, + api_base: str, + timeout_seconds: float = 10.0, + workers: int = 4, + retries: int = 2, + transport: httpx.BaseTransport | None = None, +) -> dict[str, Any]: + """Bootstrap Repo Manager identities from the retiring authoritative registrar. + + The bulk State Hub repository route is intentionally not used: direct + identity reads keep the migration bounded and let independent repositories + complete concurrently. Host paths come only from the operator-selected + local fleet root; remote State Hub path values are never trusted. + """ + parsed = urlparse(api_base) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("State Hub API base must be an absolute HTTP(S) origin") + if timeout_seconds <= 0 or workers < 1 or retries < 0: + raise ValueError("timeout_seconds/workers must be positive and retries non-negative") + paths = _discover_classified_repositories(fleet_root) + if not paths: + raise ValueError("fleet root contains no classified Git repositories") + with httpx.Client( + base_url=api_base.rstrip("/"), + timeout=httpx.Timeout(timeout_seconds), + follow_redirects=True, + transport=transport, + ) as client: + try: + health = client.get("/state/health") + health.raise_for_status() + identity = health.json() + except (httpx.HTTPError, ValueError) as exc: + raise ValueError(f"State Hub registry source is unavailable: {exc}") from exc + if not isinstance(identity, dict): + raise RegistryError("State Hub health response is not a JSON object") + if identity.get("instance_role") != "primary": + raise ValueError("State Hub registry import requires an instance_role=primary source") + + def fetch(path: Path) -> tuple[Path, dict[str, Any]]: + for attempt in range(retries + 1): + try: + response = client.get(f"/repos/{path.name}") + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise RegistryError("repository response is not a JSON object") + return path, payload + except (httpx.TimeoutException, httpx.NetworkError): + if attempt == retries: + raise + raise AssertionError("unreachable retry loop") + + rows: list[dict[str, Any]] = [] + diagnostics: list[dict[str, str]] = [] + with ThreadPoolExecutor(max_workers=workers) as executor: + pending = {executor.submit(fetch, path): path for path in paths} + for future in as_completed(pending): + path = pending[future] + try: + local_path, payload = future.result() + repository_id = str(UUID(str(payload["id"]))) + slug = str(payload["slug"]) + lifecycle = str(payload.get("status") or "active") + if lifecycle not in LIFECYCLES: + raise ValueError(f"unsupported lifecycle {lifecycle!r}") + if not _SLUG_RE.fullmatch(slug): + raise ValueError(f"invalid canonical slug {slug!r}") + except httpx.HTTPStatusError as exc: + missing = exc.response.status_code == 404 + diagnostics.append( + { + "severity": "warning" if missing else "error", + "code": ( + "registry_import.unregistered_checkout" + if missing + else "registry_import.repository_failed" + ), + "repository": path.name, + "message": str(exc)[:500], + } + ) + except (httpx.HTTPError, KeyError, TypeError, ValueError) as exc: + diagnostics.append( + { + "severity": "error", + "code": "registry_import.repository_failed", + "repository": path.name, + "message": str(exc)[:500], + } + ) + else: + rows.append( + { + "repository_id": repository_id, + "slug": slug, + "lifecycle": lifecycle, + "path": str(local_path.resolve()), + } + ) + + rows.sort(key=lambda row: row["repository_id"]) + diagnostics.sort(key=lambda row: row["repository"]) + ids = [row["repository_id"] for row in rows] + slugs = [row["slug"] for row in rows] + if len(ids) != len(set(ids)) or len(slugs) != len(set(slugs)): + diagnostics.append( + { + "severity": "error", + "code": "registry_import.identity_collision", + "repository": "fleet", + "message": "State Hub returned duplicate repository UUIDs or canonical slugs", + } + ) + return { + "ok": not any(item["severity"] == "error" for item in diagnostics), + "schema": REGISTRY_SCHEMA, + "source": { + "system": "state-hub", + "instance_role": identity.get("instance_role"), + "instance_label": identity.get("instance_label"), + }, + "generated_at": format_utc(utc_now()), + "repositories": rows, + "diagnostics": diagnostics, + } + + +def build_classification_snapshot( + registrations: tuple[RepositoryRegistration, ...], + *, + observed_at: datetime | None = None, +) -> ClassificationSnapshot: + """Observe one atomic, full fleet generation without publishing host-local paths.""" + instant = observed_at or utc_now() + generated_at = format_utc(instant) + repositories: list[dict[str, Any]] = [] + diagnostics: list[dict[str, Any]] = [] + + for registration in sorted(registrations, key=lambda item: item.repository_id): + try: + change = _observe_registration(registration, observed_at=generated_at) + except (ClassificationError, OSError, ValueError) as exc: + diagnostics.append( + { + "severity": "error", + "code": "repo_projection.source_invalid", + "message": f"{registration.slug}: {str(exc)[:900]}", + "repository_id": registration.repository_id, + } + ) + else: + repositories.append(change) + + source_revision = _sha256( + { + "registry_schema": REGISTRY_SCHEMA, + "repositories": repositories, + "diagnostics": diagnostics, + } + ) + snapshot_id = _sha256( + { + "contract_id": CONTRACT_ID, + "contract_version": CONTRACT_VERSION, + "generated_at": generated_at, + "source_revision": source_revision, + } + ) + return ClassificationSnapshot( + snapshot_id=snapshot_id, + generated_at=generated_at, + source_revision=source_revision, + repositories=tuple(repositories), + diagnostics=tuple(diagnostics), + ) + + +def projection_page( + snapshot: ClassificationSnapshot, + *, + offset: int, + page_size: int, + cursor_secret: bytes, + page_cursor: str | None, +) -> dict[str, Any]: + if not 1 <= page_size <= 500: + raise ValueError("page_size must be between 1 and 500") + if offset < 0 or offset > len(snapshot.repositories): + raise ProjectionCursorError("projection cursor offset exceeds the snapshot") + page_rows = snapshot.repositories[offset : offset + page_size] + next_offset = offset + len(page_rows) + final_page = next_offset >= len(snapshot.repositories) + next_cursor = None + if not final_page: + next_cursor = encode_projection_cursor( + snapshot_id=snapshot.snapshot_id, + offset=next_offset, + page_size=page_size, + secret=cursor_secret, + ) + return { + "contract_id": CONTRACT_ID, + "contract_version": CONTRACT_VERSION, + "source": { + "system": "repo-manager", + "classification_contract_id": CLASSIFICATION_CONTRACT_ID, + "classification_contract_version": CLASSIFICATION_CONTRACT_VERSION, + "producer_version": __version__, + }, + "snapshot": { + "snapshot_id": snapshot.snapshot_id, + "mode": "full", + "generated_at": snapshot.generated_at, + "source_revision": snapshot.source_revision, + "page_cursor": page_cursor, + "next_cursor": next_cursor, + "final_page": final_page, + "total_repository_count": len(snapshot.repositories), + }, + "repositories": list(page_rows), + "diagnostics": list(snapshot.diagnostics) if offset == 0 else [], + } + + +class ClassificationPublisher: + """Bounded in-memory holder for in-flight, immutable snapshot transfers.""" + + def __init__( + self, + registry_path: Path, + *, + cursor_secret: str | bytes, + page_size: int = 100, + max_snapshots: int = 4, + now: Callable[[], datetime] = utc_now, + ) -> None: + secret = cursor_secret.encode() if isinstance(cursor_secret, str) else cursor_secret + if len(secret) < 32: + raise ValueError("cursor_secret must contain at least 32 bytes") + if not 1 <= page_size <= 500: + raise ValueError("page_size must be between 1 and 500") + if max_snapshots < 1: + raise ValueError("max_snapshots must be positive") + self.registry_path = registry_path + self.cursor_secret = secret + self.page_size = page_size + self.max_snapshots = max_snapshots + self.now = now + self._snapshots: OrderedDict[str, ClassificationSnapshot] = OrderedDict() + self._lock = threading.Lock() + + def fetch_page(self, cursor: str | None = None) -> dict[str, Any]: + with self._lock: + return self._fetch_page(cursor) + + def _fetch_page(self, cursor: str | None = None) -> dict[str, Any]: + if cursor is None: + snapshot = build_classification_snapshot( + load_repository_registry(self.registry_path), observed_at=self.now() + ) + self._snapshots[snapshot.snapshot_id] = snapshot + self._snapshots.move_to_end(snapshot.snapshot_id) + while len(self._snapshots) > self.max_snapshots: + self._snapshots.popitem(last=False) + return projection_page( + snapshot, + offset=0, + page_size=self.page_size, + cursor_secret=self.cursor_secret, + page_cursor=None, + ) + + cursor_data = decode_projection_cursor(cursor, secret=self.cursor_secret) + snapshot = self._snapshots.get(cursor_data["snapshot_id"]) + if snapshot is None: + raise ProjectionCursorError("projection cursor snapshot has expired") + self._snapshots.move_to_end(snapshot.snapshot_id) + return projection_page( + snapshot, + offset=cursor_data["offset"], + page_size=cursor_data["page_size"], + cursor_secret=self.cursor_secret, + page_cursor=cursor, + ) + + +def encode_projection_cursor( + *, snapshot_id: str, offset: int, page_size: int, secret: bytes +) -> str: + payload = json.dumps( + {"snapshot_id": snapshot_id, "offset": offset, "page_size": page_size}, + separators=(",", ":"), + sort_keys=True, + ).encode() + signature = hmac.new(secret, payload, hashlib.sha256).digest() + return f"{_b64encode(payload)}.{_b64encode(signature)}" + + +def decode_projection_cursor(cursor: str, *, secret: bytes) -> dict[str, Any]: + try: + encoded_payload, encoded_signature = cursor.split(".", 1) + payload = _b64decode(encoded_payload) + signature = _b64decode(encoded_signature) + expected = hmac.new(secret, payload, hashlib.sha256).digest() + if not hmac.compare_digest(signature, expected): + raise ProjectionCursorError("projection cursor signature is invalid") + data = json.loads(payload) + snapshot_id = str(data["snapshot_id"]) + offset = int(data["offset"]) + page_size = int(data["page_size"]) + except ProjectionCursorError: + raise + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise ProjectionCursorError("projection cursor is malformed") from exc + if not re.fullmatch(r"[a-f0-9]{64}", snapshot_id): + raise ProjectionCursorError("projection cursor snapshot id is invalid") + if offset < 1 or not 1 <= page_size <= 500: + raise ProjectionCursorError("projection cursor bounds are invalid") + return {"snapshot_id": snapshot_id, "offset": offset, "page_size": page_size} + + +def _observe_registration( + registration: RepositoryRegistration, *, observed_at: str +) -> dict[str, Any]: + repo_root = registration.path + if not repo_root.is_dir() or not (repo_root / ".git").exists(): + raise ValueError("registered checkout is missing or is not a Git working copy") + classification_path = repo_root / ".repo-classification.yaml" + if not classification_path.is_file(): + raise ValueError("authoritative .repo-classification.yaml is missing") + classification_status = subprocess.run( + ["git", "status", "--porcelain", "--", ".repo-classification.yaml"], + cwd=repo_root, + capture_output=True, + text=True, + check=False, + ) + if classification_status.returncode != 0 or classification_status.stdout.strip(): + raise ValueError("authoritative classification has uncommitted changes") + classification = load_classification(repo_root) + if classification is None: + raise ValueError("authoritative classification is empty") + head_sha = _git_head(repo_root) + if head_sha is None: + raise ValueError("repository HEAD is unavailable") + normalized = { + "category": classification["category"], + "domain": classification["domain"], + "secondary_domains": sorted(classification.get("secondary_domains") or []), + "capability_tags": sorted(classification.get("capability_tags") or []), + "business_stake": sorted(classification.get("business_stake") or []), + "business_mechanics": sorted(classification.get("business_mechanics") or []), + } + return { + "operation": "upsert", + "repository_id": registration.repository_id, + "slug": registration.slug, + "lifecycle": registration.lifecycle, + "classification": normalized, + "revision": { + "head_sha": head_sha, + "source_fingerprint": hashlib.sha256(classification_path.read_bytes()).hexdigest(), + "observed_at": observed_at, + }, + } + + +def _git_head(repo_root: Path) -> str | None: + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=repo_root, + capture_output=True, + text=True, + check=False, + ) + value = result.stdout.strip().lower() + return value if result.returncode == 0 and re.fullmatch(r"[a-f0-9]{40,64}", value) else None + + +def _discover_classified_repositories(root: Path) -> tuple[Path, ...]: + resolved = root.expanduser().resolve() + if (resolved / ".git").exists() and (resolved / ".repo-classification.yaml").is_file(): + return (resolved,) + if not resolved.is_dir(): + raise ValueError(f"fleet root does not exist: {resolved}") + return tuple( + child + for child in sorted(resolved.iterdir()) + if child.is_dir() + and (child / ".git").exists() + and (child / ".repo-classification.yaml").is_file() + ) + + +def _sha256(value: Any) -> str: + encoded = json.dumps(value, separators=(",", ":"), sort_keys=True).encode() + return hashlib.sha256(encoded).hexdigest() + + +def _b64encode(value: bytes) -> str: + return base64.urlsafe_b64encode(value).rstrip(b"=").decode() + + +def _b64decode(value: str) -> bytes: + return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) diff --git a/src/repo_manager/runtime.py b/src/repo_manager/runtime.py new file mode 100644 index 0000000..b79a305 --- /dev/null +++ b/src/repo_manager/runtime.py @@ -0,0 +1,115 @@ +"""Read-only HTTP runtime for Repo Manager projection publication.""" + +from __future__ import annotations + +import hmac +import os +from dataclasses import dataclass +from pathlib import Path + +from fastapi import FastAPI, Header, HTTPException, Query, Response, status + +from repo_manager import __version__ +from repo_manager.repository_publisher import ( + ClassificationPublisher, + ProjectionCursorError, + RegistryError, + load_repository_registry, +) + + +@dataclass(frozen=True, slots=True) +class PublisherSettings: + registry_path: Path | None = None + cursor_secret: str | None = None + api_token: str | None = None + page_size: int = 100 + max_snapshots: int = 4 + + @classmethod + def from_env(cls) -> PublisherSettings: + registry = os.getenv("REPO_MANAGER_REGISTRY_PATH") + return cls( + registry_path=Path(registry) if registry else None, + cursor_secret=os.getenv("REPO_MANAGER_CURSOR_SECRET"), + api_token=os.getenv("REPO_MANAGER_API_TOKEN"), + page_size=int(os.getenv("REPO_MANAGER_PROJECTION_PAGE_SIZE", "100")), + max_snapshots=int(os.getenv("REPO_MANAGER_PROJECTION_MAX_SNAPSHOTS", "4")), + ) + + +def create_app(*, settings: PublisherSettings | None = None) -> FastAPI: + resolved = settings or PublisherSettings.from_env() + publisher: ClassificationPublisher | None = None + configuration_error: str | None = None + try: + if resolved.registry_path is None: + raise ValueError("REPO_MANAGER_REGISTRY_PATH is required") + if resolved.cursor_secret is None: + raise ValueError("REPO_MANAGER_CURSOR_SECRET is required") + publisher = ClassificationPublisher( + resolved.registry_path, + cursor_secret=resolved.cursor_secret, + page_size=resolved.page_size, + max_snapshots=resolved.max_snapshots, + ) + except ValueError as exc: + configuration_error = str(exc) + + app = FastAPI( + title="Repo Manager projection publisher", + version=__version__, + description="Read-only port.repo repository projection publication.", + ) + app.state.publisher = publisher + app.state.configuration_error = configuration_error + + @app.get("/healthz", tags=["system"]) + async def healthz() -> dict[str, str]: + return {"status": "ok", "service": "repo-manager", "version": __version__} + + @app.get("/readyz", tags=["system"]) + async def readyz(response: Response) -> dict[str, str | int]: + error = app.state.configuration_error + if error is None: + try: + registrations = load_repository_registry(resolved.registry_path) # type: ignore[arg-type] + except RegistryError as exc: + error = str(exc) + if error is not None: + response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE + return {"status": "degraded", "error": error} + return {"status": "ok", "repository_count": len(registrations)} + + @app.get( + "/ports/repositories/classifications", + tags=["port.repo"], + openapi_extra={"x-port-id": "port.repo", "x-direction": "out"}, + ) + def repository_classifications( + cursor: str | None = Query(default=None, max_length=1000), + authorization: str | None = Header(default=None), + ) -> dict: + if resolved.api_token is not None and not hmac.compare_digest( + authorization or "", f"Bearer {resolved.api_token}" + ): + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="unauthorized") + active = app.state.publisher + if active is None: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=app.state.configuration_error, + ) + try: + return active.fetch_page(cursor) + except ProjectionCursorError as exc: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc + except RegistryError as exc: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc) + ) from exc + + return app + + +app = create_app() diff --git a/tests/test_repository_publisher.py b/tests/test_repository_publisher.py new file mode 100644 index 0000000..ed304a5 --- /dev/null +++ b/tests/test_repository_publisher.py @@ -0,0 +1,275 @@ +from __future__ import annotations + +import subprocess +from datetime import UTC, datetime +from pathlib import Path + +import httpx +import yaml +from fastapi.testclient import TestClient + +from repo_manager.repository_publisher import ( + ClassificationPublisher, + ProjectionCursorError, + RepositoryRegistration, + build_classification_snapshot, + import_state_hub_registry, + load_repository_registry, +) +from repo_manager.runtime import PublisherSettings, create_app + +NOW = datetime(2026, 9, 1, 8, 30, tzinfo=UTC) +SECRET = "projection-test-secret-is-at-least-32-bytes" + + +def _repo(root: Path, name: str, *, domain: str = "infotech") -> Path: + repo = root / name + repo.mkdir() + subprocess.run(["git", "init", "-q", "-b", "main"], cwd=repo, check=True) + subprocess.run(["git", "config", "user.name", "Test"], cwd=repo, check=True) + subprocess.run(["git", "config", "user.email", "test@example.invalid"], cwd=repo, check=True) + (repo / ".repo-classification.yaml").write_text( + yaml.safe_dump( + { + "repo_classification": { + "category": "tooling", + "domain": domain, + "secondary_domains": ["agents"], + "capability_tags": ["repository-governance"], + "business_stake": ["technology"], + "business_mechanics": ["control"], + } + }, + sort_keys=False, + ), + encoding="utf-8", + ) + subprocess.run(["git", "add", ".repo-classification.yaml"], cwd=repo, check=True) + subprocess.run(["git", "commit", "-qm", "classify"], cwd=repo, check=True) + return repo + + +def _registry(path: Path, registrations: list[dict]) -> Path: + path.write_text( + yaml.safe_dump( + {"schema": "repo-manager.repository-registry.v1", "repositories": registrations}, + sort_keys=False, + ), + encoding="utf-8", + ) + return path + + +def test_registry_rejects_duplicate_stable_identity(tmp_path: Path) -> None: + repo = _repo(tmp_path, "one") + registry = _registry( + tmp_path / "registry.yaml", + [ + { + "repository_id": "11111111-1111-4111-8111-111111111111", + "slug": "one", + "lifecycle": "active", + "path": str(repo), + }, + { + "repository_id": "11111111-1111-4111-8111-111111111111", + "slug": "two", + "lifecycle": "active", + "path": str(repo), + }, + ], + ) + try: + load_repository_registry(registry) + except ValueError as exc: + assert "duplicate repository_id" in str(exc) + else: + raise AssertionError("duplicate stable identity was accepted") + + +def test_snapshot_is_uuid_ordered_and_carries_source_provenance(tmp_path: Path) -> None: + first = _repo(tmp_path, "first") + second = _repo(tmp_path, "second") + registrations = ( + RepositoryRegistration( + "22222222-2222-4222-8222-222222222222", "second", "active", second + ), + RepositoryRegistration( + "11111111-1111-4111-8111-111111111111", "first", "archived", first + ), + ) + snapshot = build_classification_snapshot(registrations, observed_at=NOW) + + assert [row["repository_id"] for row in snapshot.repositories] == [ + "11111111-1111-4111-8111-111111111111", + "22222222-2222-4222-8222-222222222222", + ] + assert snapshot.repositories[0]["lifecycle"] == "archived" + assert snapshot.repositories[0]["revision"]["observed_at"] == "2026-09-01T08:30:00Z" + assert len(snapshot.repositories[0]["revision"]["head_sha"]) == 40 + assert len(snapshot.repositories[0]["revision"]["source_fingerprint"]) == 64 + assert snapshot.diagnostics == () + + +def test_paging_is_bound_to_an_immutable_snapshot_and_signed(tmp_path: Path) -> None: + first = _repo(tmp_path, "first") + second = _repo(tmp_path, "second") + registry = _registry( + tmp_path / "registry.yaml", + [ + { + "repository_id": "22222222-2222-4222-8222-222222222222", + "slug": "second", + "lifecycle": "active", + "path": str(second), + }, + { + "repository_id": "11111111-1111-4111-8111-111111111111", + "slug": "first", + "lifecycle": "active", + "path": str(first), + }, + ], + ) + publisher = ClassificationPublisher( + registry, cursor_secret=SECRET, page_size=1, now=lambda: NOW + ) + + page_one = publisher.fetch_page() + cursor = page_one["snapshot"]["next_cursor"] + page_two = publisher.fetch_page(cursor) + + assert page_one["contract_id"] == "helixforge.repository-classification-projection" + assert page_one["contract_version"] == "1.0.0" + assert page_one["snapshot"]["page_cursor"] is None + assert page_one["snapshot"]["final_page"] is False + assert page_two["snapshot"]["page_cursor"] == cursor + assert page_two["snapshot"]["final_page"] is True + assert page_two["snapshot"]["next_cursor"] is None + assert page_two["snapshot"]["snapshot_id"] == page_one["snapshot"]["snapshot_id"] + assert page_one["repositories"][0]["repository_id"] < page_two["repositories"][0][ + "repository_id" + ] + + tampered = ("A" if cursor[0] != "A" else "B") + cursor[1:] + try: + publisher.fetch_page(tampered) + except ProjectionCursorError as exc: + assert "signature" in str(exc) or "malformed" in str(exc) + else: + raise AssertionError("tampered projection cursor was accepted") + + +def test_invalid_source_is_an_error_diagnostic_not_a_partial_upsert(tmp_path: Path) -> None: + repo = _repo(tmp_path, "broken") + (repo / ".repo-classification.yaml").write_text( + "repo_classification:\n category: unknown\n domain: infotech\n", + encoding="utf-8", + ) + snapshot = build_classification_snapshot( + ( + RepositoryRegistration( + "11111111-1111-4111-8111-111111111111", "broken", "active", repo + ), + ), + observed_at=NOW, + ) + assert snapshot.repositories == () + assert snapshot.diagnostics[0]["severity"] == "error" + assert snapshot.diagnostics[0]["code"] == "repo_projection.source_invalid" + + +def test_uncommitted_classification_cannot_be_published_as_head(tmp_path: Path) -> None: + repo = _repo(tmp_path, "dirty") + (repo / ".repo-classification.yaml").write_text( + "repo_classification:\n category: product\n domain: infotech\n", + encoding="utf-8", + ) + snapshot = build_classification_snapshot( + ( + RepositoryRegistration( + "11111111-1111-4111-8111-111111111111", "dirty", "active", repo + ), + ), + observed_at=NOW, + ) + assert snapshot.repositories == () + assert "uncommitted changes" in snapshot.diagnostics[0]["message"] + + +def test_http_port_reports_readiness_and_serves_the_frozen_envelope(tmp_path: Path) -> None: + repo = _repo(tmp_path, "repo-manager") + registry = _registry( + tmp_path / "registry.yaml", + [ + { + "repository_id": "11111111-1111-4111-8111-111111111111", + "slug": "repo-manager", + "lifecycle": "active", + "path": str(repo), + } + ], + ) + app = create_app( + settings=PublisherSettings( + registry_path=registry, + cursor_secret=SECRET, + api_token="publisher-token", + page_size=100, + ) + ) + with TestClient(app) as client: + assert client.get("/readyz").json() == {"status": "ok", "repository_count": 1} + assert client.get("/ports/repositories/classifications").status_code == 401 + response = client.get( + "/ports/repositories/classifications", + headers={"Authorization": "Bearer publisher-token"}, + ) + assert response.status_code == 200 + assert response.json()["snapshot"]["total_repository_count"] == 1 + methods = { + method + for method in client.get("/openapi.json").json()["paths"][ + "/ports/repositories/classifications" + ] + } + assert methods == {"get"} + + +def test_registry_bootstrap_uses_primary_direct_identity_reads(tmp_path: Path) -> None: + _repo(tmp_path, "first") + _repo(tmp_path, "second") + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.url.path) + if request.url.path == "/state/health": + return httpx.Response( + 200, + json={"status": "ok", "instance_role": "primary", "instance_label": "railiance01"}, + ) + slug = request.url.path.removeprefix("/repos/") + identifiers = { + "first": "11111111-1111-4111-8111-111111111111", + "second": "22222222-2222-4222-8222-222222222222", + } + return httpx.Response( + 200, + json={"id": identifiers[slug], "slug": slug, "status": "active"}, + ) + + result = import_state_hub_registry( + tmp_path, + api_base="http://state-hub.invalid", + workers=2, + transport=httpx.MockTransport(handler), + ) + + assert result["ok"] is True + assert len(result["repositories"]) == 2 + assert "/repos/" not in seen + assert {path for path in seen if path.startswith("/repos/")} == { + "/repos/first", + "/repos/second", + } + assert all(Path(row["path"]).is_absolute() for row in result["repositories"]) diff --git a/workplans/RMGR-WP-0013-repository-classification-publisher.md b/workplans/RMGR-WP-0013-repository-classification-publisher.md new file mode 100644 index 0000000..18c26b8 --- /dev/null +++ b/workplans/RMGR-WP-0013-repository-classification-publisher.md @@ -0,0 +1,119 @@ +--- +id: RMGR-WP-0013 +type: workplan +title: "Repository classification projection publisher" +domain: infotech +repo: repo-manager +status: active +owner: codex +topic_slug: infotech +created: "2026-09-01" +updated: "2026-09-01" +parent_project: prj-state-hub-retirement +parent_workplan: STATE-WP-0079 +related: + - HUB-WP-0006 + - RMGR-WP-0008 +--- + +# Repository classification projection publisher + +## Goal + +Publish Repo Manager's validated fleet classification view through the frozen +`helixforge.repository-classification-projection` 1.0.0 `port.repo` contract, +including registrar UUIDs, atomic snapshot provenance, and safe paging, so +hub-core can rebuild repository navigation without reading State Hub tables. + +## Freeze the publisher mapping + +```task +id: RMGR-WP-0013-T01 +status: done +priority: high +``` + +Map repository registry identity, lifecycle, checkout observation, classification +authority, revision provenance, diagnostics, and the exact hub-core envelope. +Keep host paths and credentials outside the published contract. + +**Result (2026-09-01):** the mapping pins registrar UUID, canonical slug, +lifecycle, validated file classification, Git HEAD, classification-file SHA-256, +and canonical UTC observation time. The closed envelope excludes local paths, +remotes, work records, source bodies, and credentials. + +## Implement registry-backed snapshot publication + +```task +id: RMGR-WP-0013-T02 +status: done +priority: high +``` + +Build deterministic full snapshots from an operator-owned local repository +registry, validate every classification at observation time, and publish stable +registrar UUID identities with source fingerprints and Git revisions. + +**Result (2026-09-01):** `repository_publisher.py` loads a strict private +operator registry, rejects duplicate identity and unsupported lifecycle, and +builds full UUID-ordered snapshots. The primary-State-Hub bootstrap uses +bounded direct identity reads with retries and never trusts remote host paths. + +## Implement the HTTP port and paging integrity + +```task +id: RMGR-WP-0013-T03 +status: done +priority: high +``` + +Expose a read-only FastAPI `port.repo` route. Bind opaque cursors to the exact +snapshot and page size with an HMAC, retain bounded in-flight snapshots, and +fail closed on expired, malformed, or mismatched cursors. + +**Result (2026-09-01):** the read-only FastAPI runtime exposes health, +readiness, and `GET /ports/repositories/classifications`. Cursors are +HMAC-SHA256-bound to snapshot, offset, and page size; in-flight snapshots are +bounded; optional bearer authentication is constant-time checked. + +## Prove cross-repository conformance + +```task +id: RMGR-WP-0013-T04 +status: done +priority: high +``` + +Exercise single- and multi-page transfers through hub-core's frozen consumer, +including stable ordering, provenance, malformed registry state, invalid +classification, cursor tampering, and preservation of the last good generation. + +**Result (2026-09-01):** unit and cross-repository tests pass. A live five-page +transfer carried all 123 registered classified checkouts into hub-core and was +accepted as one current generation. Evidence: +`docs/evidence/RMGR-WP-0013-live-conformance-2026-09-01.md`. + +## Package and hand off deployment + +```task +id: RMGR-WP-0013-T05 +status: wait +priority: medium +``` + +Package the publisher runtime, document registry bootstrap and refresh, record +evidence, and hand the concrete endpoint/configuration to HUB-WP-0006-T06. + +Implementation and operating documentation are complete. Deployment remains +waiting on an explicit placement/network owner: the public Core Hub cluster +cannot read the host-local Repo Manager checkout registry, and mounting a broad +home directory or reintroducing State Hub as the live classification source is +not an acceptable implicit choice. + +## Acceptance + +- [x] Every published repository is keyed by its registrar UUID +- [x] The output validates as contract 1.0.0 without hub-core coercion +- [x] Pages cannot be mixed across snapshots or page sizes +- [x] Invalid source state cannot replace hub-core's last accepted generation +- [ ] Runtime and deployment handoff are documented and reproducible