From ab936a1e98a51bdf545beeee24981193e3f4f809 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 23 Aug 2026 10:54:40 +0200 Subject: [PATCH] feat: transport authoritative workload projections Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0230c-b06c-7641-808a-e191b6d1da49 --- INTENT.md | 4 +- README.md | 6 + SCOPE.md | 16 +- WORK-RECORDS.md | 6 + docs/runtime.md | 9 + hub_core/contracts/__init__.py | 11 + .../contracts/workload_projection/__init__.py | 1 + .../workload_projection/v1_0_0/README.md | 13 + .../workload_projection/v1_0_0/__init__.py | 1 + .../v1_0_0/compatibility-matrix.json | 25 ++ .../fixtures/repo-manager-nine-workloads.json | 139 ++++++ .../openapi/workload-projection.openapi.json | 43 ++ .../workload-projection-page.schema.json | 82 ++++ .../schemas/workload-projection.schema.json | 66 +++ .../schemas/workload-resolution.schema.json | 65 +++ hub_core/mcp/server.py | 38 ++ .../versions/0004_workload_projection.py | 55 +++ hub_core/runtime/app.py | 21 + hub_core/runtime/postgres_store.py | 101 +++++ hub_core/runtime/store.py | 35 ++ hub_core/runtime/tables.py | 30 ++ hub_core/runtime/workload_projection.py | 421 ++++++++++++++++++ .../runtime/workload_projection_routes.py | 77 ++++ tests/test_mcp.py | 2 + tests/test_workload_projection.py | 271 +++++++++++ tests/test_workload_projection_contract.py | 95 ++++ ...B-WP-0007-workload-projection-transport.md | 53 ++- 27 files changed, 1671 insertions(+), 15 deletions(-) create mode 100644 hub_core/contracts/workload_projection/__init__.py create mode 100644 hub_core/contracts/workload_projection/v1_0_0/README.md create mode 100644 hub_core/contracts/workload_projection/v1_0_0/__init__.py create mode 100644 hub_core/contracts/workload_projection/v1_0_0/compatibility-matrix.json create mode 100644 hub_core/contracts/workload_projection/v1_0_0/fixtures/repo-manager-nine-workloads.json create mode 100644 hub_core/contracts/workload_projection/v1_0_0/openapi/workload-projection.openapi.json create mode 100644 hub_core/contracts/workload_projection/v1_0_0/schemas/workload-projection-page.schema.json create mode 100644 hub_core/contracts/workload_projection/v1_0_0/schemas/workload-projection.schema.json create mode 100644 hub_core/contracts/workload_projection/v1_0_0/schemas/workload-resolution.schema.json create mode 100644 hub_core/migrations/versions/0004_workload_projection.py create mode 100644 hub_core/runtime/workload_projection.py create mode 100644 hub_core/runtime/workload_projection_routes.py create mode 100644 tests/test_workload_projection.py create mode 100644 tests/test_workload_projection_contract.py diff --git a/INTENT.md b/INTENT.md index 55227de..9014ced 100644 --- a/INTENT.md +++ b/INTENT.md @@ -94,8 +94,8 @@ Current package surface (`hub_core/`): | `schemas/` | Pydantic contracts matching core models plus DoI report shapes | | `routers/` | Factory functions: domains, repos, messages, progress, capabilities, TPSC, policy | | `mcp/` | `HubCoreMCPServer` — generic orientation, messaging, capability, repo, DoI, TPSC/GDPR, risk/alert, progress tools | -| `contracts/` | Packaged `helixforge.hub-extension` schemas, named-port OpenAPI, catalogs, fixtures, compatibility | -| `migrations/` | Alembic scaffold and `0001_core_schema` for adopters | +| `contracts/` | Packaged hub-extension, repository-navigation, and workload-projection schemas, OpenAPI, fixtures, compatibility | +| `migrations/` | Alembic core/runtime scaffold plus durable repository and workload projection migrations | | `utils/` | Slugs, pagination, repo path resolution, trailing-slash routing | | `events.py` | Canonical FOS §10 risk and alert event types | diff --git a/README.md b/README.md index 7647407..a371c95 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,12 @@ matrix, and read-only OpenAPI surface are located with `repository_navigation_contract_root()`; normative rebuild and cursor rules are in `docs/repository-navigation-contract.md`. +`helixforge.workload-projection` 1.0.0 transports Repo Manager's authoritative +workload index through a separate injected `port.repo` reader. The runtime +offers GET-only list and exact-reference resolution at +`/ports/projections/workloads`, plus matching MCP tools, without importing Repo +Manager internals or inferring workload identity. + ## Runtime scaffold Install the runtime extra and start the API, MCP, or migration process through diff --git a/SCOPE.md b/SCOPE.md index 073c48e..223c99e 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -1,6 +1,6 @@ # SCOPE — hub-core -**Updated:** 2026-08-22 +**Updated:** 2026-08-23 --- @@ -41,6 +41,9 @@ the Python package remains independently importable. - **Cross-domain navigation projections** — rebuildable aggregation of Repo Manager's validated repository classifications without classification write authority +- **Authoritative workload transport** — rebuildable normalized distribution + and exact resolution of Repo Manager's workload index without semantic or + declaration authority - **Documentation** — `README.md`, `INTENT.md`, `SCOPE.md`, pointer to extraction boundary in `the-custodian` @@ -76,6 +79,8 @@ After the CUST-WP-0025 first slice (2026-06-06 — 2026-06-07): | Adopt shared slug/pagination/path utilities | `hub_core.utils` | | Expose risk/alert progress read views | `/progress/risks`, `/progress/alerts` patterns | | State Hub incremental adoption | Schemas, messages, policy, TPSC, progress, domains, capability catalog routers imported | +| Query repository classification navigation | `port.projection.query` HTTP routes plus MCP facets | +| Transport and resolve workload references | GET-only workload projection routes plus MCP exact resolution | ```bash cd ~/hub-core @@ -89,8 +94,11 @@ python3 -m venv .venv && .venv/bin/pip install -e . - **Published PyPI package** — consumed via editable path or private index only - **Complete State Hub decoupling** — dev-hub routes and models still live in `state-hub` -- **Repository classification navigation** — accepted contract; ingestion, - query/MCP, and A5 cutover remain in `HUB-WP-0006` +- **Repository classification navigation cutover** — contract, ingestion, and + query/MCP are complete; A5 production handoff remains in `HUB-WP-0006-T06` +- **Workload projection production adoption** — transport/query capability is + complete; production hosts must inject the Repo Manager workload reader and + apply migration `0004_workload_projection` - **fin-hub adoption** — planned; not verified in this repo - **Capability registry entries** — scaffold only (`capabilities: []`); no registered reusable behaviors yet @@ -108,7 +116,7 @@ python3 -m venv .venv && .venv/bin/pip install -e . | Primary consumer | `state-hub` (editable dependency, router/schema import in progress) | | Extraction workplan | `CUST-WP-0025` (custodian domain) | | Target architecture | `SHR-ARCH-IA-0001` + `SHR-ARCH-HUB-0001` | -| Runtime workplans | `HUB-WP-0004` and `HUB-WP-0005` finished; `HUB-WP-0006` active | +| Runtime workplans | `HUB-WP-0004`, `HUB-WP-0005`, and `HUB-WP-0007` finished; `HUB-WP-0006-T06` waits on cutover gates | --- diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 599ccfe..a56a7d1 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -14,6 +14,7 @@ | workplan | HUB-WP-0004 | finished | — | workplans/HUB-WP-0004-runtime-and-extension-contract.md | | workplan | HUB-WP-0005 | finished | — | workplans/HUB-WP-0005-core-hub-absorption-execution.md | | workplan | HUB-WP-0006 | active | — | workplans/HUB-WP-0006-repository-classification-navigation.md | +| workplan | HUB-WP-0007 | finished | — | workplans/HUB-WP-0007-workload-projection-transport.md | | task | HUB-WP-0001-T01 | done | — | workplans/HUB-WP-0001-statehub-bootstrap.md | | task | HUB-WP-0001-T02 | done | — | workplans/HUB-WP-0001-statehub-bootstrap.md | | task | HUB-WP-0001-T03 | done | — | workplans/HUB-WP-0001-statehub-bootstrap.md | @@ -44,3 +45,8 @@ | task | HUB-WP-0006-T04 | done | — | workplans/HUB-WP-0006-repository-classification-navigation.md | | task | HUB-WP-0006-T05 | done | — | workplans/HUB-WP-0006-repository-classification-navigation.md | | task | HUB-WP-0006-T06 | wait | — | workplans/HUB-WP-0006-repository-classification-navigation.md | +| task | HUB-WP-0007-T01 | done | — | workplans/HUB-WP-0007-workload-projection-transport.md | +| task | HUB-WP-0007-T02 | done | — | workplans/HUB-WP-0007-workload-projection-transport.md | +| task | HUB-WP-0007-T03 | done | — | workplans/HUB-WP-0007-workload-projection-transport.md | +| task | HUB-WP-0007-T04 | done | — | workplans/HUB-WP-0007-workload-projection-transport.md | +| task | HUB-WP-0007-T05 | done | — | workplans/HUB-WP-0007-workload-projection-transport.md | diff --git a/docs/runtime.md b/docs/runtime.md index 7edc433..31a4ff6 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -34,6 +34,7 @@ the API does not auto-create tables. | `port.events.interaction` | `POST /ports/events/interaction` | Accepts only cataloged interaction-family events | | `port.projection.query` | `GET /ports/projections/{id}` | Rebuildable registry/message/event projections with provenance | | `port.projection.query` | `GET /ports/projections/repository-navigation/repositories` | Snapshot-bound cross-repository classification navigation | +| `port.projection.query` | `GET /ports/projections/workloads` | Normalized authoritative workload projection and exact resolution | Available projection ids are `hub_registry`, `messages`, `progress_events`, and `interaction_events`. The two event families use distinct stores and cannot be @@ -57,6 +58,14 @@ canonical content hash. MCP exposes the same surface through `query_repository_navigation` and `get_repository_navigation_facet`. No classification-write endpoint exists. +Workload projection uses a distinct injected `WorkloadProjectionClient` and +the authoritative `helixforge.workload-reference/v1` identity pair. Migration +`0004_workload_projection` stores only normalized transport rows and source +provenance. List and `/resolve` routes require explicit rapp/name/deployable +values and return unknown instead of consulting owners, repository names, +paths, or aliases. MCP exposes `query_workloads` and +`resolve_workload_reference` over those GET-only routes. + ## Backend boundary and readiness The app is created with an injected `PortStore`. `InMemoryPortStore` remains diff --git a/hub_core/contracts/__init__.py b/hub_core/contracts/__init__.py index 53f8ae3..4af85ae 100644 --- a/hub_core/contracts/__init__.py +++ b/hub_core/contracts/__init__.py @@ -7,6 +7,8 @@ CONTRACT_ID = "helixforge.hub-extension" CONTRACT_VERSION = "0.1.0" REPOSITORY_NAVIGATION_CONTRACT_ID = "helixforge.repository-navigation" REPOSITORY_NAVIGATION_CONTRACT_VERSION = "1.0.0" +WORKLOAD_PROJECTION_CONTRACT_ID = "helixforge.workload-projection" +WORKLOAD_PROJECTION_CONTRACT_VERSION = "1.0.0" def extension_contract_root() -> Traversable: @@ -21,11 +23,20 @@ def repository_navigation_contract_root() -> Traversable: return files("hub_core.contracts.repository_navigation.v1_0_0") +def workload_projection_contract_root() -> Traversable: + """Return the packaged root for the workload-projection contract.""" + + return files("hub_core.contracts.workload_projection.v1_0_0") + + __all__ = [ "CONTRACT_ID", "CONTRACT_VERSION", "REPOSITORY_NAVIGATION_CONTRACT_ID", "REPOSITORY_NAVIGATION_CONTRACT_VERSION", + "WORKLOAD_PROJECTION_CONTRACT_ID", + "WORKLOAD_PROJECTION_CONTRACT_VERSION", "extension_contract_root", "repository_navigation_contract_root", + "workload_projection_contract_root", ] diff --git a/hub_core/contracts/workload_projection/__init__.py b/hub_core/contracts/workload_projection/__init__.py new file mode 100644 index 0000000..43f4d93 --- /dev/null +++ b/hub_core/contracts/workload_projection/__init__.py @@ -0,0 +1 @@ +"""Versioned authoritative workload projection contracts.""" diff --git a/hub_core/contracts/workload_projection/v1_0_0/README.md b/hub_core/contracts/workload_projection/v1_0_0/README.md new file mode 100644 index 0000000..9065d9b --- /dev/null +++ b/hub_core/contracts/workload_projection/v1_0_0/README.md @@ -0,0 +1,13 @@ +# helixforge.workload-projection 1.0.0 + +This package freezes hub-core's transport boundary for Repo Manager's +`helixforge.workload-reference/v1` index. It includes strict input/output JSON +schemas, a nine-declaration fixture, the read-only query OpenAPI fragment, and +the compatibility/authority matrix. + +The exact identity key is `rapp_id` plus `workload_identity.name`; optional +deployable resolution checks declared membership. Hub-core transports and +queries normalized records with source repository, declaration path, source +Git revision, and canonical UTC observation metadata. It never infers identity +from repository names, owners, paths, or aliases and never imports Repo Manager +private persistence. diff --git a/hub_core/contracts/workload_projection/v1_0_0/__init__.py b/hub_core/contracts/workload_projection/v1_0_0/__init__.py new file mode 100644 index 0000000..cc46a48 --- /dev/null +++ b/hub_core/contracts/workload_projection/v1_0_0/__init__.py @@ -0,0 +1 @@ +"""Packaged workload projection contract 1.0.0.""" diff --git a/hub_core/contracts/workload_projection/v1_0_0/compatibility-matrix.json b/hub_core/contracts/workload_projection/v1_0_0/compatibility-matrix.json new file mode 100644 index 0000000..71c0c5d --- /dev/null +++ b/hub_core/contracts/workload_projection/v1_0_0/compatibility-matrix.json @@ -0,0 +1,25 @@ +{ + "contract_id": "helixforge.workload-projection", + "current_version": "1.0.0", + "accepted_inputs": [ + { + "contract_id": "helixforge.workload-projection-source", + "version": "1.0.0", + "source_contract": "helixforge.workload-reference/v1", + "status": "current", + "coercion": false + } + ], + "authority": { + "declarations": "rapp-*/declarations/rapp.yaml", + "semantic_schema": "railiance-master/schemas/rapp.schema.json", + "index_and_resolution": "repo-manager", + "transport_and_cross_repository_query": "hub-core" + }, + "forbidden_fallbacks": ["repository_name", "owner_repo", "path_segment", "alias"], + "change_policy": { + "patch": "Clarifications and compatible constraint corrections only.", + "minor": "Additive optional transport metadata and diagnostics.", + "major": "Changed identity, authority, resolution, rebuild, cursor, or deletion semantics; removed or renamed fields; or narrowed constraints." + } +} diff --git a/hub_core/contracts/workload_projection/v1_0_0/fixtures/repo-manager-nine-workloads.json b/hub_core/contracts/workload_projection/v1_0_0/fixtures/repo-manager-nine-workloads.json new file mode 100644 index 0000000..6a947a4 --- /dev/null +++ b/hub_core/contracts/workload_projection/v1_0_0/fixtures/repo-manager-nine-workloads.json @@ -0,0 +1,139 @@ +{ + "contract_id": "helixforge.workload-projection-source", + "contract_version": "1.0.0", + "source": { + "system": "repo-manager", + "workload_contract": "helixforge.workload-reference/v1", + "workload_contract_id": "RMGR-CONTRACT-WORKLOAD-REFERENCE-0001", + "producer_revision": "b36b68bc5747aaeea412079c737f9f7e9d4b7dfb" + }, + "snapshot": { + "snapshot_id": "9999999999999999999999999999999999999999999999999999999999999999", + "generated_at": "2026-08-22T12:00:00Z", + "source_revision": "b36b68bc5747aaeea412079c737f9f7e9d4b7dfb", + "page_cursor": null, + "next_cursor": null, + "final_page": true, + "total_workload_count": 9 + }, + "workloads": [ + { + "rapp_id": "rapp-issue-core", + "name": "issue-core", + "declaration_repo": "rapp-issue-core", + "declaration_path": "rapp-issue-core/declarations/rapp.yaml", + "source_git_revision": "a78f38cfd8afb29ed23d150a4ea3e3dcd1f35381", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "issue-core", + "readiness_state": "verified", + "data_classification": "confidential", + "criticality": "high", + "deployables": ["issue-core"] + }, + { + "rapp_id": "rapp-openbao", + "name": "openbao", + "declaration_repo": "rapp-openbao", + "declaration_path": "rapp-openbao/declarations/rapp.yaml", + "source_git_revision": "6c3012f6bd12d12c88df03de69b07895e8325347", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "railiance-platform", + "readiness_state": "verified", + "data_classification": "restricted", + "criticality": "critical", + "deployables": ["openbao", "openbao-ui-gateway"] + }, + { + "rapp_id": "rapp-policy-nexus", + "name": "policy-nexus", + "declaration_repo": "rapp-policy-nexus", + "declaration_path": "rapp-policy-nexus/declarations/rapp.yaml", + "source_git_revision": "41c42e8d1a653af666da2e5e7e008a27464d9b87", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "policy-nexus", + "readiness_state": "verified", + "data_classification": "public", + "criticality": "high", + "deployables": ["policy-nexus-static"] + }, + { + "rapp_id": "rapp-postgres", + "name": "postgres", + "declaration_repo": "rapp-postgres", + "declaration_path": "rapp-postgres/declarations/rapp.yaml", + "source_git_revision": "cfc252630c53e87d01eb38b91e1d5b692e72a4b2", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "railiance-platform", + "readiness_state": "verified", + "data_classification": "restricted", + "criticality": "critical", + "deployables": ["platform-pg"] + }, + { + "rapp_id": "rapp-qonto", + "name": "qonto", + "declaration_repo": "rapp-qonto", + "declaration_path": "rapp-qonto/declarations/rapp.yaml", + "source_git_revision": "f57e60ad2a41c18f55dd9ee4ec24a56a4e0d9ff2", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "qonto-assistant", + "readiness_state": "verified", + "data_classification": "restricted", + "criticality": "critical", + "deployables": ["rapp-qonto"] + }, + { + "rapp_id": "rapp-sbom-nexus", + "name": "sbom-nexus", + "declaration_repo": "rapp-sbom-nexus", + "declaration_path": "rapp-sbom-nexus/declarations/rapp.yaml", + "source_git_revision": "1566bcf69fc0757db5efc67f78b14af10d944c1b", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "sbom-nexus", + "readiness_state": "draft", + "data_classification": "confidential", + "criticality": "high", + "deployables": ["sbom-nexus"] + }, + { + "rapp_id": "rapp-secrets-engine", + "name": "secrets-engine", + "declaration_repo": "rapp-secrets-engine", + "declaration_path": "rapp-secrets-engine/declarations/rapp.yaml", + "source_git_revision": "f71131e76b70aad02d32ab46f2d0983f030304a9", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "secrets-engine", + "readiness_state": "retired", + "data_classification": "restricted", + "criticality": "high", + "deployables": ["secrets-engine-cli"] + }, + { + "rapp_id": "rapp-tenant-engine", + "name": "tenant-engine", + "declaration_repo": "rapp-tenant-engine", + "declaration_path": "rapp-tenant-engine/declarations/rapp.yaml", + "source_git_revision": "82bbbf78c4159aa5267af683efca75cce7a57c00", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "tenant-engine", + "readiness_state": "draft", + "data_classification": "confidential", + "criticality": "high", + "deployables": ["tenant-engine"] + }, + { + "rapp_id": "rapp-user-engine", + "name": "user-engine", + "declaration_repo": "rapp-user-engine", + "declaration_path": "rapp-user-engine/declarations/rapp.yaml", + "source_git_revision": "e599b4f345cdd4725a6d0ee16efd475cf98ea616", + "observed_at": "2026-08-22T12:00:00Z", + "ownership_repo": "user-engine", + "readiness_state": "verified", + "data_classification": "confidential", + "criticality": "high", + "deployables": ["user-engine", "user-engine-pg"] + } + ], + "diagnostics": [] +} diff --git a/hub_core/contracts/workload_projection/v1_0_0/openapi/workload-projection.openapi.json b/hub_core/contracts/workload_projection/v1_0_0/openapi/workload-projection.openapi.json new file mode 100644 index 0000000..053a74e --- /dev/null +++ b/hub_core/contracts/workload_projection/v1_0_0/openapi/workload-projection.openapi.json @@ -0,0 +1,43 @@ +{ + "openapi": "3.1.0", + "info": {"title": "HelixForge workload projection", "version": "1.0.0"}, + "paths": { + "/ports/projections/workloads": { + "get": { + "operationId": "queryWorkloads", + "x-port-id": "port.projection.query", + "x-direction": "out", + "parameters": [ + {"name": "rapp_id", "in": "query", "schema": {"type": "string"}}, + {"name": "name", "in": "query", "schema": {"type": "string"}}, + {"name": "deployable", "in": "query", "schema": {"type": "string"}}, + {"name": "cursor", "in": "query", "schema": {"type": "string"}}, + {"name": "limit", "in": "query", "schema": {"type": "integer", "minimum": 1, "maximum": 500, "default": 100}} + ], + "responses": { + "200": {"description": "Stable page of normalized workload records", "content": {"application/json": {"schema": {"$ref": "../schemas/workload-projection.schema.json"}}}}, + "400": {"description": "Invalid explicit filter"}, + "409": {"description": "Cursor belongs to another generation or filter"}, + "503": {"description": "No accepted workload projection is available"} + } + } + }, + "/ports/projections/workloads/resolve": { + "get": { + "operationId": "resolveWorkloadReference", + "x-port-id": "port.projection.query", + "x-direction": "out", + "parameters": [ + {"name": "rapp_id", "in": "query", "required": true, "schema": {"type": "string"}}, + {"name": "name", "in": "query", "required": true, "schema": {"type": "string"}}, + {"name": "deployable", "in": "query", "schema": {"type": "string"}} + ], + "responses": { + "200": {"description": "Resolved or explicitly unknown reference", "content": {"application/json": {"schema": {"$ref": "../schemas/workload-resolution.schema.json"}}}}, + "400": {"description": "Invalid explicit reference"}, + "503": {"description": "No accepted workload projection is available"} + } + } + } + } +} diff --git a/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-projection-page.schema.json b/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-projection-page.schema.json new file mode 100644 index 0000000..683d270 --- /dev/null +++ b/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-projection-page.schema.json @@ -0,0 +1,82 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://schemas.helixforge.local/workload-projection/1.0.0/workload-projection-page.schema.json", + "title": "Repo Manager workload projection page", + "type": "object", + "additionalProperties": false, + "required": ["contract_id", "contract_version", "source", "snapshot", "workloads", "diagnostics"], + "properties": { + "contract_id": {"const": "helixforge.workload-projection-source"}, + "contract_version": {"const": "1.0.0"}, + "source": {"$ref": "#/$defs/source"}, + "snapshot": {"$ref": "#/$defs/snapshot"}, + "workloads": {"type": "array", "items": {"$ref": "#/$defs/workload"}}, + "diagnostics": {"type": "array", "items": {"$ref": "#/$defs/diagnostic"}} + }, + "$defs": { + "source": { + "type": "object", + "additionalProperties": false, + "required": ["system", "workload_contract", "workload_contract_id", "producer_revision"], + "properties": { + "system": {"const": "repo-manager"}, + "workload_contract": {"const": "helixforge.workload-reference/v1"}, + "workload_contract_id": {"const": "RMGR-CONTRACT-WORKLOAD-REFERENCE-0001"}, + "producer_revision": {"type": "string", "pattern": "^[a-f0-9]{40,64}$"} + } + }, + "snapshot": { + "type": "object", + "additionalProperties": false, + "required": ["snapshot_id", "generated_at", "source_revision", "page_cursor", "next_cursor", "final_page", "total_workload_count"], + "properties": { + "snapshot_id": {"type": "string", "pattern": "^[a-f0-9]{64}$"}, + "generated_at": {"$ref": "#/$defs/utcTimestamp"}, + "source_revision": {"type": "string", "pattern": "^[a-f0-9]{40,64}$"}, + "page_cursor": {"type": ["string", "null"], "maxLength": 1000}, + "next_cursor": {"type": ["string", "null"], "maxLength": 1000}, + "final_page": {"type": "boolean"}, + "total_workload_count": {"type": "integer", "minimum": 0} + } + }, + "workload": { + "type": "object", + "additionalProperties": false, + "required": ["rapp_id", "name", "declaration_repo", "declaration_path", "source_git_revision", "observed_at", "ownership_repo", "readiness_state", "data_classification", "criticality", "deployables"], + "properties": { + "rapp_id": {"type": "string", "pattern": "^rapp-[a-z0-9]+(?:-[a-z0-9]+)*$", "maxLength": 120}, + "name": {"type": "string", "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$", "maxLength": 120}, + "declaration_repo": {"type": "string", "pattern": "^rapp-[a-z0-9]+(?:-[a-z0-9]+)*$", "maxLength": 120}, + "declaration_path": {"type": "string", "pattern": "^rapp-[a-z0-9]+(?:-[a-z0-9]+)*/declarations/rapp\\.yaml$", "maxLength": 260}, + "source_git_revision": {"type": "string", "pattern": "^[a-f0-9]{40,64}$"}, + "observed_at": {"$ref": "#/$defs/utcTimestamp"}, + "ownership_repo": {"type": ["string", "null"], "minLength": 1, "maxLength": 120}, + "readiness_state": {"type": ["string", "null"], "minLength": 1, "maxLength": 80}, + "data_classification": {"type": ["string", "null"], "minLength": 1, "maxLength": 80}, + "criticality": {"type": ["string", "null"], "minLength": 1, "maxLength": 80}, + "deployables": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": {"type": "string", "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$", "maxLength": 120} + } + } + }, + "diagnostic": { + "type": "object", + "additionalProperties": false, + "required": ["severity", "code", "message"], + "properties": { + "severity": {"enum": ["info", "warning", "error"]}, + "code": {"type": "string", "pattern": "^[a-z][a-z0-9_.-]*$", "maxLength": 120}, + "message": {"type": "string", "minLength": 1, "maxLength": 1000}, + "declaration_path": {"type": "string", "maxLength": 260} + } + }, + "utcTimestamp": { + "type": "string", + "format": "date-time", + "pattern": "Z$" + } + } +} diff --git a/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-projection.schema.json b/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-projection.schema.json new file mode 100644 index 0000000..9d94e63 --- /dev/null +++ b/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-projection.schema.json @@ -0,0 +1,66 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://schemas.helixforge.local/workload-projection/1.0.0/workload-projection.schema.json", + "title": "Hub workload projection", + "type": "object", + "additionalProperties": false, + "required": ["contract_id", "contract_version", "projection_id", "projection_status", "source_snapshot", "source_checked_at", "rebuilt_at", "content_hash", "workloads", "diagnostics", "next_cursor", "total_workload_count"], + "properties": { + "contract_id": {"const": "helixforge.workload-projection"}, + "contract_version": {"const": "1.0.0"}, + "projection_id": {"const": "workloads"}, + "projection_status": {"enum": ["current", "stale"]}, + "source_snapshot": {"$ref": "#/$defs/sourceSnapshot"}, + "source_checked_at": {"$ref": "#/$defs/utcTimestamp"}, + "rebuilt_at": {"$ref": "#/$defs/utcTimestamp"}, + "content_hash": {"type": "string", "pattern": "^[a-f0-9]{64}$"}, + "workloads": {"type": "array", "items": {"$ref": "#/$defs/workload"}}, + "diagnostics": {"type": "array", "items": {"$ref": "#/$defs/diagnostic"}}, + "next_cursor": {"type": ["string", "null"], "maxLength": 1000}, + "total_workload_count": {"type": "integer", "minimum": 0} + }, + "$defs": { + "sourceSnapshot": { + "type": "object", + "additionalProperties": false, + "required": ["snapshot_id", "source_system", "source_contract", "source_revision", "generated_at"], + "properties": { + "snapshot_id": {"type": "string", "pattern": "^[a-f0-9]{64}$"}, + "source_system": {"const": "repo-manager"}, + "source_contract": {"const": "helixforge.workload-reference/v1"}, + "source_revision": {"type": "string", "pattern": "^[a-f0-9]{40,64}$"}, + "generated_at": {"$ref": "#/$defs/utcTimestamp"} + } + }, + "workload": { + "type": "object", + "additionalProperties": false, + "required": ["rapp_id", "name", "declaration_repo", "declaration_path", "source_git_revision", "observed_at", "ownership_repo", "readiness_state", "data_classification", "criticality", "deployables"], + "properties": { + "rapp_id": {"type": "string", "pattern": "^rapp-[a-z0-9]+(?:-[a-z0-9]+)*$", "maxLength": 120}, + "name": {"type": "string", "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$", "maxLength": 120}, + "declaration_repo": {"type": "string", "pattern": "^rapp-[a-z0-9]+(?:-[a-z0-9]+)*$", "maxLength": 120}, + "declaration_path": {"type": "string", "pattern": "^rapp-[a-z0-9]+(?:-[a-z0-9]+)*/declarations/rapp\\.yaml$", "maxLength": 260}, + "source_git_revision": {"type": "string", "pattern": "^[a-f0-9]{40,64}$"}, + "observed_at": {"$ref": "#/$defs/utcTimestamp"}, + "ownership_repo": {"type": ["string", "null"], "minLength": 1, "maxLength": 120}, + "readiness_state": {"type": ["string", "null"], "minLength": 1, "maxLength": 80}, + "data_classification": {"type": ["string", "null"], "minLength": 1, "maxLength": 80}, + "criticality": {"type": ["string", "null"], "minLength": 1, "maxLength": 80}, + "deployables": {"type": "array", "minItems": 1, "uniqueItems": true, "items": {"type": "string", "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$", "maxLength": 120}} + } + }, + "diagnostic": { + "type": "object", + "additionalProperties": false, + "required": ["severity", "code", "message"], + "properties": { + "severity": {"enum": ["info", "warning", "error"]}, + "code": {"type": "string", "pattern": "^[a-z][a-z0-9_.-]*$", "maxLength": 120}, + "message": {"type": "string", "minLength": 1, "maxLength": 1000}, + "declaration_path": {"type": "string", "maxLength": 260} + } + }, + "utcTimestamp": {"type": "string", "format": "date-time", "pattern": "Z$"} + } +} diff --git a/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-resolution.schema.json b/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-resolution.schema.json new file mode 100644 index 0000000..3b1716e --- /dev/null +++ b/hub_core/contracts/workload_projection/v1_0_0/schemas/workload-resolution.schema.json @@ -0,0 +1,65 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://schemas.helixforge.local/workload-projection/1.0.0/workload-resolution.schema.json", + "title": "Exact workload reference resolution", + "type": "object", + "additionalProperties": false, + "required": ["contract_id", "contract_version", "status", "reference", "source_snapshot", "content_hash", "projection_status"], + "properties": { + "contract_id": {"const": "helixforge.workload-projection"}, + "contract_version": {"const": "1.0.0"}, + "status": {"enum": ["resolved", "unknown"]}, + "reference": {"$ref": "#/$defs/reference"}, + "source_snapshot": {"$ref": "#/$defs/sourceSnapshot"}, + "content_hash": {"type": "string", "pattern": "^[a-f0-9]{64}$"}, + "projection_status": {"enum": ["current", "stale"]}, + "reason": {"enum": ["not_found", "deployable_not_declared"]}, + "workload": {"$ref": "#/$defs/workload"} + }, + "oneOf": [ + {"properties": {"status": {"const": "resolved"}}, "required": ["workload"], "not": {"required": ["reason"]}}, + {"properties": {"status": {"const": "unknown"}}, "required": ["reason"], "not": {"required": ["workload"]}} + ], + "$defs": { + "reference": { + "type": "object", + "additionalProperties": false, + "required": ["rapp_id", "name"], + "properties": { + "rapp_id": {"type": "string", "pattern": "^rapp-[a-z0-9]+(?:-[a-z0-9]+)*$"}, + "name": {"type": "string", "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$"}, + "deployable": {"type": "string", "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$"} + } + }, + "sourceSnapshot": { + "type": "object", + "additionalProperties": false, + "required": ["snapshot_id", "source_system", "source_contract", "source_revision", "generated_at"], + "properties": { + "snapshot_id": {"type": "string", "pattern": "^[a-f0-9]{64}$"}, + "source_system": {"const": "repo-manager"}, + "source_contract": {"const": "helixforge.workload-reference/v1"}, + "source_revision": {"type": "string", "pattern": "^[a-f0-9]{40,64}$"}, + "generated_at": {"type": "string", "format": "date-time", "pattern": "Z$"} + } + }, + "workload": { + "type": "object", + "additionalProperties": false, + "required": ["rapp_id", "name", "declaration_repo", "declaration_path", "source_git_revision", "observed_at", "ownership_repo", "readiness_state", "data_classification", "criticality", "deployables"], + "properties": { + "rapp_id": {"type": "string"}, + "name": {"type": "string"}, + "declaration_repo": {"type": "string"}, + "declaration_path": {"type": "string"}, + "source_git_revision": {"type": "string", "pattern": "^[a-f0-9]{40,64}$"}, + "observed_at": {"type": "string", "format": "date-time", "pattern": "Z$"}, + "ownership_repo": {"type": ["string", "null"]}, + "readiness_state": {"type": ["string", "null"]}, + "data_classification": {"type": ["string", "null"]}, + "criticality": {"type": ["string", "null"]}, + "deployables": {"type": "array", "minItems": 1, "uniqueItems": true, "items": {"type": "string"}} + } + } + } +} diff --git a/hub_core/mcp/server.py b/hub_core/mcp/server.py index 9826be7..4e3ba10 100644 --- a/hub_core/mcp/server.py +++ b/hub_core/mcp/server.py @@ -29,6 +29,8 @@ CORE_TOOL_NAMES = frozenset({ "list_domain_repos", "query_repository_navigation", "get_repository_navigation_facet", + "query_workloads", + "resolve_workload_reference", "check_repo_doi", "get_doi_summary", "register_service", @@ -344,6 +346,42 @@ class HubCoreMCPServer: ) ) + @register("query_workloads") + def query_workloads( + rapp_id: str | None = None, + name: str | None = None, + deployable: str | None = None, + cursor: str | None = None, + limit: int = 100, + ) -> str: + """Query normalized authoritative workload records without inference.""" + return self._json( + self._get( + "/ports/projections/workloads", + { + "rapp_id": rapp_id, + "name": name, + "deployable": deployable, + "cursor": cursor, + "limit": limit, + }, + ) + ) + + @register("resolve_workload_reference") + def resolve_workload_reference( + rapp_id: str, + name: str, + deployable: str | None = None, + ) -> str: + """Resolve only an exact rapp/name/deployable reference.""" + return self._json( + self._get( + "/ports/projections/workloads/resolve", + {"rapp_id": rapp_id, "name": name, "deployable": deployable}, + ) + ) + @register("check_repo_doi") def check_repo_doi(repo_slug: str, force_refresh: bool = False) -> str: return self._json( diff --git a/hub_core/migrations/versions/0004_workload_projection.py b/hub_core/migrations/versions/0004_workload_projection.py new file mode 100644 index 0000000..b81ca00 --- /dev/null +++ b/hub_core/migrations/versions/0004_workload_projection.py @@ -0,0 +1,55 @@ +"""durable authoritative workload projection + +Revision ID: 0004_workload_projection +Revises: 0003_repository_navigation +Create Date: 2026-08-23 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0004_workload_projection" +down_revision: Union[str, None] = "0003_repository_navigation" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "runtime_workload_projection_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("workload_count", sa.Integer(), nullable=False), + sa.Column("diagnostics", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + ) + op.create_table( + "runtime_workload_projection_records", + sa.Column("rapp_id", sa.String(120), primary_key=True), + sa.Column("name", sa.String(120), primary_key=True), + sa.Column("declaration_repo", sa.String(120), nullable=False), + sa.Column("declaration_path", sa.String(260), nullable=False), + sa.Column("source_git_revision", sa.String(64), nullable=False), + sa.Column("observed_at", sa.String(40), nullable=False), + sa.Column("ownership_repo", sa.String(120), nullable=True), + sa.Column("readiness_state", sa.String(80), nullable=True), + sa.Column("data_classification", sa.String(80), nullable=True), + sa.Column("criticality", sa.String(80), nullable=True), + sa.Column("deployables", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + ) + op.create_index( + "ix_runtime_workload_projection_records_declaration_repo", + "runtime_workload_projection_records", + ["declaration_repo"], + ) + + +def downgrade() -> None: + op.drop_table("runtime_workload_projection_records") + op.drop_table("runtime_workload_projection_state") diff --git a/hub_core/runtime/app.py b/hub_core/runtime/app.py index f592134..91325a7 100644 --- a/hub_core/runtime/app.py +++ b/hub_core/runtime/app.py @@ -19,6 +19,12 @@ from hub_core.runtime.repository_navigation_routes import ( ) from hub_core.runtime.store import InMemoryPortStore, PortStore from hub_core.runtime.validation import ContractValidator +from hub_core.runtime.workload_projection import ( + WorkloadProjectionClient, + WorkloadProjectionRejected, + WorkloadProjectionService, +) +from hub_core.runtime.workload_projection_routes import create_workload_projection_router def create_app( @@ -26,6 +32,7 @@ def create_app( settings: RuntimeSettings | None = None, port_store: PortStore | None = None, repo_projection_client: RepoProjectionClient | None = None, + workload_projection_client: WorkloadProjectionClient | None = None, ) -> FastAPI: resolved_settings = settings or RuntimeSettings.from_env() resolved_store = port_store or _create_store(resolved_settings) @@ -34,6 +41,10 @@ def create_app( client=repo_projection_client, store=resolved_store, ) + workload_projection = WorkloadProjectionService( + client=workload_projection_client, + store=resolved_store, + ) @asynccontextmanager async def lifespan(_: FastAPI): @@ -44,6 +55,11 @@ def create_app( # The readiness dependency reports the rejected or absent # projection while the API remains available for diagnosis. pass + if workload_projection_client is not None: + try: + await workload_projection.refresh() + except WorkloadProjectionRejected: + pass yield if owns_store and (closer := getattr(resolved_store, "aclose", None)): await closer() @@ -63,6 +79,7 @@ def create_app( ) app.state.contract_validator = ContractValidator() app.state.repository_navigation = repository_navigation + app.state.workload_projection = workload_projection @app.get("/healthz", response_model=HealthResponse, tags=["system"]) async def healthz() -> HealthResponse: @@ -76,6 +93,7 @@ def create_app( dependency_checks = { **await resolved_store.readiness_checks(), **await repository_navigation.readiness_checks(), + **await workload_projection.readiness_checks(), } ready = resolved_settings.is_ready(resolved_store.backend_name) and all( value in {"ok", "not_applicable"} for value in dependency_checks.values() @@ -90,6 +108,9 @@ def create_app( }, ) + # Exact projection routes precede the generic /projections/{projection_id} + # route so Starlette dispatch cannot shadow them. + app.include_router(create_workload_projection_router()) app.include_router(create_ports_router()) app.include_router(create_repository_navigation_router()) app.include_router(create_compatibility_router()) diff --git a/hub_core/runtime/postgres_store.py b/hub_core/runtime/postgres_store.py index 97f7d26..28b9048 100644 --- a/hub_core/runtime/postgres_store.py +++ b/hub_core/runtime/postgres_store.py @@ -22,6 +22,7 @@ from hub_core.runtime.models import ( RegistryRegistration, ) from hub_core.runtime.repository_navigation import NavigationProjection +from hub_core.runtime.workload_projection import WorkloadProjection from hub_core.runtime.tables import ( compat_api_keys, compat_hubs, @@ -33,6 +34,8 @@ from hub_core.runtime.tables import ( runtime_repository_navigation_repositories, runtime_repository_navigation_state, runtime_registrations, + runtime_workload_projection_records, + runtime_workload_projection_state, ) @@ -61,6 +64,7 @@ class PostgresPortStore: compat_hubs, compat_api_keys, runtime_repository_navigation_state, + runtime_workload_projection_state, ): await connection.execute( sa.select(sa.literal(1)).select_from(table).limit(1) @@ -342,6 +346,103 @@ class PostgresPortStore: ) ) + async def get_workload_projection(self) -> WorkloadProjection | None: + async with self.sessions() as session: + state = ( + await session.execute( + sa.select(runtime_workload_projection_state).where( + runtime_workload_projection_state.c.projection_id == "workloads" + ) + ) + ).mappings().one_or_none() + if state is None: + return None + rows = ( + await session.execute( + sa.select(runtime_workload_projection_records).order_by( + runtime_workload_projection_records.c.rapp_id, + runtime_workload_projection_records.c.name, + ) + ) + ).mappings() + workloads = tuple( + { + "rapp_id": row["rapp_id"], + "name": row["name"], + "declaration_repo": row["declaration_repo"], + "declaration_path": row["declaration_path"], + "source_git_revision": row["source_git_revision"], + "observed_at": row["observed_at"], + "ownership_repo": row["ownership_repo"], + "readiness_state": row["readiness_state"], + "data_classification": row["data_classification"], + "criticality": row["criticality"], + "deployables": list(row["deployables"] or []), + } + for row in rows + ) + return WorkloadProjection( + 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"], + workloads=workloads, + diagnostics=tuple(state["diagnostics"] or []), + ) + + async def replace_workload_projection(self, projection: WorkloadProjection) -> None: + 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, + "workload_count": len(projection.workloads), + "diagnostics": list(projection.diagnostics), + } + async with self.sessions.begin() as session: + await session.execute(runtime_workload_projection_records.delete()) + if projection.workloads: + await session.execute( + runtime_workload_projection_records.insert(), + [dict(workload) for workload in projection.workloads], + ) + exists = ( + await session.execute( + sa.select(runtime_workload_projection_state.c.projection_id).where( + runtime_workload_projection_state.c.projection_id == "workloads" + ) + ) + ).scalar_one_or_none() + if exists is None: + await session.execute( + runtime_workload_projection_state.insert().values( + projection_id="workloads", **values + ) + ) + else: + await session.execute( + runtime_workload_projection_state.update() + .where(runtime_workload_projection_state.c.projection_id == "workloads") + .values(**values) + ) + + async def mark_workload_projection_stale( + self, *, checked_at: datetime, diagnostic: Mapping[str, Any] + ) -> None: + async with self.sessions.begin() as session: + await session.execute( + runtime_workload_projection_state.update() + .where(runtime_workload_projection_state.c.projection_id == "workloads") + .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/store.py b/hub_core/runtime/store.py index 8e7c682..d3b6a73 100644 --- a/hub_core/runtime/store.py +++ b/hub_core/runtime/store.py @@ -20,6 +20,7 @@ from hub_core.runtime.models import ( RegistryRegistration, ) from hub_core.runtime.repository_navigation import NavigationProjection +from hub_core.runtime.workload_projection import WorkloadProjection class PortStore(Protocol): @@ -55,6 +56,14 @@ class PortStore(Protocol): self, *, checked_at: datetime, diagnostic: Mapping[str, Any] ) -> None: ... + async def get_workload_projection(self) -> WorkloadProjection | None: ... + + async def replace_workload_projection(self, projection: WorkloadProjection) -> None: ... + + async def mark_workload_projection_stale( + self, *, checked_at: datetime, diagnostic: Mapping[str, Any] + ) -> None: ... + class InMemoryPortStore: """Deterministic ephemeral backend for local runtime and conformance tests. @@ -72,6 +81,7 @@ class InMemoryPortStore: self._progress_events: list[dict[str, Any]] = [] self._interaction_events: list[dict[str, Any]] = [] self._repository_navigation: NavigationProjection | None = None + self._workload_projection: WorkloadProjection | None = None async def readiness_checks(self) -> dict[str, str]: return {"database": "not_applicable"} @@ -174,6 +184,31 @@ class InMemoryPortStore: diagnostics=(deepcopy(dict(diagnostic)),), ) + async def get_workload_projection(self) -> WorkloadProjection | None: + async with self._lock: + return deepcopy(self._workload_projection) + + async def replace_workload_projection(self, projection: WorkloadProjection) -> None: + async with self._lock: + self._workload_projection = deepcopy(projection) + + async def mark_workload_projection_stale( + self, *, checked_at: datetime, diagnostic: Mapping[str, Any] + ) -> None: + async with self._lock: + current = self._workload_projection + if current is None: + return + self._workload_projection = WorkloadProjection( + projection_status="stale", + source_snapshot=deepcopy(current.source_snapshot), + source_checked_at=checked_at, + rebuilt_at=current.rebuilt_at, + content_hash=current.content_hash, + workloads=deepcopy(current.workloads), + 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 0c27fe3..bbd78da 100644 --- a/hub_core/runtime/tables.py +++ b/hub_core/runtime/tables.py @@ -98,6 +98,36 @@ runtime_repository_navigation_facets = sa.Table( sa.Column("repository_ids", sa.JSON(), nullable=False), ) +runtime_workload_projection_state = sa.Table( + "runtime_workload_projection_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("workload_count", sa.Integer(), nullable=False), + sa.Column("diagnostics", sa.JSON(), nullable=False), +) + +runtime_workload_projection_records = sa.Table( + "runtime_workload_projection_records", + runtime_metadata, + sa.Column("rapp_id", sa.String(120), primary_key=True), + sa.Column("name", sa.String(120), primary_key=True), + sa.Column("declaration_repo", sa.String(120), nullable=False, index=True), + sa.Column("declaration_path", sa.String(260), nullable=False), + sa.Column("source_git_revision", sa.String(64), nullable=False), + sa.Column("observed_at", sa.String(40), nullable=False), + sa.Column("ownership_repo", sa.String(120), nullable=True), + sa.Column("readiness_state", sa.String(80), nullable=True), + sa.Column("data_classification", sa.String(80), nullable=True), + sa.Column("criticality", sa.String(80), nullable=True), + sa.Column("deployables", sa.JSON(), nullable=False), +) + compat_hubs = sa.Table( "compat_hubs", runtime_metadata, diff --git a/hub_core/runtime/workload_projection.py b/hub_core/runtime/workload_projection.py new file mode 100644 index 0000000..a3dd71d --- /dev/null +++ b/hub_core/runtime/workload_projection.py @@ -0,0 +1,421 @@ +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import json +import re +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 ( + WORKLOAD_PROJECTION_CONTRACT_ID, + WORKLOAD_PROJECTION_CONTRACT_VERSION, + workload_projection_contract_root, +) + + +class WorkloadProjectionClient(Protocol): + """Injected `port.repo` workload reader with transport owned by the host.""" + + async def fetch_workload_page(self, cursor: str | None) -> Mapping[str, Any]: ... + + +class WorkloadProjectionStore(Protocol): + async def get_workload_projection(self) -> WorkloadProjection | None: ... + + async def replace_workload_projection(self, projection: WorkloadProjection) -> None: ... + + async def mark_workload_projection_stale( + self, *, checked_at: datetime, diagnostic: Mapping[str, Any] + ) -> None: ... + + +class WorkloadProjectionRejected(ValueError): + """The authoritative workload transfer cannot safely replace the projection.""" + + +class WorkloadCursorMismatch(ValueError): + """A cursor does not belong to the active workload generation and filters.""" + + +@dataclass(frozen=True, slots=True) +class WorkloadProjection: + projection_status: Literal["current", "stale"] + source_snapshot: dict[str, Any] + source_checked_at: datetime + rebuilt_at: datetime + content_hash: str + workloads: tuple[dict[str, Any], ...] + diagnostics: tuple[dict[str, Any], ...] = () + + def to_contract(self) -> dict[str, Any]: + return { + "contract_id": WORKLOAD_PROJECTION_CONTRACT_ID, + "contract_version": WORKLOAD_PROJECTION_CONTRACT_VERSION, + "projection_id": "workloads", + "projection_status": self.projection_status, + "source_snapshot": deepcopy(self.source_snapshot), + "source_checked_at": _utc(self.source_checked_at), + "rebuilt_at": _utc(self.rebuilt_at), + "content_hash": self.content_hash, + "workloads": deepcopy(list(self.workloads)), + "diagnostics": deepcopy(list(self.diagnostics)), + "next_cursor": None, + "total_workload_count": len(self.workloads), + } + + +@dataclass(frozen=True, slots=True) +class WorkloadRefreshResult: + status: Literal["accepted", "duplicate"] + snapshot_id: str + content_hash: str + workload_count: int + + +class WorkloadProjectionService: + """Materialize and resolve normalized Repo Manager workload records.""" + + def __init__( + self, + *, + client: WorkloadProjectionClient | None, + store: WorkloadProjectionStore, + ) -> None: + self.client = client + self.store = store + self._validator = _input_validator() + self._lock = asyncio.Lock() + + async def readiness_checks(self) -> dict[str, str]: + if self.client is None: + return {"workload_projection": "not_applicable"} + projection = await self.store.get_workload_projection() + if projection is None: + return {"workload_projection": "unavailable"} + return { + "workload_projection": ( + "ok" if projection.projection_status == "current" else "stale" + ) + } + + async def refresh(self) -> WorkloadRefreshResult: + if self.client is None: + raise RuntimeError("no port.repo workload projection client is configured") + async with self._lock: + checked_at = _now() + try: + pages = await self._fetch_transfer() + return await self._apply_transfer(pages, checked_at=checked_at) + except asyncio.CancelledError: + raise + except Exception as exc: + diagnostic = { + "severity": "error", + "code": "workload_projection.rejected", + "message": str(exc)[:1000] or type(exc).__name__, + } + await self.store.mark_workload_projection_stale( + checked_at=checked_at, diagnostic=diagnostic + ) + if isinstance(exc, WorkloadProjectionRejected): + raise + raise WorkloadProjectionRejected(diagnostic["message"]) from exc + + async def query( + self, + *, + rapp_id: str | None = None, + name: str | None = None, + deployable: str | None = None, + cursor: str | None = None, + limit: int = 100, + ) -> dict[str, Any] | None: + if not 1 <= limit <= 500: + raise ValueError("limit must be between 1 and 500") + filters = _filters(rapp_id=rapp_id, name=name, deployable=deployable) + projection = await self.store.get_workload_projection() + if projection is None: + return None + filter_hash = _hash(filters) + generation_id = _hash( + [projection.source_snapshot["snapshot_id"], projection.content_hash] + ) + offset = ( + _decode_cursor(cursor, generation_id, filter_hash) + if cursor + else 0 + ) + matches = tuple( + record for record in projection.workloads if _matches(record, filters) + ) + if offset > len(matches): + raise WorkloadCursorMismatch("cursor offset exceeds the result set") + page = matches[offset : offset + limit] + next_offset = offset + len(page) + result = projection.to_contract() + result["workloads"] = deepcopy(list(page)) + result["total_workload_count"] = len(matches) + result["next_cursor"] = ( + _encode_cursor(next_offset, generation_id, filter_hash) + if next_offset < len(matches) + else None + ) + return result + + async def resolve( + self, *, rapp_id: str, name: str, deployable: str | None = None + ) -> dict[str, Any] | None: + filters = _filters(rapp_id=rapp_id, name=name, deployable=deployable) + projection = await self.store.get_workload_projection() + if projection is None: + return None + reference = {"rapp_id": rapp_id, "name": name} + if deployable is not None: + reference["deployable"] = deployable + exact = [ + record + for record in projection.workloads + if record["rapp_id"] == rapp_id and record["name"] == name + ] + metadata = { + "contract_id": WORKLOAD_PROJECTION_CONTRACT_ID, + "contract_version": WORKLOAD_PROJECTION_CONTRACT_VERSION, + "reference": reference, + "source_snapshot": deepcopy(projection.source_snapshot), + "content_hash": projection.content_hash, + "projection_status": projection.projection_status, + } + if len(exact) != 1: + return {**metadata, "status": "unknown", "reason": "not_found"} + workload = exact[0] + if deployable is not None and deployable not in workload["deployables"]: + return { + **metadata, + "status": "unknown", + "reason": "deployable_not_declared", + } + assert _matches(workload, filters) + return {**metadata, "status": "resolved", "workload": deepcopy(workload)} + + 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: set[str] = set() + while True: + raw = dict(await self.client.fetch_workload_page(cursor)) + try: + self._validator.validate(raw) + except ValidationError as exc: + location = ".".join(str(part) for part in exc.absolute_path) or "root" + raise WorkloadProjectionRejected( + f"invalid workload 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 WorkloadProjectionRejected( + "final_page must be true exactly when next_cursor is null" + ) + if snapshot["final_page"]: + return pages + if next_cursor in seen: + raise WorkloadProjectionRejected("workload projection cursor cycle detected") + seen.add(next_cursor) + cursor = next_cursor + if len(pages) >= 10000: + raise WorkloadProjectionRejected("workload transfer exceeds 10000 pages") + + async def _apply_transfer( + self, pages: list[dict[str, Any]], *, checked_at: datetime + ) -> WorkloadRefreshResult: + workloads = _validate_and_normalize(pages) + first = pages[0] + snapshot = first["snapshot"] + current = await self.store.get_workload_projection() + duplicate = bool( + current and current.source_snapshot["snapshot_id"] == snapshot["snapshot_id"] + ) + if ( + current + and not duplicate + and _parse_time(snapshot["generated_at"]) + <= _parse_time(current.source_snapshot["generated_at"]) + ): + raise WorkloadProjectionRejected( + "workload snapshot is older than the active generation" + ) + content_hash = _hash(list(workloads)) + if duplicate and current and content_hash != current.content_hash: + raise WorkloadProjectionRejected( + "snapshot_id was reused with different workload content" + ) + projection = WorkloadProjection( + projection_status="current", + source_snapshot={ + "snapshot_id": snapshot["snapshot_id"], + "source_system": "repo-manager", + "source_contract": first["source"]["workload_contract"], + "source_revision": snapshot["source_revision"], + "generated_at": snapshot["generated_at"], + }, + source_checked_at=checked_at, + rebuilt_at=_now(), + content_hash=content_hash, + workloads=workloads, + ) + await self.store.replace_workload_projection(projection) + return WorkloadRefreshResult( + status="duplicate" if duplicate else "accepted", + snapshot_id=snapshot["snapshot_id"], + content_hash=content_hash, + workload_count=len(workloads), + ) + + +def _input_validator() -> Draft202012Validator: + resource = workload_projection_contract_root().joinpath( + "schemas", "workload-projection-page.schema.json" + ) + schema = json.loads(resource.read_text(encoding="utf-8")) + Draft202012Validator.check_schema(schema) + return Draft202012Validator(schema, format_checker=FormatChecker()) + + +def _validate_and_normalize( + pages: list[dict[str, Any]], +) -> tuple[dict[str, Any], ...]: + if not pages: + raise WorkloadProjectionRejected("workload transfer contains no pages") + first = pages[0] + keys = ( + "snapshot_id", + "generated_at", + "source_revision", + "total_workload_count", + ) + expected = {key: first["snapshot"][key] for key in keys} + source = first["source"] + previous_cursor: str | None = None + records: list[dict[str, Any]] = [] + identities: set[tuple[str, str]] = set() + deployables: set[str] = set() + for index, page in enumerate(pages): + snapshot = page["snapshot"] + if any(snapshot[key] != value for key, value in expected.items()): + raise WorkloadProjectionRejected("workload snapshot metadata changed between pages") + if page["source"] != source: + raise WorkloadProjectionRejected("workload source metadata changed between pages") + if snapshot["page_cursor"] != previous_cursor: + raise WorkloadProjectionRejected(f"page {index} does not match requested cursor") + if any(item["severity"] == "error" for item in page["diagnostics"]): + raise WorkloadProjectionRejected("workload transfer contains an error diagnostic") + previous_cursor = snapshot["next_cursor"] + for raw in page["workloads"]: + record = deepcopy(raw) + identity = (record["rapp_id"], record["name"]) + if identity in identities: + raise WorkloadProjectionRejected(f"duplicate workload reference {identity!r}") + if record["declaration_repo"] != record["rapp_id"]: + raise WorkloadProjectionRejected( + f"declaration repository does not match {record['rapp_id']}" + ) + expected_path = f"{record['declaration_repo']}/declarations/rapp.yaml" + if record["declaration_path"] != expected_path: + raise WorkloadProjectionRejected( + f"declaration path is not canonical for {record['rapp_id']}" + ) + overlap = deployables.intersection(record["deployables"]) + if overlap: + raise WorkloadProjectionRejected( + f"deployable belongs to more than one workload: {sorted(overlap)[0]}" + ) + identities.add(identity) + deployables.update(record["deployables"]) + record["deployables"] = sorted(record["deployables"]) + records.append(record) + normalized = tuple(sorted(records, key=lambda item: (item["rapp_id"], item["name"]))) + if len(normalized) != expected["total_workload_count"]: + raise WorkloadProjectionRejected("total_workload_count does not match records") + if list(normalized) != records: + raise WorkloadProjectionRejected("workloads are not ordered by exact reference") + return normalized + + +def _filters( + *, rapp_id: str | None, name: str | None, deployable: str | None +) -> dict[str, str]: + values = {"rapp_id": rapp_id, "name": name, "deployable": deployable} + for key, value in values.items(): + if value is None: + continue + pattern = r"rapp-[a-z0-9]+(?:-[a-z0-9]+)*" if key == "rapp_id" else r"[a-z0-9]+(?:-[a-z0-9]+)*" + if len(value) > 120 or re.fullmatch(pattern, value) is None: + raise ValueError(f"invalid {key}") + return {key: value for key, value in values.items() if value is not None} + + +def _matches(record: Mapping[str, Any], filters: Mapping[str, str]) -> bool: + return all( + (value in record["deployables"] if key == "deployable" else record[key] == value) + for key, value in filters.items() + ) + + +def _encode_cursor(offset: int, generation_id: str, filter_hash: str) -> str: + payload = json.dumps( + {"generation_id": generation_id, "filter_hash": filter_hash, "offset": offset}, + sort_keys=True, + separators=(",", ":"), + ).encode() + encoded = base64.urlsafe_b64encode(payload).decode().rstrip("=") + checksum = hashlib.sha256(b"workload-projection/1.0.0:" + payload).hexdigest() + return f"{encoded}.{checksum}" + + +def _decode_cursor(cursor: str, generation_id: str, filter_hash: str) -> int: + try: + encoded, checksum = cursor.split(".", 1) + payload = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) + expected = hashlib.sha256(b"workload-projection/1.0.0:" + payload).hexdigest() + value = json.loads(payload) + if checksum != expected: + raise ValueError("checksum") + if value["generation_id"] != generation_id or value["filter_hash"] != filter_hash: + raise ValueError("generation or filters") + offset = int(value["offset"]) + if offset < 0: + raise ValueError("offset") + return offset + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise WorkloadCursorMismatch( + "cursor_snapshot_mismatch: restart from the first page" + ) from exc + + +def _hash(value: Any) -> str: + encoded = json.dumps(value, 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 _utc(value: datetime) -> str: + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + normalized = value.astimezone(timezone.utc) + return normalized.isoformat().replace("+00:00", "Z") + + +def _now() -> datetime: + return datetime.now(timezone.utc) diff --git a/hub_core/runtime/workload_projection_routes.py b/hub_core/runtime/workload_projection_routes.py new file mode 100644 index 0000000..fde69bd --- /dev/null +++ b/hub_core/runtime/workload_projection_routes.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Query, Request + +from hub_core.runtime.workload_projection import ( + WorkloadCursorMismatch, + WorkloadProjectionService, +) + + +def get_workload_projection_service(request: Request) -> WorkloadProjectionService: + return request.app.state.workload_projection + + +def create_workload_projection_router() -> APIRouter: + router = APIRouter(prefix="/ports/projections/workloads") + + @router.get( + "", + response_model=dict[str, Any], + tags=["workload-projection"], + openapi_extra={"x-port-id": "port.projection.query", "x-direction": "out"}, + ) + async def query_workloads( + rapp_id: str | None = None, + name: str | None = None, + deployable: str | None = None, + cursor: str | None = None, + limit: int = Query(default=100, ge=1, le=500), + service: WorkloadProjectionService = Depends(get_workload_projection_service), + ) -> dict[str, Any]: + try: + result = await service.query( + rapp_id=rapp_id, + name=name, + deployable=deployable, + cursor=cursor, + limit=limit, + ) + except WorkloadCursorMismatch as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return _available(result) + + @router.get( + "/resolve", + response_model=dict[str, Any], + tags=["workload-projection"], + openapi_extra={"x-port-id": "port.projection.query", "x-direction": "out"}, + ) + async def resolve_workload( + rapp_id: str, + name: str, + deployable: str | None = None, + service: WorkloadProjectionService = Depends(get_workload_projection_service), + ) -> dict[str, Any]: + try: + result = await service.resolve( + rapp_id=rapp_id, name=name, deployable=deployable + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return _available(result) + + return router + + +def _available(result: dict[str, Any] | None) -> dict[str, Any]: + if result is None: + raise HTTPException( + status_code=503, + detail="no accepted workload projection is available", + ) + return result diff --git a/tests/test_mcp.py b/tests/test_mcp.py index a0cb93f..163de66 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -32,6 +32,8 @@ def test_mcp_base_server_registers_orientation_doi_and_fos10_tools() -> None: "get_alerts", "query_repository_navigation", "get_repository_navigation_facet", + "query_workloads", + "resolve_workload_reference", } <= names assert names == CORE_TOOL_NAMES diff --git a/tests/test_workload_projection.py b/tests/test_workload_projection.py new file mode 100644 index 0000000..aab26b4 --- /dev/null +++ b/tests/test_workload_projection.py @@ -0,0 +1,271 @@ +from __future__ import annotations + +import asyncio +import json +from copy import deepcopy +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from jsonschema import Draft202012Validator, FormatChecker +from sqlalchemy.ext.asyncio import create_async_engine + +from hub_core.contracts import workload_projection_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.store import InMemoryPortStore +from hub_core.runtime.tables import ( + runtime_metadata, + runtime_workload_projection_records, + runtime_workload_projection_state, +) +from hub_core.runtime.workload_projection import ( + WorkloadCursorMismatch, + WorkloadProjectionRejected, + WorkloadProjectionService, +) + + +def fixture() -> dict[str, Any]: + resource = workload_projection_contract_root().joinpath( + "fixtures", "repo-manager-nine-workloads.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_workload_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_workload_page(self, cursor: str | None) -> dict[str, Any]: + raise RuntimeError("repo-manager workload index unavailable") + + +def next_snapshot( + page: dict[str, Any], *, marker: str, generated_at: str +) -> dict[str, Any]: + changed = deepcopy(page) + changed["snapshot"].update( + {"snapshot_id": marker * 64, "generated_at": generated_at} + ) + return changed + + +def test_ingests_nine_records_and_resolves_only_exact_reference() -> None: + async def run() -> None: + store = InMemoryPortStore() + service = WorkloadProjectionService(client=PageClient([fixture()]), store=store) + + result = await service.refresh() + resolved = await service.resolve( + rapp_id="rapp-qonto", name="qonto", deployable="rapp-qonto" + ) + wrong_name = await service.resolve( + rapp_id="rapp-qonto", name="qonto-assistant" + ) + wrong_deployable = await service.resolve( + rapp_id="rapp-openbao", name="openbao", deployable="rapp-openbao" + ) + + assert result.workload_count == 9 + assert resolved is not None and resolved["status"] == "resolved" + assert wrong_name is not None and wrong_name["reason"] == "not_found" + assert wrong_deployable is not None + assert wrong_deployable["reason"] == "deployable_not_declared" + assert await service.readiness_checks() == {"workload_projection": "ok"} + + asyncio.run(run()) + + +def test_deletion_and_full_rebuild_restore_equivalent_content() -> None: + async def run() -> None: + source = fixture() + store = InMemoryPortStore() + initial = WorkloadProjectionService(client=PageClient([source]), store=store) + original = await initial.refresh() + + reduced = next_snapshot( + source, marker="a", generated_at="2026-08-22T12:05:00Z" + ) + reduced["workloads"] = reduced["workloads"][:-1] + reduced["snapshot"]["total_workload_count"] = 8 + await WorkloadProjectionService( + client=PageClient([reduced]), store=store + ).refresh() + after_delete = await store.get_workload_projection() + + restored = next_snapshot( + source, marker="b", generated_at="2026-08-22T12:10:00Z" + ) + rebuilt = await WorkloadProjectionService( + client=PageClient([restored]), store=store + ).refresh() + + assert after_delete is not None and len(after_delete.workloads) == 8 + assert rebuilt.workload_count == 9 + assert rebuilt.content_hash == original.content_hash + + asyncio.run(run()) + + +def test_invalid_non_utc_input_preserves_last_generation_as_stale() -> None: + async def run() -> None: + store = InMemoryPortStore() + await WorkloadProjectionService( + client=PageClient([fixture()]), store=store + ).refresh() + before = await store.get_workload_projection() + invalid = next_snapshot( + fixture(), marker="c", generated_at="2026-08-22T14:00:00+02:00" + ) + + with pytest.raises(WorkloadProjectionRejected, match="invalid workload page"): + await WorkloadProjectionService( + client=PageClient([invalid]), store=store + ).refresh() + after = await store.get_workload_projection() + + assert before is not None and after is not None + assert after.content_hash == before.content_hash + assert after.workloads == before.workloads + assert after.projection_status == "stale" + + asyncio.run(run()) + + +def test_query_order_cursor_and_generation_binding() -> None: + async def run() -> None: + store = InMemoryPortStore() + service = WorkloadProjectionService(client=PageClient([fixture()]), store=store) + await service.refresh() + first = await service.query(limit=4) + + assert first is not None and first["next_cursor"] + assert [item["rapp_id"] for item in first["workloads"]] == sorted( + item["rapp_id"] for item in first["workloads"] + ) + second = await service.query(cursor=first["next_cursor"], limit=4) + assert second is not None and len(second["workloads"]) == 4 + + changed = next_snapshot( + fixture(), marker="d", generated_at="2026-08-22T12:15:00Z" + ) + await WorkloadProjectionService( + client=PageClient([changed]), store=store + ).refresh() + with pytest.raises(WorkloadCursorMismatch, match="cursor_snapshot_mismatch"): + await service.query(cursor=first["next_cursor"], limit=4) + + asyncio.run(run()) + + +def test_outage_without_generation_fails_readiness_closed() -> None: + async def run() -> None: + service = WorkloadProjectionService( + client=FailingClient(), store=InMemoryPortStore() + ) + with pytest.raises(WorkloadProjectionRejected, match="unavailable"): + await service.refresh() + assert await service.readiness_checks() == {"workload_projection": "unavailable"} + + asyncio.run(run()) + + +def test_http_and_mcp_backing_routes_are_read_only_and_schema_valid() -> None: + settings = RuntimeSettings(environment="test", backend="memory", allow_ephemeral=True) + app = create_app( + settings=settings, + port_store=InMemoryPortStore(), + workload_projection_client=PageClient([fixture()]), + ) + with TestClient(app) as runtime: + response = runtime.get( + "/ports/projections/workloads", params={"deployable": "openbao"} + ) + resolved = runtime.get( + "/ports/projections/workloads/resolve", + params={"rapp_id": "rapp-openbao", "name": "openbao"}, + ) + unknown = runtime.get( + "/ports/projections/workloads/resolve", + params={"rapp_id": "rapp-openbao", "name": "railiance-platform"}, + ) + ready = runtime.get("/readyz") + openapi = runtime.get("/openapi.json").json() + + schema_root = workload_projection_contract_root().joinpath("schemas") + projection_schema = json.loads( + schema_root.joinpath("workload-projection.schema.json").read_text(encoding="utf-8") + ) + resolution_schema = json.loads( + schema_root.joinpath("workload-resolution.schema.json").read_text(encoding="utf-8") + ) + Draft202012Validator(projection_schema, format_checker=FormatChecker()).validate( + response.json() + ) + Draft202012Validator(resolution_schema, format_checker=FormatChecker()).validate( + resolved.json() + ) + Draft202012Validator(resolution_schema, format_checker=FormatChecker()).validate( + unknown.json() + ) + assert response.status_code == resolved.status_code == unknown.status_code == 200 + assert [record["rapp_id"] for record in response.json()["workloads"]] == [ + "rapp-openbao" + ] + assert resolved.json()["status"] == "resolved" + assert unknown.json()["status"] == "unknown" + assert ready.json()["checks"]["workload_projection"] == "ok" + paths = { + path: item + for path, item in openapi["paths"].items() + if path.startswith("/ports/projections/workloads") + } + assert paths and all(set(item) == {"get"} for item in paths.values()) + + +def test_postgres_workload_projection_survives_reopen(tmp_path) -> None: + database_url = f"sqlite+aiosqlite:///{tmp_path / 'workloads.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 WorkloadProjectionService( + client=PageClient([fixture()]), store=first_store + ).refresh() + await first_store.aclose() + second_store = PostgresPortStore.from_url(database_url) + projection = await second_store.get_workload_projection() + await second_store.aclose() + + assert projection is not None + assert projection.content_hash == result.content_hash + assert len(projection.workloads) == 9 + assert all(item["observed_at"].endswith("Z") for item in projection.workloads) + + asyncio.run(run()) + + +def test_projection_tables_and_module_have_no_private_repo_manager_coupling() -> None: + source = ( + __import__("inspect") + .getsource(__import__("hub_core.runtime.workload_projection", fromlist=["*"])) + ) + + assert "repo_manager." not in source + assert not runtime_workload_projection_state.foreign_keys + assert not runtime_workload_projection_records.foreign_keys diff --git a/tests/test_workload_projection_contract.py b/tests/test_workload_projection_contract.py new file mode 100644 index 0000000..b92e45c --- /dev/null +++ b/tests/test_workload_projection_contract.py @@ -0,0 +1,95 @@ +import json +from typing import Any + +from jsonschema import Draft202012Validator, FormatChecker + +from hub_core.contracts import ( + WORKLOAD_PROJECTION_CONTRACT_ID, + WORKLOAD_PROJECTION_CONTRACT_VERSION, + workload_projection_contract_root, +) + + +ROOT = workload_projection_contract_root() +SCHEMAS = ROOT.joinpath("schemas") +FIXTURE = ROOT.joinpath("fixtures", "repo-manager-nine-workloads.json") +OPENAPI = ROOT.joinpath("openapi", "workload-projection.openapi.json") +COMPATIBILITY = ROOT.joinpath("compatibility-matrix.json") + + +def load(resource: Any) -> Any: + return json.loads(resource.read_text(encoding="utf-8")) + + +def validate(instance: Any, name: str) -> None: + schema = load(SCHEMAS.joinpath(name)) + Draft202012Validator.check_schema(schema) + Draft202012Validator(schema, format_checker=FormatChecker()).validate(instance) + + +def test_packaged_workload_contract_identity_and_artifacts() -> None: + assert WORKLOAD_PROJECTION_CONTRACT_ID == "helixforge.workload-projection" + assert WORKLOAD_PROJECTION_CONTRACT_VERSION == "1.0.0" + for artifact in (ROOT.joinpath("README.md"), FIXTURE, OPENAPI, COMPATIBILITY): + assert artifact.is_file() + for name in ( + "workload-projection-page.schema.json", + "workload-projection.schema.json", + "workload-resolution.schema.json", + ): + Draft202012Validator.check_schema(load(SCHEMAS.joinpath(name))) + + +def test_nine_declaration_fixture_is_valid_exact_and_canonical_utc() -> None: + fixture = load(FIXTURE) + validate(fixture, "workload-projection-page.schema.json") + workloads = fixture["workloads"] + identities = [(record["rapp_id"], record["name"]) for record in workloads] + + assert len(workloads) == fixture["snapshot"]["total_workload_count"] == 9 + assert identities == sorted(identities) + assert len(identities) == len(set(identities)) + assert all(record["declaration_repo"] == record["rapp_id"] for record in workloads) + assert all( + record["declaration_path"] + == f"{record['declaration_repo']}/declarations/rapp.yaml" + for record in workloads + ) + assert fixture["snapshot"]["generated_at"].endswith("Z") + assert all(record["observed_at"].endswith("Z") for record in workloads) + + +def test_compatibility_matrix_preserves_repo_manager_authority_and_forbids_inference() -> None: + matrix = load(COMPATIBILITY) + + assert matrix["contract_id"] == WORKLOAD_PROJECTION_CONTRACT_ID + assert matrix["accepted_inputs"] == [ + { + "contract_id": "helixforge.workload-projection-source", + "version": "1.0.0", + "source_contract": "helixforge.workload-reference/v1", + "status": "current", + "coercion": False, + } + ] + assert matrix["authority"]["index_and_resolution"] == "repo-manager" + assert set(matrix["forbidden_fallbacks"]) == { + "repository_name", + "owner_repo", + "path_segment", + "alias", + } + + +def test_workload_openapi_is_projection_query_read_only() -> None: + document = load(OPENAPI) + operations = [ + (method, operation) + for path_item in document["paths"].values() + for method, operation in path_item.items() + if method in {"get", "post", "put", "patch", "delete"} + ] + + assert document["openapi"] == "3.1.0" + assert operations and all(method == "get" for method, _ in operations) + assert all(operation["x-port-id"] == "port.projection.query" for _, operation in operations) diff --git a/workplans/HUB-WP-0007-workload-projection-transport.md b/workplans/HUB-WP-0007-workload-projection-transport.md index 755eb40..eab76ec 100644 --- a/workplans/HUB-WP-0007-workload-projection-transport.md +++ b/workplans/HUB-WP-0007-workload-projection-transport.md @@ -4,7 +4,7 @@ type: workplan title: "Authoritative workload projection transport" domain: infotech repo: hub-core -status: active +status: finished owner: codex topic_slug: custodian created: "2026-08-23" @@ -14,6 +14,7 @@ related: - HUB-WP-0006 - RMGR-WP-0010-T04 - RMGR-CONTRACT-WORKLOAD-REFERENCE-0001 +state_hub_workstream_id: "1f767e78-ab36-5e62-a23a-31af0f21a723" --- # Authoritative workload projection transport @@ -30,6 +31,7 @@ semantics, exposing Repo Manager persistence, or inferring identities. id: HUB-WP-0007-T01 status: done priority: medium +state_hub_task_id: "99787c58-f0b2-5e67-9816-ff4ea0ad6057" ``` Disposition: **approved** on 2026-08-23 by operator instruction to implement. @@ -46,8 +48,9 @@ reconstructing Repo Manager work. ```task id: HUB-WP-0007-T02 -status: progress +status: done priority: medium +state_hub_task_id: "24279a94-a23c-5468-8623-ecddcafaef3a" ``` Package a strict versioned input/output contract for records keyed by exact @@ -56,12 +59,19 @@ source repository, declaration path, source Git revision, and canonical UTC observation time. Preserve Repo Manager diagnostics and reject unsupported versions or inferred identity. +Completed 2026-08-23. Packaged `helixforge.workload-projection` 1.0.0 defines +strict Repo Manager input pages, normalized output and exact-resolution +schemas, a read-only OpenAPI fragment, authority/compatibility policy, and the +nine-declaration fixture. Canonical UTC is required at the schema boundary; +identity and path consistency are also checked semantically. + ## Implement durable rebuildable ingestion ```task id: HUB-WP-0007-T03 -status: todo +status: done priority: medium +state_hub_task_id: "cb0eb118-a968-51d4-84fc-0a9bf959a501" ``` Extend the injected `port.repo` boundary with workload pages. Deterministically @@ -69,12 +79,21 @@ replace the active generation in PostgreSQL, retain source/rebuild provenance and a canonical content hash, and keep the previous generation on invalid or unavailable input. +Completed 2026-08-23. `WorkloadProjectionService` consumes a separate injected +`port.repo` workload client, validates complete paged transfers, rejects error +diagnostics, duplicate identity/deployables, unsupported versions, noncanonical +paths, and stale/reused snapshots, then atomically replaces the generation. +Migration `0004_workload_projection` persists projection state and normalized +records without foreign keys or Repo Manager imports. Invalid input preserves +the prior generation as stale and readiness reports current/stale/unavailable. + ## Expose exact read-only HTTP and MCP resolution ```task id: HUB-WP-0007-T04 -status: todo +status: done priority: medium +state_hub_task_id: "722dfd4a-62d3-59b0-9494-08ec8f184571" ``` Expose list and exact-reference queries through `port.projection.query` and @@ -82,12 +101,20 @@ matching MCP tools. Exact resolution returns `resolved` or `unknown` without repository-name, owner, alias, or path inference. No workload mutation route is permitted. +Completed 2026-08-23. GET-only list and exact-resolution routes expose +normalized records with content/source provenance. Filters and cursors use +only explicit `rapp_id`, name, and deployable values. Resolution returns +`unknown/not_found` or `unknown/deployable_not_declared`; it never consults +owner, repository, path, or alias fallbacks. MCP tools `query_workloads` and +`resolve_workload_reference` call the same HTTP surface. + ## Prove conformance and hand off ```task id: HUB-WP-0007-T05 -status: todo +status: done priority: medium +state_hub_task_id: "04569701-6cb4-5762-ac9b-a9dec4e2bb52" ``` Ingest a nine-declaration Repo Manager fixture; prove deletion plus full rebuild @@ -96,10 +123,18 @@ reopen, provenance, deterministic hashes/order, read-only OpenAPI, and absence of Repo Manager private-model or foreign-database coupling. Run `make test` and `make conformance`, then reply with owner identifiers and evidence. +Completed 2026-08-23. Tests ingest and query all nine fixture declarations, +prove exact/unknown resolution, deletion plus equivalent full rebuild, +generation-bound cursors, UTC rejection, outage/stale behavior, durable reopen, +schema-valid HTTP responses, GET-only OpenAPI, and absence of private-model or +foreign-database coupling. `make test` passes 100 tests, live +`make conformance` passes 8/8, the wheel builds, and PostgreSQL migrations +render through `0004_workload_projection` (501 lines). + ## Acceptance - [x] Owner interface disposition and source revisions recorded -- [ ] Normalized transport contract is versioned and packaged -- [ ] Durable projection rebuild preserves authority and provenance -- [ ] HTTP/MCP exact reference queries never infer identity -- [ ] Nine-declaration and rebuild conformance evidence passes +- [x] Normalized transport contract is versioned and packaged +- [x] Durable projection rebuild preserves authority and provenance +- [x] HTTP/MCP exact reference queries never infer identity +- [x] Nine-declaration and rebuild conformance evidence passes