feat: publish repository classification projections

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c

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()

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))

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()