Add a dev memo-forwarding fixture and deduplicated intake notifications.

Pose, replay, and conflict stay on an injected transport with local retry
evidence. Notification receipts keep only memo id, question, class, and
submitter, and a resolve conflict stays readable.

Assistant: grok
Assistant-Session: 01a10329-31cb-78d3-b7a3-789e1fb83139

api/main.py

@@ -13,7 +13,7 @@ from api.database import engine
 from api.events import shutdown_publisher
 from api.services.write_idempotency import WriteIdempotencyMiddleware
 from api.routers import decisions, extension_points, intake, ops_runs, progress, state, suggestions, tasks, technical_debt, topics, workstreams, workstream_dependencies
-from api.routers import domains, repos, contributions, sbom, policy, domain_goals, repo_goals, messages, capability_requests, tpsc, services
+from api.routers import domains, repos, contributions, sbom, policy, domain_goals, repo_goals, messages, memo_notifications, capability_requests, tpsc, services
 from api.routers import token_events
 from api.routers import interface_changes
 from api.routers import flows
@@ -130,6 +130,7 @@ app.include_router(repo_goals.router)
 app.include_router(contributions.router)
 app.include_router(sbom.router)
 app.include_router(messages.router)
+app.include_router(memo_notifications.router)
 app.include_router(capability_requests.router)
 app.include_router(suggestions.router)
 app.include_router(tpsc.router)

api/models/__init__.py

@@ -24,6 +24,7 @@ from api.models.contribution import Contribution, ContributionType, Contribution
 from api.models.sbom_snapshot import SBOMSnapshot
 from api.models.sbom_entry import SBOMEntry, Ecosystem
 from api.models.agent_message import AgentMessage, MessageReceipt
+from api.models.memo_notification import MemoNotificationReceipt
 from api.models.capability_catalog import CapabilityCatalog
 from api.models.capability_request import CapabilityRequest
 from api.models.tpsc import TPSCCatalog, TPSCSnapshot, TPSCEntry
@@ -77,6 +78,7 @@ __all__ = [
     "SBOMEntry", "Ecosystem",
     "AgentMessage",
     "MessageReceipt",
+    "MemoNotificationReceipt",
     "CapabilityCatalog",
     "CapabilityRequest",
     "TPSCCatalog", "TPSCSnapshot", "TPSCEntry",

api/models/memo_notification.py

@@ -0,0 +1,37 @@
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, String, func
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from api.models.base import Base
+
+
+class MemoNotificationReceipt(Base):
+    """Durable receipt for one informed-decision delivery id.
+
+    The delivery id is the sender's stable idempotency key. The body is only
+    memo id, question, class, and submitter. Rows are retained, not expired.
+    """
+
+    __tablename__ = "memo_notification_receipts"
+
+    delivery_id: Mapped[str] = mapped_column(String(128), primary_key=True)
+    body_digest: Mapped[str] = mapped_column(String(64), nullable=False)
+    memo_id: Mapped[str] = mapped_column(String(63), nullable=False)
+    question: Mapped[str] = mapped_column(String(400), nullable=False)
+    memo_class: Mapped[str] = mapped_column(String(63), nullable=False)
+    submitter: Mapped[str] = mapped_column(String(63), nullable=False)
+    message_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("agent_messages.id", ondelete="RESTRICT"),
+        nullable=False,
+        index=True,
+    )
+    authoritative_review: Mapped[str] = mapped_column(
+        String(40), nullable=False, default="signed-in-overview"
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )

api/routers/memo_notifications.py

@@ -0,0 +1,97 @@
+"""Notification intake for informed-decision delivery ids.
+
+Intentionally absent from the 14-day write-idempotency allowlist: retention
+is this receipt row, and a conflicting reuse must stay readable by GET.
+"""
+from __future__ import annotations
+
+import json
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import JSONResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.memo_notification import MemoNotificationReceipt
+from api.services.informed_decision_contract import (
+    MAX_NOTIFICATION_BYTES,
+    ContractError,
+    parse_delivery_id,
+    parse_notification,
+)
+from api.services.memo_notification import NotificationConflict, accept_notification
+
+router = APIRouter(prefix="/informed-decision", tags=["informed-decision-notifications"])
+
+
+def _error(status: int, code: str) -> JSONResponse:
+    return JSONResponse(
+        status_code=status,
+        content={"error": code},
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+def _payload(receipt: MemoNotificationReceipt, *, status: int, replay: bool = False) -> JSONResponse:
+    return JSONResponse(
+        status_code=status,
+        content={
+            "delivery_id": receipt.delivery_id,
+            "memo_id": receipt.memo_id,
+            "question": receipt.question,
+            "memo_class": receipt.memo_class,
+            "submitter": receipt.submitter,
+            "message_id": str(receipt.message_id),
+            "status": "accepted",
+            "authoritative_review": receipt.authoritative_review,
+        },
+        headers={
+            "Cache-Control": "no-store",
+            **({"X-StateHub-Idempotency-Replay": "true"} if replay else {}),
+        },
+    )
+
+
+@router.post("/notifications")
+async def post_notification(
+    request: Request, session: AsyncSession = Depends(get_session)
+) -> JSONResponse:
+    raw = await request.body()
+    if len(raw) > MAX_NOTIFICATION_BYTES:
+        return _error(413, "notification_too_large")
+    content_type = request.headers.get("content-type", "")
+    if content_type.split(";")[0].strip().lower() != "application/json":
+        return _error(415, "json_required")
+    try:
+        delivery_id = parse_delivery_id(request.headers.get("idempotency-key"))
+        notice = parse_notification(json.loads(raw))
+    except ContractError as exc:
+        return _error(exc.status, exc.code)
+    except (json.JSONDecodeError, UnicodeError):
+        return _error(400, "invalid_json")
+    try:
+        receipt, replay = await accept_notification(session, delivery_id, notice)
+    except NotificationConflict as exc:
+        return JSONResponse(
+            status_code=409,
+            content={
+                "error": "notification_conflict",
+                "delivery_id": exc.receipt.delivery_id,
+            },
+            headers={"Cache-Control": "no-store"},
+        )
+    return _payload(receipt, status=200 if replay else 201, replay=replay)
+
+
+@router.get("/notifications/{delivery_id}")
+async def get_notification(
+    delivery_id: str, session: AsyncSession = Depends(get_session)
+) -> JSONResponse:
+    try:
+        checked = parse_delivery_id(delivery_id)
+    except ContractError as exc:
+        return _error(exc.status, exc.code)
+    receipt = await session.get(MemoNotificationReceipt, checked)
+    if receipt is None:
+        return _error(404, "notification_not_found")
+    return _payload(receipt, status=200)

api/services/informed_decision_contract.py

@@ -0,0 +1,382 @@
+"""Pure contract for informed-decision forwarding and notifications.
+
+Dev and request validation only. This module does not read a token, open a
+socket, register a submitter, or decide a hub decision. ``drafted_by`` is
+asserted provenance; it is never an authenticated identity.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+KIND = "informed-decision.memo-intake"
+VERSION = 1
+OVERVIEW_LINE = "The signed-in overview is the authoritative review entry point."
+PROVENANCE_NOTE = (
+    "drafted_by is asserted provenance, not authenticated identity. "
+    "A forwarded token proves only the hub submitter."
+)
+RESOLVING_VERBS = frozenset({"accept", "decline"})
+NOTIFICATION_FIELDS = ("memo_id", "question", "memo_class", "submitter")
+MAX_QUESTION = 400
+MAX_DRAFTED_BY = 200
+MAX_PACKET_BYTES = 256 * 1024
+MAX_ENVELOPE_BYTES = 1024 * 1024
+MAX_PACKET_ITEMS = 32
+MAX_NOTIFICATION_BYTES = 8192
+
+_KEY_RE = re.compile(r"^[A-Za-z0-9._:-]{8,128}$")
+_MEMO_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{2,62}$")
+_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{1,62}$")
+_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
+_SHA_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
+_ERROR_RE = re.compile(r"^[a-z0-9_]{1,64}$")
+
+ENVELOPE_KEYS = {
+    "kind", "version", "idempotency_key", "memo_class", "review_group",
+    "drafted_by", "serves", "approval_intent", "memo", "packet",
+}
+ENVELOPE_REQUIRED = ENVELOPE_KEYS - {"drafted_by"}
+INTENT_KEYS = {"action", "principal", "purpose", "target"}
+TARGET_KEYS = {"id", "type", "system"}
+PACKET_KEYS = {"item_id", "media_type", "content"}
+MEDIA_TYPES = {"text/markdown", "text/plain", "application/json"}
+
+# Shapes only. A hit is refused by code; the matched text is never echoed.
+_SECRET_MARKERS = (
+    re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
+    re.compile(r"\bBearer\s+[A-Za-z0-9._~+/=-]{16,}"),
+    re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}"),
+    re.compile(r"\b(hvs|hvb|hvr)\.[A-Za-z0-9_-]{20,}"),
+    re.compile(
+        r"\b(sk-(or-)?[A-Za-z0-9_-]{20,}|ghp_[A-Za-z0-9]{30,}|"
+        r"github_pat_[A-Za-z0-9_]{30,}|glpat-[A-Za-z0-9_-]{20,}|"
+        r"AKIA[0-9A-Z]{16}|xox[abpr]-[A-Za-z0-9-]{10,})"
+    ),
+    re.compile(
+        r"(?i)\b(password|passwd|client_secret|secret|api[_-]?key|access[_-]?token|token)"
+        r"\s*[:=]\s*['\"]?(?=[A-Za-z0-9_+.=-]*\d)[A-Za-z0-9_+.=-]{12,}(?=['\"\s,;)]|$)"
+    ),
+    re.compile(r"://[^/@\s:]+:[^/@\s]+@"),
+)
+
+
+class ContractError(ValueError):
+    def __init__(self, code: str, status: int = 400) -> None:
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+def canonical(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def digest(value: Any) -> str:
+    return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()
+
+
+def scan_secrets(value: Any) -> None:
+    for text in _strings(value):
+        for pattern in _SECRET_MARKERS:
+            if pattern.search(text):
+                raise ContractError("secret_marker")
+
+
+def stable_idempotency_key(decision_id: str, memo_id: str) -> str:
+    _uuid(decision_id, "hub_decision_id")
+    _memo_id(memo_id)
+    return f"state-hub.{decision_id}.{memo_id}"
+
+
+def resolution_applies(verb: str, serves: Mapping | None) -> bool:
+    """Terminal human accept/decline with resolves=on-answer only."""
+    if verb not in RESOLVING_VERBS:
+        return False
+    if not isinstance(serves, Mapping):
+        return False
+    return serves.get("resolves") == "on-answer"
+
+
+def correlate_resolution(
+    memo_id: str, *, post_status: int | None, existing: Mapping | None
+) -> str:
+    """Classify a resolve attempt. A 409 is a conflict only after the existing
+    resolution was read; an unreadable 409 stays retryable."""
+    _memo_id(memo_id)
+    if post_status is None:
+        return "failed"
+    if 200 <= post_status < 300:
+        return "resolved"
+    if post_status != 409:
+        return "failed"
+    if not isinstance(existing, Mapping):
+        return "failed"
+    rationale = existing.get("rationale")
+    if not isinstance(rationale, str):
+        return "failed"
+    if f"memo {memo_id} " in rationale:
+        return "already_resolved"
+    return "conflict"
+
+
+def safe_error_code(value: Any, fallback: str) -> str:
+    if isinstance(value, str) and _ERROR_RE.match(value):
+        return value
+    return fallback
+
+
+@dataclass(frozen=True)
+class Notification:
+    memo_id: str
+    question: str
+    memo_class: str
+    submitter: str
+
+    def as_body(self) -> dict[str, str]:
+        return {
+            "memo_id": self.memo_id,
+            "question": self.question,
+            "memo_class": self.memo_class,
+            "submitter": self.submitter,
+        }
+
+    @property
+    def body_digest(self) -> str:
+        return digest(self.as_body())
+
+    def message_subject(self) -> str:
+        return f"Decision memo {self.memo_id}"
+
+    def message_text(self) -> str:
+        return OVERVIEW_LINE + "\n" + canonical(self.as_body())
+
+
+def parse_delivery_id(value: str | None) -> str:
+    if value is None or value == "":
+        raise ContractError("delivery_id_required")
+    if not isinstance(value, str) or not _KEY_RE.match(value):
+        raise ContractError("delivery_id_invalid")
+    return value
+
+
+def parse_notification(payload: Any) -> Notification:
+    if not isinstance(payload, dict):
+        raise ContractError("invalid_notification")
+    if set(payload) - set(NOTIFICATION_FIELDS):
+        raise ContractError("unknown_field")
+    if set(NOTIFICATION_FIELDS) - set(payload):
+        raise ContractError("missing_field")
+    scan_secrets(payload)
+    if any("://" in text for text in _strings(payload)):
+        raise ContractError("notification_has_link")
+    return Notification(
+        memo_id=_memo_id(payload["memo_id"]),
+        question=_one_question(payload["question"]),
+        memo_class=_slug(payload["memo_class"], "memo_class"),
+        submitter=_slug(payload["submitter"], "submitter"),
+    )
+
+
+def build_envelope(
+    *,
+    decision_id: str,
+    memo: Mapping,
+    packet: list,
+    memo_class: str,
+    review_group: str,
+    approval_intent: Mapping,
+    drafted_by: str | None = None,
+    resolves: str = "on-answer",
+) -> dict:
+    if not isinstance(memo, Mapping) or not isinstance(memo.get("id"), str):
+        raise ContractError("invalid_memo")
+    envelope: dict[str, Any] = {
+        "kind": KIND,
+        "version": VERSION,
+        "idempotency_key": stable_idempotency_key(decision_id, memo["id"]),
+        "memo_class": memo_class,
+        "review_group": review_group,
+        "serves": {"hub_decision_id": decision_id, "resolves": resolves},
+        "approval_intent": dict(approval_intent),
+        "memo": dict(memo),
+        "packet": list(packet),
+    }
+    if drafted_by is not None:
+        envelope["drafted_by"] = drafted_by
+    return validate_envelope(envelope)
+
+
+def validate_envelope(envelope: Any) -> dict:
+    if not isinstance(envelope, dict):
+        raise ContractError("invalid_envelope")
+    extra = set(envelope) - ENVELOPE_KEYS
+    if extra:
+        raise ContractError("unknown_field")
+    missing = ENVELOPE_REQUIRED - set(envelope)
+    if missing:
+        raise ContractError("missing_field")
+    if envelope["kind"] != KIND or type(envelope["version"]) is not int or envelope["version"] != VERSION:
+        raise ContractError("unsupported_kind")
+    scan_secrets(envelope)
+    _packet(envelope["packet"])
+    _bounded_tree(envelope)
+    encoded = canonical(envelope).encode("utf-8")
+    if len(encoded) > MAX_ENVELOPE_BYTES:
+        raise ContractError("envelope_too_large")
+
+    memo = envelope["memo"]
+    if not isinstance(memo, dict):
+        raise ContractError("invalid_memo")
+    memo_id = _memo_id(memo.get("id"))
+    if "question" in memo:
+        _one_question(memo["question"])
+    decision_id = _serves(envelope["serves"])
+    key = _text(envelope["idempotency_key"], pattern=_KEY_RE, code="idempotency_key_invalid")
+    if key != stable_idempotency_key(decision_id, memo_id):
+        raise ContractError("idempotency_key_not_stable")
+    _slug(envelope["memo_class"], "memo_class")
+    _slug(envelope["review_group"], "review_group")
+    drafted_by = envelope.get("drafted_by")
+    if drafted_by is not None:
+        _text(drafted_by, max_len=MAX_DRAFTED_BY, code="drafted_by_invalid")
+    _intent(envelope["approval_intent"])
+    if "submitter" in envelope or "actor" in envelope.get("approval_intent", {}):
+        raise ContractError("unknown_field")
+    return json.loads(canonical(envelope))
+
+
+def submission_digest(envelope: Mapping) -> str:
+    body = {key: value for key, value in envelope.items() if key != "idempotency_key"}
+    return digest(body)
+
+
+def bounded_receipt(body: Any, expected_memo_id: str) -> dict[str, str]:
+    if not isinstance(body, dict):
+        raise ContractError("receipt_incomplete")
+    memo_id = body.get("memo_id")
+    approval_id = body.get("approval_id")
+    binding = body.get("approval_binding_digest")
+    if memo_id != expected_memo_id or not isinstance(approval_id, str) or not _UUID_RE.match(approval_id):
+        raise ContractError("receipt_mismatch")
+    if not isinstance(binding, str) or not _SHA_RE.match(binding):
+        raise ContractError("receipt_incomplete")
+    return {
+        "memo_id": memo_id,
+        "approval_id": approval_id,
+        "approval_binding_digest": binding,
+    }
+
+
+def _serves(serves: Any) -> str:
+    if not isinstance(serves, dict) or set(serves) != {"hub_decision_id", "resolves"}:
+        raise ContractError("invalid_serves")
+    if serves["resolves"] not in ("on-answer", "never"):
+        raise ContractError("unknown_enum")
+    return _uuid(serves["hub_decision_id"], "hub_decision_id")
+
+
+def _intent(intent: Any) -> None:
+    if not isinstance(intent, dict) or set(intent) != INTENT_KEYS:
+        raise ContractError("invalid_intent")
+    for field in ("action", "principal", "purpose"):
+        _text(intent[field], max_len=400, code="invalid_intent")
+    target = intent["target"]
+    if not isinstance(target, dict) or set(target) != TARGET_KEYS:
+        raise ContractError("invalid_intent")
+    for field in TARGET_KEYS:
+        _text(target[field], max_len=200, code="invalid_intent")
+
+
+def _packet(packet: Any) -> None:
+    if not isinstance(packet, list) or not packet or len(packet) > MAX_PACKET_ITEMS:
+        raise ContractError("packet_required" if not packet else "packet_unbounded")
+    seen: set[str] = set()
+    for item in packet:
+        if not isinstance(item, dict) or set(item) != PACKET_KEYS:
+            raise ContractError("invalid_packet")
+        if item["media_type"] not in MEDIA_TYPES:
+            raise ContractError("unsupported_media_type")
+        content = item["content"]
+        if not isinstance(content, str) or len(content.encode("utf-8")) > MAX_PACKET_BYTES:
+            raise ContractError("packet_too_large")
+        item_id = _text(item["item_id"], max_len=200, code="invalid_packet")
+        if item_id in seen:
+            raise ContractError("duplicate_packet_item")
+        seen.add(item_id)
+
+
+def _one_question(value: Any) -> str:
+    text = _text(value, max_len=MAX_QUESTION, code="question_invalid")
+    if "\n" in text or "\r" in text or "://" in text:
+        raise ContractError("question_invalid")
+    if text.count("?") != 1 or not text.endswith("?"):
+        raise ContractError("question_not_one_question")
+    return text
+
+
+def _memo_id(value: Any) -> str:
+    return _text(value, pattern=_MEMO_ID_RE, code="memo_id_invalid")
+
+
+def _slug(value: Any, code: str) -> str:
+    return _text(value, pattern=_SLUG_RE, code=code + "_invalid")
+
+
+def _uuid(value: Any, code: str) -> str:
+    return _text(value, pattern=_UUID_RE, code=code + "_invalid")
+
+
+def _text(value: Any, *, max_len: int = 128, pattern: re.Pattern[str] | None = None, code: str) -> str:
+    if not isinstance(value, str) or not value or len(value) > max_len:
+        raise ContractError(code)
+    if pattern is not None and not pattern.match(value):
+        raise ContractError(code)
+    return value
+
+
+def _bounded_tree(value: Any) -> None:
+    total = 0
+
+    def walk(node: Any, key: str | None = None) -> None:
+        nonlocal total
+        if isinstance(node, str):
+            size = len(node.encode("utf-8"))
+            if size > MAX_PACKET_BYTES:
+                raise ContractError("field_too_large")
+            if key == "question":
+                _one_question(node)
+            total += size
+        elif isinstance(node, dict):
+            for child_key, child in node.items():
+                if not isinstance(child_key, str):
+                    raise ContractError("invalid_type")
+                walk(child_key, None)
+                walk(child, child_key)
+        elif isinstance(node, list):
+            for child in node:
+                walk(child, key)
+        elif type(node) in (int, float) or node is None:
+            return
+        else:
+            raise ContractError("invalid_type")
+        if total > MAX_ENVELOPE_BYTES:
+            raise ContractError("envelope_too_large")
+
+    walk(value)
+
+
+def _strings(value: Any):
+    if isinstance(value, str):
+        yield value
+    elif isinstance(value, dict):
+        for key, child in value.items():
+            if isinstance(key, str):
+                yield key
+            yield from _strings(child)
+    elif isinstance(value, list):
+        for child in value:
+            yield from _strings(child)

api/services/memo_forwarding.py

@@ -0,0 +1,440 @@
+"""Dev-only pose, replay, and conflict fixture.
+
+Not imported by the API runtime and not scheduled. It never reads a token
+file, mints a credential, or dials a default host. Tests inject the transport
+and, when a pose should carry a header, a token callable. The callable's
+value is not retained.
+
+Choosing the workload identity that projects an intake-audience token, and
+reviewing that identity into the submitter registry, is an owner step. This
+fixture does not name or admit an account.
+"""
+from __future__ import annotations
+
+import json
+import sqlite3
+import time
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Callable, Iterator
+
+from api.services.informed_decision_contract import (
+    PROVENANCE_NOTE,
+    ContractError,
+    bounded_receipt,
+    correlate_resolution,
+    resolution_applies,
+    safe_error_code,
+    submission_digest,
+    validate_envelope,
+)
+
+Transport = Callable[[str, str, dict, dict], tuple[int, dict]]
+TokenReader = Callable[[], str]
+
+_RETRYABLE = frozenset({408, 429})
+
+
+@dataclass(frozen=True)
+class ForwardResult:
+    status: str
+    idempotency_key: str
+    receipt: dict | None
+    drafted_by: str | None
+    drafted_by_provenance: str
+    provenance_note: str
+    detail: str = ""
+
+
+class DevIntakeFixture:
+    def __init__(
+        self,
+        path: str | Path,
+        *,
+        transport: Transport,
+        intake_base_url: str,
+        hub_base_url: str,
+        token_reader: TokenReader | None = None,
+        clock: Callable[[], float] = time.time,
+    ) -> None:
+        self.path = str(path)
+        self.transport = transport
+        self.intake_base_url = _base_url(intake_base_url)
+        self.hub_base_url = _base_url(hub_base_url)
+        self.token_reader = token_reader
+        self.clock = clock
+        Path(self.path).parent.mkdir(parents=True, exist_ok=True)
+        with self._connect() as db:
+            db.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS pose_ledger (
+                    idempotency_key TEXT PRIMARY KEY,
+                    submission_digest TEXT NOT NULL,
+                    envelope_json TEXT NOT NULL,
+                    drafted_by TEXT,
+                    drafted_by_provenance TEXT NOT NULL
+                        CHECK (drafted_by_provenance IN ('asserted', 'absent')),
+                    provenance_note TEXT NOT NULL,
+                    state TEXT NOT NULL
+                        CHECK (state IN ('pending', 'delivered', 'blocked')),
+                    receipt_json TEXT,
+                    attempts INTEGER NOT NULL DEFAULT 0,
+                    next_attempt REAL NOT NULL DEFAULT 0,
+                    last_error TEXT,
+                    blocked_reason TEXT,
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS resolution_ledger (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    decision_id TEXT NOT NULL,
+                    memo_id TEXT NOT NULL,
+                    verb TEXT NOT NULL,
+                    resolves TEXT,
+                    status TEXT NOT NULL,
+                    error_type TEXT,
+                    payload_json TEXT,
+                    next_attempt REAL,
+                    created_at TEXT NOT NULL
+                );
+                """
+            )
+
+    def submit(self, envelope: dict) -> ForwardResult:
+        checked = validate_envelope(envelope)
+        key = checked["idempotency_key"]
+        body_digest = submission_digest(checked)
+        drafted_by = checked.get("drafted_by")
+        provenance = "asserted" if drafted_by else "absent"
+        now = self.clock()
+        early: tuple[str, str] | None = None
+        with self._connect() as db:
+            row = db.execute(
+                "SELECT * FROM pose_ledger WHERE idempotency_key=?", (key,)
+            ).fetchone()
+            if row is None:
+                db.execute(
+                    """INSERT INTO pose_ledger (
+                        idempotency_key, submission_digest, envelope_json, drafted_by,
+                        drafted_by_provenance, provenance_note, state, attempts,
+                        next_attempt, created_at, updated_at
+                    ) VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, 0, ?, ?)""",
+                    (
+                        key, body_digest, _dump(checked), drafted_by, provenance,
+                        PROVENANCE_NOTE, _iso(now), _iso(now),
+                    ),
+                )
+            elif row["submission_digest"] != body_digest:
+                db.execute(
+                    """UPDATE pose_ledger SET blocked_reason='idempotency_conflict',
+                       updated_at=? WHERE idempotency_key=?""",
+                    (_iso(now), key),
+                )
+                if row["state"] != "delivered":
+                    db.execute(
+                        "UPDATE pose_ledger SET state='blocked' WHERE idempotency_key=?",
+                        (key,),
+                    )
+                early = ("blocked", "idempotency_conflict")
+            elif row["state"] == "delivered":
+                early = ("replay", "replay")
+            elif row["state"] == "blocked":
+                early = ("blocked", row["blocked_reason"] or "blocked")
+            elif row["next_attempt"] > now:
+                early = ("pending", row["last_error"] or "waiting")
+        if early is not None:
+            return self._result(key, *early)
+        return self._attempt(key)
+
+    def run_due(self) -> ForwardResult | None:
+        now = self.clock()
+        with self._connect() as db:
+            row = db.execute(
+                """SELECT idempotency_key FROM pose_ledger
+                   WHERE state='pending' AND next_attempt<=?
+                   ORDER BY created_at LIMIT 1""",
+                (now,),
+            ).fetchone()
+            key = None if row is None else row["idempotency_key"]
+        if key is None:
+            return None
+        return self._attempt(key)
+
+    def pose(self, key: str) -> dict | None:
+        with self._connect() as db:
+            row = db.execute(
+                "SELECT * FROM pose_ledger WHERE idempotency_key=?", (key,)
+            ).fetchone()
+            return dict(row) if row is not None else None
+
+    def observe_resolution(
+        self, *, decision_id: str, memo_id: str, verb: str, serves: dict | None, payload: dict | None
+    ) -> str:
+        resolves = serves.get("resolves") if isinstance(serves, dict) else None
+        if not resolution_applies(verb, serves):
+            self._append_resolution(
+                decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
+                status="not_applicable", error_type=None, payload=None, retry_at=None,
+            )
+            return "not_applicable"
+        prior = self._latest_resolution(decision_id, memo_id)
+        if prior and prior["status"] in {"resolved", "already_resolved", "conflict"}:
+            return prior["status"]
+        if prior and prior["status"] == "failed":
+            if prior["next_attempt"] is not None and prior["next_attempt"] > self.clock():
+                return "failed"
+            if payload is None and prior["payload_json"]:
+                payload = json.loads(prior["payload_json"])
+        if not isinstance(payload, dict):
+            raise ValueError("resolution payload is required")
+        encoded = _dump(payload)
+        if len(encoded.encode("utf-8")) > 8192:
+            self._append_resolution(
+                decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
+                status="conflict", error_type="payload_unbounded", payload=None, retry_at=None,
+            )
+            return "conflict"
+        try:
+            status, _body = self.transport(
+                "POST",
+                f"{self.hub_base_url}/decisions/{decision_id}/resolve",
+                payload,
+                {},
+            )
+            status = int(status)
+        except OSError as exc:
+            self._append_resolution(
+                decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
+                status="failed", error_type=type(exc).__name__, payload=payload,
+                retry_at=self.clock() + _backoff(1),
+            )
+            return "failed"
+        except (TypeError, ValueError):
+            self._append_resolution(
+                decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
+                status="failed", error_type="transport_invalid", payload=payload,
+                retry_at=self.clock() + _backoff(1),
+            )
+            return "failed"
+        existing = None
+        if status == 409:
+            try:
+                read_status, existing = self.transport(
+                    "GET", f"{self.hub_base_url}/decisions/{decision_id}", {}, {}
+                )
+                if int(read_status) != 200 or not isinstance(existing, dict):
+                    existing = None
+                    raise OSError("resolution_unreadable")
+            except OSError as exc:
+                error = exc.__class__.__name__ if str(exc) != "resolution_unreadable" else "resolution_unreadable"
+                self._append_resolution(
+                    decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
+                    status="failed", error_type=error, payload=payload,
+                    retry_at=self.clock() + _backoff(1),
+                )
+                return "failed"
+        outcome = correlate_resolution(memo_id, post_status=status, existing=existing)
+        self._append_resolution(
+            decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
+            status=outcome, error_type=None if outcome != "failed" else f"http_{status}",
+            payload=payload, retry_at=None if outcome != "failed" else self.clock() + _backoff(1),
+        )
+        return outcome
+
+    def run_due_resolution(self) -> str | None:
+        now = self.clock()
+        with self._connect() as db:
+            row = db.execute(
+                """SELECT * FROM resolution_ledger
+                   WHERE status='failed' AND next_attempt IS NOT NULL AND next_attempt<=?
+                   ORDER BY id DESC LIMIT 1""",
+                (now,),
+            ).fetchone()
+            pending = dict(row) if row is not None else None
+        if pending is None:
+            return None
+        payload = json.loads(pending["payload_json"]) if pending["payload_json"] else None
+        serves = {"hub_decision_id": pending["decision_id"], "resolves": pending["resolves"]}
+        return self.observe_resolution(
+            decision_id=pending["decision_id"],
+            memo_id=pending["memo_id"],
+            verb=pending["verb"],
+            serves=serves,
+            payload=payload,
+        )
+
+    def resolutions(self) -> list[dict]:
+        with self._connect() as db:
+            rows = db.execute(
+                "SELECT * FROM resolution_ledger ORDER BY id"
+            ).fetchall()
+            return [dict(row) for row in rows]
+
+    def _attempt(self, key: str) -> ForwardResult:
+        with self._connect() as db:
+            row = db.execute(
+                "SELECT * FROM pose_ledger WHERE idempotency_key=?", (key,)
+            ).fetchone()
+            current = dict(row) if row is not None else None
+        if current is None or current["state"] != "pending":
+            return self._result(key, current["state"] if current else "blocked", "not_pending")
+        envelope = json.loads(current["envelope_json"])
+        token, token_error = self._read_token()
+        if token_error:
+            self._mark_retry(key, token_error)
+            return self._result(key, "pending", token_error)
+        headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
+        try:
+            status, body = self.transport(
+                "POST", f"{self.intake_base_url}/v1/memos", envelope, headers
+            )
+            status = int(status)
+        except OSError as exc:
+            self._mark_retry(key, type(exc).__name__)
+            return self._result(key, "pending", type(exc).__name__)
+        except (TypeError, ValueError):
+            self._mark_retry(key, "transport_invalid")
+            return self._result(key, "pending", "transport_invalid")
+        if status in (200, 201):
+            try:
+                receipt = bounded_receipt(body, envelope["memo"]["id"])
+            except ContractError as exc:
+                code = safe_error_code(exc.code, "receipt_mismatch")
+                self._mark_blocked(key, code)
+                return self._result(key, "blocked", code)
+            self._mark_delivered(key, receipt)
+            return self._result(key, "delivered", "delivered")
+        if status == 409 or (status not in _RETRYABLE and 400 <= status < 500):
+            code = safe_error_code(
+                body.get("error") if isinstance(body, dict) else None,
+                "conflict" if status == 409 else f"http_{status}",
+            )
+            self._mark_blocked(key, code)
+            return self._result(key, "blocked", code)
+        self._mark_retry(key, f"http_{status}")
+        return self._result(key, "pending", f"http_{status}")
+
+    def _read_token(self) -> tuple[str | None, str | None]:
+        if self.token_reader is None:
+            return None, "projected_token_absent"
+        try:
+            token = self.token_reader()
+        except Exception as exc:
+            return None, type(exc).__name__
+        if not isinstance(token, str) or not token.strip() or token != token.strip():
+            return None, "projected_token_absent"
+        if len(token) > 16384:
+            return None, "projected_token_absent"
+        return token, None
+
+    def _mark_retry(self, key: str, error: str) -> None:
+        now = self.clock()
+        with self._connect() as db:
+            attempts = db.execute(
+                "SELECT attempts FROM pose_ledger WHERE idempotency_key=?", (key,)
+            ).fetchone()["attempts"] + 1
+            db.execute(
+                """UPDATE pose_ledger SET state='pending', attempts=?, last_error=?,
+                   next_attempt=?, updated_at=? WHERE idempotency_key=?""",
+                (attempts, error, now + _backoff(attempts), _iso(now), key),
+            )
+
+    def _mark_blocked(self, key: str, reason: str) -> None:
+        now = self.clock()
+        with self._connect() as db:
+            db.execute(
+                """UPDATE pose_ledger SET state='blocked', blocked_reason=?, last_error=?,
+                   updated_at=? WHERE idempotency_key=?""",
+                (reason, reason, _iso(now), key),
+            )
+
+    def _mark_delivered(self, key: str, receipt: dict) -> None:
+        now = self.clock()
+        with self._connect() as db:
+            db.execute(
+                """UPDATE pose_ledger SET state='delivered', receipt_json=?, last_error=NULL,
+                   blocked_reason=NULL, updated_at=? WHERE idempotency_key=?""",
+                (_dump(receipt), _iso(now), key),
+            )
+
+    def _result(self, key: str, status: str, detail: str) -> ForwardResult:
+        row = self.pose(key)
+        receipt = json.loads(row["receipt_json"]) if row and row["receipt_json"] else None
+        return ForwardResult(
+            status=status,
+            idempotency_key=key,
+            receipt=receipt,
+            drafted_by=row["drafted_by"] if row else None,
+            drafted_by_provenance=row["drafted_by_provenance"] if row else "absent",
+            provenance_note=row["provenance_note"] if row else PROVENANCE_NOTE,
+            detail=detail,
+        )
+
+    def _append_resolution(
+        self, *, decision_id, memo_id, verb, resolves, status, error_type, payload, retry_at
+    ) -> None:
+        with self._connect() as db:
+            db.execute(
+                """UPDATE resolution_ledger SET next_attempt=NULL
+                   WHERE decision_id=? AND memo_id=? AND status='failed'""",
+                (decision_id, memo_id),
+            )
+            db.execute(
+                """INSERT INTO resolution_ledger (
+                    decision_id, memo_id, verb, resolves, status, error_type,
+                    payload_json, next_attempt, created_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                (
+                    decision_id, memo_id, verb, resolves, status, error_type,
+                    _dump(payload) if payload is not None else None,
+                    retry_at, _iso(self.clock()),
+                ),
+            )
+
+    def _latest_resolution(self, decision_id: str, memo_id: str) -> dict | None:
+        with self._connect() as db:
+            row = db.execute(
+                """SELECT * FROM resolution_ledger
+                   WHERE decision_id=? AND memo_id=? ORDER BY id DESC LIMIT 1""",
+                (decision_id, memo_id),
+            ).fetchone()
+            return dict(row) if row is not None else None
+
+    @contextmanager
+    def _connect(self) -> Iterator[sqlite3.Connection]:
+        db = sqlite3.connect(self.path)
+        db.row_factory = sqlite3.Row
+        try:
+            db.execute("PRAGMA journal_mode=WAL")
+            yield db
+            db.commit()
+        except Exception:
+            db.rollback()
+            raise
+        finally:
+            db.close()
+
+
+def _base_url(value: str) -> str:
+    if (
+        not isinstance(value, str)
+        or not (value.startswith("http://") or value.startswith("https://"))
+    ):
+        raise ValueError("base_url is required")
+    if "127.0.0.1" in value or "localhost" in value:
+        raise ValueError("base_url must be an injected test host")
+    return value.rstrip("/")
+
+
+def _backoff(attempts: int) -> float:
+    return float(min(3600, 2 ** min(attempts, 11)))
+
+
+def _dump(value: dict) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def _iso(timestamp: float) -> str:
+    return datetime.fromtimestamp(timestamp, timezone.utc).isoformat()

api/services/memo_notification.py

@@ -0,0 +1,67 @@
+"""Deduplicated informed-decision notification receipts.
+
+The signed-in overview is the review entry point. A receipt stores and
+announces only memo id, question, class, and submitter. Replays of the same
+delivery id return the retained receipt; a different body conflicts and leaves
+the original readable.
+"""
+from __future__ import annotations
+
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.models.agent_message import AgentMessage
+from api.models.memo_notification import MemoNotificationReceipt
+from api.services.informed_decision_contract import Notification
+from api.services.repository_aliases import canonicalize_repository_slug
+
+NOTIFICATION_SENDER = "informed-decision"
+NOTIFICATION_RECIPIENT = "bernd"
+
+
+class NotificationConflict(Exception):
+    def __init__(self, receipt: MemoNotificationReceipt) -> None:
+        self.receipt = receipt
+
+
+async def accept_notification(
+    session: AsyncSession, delivery_id: str, notice: Notification
+) -> tuple[MemoNotificationReceipt, bool]:
+    existing = await session.get(MemoNotificationReceipt, delivery_id)
+    if existing is not None:
+        if existing.body_digest != notice.body_digest:
+            raise NotificationConflict(existing)
+        return existing, True
+
+    message = AgentMessage(
+        from_agent=await canonicalize_repository_slug(session, NOTIFICATION_SENDER),
+        to_agent=await canonicalize_repository_slug(session, NOTIFICATION_RECIPIENT),
+        subject=notice.message_subject(),
+        body=notice.message_text(),
+        kind="message",
+    )
+    session.add(message)
+    await session.flush()
+    receipt = MemoNotificationReceipt(
+        delivery_id=delivery_id,
+        body_digest=notice.body_digest,
+        memo_id=notice.memo_id,
+        question=notice.question,
+        memo_class=notice.memo_class,
+        submitter=notice.submitter,
+        message_id=message.id,
+        authoritative_review="signed-in-overview",
+    )
+    session.add(receipt)
+    try:
+        await session.commit()
+    except IntegrityError:
+        await session.rollback()
+        existing = await session.get(MemoNotificationReceipt, delivery_id)
+        if existing is None:
+            raise
+        if existing.body_digest != notice.body_digest:
+            raise NotificationConflict(existing)
+        return existing, True
+    await session.refresh(receipt)
+    return receipt, False