diff --git a/api/main.py b/api/main.py index 17e938c..83dd63f 100644 --- a/api/main.py +++ b/api/main.py @@ -15,6 +15,7 @@ from api.services.write_idempotency import WriteIdempotencyMiddleware from api.routers import decisions, extension_points, intake, ops_runs, progress, state, suggestions, tasks, technical_debt, topics, workstreams, workstream_dependencies from api.routers import domains, repos, contributions, sbom, policy, domain_goals, repo_goals, messages, memo_notifications, capability_requests, tpsc, services from api.routers import token_events +from api.routers import message_feed from api.routers import interface_changes from api.routers import flows from api.routers import recently_on_scope @@ -129,6 +130,8 @@ app.include_router(domain_goals.router) app.include_router(repo_goals.router) app.include_router(contributions.router) app.include_router(sbom.router) +# Before messages.router so /messages/feed is not captured by /messages/{id}. +app.include_router(message_feed.router) app.include_router(messages.router) app.include_router(memo_notifications.router) app.include_router(capability_requests.router) diff --git a/api/models/__init__.py b/api/models/__init__.py index 0ce0f5d..61910c8 100644 --- a/api/models/__init__.py +++ b/api/models/__init__.py @@ -23,7 +23,7 @@ from api.models.technical_debt import TechnicalDebt, TDStatus from api.models.contribution import Contribution, ContributionType, ContributionStatus from api.models.sbom_snapshot import SBOMSnapshot from api.models.sbom_entry import SBOMEntry, Ecosystem -from api.models.agent_message import AgentMessage, MessageReceipt +from api.models.agent_message import AgentMessage, MessageReceipt, MessageTombstone from api.models.memo_notification import MemoNotificationReceipt from api.models.capability_catalog import CapabilityCatalog from api.models.capability_request import CapabilityRequest diff --git a/api/models/agent_message.py b/api/models/agent_message.py index 2ef5390..058eff6 100644 --- a/api/models/agent_message.py +++ b/api/models/agent_message.py @@ -1,7 +1,19 @@ import uuid from datetime import datetime -from sqlalchemy import DateTime, ForeignKey, Index, String, Text, text +from sqlalchemy import ( + DDL, + BigInteger, + DateTime, + FetchedValue, + ForeignKey, + Index, + Sequence, + String, + Text, + event, + text, +) from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -11,12 +23,28 @@ from api.models.base import Base, new_uuid BROADCAST = "broadcast" MESSAGE_KINDS = ("message", "news", "standing") +# STATE-WP-0098-T05: inbox change feed (docs/inbox-change-feed-contract.md). +# ``revision`` is owned by the database trigger below; application code never +# sets it. Every inbox write transaction gets exactly one revision. +INBOX_REVISION_SEQ = Sequence("inbox_revision_seq", metadata=Base.metadata) + + +def _revision_column() -> Mapped[int]: + return mapped_column( + BigInteger, + nullable=False, + index=True, + server_default=FetchedValue(), + server_onupdate=FetchedValue(), + ) + class AgentMessage(Base): __tablename__ = "agent_messages" __table_args__ = ( Index("ix_agent_messages_to_kind_archived", "to_agent", "kind", "archived_at"), ) + __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=new_uuid @@ -55,6 +83,7 @@ class AgentMessage(Base): server_default=text("now()"), nullable=False, ) + revision: Mapped[int] = _revision_column() thread_root: Mapped["AgentMessage | None"] = relationship( "AgentMessage", @@ -72,6 +101,7 @@ class MessageReceipt(Base): """ __tablename__ = "message_receipts" + __mapper_args__ = {"eager_defaults": True} message_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), @@ -88,3 +118,103 @@ class MessageReceipt(Base): acknowledged_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) + revision: Mapped[int] = _revision_column() + + +class MessageTombstone(Base): + """A deleted ``agent_messages`` row, written by the delete trigger. + + Bodies are never kept. Served by the inbox feed within the 90-day horizon + (contract D1). + """ + + __tablename__ = "message_tombstones" + + message_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True) + revision: Mapped[int] = mapped_column(BigInteger, nullable=False, index=True) + deleted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + reason: Mapped[str | None] = mapped_column(Text, nullable=True) + + +# Trigger DDL for ``Base.metadata.create_all`` (the test suite). Production +# gets the same objects from migration c4d5e6f7a8b9_inbox_feed_revision, which +# keeps its own frozen copy; tests/test_message_feed.py checks they agree. +INBOX_ADVISORY_LOCK_KEY = 5_348_490_098 + +INBOX_FEED_FUNCTIONS_SQL = f""" +CREATE OR REPLACE FUNCTION inbox_feed_revision() RETURNS bigint +LANGUAGE plpgsql AS $fn$ +DECLARE + tx text := txid_current()::text; + cached text := coalesce(current_setting('statehub.inbox_rev', true), ''); + rev bigint; +BEGIN + PERFORM pg_advisory_xact_lock({INBOX_ADVISORY_LOCK_KEY}); + IF split_part(cached, ':', 1) = tx THEN + RETURN split_part(cached, ':', 2)::bigint; + END IF; + rev := nextval('inbox_revision_seq'); + PERFORM set_config('statehub.inbox_rev', tx || ':' || rev::text, true); + RETURN rev; +END +$fn$; + +CREATE OR REPLACE FUNCTION inbox_feed_stamp_revision() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + IF TG_OP = 'UPDATE' THEN + IF NEW IS NOT DISTINCT FROM OLD THEN + RETURN NEW; + END IF; + END IF; + NEW.revision := inbox_feed_revision(); + RETURN NEW; +END +$fn$; + +CREATE OR REPLACE FUNCTION inbox_feed_tombstone() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + INSERT INTO message_tombstones (message_id, revision, deleted_at, reason) + VALUES ( + OLD.id, + inbox_feed_revision(), + now(), + nullif(current_setting('statehub.delete_reason', true), '') + ) + ON CONFLICT (message_id) DO UPDATE + SET revision = EXCLUDED.revision, + deleted_at = EXCLUDED.deleted_at, + reason = EXCLUDED.reason; + RETURN OLD; +END +$fn$; +""" + +event.listen( + AgentMessage.__table__, + "after_create", + DDL( + INBOX_FEED_FUNCTIONS_SQL + + """ +CREATE TRIGGER agent_messages_inbox_revision + BEFORE INSERT OR UPDATE ON agent_messages + FOR EACH ROW EXECUTE FUNCTION inbox_feed_stamp_revision(); +CREATE TRIGGER agent_messages_inbox_tombstone + BEFORE DELETE ON agent_messages + FOR EACH ROW EXECUTE FUNCTION inbox_feed_tombstone(); +""" + ), +) +event.listen( + MessageReceipt.__table__, + "after_create", + DDL( + INBOX_FEED_FUNCTIONS_SQL + + """ +CREATE TRIGGER message_receipts_inbox_revision + BEFORE INSERT OR UPDATE ON message_receipts + FOR EACH ROW EXECUTE FUNCTION inbox_feed_stamp_revision(); +""" + ), +) diff --git a/api/routers/message_feed.py b/api/routers/message_feed.py new file mode 100644 index 0000000..f9bd490 --- /dev/null +++ b/api/routers/message_feed.py @@ -0,0 +1,277 @@ +"""Inbox change feed (STATE-WP-0098-T05). + +Read-only pull feed for derived inbox copies (hub-core). Spec: +docs/inbox-change-feed-contract.md §2, §3, §5. Revisions are assigned by the +database trigger from migration c4d5e6f7a8b9; one write transaction = one +revision, and an advisory lock makes commit order equal revision order, so the +maximum visible revision is a safe watermark. +""" +from __future__ import annotations + +import hashlib +import json +import uuid +from datetime import datetime +from typing import Any + +from fastapi import APIRouter, Depends, Query, Response +from fastapi.responses import JSONResponse +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from api.config import settings +from api.database import get_session +from api.schemas.inbox_feed import ( + CONTRACT, + CursorError, + FeedChange, + FeedPage, + FeedSnapshot, + ForeignCursorError, + decode_cursor, + encode_cursor, +) + +router = APIRouter(prefix="/messages/feed", tags=["messages", "inbox-feed"]) + +# Contract D1: tombstones are served for 90 days. +TOMBSTONE_HORIZON_DAYS = 90 +DEFAULT_LIMIT = 500 +MAX_LIMIT = 5000 +NO_STORE = {"Cache-Control": "no-store"} + +_MESSAGE_COLUMNS = ( + "id, from_agent, to_agent, subject, body, thread_id, read_at, archived_at, " + "kind, expires_at, supersedes_id, created_at, revision" +) +_RECEIPT_COLUMNS = "message_id, agent, delivered_at, read_at, acknowledged_at, revision" +_TOMBSTONE_COLUMNS = "message_id, revision, deleted_at, reason" + +# Revisions <= this value carry tombstones that fell out of the horizon; a +# cursor below it can no longer be served losslessly (410). +_COMPACTED_SQL = f""" +SELECT coalesce(max(revision), 0) FROM message_tombstones +WHERE deleted_at < now() - interval '{TOMBSTONE_HORIZON_DAYS} days' +""" +_HIGH_WATER_SQL = """ +SELECT greatest( + (SELECT coalesce(max(revision), 0) FROM agent_messages), + (SELECT coalesce(max(revision), 0) FROM message_receipts), + (SELECT coalesce(max(revision), 0) FROM message_tombstones) +) +""" +_REVISIONS_AFTER_SQL = """ +SELECT revision FROM agent_messages WHERE revision > :after +UNION ALL +SELECT revision FROM message_receipts WHERE revision > :after +UNION ALL +SELECT revision FROM message_tombstones WHERE revision > :after AND revision > :compacted +""" + + +def _jsonable(value: Any) -> Any: + if isinstance(value, uuid.UUID): + return str(value) + if isinstance(value, datetime): + return value.isoformat() + return value + + +def _row(mapping) -> dict[str, Any]: + return {key: _jsonable(value) for key, value in mapping.items()} + + +def receipt_key(state: dict[str, Any]) -> str: + return f"{state['message_id']}:{state['agent']}" + + +def content_hash(messages, receipts, tombstones) -> str: + payload = json.dumps( + {"messages": messages, "receipts": receipts, "tombstones": tombstones}, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(payload.encode()).hexdigest() + + +async def _begin_consistent_read(session: AsyncSession) -> None: + """Start one REPEATABLE READ READ ONLY transaction for the whole response.""" + await session.connection( + execution_options={"isolation_level": "REPEATABLE READ", "postgresql_readonly": True} + ) + + +async def _scalar(session: AsyncSession, sql: str, **params) -> Any: + return (await session.execute(text(sql), params)).scalar_one() + + +async def _rows(session: AsyncSession, sql: str, **params) -> list[dict[str, Any]]: + return [_row(m) for m in (await session.execute(text(sql), params)).mappings().all()] + + +@router.get("/snapshot", response_model=FeedSnapshot) +async def feed_snapshot( + response: Response, session: AsyncSession = Depends(get_session) +) -> FeedSnapshot: + response.headers.update(NO_STORE) + await _begin_consistent_read(session) + source_time = await _scalar(session, "SELECT now()") + compacted = await _scalar(session, _COMPACTED_SQL) + high_water = await _scalar(session, _HIGH_WATER_SQL) + messages = await _rows( + session, f"SELECT {_MESSAGE_COLUMNS} FROM agent_messages ORDER BY id" + ) + receipts = await _rows( + session, + f"SELECT {_RECEIPT_COLUMNS} FROM message_receipts ORDER BY message_id, agent", + ) + tombstones = await _rows( + session, + f"SELECT {_TOMBSTONE_COLUMNS} FROM message_tombstones " + "WHERE revision > :compacted ORDER BY message_id", + compacted=compacted, + ) + return FeedSnapshot( + source_instance=settings.state_hub_instance_label, + source_time=source_time, + cursor=encode_cursor(high_water), + compacted_before=encode_cursor(compacted), + messages=messages, + receipts=receipts, + tombstones=tombstones, + count=len(messages), + content_hash=content_hash(messages, receipts, tombstones), + ) + + +def _error(status_code: int, detail: str, **extra) -> JSONResponse: + return JSONResponse( + status_code=status_code, + content={"contract": CONTRACT, "detail": detail, **extra}, + headers=NO_STORE, + ) + + +@router.get( + "", + response_model=FeedPage, + responses={400: {}, 409: {}, 410: {}}, +) +async def feed_page( + response: Response, + after: str = Query(..., description="Cursor from a snapshot or previous page"), + limit: int = Query(DEFAULT_LIMIT, ge=1, le=MAX_LIMIT), + source_instance: str | None = Query( + None, description="source_instance the cursor came from; 409 if foreign" + ), + session: AsyncSession = Depends(get_session), +): + response.headers.update(NO_STORE) + instance = settings.state_hub_instance_label + try: + after_rev = decode_cursor(after) + except ForeignCursorError as exc: + return _error(409, str(exc), resnapshot_required=True) + except CursorError as exc: + return _error(400, str(exc)) + if source_instance is not None and source_instance != instance: + return _error( + 409, + f"cursor from source_instance {source_instance!r}, this is {instance!r}", + resnapshot_required=True, + source_instance=instance, + ) + + await _begin_consistent_read(session) + source_time = await _scalar(session, "SELECT now()") + compacted = await _scalar(session, _COMPACTED_SQL) + high_water = await _scalar(session, _HIGH_WATER_SQL) + if after_rev > high_water: + # A cursor from the future: another instance, or a restored source. + return _error( + 409, + "cursor is ahead of this source's high_water; re-snapshot", + resnapshot_required=True, + source_instance=instance, + high_water=encode_cursor(high_water), + ) + if after_rev < compacted: + return _error( + 410, + "cursor is older than the tombstone horizon; re-snapshot", + resnapshot_required=True, + source_instance=instance, + compacted_before=encode_cursor(compacted), + ) + + params = {"after": after_rev, "compacted": compacted} + boundary = await _scalar( + session, + f"SELECT max(revision) FROM (SELECT revision FROM ({_REVISIONS_AFTER_SQL}) r " + "ORDER BY revision LIMIT :limit) page", + limit=limit, + **params, + ) + changes: list[FeedChange] = [] + has_more = False + if boundary is not None: + # A page never splits a revision: take every change up to the boundary. + window = {**params, "upto": boundary} + for state in await _rows( + session, + f"SELECT {_MESSAGE_COLUMNS} FROM agent_messages " + "WHERE revision > :after AND revision <= :upto " + "ORDER BY revision, created_at, id", + **window, + ): + changes.append( + FeedChange( + revision=state["revision"], entity="message", + key=state["id"], op="upsert", state=state, + ) + ) + for state in await _rows( + session, + f"SELECT {_RECEIPT_COLUMNS} FROM message_receipts " + "WHERE revision > :after AND revision <= :upto " + "ORDER BY revision, message_id, agent", + **window, + ): + changes.append( + FeedChange( + revision=state["revision"], entity="receipt", + key=receipt_key(state), op="upsert", state=state, + ) + ) + for state in await _rows( + session, + f"SELECT {_TOMBSTONE_COLUMNS} FROM message_tombstones " + "WHERE revision > :after AND revision > :compacted AND revision <= :upto " + "ORDER BY revision, message_id", + **window, + ): + changes.append( + FeedChange( + revision=state["revision"], entity="tombstone", + key=state["message_id"], op="delete", state=state, + ) + ) + # Stable sort: revision order; within a revision messages, receipts, tombstones. + changes.sort(key=lambda change: change.revision) + has_more = ( + await _scalar( + session, + f"SELECT exists(SELECT 1 FROM ({_REVISIONS_AFTER_SQL}) r " + "WHERE revision > :upto)", + **{**params, "upto": boundary}, + ) + ) + return FeedPage( + source_instance=instance, + source_time=source_time, + high_water=encode_cursor(high_water), + next_cursor=encode_cursor(boundary if boundary is not None else after_rev), + has_more=has_more, + compacted_before=encode_cursor(compacted), + changes=changes, + ) diff --git a/api/routers/messages.py b/api/routers/messages.py index eba28ea..99056dc 100644 --- a/api/routers/messages.py +++ b/api/routers/messages.py @@ -303,7 +303,9 @@ async def archive_message( session: AsyncSession = Depends(get_session), ) -> AgentMessage: message = await _get_message(message_id, session) - message.archived_at = datetime.now(timezone.utc) + # First write wins (inbox feed contract §6): re-archiving keeps archived_at. + if message.archived_at is None: + message.archived_at = datetime.now(timezone.utc) # Broadcast archive is a global withdrawal; it no longer stamps read_at. if message.read_at is None and not is_broadcast(message): message.read_at = message.archived_at diff --git a/api/schemas/inbox_feed.py b/api/schemas/inbox_feed.py new file mode 100644 index 0000000..8c78b0b --- /dev/null +++ b/api/schemas/inbox_feed.py @@ -0,0 +1,68 @@ +"""Inbox change-feed envelopes (docs/inbox-change-feed-contract.md, v1).""" +from __future__ import annotations + +import re +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel + +CONTRACT = "statehub.inbox-feed.v1" +CURSOR_VERSION = "v1" + +_CURSOR_RE = re.compile(r"^(v\d+):(\d+)$") + + +class CursorError(ValueError): + """Malformed cursor (400).""" + + +class ForeignCursorError(ValueError): + """Cursor of another contract version (409).""" + + +def encode_cursor(revision: int) -> str: + return f"{CURSOR_VERSION}:{int(revision)}" + + +def decode_cursor(cursor: str) -> int: + match = _CURSOR_RE.match(cursor or "") + if match is None: + raise CursorError(f"malformed cursor {cursor!r}") + if match.group(1) != CURSOR_VERSION: + raise ForeignCursorError( + f"cursor {cursor!r} belongs to another contract version; re-snapshot" + ) + return int(match.group(2)) + + +class FeedChange(BaseModel): + revision: int + entity: Literal["message", "receipt", "tombstone"] + key: str + op: Literal["upsert", "delete"] + state: dict[str, Any] + + +class FeedPage(BaseModel): + contract: str = CONTRACT + source_instance: str | None + source_time: datetime + high_water: str + next_cursor: str + has_more: bool + compacted_before: str + changes: list[FeedChange] + + +class FeedSnapshot(BaseModel): + contract: str = CONTRACT + source_instance: str | None + source_time: datetime + cursor: str + compacted_before: str + messages: list[dict[str, Any]] + receipts: list[dict[str, Any]] + tombstones: list[dict[str, Any]] + count: int + content_hash: str diff --git a/docs/evidence/legacy-meter-weekly-review-20261014.json b/docs/evidence/legacy-meter-weekly-review-20261014.json new file mode 100644 index 0000000..ff8ac99 --- /dev/null +++ b/docs/evidence/legacy-meter-weekly-review-20261014.json @@ -0,0 +1,803 @@ +{ + "captured_at": "2026-10-13T22:00:09.478808+00:00", + "api_base": "http://127.0.0.1:8000", + "workplan": "STATE-WP-0070", + "retired_interfaces": [], + "weekly_review": { + "generated_at": "2026-10-13T22:00:09.598862Z", + "window_start": "2026-10-06T22:00:09.404805Z", + "window_end": "2026-10-13T22:00:09.404805Z", + "cadence": "weekly", + "activity_core_handoff": { + "activity_id": "statehub-legacy-interface-review", + "cadence": "weekly", + "source_endpoint": "/legacy-meter/weekly-review", + "state_owner": "state-hub", + "scheduler_owner": "activity-core" + }, + "interfaces": [ + { + "interface": { + "id": "8051fb40-66c9-4dc4-a413-13e8573e04a3", + "interface_key": "event_subject:org.statehub.workstream.completed", + "interface_kind": "event_subject", + "legacy_since": "2026-06-04T06:09:42.198193Z", + "replacement_ref": "org.statehub.workplan.completed", + "owner_component": "state-hub.events", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "Retired by STATE-WP-0069 closeout", + "retired_at": "2026-08-31T08:55:06.871333Z", + "created_at": "2026-06-04T06:09:42.198193Z", + "updated_at": "2026-08-31T08:55:06.756087Z" + }, + "all_time": { + "calls": 247, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 247 + }, + "users": { + "unknown": 247 + }, + "components": { + "state-hub.events": 247 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-07-13T13:22:00.748140Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "cd9342fd-27e7-4a46-a1be-d41499e106a3", + "interface_key": "rest_api:DELETE /workstreams/{workstream_id}", + "interface_kind": "rest_api", + "legacy_since": "2026-06-06T17:28:13.052813Z", + "replacement_ref": "/workplans/{workplan_id}", + "owner_component": "state-hub.api", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "Retired by STATE-WP-0069 closeout", + "retired_at": "2026-08-31T08:55:08.200322Z", + "created_at": "2026-06-06T17:28:13.052813Z", + "updated_at": "2026-08-31T08:55:08.089714Z" + }, + "all_time": { + "calls": 1, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 1 + }, + "users": { + "unknown": 1 + }, + "components": { + "unknown": 1 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-06-06T17:28:13.047529Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "d0294dce-7913-4268-94c3-1b2502aefb0f", + "interface_key": "rest_api:GET /sbom/", + "interface_kind": "rest_api", + "legacy_since": "2026-08-22T17:06:38.703608Z", + "replacement_ref": "sbom-nexus:/sbom/", + "owner_component": "state-hub.sbom-compat", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "STATE-WP-0079-T05: founder-authorized retirement 2026-09-22; seven-day gate passed, replacement verified.", + "retired_at": "2026-09-22T09:22:56.503646Z", + "created_at": "2026-08-22T17:06:38.703608Z", + "updated_at": "2026-09-22T09:22:56.498576Z" + }, + "all_time": { + "calls": 7, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 7 + }, + "users": { + "unknown": 7 + }, + "components": { + "unknown": 7 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-09-11T15:49:01.762688Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "2f77ac12-8c97-4974-a2f9-2996927982d1", + "interface_key": "rest_api:GET /sbom/report/licences/", + "interface_kind": "rest_api", + "legacy_since": "2026-08-22T17:06:41.978426Z", + "replacement_ref": "sbom-nexus:/sbom/", + "owner_component": "state-hub.sbom-compat", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "STATE-WP-0079-T05: user-authorized retirement 2026-09-05; fresh seven-day gate passed, replacement_verified=true; direct SBOM Nexus endpoints returned HTTP 200 from the State Hub pod. Compatibility storage/routes retained for the final cutover.", + "retired_at": "2026-09-05T08:04:48.423727Z", + "created_at": "2026-08-22T17:06:41.978426Z", + "updated_at": "2026-09-05T08:04:48.418993Z" + }, + "all_time": { + "calls": 5, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 5 + }, + "users": { + "unknown": 5 + }, + "components": { + "unknown": 5 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-08-28T18:46:10.642164Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "b627edf5-3302-4366-ba2e-547d32f634b5", + "interface_key": "rest_api:GET /sbom/snapshots/", + "interface_kind": "rest_api", + "legacy_since": "2026-08-22T17:06:37.057774Z", + "replacement_ref": "sbom-nexus:/sbom/", + "owner_component": "state-hub.sbom-compat", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "STATE-WP-0079-T05: user-authorized retirement 2026-09-05; fresh seven-day gate passed, replacement_verified=true; direct SBOM Nexus endpoints returned HTTP 200 from the State Hub pod. Compatibility storage/routes retained for the final cutover.", + "retired_at": "2026-09-05T08:04:48.506447Z", + "created_at": "2026-08-22T17:06:37.057774Z", + "updated_at": "2026-09-05T08:04:48.501706Z" + }, + "all_time": { + "calls": 7, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 7 + }, + "users": { + "unknown": 7 + }, + "components": { + "unknown": 7 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-08-28T18:46:10.694185Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "d03ee274-cd6f-4cd5-8f22-e4d298416299", + "interface_key": "rest_api:GET /sbom/snapshots/{snapshot_id}", + "interface_kind": "rest_api", + "legacy_since": "2026-08-22T17:23:45.969876Z", + "replacement_ref": "sbom-nexus:/sbom/", + "owner_component": "state-hub.sbom-compat", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "Retired by STATE-WP-0069 closeout", + "retired_at": "2026-08-31T08:55:08.470106Z", + "created_at": "2026-08-22T17:23:45.969876Z", + "updated_at": "2026-08-31T08:55:08.456139Z" + }, + "all_time": { + "calls": 1, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 1 + }, + "users": { + "unknown": 1 + }, + "components": { + "unknown": 1 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-08-22T17:23:45.967492Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "a6cb291e-8312-42f8-8a66-fde369482881", + "interface_key": "rest_api:GET /sbom/{repo_slug}", + "interface_kind": "rest_api", + "legacy_since": "2026-08-22T17:23:45.089542Z", + "replacement_ref": "sbom-nexus:/sbom/", + "owner_component": "state-hub.sbom-compat", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "Retired by STATE-WP-0069 closeout", + "retired_at": "2026-08-31T08:55:08.754914Z", + "created_at": "2026-08-22T17:23:45.089542Z", + "updated_at": "2026-08-31T08:55:08.750826Z" + }, + "all_time": { + "calls": 1, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 1 + }, + "users": { + "unknown": 1 + }, + "components": { + "unknown": 1 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-08-22T17:23:45.670511Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "cad13f19-ddb1-49b9-86d0-8789b7599f3a", + "interface_key": "rest_api:GET /tasks/?workstream_id", + "interface_kind": "rest_api", + "legacy_since": "2026-08-29T10:01:31.546213Z", + "replacement_ref": "/tasks/?workplan_id=", + "owner_component": "state-hub.api", + "status": "legacy", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": null, + "retired_at": null, + "created_at": "2026-08-29T10:01:31.546213Z", + "updated_at": "2026-08-29T10:01:31.546213Z" + }, + "all_time": { + "calls": 53, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 53 + }, + "users": { + "unknown": 53 + }, + "components": { + "unknown": 53 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-10-04T18:32:36.539797Z", + "retirement_candidate": true, + "retirement_reason": "no usage in review window; quiet 9d (>= 7d required for 53 all-time call(s))" + }, + { + "interface": { + "id": "3e4a3d0b-08fa-45c3-91e2-4f514a914b97", + "interface_key": "rest_api:GET /workstreams/", + "interface_kind": "rest_api", + "legacy_since": "2026-06-04T00:26:14.533764Z", + "replacement_ref": "/workplans/", + "owner_component": "state-hub.api", + "status": "legacy", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": null, + "retired_at": null, + "created_at": "2026-06-04T00:26:14.533764Z", + "updated_at": "2026-06-04T00:26:14.533764Z" + }, + "all_time": { + "calls": 830269, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 830269 + }, + "users": { + "unknown": 830269 + }, + "components": { + "unknown": 830269 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-08-21T13:50:37.994459Z", + "retirement_candidate": false, + "retirement_reason": "quiet 53d of 60d required for 830269 all-time call(s)" + }, + { + "interface": { + "id": "e4c140ec-7479-46f9-b185-4b533f19d639", + "interface_key": "rest_api:GET /workstreams/workplan-index", + "interface_kind": "rest_api", + "legacy_since": "2026-06-04T05:20:58.321869Z", + "replacement_ref": "/workplans/index", + "owner_component": "state-hub.api", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "Retired by STATE-WP-0069 closeout", + "retired_at": "2026-08-31T08:55:08.939572Z", + "created_at": "2026-06-04T05:20:58.321869Z", + "updated_at": "2026-08-31T08:55:08.933482Z" + }, + "all_time": { + "calls": 40, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 40 + }, + "users": { + "unknown": 40 + }, + "components": { + "unknown": 40 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-09-01T18:42:55.879795Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "3a00b19e-e7db-4403-98aa-a77c9ab61ecc", + "interface_key": "rest_api:GET /workstreams/{workstream_id}", + "interface_kind": "rest_api", + "legacy_since": "2026-06-04T00:25:59.014966Z", + "replacement_ref": "/workplans/{workplan_id}", + "owner_component": "state-hub.api", + "status": "legacy", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": null, + "retired_at": null, + "created_at": "2026-06-04T00:25:59.014966Z", + "updated_at": "2026-06-04T00:25:59.014966Z" + }, + "all_time": { + "calls": 3411478, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 3411478 + }, + "users": { + "unknown": 3411478 + }, + "components": { + "unknown": 3411478 + } + }, + "window": { + "calls": 1, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 1 + }, + "users": { + "unknown": 1 + }, + "components": { + "unknown": 1 + } + }, + "last_seen_at": "2026-10-08T14:34:43.410356Z", + "retirement_candidate": false, + "retirement_reason": "1 call(s) in review window" + }, + { + "interface": { + "id": "69e1a255-a8b9-4101-8a68-b1bed23abfda", + "interface_key": "rest_api:GET /workstreams/{workstream_id}/dependencies/", + "interface_kind": "rest_api", + "legacy_since": "2026-06-04T00:25:59.291135Z", + "replacement_ref": "/workplans/{workplan_id}/dependencies/", + "owner_component": "state-hub.api", + "status": "legacy", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": null, + "retired_at": null, + "created_at": "2026-06-04T00:25:59.291135Z", + "updated_at": "2026-06-04T00:25:59.291135Z" + }, + "all_time": { + "calls": 1643952, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 1643952 + }, + "users": { + "unknown": 1643952 + }, + "components": { + "unknown": 1643952 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-08-21T13:50:30.082480Z", + "retirement_candidate": false, + "retirement_reason": "quiet 53d of 60d required for 1643952 all-time call(s)" + }, + { + "interface": { + "id": "44734299-5fc3-4da9-a412-bf9884a56a44", + "interface_key": "rest_api:PATCH /messages/{id}/read broadcast-without-reader", + "interface_kind": "rest_api", + "legacy_since": "2026-09-23T15:30:53.835158Z", + "replacement_ref": "PATCH /messages/{id}/read?reader=", + "owner_component": "state-hub.api", + "status": "legacy", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": null, + "retired_at": null, + "created_at": "2026-09-23T15:30:53.835158Z", + "updated_at": "2026-09-23T15:30:53.835158Z" + }, + "all_time": { + "calls": 82, + "tenant_count": 1, + "user_count": 1, + "component_count": 2, + "tenants": { + "unknown": 82 + }, + "users": { + "unknown": 82 + }, + "components": { + "unknown": 77, + "prj-helixforge-factory": 5 + } + }, + "window": { + "calls": 22, + "tenant_count": 1, + "user_count": 1, + "component_count": 2, + "tenants": { + "unknown": 22 + }, + "users": { + "unknown": 22 + }, + "components": { + "unknown": 17, + "prj-helixforge-factory": 5 + } + }, + "last_seen_at": "2026-10-13T20:58:21.459537Z", + "retirement_candidate": false, + "retirement_reason": "22 call(s) in review window" + }, + { + "interface": { + "id": "b1aae931-b51f-4a85-a723-865dd65f6132", + "interface_key": "rest_api:PATCH /workstreams/{workstream_id}", + "interface_kind": "rest_api", + "legacy_since": "2026-06-04T06:09:41.901035Z", + "replacement_ref": "/workplans/{workplan_id}", + "owner_component": "state-hub.api", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "STATE-WP-0079-T05: founder-authorized retirement 2026-09-22; seven-day gate passed, replacement verified.", + "retired_at": "2026-09-22T09:22:56.610064Z", + "created_at": "2026-06-04T06:09:41.901035Z", + "updated_at": "2026-09-22T09:22:56.603117Z" + }, + "all_time": { + "calls": 584, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 584 + }, + "users": { + "unknown": 584 + }, + "components": { + "unknown": 584 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-08-16T16:33:34.045078Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "ed6451c9-d2cc-4486-9afe-d3852782cabc", + "interface_key": "rest_api:POST /workstreams/", + "interface_kind": "rest_api", + "legacy_since": "2026-06-04T07:40:24.168535Z", + "replacement_ref": "/workplans/", + "owner_component": "state-hub.api", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "STATE-WP-0079-T05: founder-authorized retirement 2026-09-22; seven-day gate passed, replacement verified.", + "retired_at": "2026-09-22T09:22:56.716384Z", + "created_at": "2026-06-04T07:40:24.168535Z", + "updated_at": "2026-09-22T09:22:56.710100Z" + }, + "all_time": { + "calls": 811, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 811 + }, + "users": { + "unknown": 811 + }, + "components": { + "unknown": 811 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-08-16T01:03:41.093965Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + }, + { + "interface": { + "id": "281fd706-b192-4194-8055-2d8733c115f9", + "interface_key": "rest_api:POST /workstreams/{workstream_id}/dependencies/", + "interface_kind": "rest_api", + "legacy_since": "2026-06-04T22:57:37.149207Z", + "replacement_ref": "/workplans/{workplan_id}/dependencies/", + "owner_component": "state-hub.api", + "status": "retired", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": "Retired by STATE-WP-0069 closeout", + "retired_at": "2026-08-31T08:55:09.130713Z", + "created_at": "2026-06-04T22:57:37.149207Z", + "updated_at": "2026-08-31T08:55:09.123888Z" + }, + "all_time": { + "calls": 4975, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 4975 + }, + "users": { + "unknown": 4975 + }, + "components": { + "unknown": 4975 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-07-07T20:31:35.389791Z", + "retirement_candidate": false, + "retirement_reason": "already retired" + } + ], + "retirement_candidates": [ + { + "interface": { + "id": "cad13f19-ddb1-49b9-86d0-8789b7599f3a", + "interface_key": "rest_api:GET /tasks/?workstream_id", + "interface_kind": "rest_api", + "legacy_since": "2026-08-29T10:01:31.546213Z", + "replacement_ref": "/tasks/?workplan_id=", + "owner_component": "state-hub.api", + "status": "legacy", + "replacement_verified": true, + "manual_hold": false, + "hold_reason": null, + "notes": null, + "retired_at": null, + "created_at": "2026-08-29T10:01:31.546213Z", + "updated_at": "2026-08-29T10:01:31.546213Z" + }, + "all_time": { + "calls": 53, + "tenant_count": 1, + "user_count": 1, + "component_count": 1, + "tenants": { + "unknown": 53 + }, + "users": { + "unknown": 53 + }, + "components": { + "unknown": 53 + } + }, + "window": { + "calls": 0, + "tenant_count": 0, + "user_count": 0, + "component_count": 0, + "tenants": {}, + "users": {}, + "components": {} + }, + "last_seen_at": "2026-10-04T18:32:36.539797Z", + "retirement_candidate": true, + "retirement_reason": "no usage in review window; quiet 9d (>= 7d required for 53 all-time call(s))" + } + ] + }, + "days": 7 +} diff --git a/docs/inbox-change-feed-contract.md b/docs/inbox-change-feed-contract.md index 9fa11b9..9f4d631 100644 --- a/docs/inbox-change-feed-contract.md +++ b/docs/inbox-change-feed-contract.md @@ -1,7 +1,9 @@ # State Hub inbox change-feed contract -Status: **accepted** (founder, 2026-10-13: D1-D4 as recommended). Not yet -implemented; implementation is `STATE-WP-0098-T05`. Deliverable of `STATE-WP-0098-T03`. +Status: **implemented (not deployed)** (2026-10-14, `STATE-WP-0098-T05`; +migration `c4d5e6f7a8b9`). Accepted by the founder 2026-10-13: D1-D4 as +recommended. Deliverable of `STATE-WP-0098-T03`. Deploy is a separate confirmed +step. Implementation details and deviations are in §8. It is the State Hub input for `HUB-WP-0011-T01` (the first row of its "Dependency review — 2026-10-04" table). Date: 2026-10-13. @@ -148,7 +150,7 @@ Only the transitions below can occur in the current State Hub model. | `created` | `POST /messages/`, `POST /{id}/reply` | `message` upsert with no prior row; `thread_id`, `supersedes_id`, `kind` and `expires_at` set at creation | | `read` (direct) | `PATCH /{id}/read` without a reader; a reply marks the original as read; archiving a direct message stamps `read_at` if it is null | `message` upsert where `read_at` goes from null to a value. There is one global `read_at` and no reader identity | | `receipt` (broadcast, per reader) | `PATCH /{id}/read?reader=`, `POST /{id}/ack`, a reply to a broadcast, and **`GET /messages/?unread_only=true` (stamps `delivered_at`)** | `receipt` upsert keyed `(message_id, agent)`. The fields are first-write-wins (`COALESCE`) | -| `archived` | `PATCH /{id}/archive`, and supersession of a broadcast | `message` upsert where `archived_at` is set. Re-archiving today *changes* `archived_at`, which arrives as a normal higher-revision upsert | +| `archived` | `PATCH /{id}/archive`, and supersession of a broadcast | `message` upsert where `archived_at` is set. Since T05, re-archiving keeps the first `archived_at` and produces no new revision | | `superseded` | sending a broadcast with `supersedes_id` archives its predecessor in the same transaction | one revision that holds the new `message` and the predecessor's archive upsert | | reply / thread link | `thread_id` points at the root; a reply inherits the root | carried in `created`. There is no separate link event | | `deleted` | **no API route deletes messages.** Only a manual DB delete can | `tombstone` record. The delete cascades: receipts are dropped, and `thread_id`/`supersedes_id` on other rows become NULL. Those referencing rows arrive as upserts in the same revision | @@ -294,8 +296,8 @@ them: - `thread_id` and `supersedes_id` may change only from a value to NULL, and only in a revision that tombstones the referenced message. - `read_at` and the receipt fields may change only from null to a value. -- `archived_at` may change from null to a value, or to a later value - (re-archive). It never returns to null. +- `archived_at` may change only from null to a value (first-write-wins since + T05). It never returns to null. Anything else is refused, and hub-core re-snapshots. @@ -304,11 +306,10 @@ and the cursor is updated in the same transaction. Thread roots arrive in the same or an earlier revision, because a reply cannot exist before its root, so the pilot's thread-closure check still holds. -Small State Hub fix (part of the implementation): make `archive_message` -first-write-wins for `archived_at`. The re-archive rule can then tighten to -"null to value only". +Small State Hub fix (done in T05): `archive_message` is first-write-wins for +`archived_at`, so the re-archive rule above is "null to value only". -## 7. State Hub implementation sketch (not implemented) +## 7. State Hub implementation sketch (implemented 2026-10-14, see §8) | File | Change | Size | | --- | --- | --- | @@ -328,3 +329,51 @@ SQLite. If it does, the trigger tests need the Postgres test fixture. The hub-core side (`source_revision` column, page applier, cursor store, staleness) is not sketched here. It belongs to HUB-WP-0011-T01. + +## 8. Implementation notes (2026-10-14) + +Implemented as `STATE-WP-0098-T05`. Files: migration +`migrations/versions/c4d5e6f7a8b9_inbox_feed_revision.py`, +`api/models/agent_message.py`, `api/schemas/inbox_feed.py`, +`api/routers/message_feed.py`, `api/main.py`, `api/routers/messages.py`, +`tests/test_message_feed.py`. Where this section and §1-§7 differ, this +section describes what the code does. + +- **Revision cache.** `inbox_feed_revision()` caches `":"` in + the transaction-local GUC `statehub.inbox_rev` and reuses it only when the + txid matches. A savepoint that rolls back after drawing a revision can make a + transaction use two revisions; both stay ordered under the advisory lock + (key `5348490098`). +- **Keys and state.** `key` is the message UUID for `message` and `tombstone`, + and `":"` for `receipt`. `state` is the full row and + includes `revision`. Snapshot rows have the same shape as `state`. Within a + revision the order is messages, receipts, tombstones. +- **Foreign cursor (409).** The cursor stays `v1:` and cannot carry + the instance. The server answers 409 (`resnapshot_required: true`) when the + cursor has another version prefix (for example `v2:`), when the optional + query parameter `source_instance` differs from this instance's + `instance_label`, or when the cursor is ahead of `high_water` (another + instance or a restored source). A malformed cursor gets 400. +- **Horizon (410).** `compacted_before` is `v1:`, where `R` is the highest + revision of any tombstone older than 90 days (`v1:0` if there is none). + Those tombstones are no longer served, and `after < R` gets 410 with + `resnapshot_required: true`. The horizon is enforced logically. Nothing + purges tombstones physically (D2). The snapshot also carries + `compacted_before` (additive). +- **Paging.** `limit` defaults to 500, maximum 5000. A page may exceed it to + finish a revision. An empty page returns `next_cursor` equal to `after`. +- **content_hash** is sha256 over canonical JSON (`sort_keys`, compact + separators) of `{messages, receipts, tombstones}`. Rows are sorted by key and + timestamps are ISO strings. It is deterministic but not byte-identical to the + pilot's hash. +- **Delete reason.** The tombstone `reason` comes from + `SET LOCAL statehub.delete_reason = '...'` if the deleting session sets it, + and is NULL otherwise. +- **Not covered.** A direct `DELETE` on `message_receipts` writes no tombstone, + and no API route does this. Receipts disappear only through the message + delete cascade, which the message tombstone implies. +- **Tests and create_all.** The test suite builds its schema with + `Base.metadata.create_all`. The models attach the same trigger DDL via + `after_create`. The migration keeps its own frozen copy, and a test asserts + the two are identical. It also runs the migration's downgrade and upgrade on + the test database in a rolled-back transaction. diff --git a/migrations/versions/c4d5e6f7a8b9_inbox_feed_revision.py b/migrations/versions/c4d5e6f7a8b9_inbox_feed_revision.py new file mode 100644 index 0000000..7996591 --- /dev/null +++ b/migrations/versions/c4d5e6f7a8b9_inbox_feed_revision.py @@ -0,0 +1,147 @@ +"""inbox change-feed revisions and tombstones (STATE-WP-0098-T05) + +Implements §2 of docs/inbox-change-feed-contract.md (accepted 2026-10-13, D4): + +- sequence ``inbox_revision_seq``; +- ``revision bigint NOT NULL`` on ``agent_messages`` and ``message_receipts``, + backfilled with ONE sequence value (one snapshot = one revision), indexed; +- ``message_tombstones``; +- a BEFORE INSERT OR UPDATE trigger on both tables that skips no-op updates, + takes a transaction-scoped advisory lock (commit order = revision order) and + assigns one revision per transaction (cached per txid in a txn-local GUC); +- a BEFORE DELETE trigger on ``agent_messages`` that writes a tombstone. + +The SQL is a frozen copy of ``api.models.agent_message.INBOX_FEED_FUNCTIONS_SQL``; +tests/test_message_feed.py checks the two agree. + +Revision ID: c4d5e6f7a8b9 +Revises: b3c4d5e6f7a8 +""" +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import UUID + +revision = "c4d5e6f7a8b9" +down_revision = "b3c4d5e6f7a8" +branch_labels = None +depends_on = None + +INBOX_ADVISORY_LOCK_KEY = 5_348_490_098 + +FUNCTIONS_SQL = f""" +CREATE OR REPLACE FUNCTION inbox_feed_revision() RETURNS bigint +LANGUAGE plpgsql AS $fn$ +DECLARE + tx text := txid_current()::text; + cached text := coalesce(current_setting('statehub.inbox_rev', true), ''); + rev bigint; +BEGIN + PERFORM pg_advisory_xact_lock({INBOX_ADVISORY_LOCK_KEY}); + IF split_part(cached, ':', 1) = tx THEN + RETURN split_part(cached, ':', 2)::bigint; + END IF; + rev := nextval('inbox_revision_seq'); + PERFORM set_config('statehub.inbox_rev', tx || ':' || rev::text, true); + RETURN rev; +END +$fn$; + +CREATE OR REPLACE FUNCTION inbox_feed_stamp_revision() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + IF TG_OP = 'UPDATE' THEN + IF NEW IS NOT DISTINCT FROM OLD THEN + RETURN NEW; + END IF; + END IF; + NEW.revision := inbox_feed_revision(); + RETURN NEW; +END +$fn$; + +CREATE OR REPLACE FUNCTION inbox_feed_tombstone() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + INSERT INTO message_tombstones (message_id, revision, deleted_at, reason) + VALUES ( + OLD.id, + inbox_feed_revision(), + now(), + nullif(current_setting('statehub.delete_reason', true), '') + ) + ON CONFLICT (message_id) DO UPDATE + SET revision = EXCLUDED.revision, + deleted_at = EXCLUDED.deleted_at, + reason = EXCLUDED.reason; + RETURN OLD; +END +$fn$; +""" + +TRIGGERS_SQL = """ +CREATE TRIGGER agent_messages_inbox_revision + BEFORE INSERT OR UPDATE ON agent_messages + FOR EACH ROW EXECUTE FUNCTION inbox_feed_stamp_revision(); +CREATE TRIGGER agent_messages_inbox_tombstone + BEFORE DELETE ON agent_messages + FOR EACH ROW EXECUTE FUNCTION inbox_feed_tombstone(); +CREATE TRIGGER message_receipts_inbox_revision + BEFORE INSERT OR UPDATE ON message_receipts + FOR EACH ROW EXECUTE FUNCTION inbox_feed_stamp_revision(); +""" + +_TABLES = ("agent_messages", "message_receipts") + + +def upgrade() -> None: + op.execute(sa.text("CREATE SEQUENCE inbox_revision_seq AS bigint")) + for table in _TABLES: + op.add_column(table, sa.Column("revision", sa.BigInteger(), nullable=True)) + # One sequence value for all existing rows: the pre-feed state is one revision. + op.execute( + sa.text( + """ + DO $$ + DECLARE base bigint := nextval('inbox_revision_seq'); + BEGIN + UPDATE agent_messages SET revision = base; + UPDATE message_receipts SET revision = base; + END + $$ + """ + ) + ) + for table in _TABLES: + op.alter_column(table, "revision", nullable=False) + op.create_index(f"ix_{table}_revision", table, ["revision"]) + op.create_table( + "message_tombstones", + sa.Column("message_id", UUID(as_uuid=True), primary_key=True), + sa.Column("revision", sa.BigInteger(), nullable=False), + sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("reason", sa.Text(), nullable=True), + ) + op.create_index("ix_message_tombstones_revision", "message_tombstones", ["revision"]) + op.execute(sa.text(FUNCTIONS_SQL)) + op.execute(sa.text(TRIGGERS_SQL)) + + +def downgrade() -> None: + op.execute( + sa.text( + """ + DROP TRIGGER IF EXISTS message_receipts_inbox_revision ON message_receipts; + DROP TRIGGER IF EXISTS agent_messages_inbox_tombstone ON agent_messages; + DROP TRIGGER IF EXISTS agent_messages_inbox_revision ON agent_messages; + DROP FUNCTION IF EXISTS inbox_feed_tombstone(); + DROP FUNCTION IF EXISTS inbox_feed_stamp_revision(); + DROP FUNCTION IF EXISTS inbox_feed_revision(); + """ + ) + ) + op.drop_index("ix_message_tombstones_revision", table_name="message_tombstones") + op.drop_table("message_tombstones") + for table in reversed(_TABLES): + op.drop_index(f"ix_{table}_revision", table_name=table) + op.drop_column(table, "revision") + op.execute(sa.text("DROP SEQUENCE IF EXISTS inbox_revision_seq")) diff --git a/tests/test_message_feed.py b/tests/test_message_feed.py new file mode 100644 index 0000000..f9468ff --- /dev/null +++ b/tests/test_message_feed.py @@ -0,0 +1,412 @@ +"""STATE-WP-0098-T05: inbox change feed (docs/inbox-change-feed-contract.md).""" +from __future__ import annotations + +import importlib.util +import os +import uuid +from pathlib import Path + +import pytest +import pytest_asyncio +import sqlalchemy +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from api.models.agent_message import AgentMessage, MessageReceipt +from api.services.message_receipts import upsert_receipt + +CONTRACT = "statehub.inbox-feed.v1" + + +@pytest_asyncio.fixture(autouse=True) +async def _clean_tombstones(test_engine): + # conftest deletes agent_messages after each test, which (correctly) leaves + # tombstones behind; start every feed test from an empty tombstone table. + async with test_engine.begin() as conn: + await conn.execute(text("DELETE FROM message_tombstones")) + yield + + +@pytest.fixture +def factory(test_engine): + return async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + + +async def _send(client, to_agent="state-hub", **payload): + body = {"from_agent": "the-custodian", "subject": "s", "body": "b", "to_agent": to_agent, **payload} + resp = await client.post("/messages/", json=body) + assert resp.status_code == 201, resp.text + return resp.json() + + +async def _revisions(factory, model=AgentMessage): + async with factory() as session: + if model is AgentMessage: + rows = await session.execute(select(AgentMessage.id, AgentMessage.revision)) + return {str(i): r for i, r in rows.all()} + rows = await session.execute( + select(MessageReceipt.message_id, MessageReceipt.agent, MessageReceipt.revision) + ) + return {f"{m}:{a}": r for m, a, r in rows.all()} + + +async def _snapshot(client): + resp = await client.get("/messages/feed/snapshot") + assert resp.status_code == 200, resp.text + assert resp.headers["cache-control"] == "no-store" + return resp.json() + + +async def _page(client, after, limit=500, expect=200, **params): + resp = await client.get("/messages/feed", params={"after": after, "limit": limit, **params}) + assert resp.status_code == expect, resp.text + assert resp.headers["cache-control"] == "no-store" + return resp.json() + + +async def _drain(client, after, limit=2): + pages = [] + while True: + page = await _page(client, after, limit=limit) + pages.append(page) + after = page["next_cursor"] + if not page["has_more"]: + return pages + + +def _rev(cursor): + return int(cursor.split(":", 1)[1]) + + +@pytest.mark.asyncio +async def test_one_revision_per_transaction(factory): + async with factory() as session: + a = AgentMessage(from_agent="x", to_agent="y", subject="a", body="a") + b = AgentMessage(from_agent="x", to_agent="broadcast", subject="b", body="b", kind="news") + session.add_all([a, b]) + await session.flush() + await upsert_receipt(session, [b.id], "reader-1", delivered=True) + await session.commit() + assert a.revision == b.revision # eager_defaults: RETURNING, no lazy load + revs = await _revisions(factory) + receipt_revs = await _revisions(factory, MessageReceipt) + assert set(revs.values()) == {a.revision} + assert set(receipt_revs.values()) == {a.revision} + + async with factory() as session: + c = AgentMessage(from_agent="x", to_agent="y", subject="c", body="c") + session.add(c) + await session.commit() + assert c.revision > a.revision + + +@pytest.mark.asyncio +async def test_no_revision_bump_on_noop_update(client, factory): + direct = await _send(client) + news = await _send(client, to_agent="broadcast") + assert (await client.patch(f"/messages/{direct['id']}/read", json={})).status_code == 200 + resp = await client.patch(f"/messages/{news['id']}/read", json={"reader": "reader-1"}) + assert resp.status_code == 200 + before = await _revisions(factory) + receipts_before = await _revisions(factory, MessageReceipt) + + # COALESCE upsert that changes nothing, and a raw UPDATE setting equal values. + async with factory() as session: + await upsert_receipt(session, [uuid.UUID(news["id"])], "reader-1", read=True, delivered=False) + await session.execute( + text("UPDATE agent_messages SET read_at = read_at, subject = subject") + ) + await session.commit() + assert await _revisions(factory) == before + assert await _revisions(factory, MessageReceipt) == receipts_before + + # A real change bumps exactly the touched receipt. + async with factory() as session: + await upsert_receipt(session, [uuid.UUID(news["id"])], "reader-1", acknowledged=True) + await session.commit() + after = await _revisions(factory, MessageReceipt) + key = f"{news['id']}:reader-1" + assert after[key] > receipts_before[key] + + +@pytest.mark.asyncio +async def test_archive_is_first_write_wins(client, factory): + msg = await _send(client) + first = await client.patch(f"/messages/{msg['id']}/archive") + assert first.status_code == 200 + rev = (await _revisions(factory))[msg["id"]] + second = await client.patch(f"/messages/{msg['id']}/archive") + assert second.status_code == 200 + assert second.json()["archived_at"] == first.json()["archived_at"] + assert (await _revisions(factory))[msg["id"]] == rev + + +@pytest.mark.asyncio +async def test_delete_writes_tombstone_in_same_revision(client, factory): + root = await _send(client) + reply = ( + await client.post(f"/messages/{root['id']}/reply", json={"from_agent": "state-hub", "body": "r"}) + ).json() + news = await _send(client, to_agent="broadcast") + await upsert_receipt_via_api(client, news["id"]) + cursor = (await _snapshot(client))["cursor"] + + async with factory() as session: + await session.execute(text("SET LOCAL statehub.delete_reason = 'test cleanup'")) + await session.execute( + text("DELETE FROM agent_messages WHERE id IN (:a, :b)"), + {"a": root["id"], "b": news["id"]}, + ) + await session.commit() + + page = await _page(client, cursor) + revisions = {c["revision"] for c in page["changes"]} + assert len(revisions) == 1 + by_entity = {(c["entity"], c["key"]): c for c in page["changes"]} + root_tomb = by_entity[("tombstone", root["id"])] + assert root_tomb["op"] == "delete" + assert root_tomb["state"]["reason"] == "test cleanup" + assert "body" not in root_tomb["state"] + assert ("tombstone", news["id"]) in by_entity + # The cascade SET NULL on the reply arrives as an upsert in the same revision. + reply_change = by_entity[("message", reply["id"])] + assert reply_change["state"]["thread_id"] is None + # Cascaded receipts are gone from the snapshot. + snap = await _snapshot(client) + assert snap["receipts"] == [] + assert {t["message_id"] for t in snap["tombstones"]} == {root["id"], news["id"]} + + +async def upsert_receipt_via_api(client, message_id): + resp = await client.post(f"/messages/{message_id}/ack", json={"agent": "reader-1"}) + assert resp.status_code in (200, 201), resp.text + + +@pytest.mark.asyncio +async def test_snapshot_and_feed_parity(client, factory): + first = await _snapshot(client) + assert first["contract"] == CONTRACT + assert first["messages"] == [] and first["count"] == 0 + + m1 = await _send(client) + m2 = await _send(client, to_agent="broadcast", kind="standing") + await _send(client, to_agent="broadcast", kind="standing", supersedes_id=m2["id"]) + await client.patch(f"/messages/{m1['id']}/read", json={}) + await client.get("/messages/", params={"to_agent": "reader-2", "unread_only": "true"}) + await upsert_receipt_via_api(client, m2["id"]) + await client.post(f"/messages/{m1['id']}/reply", json={"from_agent": "state-hub", "body": "r"}) + doomed = await _send(client) + async with factory() as session: + await session.execute(text("DELETE FROM agent_messages WHERE id = :i"), {"i": doomed["id"]}) + await session.commit() + + snap = await _snapshot(client) + pages = await _drain(client, "v1:0", limit=2) + assert len(pages) > 1 + messages, receipts, tombstones = {}, {}, {} + seen = [] + for page in pages: + assert page["contract"] == CONTRACT + for change in page["changes"]: + seen.append(change["revision"]) + if change["entity"] == "message": + messages[change["key"]] = change["state"] + elif change["entity"] == "receipt": + receipts[change["key"]] = change["state"] + else: + tombstones[change["key"]] = change["state"] + messages.pop(change["key"], None) + for key in [k for k in receipts if k.startswith(change["key"] + ":")]: + receipts.pop(key) + assert seen == sorted(seen) + assert pages[-1]["next_cursor"] == snap["cursor"] == pages[-1]["high_water"] + assert sorted(messages.values(), key=lambda m: m["id"]) == snap["messages"] + assert sorted(receipts.values(), key=lambda r: (r["message_id"], r["agent"])) == snap["receipts"] + assert list(tombstones.values()) == snap["tombstones"] + assert snap["count"] == len(snap["messages"]) == 4 + assert len(snap["content_hash"]) == 64 + # Same state, same hash; a change moves it. + assert (await _snapshot(client))["content_hash"] == snap["content_hash"] + await client.patch(f"/messages/{m2['id']}/read", json={"reader": "reader-3"}) + assert (await _snapshot(client))["content_hash"] != snap["content_hash"] + + +@pytest.mark.asyncio +async def test_page_never_splits_a_revision(factory, client): + async with factory() as session: + session.add_all( + [AgentMessage(from_agent="x", to_agent="y", subject=str(i), body="b") for i in range(3)] + ) + await session.commit() + await _send(client) + + page = await _page(client, "v1:0", limit=1) + assert len(page["changes"]) == 3 + assert len({c["revision"] for c in page["changes"]}) == 1 + assert page["has_more"] is True + assert page["next_cursor"] == f"v1:{page['changes'][0]['revision']}" + last = await _page(client, page["next_cursor"], limit=1) + assert len(last["changes"]) == 1 and last["has_more"] is False + assert last["next_cursor"] == last["high_water"] + + +@pytest.mark.asyncio +async def test_replay_and_empty_page_heartbeat(client): + await _send(client) + await _send(client) + snap = await _snapshot(client) + a = await _page(client, "v1:0") + b = await _page(client, "v1:0") + assert a["changes"] == b["changes"] and a["next_cursor"] == b["next_cursor"] + + empty = await _page(client, snap["cursor"]) + assert empty["changes"] == [] and empty["has_more"] is False + assert empty["next_cursor"] == snap["cursor"] == empty["high_water"] + assert empty["source_time"] and "source_instance" in empty + + +@pytest.mark.asyncio +async def test_cursor_older_than_tombstone_horizon_is_410(client, factory): + await _send(client) + async with factory() as session: + await session.execute( + text( + "INSERT INTO message_tombstones (message_id, revision, deleted_at) " + "VALUES (:i, nextval('inbox_revision_seq'), now() - interval '91 days')" + ), + {"i": str(uuid.uuid4())}, + ) + await session.commit() + fresh = await _send(client) + snap = await _snapshot(client) + compacted = _rev(snap["compacted_before"]) + assert compacted > 0 + assert snap["tombstones"] == [] # out of horizon, not served + + gone = await _page(client, f"v1:{compacted - 1}", expect=410) + assert gone["resnapshot_required"] is True + ok = await _page(client, f"v1:{compacted}") + assert [c["key"] for c in ok["changes"]] == [fresh["id"]] + assert ok["compacted_before"] == snap["compacted_before"] + + +@pytest.mark.asyncio +async def test_foreign_cursor_is_409_and_malformed_is_400(client): + await _send(client) + snap = await _snapshot(client) + body = await _page(client, "v2:1", expect=409) + assert body["resnapshot_required"] is True + await _page(client, snap["cursor"], source_instance="some-other-hub", expect=409) + await _page(client, f"v1:{_rev(snap['cursor']) + 1000}", expect=409) + await _page(client, "garbage", expect=400) + if snap["source_instance"] is not None: + await _page(client, snap["cursor"], source_instance=snap["source_instance"]) + + +@pytest.mark.asyncio +async def test_feed_routes_do_not_collide_with_message_routes(client): + resp = await client.post("/messages/feed/snapshot") + assert resp.status_code == 405 + assert (await client.get("/messages/feed")).status_code == 422 # after is required + + +def _load_migration(): + path = Path(__file__).parent.parent / "migrations/versions/c4d5e6f7a8b9_inbox_feed_revision.py" + spec = importlib.util.spec_from_file_location("inbox_feed_migration", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +_FUNCS = "('inbox_feed_revision','inbox_feed_stamp_revision','inbox_feed_tombstone')" + + +def test_migration_downgrade_upgrade_and_ddl_parity(): + """Run the migration on the test DB inside a rolled-back transaction.""" + from api.models import agent_message + + migration = _load_migration() + assert migration.FUNCTIONS_SQL == agent_message.INBOX_FEED_FUNCTIONS_SQL + url = os.getenv( + "TEST_DATABASE_URL", "postgresql+asyncpg://custodian:changeme@127.0.0.1:5432/custodian_test" + ).replace("+asyncpg", "+psycopg2") + engine = sqlalchemy.create_engine(url) + try: + with engine.connect() as conn: + trans = conn.begin() + try: + conn.execute( + text( + "INSERT INTO agent_messages (id, from_agent, to_agent, subject, body) " + "VALUES (gen_random_uuid(), 'x', 'y', 's', 'b'), " + "(gen_random_uuid(), 'x', 'y', 's2', 'b')" + ) + ) + ops = Operations(MigrationContext.configure(conn)) + migration.op = ops + migration.downgrade() + cols = {c["name"] for c in sqlalchemy.inspect(conn).get_columns("agent_messages")} + assert "revision" not in cols + assert not sqlalchemy.inspect(conn).has_table("message_tombstones") + migration.upgrade() + revs = conn.execute(text("SELECT DISTINCT revision FROM agent_messages")).scalars().all() + assert len(revs) == 1 and revs[0] is not None # one backfill revision + triggers = set( + conn.execute( + text( + "SELECT tgname FROM pg_trigger WHERE NOT tgisinternal " + "AND tgrelid IN ('agent_messages'::regclass, 'message_receipts'::regclass)" + ) + ).scalars() + ) + assert triggers == { + "agent_messages_inbox_revision", + "agent_messages_inbox_tombstone", + "message_receipts_inbox_revision", + } + conn.execute(text("DELETE FROM agent_messages")) + assert conn.execute(text("SELECT count(*) FROM message_tombstones")).scalar() == 2 + finally: + trans.rollback() + finally: + engine.dispose() + + +@pytest.mark.asyncio +async def test_snapshot_transaction_is_repeatable_read_read_only(factory): + from api.routers.message_feed import _begin_consistent_read + + async with factory() as session: + await _begin_consistent_read(session) + assert (await session.execute(text("SHOW transaction_isolation"))).scalar() == "repeatable read" + assert (await session.execute(text("SHOW transaction_read_only"))).scalar() == "on" + + +@pytest.mark.asyncio +async def test_advisory_lock_serializes_writers_in_revision_order(factory): + import asyncio + + first = factory() + second = factory() + try: + a = AgentMessage(from_agent="x", to_agent="y", subject="a", body="a") + first.add(a) + await first.flush() # holds the inbox lock until commit + + async def write_b(): + b = AgentMessage(from_agent="x", to_agent="y", subject="b", body="b") + second.add(b) + await second.commit() + return b + + task = asyncio.create_task(write_b()) + await asyncio.sleep(0.5) + assert not task.done() # blocked on pg_advisory_xact_lock + await first.commit() + b = await asyncio.wait_for(task, timeout=10) + assert b.revision > a.revision + finally: + await first.close() + await second.close() diff --git a/workplans/STATE-WP-0094-routed-gates-and-owner-returns.md b/workplans/STATE-WP-0094-routed-gates-and-owner-returns.md index 38efb8c..2695843 100644 --- a/workplans/STATE-WP-0094-routed-gates-and-owner-returns.md +++ b/workplans/STATE-WP-0094-routed-gates-and-owner-returns.md @@ -82,7 +82,8 @@ that evidenced omission. Primary deployment and normal-sync parity remain T05. ```task id: STATE-WP-0094-T05 status: wait -blocking_reason: "Memo 3 apply trials need a read-only Forgejo token for the Activity Core worker (draft CCR-2026-0034 in railiance-platform); then memo 4 schedule and 48 h freshness evidence" +blocked_on: message-from:railiance-platform +blocking_reason: "Memo 3 apply trials need a read-only Forgejo token for the Activity Core worker; then memo 4 schedule and 48 h freshness evidence. 2026-10-14: CCR-2026-0034 in railiance-platform is the keycape operator reader (applied), not this token; no Forgejo-read CCR for the Activity Core worker found." priority: medium state_hub_task_id: "ec0acd06-0e2f-5b5c-ae8a-6a3241041d4c" ``` diff --git a/workplans/STATE-WP-0098-hub-core-receipt-inputs.md b/workplans/STATE-WP-0098-hub-core-receipt-inputs.md index 703c71d..071eb1a 100644 --- a/workplans/STATE-WP-0098-hub-core-receipt-inputs.md +++ b/workplans/STATE-WP-0098-hub-core-receipt-inputs.md @@ -4,7 +4,7 @@ type: workplan title: "Supply State Hub inputs that hub-core receipt tasks wait on" domain: infotech repo: state-hub -status: active +status: blocked owner: claude-code topic_slug: infotech created: "2026-10-13" @@ -42,7 +42,9 @@ plus a JSON case list, and notify hub-core. ```task id: STATE-WP-0098-T02 -status: progress +status: wait +blocked_on: message-from:hub-core +blocking_reason: "Owner map delivered and switch-plan requests sent 2026-10-13 to hub-core (5e2f749b), activity-core (13cff5ff), repo-manager (0c49c989), sbom-nexus (63ce4ead) and policy-nexus (d5dda84d); waiting for their read smokes and reversible switch plans." priority: medium state_hub_task_id: "e6ef3381-9867-52f6-8c9c-2aa24f9a16b3" ``` @@ -74,7 +76,7 @@ goes to the founder before hub-core implements against it. ```task id: STATE-WP-0098-T05 -status: todo +status: done priority: high state_hub_task_id: "14c66646-2a20-5dc2-b9d7-6d30a3dfa6b1" ``` @@ -89,13 +91,23 @@ Implement `docs/inbox-change-feed-contract.md` as accepted on 2026-10-13: - first-write-wins `archived_at` - tests on PostgreSQL +2026-10-14: implemented, not committed and not deployed. Migration +`c4d5e6f7a8b9` (down `b3c4d5e6f7a8`); `api/models/agent_message.py`, +`api/schemas/inbox_feed.py`, `api/routers/message_feed.py`, `api/main.py`, +`api/routers/messages.py` (first-write-wins archive), `tests/test_message_feed.py` +(13 tests). Message tests 42 passed; full suite 1087 passed. Contract status is now +"implemented (not deployed)", and its §8 lists the deviations. Deploy is a +separate confirmed step. + The deploy is a separate, confirmed step. ## Named single reader and parity window ```task id: STATE-WP-0098-T04 -status: todo +status: wait +depends_on: [HUB-WP-0011-T01] +blocking_reason: "The first reader is chosen once hub-core implements the accepted feed contract (HUB-WP-0011-T01) and the feed is deployed. Per D3, choose a reader that does not rely on broadcast delivery stamps." priority: medium state_hub_task_id: "077797c8-289b-55ec-9597-28f223891b02" ```