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 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 63291@bnt-lap001
Assistant-Session: 8bd77868-ca68-4f49-bb1e-d539ecc0d703

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",

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
+    )

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)
 
 

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=<agent>"
+UNATTRIBUTED_BROADCAST_WARNING = '299 - "broadcast mark-read needs ?reader=<agent>"'
+
+
+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,

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,
     )

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",
 ]

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):

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