Compatibility runtime candidate; projections are authoritative.
" + ) + + return router + + +def _enabled(request: Request, group: str) -> None: + if group not in request.app.state.settings.v2_groups: + raise HTTPException(status_code=404, detail="compatibility group is disabled") + + +async def _protected( + request: Request, + group: str, + authorization: str | None, + *, + write: bool = False, +) -> None: + _enabled(request, group) + if write and group not in request.app.state.settings.v2_write_groups: + raise HTTPException(status_code=503, detail="compatibility group is read-only") + if not authorization or not authorization.startswith("Bearer "): + raise _unauthorized("Missing bearer token") + token = authorization.removeprefix("Bearer ").strip() + configured = request.app.state.settings.api_token + if configured and hmac.compare_digest(token, configured): + return + try: + valid = await _store(request).validate_token(token) + except Exception as exc: + raise HTTPException(status_code=503, detail="authorization dependency unavailable") from exc + if not valid: + raise _unauthorized("Invalid bearer token") + + +def _store(request: Request) -> SQLCompatibilityStore: + store = request.app.state.compat_store + if store is None: + raise HTTPException(status_code=503, detail="compatibility store unavailable") + return store + + +def _unauthorized(message: str) -> HTTPException: + return HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail={"code": "unauthorized", "message": message}, + ) + + +def _page(values: list[dict[str, Any]]) -> dict[str, Any]: + return {"data": values, "count": len(values)} + + +def _hub(row: Any) -> dict[str, Any]: + value = dict(row["body"] or {}) + value.update( + id=row["id"], slug=row["slug"], name=row["name"], domain=row["domain"], + hubKind=row["hub_kind"], hubFamily=row["hub_family"], + vsmFunction=row["vsm_function"], vsmSystem=row["vsm_system"], + status=row["status"], description=row["description"], + ) + return {key: item for key, item in value.items() if item is not None} + + +def _manifest(row: Any) -> dict[str, Any]: + value = dict(row["body"] or {}) + value.update( + id=row["id"], hubId=row["hub_id"], hubSlug=row["hub_slug"], + manifestVersion=row["manifest_version"], status=row["status"], + ) + return {key: item for key, item in value.items() if item is not None} + + +def _consumer(row: Any) -> dict[str, Any]: + value = dict(row["body"] or {}) + value.update( + id=row["id"], slug=row["slug"], name=row["name"], + description=row["description"], + hubCapabilityManifestId=row["hub_capability_manifest_id"], + rateLimitPerMinute=row["rate_limit_per_minute"], quotaPerDay=row["quota_per_day"], + keyPrefix=row["key_prefix"], status=row["status"], + ) + return {key: item for key, item in value.items() if item is not None} + + +def _widget(row: Any) -> dict[str, Any]: + value = dict(row["body"] or {}) + value.update( + id=row["id"], hubId=row["hub_id"], name=row["name"], + widgetType=row["widget_type"], capabilityRef=row["capability_ref"], + viewContext=row["view_context"], policyScope=row["policy_scope"], + status=row["status"], + ) + return {key: item for key, item in value.items() if item is not None} + + +def _hash_token(token: str) -> str: + return hashlib.sha256(token.encode()).hexdigest() + + +def _now() -> datetime: + return datetime.now(timezone.utc) diff --git a/hub_core/runtime/compat_catalogs.py b/hub_core/runtime/compat_catalogs.py new file mode 100644 index 0000000..1e346e0 --- /dev/null +++ b/hub_core/runtime/compat_catalogs.py @@ -0,0 +1,66 @@ +from __future__ import annotations + + +def item(slug: str, name: str, description: str | None = None) -> dict[str, str]: + value = {"slug": slug, "name": name} + if description is not None: + value["description"] = description + return value + + +WIDGET_TYPES = [ + item("status-summary", "Status Summary"), + item("workplan-board", "Workplan Board"), + item("event-stream", "Event Stream"), + item("action", "Action Control", "Button, link, or trigger widget"), + item("chart", "Chart", "Data visualisation chart widget"), + item("chat", "Chat Region", "Conversational interaction region"), + item("diff", "Diff / Review", "Code diff or change review element"), + item("form", "Form", "Data entry form widget"), + *[item(f"ops-{name}", f"ops-{name}") for name in ( + "backup-set", "cluster", "endpoint", "environment", "host", "incident", + "migration-wave", "readiness-gate", "release", "risk", "runbook", + "secret-set", "service", "service-catalog", + )], + item("panel", "Status Panel", "Summary or status information panel"), + item("recommendation", "Recommendation", "AI or system recommendation block"), + item("table", "Table", "Tabular data display widget"), + item("workflow-step", "Workflow Step", "Single step in a multi-step workflow"), +] + +EVENT_TYPES = [ + item("interaction.event", "Interaction Event"), + item("workplan.progress", "Workplan Progress"), + item("evidence.recorded", "Evidence Recorded"), + *[ + item(slug, slug.replace("_", " ").title()) + for slug in ( + "abandoned", "accepted_recommendation", "blocked_by_policy", "clicked", + "commented", "escalated", "failed", "flagged_confusing", "flagged_helpful", + "focused", "rejected_recommendation", "retracted", "retried", "submitted", + "viewed", + ) + ], + *[ + item(f"ops-{name}", f"ops-{name}") + for name in ( + "backup-verified", "drift-detected", "endpoint-verified", "health-checked", + "inventory-registered", "inventory-updated", "migration-gate-failed", + "migration-gate-passed", "readiness-gate-updated", "release-observed", + "restore-tested", "risk-accepted", "risk-raised", "runbook-executed", + "service-discovered", + ) + ], +] + +ANNOTATION_CATEGORIES = [ + item("operator-note", "Operator Note"), + item("compatibility-delta", "Compatibility Delta"), + item("migration-evidence", "Migration Evidence"), +] + +POLICY_SCOPES = [ + item("public-read", "Public Read"), + item("hub-write", "Hub Write"), + item("operator-admin", "Operator Admin"), +] diff --git a/hub_core/runtime/config.py b/hub_core/runtime/config.py index af919d7..c0fd378 100644 --- a/hub_core/runtime/config.py +++ b/hub_core/runtime/config.py @@ -11,6 +11,10 @@ def _env_bool(name: str, default: bool) -> bool: return raw.strip().lower() in {"1", "true", "yes", "on"} +def _env_set(name: str) -> frozenset[str]: + return frozenset(value.strip() for value in os.getenv(name, "").split(",") if value.strip()) + + @dataclass(frozen=True, slots=True) class RuntimeSettings: environment: str = "development" @@ -23,6 +27,21 @@ class RuntimeSettings: mcp_port: int = 8011 mcp_transport: str = "http" database_url: str | None = None + api_token: str | None = None + 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: + overlap = self.v2_write_groups & self.legacy_write_groups + if overlap: + joined = ", ".join(sorted(overlap)) + raise ValueError(f"hub-core and Core Hub write groups overlap: {joined}") + unknown = self.v2_write_groups - self.v2_groups + if unknown: + joined = ", ".join(sorted(unknown)) + raise ValueError(f"hub-core write groups are not enabled: {joined}") @classmethod def from_env(cls) -> RuntimeSettings: @@ -41,6 +60,11 @@ class RuntimeSettings: mcp_port=int(os.getenv("HUB_CORE_MCP_PORT", "8011")), 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"), + 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"), + legacy_health=_env_bool("HUB_CORE_LEGACY_HEALTH", False), ) def readiness_checks(self, store_backend: str) -> dict[str, str]: @@ -51,6 +75,9 @@ class RuntimeSettings: "active_backend": store_backend, "ephemeral_backend": "allowed" if ephemeral_allowed else "not_allowed", "contract": "helixforge.hub-extension/0.1.0", + "v2_groups": ",".join(sorted(self.v2_groups)) or "none", + "v2_write_groups": ",".join(sorted(self.v2_write_groups)) or "none", + "legacy_write_groups": ",".join(sorted(self.legacy_write_groups)) or "none", } def is_ready(self, store_backend: str) -> bool: diff --git a/hub_core/runtime/migration.py b/hub_core/runtime/migration.py new file mode 100644 index 0000000..cb390b1 --- /dev/null +++ b/hub_core/runtime/migration.py @@ -0,0 +1,438 @@ +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from uuid import NAMESPACE_URL, uuid4, uuid5 + +import sqlalchemy as sa + +from hub_core.runtime.postgres_store import PostgresPortStore +from hub_core.runtime.tables import ( + compat_api_consumers, + compat_api_keys, + compat_hubs, + compat_manifests, + compat_migration_runs, + compat_widgets, + runtime_audit_ledger, + runtime_import_runs, + runtime_interaction_events, +) + +SCHEMA_VERSION = "core-hub.migration.v1" +COLLECTIONS = ( + "hubs", + "hubCapabilityManifests", + "apiConsumers", + "apiKeys", + "widgets", + "interactionEvents", + "migrationRuns", +) +SECRET_FIELDS = {"fullKey", "rawKey", "apiKey", "secret", "token", "authorization"} + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=_json_default) + + +def _json_default(value: Any) -> str: + if isinstance(value, datetime): + return _timestamp(value) + raise TypeError(f"cannot encode {type(value).__name__}") + + +def _timestamp(value: datetime) -> str: + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _datetime(value: Any, default: datetime | None = None) -> datetime: + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + if isinstance(value, str) and value: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + return default or _now() + + +def _records(bundle: dict[str, Any], collection: str) -> list[dict[str, Any]]: + records = bundle.get("records", {}) + values = records.get(collection, []) if isinstance(records, dict) else [] + return [dict(value) for value in values] if isinstance(values, list) else [] + + +def _contains_secret(value: Any) -> bool: + if isinstance(value, dict): + return any(key in SECRET_FIELDS or _contains_secret(child) for key, child in value.items()) + if isinstance(value, list): + return any(_contains_secret(child) for child in value) + return False + + +def _canonical_record(record: dict[str, Any]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in record.items(): + if key in {"createdAt", "updatedAt", "occurredAt", "recordedAt"} and value: + result[key] = _timestamp(_datetime(value)) + elif isinstance(value, dict): + result[key] = _canonical_record(value) + elif isinstance(value, list): + result[key] = [ + _canonical_record(item) if isinstance(item, dict) else item for item in value + ] + else: + result[key] = value + return result + + +def _identity(record: dict[str, Any]) -> str: + return str(record.get("id") or record.get("slug") or record.get("keyHash") or "") + + +def content_hashes(bundle: dict[str, Any]) -> dict[str, str]: + hashes: dict[str, str] = {} + for collection in COLLECTIONS: + values = [_canonical_record(value) for value in _records(bundle, collection)] + values.sort(key=_identity) + hashes[collection] = hashlib.sha256(_json(values).encode()).hexdigest() + return hashes + + +def identity_sets(bundle: dict[str, Any]) -> dict[str, list[str]]: + return { + collection: sorted(_identity(value) for value in _records(bundle, collection)) + for collection in COLLECTIONS + } + + +def bundle_digest(bundle: dict[str, Any]) -> str: + value = {key: child for key, child in bundle.items() if key != "bundleSha256"} + return hashlib.sha256(_json(value).encode()).hexdigest() + + +def validate_bundle(bundle: dict[str, Any]) -> dict[str, Any]: + errors: list[str] = [] + warnings: list[str] = [] + unknown = set(bundle) - { + "schemaVersion", + "source", + "sourceRevision", + "exportedAt", + "highWaterMark", + "bundleSha256", + "records", + } + if unknown: + errors.append(f"unknown bundle fields: {', '.join(sorted(unknown))}") + if bundle.get("schemaVersion") != SCHEMA_VERSION: + errors.append(f"schemaVersion must be {SCHEMA_VERSION}") + if not isinstance(bundle.get("records"), dict): + errors.append("records must be an object") + else: + unknown_collections = set(bundle["records"]) - set(COLLECTIONS) + if unknown_collections: + errors.append(f"unknown collections: {', '.join(sorted(unknown_collections))}") + claimed = bundle.get("bundleSha256") + calculated = bundle_digest(bundle) + if claimed and claimed != calculated: + errors.append("bundleSha256 does not match canonical bundle content") + + required = { + "hubs": ("slug", "name"), + "hubCapabilityManifests": ("id",), + "apiConsumers": ("name",), + "apiKeys": ("id", "keyPrefix", "keyHash"), + "widgets": ("id", "name"), + "interactionEvents": ("id", "widgetId", "eventType"), + "migrationRuns": ("id", "source", "schemaVersion", "bundleSha256"), + } + for collection in COLLECTIONS: + seen: set[str] = set() + for index, record in enumerate(_records(bundle, collection)): + path = f"records.{collection}[{index}]" + for field in required[collection]: + if record.get(field) in (None, ""): + errors.append(f"{path}.{field} is required") + identity = _identity(record) + if identity and identity in seen: + errors.append(f"duplicate {collection} identity: {identity}") + seen.add(identity) + if _contains_secret(record): + errors.append(f"{path} contains secret-shaped material") + for record in _records(bundle, "hubCapabilityManifests"): + if not record.get("hubId") and not record.get("hubSlug"): + errors.append("manifest must include hubId or hubSlug") + for record in _records(bundle, "apiConsumers"): + if not record.get("id") and not record.get("slug"): + errors.append("API consumer must include id or slug") + for record in _records(bundle, "apiKeys"): + if not record.get("apiConsumerId") and not record.get("apiConsumerSlug"): + errors.append("API key must include apiConsumerId or apiConsumerSlug") + for record in _records(bundle, "widgets"): + if not record.get("hubId") and not record.get("hubSlug"): + errors.append("widget must include hubId or hubSlug") + + return { + "ok": not errors, + "schemaVersion": bundle.get("schemaVersion"), + "source": bundle.get("source", "unknown"), + "bundleSha256": calculated, + "highWaterMark": bundle.get("highWaterMark"), + "counts": {name: {"input": len(_records(bundle, name))} for name in COLLECTIONS}, + "identitySets": identity_sets(bundle), + "contentHashes": content_hashes(bundle), + "errors": errors, + "warnings": warnings, + } + + +async def import_bundle( + store: PostgresPortStore, bundle: dict[str, Any], *, dry_run: bool = False +) -> dict[str, Any]: + report = validate_bundle(bundle) + report["dryRun"] = dry_run + report["idempotent"] = False + report["counts"] = { + name: {"input": len(_records(bundle, name)), "created": 0, "updated": 0, "skipped": 0} + for name in COLLECTIONS + } + if not report["ok"]: + return report + + async with store.sessions() as session: + prior = ( + await session.execute( + sa.select(runtime_import_runs).where( + runtime_import_runs.c.bundle_sha256 == report["bundleSha256"] + ) + ) + ).mappings().first() + if prior and not dry_run: + report["idempotent"] = True + report["migrationRunId"] = prior["id"] + for name in COLLECTIONS: + report["counts"][name]["skipped"] = report["counts"][name]["input"] + return report + + async with store.sessions.begin() as session: + hubs = await _maps(session, compat_hubs, "id", "slug") + await _import_hubs(session, bundle, report, hubs, dry_run) + hubs = await _maps(session, compat_hubs, "id", "slug") if not dry_run else hubs + await _import_manifests(session, bundle, report, hubs, dry_run) + consumers = await _maps(session, compat_api_consumers, "id", "slug") + await _import_consumers(session, bundle, report, consumers, dry_run) + consumers = ( + await _maps(session, compat_api_consumers, "id", "slug") if not dry_run else consumers + ) + await _import_keys(session, bundle, report, consumers, dry_run) + await _import_widgets(session, bundle, report, hubs, dry_run) + await _import_events(session, bundle, report, dry_run) + await _import_source_runs(session, bundle, report, dry_run) + if dry_run: + await session.rollback() + return report + run_id = str(uuid4()) + await session.execute( + runtime_import_runs.insert().values( + id=run_id, + source=report["source"], + schema_version=SCHEMA_VERSION, + bundle_sha256=report["bundleSha256"], + high_water_mark=report["highWaterMark"], + counts=report["counts"], + content_hashes=report["contentHashes"], + status="imported", + created_at=_now(), + ) + ) + await session.execute( + runtime_audit_ledger.insert().values( + id=str(uuid4()), + action="migration.imported", + subject_type="migration_bundle", + subject_id=report["bundleSha256"], + correlation_id=None, + payload_hash=hashlib.sha256(_json(report["contentHashes"]).encode()).hexdigest(), + detail={"source": report["source"], "counts": report["counts"]}, + recorded_at=_now(), + ) + ) + report["migrationRunId"] = run_id + return report + + +async def _maps(session: Any, table: sa.Table, *keys: str) -> dict[str, dict[str, Any]]: + rows = (await session.execute(sa.select(table))).mappings().all() + return {str(row[key]): dict(row) for row in rows for key in keys if row[key]} + + +def _mark(report: dict[str, Any], name: str, existing: Any) -> None: + report["counts"][name]["updated" if existing else "created"] += 1 + + +async def _upsert(session: Any, table: sa.Table, row: dict[str, Any], existing: Any) -> None: + if existing: + await session.execute(table.update().where(table.c.id == existing["id"]).values(**row)) + else: + await session.execute(table.insert().values(**row)) + + +def _parent_id(record: dict[str, Any], rows: dict[str, dict[str, Any]], kind: str) -> str: + parent = record.get(f"{kind}Id") or record.get(f"{kind}Slug") + if parent in rows: + return str(rows[str(parent)]["id"]) + raise ValueError(f"unresolved {kind} reference: {parent}") + + +async def _import_hubs(session: Any, bundle: dict[str, Any], report: dict, rows: dict, dry: bool) -> None: + for record in _records(bundle, "hubs"): + existing = rows.get(str(record.get("id"))) or rows.get(str(record.get("slug"))) + _mark(report, "hubs", existing) + if dry: + continue + now = _now() + row = { + "id": existing["id"] if existing else str(record.get("id") or uuid4()), + "slug": record["slug"], "name": record["name"], "domain": record.get("domain"), + "hub_kind": record.get("hubKind"), "hub_family": record.get("hubFamily"), + "vsm_function": record.get("vsmFunction"), "vsm_system": record.get("vsmSystem"), + "status": record.get("status", "active"), "description": record.get("description"), + "body": record, "created_at": _datetime(record.get("createdAt"), now), + "updated_at": _datetime(record.get("updatedAt"), now), + } + await _upsert(session, compat_hubs, row, existing) + + +async def _import_manifests(session: Any, bundle: dict, report: dict, hubs: dict, dry: bool) -> None: + current = await _maps(session, compat_manifests, "id") + for record in _records(bundle, "hubCapabilityManifests"): + existing = current.get(str(record["id"])); _mark(report, "hubCapabilityManifests", existing) + if dry: continue + now = _now(); hub_id = _parent_id(record, hubs, "hub") + row = {"id": record["id"], "hub_id": hub_id, "hub_slug": hubs[hub_id]["slug"], + "manifest_version": record.get("manifestVersion", "0.1.0"), + "status": record.get("status", "draft"), "body": record, + "created_at": _datetime(record.get("createdAt"), now), + "updated_at": _datetime(record.get("updatedAt"), now)} + await _upsert(session, compat_manifests, row, existing) + + +async def _import_consumers(session: Any, bundle: dict, report: dict, rows: dict, dry: bool) -> None: + for record in _records(bundle, "apiConsumers"): + existing = rows.get(str(record.get("id"))) or rows.get(str(record.get("slug"))) + _mark(report, "apiConsumers", existing) + if dry: continue + row = {"id": existing["id"] if existing else str(record.get("id") or uuid4()), + "slug": record.get("slug"), "name": record["name"], + "description": record.get("description"), + "hub_capability_manifest_id": record.get("hubCapabilityManifestId"), + "rate_limit_per_minute": record.get("rateLimitPerMinute"), + "quota_per_day": record.get("quotaPerDay"), "key_prefix": record.get("keyPrefix"), + "status": record.get("status", "active"), "body": record, + "created_at": _datetime(record.get("createdAt"))} + await _upsert(session, compat_api_consumers, row, existing) + + +async def _import_keys(session: Any, bundle: dict, report: dict, consumers: dict, dry: bool) -> None: + current = await _maps(session, compat_api_keys, "id", "key_hash") + for record in _records(bundle, "apiKeys"): + existing = current.get(str(record["id"])) or current.get(str(record["keyHash"])) + _mark(report, "apiKeys", existing) + if dry: continue + consumer_id = _parent_id(record, consumers, "apiConsumer") + row = {"id": record["id"], "api_consumer_id": consumer_id, + "key_prefix": record["keyPrefix"], "key_hash": record["keyHash"], + "scopes": record.get("scopes"), "status": record.get("status", "active"), + "created_at": _datetime(record.get("createdAt"))} + await _upsert(session, compat_api_keys, row, existing) + + +async def _import_widgets(session: Any, bundle: dict, report: dict, hubs: dict, dry: bool) -> None: + current = await _maps(session, compat_widgets, "id") + for record in _records(bundle, "widgets"): + existing = current.get(str(record["id"])); _mark(report, "widgets", existing) + if dry: continue + row = {"id": record["id"], "hub_id": _parent_id(record, hubs, "hub"), + "name": record["name"], "widget_type": record.get("widgetType"), + "capability_ref": record.get("capabilityRef"), + "view_context": record.get("viewContext"), "policy_scope": record.get("policyScope"), + "status": record.get("status", "active"), "body": record, + "created_at": _datetime(record.get("createdAt"))} + await _upsert(session, compat_widgets, row, existing) + + +async def _import_events(session: Any, bundle: dict, report: dict, dry: bool) -> None: + current = await _maps(session, runtime_interaction_events, "id") + for record in _records(bundle, "interactionEvents"): + existing = current.get(str(record["id"])); _mark(report, "interactionEvents", existing) + if dry: continue + created = _datetime(record.get("createdAt") or record.get("occurredAt")) + row = {"id": record["id"], "schema_version": "0.1.0", + "correlation_id": str(record.get("correlationId") or uuid5(NAMESPACE_URL, record["id"])), + "event_type": record["eventType"], "occurred_at": created, + "subject_refs": {"widget": record["widgetId"]}, "payload": {"legacy": record}, + "recorded_at": _datetime(record.get("recordedAt"), created)} + await _upsert(session, runtime_interaction_events, row, existing) + + +async def _import_source_runs(session: Any, bundle: dict, report: dict, dry: bool) -> None: + current = await _maps(session, compat_migration_runs, "id") + for record in _records(bundle, "migrationRuns"): + existing = current.get(str(record["id"])); _mark(report, "migrationRuns", existing) + if dry: continue + row = {"id": record["id"], "source": record["source"], + "schema_version": record["schemaVersion"], "bundle_sha256": record["bundleSha256"], + "dry_run": bool(record.get("dryRun", False)), "status": record.get("status", "imported"), + "counts": record.get("counts", {}), "diagnostics": record.get("diagnostics", {}), + "created_at": _datetime(record.get("createdAt"))} + await _upsert(session, compat_migration_runs, row, existing) + + +async def export_bundle(store: PostgresPortStore, *, source_revision: str | None = None) -> dict[str, Any]: + async with store.sessions() as session: + hubs = (await session.execute(sa.select(compat_hubs).order_by(compat_hubs.c.id))).mappings().all() + manifests = (await session.execute(sa.select(compat_manifests).order_by(compat_manifests.c.id))).mappings().all() + consumers = (await session.execute(sa.select(compat_api_consumers).order_by(compat_api_consumers.c.id))).mappings().all() + keys = (await session.execute(sa.select(compat_api_keys).order_by(compat_api_keys.c.id))).mappings().all() + widgets = (await session.execute(sa.select(compat_widgets).order_by(compat_widgets.c.id))).mappings().all() + events = (await session.execute(sa.select(runtime_interaction_events).order_by(runtime_interaction_events.c.id))).mappings().all() + runs = (await session.execute(sa.select(compat_migration_runs).order_by(compat_migration_runs.c.id))).mappings().all() + records = { + "hubs": [_body(row, created=True, updated=True) for row in hubs], + "hubCapabilityManifests": [_body(row, created=True, updated=True) for row in manifests], + "apiConsumers": [_body(row, created=True) for row in consumers], + "apiKeys": [{"id": row["id"], "apiConsumerId": row["api_consumer_id"], "keyPrefix": row["key_prefix"], "keyHash": row["key_hash"], "scopes": row["scopes"], "status": row["status"], "createdAt": _timestamp(row["created_at"])} for row in keys], + "widgets": [_body(row, created=True) for row in widgets], + "interactionEvents": [{**dict((row["payload"] or {}).get("legacy") or {}), "id": row["id"], "widgetId": (row["subject_refs"] or {}).get("widget"), "eventType": row["event_type"], "occurredAt": _timestamp(row["occurred_at"]), "recordedAt": _timestamp(row["recorded_at"]), "correlationId": row["correlation_id"]} for row in events], + "migrationRuns": [{"id": row["id"], "source": row["source"], "schemaVersion": row["schema_version"], "bundleSha256": row["bundle_sha256"], "dryRun": row["dry_run"], "status": row["status"], "counts": row["counts"], "diagnostics": row["diagnostics"], "createdAt": _timestamp(row["created_at"])} for row in runs], + } + high_water = max((_timestamp(row["recorded_at"]) for row in events), default=None) + bundle: dict[str, Any] = {"schemaVersion": SCHEMA_VERSION, "source": "hub-core", "exportedAt": _timestamp(_now()), "highWaterMark": high_water, "records": records} + if source_revision: + bundle["sourceRevision"] = source_revision + bundle["bundleSha256"] = bundle_digest(bundle) + return bundle + + +def _body(row: Any, *, created: bool = False, updated: bool = False) -> dict[str, Any]: + value = dict(row["body"] or {}) + if created: + value["createdAt"] = _timestamp(row["created_at"]) + if updated: + value["updatedAt"] = _timestamp(row["updated_at"]) + return value + + +def load_bundle(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text()) + if not isinstance(value, dict): + raise ValueError("migration bundle must be a JSON object") + return value diff --git a/hub_core/runtime/postgres_store.py b/hub_core/runtime/postgres_store.py new file mode 100644 index 0000000..3d0d373 --- /dev/null +++ b/hub_core/runtime/postgres_store.py @@ -0,0 +1,321 @@ +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from copy import deepcopy +from datetime import datetime, timezone +from typing import Any +from uuid import UUID, uuid4 + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine + +from hub_core.contracts import CONTRACT_VERSION +from hub_core.runtime.models import ( + EventCommand, + MessageCommand, + PortAccepted, + PortCollection, + PortRecord, + Provenance, + RegistryRegistration, +) +from hub_core.runtime.tables import ( + runtime_audit_ledger, + runtime_interaction_events, + runtime_messages, + runtime_progress_events, + runtime_registrations, +) + + +class PostgresPortStore: + """Durable named-port store for PostgreSQL-compatible async engines.""" + + backend_name = "postgresql" + + def __init__(self, engine: AsyncEngine) -> None: + self.engine = engine + self.sessions = async_sessionmaker(engine, expire_on_commit=False) + + @classmethod + def from_url(cls, database_url: str) -> PostgresPortStore: + return cls(create_async_engine(database_url, pool_pre_ping=True)) + + async def readiness_checks(self) -> dict[str, str]: + try: + async with self.engine.connect() as connection: + await connection.execute(sa.text("SELECT 1")) + except Exception: + return {"database": "unavailable"} + return {"database": "ok"} + + async def aclose(self) -> None: + await self.engine.dispose() + + async def register_extension( + self, + registration: RegistryRegistration, + correlation_id: UUID, + ) -> PortAccepted: + hub_slug = str(registration.descriptor["hub_slug"]) + value = registration.model_dump(mode="json") + now = _now() + async with self.sessions.begin() as session: + current = ( + await session.execute( + sa.select(runtime_registrations.c.package).where( + runtime_registrations.c.hub_slug == hub_slug + ) + ) + ).scalar_one_or_none() + duplicate = current == value + if current is None: + await session.execute( + runtime_registrations.insert().values( + hub_slug=hub_slug, + package=value, + created_at=now, + updated_at=now, + ) + ) + elif not duplicate: + await session.execute( + runtime_registrations.update() + .where(runtime_registrations.c.hub_slug == hub_slug) + .values(package=value, updated_at=now) + ) + await self._audit( + session, + action="registry.duplicate" if duplicate else "registry.accepted", + subject_type="registration", + subject_id=hub_slug, + correlation_id=correlation_id, + value=value, + ) + return PortAccepted( + id=hub_slug, + status="duplicate" if duplicate else "accepted", + correlation_id=correlation_id, + ) + + async def send_message(self, command: MessageCommand) -> PortAccepted: + message_id = uuid4() + created_at = _now() + value = { + "id": str(message_id), + "created_at": created_at.isoformat(), + **command.model_dump(mode="json"), + } + stored = { + **value, + "created_at": created_at, + } + async with self.sessions.begin() as session: + await session.execute(runtime_messages.insert().values(**stored)) + await self._audit( + session, + action="message.accepted", + subject_type="message", + subject_id=str(message_id), + correlation_id=command.correlation_id, + value=value, + ) + return PortAccepted( + id=str(message_id), status="accepted", correlation_id=command.correlation_id + ) + + async def list_messages( + self, address: str, conversation_id: UUID | None + ) -> PortCollection: + statement = sa.select(runtime_messages).order_by( + runtime_messages.c.created_at, runtime_messages.c.id + ) + if conversation_id is not None: + statement = statement.where( + runtime_messages.c.conversation_id == str(conversation_id) + ) + async with self.sessions() as session: + rows = (await session.execute(statement)).mappings().all() + values = [ + _message_value(row) + for row in rows + if address in list(row["to_addresses"] or []) + ] + return PortCollection(items=[self._record("message", value) for value in values]) + + async def append_progress(self, command: EventCommand) -> PortAccepted: + return await self._append_event(command, runtime_progress_events, "progress") + + async def append_interaction(self, command: EventCommand) -> PortAccepted: + return await self._append_event(command, runtime_interaction_events, "interaction") + + async def query_projection(self, projection_id: str) -> PortRecord | None: + async with self.sessions() as session: + if projection_id == "hub_registry": + rows = ( + await session.execute( + sa.select(runtime_registrations.c.package).order_by( + runtime_registrations.c.hub_slug + ) + ) + ).scalars() + items = [deepcopy(value) for value in rows] + elif projection_id == "messages": + rows = ( + await session.execute( + sa.select(runtime_messages).order_by( + runtime_messages.c.created_at, runtime_messages.c.id + ) + ) + ).mappings() + items = [_message_value(row) for row in rows] + elif projection_id in {"progress_events", "interaction_events"}: + table = ( + runtime_progress_events + if projection_id == "progress_events" + else runtime_interaction_events + ) + family = "progress" if projection_id == "progress_events" else "interaction" + rows = ( + await session.execute( + sa.select(table).order_by(table.c.recorded_at, table.c.id) + ) + ).mappings() + items = [_event_value(row, family) for row in rows] + else: + return None + return self._record( + projection_id, + { + "projection_id": projection_id, + "items": items, + "rebuild_from": _rebuild_sources(projection_id), + }, + ) + + async def _append_event( + self, + command: EventCommand, + table: sa.Table, + family: str, + ) -> PortAccepted: + event_id = uuid4() + recorded_at = _now() + value = { + "id": str(event_id), + "family": family, + "recorded_at": recorded_at.isoformat(), + **command.model_dump(mode="json"), + } + stored = { + "id": str(event_id), + "schema_version": command.schema_version, + "correlation_id": str(command.correlation_id), + "event_type": command.event_type, + "occurred_at": command.occurred_at, + "subject_refs": command.subject_refs, + "payload": command.payload, + "recorded_at": recorded_at, + } + async with self.sessions.begin() as session: + await session.execute(table.insert().values(**stored)) + await self._audit( + session, + action=f"event.{family}.accepted", + subject_type=f"{family}_event", + subject_id=str(event_id), + correlation_id=command.correlation_id, + value=value, + ) + return PortAccepted( + id=str(event_id), status="accepted", correlation_id=command.correlation_id + ) + + async def _audit( + self, + session: Any, + *, + action: str, + subject_type: str, + subject_id: str, + correlation_id: UUID | None, + value: Mapping[str, Any], + ) -> None: + await session.execute( + runtime_audit_ledger.insert().values( + id=str(uuid4()), + action=action, + subject_type=subject_type, + subject_id=subject_id, + correlation_id=str(correlation_id) if correlation_id else None, + payload_hash=_hash(value), + detail={"schema_version": CONTRACT_VERSION}, + recorded_at=_now(), + ) + ) + + def _record(self, kind: str, value: dict[str, Any]) -> PortRecord: + record_id = str(value.get("id") or kind) + return PortRecord( + id=record_id, + data=deepcopy(value), + provenance=Provenance( + source_system="hub-core-postgresql", + source_ref=f"postgresql://hub_runtime/{kind}/{record_id}", + schema_version=CONTRACT_VERSION, + content_hash=_hash(value), + indexed_at=_now(), + ), + ) + + +def _message_value(row: Mapping[str, Any]) -> dict[str, Any]: + return { + "id": str(row["id"]), + "created_at": _iso(row["created_at"]), + "schema_version": row["schema_version"], + "correlation_id": str(row["correlation_id"]), + "conversation_id": str(row["conversation_id"]) if row["conversation_id"] else None, + "from_address": row["from_address"], + "to_addresses": list(row["to_addresses"] or []), + "body": row["body"], + "subject_refs": dict(row["subject_refs"] or {}), + } + + +def _event_value(row: Mapping[str, Any], family: str) -> dict[str, Any]: + return { + "id": str(row["id"]), + "family": family, + "recorded_at": _iso(row["recorded_at"]), + "schema_version": row["schema_version"], + "correlation_id": str(row["correlation_id"]), + "event_type": row["event_type"], + "occurred_at": _iso(row["occurred_at"]), + "subject_refs": dict(row["subject_refs"] or {}), + "payload": dict(row["payload"] or {}), + } + + +def _hash(value: Mapping[str, Any]) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode() + return hashlib.sha256(encoded).hexdigest() + + +def _iso(value: Any) -> str: + return value.isoformat() if isinstance(value, datetime) else str(value) + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _rebuild_sources(projection_id: str) -> list[str]: + return { + "hub_registry": ["runtime_registrations"], + "messages": ["runtime_messages"], + "progress_events": ["runtime_progress_events"], + "interaction_events": ["runtime_interaction_events"], + }[projection_id] diff --git a/hub_core/runtime/store.py b/hub_core/runtime/store.py index 25dfe75..c0e8447 100644 --- a/hub_core/runtime/store.py +++ b/hub_core/runtime/store.py @@ -26,6 +26,8 @@ class PortStore(Protocol): backend_name: str + async def readiness_checks(self) -> dict[str, str]: ... + async def register_extension( self, registration: RegistryRegistration, @@ -59,6 +61,9 @@ class InMemoryPortStore: self._progress_events: list[dict[str, Any]] = [] self._interaction_events: list[dict[str, Any]] = [] + async def readiness_checks(self) -> dict[str, str]: + return {"database": "not_applicable"} + async def register_extension( self, registration: RegistryRegistration, diff --git a/hub_core/runtime/tables.py b/hub_core/runtime/tables.py new file mode 100644 index 0000000..d268a6c --- /dev/null +++ b/hub_core/runtime/tables.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import sqlalchemy as sa + + +runtime_metadata = sa.MetaData() + +runtime_registrations = sa.Table( + "runtime_registrations", + runtime_metadata, + sa.Column("hub_slug", sa.String(120), primary_key=True), + sa.Column("package", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), +) + +runtime_messages = sa.Table( + "runtime_messages", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("schema_version", sa.String(40), nullable=False), + sa.Column("correlation_id", sa.String(36), nullable=False, index=True), + sa.Column("conversation_id", sa.String(36), nullable=True, index=True), + sa.Column("from_address", sa.String(240), nullable=False), + sa.Column("to_addresses", sa.JSON(), nullable=False), + sa.Column("body", sa.Text(), nullable=False), + sa.Column("subject_refs", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, index=True), +) + + +def _event_table(name: str) -> sa.Table: + return sa.Table( + name, + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("schema_version", sa.String(40), nullable=False), + sa.Column("correlation_id", sa.String(36), nullable=False, index=True), + sa.Column("event_type", sa.String(160), nullable=False, index=True), + sa.Column("occurred_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("subject_refs", sa.JSON(), nullable=False), + sa.Column("payload", sa.JSON(), nullable=False), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False, index=True), + ) + + +runtime_progress_events = _event_table("runtime_progress_events") +runtime_interaction_events = _event_table("runtime_interaction_events") + +runtime_audit_ledger = sa.Table( + "runtime_audit_ledger", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("action", sa.String(120), nullable=False, index=True), + sa.Column("subject_type", sa.String(80), nullable=False), + sa.Column("subject_id", sa.String(240), nullable=False, index=True), + sa.Column("correlation_id", sa.String(36), nullable=True, index=True), + sa.Column("payload_hash", sa.String(64), nullable=False), + sa.Column("detail", sa.JSON(), nullable=False), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False, index=True), +) + +compat_hubs = sa.Table( + "compat_hubs", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("slug", sa.String(120), nullable=False, unique=True, index=True), + sa.Column("name", sa.String(240), nullable=False), + sa.Column("domain", sa.String(240), nullable=True), + sa.Column("hub_kind", sa.String(80), nullable=True), + sa.Column("hub_family", sa.String(80), nullable=True), + sa.Column("vsm_function", sa.String(80), nullable=True), + sa.Column("vsm_system", sa.String(80), nullable=True), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("body", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), +) + +compat_manifests = sa.Table( + "compat_manifests", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("hub_id", sa.String(36), sa.ForeignKey("compat_hubs.id"), nullable=True, index=True), + sa.Column("hub_slug", sa.String(120), nullable=True, index=True), + sa.Column("manifest_version", sa.String(40), nullable=False), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("body", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), +) + +compat_api_consumers = sa.Table( + "compat_api_consumers", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("slug", sa.String(120), nullable=True, unique=True, index=True), + sa.Column("name", sa.String(240), nullable=False, index=True), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("hub_capability_manifest_id", sa.String(36), nullable=True, index=True), + sa.Column("rate_limit_per_minute", sa.Integer(), nullable=True), + sa.Column("quota_per_day", sa.Integer(), nullable=True), + sa.Column("key_prefix", sa.String(32), nullable=True), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("body", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), +) + +compat_api_keys = sa.Table( + "compat_api_keys", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column( + "api_consumer_id", + sa.String(36), + sa.ForeignKey("compat_api_consumers.id"), + nullable=False, + index=True, + ), + sa.Column("key_prefix", sa.String(32), nullable=False, index=True), + sa.Column("key_hash", sa.String(64), nullable=False, unique=True), + sa.Column("scopes", sa.Text(), nullable=True), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), +) + +compat_widgets = sa.Table( + "compat_widgets", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("hub_id", sa.String(36), sa.ForeignKey("compat_hubs.id"), nullable=False, index=True), + sa.Column("name", sa.String(240), nullable=False), + sa.Column("widget_type", sa.String(120), nullable=True, index=True), + sa.Column("capability_ref", sa.String(240), nullable=True, index=True), + sa.Column("view_context", sa.String(500), nullable=True), + sa.Column("policy_scope", sa.String(120), nullable=True), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("body", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), +) + +compat_migration_runs = sa.Table( + "compat_migration_runs", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("source", sa.String(120), nullable=False, index=True), + sa.Column("schema_version", sa.String(80), nullable=False), + sa.Column("bundle_sha256", sa.String(64), nullable=False, index=True), + sa.Column("dry_run", sa.Boolean(), nullable=False), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("counts", sa.JSON(), nullable=False), + sa.Column("diagnostics", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), +) + +runtime_import_runs = sa.Table( + "runtime_import_runs", + runtime_metadata, + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("source", sa.String(120), nullable=False, index=True), + sa.Column("schema_version", sa.String(80), nullable=False), + sa.Column("bundle_sha256", sa.String(64), nullable=False, unique=True, index=True), + sa.Column("high_water_mark", sa.String(200), nullable=True), + sa.Column("counts", sa.JSON(), nullable=False), + sa.Column("content_hashes", sa.JSON(), nullable=False), + sa.Column("status", sa.String(40), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), +) diff --git a/pyproject.toml b/pyproject.toml index 7783ece..07a2c57 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ hub-core = "hub_core.runtime.cli:main" [dependency-groups] dev = [ + "aiosqlite>=0.20.0", "pytest>=8.0.0", ] diff --git a/tests/fixtures/core-hub-migration.v1.json b/tests/fixtures/core-hub-migration.v1.json new file mode 100644 index 0000000..e757b41 --- /dev/null +++ b/tests/fixtures/core-hub-migration.v1.json @@ -0,0 +1,16 @@ +{ + "schemaVersion": "core-hub.migration.v1", + "source": "core-hub-test-fixture", + "sourceRevision": "fixture-revision", + "exportedAt": "2026-08-21T12:00:00Z", + "highWaterMark": "2026-08-21T11:00:00Z", + "records": { + "hubs": [{"id": "11111111-1111-4111-8111-111111111111", "slug": "ops-hub", "name": "Ops Hub", "createdAt": "2026-08-20T10:00:00Z", "updatedAt": "2026-08-20T11:00:00Z"}], + "hubCapabilityManifests": [{"id": "22222222-2222-4222-8222-222222222222", "hubId": "11111111-1111-4111-8111-111111111111", "manifestVersion": "1.0", "status": "active"}], + "apiConsumers": [{"id": "33333333-3333-4333-8333-333333333333", "slug": "ops-hub", "name": "ops-hub", "hubCapabilityManifestId": "22222222-2222-4222-8222-222222222222"}], + "apiKeys": [{"id": "44444444-4444-4444-8444-444444444444", "apiConsumerId": "33333333-3333-4333-8333-333333333333", "keyPrefix": "ch_fixture", "keyHash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "scopes": "framework:read", "status": "active"}], + "widgets": [{"id": "55555555-5555-4555-8555-555555555555", "hubId": "11111111-1111-4111-8111-111111111111", "name": "Readiness", "widgetType": "ops-readiness-gate"}], + "interactionEvents": [{"id": "66666666-6666-4666-8666-666666666666", "widgetId": "55555555-5555-4555-8555-555555555555", "eventType": "ops-endpoint-verified", "metadata": {"expectedStatus": 401}, "createdAt": "2026-08-21T11:00:00Z"}], + "migrationRuns": [{"id": "77777777-7777-4777-8777-777777777777", "source": "inter-hub", "schemaVersion": "core-hub.migration.v1", "bundleSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "dryRun": false, "status": "imported", "counts": {}, "diagnostics": {}, "createdAt": "2026-08-19T09:00:00Z"}] + } +} diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py new file mode 100644 index 0000000..4f802bb --- /dev/null +++ b/tests/test_compatibility.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import asyncio + +from fastapi.testclient import TestClient +from sqlalchemy.ext.asyncio import create_async_engine + +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.tables import runtime_metadata + +GROUPS = frozenset({"system", "registry", "credentials", "interaction", "deferred", "operator"}) +HEADERS = {"Authorization": "Bearer operator-token"} + + +def compatibility_client(tmp_path, *, write_groups: frozenset[str] = GROUPS) -> tuple[TestClient, PostgresPortStore]: + database_url = f"sqlite+aiosqlite:///{tmp_path / 'compat.db'}" + + async def create_schema() -> None: + engine = create_async_engine(database_url) + async with engine.begin() as connection: + await connection.run_sync(runtime_metadata.create_all) + await engine.dispose() + + asyncio.run(create_schema()) + store = PostgresPortStore.from_url(database_url) + settings = RuntimeSettings( + environment="production", + backend="postgresql", + allow_ephemeral=False, + database_url=database_url, + api_token="operator-token", + v2_groups=GROUPS, + v2_write_groups=write_groups, + legacy_health=True, + ) + return TestClient(create_app(settings=settings, port_store=store)), store + + +def test_route_group_configuration_rejects_two_writers() -> None: + try: + RuntimeSettings( + v2_groups=frozenset({"registry"}), + v2_write_groups=frozenset({"registry"}), + legacy_write_groups=frozenset({"registry"}), + ) + except ValueError as exc: + assert "write groups overlap" in str(exc) + else: + raise AssertionError("overlapping writers were accepted") + + +def test_system_compatibility_and_protected_auth(tmp_path) -> None: + client, store = compatibility_client(tmp_path) + + assert client.get("/healthz").json()["service"] == "core-hub" + assert client.get("/api/v2/widget-types").status_code == 200 + denied = client.get("/api/v2/hubs") + assert denied.status_code == 401 + assert denied.json()["detail"]["code"] == "unauthorized" + document = client.get("/api/v2/openapi.json").json() + assert "/api/v2/hubs" in document["paths"] + assert "/hubs" in document["paths"] + asyncio.run(store.aclose()) + + +def test_ops_hub_bootstrap_and_dynamic_key(tmp_path) -> None: + client, store = compatibility_client(tmp_path) + + hub = client.post( + "/api/v2/hubs", + headers=HEADERS, + json={"slug": "ops-hub", "name": "Ops Hub", "domain": "ops.coulomb.social"}, + ).json() + manifest = client.post( + "/api/v2/hub-capability-manifests", + headers=HEADERS, + json={"hubId": hub["id"], "manifestVersion": "1.0"}, + ).json() + activated = client.post( + f"/api/v2/hub-capability-manifests/{manifest['id']}/activate", headers=HEADERS + ) + assert activated.json()["status"] == "active" + consumer = client.post( + "/api/v2/api-consumers", + headers=HEADERS, + json={"name": "ops-hub", "hubCapabilityManifestId": manifest["id"]}, + ).json() + issued = client.post( + f"/api/v2/api-consumers/{consumer['id']}/api-keys", + headers=HEADERS, + json={"scopes": "framework:read hub:ops-hub:write"}, + ).json() + runtime_headers = {"Authorization": f"Bearer {issued['fullKey']}"} + widget = client.post( + "/api/v2/widgets", + headers=runtime_headers, + json={"hubId": hub["id"], "name": "Readiness", "widgetType": "ops-readiness-gate"}, + ).json() + event = client.post( + "/api/v2/interaction-events", + headers=runtime_headers, + json={ + "widgetId": widget["id"], + "eventType": "ops-endpoint-verified", + "metadata": {"expectedStatus": 401}, + }, + ) + + assert event.status_code == 201 + assert client.get("/api/v2/hub-registry", headers=runtime_headers).status_code == 200 + assert client.get("/api/v2/interaction-events", headers=runtime_headers).json()["count"] == 1 + assert "fullKey" not in str(client.get("/api/v2/api-consumers", headers=HEADERS).json()) + asyncio.run(store.aclose()) + + +def test_read_only_group_rejects_write(tmp_path) -> None: + client, store = compatibility_client(tmp_path, write_groups=frozenset()) + response = client.post( + "/api/v2/hubs", headers=HEADERS, json={"slug": "ops-hub", "name": "Ops Hub"} + ) + + assert response.status_code == 503 + assert response.json()["detail"] == "compatibility group is read-only" + asyncio.run(store.aclose()) diff --git a/tests/test_migration.py b/tests/test_migration.py new file mode 100644 index 0000000..0694a31 --- /dev/null +++ b/tests/test_migration.py @@ -0,0 +1,97 @@ +import json +import asyncio + +from sqlalchemy.ext.asyncio import create_async_engine + +from hub_core.runtime.migration import ( + bundle_digest, + export_bundle, + import_bundle, + validate_bundle, +) +from hub_core.runtime.postgres_store import PostgresPortStore +from hub_core.runtime.tables import runtime_metadata + + +def bundle() -> dict: + value = { + "schemaVersion": "core-hub.migration.v1", + "source": "core-hub", + "sourceRevision": "abc123", + "exportedAt": "2026-08-21T12:00:00Z", + "highWaterMark": "2026-08-21T11:00:00Z", + "records": { + "hubs": [{"id": "hub-1", "slug": "ops-hub", "name": "Ops Hub", "createdAt": "2026-08-20T10:00:00Z", "updatedAt": "2026-08-20T11:00:00Z"}], + "hubCapabilityManifests": [{"id": "manifest-1", "hubId": "hub-1", "manifestVersion": "1.0", "status": "active"}], + "apiConsumers": [{"id": "consumer-1", "slug": "ops-hub", "name": "ops-hub", "hubCapabilityManifestId": "manifest-1"}], + "apiKeys": [{"id": "key-1", "apiConsumerId": "consumer-1", "keyPrefix": "ch_prefix", "keyHash": "a" * 64, "scopes": "framework:read"}], + "widgets": [{"id": "widget-1", "hubId": "hub-1", "name": "Readiness", "widgetType": "ops-readiness-gate"}], + "interactionEvents": [{"id": "event-1", "widgetId": "widget-1", "eventType": "ops-endpoint-verified", "metadata": {"status": 401}, "createdAt": "2026-08-21T11:00:00Z"}], + "migrationRuns": [{"id": "run-1", "source": "inter-hub", "schemaVersion": "core-hub.migration.v1", "bundleSha256": "b" * 64, "dryRun": False, "status": "imported", "counts": {}, "diagnostics": {}, "createdAt": "2026-08-19T09:00:00Z"}], + }, + } + value["bundleSha256"] = bundle_digest(value) + return value + + +async def make_store(tmp_path) -> PostgresPortStore: + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'migration.db'}") + async with engine.begin() as connection: + await connection.run_sync(runtime_metadata.create_all) + return PostgresPortStore(engine) + + +def test_validation_rejects_secret_material_and_digest_drift(): + value = bundle() + value["records"]["apiKeys"][0]["fullKey"] = "must-not-move" + + report = validate_bundle(value) + + assert report["ok"] is False + assert any("secret-shaped" in error for error in report["errors"]) + assert any("bundleSha256" in error for error in report["errors"]) + + +def test_import_is_idempotent_and_exports_all_seven_collections(tmp_path): + async def run(): + store = await make_store(tmp_path) + try: + return ( + await import_bundle(store, bundle()), + await import_bundle(store, bundle()), + await export_bundle(store, source_revision="target-revision"), + ) + finally: + await store.aclose() + + first, second, exported = asyncio.run(run()) + + assert first["ok"] is True + assert first["counts"]["migrationRuns"]["created"] == 1 + assert second["idempotent"] is True + assert second["migrationRunId"] == first["migrationRunId"] + assert set(exported["records"]) == { + "hubs", "hubCapabilityManifests", "apiConsumers", "apiKeys", + "widgets", "interactionEvents", "migrationRuns", + } + assert exported["records"]["apiKeys"][0]["keyHash"] == "a" * 64 + assert "fullKey" not in json.dumps(exported) + assert validate_bundle(exported)["ok"] is True + + +def test_dry_run_does_not_record_import(tmp_path): + async def run(): + store = await make_store(tmp_path) + try: + return ( + await import_bundle(store, bundle(), dry_run=True), + await import_bundle(store, bundle()), + ) + finally: + await store.aclose() + + report, followup = asyncio.run(run()) + + assert report["dryRun"] is True + assert "migrationRunId" not in report + assert followup["idempotent"] is False diff --git a/tests/test_postgres_store.py b/tests/test_postgres_store.py new file mode 100644 index 0000000..62e8d81 --- /dev/null +++ b/tests/test_postgres_store.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import asyncio +import json +from datetime import datetime, timezone +from uuid import uuid4 + +from fastapi.testclient import TestClient +from sqlalchemy.ext.asyncio import create_async_engine + +from hub_core.contracts import extension_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.tables import runtime_audit_ledger, runtime_metadata + + +def test_durable_store_survives_reopen_and_keeps_event_families_separate(tmp_path) -> None: + database_url = f"sqlite+aiosqlite:///{tmp_path / 'runtime.db'}" + + async def create_schema() -> None: + engine = create_async_engine(database_url) + async with engine.begin() as connection: + await connection.run_sync(runtime_metadata.create_all) + await engine.dispose() + + asyncio.run(create_schema()) + package = json.loads( + extension_contract_root() + .joinpath("fixtures", "ops-hub.extension.json") + .read_text(encoding="utf-8") + ) + settings = RuntimeSettings( + environment="production", + backend="postgresql", + allow_ephemeral=False, + database_url=database_url, + ) + first_store = PostgresPortStore.from_url(database_url) + first = TestClient(create_app(settings=settings, port_store=first_store)) + + correlation_id = str(uuid4()) + assert first.get("/readyz").status_code == 200 + assert first.post( + "/ports/registry/registrations", + headers={"X-Correlation-ID": correlation_id}, + json=package, + ).status_code == 202 + assert first.post( + "/ports/events/progress", json=_event("hub.progress.recorded") + ).status_code == 202 + assert first.post( + "/ports/events/interaction", json=_event("hub.interaction.recorded") + ).status_code == 202 + asyncio.run(first_store.aclose()) + + second_store = PostgresPortStore.from_url(database_url) + second = TestClient(create_app(settings=settings, port_store=second_store)) + registry = second.get("/ports/projections/hub_registry").json() + progress = second.get("/ports/projections/progress_events").json() + interaction = second.get("/ports/projections/interaction_events").json() + + assert registry["data"]["items"][0]["descriptor"]["hub_slug"] == "ops-hub" + assert {item["family"] for item in progress["data"]["items"]} == {"progress"} + assert {item["family"] for item in interaction["data"]["items"]} == {"interaction"} + assert progress["provenance"]["source_system"] == "hub-core-postgresql" + + async def audit_count() -> int: + async with second_store.sessions() as session: + return len((await session.execute(runtime_audit_ledger.select())).all()) + + assert asyncio.run(audit_count()) == 3 + asyncio.run(second_store.aclose()) + + +def test_postgresql_readiness_fails_when_database_is_unavailable() -> None: + database_url = "sqlite+aiosqlite:////definitely-missing-parent/runtime.db" + store = PostgresPortStore.from_url(database_url) + settings = RuntimeSettings( + environment="production", + backend="postgresql", + allow_ephemeral=False, + database_url=database_url, + ) + response = TestClient(create_app(settings=settings, port_store=store)).get("/readyz") + + assert response.status_code == 503 + assert response.json()["checks"]["database"] == "unavailable" + asyncio.run(store.aclose()) + + +def _event(event_type: str) -> dict: + return { + "schema_version": "0.1.0", + "correlation_id": str(uuid4()), + "event_type": event_type, + "occurred_at": datetime.now(timezone.utc).isoformat(), + "subject_refs": {"hub": "ops-hub"}, + "payload": {"result": "ok"}, + } diff --git a/uv.lock b/uv.lock index 8b5baac..c8653e5 100644 --- a/uv.lock +++ b/uv.lock @@ -19,6 +19,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/67/cd/0d76dfc5de72bde52f55f53e925c7d152d9c7906634ec1e0cbc7e8d4ad93/aiofile-3.11.1-py3-none-any.whl", hash = "sha256:ce77d14ac07f77bc2b757834a5c129321f3f705c474593deed5ab209079a52c9", size = 20446 }, ] +[[package]] +name = "aiosqlite" +version = "0.22.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/4e/8a/64761f4005f17809769d23e518d915db74e6310474e733e3593cfc854ef1/aiosqlite-0.22.1.tar.gz", hash = "sha256:043e0bd78d32888c0a9ca90fc788b38796843360c855a7262a532813133a0650", size = 14821 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/00/b7/e3bf5133d697a08128598c8d0abc5e16377b51465a33756de24fa7dee953/aiosqlite-0.22.1-py3-none-any.whl", hash = "sha256:21c002eb13823fad740196c5a2e9d8e62f6243bd9e7e4a1f87fb5e44ecb4fceb", size = 17405 }, +] + [[package]] name = "alembic" version = "1.19.1" @@ -648,6 +657,7 @@ runtime = [ [package.dev-dependencies] dev = [ + { name = "aiosqlite" }, { name = "pytest" }, ] @@ -666,7 +676,10 @@ requires-dist = [ ] [package.metadata.requires-dev] -dev = [{ name = "pytest", specifier = ">=8.0.0" }] +dev = [ + { name = "aiosqlite", specifier = ">=0.20.0" }, + { name = "pytest", specifier = ">=8.0.0" }, +] [[package]] name = "idna" diff --git a/workplans/HUB-WP-0005-core-hub-absorption-execution.md b/workplans/HUB-WP-0005-core-hub-absorption-execution.md new file mode 100644 index 0000000..fb35e1e --- /dev/null +++ b/workplans/HUB-WP-0005-core-hub-absorption-execution.md @@ -0,0 +1,108 @@ +--- +id: HUB-WP-0005 +type: workplan +title: "Core Hub absorption execution" +domain: infotech +repo: hub-core +status: active +owner: codex +topic_slug: custodian +created: "2026-08-21" +updated: "2026-08-21" +parent_project: prj-state-hub-retirement +parent_workplan: CORE-WP-0010 +related: + - HUB-WP-0004 + - CORE-WP-0010 + - RAPPCOREHUB-WP-0002 +--- + +# Core Hub absorption execution + +## Goal + +Implement the accepted S0–S6 plan in `docs/core-hub-absorption-plan.md`, move +Core Hub compatibility and durable framework authority into hub-core without +concurrent production writers, and retain a tested rollback until the final +stabilization gate closes. + +## S0 durable runtime and audit boundary + +```task +id: HUB-WP-0005-T01 +status: done +priority: high +``` + +Add hub-core-owned PostgreSQL models/migrations, a durable `PortStore`, audit +ledger, dependency-aware readiness, and production-safe lifecycle. The memory +backend remains local/conformance only. + +## S0 compatibility shell and migration tooling + +```task +id: HUB-WP-0005-T02 +status: done +priority: high +``` + +Add legacy bearer authentication, non-overlapping route/write group controls, +versioned import/export/delta bundles, idempotent imports, canonical count/hash +comparison, and safe evidence. Never serialize a raw key or secret. + +## S1 system and discovery compatibility + +```task +id: HUB-WP-0005-T03 +status: done +priority: high +``` + +Implement the health/readiness, catalog, OpenAPI/docs, and compatibility alias +surface with normalized response comparison against Core Hub. + +## S2 and S3 registry and credential compatibility + +```task +id: HUB-WP-0005-T04 +status: done +priority: high +``` + +Absorb hubs, manifests/activation, hub registry, API consumers, key issuance, +and token compatibility with stable identities, audit, fail-closed auth, and +reverse-delta support. + +## S4 and S5 interaction and operator compatibility + +```task +id: HUB-WP-0005-T05 +status: done +priority: high +``` + +Absorb widgets and interaction events through the named interaction port, +preserve explicitly empty deferred collections, and rebuild the operator +surface without reads from the legacy schema. + +## S6 production cutover and retirement handoff + +```task +id: HUB-WP-0005-T06 +status: wait +priority: high +``` + +Publish immutable artifacts, pass all conformance and consumer gates, support +the operator-approved production writer cutover, exercise rollback, and hand +the seven-day stabilization/archive residual to CORE-WP-0010. Wait until +RAPPCOREHUB-WP-0002 provides the private candidate and route controls. + +## Acceptance + +- [x] Production readiness requires durable PostgreSQL and auth dependencies +- [x] Import is idempotent; counts, identities, hashes, and provenance compare +- [x] All 22 checked-in Core Hub OpenAPI paths have governed dispositions +- [ ] ops-hub and activity-core gates pass on hub-core +- [ ] Production runs the immutable hub-core image with one writer per group +- [ ] Rollback is exercised and stabilization residual is recorded