Close HUB-WP-0009 conformance gaps (C2, C7, C9, C10); mark blocked workplans
Some checks are pending
CI Smoke / host-smoke (push) Waiting to run
CI Smoke / pytest-smoke (push) Waiting to run

Implements the four residual conformance checks left open by the T04
minimal vertical:

- C2: GET /ports/registry/registrations/{hub_slug} resolves missing (404),
  ambiguous (shared reuse_surface_id across hub_slugs), and stale
  (deprecated/retired descriptor) registrations; a new .../audit route
  exposes queryable registration history from the existing in-memory
  history and the PostgreSQL runtime_audit_ledger.
- C7: harness proof that disabled compatibility groups deny access
  (404) with no fixture credentials involved, matching the existing
  fail-closed compat router behavior.
- C9: harness proof plus a dedicated test that /readyz degrades only on
  an unavailable configured dependency while unrelated disabled
  projections stay non-blocking.
- C10: ContractValidator now negotiates contract_version_min/max against
  the runtime's contract version and rejects incompatible or inverted
  ranges with an explicit 422 instead of silently accepting them.

HUB-WP-0009 is now finished. HUB-WP-0006 is marked blocked: its only open
task (T06) has no remaining hub-core code path and waits on an external
Forgejo identity/production deployment gate. HUB-WP-0011 is marked
blocked: T02/T03 already waited on external credential/deployment
review, and T01 needs a source/destination ownership and retention
decision against live message data before it can be implemented safely.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 310936@bnt-lap001
Assistant-Session: 00cd9abe-09a0-416b-88e0-f907b9101629
This commit is contained in:
tegwick 2026-09-27 23:59:35 +02:00
parent b0e89592c6
commit e89d621f18
11 changed files with 496 additions and 20 deletions

View file

@ -163,8 +163,105 @@ class ConformanceHarness:
),
)
)
ambiguous_slug = f"{self.package['descriptor']['hub_slug']}-conformance-ambiguous"
ambiguous_package = _with_hub_slug(self.package, ambiguous_slug)
self.target.post(
"/ports/registry/registrations",
headers={"X-Correlation-ID": str(uuid4())},
json=ambiguous_package,
)
resolution = self.target.get(
f"/ports/registry/registrations/{self.package['descriptor']['hub_slug']}"
)
missing = self.target.get("/ports/registry/registrations/hub-that-was-never-registered")
audit = self.target.get(
f"/ports/registry/registrations/{self.package['descriptor']['hub_slug']}/audit"
)
results.append(
self._check(
"C2",
2,
"registry resolution reports missing, ambiguous, and audited registrations",
lambda: _assert_registry_resolution(resolution, missing, audit, ambiguous_slug),
)
)
results.append(
self._check(
"C7",
2,
"disabled compatibility groups deny access without needing fixture credentials",
self._raw_port_policy_denies_by_default,
)
)
results.append(
self._check(
"C9",
2,
"readiness reports each dependency without blocking on unrelated disabled groups",
self._dependency_aware_readiness,
)
)
incompatible_package = _with_contract_version_range(
self.package, "9.9.9", "9.9.9", hub_slug=f"{ambiguous_slug}-incompatible"
)
incompatible_response = self.target.post(
"/ports/registry/registrations",
headers={"X-Correlation-ID": str(uuid4())},
json=incompatible_package,
)
results.append(
self._check(
"C10",
2,
"registration outside the runtime's supported contract range is rejected",
lambda: _assert_status_and_incompatibility(incompatible_response),
)
)
return ConformanceReport(contract_version=CONTRACT_VERSION, checks=tuple(results))
def _raw_port_policy_denies_by_default(self) -> None:
read_only_group = self.target.get("/api/v2/hubs")
operator_console = self.target.get("/console")
if read_only_group.status_code != 404:
raise AssertionError(
"disabled registry compatibility group did not deny access "
f"(status {read_only_group.status_code})"
)
if operator_console.status_code != 404:
raise AssertionError(
"disabled operator compatibility group did not deny access "
f"(status {operator_console.status_code})"
)
def _dependency_aware_readiness(self) -> None:
response = self.target.get("/readyz")
if response.status_code not in {200, 503}:
raise AssertionError(f"/readyz returned unexpected status {response.status_code}")
body = response.json()
checks = body.get("checks", {})
dependency_keys = ("database", "repo_manager_projection", "workload_projection")
dependency_values = {}
for required in dependency_keys:
if required not in checks:
raise AssertionError(f"readiness checks missing '{required}'")
if checks[required] not in {"ok", "stale", "unavailable", "not_applicable"}:
raise AssertionError(f"readiness check '{required}' has unexpected value")
dependency_values[required] = checks[required]
if checks.get("authorization") not in {"ok", "unavailable"}:
raise AssertionError("readiness check 'authorization' has unexpected value")
all_non_blocking = checks.get("authorization") == "ok" and all(
value in {"ok", "not_applicable"} for value in dependency_values.values()
)
if all_non_blocking and body.get("status") != "ok":
raise AssertionError("readiness reported degraded with no failing dependency")
if not all_non_blocking and body.get("status") != "degraded":
raise AssertionError("readiness reported ok while a dependency is unavailable or stale")
def _schema_validate(self) -> None:
_validator(self.schema_root.joinpath("hub-descriptor.schema.json")).validate(
self.package["descriptor"]
@ -318,6 +415,49 @@ def _assert_projection_rebuild(
raise AssertionError("message authority fixture missing from message projection")
def _with_hub_slug(package: Mapping[str, Any], hub_slug: str) -> dict[str, Any]:
descriptor = {**package["descriptor"], "hub_slug": hub_slug}
return {**package, "descriptor": descriptor}
def _with_contract_version_range(
package: Mapping[str, Any],
version_min: str,
version_max: str,
*,
hub_slug: str,
) -> dict[str, Any]:
descriptor = {
**package["descriptor"],
"hub_slug": hub_slug,
"contract_version_min": version_min,
"contract_version_max": version_max,
}
return {**package, "descriptor": descriptor}
def _assert_registry_resolution(
resolution: ResponseLike,
missing: ResponseLike,
audit: ResponseLike,
ambiguous_slug: str,
) -> None:
_expect_status(missing, 404, "unregistered hub_slug resolution")
_expect_status(resolution, 200, "registered hub_slug resolution")
data = resolution.json().get("data", {})
if data.get("resolution") != "ambiguous":
raise AssertionError("second registration sharing reuse_surface_id was not flagged ambiguous")
if ambiguous_slug not in data.get("ambiguous_with", []):
raise AssertionError("ambiguous resolution did not name the conflicting hub_slug")
_expect_status(audit, 200, "registration audit history")
if not audit.json().get("items"):
raise AssertionError("registration audit history is empty")
def _assert_status_and_incompatibility(response: ResponseLike) -> None:
_expect_status(response, 422, "out-of-range contract version registration")
def _projection_items(response: ResponseLike, projection_id: str) -> list[dict[str, Any]]:
_expect_status(response, 200, f"{projection_id} projection")
items = response.json().get("data", {}).get("items")

View file

@ -47,6 +47,36 @@ def create_ports_router() -> APIRouter:
raise HTTPException(status_code=422, detail=str(exc)) from exc
return await store.register_extension(body, x_correlation_id)
@router.get(
"/registry/registrations/{hub_slug}",
response_model=PortRecord,
tags=["registry"],
openapi_extra={"x-port-id": "port.registry", "x-direction": "out"},
)
async def resolve_registration(
hub_slug: str,
store: PortStore = Depends(get_port_store),
) -> PortRecord:
resolved = await store.resolve_registration(hub_slug)
if resolved is None:
raise HTTPException(status_code=404, detail=f"Registration '{hub_slug}' not found")
return resolved
@router.get(
"/registry/registrations/{hub_slug}/audit",
response_model=PortCollection,
tags=["registry"],
openapi_extra={"x-port-id": "port.registry", "x-direction": "out"},
)
async def registration_audit(
hub_slug: str,
store: PortStore = Depends(get_port_store),
) -> PortCollection:
audit = await store.list_registration_audit(hub_slug)
if not audit.items:
raise HTTPException(status_code=404, detail=f"No audit history for '{hub_slug}'")
return audit
@router.get(
"/messaging/messages",
response_model=PortCollection,

View file

@ -22,6 +22,7 @@ from hub_core.runtime.models import (
RegistryRegistration,
)
from hub_core.runtime.repository_navigation import NavigationProjection
from hub_core.runtime.store import _resolve_registration_record
from hub_core.runtime.workload_projection import WorkloadProjection
from hub_core.runtime.tables import (
compat_api_keys,
@ -443,6 +444,45 @@ class PostgresPortStore:
)
)
async def resolve_registration(self, hub_slug: str) -> PortRecord | None:
async with self.sessions() as session:
rows = (
await session.execute(
sa.select(runtime_registrations.c.hub_slug, runtime_registrations.c.package)
)
).all()
registrations = {row.hub_slug: row.package for row in rows}
value = registrations.get(hub_slug)
if value is None:
return None
return _resolve_registration_record("hub-core-postgresql", hub_slug, value, registrations)
async def list_registration_audit(self, hub_slug: str) -> PortCollection:
async with self.sessions() as session:
rows = (
await session.execute(
sa.select(runtime_audit_ledger)
.where(
runtime_audit_ledger.c.subject_type == "registration",
runtime_audit_ledger.c.subject_id == hub_slug,
)
.order_by(runtime_audit_ledger.c.recorded_at, runtime_audit_ledger.c.id)
)
).mappings()
entries = [
{
"id": str(row["id"]),
"action": row["action"],
"hub_slug": hub_slug,
"correlation_id": row["correlation_id"],
"recorded_at": _iso(row["recorded_at"]),
}
for row in rows
]
return PortCollection(
items=[self._record("registration_audit", entry) for entry in entries]
)
async def _append_event(
self,
command: EventCommand,

View file

@ -64,6 +64,10 @@ class PortStore(Protocol):
self, *, checked_at: datetime, diagnostic: Mapping[str, Any]
) -> None: ...
async def resolve_registration(self, hub_slug: str) -> PortRecord | None: ...
async def list_registration_audit(self, hub_slug: str) -> PortCollection: ...
class InMemoryPortStore:
"""Deterministic ephemeral backend for local runtime and conformance tests.
@ -77,6 +81,7 @@ class InMemoryPortStore:
def __init__(self) -> None:
self._lock = asyncio.Lock()
self._registrations: dict[str, dict[str, Any]] = {}
self._registration_audit: dict[str, list[dict[str, Any]]] = {}
self._messages: list[dict[str, Any]] = []
self._progress_events: list[dict[str, Any]] = []
self._interaction_events: list[dict[str, Any]] = []
@ -96,12 +101,36 @@ class InMemoryPortStore:
async with self._lock:
duplicate = self._registrations.get(hub_slug) == value
self._registrations[hub_slug] = deepcopy(value)
self._registration_audit.setdefault(hub_slug, []).append(
{
"id": str(uuid4()),
"action": "registry.duplicate" if duplicate else "registry.accepted",
"hub_slug": hub_slug,
"correlation_id": str(correlation_id),
"recorded_at": _now().isoformat(),
}
)
return PortAccepted(
id=hub_slug,
status="duplicate" if duplicate else "accepted",
correlation_id=correlation_id,
)
async def resolve_registration(self, hub_slug: str) -> PortRecord | None:
async with self._lock:
value = self._registrations.get(hub_slug)
if value is None:
return None
registrations = deepcopy(self._registrations)
return _resolve_registration_record("hub-core-memory", hub_slug, value, registrations)
async def list_registration_audit(self, hub_slug: str) -> PortCollection:
async with self._lock:
entries = deepcopy(self._registration_audit.get(hub_slug, []))
return PortCollection(
items=[self._record("registration_audit", entry) for entry in entries]
)
async def send_message(self, command: MessageCommand) -> PortAccepted:
message_id = uuid4()
value = {
@ -250,6 +279,48 @@ def _now() -> datetime:
return datetime.now(timezone.utc)
def _resolve_registration_record(
source_system: str,
hub_slug: str,
value: Mapping[str, Any],
registrations: Mapping[str, Mapping[str, Any]],
) -> PortRecord:
descriptor = value["descriptor"]
reuse_surface_id = descriptor.get("reuse_surface_id")
ambiguous_with = sorted(
other_slug
for other_slug, other_value in registrations.items()
if other_slug != hub_slug
and other_value["descriptor"].get("reuse_surface_id") == reuse_surface_id
)
stale = descriptor.get("status") in {"deprecated", "retired"}
if ambiguous_with:
resolution = "ambiguous"
elif stale:
resolution = "stale"
else:
resolution = "ok"
data = {
"hub_slug": hub_slug,
"resolution": resolution,
"ambiguous_with": ambiguous_with,
"descriptor": deepcopy(dict(descriptor)),
"manifest": deepcopy(dict(value["manifest"])),
}
encoded = json.dumps(data, sort_keys=True, separators=(",", ":")).encode()
return PortRecord(
id=hub_slug,
data=data,
provenance=Provenance(
source_system=source_system,
source_ref=f"{source_system.replace('hub-core-', '')}://registration/{hub_slug}",
schema_version=CONTRACT_VERSION,
content_hash=hashlib.sha256(encoded).hexdigest(),
indexed_at=_now(),
),
)
def _rebuild_sources(projection_id: str) -> list[str]:
return {
"hub_registry": ["hub_descriptors", "hub_manifests"],

View file

@ -5,14 +5,15 @@ from typing import Any
from jsonschema import Draft202012Validator, FormatChecker
from hub_core.contracts import extension_contract_root
from hub_core.contracts import CONTRACT_VERSION, extension_contract_root
from hub_core.runtime.models import RegistryRegistration
class ContractValidator:
"""Validate runtime registration input against the packaged contract."""
def __init__(self) -> None:
def __init__(self, *, runtime_contract_version: str = CONTRACT_VERSION) -> None:
self._runtime_contract_version = _parse_semver(runtime_contract_version)
contract_root = extension_contract_root()
schema_root = contract_root.joinpath("schemas")
self._descriptor = _validator(schema_root.joinpath("hub-descriptor.schema.json"))
@ -31,6 +32,21 @@ class ContractValidator:
manifest_id = registration.manifest.get("reuse_surface_id")
if descriptor_id != manifest_id:
raise ValueError("descriptor and manifest reuse_surface_id must match")
self._negotiate_contract_version(registration.descriptor)
def _negotiate_contract_version(self, descriptor: dict[str, Any]) -> None:
version_min = _parse_semver(descriptor["contract_version_min"])
version_max = _parse_semver(descriptor["contract_version_max"])
if version_min > version_max:
raise ValueError(
"descriptor contract_version_min must not exceed contract_version_max"
)
if not (version_min <= self._runtime_contract_version <= version_max):
raise ValueError(
"descriptor requires contract version range "
f"{descriptor['contract_version_min']}-{descriptor['contract_version_max']}, "
f"incompatible with runtime contract version {CONTRACT_VERSION}"
)
def validate_event_family(self, event_type: str, expected_family: str) -> None:
actual_family = self._event_families.get(event_type)
@ -46,3 +62,9 @@ def _validator(resource: Any) -> Draft202012Validator:
schema = json.loads(resource.read_text(encoding="utf-8"))
Draft202012Validator.check_schema(schema)
return Draft202012Validator(schema, format_checker=FormatChecker())
def _parse_semver(value: str) -> tuple[int, int, int]:
core = value.split("+", 1)[0].split("-", 1)[0]
major, minor, patch = core.split(".")
return (int(major), int(minor), int(patch))