feat: ingest repository navigation projections

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0230c-b06c-7641-808a-e191b6d1da49

WORK-RECORDS.md

@@ -40,7 +40,7 @@
 | task | HUB-WP-0005-T06 | done | — | workplans/HUB-WP-0005-core-hub-absorption-execution.md |
 | task | HUB-WP-0006-T01 | done | — | workplans/HUB-WP-0006-repository-classification-navigation.md |
 | task | HUB-WP-0006-T02 | done | — | workplans/HUB-WP-0006-repository-classification-navigation.md |
-| task | HUB-WP-0006-T03 | todo | — | workplans/HUB-WP-0006-repository-classification-navigation.md |
+| task | HUB-WP-0006-T03 | done | — | workplans/HUB-WP-0006-repository-classification-navigation.md |
 | task | HUB-WP-0006-T04 | todo | — | workplans/HUB-WP-0006-repository-classification-navigation.md |
 | task | HUB-WP-0006-T05 | todo | — | workplans/HUB-WP-0006-repository-classification-navigation.md |
 | task | HUB-WP-0006-T06 | wait | — | workplans/HUB-WP-0006-repository-classification-navigation.md |

hub_core/migrations/versions/0003_repository_navigation.py

@@ -0,0 +1,89 @@
+"""durable repository navigation projection
+
+Revision ID: 0003_repository_navigation
+Revises: 0002_runtime_ports
+Create Date: 2026-08-22
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0003_repository_navigation"
+down_revision: Union[str, None] = "0002_runtime_ports"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "runtime_repository_navigation_state",
+        sa.Column("projection_id", sa.String(80), primary_key=True),
+        sa.Column("projection_status", sa.String(20), nullable=False),
+        sa.Column("snapshot_id", sa.String(64), nullable=False, unique=True),
+        sa.Column(
+            "source_snapshot", postgresql.JSONB(astext_type=sa.Text()), nullable=False
+        ),
+        sa.Column("source_checked_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("rebuilt_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("content_hash", sa.String(64), nullable=False),
+        sa.Column("repository_count", sa.Integer(), nullable=False),
+        sa.Column("diagnostics", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    )
+    op.create_table(
+        "runtime_repository_navigation_repositories",
+        sa.Column("repository_id", sa.String(36), primary_key=True),
+        sa.Column("slug", sa.String(120), nullable=False, unique=True),
+        sa.Column("lifecycle", sa.String(20), nullable=False),
+        sa.Column("primary_domain", sa.String(40), nullable=False),
+        sa.Column(
+            "secondary_domains", postgresql.JSONB(astext_type=sa.Text()), nullable=False
+        ),
+        sa.Column("category", sa.String(40), nullable=False),
+        sa.Column(
+            "capability_tags", postgresql.JSONB(astext_type=sa.Text()), nullable=False
+        ),
+        sa.Column(
+            "business_stake", postgresql.JSONB(astext_type=sa.Text()), nullable=False
+        ),
+        sa.Column(
+            "business_mechanics", postgresql.JSONB(astext_type=sa.Text()), nullable=False
+        ),
+        sa.Column("provenance", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    )
+    op.create_index(
+        "ix_runtime_repository_navigation_repositories_slug",
+        "runtime_repository_navigation_repositories",
+        ["slug"],
+    )
+    op.create_index(
+        "ix_runtime_repository_navigation_repositories_lifecycle",
+        "runtime_repository_navigation_repositories",
+        ["lifecycle"],
+    )
+    op.create_index(
+        "ix_runtime_repository_navigation_repositories_primary_domain",
+        "runtime_repository_navigation_repositories",
+        ["primary_domain"],
+    )
+    op.create_index(
+        "ix_runtime_repository_navigation_repositories_category",
+        "runtime_repository_navigation_repositories",
+        ["category"],
+    )
+    op.create_table(
+        "runtime_repository_navigation_facets",
+        sa.Column("kind", sa.String(40), primary_key=True),
+        sa.Column("value", sa.String(120), primary_key=True),
+        sa.Column("repository_count", sa.Integer(), nullable=False),
+        sa.Column(
+            "repository_ids", postgresql.JSONB(astext_type=sa.Text()), nullable=False
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("runtime_repository_navigation_facets")
+    op.drop_table("runtime_repository_navigation_repositories")
+    op.drop_table("runtime_repository_navigation_state")

hub_core/runtime/app.py

@@ -9,6 +9,11 @@ from hub_core.runtime.config import RuntimeSettings
 from hub_core.runtime.compat import SQLCompatibilityStore, create_compatibility_router
 from hub_core.runtime.models import HealthResponse, ReadinessResponse
 from hub_core.runtime.ports import create_ports_router
+from hub_core.runtime.repository_navigation import (
+    ProjectionRejected,
+    RepoProjectionClient,
+    RepositoryNavigationService,
+)
 from hub_core.runtime.store import InMemoryPortStore, PortStore
 from hub_core.runtime.validation import ContractValidator
 
@@ -17,13 +22,25 @@ def create_app(
     *,
     settings: RuntimeSettings | None = None,
     port_store: PortStore | None = None,
+    repo_projection_client: RepoProjectionClient | None = None,
 ) -> FastAPI:
     resolved_settings = settings or RuntimeSettings.from_env()
     resolved_store = port_store or _create_store(resolved_settings)
     owns_store = port_store is None
+    repository_navigation = RepositoryNavigationService(
+        client=repo_projection_client,
+        store=resolved_store,
+    )
 
     @asynccontextmanager
     async def lifespan(_: FastAPI):
+        if repo_projection_client is not None:
+            try:
+                await repository_navigation.refresh()
+            except ProjectionRejected:
+                # The readiness dependency reports the rejected or absent
+                # projection while the API remains available for diagnosis.
+                pass
         yield
         if owns_store and (closer := getattr(resolved_store, "aclose", None)):
             await closer()
@@ -42,6 +59,7 @@ def create_app(
         else None
     )
     app.state.contract_validator = ContractValidator()
+    app.state.repository_navigation = repository_navigation
 
     @app.get("/healthz", response_model=HealthResponse, tags=["system"])
     async def healthz() -> HealthResponse:
@@ -52,7 +70,10 @@ def create_app(
 
     @app.get("/readyz", response_model=ReadinessResponse, tags=["system"])
     async def readyz(response: Response) -> ReadinessResponse:
-        dependency_checks = await resolved_store.readiness_checks()
+        dependency_checks = {
+            **await resolved_store.readiness_checks(),
+            **await repository_navigation.readiness_checks(),
+        }
         ready = resolved_settings.is_ready(resolved_store.backend_name) and all(
             value in {"ok", "not_applicable"} for value in dependency_checks.values()
         )

hub_core/runtime/postgres_store.py

@@ -21,6 +21,7 @@ from hub_core.runtime.models import (
     Provenance,
     RegistryRegistration,
 )
+from hub_core.runtime.repository_navigation import NavigationProjection
 from hub_core.runtime.tables import (
     compat_api_keys,
     compat_hubs,
@@ -28,6 +29,9 @@ from hub_core.runtime.tables import (
     runtime_interaction_events,
     runtime_messages,
     runtime_progress_events,
+    runtime_repository_navigation_facets,
+    runtime_repository_navigation_repositories,
+    runtime_repository_navigation_state,
     runtime_registrations,
 )
 
@@ -52,7 +56,12 @@ class PostgresPortStore:
                 # A live connection alone can hide a mis-granted runtime lease.
                 # Exercise the durable port, compatibility, and authorization
                 # tables that protected traffic actually depends on.
-                for table in (runtime_audit_ledger, compat_hubs, compat_api_keys):
+                for table in (
+                    runtime_audit_ledger,
+                    compat_hubs,
+                    compat_api_keys,
+                    runtime_repository_navigation_state,
+                ):
                     await connection.execute(
                         sa.select(sa.literal(1)).select_from(table).limit(1)
                     )
@@ -204,6 +213,135 @@ class PostgresPortStore:
             },
         )
 
+    async def get_repository_navigation(self) -> NavigationProjection | None:
+        async with self.sessions() as session:
+            state = (
+                await session.execute(
+                    sa.select(runtime_repository_navigation_state).where(
+                        runtime_repository_navigation_state.c.projection_id
+                        == "repository_navigation"
+                    )
+                )
+            ).mappings().one_or_none()
+            if state is None:
+                return None
+            repository_rows = (
+                await session.execute(
+                    sa.select(runtime_repository_navigation_repositories).order_by(
+                        runtime_repository_navigation_repositories.c.repository_id
+                    )
+                )
+            ).mappings()
+            facet_rows = (
+                await session.execute(
+                    sa.select(runtime_repository_navigation_facets).order_by(
+                        runtime_repository_navigation_facets.c.kind,
+                        runtime_repository_navigation_facets.c.value,
+                    )
+                )
+            ).mappings()
+            repositories = tuple(
+                {
+                    "repository_id": str(row["repository_id"]),
+                    "slug": row["slug"],
+                    "lifecycle": row["lifecycle"],
+                    "primary_domain": row["primary_domain"],
+                    "secondary_domains": list(row["secondary_domains"] or []),
+                    "category": row["category"],
+                    "capability_tags": list(row["capability_tags"] or []),
+                    "business_stake": list(row["business_stake"] or []),
+                    "business_mechanics": list(row["business_mechanics"] or []),
+                    "provenance": dict(row["provenance"] or {}),
+                }
+                for row in repository_rows
+            )
+            facets = tuple(
+                {
+                    "kind": row["kind"],
+                    "value": row["value"],
+                    "repository_count": row["repository_count"],
+                    "repository_ids": list(row["repository_ids"] or []),
+                }
+                for row in facet_rows
+            )
+        return NavigationProjection(
+            projection_status=state["projection_status"],
+            source_snapshot=dict(state["source_snapshot"]),
+            source_checked_at=state["source_checked_at"],
+            rebuilt_at=state["rebuilt_at"],
+            content_hash=state["content_hash"],
+            repositories=repositories,
+            facets=facets,
+            diagnostics=tuple(state["diagnostics"] or []),
+        )
+
+    async def replace_repository_navigation(
+        self, projection: NavigationProjection
+    ) -> None:
+        state_values = {
+            "projection_status": projection.projection_status,
+            "snapshot_id": projection.source_snapshot["snapshot_id"],
+            "source_snapshot": projection.source_snapshot,
+            "source_checked_at": projection.source_checked_at,
+            "rebuilt_at": projection.rebuilt_at,
+            "content_hash": projection.content_hash,
+            "repository_count": len(projection.repositories),
+            "diagnostics": list(projection.diagnostics),
+        }
+        async with self.sessions.begin() as session:
+            await session.execute(runtime_repository_navigation_facets.delete())
+            await session.execute(runtime_repository_navigation_repositories.delete())
+            if projection.repositories:
+                await session.execute(
+                    runtime_repository_navigation_repositories.insert(),
+                    [dict(repository) for repository in projection.repositories],
+                )
+            if projection.facets:
+                await session.execute(
+                    runtime_repository_navigation_facets.insert(),
+                    [dict(facet) for facet in projection.facets],
+                )
+            exists = (
+                await session.execute(
+                    sa.select(runtime_repository_navigation_state.c.projection_id).where(
+                        runtime_repository_navigation_state.c.projection_id
+                        == "repository_navigation"
+                    )
+                )
+            ).scalar_one_or_none()
+            if exists is None:
+                await session.execute(
+                    runtime_repository_navigation_state.insert().values(
+                        projection_id="repository_navigation", **state_values
+                    )
+                )
+            else:
+                await session.execute(
+                    runtime_repository_navigation_state.update()
+                    .where(
+                        runtime_repository_navigation_state.c.projection_id
+                        == "repository_navigation"
+                    )
+                    .values(**state_values)
+                )
+
+    async def mark_repository_navigation_stale(
+        self, *, checked_at: datetime, diagnostic: Mapping[str, Any]
+    ) -> None:
+        async with self.sessions.begin() as session:
+            await session.execute(
+                runtime_repository_navigation_state.update()
+                .where(
+                    runtime_repository_navigation_state.c.projection_id
+                    == "repository_navigation"
+                )
+                .values(
+                    projection_status="stale",
+                    source_checked_at=checked_at,
+                    diagnostics=[dict(diagnostic)],
+                )
+            )
+
     async def _append_event(
         self,
         command: EventCommand,

hub_core/runtime/repository_navigation.py

@@ -0,0 +1,357 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+from collections.abc import Mapping
+from copy import deepcopy
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Literal, Protocol
+
+from jsonschema import Draft202012Validator, FormatChecker, ValidationError
+
+from hub_core.contracts import (
+    REPOSITORY_NAVIGATION_CONTRACT_ID,
+    REPOSITORY_NAVIGATION_CONTRACT_VERSION,
+    repository_navigation_contract_root,
+)
+
+
+class RepoProjectionClient(Protocol):
+    """Injected `port.repo` reader; transport and credentials stay outside the projection."""
+
+    async def fetch_classification_page(
+        self, cursor: str | None
+    ) -> Mapping[str, Any]: ...
+
+
+class RepositoryNavigationStore(Protocol):
+    async def get_repository_navigation(self) -> NavigationProjection | None: ...
+
+    async def replace_repository_navigation(
+        self, projection: NavigationProjection
+    ) -> None: ...
+
+    async def mark_repository_navigation_stale(
+        self, *, checked_at: datetime, diagnostic: Mapping[str, Any]
+    ) -> None: ...
+
+
+class ProjectionRejected(ValueError):
+    """The upstream transfer cannot safely update the active projection."""
+
+
+@dataclass(frozen=True, slots=True)
+class NavigationProjection:
+    projection_status: Literal["current", "stale"]
+    source_snapshot: dict[str, Any]
+    source_checked_at: datetime
+    rebuilt_at: datetime
+    content_hash: str
+    repositories: tuple[dict[str, Any], ...]
+    facets: tuple[dict[str, Any], ...]
+    diagnostics: tuple[dict[str, Any], ...] = ()
+
+    def to_contract(self) -> dict[str, Any]:
+        return {
+            "contract_id": REPOSITORY_NAVIGATION_CONTRACT_ID,
+            "contract_version": REPOSITORY_NAVIGATION_CONTRACT_VERSION,
+            "projection_id": "repository_navigation",
+            "projection_status": self.projection_status,
+            "source_snapshot": deepcopy(self.source_snapshot),
+            "source_checked_at": self.source_checked_at.isoformat(),
+            "rebuilt_at": self.rebuilt_at.isoformat(),
+            "content_hash": self.content_hash,
+            "repositories": deepcopy(list(self.repositories)),
+            "facets": deepcopy(list(self.facets)),
+            "legacy_topic_aliases": [],
+            "diagnostics": deepcopy(list(self.diagnostics)),
+            "next_cursor": None,
+            "total_repository_count": len(self.repositories),
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class NavigationRefreshResult:
+    status: Literal["accepted", "duplicate"]
+    snapshot_id: str
+    content_hash: str
+    repository_count: int
+
+
+class RepositoryNavigationService:
+    """Validate and atomically materialize Repo Manager classification snapshots."""
+
+    def __init__(
+        self,
+        *,
+        client: RepoProjectionClient | None,
+        store: RepositoryNavigationStore,
+    ) -> None:
+        self.client = client
+        self.store = store
+        self._validator = _input_validator()
+        self._lock = asyncio.Lock()
+        self._last_error: str | None = None
+
+    async def readiness_checks(self) -> dict[str, str]:
+        if self.client is None:
+            return {"repo_manager_projection": "not_applicable"}
+        current = await self.store.get_repository_navigation()
+        if current is None:
+            return {"repo_manager_projection": "unavailable"}
+        return {
+            "repo_manager_projection": (
+                "ok" if current.projection_status == "current" else "stale"
+            )
+        }
+
+    async def refresh(self) -> NavigationRefreshResult:
+        if self.client is None:
+            raise RuntimeError("no port.repo projection client is configured")
+        async with self._lock:
+            checked_at = _now()
+            try:
+                pages = await self._fetch_transfer()
+                result = await self._apply_transfer(pages, checked_at=checked_at)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                diagnostic = {
+                    "severity": "error",
+                    "code": "repo_projection.rejected",
+                    "message": str(exc)[:1000] or type(exc).__name__,
+                }
+                self._last_error = diagnostic["message"]
+                await self.store.mark_repository_navigation_stale(
+                    checked_at=checked_at,
+                    diagnostic=diagnostic,
+                )
+                if isinstance(exc, ProjectionRejected):
+                    raise
+                raise ProjectionRejected(diagnostic["message"]) from exc
+            self._last_error = None
+            return result
+
+    async def _fetch_transfer(self) -> list[dict[str, Any]]:
+        assert self.client is not None
+        pages: list[dict[str, Any]] = []
+        cursor: str | None = None
+        seen_cursors: set[str] = set()
+        while True:
+            raw = dict(await self.client.fetch_classification_page(cursor))
+            try:
+                self._validator.validate(raw)
+            except ValidationError as exc:
+                location = ".".join(str(part) for part in exc.absolute_path) or "root"
+                raise ProjectionRejected(f"invalid projection page at {location}: {exc.message}") from exc
+            pages.append(raw)
+            snapshot = raw["snapshot"]
+            next_cursor = snapshot["next_cursor"]
+            if snapshot["final_page"] != (next_cursor is None):
+                raise ProjectionRejected("final_page must be true exactly when next_cursor is null")
+            if snapshot["final_page"]:
+                return pages
+            if next_cursor in seen_cursors:
+                raise ProjectionRejected("projection cursor cycle detected")
+            seen_cursors.add(next_cursor)
+            cursor = next_cursor
+            if len(pages) >= 10000:
+                raise ProjectionRejected("projection transfer exceeds the 10000-page limit")
+
+    async def _apply_transfer(
+        self,
+        pages: list[dict[str, Any]],
+        *,
+        checked_at: datetime,
+    ) -> NavigationRefreshResult:
+        _validate_transfer_semantics(pages)
+        first_snapshot = pages[0]["snapshot"]
+        current = await self.store.get_repository_navigation()
+        snapshot_id = first_snapshot["snapshot_id"]
+        duplicate_snapshot = bool(
+            current and current.source_snapshot["snapshot_id"] == snapshot_id
+        )
+        if (
+            current
+            and not duplicate_snapshot
+            and _parse_time(first_snapshot["generated_at"])
+            <= _parse_time(current.source_snapshot["generated_at"])
+        ):
+            raise ProjectionRejected("projection snapshot is older than the active generation")
+
+        mode = first_snapshot["mode"]
+        if mode == "incremental" and current is None:
+            raise ProjectionRejected("incremental projection requires an accepted base generation")
+        repositories = {
+            item["repository_id"]: deepcopy(item)
+            for item in (current.repositories if current and mode == "incremental" else ())
+        }
+        for page in pages:
+            for change in page["repositories"]:
+                repository_id = change["repository_id"]
+                if change["operation"] == "delete":
+                    repositories.pop(repository_id, None)
+                else:
+                    repositories[repository_id] = _normalize_repository(change)
+
+        normalized = tuple(repositories[key] for key in sorted(repositories))
+        content_hash = _content_hash(normalized)
+        if mode == "full" and first_snapshot["total_repository_count"] != len(normalized):
+            raise ProjectionRejected("full snapshot total_repository_count does not match upserts")
+        if duplicate_snapshot and current and content_hash != current.content_hash:
+            raise ProjectionRejected("snapshot_id was reused with different projection content")
+        projection = NavigationProjection(
+            projection_status="current",
+            source_snapshot={
+                "snapshot_id": snapshot_id,
+                "source_system": "repo-manager",
+                "source_contract_id": pages[0]["source"]["classification_contract_id"],
+                "source_contract_version": pages[0]["source"][
+                    "classification_contract_version"
+                ],
+                "source_revision": first_snapshot["source_revision"],
+                "generated_at": first_snapshot["generated_at"],
+            },
+            source_checked_at=checked_at,
+            rebuilt_at=_now(),
+            content_hash=content_hash,
+            repositories=normalized,
+            facets=_derive_facets(normalized),
+        )
+        await self.store.replace_repository_navigation(projection)
+        return NavigationRefreshResult(
+            status="duplicate" if duplicate_snapshot else "accepted",
+            snapshot_id=snapshot_id,
+            content_hash=content_hash,
+            repository_count=len(normalized),
+        )
+
+
+def _input_validator() -> Draft202012Validator:
+    resource = repository_navigation_contract_root().joinpath(
+        "schemas", "repository-classification-page.schema.json"
+    )
+    schema = json.loads(resource.read_text(encoding="utf-8"))
+    Draft202012Validator.check_schema(schema)
+    return Draft202012Validator(schema, format_checker=FormatChecker())
+
+
+def _validate_transfer_semantics(pages: list[dict[str, Any]]) -> None:
+    if not pages:
+        raise ProjectionRejected("projection transfer contains no pages")
+    first = pages[0]
+    expected = {
+        key: first["snapshot"][key]
+        for key in (
+            "snapshot_id",
+            "mode",
+            "generated_at",
+            "source_revision",
+            "total_repository_count",
+        )
+    }
+    expected_source = first["source"]
+    seen_ids: set[str] = set()
+    ordered_ids: list[str] = []
+    previous_cursor: str | None = None
+    for index, page in enumerate(pages):
+        snapshot = page["snapshot"]
+        if any(snapshot[key] != value for key, value in expected.items()):
+            raise ProjectionRejected("projection snapshot metadata changed between pages")
+        if page["source"] != expected_source:
+            raise ProjectionRejected("projection source metadata changed between pages")
+        if snapshot["page_cursor"] != previous_cursor:
+            raise ProjectionRejected(f"page {index} does not match the requested cursor")
+        if any(item["severity"] == "error" for item in page["diagnostics"]):
+            raise ProjectionRejected("projection transfer contains an error diagnostic")
+        previous_cursor = snapshot["next_cursor"]
+        for change in page["repositories"]:
+            repository_id = change["repository_id"]
+            if repository_id in seen_ids:
+                raise ProjectionRejected(f"duplicate repository_id {repository_id}")
+            seen_ids.add(repository_id)
+            ordered_ids.append(repository_id)
+            if (
+                change["operation"] == "upsert"
+                and change["classification"]["domain"]
+                in change["classification"].get("secondary_domains", [])
+            ):
+                raise ProjectionRejected(
+                    f"repository {repository_id} repeats its primary domain as secondary"
+                )
+            if snapshot["mode"] == "full" and change["operation"] != "upsert":
+                raise ProjectionRejected("full snapshots may contain only upserts")
+    if ordered_ids != sorted(ordered_ids):
+        raise ProjectionRejected("repository changes are not ordered by repository_id")
+    if expected["mode"] == "full" and expected["total_repository_count"] is None:
+        raise ProjectionRejected("full snapshot requires total_repository_count")
+
+
+def _normalize_repository(change: Mapping[str, Any]) -> dict[str, Any]:
+    classification = change["classification"]
+    repository_id = str(change["repository_id"])
+    revision = change["revision"]
+    return {
+        "repository_id": repository_id,
+        "slug": change["slug"],
+        "lifecycle": change["lifecycle"],
+        "primary_domain": classification["domain"],
+        "secondary_domains": sorted(classification.get("secondary_domains", [])),
+        "category": classification["category"],
+        "capability_tags": sorted(classification.get("capability_tags", [])),
+        "business_stake": sorted(classification.get("business_stake", [])),
+        "business_mechanics": sorted(classification.get("business_mechanics", [])),
+        "provenance": {
+            "source_system": "repo-manager",
+            "source_ref": f"repo-manager://repositories/{repository_id}",
+            "classification_contract_version": "1.0",
+            "repository_revision": revision["head_sha"],
+            "source_fingerprint": revision["source_fingerprint"],
+            "observed_at": revision["observed_at"],
+        },
+    }
+
+
+def _derive_facets(
+    repositories: tuple[dict[str, Any], ...],
+) -> tuple[dict[str, Any], ...]:
+    memberships: dict[tuple[str, str], list[str]] = {}
+    for repository in repositories:
+        repository_id = repository["repository_id"]
+        values: tuple[tuple[str, list[str]], ...] = (
+            ("primary_domain", [repository["primary_domain"]]),
+            ("secondary_domain", repository["secondary_domains"]),
+            ("category", [repository["category"]]),
+            ("capability_tag", repository["capability_tags"]),
+            ("business_stake", repository["business_stake"]),
+            ("business_mechanic", repository["business_mechanics"]),
+        )
+        for kind, facet_values in values:
+            for value in facet_values:
+                memberships.setdefault((kind, value), []).append(repository_id)
+    return tuple(
+        {
+            "kind": kind,
+            "value": value,
+            "repository_count": len(repository_ids),
+            "repository_ids": repository_ids,
+        }
+        for (kind, value), repository_ids in sorted(memberships.items())
+    )
+
+
+def _content_hash(repositories: tuple[dict[str, Any], ...]) -> str:
+    encoded = json.dumps(
+        list(repositories), sort_keys=True, separators=(",", ":")
+    ).encode()
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _parse_time(value: str) -> datetime:
+    return datetime.fromisoformat(value.replace("Z", "+00:00"))
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)

hub_core/runtime/store.py

@@ -19,6 +19,7 @@ from hub_core.runtime.models import (
     Provenance,
     RegistryRegistration,
 )
+from hub_core.runtime.repository_navigation import NavigationProjection
 
 
 class PortStore(Protocol):
@@ -44,6 +45,16 @@ class PortStore(Protocol):
 
     async def query_projection(self, projection_id: str) -> PortRecord | None: ...
 
+    async def get_repository_navigation(self) -> NavigationProjection | None: ...
+
+    async def replace_repository_navigation(
+        self, projection: NavigationProjection
+    ) -> None: ...
+
+    async def mark_repository_navigation_stale(
+        self, *, checked_at: datetime, diagnostic: Mapping[str, Any]
+    ) -> None: ...
+
 
 class InMemoryPortStore:
     """Deterministic ephemeral backend for local runtime and conformance tests.
@@ -60,6 +71,7 @@ class InMemoryPortStore:
         self._messages: list[dict[str, Any]] = []
         self._progress_events: list[dict[str, Any]] = []
         self._interaction_events: list[dict[str, Any]] = []
+        self._repository_navigation: NavigationProjection | None = None
 
     async def readiness_checks(self) -> dict[str, str]:
         return {"database": "not_applicable"}
@@ -134,6 +146,34 @@ class InMemoryPortStore:
             },
         )
 
+    async def get_repository_navigation(self) -> NavigationProjection | None:
+        async with self._lock:
+            return deepcopy(self._repository_navigation)
+
+    async def replace_repository_navigation(
+        self, projection: NavigationProjection
+    ) -> None:
+        async with self._lock:
+            self._repository_navigation = deepcopy(projection)
+
+    async def mark_repository_navigation_stale(
+        self, *, checked_at: datetime, diagnostic: Mapping[str, Any]
+    ) -> None:
+        async with self._lock:
+            current = self._repository_navigation
+            if current is None:
+                return
+            self._repository_navigation = NavigationProjection(
+                projection_status="stale",
+                source_snapshot=deepcopy(current.source_snapshot),
+                source_checked_at=checked_at,
+                rebuilt_at=current.rebuilt_at,
+                content_hash=current.content_hash,
+                repositories=deepcopy(current.repositories),
+                facets=deepcopy(current.facets),
+                diagnostics=(deepcopy(dict(diagnostic)),),
+            )
+
     async def _append_event(
         self,
         command: EventCommand,

hub_core/runtime/tables.py

@@ -60,6 +60,44 @@ runtime_audit_ledger = sa.Table(
     sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False, index=True),
 )
 
+runtime_repository_navigation_state = sa.Table(
+    "runtime_repository_navigation_state",
+    runtime_metadata,
+    sa.Column("projection_id", sa.String(80), primary_key=True),
+    sa.Column("projection_status", sa.String(20), nullable=False),
+    sa.Column("snapshot_id", sa.String(64), nullable=False, unique=True),
+    sa.Column("source_snapshot", sa.JSON(), nullable=False),
+    sa.Column("source_checked_at", sa.DateTime(timezone=True), nullable=False),
+    sa.Column("rebuilt_at", sa.DateTime(timezone=True), nullable=False),
+    sa.Column("content_hash", sa.String(64), nullable=False),
+    sa.Column("repository_count", sa.Integer(), nullable=False),
+    sa.Column("diagnostics", sa.JSON(), nullable=False),
+)
+
+runtime_repository_navigation_repositories = sa.Table(
+    "runtime_repository_navigation_repositories",
+    runtime_metadata,
+    sa.Column("repository_id", sa.String(36), primary_key=True),
+    sa.Column("slug", sa.String(120), nullable=False, unique=True, index=True),
+    sa.Column("lifecycle", sa.String(20), nullable=False, index=True),
+    sa.Column("primary_domain", sa.String(40), nullable=False, index=True),
+    sa.Column("secondary_domains", sa.JSON(), nullable=False),
+    sa.Column("category", sa.String(40), nullable=False, index=True),
+    sa.Column("capability_tags", sa.JSON(), nullable=False),
+    sa.Column("business_stake", sa.JSON(), nullable=False),
+    sa.Column("business_mechanics", sa.JSON(), nullable=False),
+    sa.Column("provenance", sa.JSON(), nullable=False),
+)
+
+runtime_repository_navigation_facets = sa.Table(
+    "runtime_repository_navigation_facets",
+    runtime_metadata,
+    sa.Column("kind", sa.String(40), primary_key=True),
+    sa.Column("value", sa.String(120), primary_key=True),
+    sa.Column("repository_count", sa.Integer(), nullable=False),
+    sa.Column("repository_ids", sa.JSON(), nullable=False),
+)
+
 compat_hubs = sa.Table(
     "compat_hubs",
     runtime_metadata,

tests/test_repository_navigation_ingestion.py

@@ -0,0 +1,235 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from copy import deepcopy
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from hub_core.contracts import repository_navigation_contract_root
+from hub_core.runtime.app import create_app
+from hub_core.runtime.config import RuntimeSettings
+from hub_core.runtime.postgres_store import PostgresPortStore
+from hub_core.runtime.repository_navigation import (
+    ProjectionRejected,
+    RepositoryNavigationService,
+)
+from hub_core.runtime.store import InMemoryPortStore
+from hub_core.runtime.tables import runtime_metadata
+
+
+def fixture_page() -> dict[str, Any]:
+    resource = repository_navigation_contract_root().joinpath(
+        "fixtures", "repository-classification-page.json"
+    )
+    return json.loads(resource.read_text(encoding="utf-8"))
+
+
+class PageClient:
+    def __init__(self, pages: list[dict[str, Any]]) -> None:
+        self.pages = pages
+        self.calls: list[str | None] = []
+
+    async def fetch_classification_page(self, cursor: str | None) -> dict[str, Any]:
+        self.calls.append(cursor)
+        for page in self.pages:
+            if page["snapshot"]["page_cursor"] == cursor:
+                return deepcopy(page)
+        raise RuntimeError(f"unexpected cursor {cursor!r}")
+
+
+class FailingClient:
+    async def fetch_classification_page(self, cursor: str | None) -> dict[str, Any]:
+        raise RuntimeError("repo-manager unavailable")
+
+
+def test_full_rebuild_is_deterministic_and_duplicate_delivery_is_idempotent() -> None:
+    async def run() -> None:
+        page = fixture_page()
+        store = InMemoryPortStore()
+        service = RepositoryNavigationService(client=PageClient([page]), store=store)
+
+        first = await service.refresh()
+        duplicate = await service.refresh()
+        projection = await store.get_repository_navigation()
+
+        assert first.status == "accepted"
+        assert duplicate.status == "duplicate"
+        assert duplicate.content_hash == first.content_hash
+        assert projection is not None
+        assert [row["repository_id"] for row in projection.repositories] == sorted(
+            row["repository_id"] for row in projection.repositories
+        )
+        assert {facet["kind"] for facet in projection.facets} == {
+            "primary_domain",
+            "secondary_domain",
+            "category",
+            "capability_tag",
+            "business_stake",
+            "business_mechanic",
+        }
+        assert await service.readiness_checks() == {"repo_manager_projection": "ok"}
+
+    asyncio.run(run())
+
+
+def test_incremental_upsert_and_delete_rebuild_facets_atomically() -> None:
+    async def run() -> None:
+        initial = fixture_page()
+        store = InMemoryPortStore()
+        first_service = RepositoryNavigationService(
+            client=PageClient([initial]), store=store
+        )
+        await first_service.refresh()
+
+        incremental = fixture_page()
+        incremental["snapshot"].update(
+            {
+                "snapshot_id": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",
+                "mode": "incremental",
+                "generated_at": "2026-08-22T12:05:00Z",
+                "source_revision": "123456789abcdef0123456789abcdef012345678",
+                "total_repository_count": None,
+            }
+        )
+        delete = {
+            "operation": "delete",
+            "repository_id": "11111111-1111-4111-8111-111111111111",
+            "slug": "hub-core",
+            "revision": deepcopy(initial["repositories"][0]["revision"]),
+        }
+        upsert = deepcopy(initial["repositories"][1])
+        upsert["repository_id"] = "33333333-3333-4333-8333-333333333333"
+        upsert["slug"] = "classification-catalog"
+        upsert["classification"]["category"] = "product"
+        incremental["repositories"] = [delete, upsert]
+
+        result = await RepositoryNavigationService(
+            client=PageClient([incremental]), store=store
+        ).refresh()
+        projection = await store.get_repository_navigation()
+
+        assert result.status == "accepted"
+        assert projection is not None
+        assert [row["slug"] for row in projection.repositories] == [
+            "repo-manager",
+            "classification-catalog",
+        ]
+        assert any(
+            facet["kind"] == "category"
+            and facet["value"] == "product"
+            and facet["repository_count"] == 1
+            for facet in projection.facets
+        )
+
+    asyncio.run(run())
+
+
+def test_rejected_version_preserves_last_projection_and_marks_it_stale() -> None:
+    async def run() -> None:
+        store = InMemoryPortStore()
+        initial = fixture_page()
+        await RepositoryNavigationService(
+            client=PageClient([initial]), store=store
+        ).refresh()
+        before = await store.get_repository_navigation()
+
+        unsupported = fixture_page()
+        unsupported["contract_version"] = "2.0.0"
+        service = RepositoryNavigationService(
+            client=PageClient([unsupported]), store=store
+        )
+        with pytest.raises(ProjectionRejected, match="invalid projection page"):
+            await service.refresh()
+        after = await store.get_repository_navigation()
+
+        assert before is not None and after is not None
+        assert after.content_hash == before.content_hash
+        assert after.repositories == before.repositories
+        assert after.projection_status == "stale"
+        assert after.diagnostics[0]["code"] == "repo_projection.rejected"
+        assert await service.readiness_checks() == {"repo_manager_projection": "stale"}
+
+    asyncio.run(run())
+
+
+def test_upstream_outage_without_a_generation_fails_readiness_closed() -> None:
+    async def run() -> None:
+        store = InMemoryPortStore()
+        service = RepositoryNavigationService(client=FailingClient(), store=store)
+
+        with pytest.raises(ProjectionRejected, match="repo-manager unavailable"):
+            await service.refresh()
+
+        assert await store.get_repository_navigation() is None
+        assert await service.readiness_checks() == {
+            "repo_manager_projection": "unavailable"
+        }
+
+    asyncio.run(run())
+
+
+def test_injected_port_repo_client_refreshes_at_startup_and_controls_readiness() -> None:
+    settings = RuntimeSettings(
+        environment="test", backend="memory", allow_ephemeral=True
+    )
+    store = InMemoryPortStore()
+    app = create_app(
+        settings=settings,
+        port_store=store,
+        repo_projection_client=PageClient([fixture_page()]),
+    )
+
+    with TestClient(app) as runtime:
+        response = runtime.get("/readyz")
+
+    assert response.status_code == 200
+    assert response.json()["checks"]["repo_manager_projection"] == "ok"
+
+
+def test_failed_startup_refresh_keeps_api_up_for_diagnosis_but_not_ready() -> None:
+    settings = RuntimeSettings(
+        environment="test", backend="memory", allow_ephemeral=True
+    )
+    app = create_app(
+        settings=settings,
+        port_store=InMemoryPortStore(),
+        repo_projection_client=FailingClient(),
+    )
+
+    with TestClient(app) as runtime:
+        assert runtime.get("/healthz").status_code == 200
+        response = runtime.get("/readyz")
+
+    assert response.status_code == 503
+    assert response.json()["checks"]["repo_manager_projection"] == "unavailable"
+
+
+def test_postgres_projection_survives_store_reopen(tmp_path) -> None:
+    database_url = f"sqlite+aiosqlite:///{tmp_path / 'navigation.db'}"
+
+    async def run() -> None:
+        engine = create_async_engine(database_url)
+        async with engine.begin() as connection:
+            await connection.run_sync(runtime_metadata.create_all)
+        await engine.dispose()
+
+        first_store = PostgresPortStore.from_url(database_url)
+        result = await RepositoryNavigationService(
+            client=PageClient([fixture_page()]), store=first_store
+        ).refresh()
+        await first_store.aclose()
+
+        second_store = PostgresPortStore.from_url(database_url)
+        projection = await second_store.get_repository_navigation()
+        await second_store.aclose()
+
+        assert projection is not None
+        assert projection.content_hash == result.content_hash
+        assert len(projection.repositories) == 2
+        assert projection.source_snapshot["source_system"] == "repo-manager"
+
+    asyncio.run(run())

workplans/HUB-WP-0006-repository-classification-navigation.md

@@ -74,7 +74,7 @@ validate, the wheel contains all eight artifacts, and the full suite passes
 
 ```task
 id: HUB-WP-0006-T03
-status: todo
+status: done
 priority: high
 state_hub_task_id: "2884eb0f-046f-4122-a422-8a06c7c32298"
 ```
@@ -84,6 +84,17 @@ upsert/delete behavior, durable PostgreSQL projection tables, content hashes,
 and dependency-aware readiness. Reject invalid or unsupported projection
 versions without coercion.
 
+Completed 2026-08-22. `RepositoryNavigationService` consumes an injected
+`port.repo` client, validates the exact v1.0.0 envelope plus cross-page
+semantics, and applies deterministic full or incremental generations with
+idempotent duplicate handling. Migration `0003_repository_navigation`
+persists active state, normalized repository rows, and all derived facets in
+one atomic transaction. Invalid versions and transfers preserve the last
+generation as stale; readiness distinguishes current, stale, unavailable, and
+disabled dependencies. PostgreSQL migration SQL renders through head and the
+77-test suite covers durable reopen, startup injection, outage, rejection,
+full rebuild, upsert, and deletion.
+
 ## Expose projection query and MCP navigation
 
 ```task