diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 6273d91..762a828 100644 --- a/WORK-RECORDS.md +++ b/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 | diff --git a/hub_core/migrations/versions/0003_repository_navigation.py b/hub_core/migrations/versions/0003_repository_navigation.py new file mode 100644 index 0000000..61bac88 --- /dev/null +++ b/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") diff --git a/hub_core/runtime/app.py b/hub_core/runtime/app.py index da5553b..1711376 100644 --- a/hub_core/runtime/app.py +++ b/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() ) diff --git a/hub_core/runtime/postgres_store.py b/hub_core/runtime/postgres_store.py index b2b6403..97f7d26 100644 --- a/hub_core/runtime/postgres_store.py +++ b/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, diff --git a/hub_core/runtime/repository_navigation.py b/hub_core/runtime/repository_navigation.py new file mode 100644 index 0000000..8f487a2 --- /dev/null +++ b/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) diff --git a/hub_core/runtime/store.py b/hub_core/runtime/store.py index c0e8447..8e7c682 100644 --- a/hub_core/runtime/store.py +++ b/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, diff --git a/hub_core/runtime/tables.py b/hub_core/runtime/tables.py index d268a6c..0c27fe3 100644 --- a/hub_core/runtime/tables.py +++ b/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, diff --git a/tests/test_repository_navigation_ingestion.py b/tests/test_repository_navigation_ingestion.py new file mode 100644 index 0000000..9180991 --- /dev/null +++ b/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()) diff --git a/workplans/HUB-WP-0006-repository-classification-navigation.md b/workplans/HUB-WP-0006-repository-classification-navigation.md index e6a3857..3545985 100644 --- a/workplans/HUB-WP-0006-repository-classification-navigation.md +++ b/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