From 49eeddbfed7a6dcfb15a26cf4de48357d45b10d5 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 3 Oct 2026 21:25:46 +0200 Subject: [PATCH] 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 | 3 +- api/models/__init__.py | 2 + api/models/memo_notification.py | 37 ++ api/routers/memo_notifications.py | 97 ++++ api/services/informed_decision_contract.py | 382 +++++++++++++++ api/services/memo_forwarding.py | 440 ++++++++++++++++++ api/services/memo_notification.py | 67 +++ ...a0b1c2d3e4f5_memo_notification_receipts.py | 53 +++ tests/test_memo_forwarding.py | 380 +++++++++++++++ tests/test_memo_notification.py | 152 ++++++ 10 files changed, 1612 insertions(+), 1 deletion(-) create mode 100644 api/models/memo_notification.py create mode 100644 api/routers/memo_notifications.py create mode 100644 api/services/informed_decision_contract.py create mode 100644 api/services/memo_forwarding.py create mode 100644 api/services/memo_notification.py create mode 100644 migrations/versions/a0b1c2d3e4f5_memo_notification_receipts.py create mode 100644 tests/test_memo_forwarding.py create mode 100644 tests/test_memo_notification.py diff --git a/api/main.py b/api/main.py index fecd89c..17e938c 100644 --- a/api/main.py +++ b/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) diff --git a/api/models/__init__.py b/api/models/__init__.py index 3e06ef0..0ce0f5d 100644 --- a/api/models/__init__.py +++ b/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", diff --git a/api/models/memo_notification.py b/api/models/memo_notification.py new file mode 100644 index 0000000..18d9c55 --- /dev/null +++ b/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 + ) diff --git a/api/routers/memo_notifications.py b/api/routers/memo_notifications.py new file mode 100644 index 0000000..3c6d8a0 --- /dev/null +++ b/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) diff --git a/api/services/informed_decision_contract.py b/api/services/informed_decision_contract.py new file mode 100644 index 0000000..f5c8b48 --- /dev/null +++ b/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) diff --git a/api/services/memo_forwarding.py b/api/services/memo_forwarding.py new file mode 100644 index 0000000..d8e9311 --- /dev/null +++ b/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() diff --git a/api/services/memo_notification.py b/api/services/memo_notification.py new file mode 100644 index 0000000..7f2d0ea --- /dev/null +++ b/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 diff --git a/migrations/versions/a0b1c2d3e4f5_memo_notification_receipts.py b/migrations/versions/a0b1c2d3e4f5_memo_notification_receipts.py new file mode 100644 index 0000000..ec770d7 --- /dev/null +++ b/migrations/versions/a0b1c2d3e4f5_memo_notification_receipts.py @@ -0,0 +1,53 @@ +"""memo notification receipts + +Revision ID: a0b1c2d3e4f5 +Revises: f9a0b1c2d3e4 +Create Date: 2026-10-03 + +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import UUID + +revision = "a0b1c2d3e4f5" +down_revision = "f9a0b1c2d3e4" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "memo_notification_receipts", + sa.Column("delivery_id", sa.String(length=128), primary_key=True), + sa.Column("body_digest", sa.String(length=64), nullable=False), + sa.Column("memo_id", sa.String(length=63), nullable=False), + sa.Column("question", sa.String(length=400), nullable=False), + sa.Column("memo_class", sa.String(length=63), nullable=False), + sa.Column("submitter", sa.String(length=63), nullable=False), + sa.Column( + "message_id", + UUID(as_uuid=True), + sa.ForeignKey("agent_messages.id", ondelete="RESTRICT"), + nullable=False, + ), + sa.Column("authoritative_review", sa.String(length=40), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + ) + op.create_index( + "ix_memo_notification_receipts_message_id", + "memo_notification_receipts", + ["message_id"], + ) + + +def downgrade() -> None: + op.drop_index( + "ix_memo_notification_receipts_message_id", + table_name="memo_notification_receipts", + ) + op.drop_table("memo_notification_receipts") diff --git a/tests/test_memo_forwarding.py b/tests/test_memo_forwarding.py new file mode 100644 index 0000000..e09a7b7 --- /dev/null +++ b/tests/test_memo_forwarding.py @@ -0,0 +1,380 @@ +"""Dev fixture for hub memo forwarding. No live intake, token, or submitter.""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from api.services.informed_decision_contract import ( + PROVENANCE_NOTE, + ContractError, + build_envelope, + correlate_resolution, + resolution_applies, + stable_idempotency_key, + validate_envelope, +) +from api.services.memo_forwarding import DevIntakeFixture +from api.services.write_idempotency import route_rule_for + +DECISION = "62aa95e4-3ccc-4867-8806-5e3db005b483" +MEMO = "infd-20261003-p01" +OTHER = "infd-20261003-p02" +APPROVAL = "ba5ce2d8-8b6d-40be-af89-2e8c147029a3" +BINDING = "sha256:" + "ab" * 32 +TOKEN = "dev-fixture-not-a-credential" + + +class Clock: + def __init__(self) -> None: + self.now = 1_000.0 + + def __call__(self) -> float: + return self.now + + +class Transport: + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + def __call__(self, method, url, body, headers): + self.calls.append((method, url, body, dict(headers))) + item = self.responses.pop(0) + if isinstance(item, Exception): + raise item + return item + + +def envelope(**overrides): + memo = {"id": MEMO, "question": "Accept the forwarded draft?"} + packet = [{"item_id": "brief", "media_type": "text/plain", "content": "Bounded brief."}] + intent = { + "action": "accept", + "principal": "pending-human-session", + "purpose": "Review the draft.", + "target": {"id": MEMO, "type": "decision-memo", "system": "informed-decision"}, + } + fields = dict( + decision_id=DECISION, + memo=memo, + packet=packet, + memo_class="work-record-accept", + review_group="net-kingdom-admins", + approval_intent=intent, + drafted_by="claude@the-custodian", + resolves="on-answer", + ) + fields.update(overrides) + return build_envelope(**fields) + + +def receipt(memo_id=MEMO): + return { + "memo_id": memo_id, + "approval_id": APPROVAL, + "approval_binding_digest": BINDING, + "echo": "should-not-be-kept", + } + + +def fixture(tmp_path, transport, clock=None, token=lambda: TOKEN): + return DevIntakeFixture( + tmp_path / "forward.sqlite", + transport=transport, + intake_base_url="http://intake.test", + hub_base_url="http://hub.test", + token_reader=token, + clock=clock or Clock(), + ) + + +def stored_bytes(path) -> bytes: + data = Path(path).read_bytes() + for suffix in ("-wal", "-shm"): + side = Path(str(path) + suffix) + if side.exists(): + data += side.read_bytes() + return data + + +def test_idempotency_key_is_stable_and_drafted_by_is_not_identity(): + first = envelope() + second = envelope() + assert first["idempotency_key"] == second["idempotency_key"] + assert first["idempotency_key"] == stable_idempotency_key(DECISION, MEMO) + assert "drafted_by" not in first["idempotency_key"] + assert "submitter" not in first + changed = dict(first) + changed["idempotency_key"] = "unstable-key-value" + with pytest.raises(ContractError) as error: + validate_envelope(changed) + assert error.value.code == "idempotency_key_not_stable" + + +def test_envelope_is_bounded_and_does_not_echo_secrets(): + huge = envelope() + huge["packet"][0]["content"] = "x" * (256 * 1024 + 1) + with pytest.raises(ContractError) as too_big: + validate_envelope(huge) + assert too_big.value.code == "packet_too_large" + leaked = envelope() + leaked["packet"][0]["content"] = "-----BEGIN PRIVATE KEY-----" + with pytest.raises(ContractError) as secret: + validate_envelope(leaked) + assert secret.value.code == "secret_marker" + assert "PRIVATE KEY" not in str(secret.value) + extra = envelope() + extra["highlights"] = ["not-a-notice"] + with pytest.raises(ContractError) as unknown: + validate_envelope(extra) + assert unknown.value.code == "unknown_field" + + +def test_pose_replay_conflict_and_asserted_provenance(tmp_path): + clock = Clock() + transport = Transport([(201, receipt())]) + fix = fixture(tmp_path, transport, clock) + posed = fix.submit(envelope()) + assert posed.status == "delivered" + assert posed.drafted_by == "claude@the-custodian" + assert posed.drafted_by_provenance == "asserted" + assert posed.provenance_note == PROVENANCE_NOTE + assert "asserted provenance" in posed.provenance_note + assert set(posed.receipt) == {"memo_id", "approval_id", "approval_binding_digest"} + sent = transport.calls[0] + assert sent[0:2] == ("POST", "http://intake.test/v1/memos") + assert sent[2]["drafted_by"] == "claude@the-custodian" + assert "drafted_by_provenance" not in sent[2] + assert "submitter" not in sent[2] + assert sent[3]["Authorization"] == f"Bearer {TOKEN}" + assert TOKEN.encode() not in stored_bytes(fix.path) + + restarted = DevIntakeFixture( + fix.path, + transport=Transport([(201, receipt())]), + intake_base_url="http://intake.test", + hub_base_url="http://hub.test", + token_reader=lambda: TOKEN, + clock=clock, + ) + replay = restarted.submit(envelope()) + assert replay.status == "replay" + assert replay.receipt == posed.receipt + assert restarted.transport.calls == [] + + conflict = envelope() + conflict["packet"][0]["content"] = "Different brief." + blocked = restarted.submit(conflict) + assert blocked.status == "blocked" + assert blocked.detail == "idempotency_conflict" + kept = restarted.pose(posed.idempotency_key) + assert kept["state"] == "delivered" + assert "Different brief." not in kept["envelope_json"] + assert restarted.transport.calls == [] + + +def test_delivery_failure_keeps_local_evidence_and_retries(tmp_path): + clock = Clock() + transport = Transport([ + ConnectionRefusedError("down"), + (503, {"error": "intake_unavailable", "body": "LEAKED-BODY"}), + (201, receipt()), + ]) + fix = fixture(tmp_path, transport, clock) + failed = fix.submit(envelope()) + assert failed.status == "pending" + assert failed.detail == "ConnectionRefusedError" + row = fix.pose(failed.idempotency_key) + assert row["state"] == "pending" + assert "Bounded brief." in row["envelope_json"] + assert "LEAKED-BODY" not in row["envelope_json"] + assert fix.run_due() is None + + clock.now += 2 + again = fix.run_due() + assert again.status == "pending" + assert again.detail == "http_503" + assert b"LEAKED-BODY" not in stored_bytes(fix.path) + assert fix.run_due() is None + + clock.now += 4 + delivered = fix.run_due() + assert delivered.status == "delivered" + assert delivered.receipt["memo_id"] == MEMO + assert fix.run_due() is None + assert len(transport.calls) == 3 + + +def test_intake_conflict_blocks_without_retry(tmp_path): + transport = Transport([(409, {"error": "idempotency_conflict", "packet": "LEAK"})]) + fix = fixture(tmp_path, transport) + blocked = fix.submit(envelope()) + assert blocked.status == "blocked" + assert blocked.detail == "idempotency_conflict" + assert b"LEAK" not in stored_bytes(fix.path) + assert fix.run_due() is None + assert len(transport.calls) == 1 + + +def test_absent_token_is_not_invented(tmp_path): + transport = Transport([(201, receipt())]) + fix = fixture(tmp_path, transport, token=None) + pending = fix.submit(envelope()) + assert pending.status == "pending" + assert pending.detail == "projected_token_absent" + assert transport.calls == [] + assert fix.pose(pending.idempotency_key)["state"] == "pending" + + def boom(): + raise RuntimeError("sk-or-this-must-not-be-stored-0123456789") + + fix.token_reader = boom + fix.clock = Clock() + fix.clock.now = 1_000 + 2 + retried = fix.run_due() + assert retried.status == "pending" + assert retried.detail == "RuntimeError" + assert transport.calls == [] + assert b"sk-or-this-must-not-be-stored" not in stored_bytes(fix.path) + + +def test_live_hosts_and_runtime_binding_are_refused(tmp_path): + with pytest.raises(ValueError): + DevIntakeFixture( + tmp_path / "no.sqlite", + transport=Transport([]), + intake_base_url="http://127.0.0.1:8000", + hub_base_url="http://hub.test", + ) + root = Path(__file__).resolve().parents[1] / "api" + main = (root / "main.py").read_text() + assert "memo_notifications" in main + assert "memo_forwarding" not in main + assert "DevIntakeFixture" not in main + owned = [ + root / "main.py", + root / "services" / "memo_forwarding.py", + root / "services" / "informed_decision_contract.py", + root / "services" / "memo_notification.py", + root / "routers" / "memo_notifications.py", + root / "models" / "memo_notification.py", + ] + for path in owned: + text = path.read_text() + assert "INFD_HUB_WRITEBACK" not in text + assert "IntakeDeliveryWorker" not in text + assert "/var/run/secrets" not in text + assert "token_file" not in text + assert route_rule_for("POST", "/informed-decision/notifications") is None + + +@pytest.mark.parametrize("verb,serves", [ + ("return", {"hub_decision_id": DECISION, "resolves": "on-answer"}), + ("discuss", {"hub_decision_id": DECISION, "resolves": "on-answer"}), + ("accept", {"hub_decision_id": DECISION, "resolves": "never"}), + ("decline", None), +]) +def test_non_terminal_answers_do_not_resolve(tmp_path, verb, serves): + assert resolution_applies(verb, serves) is False + transport = Transport([(200, {})]) + fix = fixture(tmp_path, transport) + status = fix.observe_resolution( + decision_id=DECISION, memo_id=MEMO, verb=verb, serves=serves, payload={"rationale": "no"}, + ) + assert status == "not_applicable" + assert transport.calls == [] + + +def test_resolution_failure_retries_and_other_memo_blocks(tmp_path): + clock = Clock() + payload = { + "decided_by": "Bernd Worsch via informed-decision", + "rationale": f"accept in informed-decision memo {MEMO} v1; ", + "write_log": False, + } + transport = Transport([ + ConnectionRefusedError("down"), + (200, {"status": "resolved"}), + (409, {"error": "already resolved"}), + (200, {"rationale": "resolved by hand"}), + ]) + fix = fixture(tmp_path, transport, clock) + failed = fix.observe_resolution( + decision_id=DECISION, memo_id=MEMO, verb="accept", + serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload, + ) + assert failed == "failed" + assert fix.resolutions()[0]["payload_json"] + assert fix.run_due_resolution() is None + clock.now += 2 + assert fix.run_due_resolution() == "resolved" + assert [row["status"] for row in fix.resolutions()] == ["failed", "resolved"] + assert fix.run_due_resolution() is None + + conflict = fix.observe_resolution( + decision_id=DECISION, memo_id=OTHER, verb="decline", + serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload, + ) + assert conflict == "conflict" + calls = len(transport.calls) + assert transport.calls[-2][0] == "POST" + assert transport.calls[-1][0] == "GET" + assert fix.observe_resolution( + decision_id=DECISION, memo_id=OTHER, verb="decline", + serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload, + ) == "conflict" + assert len(transport.calls) == calls + + +def test_unreadable_409_stays_retryable(tmp_path): + clock = Clock() + transport = Transport([ + (409, {}), + (503, {}), + (409, {}), + (200, {"rationale": f"decline in informed-decision memo {MEMO} v1; "}), + ]) + fix = fixture(tmp_path, transport, clock) + payload = {"rationale": "kept locally", "decided_by": "reviewer", "write_log": False} + assert fix.observe_resolution( + decision_id=DECISION, memo_id=MEMO, verb="decline", + serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload, + ) == "failed" + assert fix.resolutions()[0]["error_type"] == "resolution_unreadable" + clock.now += 2 + assert fix.run_due_resolution() == "already_resolved" + + +async def test_hub_409_still_returns_the_existing_resolution(client): + domain = await client.post("/domains/", json={"slug": "testdomain", "name": "Test Domain"}) + assert domain.status_code == 201, domain.text + topic = await client.post("/topics/", json={ + "slug": "testtopic", "title": "Test Topic", "domain": "testdomain", + }) + assert topic.status_code == 201, topic.text + created = await client.post("/decisions/", json={ + "title": "Should the draft stand?", "topic_id": topic.json()["id"], + }) + assert created.status_code == 201, created.text + decision_id = created.json()["id"] + rationale = f"accept in informed-decision memo {MEMO} v1; approval {APPROVAL}" + resolved = await client.post(f"/decisions/{decision_id}/resolve", json={ + "rationale": rationale, + "decided_by": "Bernd Worsch via informed-decision", + "write_log": False, + }) + assert resolved.status_code == 200, resolved.text + conflict = await client.post(f"/decisions/{decision_id}/resolve", json={ + "rationale": "resolved by hand", + "decided_by": "someone else", + "write_log": False, + }) + assert conflict.status_code == 409 + current = await client.get(f"/decisions/{decision_id}") + assert current.status_code == 200 + body = current.json() + assert body["rationale"] == rationale + assert correlate_resolution(MEMO, post_status=409, existing=body) == "already_resolved" + assert correlate_resolution(OTHER, post_status=409, existing=body) == "conflict" diff --git a/tests/test_memo_notification.py b/tests/test_memo_notification.py new file mode 100644 index 0000000..f718f8f --- /dev/null +++ b/tests/test_memo_notification.py @@ -0,0 +1,152 @@ +"""Deduplicated informed-decision notification contract.""" +from __future__ import annotations + +import json + +NOTICE = { + "memo_id": "infd-20261003-p01", + "question": "Accept the forwarded draft?", + "memo_class": "work-record-accept", + "submitter": "example-submitter", +} +KEY = "state-hub.delivery.0001" + + +async def _messages(client): + listed = await client.get("/messages/", params={"to_agent": "bernd"}) + assert listed.status_code == 200, listed.text + return listed.json() + + +async def test_notification_is_deduplicated_and_limited_to_four_fields(client): + first = await client.post( + "/informed-decision/notifications", + json=NOTICE, + headers={"Idempotency-Key": KEY}, + ) + assert first.status_code == 201, first.text + body = first.json() + assert body["status"] == "accepted" + assert body["authoritative_review"] == "signed-in-overview" + assert body["delivery_id"] == KEY + + raw = json.dumps({ + "submitter": NOTICE["submitter"], + "memo_class": NOTICE["memo_class"], + "question": NOTICE["question"], + "memo_id": NOTICE["memo_id"], + }) + second = await client.post( + "/informed-decision/notifications", + content=raw, + headers={"Idempotency-Key": KEY, "Content-Type": "application/json"}, + ) + assert second.status_code == 200, second.text + assert second.headers["x-statehub-idempotency-replay"] == "true" + assert second.json()["message_id"] == body["message_id"] + + messages = await _messages(client) + assert len(messages) == 1 + text = messages[0]["body"] + assert text.startswith("The signed-in overview is the authoritative review entry point.\n") + assert set(json.loads(text.split("\n", 1)[1])) == set(NOTICE) + assert "://" not in text + assert "packet" not in text + assert "highlight" not in text + assert messages[0]["from_agent"] == "informed-decision" + assert messages[0]["to_agent"] == "bernd" + + +async def test_conflict_leaves_the_original_receipt_readable(client): + created = await client.post( + "/informed-decision/notifications", + json=NOTICE, + headers={"Idempotency-Key": KEY}, + ) + assert created.status_code == 201, created.text + changed = dict(NOTICE, question="Decline the forwarded draft?") + conflict = await client.post( + "/informed-decision/notifications", + json=changed, + headers={"Idempotency-Key": KEY}, + ) + assert conflict.status_code == 409 + assert conflict.json() == {"error": "notification_conflict", "delivery_id": KEY} + assert "Decline" not in conflict.text + current = await client.get(f"/informed-decision/notifications/{KEY}") + assert current.status_code == 200 + assert current.json()["question"] == NOTICE["question"] + assert len(await _messages(client)) == 1 + + +async def test_notification_refuses_extra_secret_and_link_without_echo(client): + extra = dict(NOTICE, packet="SECRET-PACKET-VALUE") + refused = await client.post( + "/informed-decision/notifications", + json=extra, + headers={"Idempotency-Key": KEY}, + ) + assert refused.status_code == 400 + assert refused.json() == {"error": "unknown_field"} + assert "SECRET-PACKET-VALUE" not in refused.text + + secret = dict(NOTICE, question="Is Bearer " + ("A" * 16) + " acceptable?") + marked = await client.post( + "/informed-decision/notifications", + json=secret, + headers={"Idempotency-Key": "state-hub.delivery.0002"}, + ) + assert marked.status_code == 400 + assert marked.json() == {"error": "secret_marker"} + assert "AAAA" not in marked.text + + linked = dict(NOTICE, question="Open https://overview.example/review?") + link = await client.post( + "/informed-decision/notifications", + json=linked, + headers={"Idempotency-Key": "state-hub.delivery.0003"}, + ) + assert link.status_code == 400 + assert link.json() == {"error": "notification_has_link"} + assert "overview.example" not in link.text + assert await _messages(client) == [] + + +async def test_delivery_id_and_size_are_required(client): + missing = await client.post("/informed-decision/notifications", json=NOTICE) + assert missing.status_code == 400 + assert missing.json()["error"] == "delivery_id_required" + short = await client.post( + "/informed-decision/notifications", + json=NOTICE, + headers={"Idempotency-Key": "short"}, + ) + assert short.status_code == 400 + assert short.json()["error"] == "delivery_id_invalid" + huge = await client.post( + "/informed-decision/notifications", + content=b"{" + b"x" * 9000, + headers={"Idempotency-Key": KEY, "Content-Type": "application/json"}, + ) + assert huge.status_code == 413 + assert huge.json()["error"] == "notification_too_large" + assert b"xxx" not in huge.content + missing_receipt = await client.get("/informed-decision/notifications/" + KEY) + assert missing_receipt.status_code == 404 + + +async def test_distinct_delivery_ids_are_distinct_notices(client): + first = await client.post( + "/informed-decision/notifications", + json=NOTICE, + headers={"Idempotency-Key": "state-hub.delivery.0001"}, + ) + second_body = dict(NOTICE, memo_id="infd-20261003-p02", question="Decline the other draft?") + second = await client.post( + "/informed-decision/notifications", + json=second_body, + headers={"Idempotency-Key": "state-hub.delivery.0002"}, + ) + assert first.status_code == 201 and second.status_code == 201 + assert first.json()["message_id"] != second.json()["message_id"] + assert len(await _messages(client)) == 2