Register informed-decision.memo.posed for the platform tenant.

Accept the retained pose envelope and refuse packet, brief, and secret
content. Heartbeat and reconciliation follow the existing load-bearing
class pattern. This does not connect the sink or claim runtime readiness.

Assistant: grok
Assistant-Session: 01a10329-31cb-78d3-b7a3-787a69c89d54

tests/fixtures/informed-decision-memo-posed/duplicate.json

@@ -0,0 +1,17 @@
+{
+  "id": "posed-fixture-1",
+  "type": "informed-decision.memo.posed",
+  "source": "informed-decision",
+  "subject": "memo-fixture-1",
+  "tenant": "tenant:platform",
+  "correlation_id": "approval-fixture-1",
+  "occurred_at": "2026-10-03T12:00:00+00:00",
+  "data": {
+    "memo_id": "memo-fixture-1",
+    "approval_id": "approval-fixture-1",
+    "approval_binding_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "submitter": "state-hub",
+    "memo_class": "fixture-class",
+    "submission_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+  }
+}

tests/fixtures/informed-decision-memo-posed/positive.json

@@ -0,0 +1,17 @@
+{
+  "id": "posed-fixture-1",
+  "type": "informed-decision.memo.posed",
+  "source": "informed-decision",
+  "subject": "memo-fixture-1",
+  "tenant": "tenant:platform",
+  "correlation_id": "approval-fixture-1",
+  "occurred_at": "2026-10-03T12:00:00+00:00",
+  "data": {
+    "memo_id": "memo-fixture-1",
+    "approval_id": "approval-fixture-1",
+    "approval_binding_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "submitter": "state-hub",
+    "memo_class": "fixture-class",
+    "submission_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+  }
+}

tests/fixtures/informed-decision-memo-posed/refusal-brief.json

@@ -0,0 +1,18 @@
+{
+  "id": "posed-refusal-brief",
+  "type": "informed-decision.memo.posed",
+  "source": "informed-decision",
+  "subject": "memo-fixture-1",
+  "tenant": "tenant:platform",
+  "correlation_id": "approval-fixture-1",
+  "occurred_at": "2026-10-03T12:00:00+00:00",
+  "data": {
+    "memo_id": "memo-fixture-1",
+    "approval_id": "approval-fixture-1",
+    "approval_binding_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "submitter": "state-hub",
+    "memo_class": "fixture-class",
+    "submission_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+    "brief": "fixture-brief"
+  }
+}

tests/fixtures/informed-decision-memo-posed/refusal-packet.json

@@ -0,0 +1,18 @@
+{
+  "id": "posed-refusal-packet",
+  "type": "informed-decision.memo.posed",
+  "source": "informed-decision",
+  "subject": "memo-fixture-1",
+  "tenant": "tenant:platform",
+  "correlation_id": "approval-fixture-1",
+  "occurred_at": "2026-10-03T12:00:00+00:00",
+  "data": {
+    "memo_id": "memo-fixture-1",
+    "approval_id": "approval-fixture-1",
+    "approval_binding_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "submitter": "state-hub",
+    "memo_class": "fixture-class",
+    "submission_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+    "packet": "fixture-packet"
+  }
+}

tests/fixtures/informed-decision-memo-posed/refusal-secret.json

@@ -0,0 +1,18 @@
+{
+  "id": "posed-refusal-secret",
+  "type": "informed-decision.memo.posed",
+  "source": "informed-decision",
+  "subject": "memo-fixture-1",
+  "tenant": "tenant:platform",
+  "correlation_id": "approval-fixture-1",
+  "occurred_at": "2026-10-03T12:00:00+00:00",
+  "data": {
+    "memo_id": "memo-fixture-1",
+    "approval_id": "approval-fixture-1",
+    "approval_binding_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "submitter": "state-hub",
+    "memo_class": "fixture-class",
+    "submission_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+    "password": "fixture-not-a-credential"
+  }
+}

tests/fixtures/informed-decision-memo-posed/refusal-tenant.json

@@ -0,0 +1,17 @@
+{
+  "id": "posed-refusal-tenant",
+  "type": "informed-decision.memo.posed",
+  "source": "informed-decision",
+  "subject": "memo-fixture-1",
+  "tenant": "platform",
+  "correlation_id": "approval-fixture-1",
+  "occurred_at": "2026-10-03T12:00:00+00:00",
+  "data": {
+    "memo_id": "memo-fixture-1",
+    "approval_id": "approval-fixture-1",
+    "approval_binding_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "submitter": "state-hub",
+    "memo_class": "fixture-class",
+    "submission_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+  }
+}

tests/test_informed_decision_memo_posed.py

@@ -0,0 +1,261 @@
+"""Local registration of informed-decision.memo.posed.
+
+Same sender pattern as the other registered sources: scope binds the
+credential, Idempotency-Key receipts are 202 then 200, and a low-volume
+load-bearing class is a heartbeat class counted by /v1/reconciliation.
+"""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from audit_core.ingestion import (
+    CONTENT_REFUSED,
+    POSED_EVENT_TYPE,
+    IngestionApplication,
+)
+from audit_core.senders import SenderRegistry
+from audit_core.sqlite_backend import SQLiteAuditBackend
+from audit_core.stream_findings import HEARTBEAT_ACTION, NEVER_HEARTBEAT
+
+from test_ingestion import invoke
+from test_senders import SCOPE_FILE
+
+FIXTURES = Path(__file__).resolve().parent / "fixtures" / "informed-decision-memo-posed"
+WINDOW = (
+    "/v1/reconciliation?source=informed-decision&tenant=tenant:platform"
+    "&since=2026-01-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00"
+)
+
+
+def _load(name: str) -> dict:
+    return json.loads((FIXTURES / name).read_text())
+
+
+def _registry() -> SenderRegistry:
+    return SenderRegistry.from_env({
+        "AUDIT_CORE_SENDERS": json.dumps([
+            {
+                "name": "informed-decision",
+                "tokens": ["fixture-only"],
+                "sources": ["*"],
+                "tenants": ["*"],
+                "may_read": True,
+                "secret_policy": "redact",
+                "evidence_kind": "load-bearing",
+            },
+            {
+                "name": "user-engine",
+                "tokens": ["user-fixture"],
+                "sources": ["*"],
+                "tenants": ["*"],
+                "may_read": False,
+                "secret_policy": "redact",
+            },
+            {
+                "name": "operator",
+                "tokens": ["operator-fixture"],
+                "sources": ["informed-decision"],
+                "tenants": ["*"],
+                "may_write": False,
+                "may_read": True,
+            },
+        ]),
+        "AUDIT_CORE_SENDERS_SCOPE_PATH": str(SCOPE_FILE),
+    })
+
+
+@pytest.fixture
+def app(tmp_path):
+    backend = SQLiteAuditBackend(str(tmp_path / "posed.db"))
+    return IngestionApplication(backend, _registry()), backend
+
+
+def _post(app, payload, *, token="fixture-only", key=None):
+    return invoke(app, payload, token=token, key=payload["id"] if key is None else key)
+
+
+def test_scope_registration_keeps_sender_custody_and_adds_the_class():
+    identity = _registry().authenticate("Bearer fixture-only")
+    assert identity.name == "informed-decision"
+    assert identity.is_load_bearing
+    assert identity.completeness_trade is None
+    assert identity.secret_policy == "redact"
+    assert identity.may_write and not identity.may_read
+    assert identity.permits_source("informed-decision")
+    assert not identity.permits_source("user-engine")
+    assert identity.permits_tenant("tenant:platform")
+    assert not identity.permits_tenant("platform")
+    assert dict(identity.heartbeat_classes) == {POSED_EVENT_TYPE: 86400}
+
+
+def test_positive_fixture_is_accepted_and_duplicate_fixture_reconciles(app):
+    application, backend = app
+    positive = _load("positive.json")
+    duplicate = _load("duplicate.json")
+    assert duplicate == positive
+
+    status, body = _post(application, positive)
+    assert status.startswith("202")
+    assert body == {"status": "accepted", "reference": "audit:posed-fixture-1"}
+
+    status, body = _post(application, duplicate)
+    assert status.startswith("200")
+    assert body == {"status": "duplicate", "reference": "audit:posed-fixture-1"}
+    assert backend.db.execute("SELECT COUNT(*) FROM events").fetchone()[0] == 1
+
+    status, stored = invoke(
+        application, None, path="/v1/events/posed-fixture-1", method="GET",
+        body=b"", token="operator-fixture",
+    )
+    assert status.startswith("200")
+    assert stored["action"] == POSED_EVENT_TYPE
+    assert stored["source"] == "informed-decision"
+    assert stored["tenant"] == "tenant:platform"
+    assert stored["details"]["data"] == positive["data"]
+    assert "redaction" not in stored["details"]
+
+    status, denied = invoke(
+        application, None, path="/v1/events/posed-fixture-1", method="GET",
+        body=b"", token="fixture-only",
+    )
+    assert status.startswith("403")
+    assert denied["error"] == "read_forbidden"
+
+
+def test_idempotency_key_must_equal_the_event_id(app):
+    application, _ = app
+    status, body = _post(application, _load("positive.json"), key="other-key")
+    assert status.startswith("400")
+    assert body["error"] == "idempotency_key_mismatch"
+
+
+def _get(app, path, *, token="fixture-only"):
+    import io
+
+    base, _, query = path.partition("?")
+    environ = {
+        "PATH_INFO": base,
+        "QUERY_STRING": query,
+        "REQUEST_METHOD": "GET",
+        "CONTENT_LENGTH": "0",
+        "wsgi.input": io.BytesIO(b""),
+        "HTTP_AUTHORIZATION": f"Bearer {token}",
+    }
+    result = {}
+    out = b"".join(app(environ, lambda status, headers: result.update(status=status)))
+    return result["status"], (json.loads(out) if out else {})
+
+
+def test_reconciliation_counts_the_class_and_returns_no_payload(app):
+    application, _ = app
+    assert _post(application, _load("positive.json"))[0].startswith("202")
+    status, body = _get(application, WINDOW)
+    assert status.startswith("200")
+    assert body["source"] == "informed-decision"
+    assert body["tenant"] == "tenant:platform"
+    assert body["counts"] == [{"class": POSED_EVENT_TYPE, "count": 1}]
+    assert "memo-fixture-1" not in json.dumps(body)
+    assert "completeness" in body["means"]
+
+
+def test_a_missing_heartbeat_for_the_class_is_a_finding_until_one_arrives(app):
+    application, _ = app
+    status, body = invoke(
+        application, None, path="/v1/stream-findings", method="GET",
+        body=b"", token="operator-fixture",
+    )
+    assert status.startswith("200")
+    posed = [row for row in body["stream_findings"] if row["class"] == POSED_EVENT_TYPE]
+    assert [row["kind"] for row in posed] == [NEVER_HEARTBEAT]
+    assert posed[0]["source"] == "informed-decision"
+    assert posed[0]["expected_within_seconds"] == 86400
+    assert "not proof of suppression" in posed[0]["means"].lower()
+
+    heartbeat = {
+        "id": "hb-posed-1",
+        "type": HEARTBEAT_ACTION,
+        "source": "informed-decision",
+        "subject": "informed-decision",
+        "tenant": "tenant:platform",
+        "correlation_id": "hb-posed-1",
+        "occurred_at": "2026-10-03T12:00:00+00:00",
+        "data": {"class": POSED_EVENT_TYPE, "assertion": "nothing-to-report"},
+    }
+    assert _post(application, heartbeat)[0].startswith("202")
+    _, cleared = invoke(
+        application, None, path="/v1/stream-findings", method="GET",
+        body=b"", token="operator-fixture",
+    )
+    assert cleared["stream_findings"] == []
+
+
+def test_another_sender_cannot_write_the_class(app):
+    application, backend = app
+    status, body = _post(application, _load("positive.json"), token="user-fixture")
+    assert status.startswith("400")
+    assert body["error"] == "source_not_allowed"
+    assert backend.db.execute("SELECT COUNT(*) FROM events").fetchone()[0] == 0
+
+
+@pytest.mark.parametrize("name,error,marker", [
+    ("refusal-packet.json", CONTENT_REFUSED, "fixture-packet"),
+    ("refusal-brief.json", CONTENT_REFUSED, "fixture-brief"),
+    ("refusal-secret.json", "secret_shaped_field", "fixture-not-a-credential"),
+])
+def test_refusal_fixtures_are_not_retained(app, name, error, marker):
+    application, backend = app
+    payload = _load(name)
+    status, body = _post(application, payload)
+    assert status.startswith("400")
+    assert body["error"] == error
+    assert backend.db.execute("SELECT COUNT(*) FROM events").fetchone()[0] == 0
+
+    _, letters = invoke(
+        application, None, path="/v1/dead-letters", method="GET",
+        body=b"", token="operator-fixture",
+    )
+    entry = letters["dead_letters"][0]
+    assert entry["reason"] == error
+    assert entry["payload_withheld"] is True
+    assert entry["payload"] is None
+    assert marker not in json.dumps(letters)
+
+
+def test_bare_platform_tenant_is_not_the_registered_tenant(app):
+    application, _ = app
+    status, body = _post(application, _load("refusal-tenant.json"))
+    assert status.startswith("400")
+    assert body["error"] == "tenant_not_allowed"
+
+
+def test_other_classes_still_follow_the_sender_redaction_policy(app):
+    """The posed refusal is not a new rule for the sender's other classes."""
+    application, _ = app
+    payload = _load("positive.json")
+    payload = {
+        **payload,
+        "id": "presentation-1",
+        "type": "informed-decision.presentation",
+        "data": {"packet": "still-stored", "auth_token": "fixture-token"},
+    }
+    assert _post(application, payload)[0].startswith("202")
+    _, stored = invoke(
+        application, None, path="/v1/events/presentation-1", method="GET",
+        body=b"", token="operator-fixture",
+    )
+    assert stored["details"]["data"]["packet"] == "still-stored"
+    assert stored["details"]["data"]["auth_token"] == "[redacted]"
+
+
+def test_both_backends_withhold_posed_content_refusals():
+    from audit_core.sqlite_backend import WITHHOLD_PAYLOAD_REASONS as sqlite_reasons
+
+    assert "content_refused" in sqlite_reasons
+    assert "secret_shaped_field" in sqlite_reasons
+    try:
+        from audit_core.postgres_backend import WITHHOLD_PAYLOAD_REASONS as postgres_reasons
+    except ImportError:
+        pytest.skip("postgres backend extra is not installed")
+    assert postgres_reasons == sqlite_reasons

tests/test_senders.py

@@ -353,7 +353,12 @@ def test_informed_decision_is_declared_load_bearing():
     assert entry["may_read"] is False
     # §9.6 permits no completeness trade for a load-bearing source.
     assert "completeness_trade" not in entry
+    assert "emission_cadence" not in entry
     assert "tokens" not in entry and "token" not in entry
+    # Same interval this sender already publishes for its low-volume classes,
+    # and the same interval the approval-engine registration tests use. Not a
+    # new custody class and not a rate declaration.
+    assert entry["heartbeat_classes"] == {"informed-decision.memo.posed": 86400}
 
 
 def test_informed_decision_scope_entry_admits_nothing_without_a_token():