From f582773f5a465359c188cfdeb0f651eb65b459f7 Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 1 Sep 2026 00:49:30 +0200 Subject: [PATCH] feat: consume Repo Manager classification publisher Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c --- docs/runtime.md | 8 +++ hub_core/runtime/app.py | 47 ++++++++++++- hub_core/runtime/config.py | 16 +++++ hub_core/runtime/repo_manager_client.py | 50 ++++++++++++++ tests/test_repo_manager_client.py | 68 +++++++++++++++++++ ...06-repository-classification-navigation.md | 15 ++++ 6 files changed, 201 insertions(+), 3 deletions(-) create mode 100644 hub_core/runtime/repo_manager_client.py create mode 100644 tests/test_repo_manager_client.py diff --git a/docs/runtime.md b/docs/runtime.md index 31a4ff6..88d2336 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -49,6 +49,14 @@ credentials never enter projection payloads. Migration `0003_repository_navigation` stores active state, normalized repositories, and derived facets atomically. +Production can configure the built-in HTTP adapter with +`HUB_CORE_REPO_MANAGER_BASE_URL`; optional +`HUB_CORE_REPO_MANAGER_API_TOKEN` bearer authentication and +`HUB_CORE_REPO_MANAGER_TIMEOUT_SECONDS` remain transport concerns. The runtime +refreshes at startup and every `HUB_CORE_REPO_PROJECTION_REFRESH_SECONDS` +(default 300). A failed scheduled refresh preserves the last generation as +stale; set the interval to `0` only when an external scheduler owns refresh. + The repository query accepts repeated primary/secondary domain, category, capability-tag, business-stake, and business-mechanic filters. Values within a family are ORed and families are ANDed. Opaque cursors are bound to the active diff --git a/hub_core/runtime/app.py b/hub_core/runtime/app.py index 91325a7..5e9efbd 100644 --- a/hub_core/runtime/app.py +++ b/hub_core/runtime/app.py @@ -1,6 +1,7 @@ from __future__ import annotations -from contextlib import asynccontextmanager +import asyncio +from contextlib import asynccontextmanager, suppress from fastapi import FastAPI, Response, status @@ -17,6 +18,7 @@ from hub_core.runtime.repository_navigation import ( from hub_core.runtime.repository_navigation_routes import ( create_repository_navigation_router, ) +from hub_core.runtime.repo_manager_client import HTTPRepoProjectionClient from hub_core.runtime.store import InMemoryPortStore, PortStore from hub_core.runtime.validation import ContractValidator from hub_core.runtime.workload_projection import ( @@ -37,8 +39,20 @@ def create_app( resolved_settings = settings or RuntimeSettings.from_env() resolved_store = port_store or _create_store(resolved_settings) owns_store = port_store is None + resolved_repo_projection_client = repo_projection_client + owns_repo_projection_client = False + if ( + resolved_repo_projection_client is None + and resolved_settings.repo_manager_base_url is not None + ): + resolved_repo_projection_client = HTTPRepoProjectionClient( + resolved_settings.repo_manager_base_url, + api_token=resolved_settings.repo_manager_api_token, + timeout_seconds=resolved_settings.repo_manager_timeout_seconds, + ) + owns_repo_projection_client = True repository_navigation = RepositoryNavigationService( - client=repo_projection_client, + client=resolved_repo_projection_client, store=resolved_store, ) workload_projection = WorkloadProjectionService( @@ -48,19 +62,33 @@ def create_app( @asynccontextmanager async def lifespan(_: FastAPI): - if repo_projection_client is not None: + refresh_task: asyncio.Task[None] | None = None + if resolved_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 + if resolved_settings.repo_projection_refresh_seconds > 0: + refresh_task = asyncio.create_task( + _refresh_repository_projection( + repository_navigation, + resolved_settings.repo_projection_refresh_seconds, + ) + ) if workload_projection_client is not None: try: await workload_projection.refresh() except WorkloadProjectionRejected: pass yield + if refresh_task is not None: + refresh_task.cancel() + with suppress(asyncio.CancelledError): + await refresh_task + if owns_repo_projection_client: + await resolved_repo_projection_client.aclose() # type: ignore[union-attr] if owns_store and (closer := getattr(resolved_store, "aclose", None)): await closer() @@ -117,6 +145,19 @@ def create_app( return app +async def _refresh_repository_projection( + service: RepositoryNavigationService, + interval_seconds: float, +) -> None: + while True: + await asyncio.sleep(interval_seconds) + try: + await service.refresh() + except ProjectionRejected: + # The service records stale state and diagnostics; retry on schedule. + pass + + def _create_store(settings: RuntimeSettings) -> PortStore: if settings.backend == "memory": return InMemoryPortStore() diff --git a/hub_core/runtime/config.py b/hub_core/runtime/config.py index ed6663d..ca04e5e 100644 --- a/hub_core/runtime/config.py +++ b/hub_core/runtime/config.py @@ -28,12 +28,20 @@ class RuntimeSettings: mcp_transport: str = "http" database_url: str | None = None api_token: str | None = None + repo_manager_base_url: str | None = None + repo_manager_api_token: str | None = None + repo_manager_timeout_seconds: float = 10.0 + repo_projection_refresh_seconds: float = 300.0 v2_groups: frozenset[str] = frozenset() v2_write_groups: frozenset[str] = frozenset() legacy_write_groups: frozenset[str] = frozenset() legacy_health: bool = False def __post_init__(self) -> None: + if self.repo_manager_timeout_seconds <= 0: + raise ValueError("Repo Manager timeout must be positive") + if self.repo_projection_refresh_seconds < 0: + raise ValueError("repository projection refresh interval must not be negative") overlap = self.v2_write_groups & self.legacy_write_groups if overlap: joined = ", ".join(sorted(overlap)) @@ -61,6 +69,14 @@ class RuntimeSettings: mcp_transport=os.getenv("HUB_CORE_MCP_TRANSPORT", "http"), database_url=os.getenv("HUB_CORE_DATABASE_URL") or os.getenv("DATABASE_URL"), api_token=os.getenv("HUB_CORE_API_TOKEN") or os.getenv("CORE_HUB_API_TOKEN"), + repo_manager_base_url=os.getenv("HUB_CORE_REPO_MANAGER_BASE_URL"), + repo_manager_api_token=os.getenv("HUB_CORE_REPO_MANAGER_API_TOKEN"), + repo_manager_timeout_seconds=float( + os.getenv("HUB_CORE_REPO_MANAGER_TIMEOUT_SECONDS", "10") + ), + repo_projection_refresh_seconds=float( + os.getenv("HUB_CORE_REPO_PROJECTION_REFRESH_SECONDS", "300") + ), v2_groups=_env_set("HUB_CORE_V2_GROUPS"), v2_write_groups=_env_set("HUB_CORE_V2_WRITE_GROUPS"), legacy_write_groups=_env_set("CORE_HUB_V2_WRITE_GROUPS"), diff --git a/hub_core/runtime/repo_manager_client.py b/hub_core/runtime/repo_manager_client.py new file mode 100644 index 0000000..4f9f9b1 --- /dev/null +++ b/hub_core/runtime/repo_manager_client.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any +from urllib.parse import urlparse + +import httpx + + +class HTTPRepoProjectionClient: + """HTTP adapter for Repo Manager's read-only ``port.repo`` publisher.""" + + def __init__( + self, + base_url: str, + *, + api_token: str | None = None, + timeout_seconds: float = 10.0, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + parsed = urlparse(base_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("Repo Manager base URL must be an absolute HTTP(S) origin") + if timeout_seconds <= 0: + raise ValueError("Repo Manager timeout must be positive") + headers = {"Authorization": f"Bearer {api_token}"} if api_token else None + self._client = httpx.AsyncClient( + base_url=base_url.rstrip("/"), + headers=headers, + timeout=httpx.Timeout(timeout_seconds), + follow_redirects=True, + transport=transport, + ) + + async def fetch_classification_page( + self, cursor: str | None + ) -> Mapping[str, Any]: + params = {"cursor": cursor} if cursor is not None else None + response = await self._client.get( + "/ports/repositories/classifications", + params=params, + ) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise ValueError("Repo Manager classification page must be a JSON object") + return payload + + async def aclose(self) -> None: + await self._client.aclose() diff --git a/tests/test_repo_manager_client.py b/tests/test_repo_manager_client.py new file mode 100644 index 0000000..1a55002 --- /dev/null +++ b/tests/test_repo_manager_client.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +import asyncio +import json + +import httpx +import pytest + +from hub_core.runtime.config import RuntimeSettings +from hub_core.runtime.repo_manager_client import HTTPRepoProjectionClient + + +def test_http_repo_projection_client_forwards_cursor_and_bearer_token() -> None: + seen: list[httpx.Request] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(200, json={"contract_id": "example"}) + + async def exercise() -> dict: + client = HTTPRepoProjectionClient( + "https://repo-manager.invalid", + api_token="test-token", + transport=httpx.MockTransport(handler), + ) + try: + return dict(await client.fetch_classification_page("signed-cursor")) + finally: + await client.aclose() + + payload = asyncio.run(exercise()) + + assert payload == {"contract_id": "example"} + assert seen[0].url.path == "/ports/repositories/classifications" + assert seen[0].url.params["cursor"] == "signed-cursor" + assert seen[0].headers["authorization"] == "Bearer test-token" + + +def test_http_repo_projection_client_rejects_non_object_json() -> None: + async def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=json.dumps([])) + + async def exercise() -> None: + client = HTTPRepoProjectionClient( + "http://repo-manager.invalid", + transport=httpx.MockTransport(handler), + ) + try: + await client.fetch_classification_page(None) + finally: + await client.aclose() + + with pytest.raises(ValueError, match="JSON object"): + asyncio.run(exercise()) + + +def test_runtime_settings_load_repo_projection_transport(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("HUB_CORE_REPO_MANAGER_BASE_URL", "http://repo-manager:8020") + monkeypatch.setenv("HUB_CORE_REPO_MANAGER_API_TOKEN", "token") + monkeypatch.setenv("HUB_CORE_REPO_MANAGER_TIMEOUT_SECONDS", "4.5") + monkeypatch.setenv("HUB_CORE_REPO_PROJECTION_REFRESH_SECONDS", "45") + + settings = RuntimeSettings.from_env() + + assert settings.repo_manager_base_url == "http://repo-manager:8020" + assert settings.repo_manager_api_token == "token" + assert settings.repo_manager_timeout_seconds == 4.5 + assert settings.repo_projection_refresh_seconds == 45.0 diff --git a/workplans/HUB-WP-0006-repository-classification-navigation.md b/workplans/HUB-WP-0006-repository-classification-navigation.md index 91ac259..f698894 100644 --- a/workplans/HUB-WP-0006-repository-classification-navigation.md +++ b/workplans/HUB-WP-0006-repository-classification-navigation.md @@ -166,6 +166,21 @@ therefore still required before comparison can start. Hub-core commits off until those three gates close; rollback remains the flag set to off plus the prior State Hub read route. +Publisher checkpoint 2026-09-01: Repo Manager revision +`10d87e21f50a0205b62688604c78c6f43a7171e5` now emits the exact frozen +envelope through a read-only, bearer-capable HTTP `port.repo` route with stable +registrar UUIDs and HMAC-bound paging. Hub-core's built-in HTTP client refreshes +at startup and on a configured interval. Live conformance accepted 123 +repositories over five pages as one current generation; Repo Manager evidence: +`docs/evidence/RMGR-WP-0013-live-conformance-2026-09-01.md`. + +T06 still waits only on production placement/networking and deployment of the +two committed runtimes. The public Core Hub cluster does not own the host-local +Repo Manager checkout registry; do not solve that by mounting a broad home +directory or restoring State Hub as the classification authority. Once a +scoped publisher endpoint is deployed and the candidate migration/image is +rolled out, comparison and `RM_SLICE_TOPICSPINE` can proceed. + ## Acceptance - [x] Repo Manager v1.0 authority and projection contract accepted