From ef541f58cf7610092819c20838254fbfb4bf5554 Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 22 Sep 2026 00:26:33 +0200 Subject: [PATCH] STATE-WP-0093: per-recipient broadcast receipts and standing notices (T01-T06). Founder approved T01 on 2026-09-22 (D2, D3, D6 as recommended). - T02: message_receipts table; kind/expires_at/supersedes_id on agent_messages; migration d7e8f9a0b1c2 archives existing broadcasts, leaves direct messages untouched, reversible. - T03: reader-aware mark-read (unattributed broadcast mark-read is a metered, deprecated no-op), delivery receipts on the scoped unread inbox, POST /messages/{id}/ack, news/standing kinds, expiry, supersede, broadcast archive no longer stamps read_at, reply writes the replier's receipt. - T04 (state-hub part): Codex MCP reader param and acknowledge_notice; hub-core part handed off (message 69fc387c). - T05: GET /messages/notices, standing_notices in /state/summary, dashboard standing-notices panel. - T06: 17 new tests; full suite green. Co-Authored-By: Claude Opus 5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 63291@bnt-lap001 Assistant-Session: 8bd77868-ca68-4f49-bb1e-d539ecc0d703 --- api/models/__init__.py | 3 +- api/models/agent_message.py | 48 ++- api/routers/capability_requests.py | 7 +- api/routers/messages.py | 231 ++++++++++- api/routers/state.py | 21 +- api/schemas/agent_message.py | 73 +++- api/schemas/state.py | 2 + api/services/message_receipts.py | 212 ++++++++++ dashboard/src/data/notices.json.py | 15 + dashboard/src/docs/dashboard.md | 1 + dashboard/src/inbox.md | 78 +++- mcp_server/TOOLS.md | 11 +- mcp_server/codex_server.py | 20 +- ...f9a0b1c2_broadcast_receipts_and_notices.py | 92 ++++ tests/test_codex_mcp_server.py | 16 + tests/test_hub_core_imports.py | 8 +- tests/test_message_receipts.py | 392 ++++++++++++++++++ ...broadcast-receipts-and-standing-notices.md | 78 +++- 18 files changed, 1256 insertions(+), 52 deletions(-) create mode 100644 api/services/message_receipts.py create mode 100755 dashboard/src/data/notices.json.py create mode 100644 migrations/versions/d7e8f9a0b1c2_broadcast_receipts_and_notices.py create mode 100644 tests/test_message_receipts.py diff --git a/api/models/__init__.py b/api/models/__init__.py index 1f8b110..3e06ef0 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 +from api.models.agent_message import AgentMessage, MessageReceipt from api.models.capability_catalog import CapabilityCatalog from api.models.capability_request import CapabilityRequest from api.models.tpsc import TPSCCatalog, TPSCSnapshot, TPSCEntry @@ -76,6 +76,7 @@ __all__ = [ "SBOMSnapshot", "SBOMEntry", "Ecosystem", "AgentMessage", + "MessageReceipt", "CapabilityCatalog", "CapabilityRequest", "TPSCCatalog", "TPSCSnapshot", "TPSCEntry", diff --git a/api/models/agent_message.py b/api/models/agent_message.py index d521165..2ef5390 100644 --- a/api/models/agent_message.py +++ b/api/models/agent_message.py @@ -1,15 +1,22 @@ import uuid from datetime import datetime -from sqlalchemy import DateTime, ForeignKey, String, Text, text +from sqlalchemy import DateTime, ForeignKey, Index, String, Text, text from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import Mapped, mapped_column, relationship from api.models.base import Base, new_uuid +BROADCAST = "broadcast" +MESSAGE_KINDS = ("message", "news", "standing") + + class AgentMessage(Base): __tablename__ = "agent_messages" + __table_args__ = ( + Index("ix_agent_messages_to_kind_archived", "to_agent", "kind", "archived_at"), + ) id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=new_uuid @@ -30,6 +37,19 @@ class AgentMessage(Base): archived_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) + # STATE-WP-0093: kind is only meaningful for broadcasts + # (message | news | standing); direct messages stay "message". + kind: Mapped[str] = mapped_column( + String(20), nullable=False, default="message", server_default="message" + ) + expires_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + supersedes_id: Mapped[uuid.UUID | None] = mapped_column( + UUID(as_uuid=True), + ForeignKey("agent_messages.id", ondelete="SET NULL"), + nullable=True, + ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=text("now()"), @@ -42,3 +62,29 @@ class AgentMessage(Base): foreign_keys=[thread_id], lazy="select", ) + + +class MessageReceipt(Base): + """Per-reader state for broadcast messages (STATE-WP-0093). + + Written only for broadcasts; direct messages keep the global + ``AgentMessage.read_at``. + """ + + __tablename__ = "message_receipts" + + message_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("agent_messages.id", ondelete="CASCADE"), + primary_key=True, + ) + agent: Mapped[str] = mapped_column(String(100), primary_key=True, index=True) + delivered_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + read_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + acknowledged_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) diff --git a/api/routers/capability_requests.py b/api/routers/capability_requests.py index 271ed41..74b13ba 100644 --- a/api/routers/capability_requests.py +++ b/api/routers/capability_requests.py @@ -7,7 +7,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from api.database import get_session from api.flow_defs import assertion_result_to_dict, evaluate_transition, flow_result_to_dict -from api.models.agent_message import AgentMessage +from api.models.agent_message import BROADCAST, AgentMessage +from api.services.message_receipts import NEWS_DEFAULT_TTL, utcnow from api.models.capability_catalog import CapabilityCatalog from api.models.capability_request import CapabilityRequest from api.models.domain import Domain @@ -372,6 +373,10 @@ def _add_notification( subject=subject, body=body, ) + if to_agent == BROADCAST: + # STATE-WP-0093: system broadcasts are news (seen once per reader). + msg.kind = "news" + msg.expires_at = utcnow() + NEWS_DEFAULT_TTL session.add(msg) diff --git a/api/routers/messages.py b/api/routers/messages.py index d8334cb..cfe102c 100644 --- a/api/routers/messages.py +++ b/api/routers/messages.py @@ -1,12 +1,29 @@ from datetime import datetime, timezone -from fastapi import APIRouter, Depends, HTTPException, status -from sqlalchemy import or_, select +from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response, status +from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession from api.database import get_session -from api.models.agent_message import AgentMessage -from api.schemas.agent_message import MessageCreate, MessageRead, MessageReply +from api.models.agent_message import BROADCAST, MESSAGE_KINDS, AgentMessage +from api.schemas.agent_message import ( + MessageAck, + MessageCreate, + MessageMarkRead, + MessageRead, + MessageReply, + NoticeStatus, +) +from api.services.legacy_meter import identity_from_request, record_legacy_usage +from api.services.message_receipts import ( + NEWS_DEFAULT_TTL, + broadcast_unread_clause, + is_broadcast, + notice_statuses, + receipts_for, + upsert_receipt, + utcnow, +) from api.services.repository_aliases import ( canonicalize_repository_slug, resolve_repository_slug, @@ -16,6 +33,38 @@ from hub_core.models.message_identity_alias import MessageIdentityAlias router = APIRouter(prefix="/messages", tags=["messages"]) +UNATTRIBUTED_BROADCAST_READ_KEY = "rest_api:PATCH /messages/{id}/read broadcast-without-reader" +UNATTRIBUTED_BROADCAST_READ_REPLACEMENT = "PATCH /messages/{id}/read?reader=" +UNATTRIBUTED_BROADCAST_WARNING = '299 - "broadcast mark-read needs ?reader="' + + +async def _reader_values(session: AsyncSession, reader: str) -> tuple[str, tuple[str, ...]]: + """Canonical reader slug plus every historical slug it answers to.""" + resolution = await resolve_repository_slug(session, reader, required=False) + if resolution is None: + return reader, (reader,) + return resolution.canonical_slug, tuple(resolution.slug_values) + + +async def _read_view( + session: AsyncSession, + message: AgentMessage, + reader_values: tuple[str, ...] | None, + reader: str | None, +) -> MessageRead: + view = MessageRead.model_validate(message) + if reader_values is None or not is_broadcast(message): + return view + state = (await receipts_for(session, [message.id], reader_values)).get(message.id, {}) + return view.model_copy( + update={ + "reader": reader, + "delivered_at": state.get("delivered_at"), + "read_at": state.get("read_at"), + "acknowledged_at": state.get("acknowledged_at"), + } + ) + async def _get_message(reference: str, session: AsyncSession) -> AgentMessage: message_id = await resolve_message_reference( @@ -43,6 +92,38 @@ async def send_message( payload = body.model_dump() payload["from_agent"] = await canonicalize_repository_slug(session, body.from_agent) payload["to_agent"] = await canonicalize_repository_slug(session, body.to_agent) + if payload["to_agent"] == BROADCAST: + kind = body.kind or "news" + if kind == "message": + kind = "news" + if kind not in MESSAGE_KINDS: + raise HTTPException( + status_code=422, + detail=f"kind must be one of {', '.join(MESSAGE_KINDS[1:])} for broadcasts", + ) + payload["kind"] = kind + if kind == "news" and body.expires_at is None: + payload["expires_at"] = utcnow() + NEWS_DEFAULT_TTL + if body.supersedes_id is not None: + predecessor = await session.get(AgentMessage, body.supersedes_id) + if predecessor is None or not is_broadcast(predecessor): + raise HTTPException( + status_code=404, + detail=f"Superseded broadcast {body.supersedes_id} not found", + ) + if predecessor.archived_at is None: + predecessor.archived_at = utcnow() + else: + if body.kind not in (None, "message"): + raise HTTPException( + status_code=422, detail="kind news/standing is only valid for broadcasts" + ) + if body.expires_at is not None or body.supersedes_id is not None: + raise HTTPException( + status_code=422, + detail="expires_at and supersedes_id are only valid for broadcasts", + ) + payload["kind"] = "message" message = AgentMessage(**payload) session.add(message) await session.commit() @@ -57,24 +138,74 @@ async def list_messages( unread_only: bool = False, limit: int = 50, session: AsyncSession = Depends(get_session), -) -> list[AgentMessage]: +) -> list[MessageRead]: + now = utcnow() query = select(AgentMessage).where(AgentMessage.archived_at.is_(None)) + reader: str | None = None + reader_values: tuple[str, ...] | None = None if to_agent: - resolution = await resolve_repository_slug(session, to_agent, required=False) - values = resolution.slug_values if resolution else (to_agent,) - query = query.where( - or_(AgentMessage.to_agent.in_(values), AgentMessage.to_agent == "broadcast") - ) + reader, reader_values = await _reader_values(session, to_agent) + direct = AgentMessage.to_agent.in_(reader_values) + if unread_only: + query = query.where( + or_( + and_(direct, AgentMessage.read_at.is_(None)), + broadcast_unread_clause(reader_values, now), + ) + ) + else: + query = query.where( + or_( + direct, + and_( + AgentMessage.to_agent == BROADCAST, + or_(AgentMessage.expires_at.is_(None), AgentMessage.expires_at > now), + ), + ) + ) + elif unread_only: + query = query.where(AgentMessage.read_at.is_(None)) if from_agent: resolution = await resolve_repository_slug(session, from_agent, required=False) values = resolution.slug_values if resolution else (from_agent,) query = query.where(AgentMessage.from_agent.in_(values)) - if unread_only: - query = query.where(AgentMessage.read_at.is_(None)) result = await session.execute( query.order_by(AgentMessage.created_at.desc()).limit(limit) ) - return list(result.scalars().all()) + messages = list(result.scalars().all()) + if reader_values is None or reader is None: + return [MessageRead.model_validate(m) for m in messages] + + broadcast_ids = [m.id for m in messages if is_broadcast(m)] + if unread_only and broadcast_ids and reader != BROADCAST: + # D3 (founder-approved 2026-09-22): the orientation inbox call records + # an idempotent delivery receipt for each broadcast it returns. + await upsert_receipt(session, broadcast_ids, reader, delivered=True, at=now) + await session.commit() + receipts = await receipts_for(session, broadcast_ids, reader_values) + views = [] + for message in messages: + view = MessageRead.model_validate(message) + if is_broadcast(message): + state = receipts.get(message.id, {}) + view = view.model_copy( + update={ + "reader": reader, + "delivered_at": state.get("delivered_at"), + "read_at": state.get("read_at"), + "acknowledged_at": state.get("acknowledged_at"), + } + ) + views.append(view) + return views + + +@router.get("/notices", response_model=list[NoticeStatus]) +async def list_notices( + session: AsyncSession = Depends(get_session), +) -> list[NoticeStatus]: + """Live standing notices with acknowledged / delivered-only / unreached repos.""" + return await notice_statuses(session) @router.get("/thread/{thread_id}", response_model=list[MessageRead]) @@ -100,14 +231,66 @@ async def get_thread( @router.patch("/{message_id}/read", response_model=MessageRead) async def mark_read( message_id: str, + request: Request, + response: Response, + reader: str | None = None, + ack: bool = False, + body: MessageMarkRead | None = Body(default=None), session: AsyncSession = Depends(get_session), -) -> AgentMessage: +) -> AgentMessage | MessageRead: message = await _get_message(message_id, session) - if message.read_at is None: - message.read_at = datetime.now(timezone.utc) - await session.commit() - await session.refresh(message) - return message + if not is_broadcast(message): + if message.read_at is None: + message.read_at = datetime.now(timezone.utc) + await session.commit() + await session.refresh(message) + return message + + reader = reader or (body.reader if body else None) + ack = ack or bool(body and body.ack) + if not reader: + view = MessageRead.model_validate(message) + # D2: an unattributed mark-read never hides a broadcast from anyone. + response.headers["Deprecation"] = "true" + response.headers["Warning"] = UNATTRIBUTED_BROADCAST_WARNING + response.headers["X-StateHub-Replacement"] = UNATTRIBUTED_BROADCAST_READ_REPLACEMENT + try: + await record_legacy_usage( + session, + interface_key=UNATTRIBUTED_BROADCAST_READ_KEY, + interface_kind="rest_api", + replacement_ref=UNATTRIBUTED_BROADCAST_READ_REPLACEMENT, + owner_component="state-hub.api", + replacement_verified=True, + identity=identity_from_request(request), + ) + except Exception: + await session.rollback() + return view + + canonical, values = await _reader_values(session, reader) + await upsert_receipt(session, [message.id], canonical, read=True, acknowledged=ack) + await session.commit() + return await _read_view(session, message, values, canonical) + + +@router.post("/{message_id}/ack", response_model=MessageRead) +async def acknowledge_message( + message_id: str, + body: MessageAck, + session: AsyncSession = Depends(get_session), +) -> MessageRead: + """Acknowledge a broadcast (clears a standing notice for that agent).""" + message = await _get_message(message_id, session) + if not is_broadcast(message): + raise HTTPException( + status_code=400, + detail="ack applies to broadcasts only; use PATCH /messages/{id}/read", + ) + canonical, values = await _reader_values(session, body.agent) + await upsert_receipt(session, [message.id], canonical, acknowledged=True) + await session.commit() + return await _read_view(session, message, values, canonical) @router.patch("/{message_id}/archive", response_model=MessageRead) @@ -117,7 +300,8 @@ async def archive_message( ) -> AgentMessage: message = await _get_message(message_id, session) message.archived_at = datetime.now(timezone.utc) - if message.read_at is None: + # 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 await session.commit() await session.refresh(message) @@ -135,10 +319,13 @@ async def reply_to_message( session: AsyncSession = Depends(get_session), ) -> AgentMessage: original = await _get_message(message_id, session) - if original.read_at is None: + replier = await canonicalize_repository_slug(session, body.from_agent) + if is_broadcast(original): + await upsert_receipt(session, [original.id], replier, read=True) + elif original.read_at is None: original.read_at = datetime.now(timezone.utc) reply = AgentMessage( - from_agent=await canonicalize_repository_slug(session, body.from_agent), + from_agent=replier, to_agent=await canonicalize_repository_slug(session, original.from_agent), subject=f"Re: {original.subject}", body=body.body, diff --git a/api/routers/state.py b/api/routers/state.py index c78f238..1545449 100644 --- a/api/routers/state.py +++ b/api/routers/state.py @@ -9,6 +9,7 @@ from sqlalchemy.orm import noload, selectinload from api.config import settings from api.database import get_session +from api.services.message_receipts import standing_notice_digests from api.services.schema_state import schema_state from api.flow_defs import assertion_result_to_dict, load_flow from api.models.capability_request import CapabilityRequest @@ -136,6 +137,18 @@ def _apply_summary_flavor_view( ) +async def _live_summary_sections( + session: AsyncSession, *, refresh: bool = False +) -> dict[str, object]: + """Sections computed per request, outside the revision-keyed cache.""" + return { + "ops_runs": await get_ops_run_projection(refresh=refresh), + # STATE-WP-0093 D5: receipts are written on inbox reads, which do not + # bump the summary revision, so standing-notice counts are live. + "standing_notices": await standing_notice_digests(session), + } + + @router.get("/summary", response_model=StateSummary) async def get_summary( request: Request, @@ -155,7 +168,7 @@ async def get_summary( if cache_status == "hit-revision" and cached is not None: _summary_cache_headers(response, cache_status="hit-revision", revision=revision_token) return _apply_summary_flavor_view( - cached.model_copy(update={"ops_runs": await get_ops_run_projection()}), + cached.model_copy(update=await _live_summary_sections(session)), include_residuals=include_residuals, flavor=flavor, ) @@ -164,7 +177,7 @@ async def get_summary( result = await apply_progress_section(session, cached, revision) _summary_cache_headers(response, cache_status="hit-revision", revision=revision_token) return _apply_summary_flavor_view( - result.model_copy(update={"ops_runs": await get_ops_run_projection()}), + result.model_copy(update=await _live_summary_sections(session)), include_residuals=include_residuals, flavor=flavor, ) @@ -173,7 +186,7 @@ async def get_summary( cache.schedule_refresh(revision) _summary_cache_headers(response, cache_status="stale", revision=revision_token) return _apply_summary_flavor_view( - cached.model_copy(update={"ops_runs": await get_ops_run_projection()}), + cached.model_copy(update=await _live_summary_sections(session)), include_residuals=include_residuals, flavor=flavor, ) @@ -182,7 +195,7 @@ async def get_summary( cache.store(result, revision) _summary_cache_headers(response, cache_status="miss", revision=revision_token) return _apply_summary_flavor_view( - result.model_copy(update={"ops_runs": await get_ops_run_projection(refresh=force_refresh)}), + result.model_copy(update=await _live_summary_sections(session, refresh=force_refresh)), include_residuals=include_residuals, flavor=flavor, ) diff --git a/api/schemas/agent_message.py b/api/schemas/agent_message.py index 135b733..baf0856 100644 --- a/api/schemas/agent_message.py +++ b/api/schemas/agent_message.py @@ -1,7 +1,78 @@ -from hub_core.schemas.agent_message import MessageCreate, MessageRead, MessageReply +"""Agent message schemas. + +The base shapes come from hub-core. STATE-WP-0093 extends them locally with +broadcast kinds, expiry, supersede and per-reader receipt fields; direct +messages keep the hub-core semantics (the new fields stay at their defaults). +""" +import uuid +from datetime import datetime + +from hub_core.schemas.agent_message import MessageCreate as _HubMessageCreate +from hub_core.schemas.agent_message import MessageRead as _HubMessageRead +from hub_core.schemas.agent_message import MessageReply +from pydantic import BaseModel + + +class MessageCreate(_HubMessageCreate): + # Only meaningful for broadcasts: news | standing. A broadcast without a + # kind defaults to ``news``; direct messages must leave it unset or + # ``message``. + kind: str | None = None + expires_at: datetime | None = None + supersedes_id: uuid.UUID | None = None + + +class MessageRead(_HubMessageRead): + kind: str = "message" + expires_at: datetime | None = None + supersedes_id: uuid.UUID | None = None + # Per-reader receipt state; populated for broadcasts when the request is + # scoped to a reader (``to_agent`` on list, ``reader`` on mark-read/ack). + # For such responses ``read_at`` is the reader's own read time. + reader: str | None = None + delivered_at: datetime | None = None + acknowledged_at: datetime | None = None + + +class MessageMarkRead(BaseModel): + reader: str | None = None + ack: bool = False + + +class MessageAck(BaseModel): + agent: str + + +class NoticeStatus(BaseModel): + id: uuid.UUID + from_agent: str + subject: str + body: str + created_at: datetime + expires_at: datetime | None = None + supersedes_id: uuid.UUID | None = None + acknowledged: list[str] = [] + delivered_only: list[str] = [] + unreached: list[str] = [] + acked_count: int = 0 + delivered_only_count: int = 0 + unreached_count: int = 0 + + +class StandingNoticeDigest(BaseModel): + id: uuid.UUID + subject: str + expires_at: datetime | None = None + acked: int = 0 + pending: int = 0 + __all__ = [ + "MessageAck", "MessageCreate", + "MessageMarkRead", "MessageRead", "MessageReply", + "NoticeStatus", + "StandingNoticeDigest", ] diff --git a/api/schemas/state.py b/api/schemas/state.py index 3b9e53e..7a826ee 100644 --- a/api/schemas/state.py +++ b/api/schemas/state.py @@ -10,6 +10,7 @@ from api.schemas.progress_event import ProgressEventRead from api.schemas.task import TaskRead from api.schemas.topic import TopicWithWorkstreams from api.schemas.suggestion import RankedSuggestionDigest +from api.schemas.agent_message import StandingNoticeDigest from api.schemas.workstream import WorkstreamWithDeps from api.schemas.ops_run import OpsRunProjection @@ -94,6 +95,7 @@ class StateSummary(BaseModel): open_capability_requests: int = 0 ranked_suggestions: list[RankedSuggestionDigest] = [] ops_runs: OpsRunProjection | None = None + standing_notices: list[StandingNoticeDigest] = [] class DashboardWorkplanRow(BaseModel): diff --git a/api/services/message_receipts.py b/api/services/message_receipts.py new file mode 100644 index 0000000..8e12080 --- /dev/null +++ b/api/services/message_receipts.py @@ -0,0 +1,212 @@ +"""Per-recipient broadcast receipts and standing notices (STATE-WP-0093). + +Broadcast visibility is per reader: a ``message_receipts`` row per +(message, agent) records delivery, read and acknowledgement. Direct messages +never get receipts and keep the global ``AgentMessage.read_at``. + +- ``news`` broadcasts stay in a reader's unread inbox until that reader has + any receipt (delivery at orientation is enough). +- ``standing`` broadcasts stay until the reader acknowledges, the notice + expires, or it is superseded (which archives it). +""" +from __future__ import annotations + +import uuid +from collections.abc import Iterable, Sequence +from datetime import datetime, timedelta, timezone + +from sqlalchemy import and_, exists, func, not_, or_, select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from api.models.agent_message import BROADCAST, AgentMessage, MessageReceipt +from api.models.managed_repo import ManagedRepo +from api.schemas.agent_message import NoticeStatus, StandingNoticeDigest + +NEWS_DEFAULT_TTL = timedelta(days=30) + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def is_broadcast(message: AgentMessage) -> bool: + return message.to_agent == BROADCAST + + +def not_expired(now: datetime): + return or_(AgentMessage.expires_at.is_(None), AgentMessage.expires_at > now) + + +def broadcast_unread_clause(reader_values: Sequence[str], now: datetime): + """SQL clause: broadcast is still unread for a reader (any of its slugs).""" + has_any_receipt = exists().where( + and_( + MessageReceipt.message_id == AgentMessage.id, + MessageReceipt.agent.in_(reader_values), + ) + ) + has_ack = exists().where( + and_( + MessageReceipt.message_id == AgentMessage.id, + MessageReceipt.agent.in_(reader_values), + MessageReceipt.acknowledged_at.is_not(None), + ) + ) + return and_( + AgentMessage.to_agent == BROADCAST, + not_expired(now), + or_( + and_(AgentMessage.kind == "standing", not_(has_ack)), + and_(AgentMessage.kind != "standing", not_(has_any_receipt)), + ), + ) + + +async def upsert_receipt( + session: AsyncSession, + message_ids: Iterable[uuid.UUID], + agent: str, + *, + delivered: bool = False, + read: bool = False, + acknowledged: bool = False, + at: datetime | None = None, +) -> None: + """Idempotently stamp receipt fields; the first timestamp wins.""" + ids = list(message_ids) + if not ids or not agent or agent == BROADCAST: + return + at = at or utcnow() + rows = [ + { + "message_id": mid, + "agent": agent, + "delivered_at": at if delivered else None, + "read_at": at if (read or acknowledged) else None, + "acknowledged_at": at if acknowledged else None, + } + for mid in ids + ] + stmt = pg_insert(MessageReceipt).values(rows) + table = MessageReceipt.__table__ + stmt = stmt.on_conflict_do_update( + index_elements=[table.c.message_id, table.c.agent], + set_={ + col: func.coalesce(table.c[col], stmt.excluded[col]) + for col in ("delivered_at", "read_at", "acknowledged_at") + }, + ) + await session.execute(stmt) + + +async def receipts_for( + session: AsyncSession, + message_ids: Iterable[uuid.UUID], + reader_values: Sequence[str], +) -> dict[uuid.UUID, dict[str, datetime | None]]: + """Merged receipt state per message across a reader's historical slugs.""" + ids = list(message_ids) + if not ids: + return {} + result = await session.execute( + select(MessageReceipt).where( + MessageReceipt.message_id.in_(ids), + MessageReceipt.agent.in_(reader_values), + ) + ) + merged: dict[uuid.UUID, dict[str, datetime | None]] = {} + for receipt in result.scalars().all(): + state = merged.setdefault( + receipt.message_id, + {"delivered_at": None, "read_at": None, "acknowledged_at": None}, + ) + for field in state: + value = getattr(receipt, field) + if value is not None and (state[field] is None or value < state[field]): + state[field] = value + return merged + + +async def _active_repo_slugs(session: AsyncSession) -> list[str]: + result = await session.execute( + select(ManagedRepo.slug).where(ManagedRepo.status == "active").order_by(ManagedRepo.slug) + ) + return list(result.scalars().all()) + + +async def live_standing_notices( + session: AsyncSession, now: datetime | None = None +) -> list[AgentMessage]: + now = now or utcnow() + result = await session.execute( + select(AgentMessage) + .where( + AgentMessage.to_agent == BROADCAST, + AgentMessage.kind == "standing", + AgentMessage.archived_at.is_(None), + not_expired(now), + ) + .order_by(AgentMessage.created_at.desc()) + ) + return list(result.scalars().all()) + + +async def notice_statuses(session: AsyncSession) -> list[NoticeStatus]: + notices = await live_standing_notices(session) + if not notices: + return [] + repos = await _active_repo_slugs(session) + result = await session.execute( + select(MessageReceipt).where( + MessageReceipt.message_id.in_([n.id for n in notices]) + ) + ) + by_notice: dict[uuid.UUID, list[MessageReceipt]] = {} + for receipt in result.scalars().all(): + by_notice.setdefault(receipt.message_id, []).append(receipt) + statuses: list[NoticeStatus] = [] + for notice in notices: + receipts = by_notice.get(notice.id, []) + acked = sorted({r.agent for r in receipts if r.acknowledged_at is not None}) + seen = {r.agent for r in receipts} + delivered_only = sorted(seen - set(acked)) + unreached = [slug for slug in repos if slug not in seen] + statuses.append( + NoticeStatus( + id=notice.id, + from_agent=notice.from_agent, + subject=notice.subject, + body=notice.body, + created_at=notice.created_at, + expires_at=notice.expires_at, + supersedes_id=notice.supersedes_id, + acknowledged=acked, + delivered_only=delivered_only, + unreached=unreached, + acked_count=len(acked), + delivered_only_count=len(delivered_only), + unreached_count=len(unreached), + ) + ) + return statuses + + +async def standing_notice_digests(session: AsyncSession) -> list[StandingNoticeDigest]: + statuses = await notice_statuses(session) + if not statuses: + return [] + repos = set(await _active_repo_slugs(session)) + digests = [] + for status in statuses: + acked_repos = len(repos.intersection(status.acknowledged)) + digests.append( + StandingNoticeDigest( + id=status.id, + subject=status.subject, + expires_at=status.expires_at, + acked=acked_repos, + pending=len(repos) - acked_repos, + ) + ) + return digests diff --git a/dashboard/src/data/notices.json.py b/dashboard/src/data/notices.json.py new file mode 100755 index 0000000..536d101 --- /dev/null +++ b/dashboard/src/data/notices.json.py @@ -0,0 +1,15 @@ +#!/usr/bin/env python3 +"""Observable data loader: fetches /messages/notices (live standing notices).""" +import json +import os +import urllib.error +import urllib.request + +API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/") + +try: + with urllib.request.urlopen(f"{API_BASE}/messages/notices", timeout=10) as resp: + data = json.loads(resp.read()) + print(json.dumps(data)) +except urllib.error.URLError as e: + print(json.dumps({"error": str(e), "notices": []})) diff --git a/dashboard/src/docs/dashboard.md b/dashboard/src/docs/dashboard.md index 656f9f6..85a70e1 100644 --- a/dashboard/src/docs/dashboard.md +++ b/dashboard/src/docs/dashboard.md @@ -68,6 +68,7 @@ Current loaders: | `decisions.json.py` | `/decisions/` | | `domains.json.py` | `/domains/` | | `messages.json.py` | `/messages/` | +| `notices.json.py` | `/messages/notices` | | `progress.json.py` | `/progress/` | | `repos.json.py` | `/repos/` | | `sbom.json.py` | `/sbom/aggregated` | diff --git a/dashboard/src/inbox.md b/dashboard/src/inbox.md index b089015..7a33ca0 100644 --- a/dashboard/src/inbox.md +++ b/dashboard/src/inbox.md @@ -11,14 +11,16 @@ import {API, apiFetch, pollDelay, waitForVisible} from "./components/config.js"; const inboxState = (async function*() { let failures = 0; while (true) { - let messages = [], ok = false; + let messages = [], notices = [], ok = false; try { const resp = await apiFetch("/messages/?limit=100"); ok = resp.ok; if (ok) messages = await resp.json(); + const nresp = await apiFetch("/messages/notices"); + if (nresp.ok) notices = await nresp.json(); } catch {} failures = ok ? 0 : failures + 1; - yield {messages, ok, ts: new Date()}; + yield {messages, notices, ok, ts: new Date()}; await waitForVisible(pollDelay({ok, failures})); } })(); @@ -26,12 +28,17 @@ const inboxState = (async function*() { ```js const messages = inboxState.messages ?? []; +const notices = inboxState.notices ?? []; const _ok = inboxState.ok ?? false; const _ts = inboxState.ts; -const unread = messages.filter(m => !m.read_at && !m.archived_at); -const read = messages.filter(m => m.read_at && !m.archived_at); -const archived = messages.filter(m => m.archived_at); +// Broadcast read state is per reader (STATE-WP-0093): the dashboard has no +// reader identity, so live broadcasts get their own section. +const isBc = m => m.to_agent === "broadcast"; +const unread = messages.filter(m => !isBc(m) && !m.read_at && !m.archived_at); +const read = messages.filter(m => !isBc(m) && m.read_at && !m.archived_at); +const broadcasts = messages.filter(m => isBc(m) && !m.archived_at); +const archived = messages.filter(m => m.archived_at); // Group unread by agent for KPI const agentCounts = {}; @@ -55,6 +62,12 @@ const _kpiBox = html`
${unread.length}
+
+ standing notices +
+
${notices.length}
+
+
total
@@ -85,6 +98,39 @@ Inter-agent coordination messages. Agents send messages via `send_message()` MCP --- +## Standing notices + +Live `standing` broadcasts stay in every agent's inbox until that agent +acknowledges them (`POST /messages/{id}/ack`). Repositories listed as not yet +acknowledged are the ones still out of date. + +```js +function repoList(label, slugs, color) { + if (!slugs.length) return ""; + return html`
${label} (${slugs.length}) +
${slugs.join(", ")}
`; +} + +if (notices.length === 0) { + display(html`

No live standing notices.

`); +} else { + display(html`
${notices.map(n => html`
+
+ ${n.from_agent} + standing + expires ${n.expires_at ? new Date(n.expires_at).toLocaleDateString() : "never"} +
+
${n.subject}
+
${n.acked_count} acknowledged · ${n.delivered_only_count} delivered, not acknowledged · ${n.unreached_count} not reached
+ ${repoList("Not yet acknowledged — delivered", n.delivered_only, "#d97706")} + ${repoList("Not yet acknowledged — never reached", n.unreached, "#dc2626")} + ${repoList("Acknowledged", n.acknowledged, "#059669")} +
`)}
`); +} +``` + +--- + ## Unread ```js @@ -96,7 +142,7 @@ function fmtDate(s) { function renderMessage(m, showMarkRead = false) { const isBroadcast = m.to_agent === "broadcast"; - const borderColor = !m.read_at ? "#d97706" : "#6b7280"; + const borderColor = isBroadcast ? "#7c3aed" : (!m.read_at ? "#d97706" : "#6b7280"); async function onMarkRead() { await fetch(`${API}/messages/${m.id}/read`, {method: "PATCH"}); @@ -111,9 +157,10 @@ function renderMessage(m, showMarkRead = false) { ${m.from_agent} → ${m.to_agent} + ${isBroadcast ? html`${m.kind}` : ""} ${fmtDate(m.created_at)}
- ${showMarkRead ? html`` : ""} + ${showMarkRead && !isBroadcast ? html`` : ""}
@@ -135,6 +182,18 @@ if (unread.length === 0) { --- +## Broadcasts + +```js +if (broadcasts.length === 0) { + display(html`

No live broadcasts.

`); +} else { + display(html`
${broadcasts.map(m => renderMessage(m, false))}
`); +} +``` + +--- + ## Read ```js @@ -186,4 +245,9 @@ if (archived.length === 0) { .msg-body { white-space: pre-wrap; margin: 0.4rem 0 0; font-family: var(--mono); font-size: 0.8rem; background: var(--theme-background); padding: 0.5rem; border-radius: 4px; } .msg-thread { font-size: 0.7rem; color: var(--theme-foreground-muted, #999); margin-top: 0.2rem; } .dim { color: gray; font-style: italic; } +.msg-kind { font-size: 0.7rem; font-weight: 600; color: #7c3aed; border: 1px solid #c4b5fd; border-radius: 4px; padding: 0 0.3rem; } +.notice-counts { font-size: 0.8rem; color: var(--theme-foreground-muted, #555); margin: 0.2rem 0; } +.notice-list { font-size: 0.8rem; margin-top: 0.2rem; } +.notice-list summary { cursor: pointer; font-weight: 600; } +.notice-slugs { font-family: var(--mono); font-size: 0.75rem; padding: 0.3rem 0; } diff --git a/mcp_server/TOOLS.md b/mcp_server/TOOLS.md index 5875faf..7db22c6 100644 --- a/mcp_server/TOOLS.md +++ b/mcp_server/TOOLS.md @@ -247,8 +247,15 @@ Use `"broadcast"` as `to_agent` to send to all agents. |------|----------|-------------| | `get_messages(to_agent?, from_agent?, unread_only?, limit?)` | `to_agent`: your agent name; `unread_only`: True recommended at session start | Check for pending coordination messages. | | `send_message(from_agent, to_agent, subject, body, thread_id?)` | all except `thread_id` required | Send a coordination message to another agent (or broadcast). | -| `mark_message_read(message_id)` | `message_id`: UUID | Mark a message as read after acting on it. | -| `reply_to_message(message_id, from_agent, body)` | all required | Reply in-thread; marks original as read. | +| `mark_message_read(message_id, reader?)` | `message_id`: UUID; `reader`: your repo slug | Mark a message as read after acting on it. For broadcasts pass `reader` — read state is per reader; without it the call is a deprecated no-op for broadcasts (STATE-WP-0093). | +| `acknowledge_notice(message_id, agent)` | both required | Acknowledge a `standing` broadcast notice after acting on it; clears it from `agent`'s inbox (`POST /messages/{id}/ack`). Codex server now; main server pending hub-core. | +| `reply_to_message(message_id, from_agent, body)` | all required | Reply in-thread; marks original as read (for a broadcast: for the replier only). | + +Broadcast kinds (STATE-WP-0093): `news` (default for `to_agent="broadcast"`, +30-day default expiry) appears once per reader — the unread inbox call records +delivery; `standing` stays in each reader's unread inbox until acknowledged, +expired or superseded (`supersedes_id` archives the predecessor). Live notices +and who has acknowledged them: `GET /messages/notices`. Dashboard: `http://localhost:3000/inbox` diff --git a/mcp_server/codex_server.py b/mcp_server/codex_server.py index ce2883f..6d2a154 100644 --- a/mcp_server/codex_server.py +++ b/mcp_server/codex_server.py @@ -10,6 +10,7 @@ from __future__ import annotations import json import os from typing import Any +from urllib.parse import quote import httpx from fastmcp import FastMCP @@ -59,9 +60,22 @@ def get_messages(to_agent: str, unread_only: bool = True) -> str: @mcp.tool() -def mark_message_read(message_id: str) -> str: - """Mark one coordination message as read.""" - return _json(_request("PATCH", f"/messages/{message_id}/read", {})) +def mark_message_read(message_id: str, reader: str | None = None) -> str: + """Mark one coordination message as read. + + Pass ``reader`` (your repository slug) so a broadcast is marked read for + you only; without it a broadcast stays visible and the call is deprecated. + """ + path = f"/messages/{message_id}/read" + if reader: + path += f"?reader={quote(reader, safe='')}" + return _json(_request("PATCH", path, {})) + + +@mcp.tool() +def acknowledge_notice(message_id: str, agent: str) -> str: + """Acknowledge a standing notice after acting on it (clears it for ``agent``).""" + return _json(_request("POST", f"/messages/{message_id}/ack", {"agent": agent})) @mcp.tool() diff --git a/migrations/versions/d7e8f9a0b1c2_broadcast_receipts_and_notices.py b/migrations/versions/d7e8f9a0b1c2_broadcast_receipts_and_notices.py new file mode 100644 index 0000000..5a449b4 --- /dev/null +++ b/migrations/versions/d7e8f9a0b1c2_broadcast_receipts_and_notices.py @@ -0,0 +1,92 @@ +"""per-recipient broadcast receipts and standing notices (STATE-WP-0093) + +Creates ``message_receipts`` (per-reader delivered/read/acknowledged state for +broadcasts) and adds ``kind``, ``expires_at`` and ``supersedes_id`` to +``agent_messages``. Existing broadcasts are backfilled to ``kind='news'`` and, +per the founder decision on D6 (2026-09-22), archived so that none of them +re-surfaces once broadcast visibility stops depending on the global +``read_at``. Direct messages are not touched. + +Downgrade drops the table, index and columns. Archive stamps stay (they are +reversible through the API and harmless under the old global read state). + +Revision ID: d7e8f9a0b1c2 +Revises: c6f7a8b9d0e1 +""" +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision = "d7e8f9a0b1c2" +down_revision = "c6f7a8b9d0e1" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "agent_messages", + sa.Column( + "kind", + sa.String(length=20), + nullable=False, + server_default="message", + ), + ) + op.add_column( + "agent_messages", + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + ) + op.add_column( + "agent_messages", + sa.Column( + "supersedes_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey( + "agent_messages.id", + ondelete="SET NULL", + name="fk_agent_messages_supersedes_id", + ), + nullable=True, + ), + ) + op.create_index( + "ix_agent_messages_to_kind_archived", + "agent_messages", + ["to_agent", "kind", "archived_at"], + ) + op.create_table( + "message_receipts", + sa.Column( + "message_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("agent_messages.id", ondelete="CASCADE"), + primary_key=True, + ), + sa.Column("agent", sa.String(length=100), primary_key=True), + sa.Column("delivered_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("read_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("acknowledged_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index("ix_message_receipts_agent", "message_receipts", ["agent"]) + + # D6: existing broadcasts become news and are archived (only broadcasts). + op.execute( + "UPDATE agent_messages SET kind = 'news' WHERE to_agent = 'broadcast'" + ) + op.execute( + "UPDATE agent_messages SET archived_at = now() " + "WHERE to_agent = 'broadcast' AND archived_at IS NULL" + ) + + +def downgrade() -> None: + op.drop_index("ix_message_receipts_agent", table_name="message_receipts") + op.drop_table("message_receipts") + op.drop_index("ix_agent_messages_to_kind_archived", table_name="agent_messages") + op.drop_constraint( + "fk_agent_messages_supersedes_id", "agent_messages", type_="foreignkey" + ) + op.drop_column("agent_messages", "supersedes_id") + op.drop_column("agent_messages", "expires_at") + op.drop_column("agent_messages", "kind") diff --git a/tests/test_codex_mcp_server.py b/tests/test_codex_mcp_server.py index 5c41efb..c0071f8 100644 --- a/tests/test_codex_mcp_server.py +++ b/tests/test_codex_mcp_server.py @@ -9,6 +9,7 @@ def test_codex_server_has_only_repository_coordination_tools() -> None: "get_domain_summary", "get_messages", "mark_message_read", + "acknowledge_notice", "add_progress_event", "record_decision", "update_task_status", @@ -17,3 +18,18 @@ def test_codex_server_has_only_repository_coordination_tools() -> None: def test_codex_server_uses_distinct_server_identity() -> None: assert codex_server.mcp.name == "dev-hub-codex" + + +def test_mark_message_read_passes_reader(monkeypatch) -> None: + calls = [] + monkeypatch.setattr( + codex_server, "_request", lambda method, path, body=None: calls.append((method, path, body)) or {} + ) + codex_server.mark_message_read("abc", reader="state-hub") + codex_server.mark_message_read("abc") + codex_server.acknowledge_notice("abc", "state-hub") + assert calls == [ + ("PATCH", "/messages/abc/read?reader=state-hub", {}), + ("PATCH", "/messages/abc/read", {}), + ("POST", "/messages/abc/ack", {"agent": "state-hub"}), + ] diff --git a/tests/test_hub_core_imports.py b/tests/test_hub_core_imports.py index b3765da..f241d68 100644 --- a/tests/test_hub_core_imports.py +++ b/tests/test_hub_core_imports.py @@ -37,8 +37,12 @@ from hub_core.schemas.tpsc import ( ) -def test_state_hub_reexports_core_message_schema() -> None: - assert MessageCreate is CoreMessageCreate +def test_state_hub_message_schema_extends_core() -> None: + # STATE-WP-0093 extends the hub-core shape locally (kind, expires_at, + # supersedes_id) until hub-core absorbs the fields; then this returns to + # an identity re-export. + assert issubclass(MessageCreate, CoreMessageCreate) + assert set(CoreMessageCreate.model_fields) <= set(MessageCreate.model_fields) def test_state_hub_reexports_core_doi_schema() -> None: diff --git a/tests/test_message_receipts.py b/tests/test_message_receipts.py new file mode 100644 index 0000000..7c7f016 --- /dev/null +++ b/tests/test_message_receipts.py @@ -0,0 +1,392 @@ +"""STATE-WP-0093: per-recipient broadcast receipts and standing notices.""" +from __future__ import annotations + +import importlib +import uuid +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import inspect, select, text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from api.models.agent_message import AgentMessage, MessageReceipt +from api.models.legacy_meter import LegacyInterface +from api.models.managed_repo import ManagedRepo +from api.models.repository_rename import RepositorySlug +from api.routers.messages import UNATTRIBUTED_BROADCAST_READ_KEY +from tests.conftest import create_test_domain, create_test_repo + + +async def _send(client, **payload): + body = {"from_agent": "the-custodian", "subject": "s", "body": "b", **payload} + body.setdefault("to_agent", "broadcast") + resp = await client.post("/messages/", json=body) + assert resp.status_code == 201, resp.text + return resp.json() + + +async def _inbox(client, agent, unread_only=True): + resp = await client.get( + "/messages/", params={"to_agent": agent, "unread_only": str(unread_only).lower()} + ) + assert resp.status_code == 200, resp.text + return resp.json() + + +def _ids(messages): + return {m["id"] for m in messages} + + +async def _receipts(test_engine, message_id): + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + result = await session.execute( + select(MessageReceipt).where(MessageReceipt.message_id == uuid.UUID(message_id)) + ) + return {r.agent: r for r in result.scalars().all()} + + +@pytest.mark.asyncio +async def test_broadcast_defaults_to_news_with_expiry(client): + msg = await _send(client) + assert msg["kind"] == "news" + expires = datetime.fromisoformat(msg["expires_at"]) + assert timedelta(days=29) < expires - datetime.now(timezone.utc) <= timedelta(days=30) + + +@pytest.mark.asyncio +async def test_direct_message_rejects_broadcast_fields(client): + resp = await client.post( + "/messages/", + json={"from_agent": "a", "to_agent": "b", "subject": "s", "body": "b", "kind": "standing"}, + ) + assert resp.status_code == 422 + direct = await _send(client, to_agent="b") + assert direct["kind"] == "message" + assert direct["expires_at"] is None + + +@pytest.mark.asyncio +async def test_broadcast_read_by_a_stays_unread_for_b(client, test_engine): + msg = await _send(client, kind="standing") + resp = await client.patch(f"/messages/{msg['id']}/read", params={"reader": "agent-a"}) + assert resp.status_code == 200 + body = resp.json() + assert body["reader"] == "agent-a" + assert body["read_at"] is not None + assert msg["id"] in _ids(await _inbox(client, "agent-b")) + receipts = await _receipts(test_engine, msg["id"]) + assert set(receipts) == {"agent-a", "agent-b"} # b: delivery receipt only + assert receipts["agent-b"].read_at is None + # Global read_at never set for a broadcast. + listing = (await client.get("/messages/")).json() + assert next(m for m in listing if m["id"] == msg["id"])["read_at"] is None + + +@pytest.mark.asyncio +async def test_mark_read_reader_in_json_body(client, test_engine): + msg = await _send(client, kind="standing") + resp = await client.patch(f"/messages/{msg['id']}/read", json={"reader": "agent-a"}) + assert resp.status_code == 200 + assert (await _receipts(test_engine, msg["id"]))["agent-a"].read_at is not None + + +@pytest.mark.asyncio +async def test_unattributed_broadcast_mark_read_is_noop_with_deprecation(client, test_engine): + msg = await _send(client, kind="standing") + resp = await client.patch(f"/messages/{msg['id']}/read", json={}) + assert resp.status_code == 200 + assert resp.headers["Deprecation"] == "true" + assert "reader" in resp.headers["Warning"] + assert resp.json()["read_at"] is None + assert await _receipts(test_engine, msg["id"]) == {} + assert msg["id"] in _ids(await _inbox(client, "agent-a")) + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + meter = ( + await session.execute( + select(LegacyInterface).where( + LegacyInterface.interface_key == UNATTRIBUTED_BROADCAST_READ_KEY + ) + ) + ).scalar_one_or_none() + assert meter is not None + + +@pytest.mark.asyncio +async def test_direct_message_read_archive_reply_unchanged(client, test_engine): + direct = await _send(client, to_agent="agent-b") + assert direct["id"] in _ids(await _inbox(client, "agent-b")) + resp = await client.patch(f"/messages/{direct['id']}/read", json={}) + assert resp.status_code == 200 + assert "Deprecation" not in resp.headers + assert resp.json()["read_at"] is not None + assert direct["id"] not in _ids(await _inbox(client, "agent-b")) + assert await _receipts(test_engine, direct["id"]) == {} + + second = await _send(client, to_agent="agent-b") + reply = await client.post( + f"/messages/{second['id']}/reply", json={"from_agent": "agent-b", "body": "ok"} + ) + assert reply.status_code == 201 + assert reply.json()["to_agent"] == "the-custodian" + assert second["id"] not in _ids(await _inbox(client, "agent-b")) + assert await _receipts(test_engine, second["id"]) == {} + + third = await _send(client, to_agent="agent-b") + archived = (await client.patch(f"/messages/{third['id']}/archive")).json() + assert archived["archived_at"] is not None + assert archived["read_at"] == archived["archived_at"] + + +@pytest.mark.asyncio +async def test_reply_to_broadcast_writes_receipt_for_replier_only(client, test_engine): + msg = await _send(client, kind="standing") + reply = await client.post( + f"/messages/{msg['id']}/reply", json={"from_agent": "agent-a", "body": "noted"} + ) + assert reply.status_code == 201 + receipts = await _receipts(test_engine, msg["id"]) + assert set(receipts) == {"agent-a"} + assert receipts["agent-a"].read_at is not None + listing = (await client.get("/messages/")).json() + assert next(m for m in listing if m["id"] == msg["id"])["read_at"] is None + assert msg["id"] in _ids(await _inbox(client, "agent-b")) + + +@pytest.mark.asyncio +async def test_archive_broadcast_does_not_set_read_at(client): + msg = await _send(client) + archived = (await client.patch(f"/messages/{msg['id']}/archive")).json() + assert archived["archived_at"] is not None + assert archived["read_at"] is None + assert msg["id"] not in _ids(await _inbox(client, "agent-a")) + + +@pytest.mark.asyncio +async def test_delivery_receipt_only_for_scoped_unread_listing(client, test_engine): + msg = await _send(client, kind="standing") + await client.get("/messages/") + await client.get("/messages/", params={"unread_only": "true"}) + await _inbox(client, "agent-a", unread_only=False) + assert await _receipts(test_engine, msg["id"]) == {} + + first = await _inbox(client, "agent-a") + delivered = (await _receipts(test_engine, msg["id"]))["agent-a"].delivered_at + assert delivered is not None + assert first[0]["delivered_at"] is not None + await _inbox(client, "agent-a") + receipts = await _receipts(test_engine, msg["id"]) + assert len(receipts) == 1 + assert receipts["agent-a"].delivered_at == delivered # first timestamp wins + + +@pytest.mark.asyncio +async def test_news_disappears_after_delivery_standing_persists_until_ack(client): + news = await _send(client, kind="news") + standing = await _send(client, kind="standing") + first = _ids(await _inbox(client, "agent-a")) + assert {news["id"], standing["id"]} <= first + second = _ids(await _inbox(client, "agent-a")) + assert news["id"] not in second + assert standing["id"] in second + # Plain read does not clear a standing notice. + await client.patch(f"/messages/{standing['id']}/read", params={"reader": "agent-a"}) + assert standing["id"] in _ids(await _inbox(client, "agent-a")) + ack = await client.post(f"/messages/{standing['id']}/ack", json={"agent": "agent-a"}) + assert ack.status_code == 200 + assert ack.json()["acknowledged_at"] is not None + assert standing["id"] not in _ids(await _inbox(client, "agent-a")) + assert standing["id"] in _ids(await _inbox(client, "agent-b")) + + +@pytest.mark.asyncio +async def test_ack_via_mark_read_flag_and_ack_rejects_direct(client): + standing = await _send(client, kind="standing") + await client.patch( + f"/messages/{standing['id']}/read", params={"reader": "agent-a", "ack": "true"} + ) + assert standing["id"] not in _ids(await _inbox(client, "agent-a")) + direct = await _send(client, to_agent="agent-a") + resp = await client.post(f"/messages/{direct['id']}/ack", json={"agent": "agent-a"}) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_expired_broadcast_hidden(client): + past = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat() + expired = await _send(client, kind="standing", expires_at=past) + assert expired["id"] not in _ids(await _inbox(client, "agent-a")) + assert expired["id"] not in _ids(await _inbox(client, "agent-a", unread_only=False)) + assert (await client.get("/messages/notices")).json() == [] + + +@pytest.mark.asyncio +async def test_supersede_archives_predecessor(client): + old = await _send(client, kind="standing", subject="v1") + new = await _send(client, kind="standing", subject="v2", supersedes_id=old["id"]) + assert new["supersedes_id"] == old["id"] + inbox = _ids(await _inbox(client, "agent-a")) + assert new["id"] in inbox and old["id"] not in inbox + notices = (await client.get("/messages/notices")).json() + assert [n["id"] for n in notices] == [new["id"]] + missing = await client.post( + "/messages/", + json={ + "from_agent": "x", "to_agent": "broadcast", "subject": "s", "body": "b", + "kind": "standing", "supersedes_id": str(uuid.uuid4()), + }, + ) + assert missing.status_code == 404 + + +@pytest.mark.asyncio +async def test_notices_view_and_summary_counts(client): + domain = await create_test_domain(client) + for slug in ("repo-a", "repo-b", "repo-c"): + await create_test_repo(client, domain_slug=domain["slug"], slug=slug) + notice = await _send(client, kind="standing", subject="Read the orientation") + await _send(client, kind="news", subject="not a notice") + await client.post(f"/messages/{notice['id']}/ack", json={"agent": "repo-a"}) + await _inbox(client, "repo-b") + + notices = (await client.get("/messages/notices")).json() + assert len(notices) == 1 + status = notices[0] + assert status["acknowledged"] == ["repo-a"] + assert status["delivered_only"] == ["repo-b"] + assert status["unreached"] == ["repo-c"] + assert (status["acked_count"], status["delivered_only_count"], status["unreached_count"]) == (1, 1, 1) + + summary = (await client.get("/state/summary")).json() + assert summary["standing_notices"] == [ + { + "id": notice["id"], + "subject": "Read the orientation", + "expires_at": None, + "acked": 1, + "pending": 2, + } + ] + + +@pytest.mark.asyncio +async def test_renamed_repo_reader_resolves_to_canonical_receipt(client, test_engine): + domain = await create_test_domain(client) + repo = await create_test_repo(client, domain_slug=domain["slug"], slug="flex-auth") + notice = await _send(client, kind="standing") + await client.post(f"/messages/{notice['id']}/ack", json={"agent": "flex-auth"}) + + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + managed = await session.get(ManagedRepo, uuid.UUID(repo["id"])) + old = ( + await session.execute(select(RepositorySlug).where(RepositorySlug.slug == "flex-auth")) + ).scalar_one() + old.kind = "alias" + old.protected = True + managed.slug = "access-engine" + managed.name = "Access Engine" + session.add( + RepositorySlug(repo_id=managed.id, slug="access-engine", kind="canonical", protected=True) + ) + await session.commit() + + # The pre-rename ack still counts for the new name, and new writes via the + # old name land on the canonical slug. + assert notice["id"] not in _ids(await _inbox(client, "access-engine")) + second = await _send(client, kind="standing", subject="second") + await client.patch(f"/messages/{second['id']}/read", params={"reader": "flex-auth"}) + assert set(await _receipts(test_engine, second["id"])) == {"access-engine"} + + +@pytest.mark.asyncio +async def test_capability_request_broadcast_is_news(client, test_engine): + await create_test_domain(client) + resp = await client.post( + "/capability-requests/", + json={ + "title": "Need a thing", + "requesting_agent": "agent-a", + "requesting_domain": "infotech", + "capability_type": "service", + "description": "unmatched", + }, + ) + assert resp.status_code == 201, resp.text + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + msgs = ( + await session.execute(select(AgentMessage).where(AgentMessage.to_agent == "broadcast")) + ).scalars().all() + assert msgs and all(m.kind == "news" and m.expires_at is not None for m in msgs) + + +@pytest.mark.asyncio +async def test_migration_archives_broadcasts_and_is_reversible(test_engine): + migration = importlib.import_module( + "migrations.versions.d7e8f9a0b1c2_broadcast_receipts_and_notices" + ) + from alembic.migration import MigrationContext + from alembic.operations import Operations + + schema = f"receipts_migration_{uuid.uuid4().hex}" + direct_id, read_bc, unread_bc, archived_bc = (uuid.uuid4() for _ in range(4)) + old_archive = datetime(2026, 1, 1, tzinfo=timezone.utc) + + async with test_engine.connect() as connection: + def run(sync_connection): + quoted = sync_connection.dialect.identifier_preparer.quote(schema) + sync_connection.exec_driver_sql(f"CREATE SCHEMA {quoted}") + sync_connection.exec_driver_sql(f"SET LOCAL search_path TO {quoted}") + sync_connection.exec_driver_sql( + "CREATE TABLE agent_messages (" + "id uuid PRIMARY KEY, from_agent varchar(100) NOT NULL, " + "to_agent varchar(100) NOT NULL, subject varchar(500) NOT NULL, " + "body text NOT NULL, thread_id uuid REFERENCES agent_messages(id), " + "read_at timestamptz, archived_at timestamptz, " + "created_at timestamptz NOT NULL DEFAULT now())" + ) + sync_connection.execute( + text( + "INSERT INTO agent_messages (id, from_agent, to_agent, subject, body, read_at, archived_at) VALUES " + "(:d, 'a', 'b', 's', 'b', NULL, NULL), " + "(:r, 'gate-house', 'broadcast', 's', 'b', now(), NULL), " + "(:u, 'system', 'broadcast', 's', 'b', NULL, NULL), " + "(:x, 'system', 'broadcast', 's', 'b', now(), :old)" + ), + {"d": direct_id, "r": read_bc, "u": unread_bc, "x": archived_bc, "old": old_archive}, + ) + original_op = migration.op + migration.op = Operations(MigrationContext.configure(sync_connection)) + try: + migration.upgrade() + assert "message_receipts" in inspect(sync_connection).get_table_names(schema=schema) + rows = { + r.id: r + for r in sync_connection.execute( + text("SELECT id, kind, read_at, archived_at FROM agent_messages") + ) + } + assert rows[direct_id].kind == "message" + assert rows[direct_id].archived_at is None + assert rows[direct_id].read_at is None + for bc in (read_bc, unread_bc, archived_bc): + assert rows[bc].kind == "news" + assert rows[bc].archived_at is not None + assert rows[archived_bc].archived_at == old_archive + assert rows[unread_bc].read_at is None # archive does not stamp read_at + + migration.downgrade() + columns = { + c["name"] for c in inspect(sync_connection).get_columns("agent_messages", schema=schema) + } + assert not {"kind", "expires_at", "supersedes_id"} & columns + assert "message_receipts" not in inspect(sync_connection).get_table_names(schema=schema) + count = sync_connection.execute(text("SELECT count(*) FROM agent_messages")).scalar() + assert count == 4 + finally: + migration.op = original_op + + await connection.run_sync(run) + await connection.rollback() diff --git a/workplans/STATE-WP-0093-broadcast-receipts-and-standing-notices.md b/workplans/STATE-WP-0093-broadcast-receipts-and-standing-notices.md index bff0239..5e71b6f 100644 --- a/workplans/STATE-WP-0093-broadcast-receipts-and-standing-notices.md +++ b/workplans/STATE-WP-0093-broadcast-receipts-and-standing-notices.md @@ -4,12 +4,12 @@ type: workplan title: "Per-recipient broadcast receipts and standing notices" domain: infotech repo: state-hub -status: proposed +status: active owner: claude topic_slug: infotech flavor: implementation created: "2026-09-21" -updated: "2026-09-21" +updated: "2026-09-22" related: - STATE-WP-0091 origin: founder-direction @@ -150,7 +150,7 @@ acknowledgement needs one instruction line, changed once: ```task id: STATE-WP-0093-T01 -status: wait +status: done priority: high state_hub_task_id: "94eec945-b728-50cb-aefb-60eca5aeccd9" ``` @@ -162,11 +162,28 @@ broadcasts (D6). Record the choices as decisions in this file. Done when D2, D3 and D6 each have a recorded founder choice and this workplan moves to `ready`. +**Founder decisions (Bernd Worsch, 2026-09-22) — all accepted as recommended:** + +1. **D3 ACCEPTED:** `GET /messages/?to_agent=X&unread_only=true` records an + idempotent `delivered_at` receipt for X on each broadcast it returns (a + deliberate write inside the GET; no protocol change for delivery tracking). +2. **D6 ACCEPTED:** the migration ARCHIVES the 5 existing broadcasts + (reversible; nothing re-surfaces). T08 asks gate-house whether to + republish its v0.7 "start here" note as a standing notice. +3. **D2/D7 ACCEPTED:** unattributed mark-read of a broadcast is a visibility + no-op with deprecation headers; one line in + `scripts/project_rules/session-protocol.template` and + `agents-codex.template` (inbox curl gains `?reader={REPO_SLUG}` plus an + ack note), carried ONCE by `update_agent_instruction_files.py`. The fleet + propagation run itself is T08, not part of the implementation session. + +T07 (release) still waits on its own founder go-ahead. + ## Schema and migration ```task id: STATE-WP-0093-T02 -status: wait +status: progress priority: high state_hub_task_id: "6680d408-ebce-5b2c-8a97-84f0bf011754" ``` @@ -182,11 +199,21 @@ Downgrade drops the table and columns (archive stamps stay). Done when `alembic upgrade head` and `downgrade -1` both run clean on a copy of a production dump, and the 5 broadcasts end in the D6 state. +**2026-09-22 (progress):** `MessageReceipt` model and Alembic revision +`d7e8f9a0b1c2` (on `c6f7a8b9d0e1`) implemented: `message_receipts`, +`kind`/`expires_at`/`supersedes_id`, both indexes, broadcasts backfilled to +`news` and archived (D6); direct messages untouched. Verified on a scratch +local database built from the full migration chain (upgrade head, downgrade +-1, upgrade head all clean; a read broadcast ended `news` + archived, a direct +message unchanged) and by an isolated-schema up/down test. **Remaining:** the +rehearsal on a copy of a production dump — not run from the implementation +session (no production access); do it in the T07 preflight. + ## API: receipts, notices, ack ```task id: STATE-WP-0093-T03 -status: wait +status: done priority: high state_hub_task_id: "33010aa6-326d-5f52-9ff3-32b09c9e88c4" ``` @@ -201,11 +228,21 @@ Capability-request broadcasts send `kind='news'`. Done when direct-message behavior is byte-for-byte unchanged in the existing test suite and T06's new tests pass. +**2026-09-22 (done):** D2-D4 in `api/routers/messages.py`, +`api/services/message_receipts.py`, `api/schemas/agent_message.py` (local +subclasses of the hub-core schemas). `PATCH /read?reader=&ack=` (also JSON +body), unattributed broadcast mark-read = no-op + `Deprecation`/`Warning` +headers + legacy-meter key `rest_api:PATCH /messages/{id}/read +broadcast-without-reader`, broadcast reply writes a receipt for the replier, +`POST /messages/{id}/ack`, news default expiry 30d, supersede archives the +predecessor, expiry filtered at query time, capability-request broadcasts are +`news`. Direct-message paths unchanged; existing suite green. + ## MCP parity (state-hub and hub-core) ```task id: STATE-WP-0093-T04 -status: wait +status: progress priority: medium state_hub_task_id: "43529a93-6e49-51c8-a358-726f7e737fab" ``` @@ -222,11 +259,21 @@ Done when both MCP servers can mark a broadcast read per reader and acknowledge a standing notice, with tests in `tests/test_codex_mcp_server.py` and hub-core's suite. +**2026-09-22 (progress — split):** state-hub part done: +`mcp_server/codex_server.py` `mark_message_read(message_id, reader=None)` and +new `acknowledge_notice(message_id, agent)`, tests in +`tests/test_codex_mcp_server.py`, `mcp_server/TOOLS.md` updated (the Codex +server has no `send_message`). hub-core part handed off via State Hub message +`69fc387c-5bd3-47bc-9a3f-d3d83b6dc213` (addendum +`f8eecebf-cf76-44f9-832e-20401b1b37ec`: absorb the schema fields so state-hub +can return to a pure re-export). **Remaining:** hub-core change + release, +then bump the pinned hub-core here. + ## Visibility: notices view, summary, dashboard ```task id: STATE-WP-0093-T05 -status: wait +status: progress priority: medium state_hub_task_id: "fdb44fd9-ab44-5e26-86b3-a4f0f8a4e701" ``` @@ -237,11 +284,20 @@ Implement D5: `GET /messages/notices`, `standing_notices` in Done when a live standing notice shows acked / delivered-only / unreached repo lists on the endpoint and the dashboard. +**2026-09-22 (progress):** `GET /messages/notices` (acknowledged / +delivered_only / unreached against active `managed_repos`), +`standing_notices` in `/state/summary` (computed per request, outside the +revision cache), dashboard inbox "Standing notices" panel + separate +"Broadcasts" section (no global mark-read for broadcasts), loader +`dashboard/src/data/notices.json.py`. Endpoint and summary covered by tests; +dashboard build passes. **Remaining:** check the panel against a live notice +after T07. + ## Tests ```task id: STATE-WP-0093-T06 -status: wait +status: done priority: high state_hub_task_id: "d68aa1fa-0f09-5d64-a4fc-cbc24180f2c3" ``` @@ -257,6 +313,12 @@ migration state of pre-existing broadcasts. Done when the new tests pass and the full suite is green. +**2026-09-22 (done):** `tests/test_message_receipts.py` (17 tests: all +listed cases, incl. reader in JSON body, ack via mark-read flag, capability +broadcast as news, isolated-schema migration up/down). Full suite: 894 passed +(`tests/test_hub_core_imports.py` message-schema check relaxed from identity +to subclass, see T04). + ## Release through the normal promotion path ```task