From 6fb5ce285c4d2cf94a389e9ef7d1fc10f6be1268 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 5 Sep 2026 10:37:09 +0200 Subject: [PATCH] feat: add guarded State Hub inbox read projection pilot Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ed7-828d-7ca0-a8d4-0c3e5a0c4102 --- docs/statehub-inbox-read-pilot.md | 67 ++++++++++ hub_core/runtime/app.py | 3 + hub_core/runtime/config.py | 6 + hub_core/runtime/inbox_projection.py | 125 ++++++++++++++++++ tests/test_inbox_projection.py | 93 +++++++++++++ .../HUB-WP-0010-statehub-inbox-read-pilot.md | 49 +++++++ 6 files changed, 343 insertions(+) create mode 100644 docs/statehub-inbox-read-pilot.md create mode 100644 hub_core/runtime/inbox_projection.py create mode 100644 tests/test_inbox_projection.py create mode 100644 workplans/HUB-WP-0010-statehub-inbox-read-pilot.md diff --git a/docs/statehub-inbox-read-pilot.md b/docs/statehub-inbox-read-pilot.md new file mode 100644 index 0000000..f15c9f4 --- /dev/null +++ b/docs/statehub-inbox-read-pilot.md @@ -0,0 +1,67 @@ +# State Hub inbox read pilot + +`GET /ports/projections/statehub-inbox` is an opt-in, authenticated read-only +projection over the existing `agent_messages` model. It is distinct from +`/ports/messaging/messages`, which accepts new message envelopes and does not +implement State Hub's read/archive lifecycle. + +Enable only after validating and importing a consistent source snapshot: +`HUB_CORE_STATEHUB_INBOX_READS=1`, `HUB_CORE_STATEHUB_INBOX_AGENT=state-hub`. +PostgreSQL and the configured operator bearer token are required. Disabled by +default. No POST/PATCH/DELETE exists on the pilot endpoint. Imported history +never emits messages or notifications. + +The response is the State Hub `MessageRead` list shape. Parameters are +`to_agent` (must match the configured pilot agent), exact `from_agent`, +`unread_only` and `limit` (1–1000, default 50). Broadcast messages are included; +archived messages are excluded. Results order by created_at descending, then ID +for ties. Query transactions are read-only. Responses carry `Cache-Control: +no-store` and `X-Hub-Core-Read-Mode: snapshot-pilot`. + +This proves one literal agent inbox. It does not claim repository rename/alias +resolution, unscoped fleet reads, thread lookup, continuous freshness, or writer +cutover. The importer preserves thread roots so later history work need not +reconstruct dangling identities. + +## Snapshot and import + +Capture messages with one source PostgreSQL read-only, repeatable-read +transaction. Preserve id, sender, recipient, subject, body, thread_id, read_at, +archived_at and created_at, including archived/thread-root records. Keep the +raw snapshot in private temporary storage; commit only hashes and counts. + +The bundle has schema `hub-core.statehub-inbox-snapshot.v1`, `source`, +`captured_at`, `messages`, `count`, and `content_hash`. Use `snapshot_hash` from +`hub_core.runtime.inbox_projection` for canonical UTC/UUID normalization. +`import_snapshot(session, bundle, expected_hash=..., apply=False)` validates; +`apply=True` imports under a caller-owned transaction. The caller must commit +only on success and roll back any exception. PostgreSQL apply locks the target +message table before inspecting it. Different existing content refuses all +changes; exact repeats are no-ops. Thread roots insert before replies. Duplicate +IDs, malformed rows, hash/count mismatch, absent provenance, missing roots and +cycles are rejected. + +Use the existing workload database credential delivery inside the receiving +pod; never export database credentials into evidence. No schema migration or +new secret is required. This is a one-time migration helper, not a message +sending API or an automated synchronization loop. + +## Pilot sequence and rollback + +1. Keep State Hub as sole message writer; capture source revision, instance, + repeatable-read snapshot, data hash and expected filtered inbox results. +2. Import only into an empty receiver, verify all row hashes/counts and replay + as a no-op. Deploy the disabled reader image, then enable the private pilot. +3. Compare the same frozen source rows with authenticated receiver responses: + all/unread/sender/limit cases, broadcast inclusion and archive exclusion. + Check unauthenticated rejection and absence of writes. A later live State + Hub read is a freshness observation, not the original snapshot oracle. +4. Disable the reader flag and verify the endpoint returns 404; re-enable and + recheck the same hash. Retain the imported rows for review; do not delete + history as a rollback technique. Runtime rollback uses the previous image. + +Transport from State Hub must be explicitly admitted by rapp-core-hub's +NetworkPolicy. Scope any pilot exception to State Hub API pod labels and the +candidate port. No public Ingress or production message-writer change is part +of this pilot. The operator bearer stays in its existing credential lane; +production caller credentials and ongoing freshness require their own cutover. diff --git a/hub_core/runtime/app.py b/hub_core/runtime/app.py index 5e9efbd..5f94b18 100644 --- a/hub_core/runtime/app.py +++ b/hub_core/runtime/app.py @@ -138,6 +138,9 @@ def create_app( # Exact projection routes precede the generic /projections/{projection_id} # route so Starlette dispatch cannot shadow them. + if resolved_settings.statehub_inbox_reads: + from hub_core.runtime.inbox_projection import create_inbox_projection_router + app.include_router(create_inbox_projection_router()) app.include_router(create_workload_projection_router()) app.include_router(create_ports_router()) app.include_router(create_repository_navigation_router()) diff --git a/hub_core/runtime/config.py b/hub_core/runtime/config.py index ca04e5e..a4ed940 100644 --- a/hub_core/runtime/config.py +++ b/hub_core/runtime/config.py @@ -36,8 +36,12 @@ class RuntimeSettings: v2_write_groups: frozenset[str] = frozenset() legacy_write_groups: frozenset[str] = frozenset() legacy_health: bool = False + statehub_inbox_reads: bool = False + statehub_inbox_agent: str = "state-hub" def __post_init__(self) -> None: + if self.statehub_inbox_reads and (self.backend != "postgresql" or not self.api_token): + raise ValueError("State Hub inbox reads require PostgreSQL and operator token") if self.repo_manager_timeout_seconds <= 0: raise ValueError("Repo Manager timeout must be positive") if self.repo_projection_refresh_seconds < 0: @@ -81,6 +85,8 @@ class RuntimeSettings: 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), + statehub_inbox_reads=_env_bool("HUB_CORE_STATEHUB_INBOX_READS", False), + statehub_inbox_agent=os.getenv("HUB_CORE_STATEHUB_INBOX_AGENT", "state-hub"), ) def readiness_checks(self, store_backend: str) -> dict[str, str]: diff --git a/hub_core/runtime/inbox_projection.py b/hub_core/runtime/inbox_projection.py new file mode 100644 index 0000000..d7ce9d2 --- /dev/null +++ b/hub_core/runtime/inbox_projection.py @@ -0,0 +1,125 @@ +"""Opt-in, authenticated State Hub inbox read pilot; no message writers.""" +from __future__ import annotations + +import hashlib +import hmac +import json +from datetime import datetime, timezone +from typing import Any + +from fastapi import APIRouter, Header, HTTPException, Query, Request, Response +from sqlalchemy import select, text + +from hub_core.models.agent_message import AgentMessage +from hub_core.schemas.agent_message import MessageRead + +SCHEMA = "hub-core.statehub-inbox-snapshot.v1" + + +def normalize_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows = [] + for row in messages: + model = MessageRead.model_validate(row) + for field in ("created_at", "read_at", "archived_at"): + value = getattr(model, field) + if value is not None: + # SQLite test adapters return naive UTC; PostgreSQL retains tzinfo. + setattr(model, field, value.replace(tzinfo=timezone.utc) if value.tzinfo is None + else value.astimezone(timezone.utc)) + rows.append(model.model_dump(mode="json")) + if len({row["id"] for row in rows}) != len(rows): + raise ValueError("duplicate message IDs") + return sorted(rows, key=lambda row: row["id"]) + + +def snapshot_hash(messages: list[dict[str, Any]]) -> str: + return hashlib.sha256(json.dumps(normalize_messages(messages), sort_keys=True, + separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() + + +async def import_snapshot(session, bundle: dict, *, expected_hash: str, apply: bool = False) -> dict: + """Import into an empty receiver only; exact retries are read-only no-ops. + + Caller owns the transaction. This function never sends notifications, changes + existing message rows, or enables a runtime route. + """ + if bundle.get("schema") != SCHEMA: + raise ValueError("unsupported inbox snapshot schema") + rows = normalize_messages(bundle["messages"]) + digest = snapshot_hash(rows) + if digest != expected_hash or digest != bundle.get("content_hash"): + raise ValueError("snapshot hash mismatch") + if len(rows) != bundle.get("count"): + raise ValueError("snapshot count mismatch") + if not bundle.get("source") or not bundle.get("captured_at"): + raise ValueError("snapshot provenance is required") + ids = {row["id"] for row in rows} + if any(row["thread_id"] and row["thread_id"] not in ids for row in rows): + raise ValueError("snapshot must include all thread roots") + # Lock before inspecting emptiness so concurrent imports cannot both pass. + if apply and session.bind.dialect.name == "postgresql": + await session.execute(text("LOCK TABLE agent_messages IN EXCLUSIVE MODE")) + existing = list((await session.execute(select(AgentMessage))).scalars()) + if existing: + current = [MessageRead.model_validate(row).model_dump(mode="json") for row in existing] + if snapshot_hash(current) != digest: + raise ValueError("receiver contains different messages; refusing overwrite") + return {"status": "noop", "count": len(rows), "content_hash": digest} + remaining = {row["id"]: row for row in rows} + inserted: set[str] = set() + levels: list[list[dict]] = [] + while remaining: + level = [row for row in remaining.values() if not row["thread_id"] or row["thread_id"] in inserted] + if not level: + raise ValueError("cyclic message threading") + levels.append(level) + for row in level: + inserted.add(row["id"]) + del remaining[row["id"]] + if apply: + for level in levels: + session.add_all([AgentMessage(**MessageRead.model_validate(row).model_dump()) for row in level]) + await session.flush() + return {"status": "applied" if apply else "validated", "count": len(rows), "content_hash": digest} + + +def create_inbox_projection_router() -> APIRouter: + router = APIRouter() + + @router.get("/ports/projections/statehub-inbox", response_model=list[MessageRead], tags=["projections"]) + async def inbox( + request: Request, + response: Response, + to_agent: str = "state-hub", + from_agent: str | None = None, + unread_only: bool = False, + limit: int = Query(50, ge=1, le=1000), + authorization: str | None = Header(None), + ): + settings = request.app.state.settings + token = settings.api_token + supplied = (authorization or "").removeprefix("Bearer ") + if not token or not (authorization or "").startswith("Bearer ") or not hmac.compare_digest(supplied, token): + raise HTTPException(401, "inbox pilot requires operator bearer authentication", + headers={"WWW-Authenticate": "Bearer"}) + if to_agent != settings.statehub_inbox_agent: + raise HTTPException(422, "inbox pilot is restricted to its configured agent") + query = select(AgentMessage).where( + AgentMessage.archived_at.is_(None), + (AgentMessage.to_agent == to_agent) | (AgentMessage.to_agent == "broadcast"), + ) + if from_agent: + query = query.where(AgentMessage.from_agent == from_agent) + if unread_only: + query = query.where(AgentMessage.read_at.is_(None)) + query = query.order_by(AgentMessage.created_at.desc(), AgentMessage.id).limit(limit) + async with request.app.state.port_store.sessions() as session: + # The pilot may never mutate its imported source history. + if session.bind.dialect.name == "postgresql": + await session.execute(text("SET TRANSACTION READ ONLY")) + rows = list((await session.execute(query)).scalars()) + response.headers["Cache-Control"] = "no-store" + response.headers["X-Hub-Core-Read-Mode"] = "snapshot-pilot" + return rows + + return router diff --git a/tests/test_inbox_projection.py b/tests/test_inbox_projection.py new file mode 100644 index 0000000..c1bc4c6 --- /dev/null +++ b/tests/test_inbox_projection.py @@ -0,0 +1,93 @@ +import asyncio +from datetime import datetime, timezone +from uuid import uuid4 + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker + +from hub_core.models.agent_message import AgentMessage +from hub_core.runtime.app import create_app +from hub_core.runtime.config import RuntimeSettings +from hub_core.runtime.inbox_projection import SCHEMA, import_snapshot, snapshot_hash +from hub_core.runtime.postgres_store import PostgresPortStore +from hub_core.runtime.store import InMemoryPortStore + + +def message(**extra): + return dict(id=str(uuid4()), from_agent="sender", to_agent="state-hub", subject="Subject", + body="Preserved body", thread_id=None, read_at=None, archived_at=None, + created_at="2026-09-05T08:00:00Z", **extra) + + +def bundle(rows): + return dict(schema=SCHEMA, source="state-hub/primary", captured_at="2026-09-05T08:01:00Z", + messages=rows, count=len(rows), content_hash=snapshot_hash(rows)) + + +def test_snapshot_import_and_read_contract(tmp_path): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'inbox.db'}") + factory = async_sessionmaker(engine, expire_on_commit=False) + root = message() + read = {**message(), "read_at": "2026-09-05T08:00:01Z", "thread_id": root["id"]} + archived = {**message(), "archived_at": "2026-09-05T08:00:02Z"} + broadcast = {**message(), "to_agent": "broadcast"} + other = {**message(), "to_agent": "other"} + data = bundle([root, read, archived, broadcast, other]) + + async def prepare(): + async with engine.begin() as conn: + await conn.run_sync(AgentMessage.__table__.create) + async with factory.begin() as session: + result = await import_snapshot(session, data, expected_hash=data['content_hash'], apply=True) + assert result['status'] == 'applied' + async with factory.begin() as session: + assert (await import_snapshot(session, data, expected_hash=data['content_hash'], apply=True))['status'] == 'noop' + async with factory.begin() as session: + changed = bundle([root]) + with pytest.raises(ValueError, match='refusing overwrite'): + await import_snapshot(session, changed, expected_hash=changed['content_hash'], apply=True) + asyncio.run(prepare()) + settings = RuntimeSettings(backend='postgresql', statehub_inbox_reads=True, api_token='test-reader') + client = TestClient(create_app(settings=settings, port_store=PostgresPortStore(engine))) + url='/ports/projections/statehub-inbox' + headers={'Authorization':'Bearer test-reader'} + assert client.get(url).status_code == 401 + assert client.get(url,headers={'Authorization':'Bearer wrong'}).status_code == 401 + response=client.get(url,headers=headers) + assert response.status_code == 200 + assert {r['id'] for r in response.json()} == {root['id'],read['id'],broadcast['id']} + assert [r['id'] for r in response.json()] == sorted([root['id'],read['id'],broadcast['id']]) + assert len(client.get(url,headers=headers,params={'unread_only':True}).json()) == 2 + assert client.get(url,headers=headers,params={'from_agent':'missing'}).json() == [] + assert len(client.get(url,headers=headers,params={'limit':1}).json()) == 1 + assert client.get(url,headers=headers,params={'to_agent':'other'}).status_code == 422 + assert client.post(url,headers=headers,json={}).status_code == 405 + assert response.headers['x-hub-core-read-mode'] == 'snapshot-pilot' + asyncio.run(engine.dispose()) + + +def test_disabled_and_unconfigured_reader(): + c=TestClient(create_app(settings=RuntimeSettings(),port_store=InMemoryPortStore())) + assert c.get('/ports/projections/statehub-inbox').status_code == 404 + with pytest.raises(ValueError,match='PostgreSQL and operator token'): + RuntimeSettings(statehub_inbox_reads=True) + + +@pytest.mark.parametrize('defect',['hash','duplicate','missing_root','cycle']) +def test_invalid_snapshot_fails_before_database_access(defect): + row=message() + data=bundle([row]) + if defect=='hash': data['content_hash']='wrong' + if defect=='duplicate': data['messages'].append(row) + if defect=='missing_root': + row['thread_id']=str(uuid4()); data=bundle([row]) + if defect=='cycle': + row['thread_id']=row['id']; data=bundle([row]) + class EmptySession: + async def execute(self,*args): + class Result: + def scalars(self): return [] + return Result() + with pytest.raises(ValueError): + asyncio.run(import_snapshot(EmptySession(),data,expected_hash=data['content_hash'])) diff --git a/workplans/HUB-WP-0010-statehub-inbox-read-pilot.md b/workplans/HUB-WP-0010-statehub-inbox-read-pilot.md new file mode 100644 index 0000000..6b19424 --- /dev/null +++ b/workplans/HUB-WP-0010-statehub-inbox-read-pilot.md @@ -0,0 +1,49 @@ +--- +id: HUB-WP-0010 +type: workplan +title: "State Hub inbox read projection pilot" +domain: infotech +repo: hub-core +status: active +owner: codex +topic_slug: infotech +created: "2026-09-05" +updated: "2026-09-05" +related: + - STATE-WP-0079 +quality_dor: DoR-Ok +quality_dor_at: "2026-09-05" +quality_dor_by: codex +quality_dor_note: >- + STATE-WP-0079-T08 requests one read-only inbox parity proof. The runtime has + an empty agent_messages table and an operator token; named-port messages + are a distinct envelope contract. Scope excludes production message writers. +--- + +## Deliver the opt-in receiver contract + +```task +id: HUB-WP-0010-T01 +status: progress +priority: high +``` + +Add an authenticated, GET-only `/ports/projections/statehub-inbox` pilot for one +configured agent, preserving message payloads, broadcast inclusion, sender and +unread filters, archive exclusion, ordering and limits. Default disabled. +State Hub remains the writer. Existing `/ports/messaging/messages` is unchanged. + +## Import and verify one consistent historical snapshot + +```task +id: HUB-WP-0010-T02 +status: todo +priority: high +``` + +Validate source/count/hash/thread closure before importing to an empty receiver. +Preserve all IDs/timestamps/flags. Exact retries are no-ops; different existing +rows refuse overwrite. Verify parity over a fixed source snapshot with read-only +queries, auth failures and rollback by disabling the pilot. Packaging/deployment +belongs to rapp-core-hub. Do not claim continuous freshness or switch clients +from the sole production message writer based on a one-time snapshot.