Evaluate declared emission cadence and close AUDIT-WP-0009 (T05)

The contract T05 waited on is published: info-tech-canon
emission-cadence wire schema 0.1, contract digest b08b4d95fc4b0bd3.
A source-owned declaration now travels as emission_cadence on the
sender registration; expected-rate entries raise below_declared_cadence
on /v1/stream-findings from the same counts /v1/reconciliation returns.
heartbeat-or-reconciliation entries are validated and left to T04/T06.

Records the observer evaluation of net-kingdom's local-identity
declaration: structurally clean, not operationally evaluated, one
heartbeat event_class mapping incompatibility.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 151986@bnt-lap001
Assistant-Session: ccd02b6b-80ae-48e5-8cad-9c8f74d21a67

tests/test_emission_cadence.py

@@ -0,0 +1,219 @@
+"""AUDIT-WP-0009-T05 — declared emission cadence, evaluated by the observer."""
+
+import copy
+import json
+
+import pytest
+
+from audit_core.emission_cadence import (
+    BELOW_CADENCE,
+    CONTRACT_DIGEST,
+    duration_seconds,
+    evaluate,
+    parse_declaration,
+)
+from audit_core.ingestion import IngestionApplication
+from audit_core.senders import SenderIdentity, SenderRegistry, _identity_from
+from audit_core.sqlite_backend import SQLiteAuditBackend
+from tests.test_stream_findings import emit, invoke
+
+RATE = {
+    "schema_version": "0.1",
+    "declaration_id": "approval-engine.audit.v1",
+    "source": "approval-engine",
+    "stream_id": "approval-engine.audit",
+    "sources": [
+        {
+            "source_id": "approval-engine.issued",
+            "event_class": "approval.issued",
+            "form": "expected-rate",
+            "window": "PT1H",
+            "expected_min": 2,
+            "drop_below": "finding",
+        },
+        {
+            "source_id": "approval-engine.revocation",
+            "event_class": "approval.revocation",
+            "form": "heartbeat-or-reconciliation",
+            "heartbeat": {
+                "event_class": "approval-engine.heartbeat",
+                "interval": "24h",
+                "assertion": "nothing-to-report",
+                "missing": "finding",
+            },
+        },
+    ],
+}
+
+# net-kingdom commit 116643f, local-identity/emission-cadence.yaml, as JSON.
+# The one real source-owned declaration published so far.
+NET_KINGDOM_LOCAL_IDENTITY = {
+    "schema_version": "0.1",
+    "declaration_id": "net-kingdom.local-identity.audit.v1",
+    "source": "net-kingdom",
+    "stream_id": "net-kingdom.local-identity.audit",
+    "extensions": {"netkingdom": {"contract_digest": "972c0b6701d1693f",
+                                  "contract_document_version": "0.2.0"}},
+    "sources": [
+        {
+            "source_id": f"net-kingdom.local-identity.audit.{name}",
+            "source_system": "local-identity",
+            "event_class": cls,
+            "form": "heartbeat-or-reconciliation",
+            "reconciliation": {
+                "compare_local": f"audit_log_counts.{cls}",
+                "compare_observed": f"evidence_counts.{cls}",
+                "divergence": "finding",
+            },
+            "extensions": {"netkingdom": {"evidence_class": "load-bearing",
+                                          "rate_monitoring": "forbidden",
+                                          "completeness_claimed": False,
+                                          "heartbeat_emitted": False}},
+        }
+        for name, cls in (("token-issued", "serve/token.token_issued"),
+                          ("token-revoked", "revoke-token"))
+    ],
+}
+
+
+def _sender(declaration=RATE, sources=("approval-engine",)):
+    return SenderIdentity(
+        name="approval-engine", tokens=("approval",), sources=frozenset(sources),
+        tenants=frozenset({"tenant:platform"}),
+        emission_cadence=parse_declaration(declaration),
+    )
+
+
+@pytest.mark.parametrize("text,seconds", [
+    ("PT1H", 3600), ("P1D", 86400), ("P1DT30M", 88200), ("90s", 90), ("5m", 300), ("1d", 86400),
+])
+def test_durations_follow_the_schema_pattern(text, seconds):
+    assert duration_seconds(text) == seconds
+
+
+@pytest.mark.parametrize("text", ["P", "PT", "PT0S", "0s", "1w", "3600", "P1H"])
+def test_durations_the_schema_refuses_are_refused(text):
+    with pytest.raises(ValueError):
+        duration_seconds(text)
+
+
+def test_both_forms_parse_and_only_rates_are_indexed_for_evaluation():
+    declaration = parse_declaration(RATE)
+    assert [(r.event_class, r.window_seconds, r.expected_min) for r in declaration.rates] == [
+        ("approval.issued", 3600, 2)
+    ]
+    assert declaration.summary()["heartbeat_or_reconciliation_classes"] == ["approval.revocation"]
+    assert declaration.summary()["contract_digest"] == CONTRACT_DIGEST
+
+
+def test_the_published_net_kingdom_declaration_is_accepted():
+    """Observer evaluation of the one real declaration: it fits the contract."""
+    declaration = parse_declaration(NET_KINGDOM_LOCAL_IDENTITY)
+    assert declaration.rates == ()
+    assert len(declaration.other_entries) == 2
+
+
+def _broken(mutate):
+    doc = copy.deepcopy(RATE)
+    mutate(doc)
+    return doc
+
+
+@pytest.mark.parametrize("mutate", [
+    lambda d: d.update(schema_version="0.2"),
+    lambda d: d.pop("declaration_id"),
+    lambda d: d.update(sources=[]),
+    lambda d: d.update(unknown=1),
+    lambda d: d["sources"][0].pop("expected_min"),
+    lambda d: d["sources"][0].pop("drop_below"),
+    lambda d: d["sources"][0].update(window_seconds=3600),           # both window forms
+    lambda d: d["sources"][0].pop("window"),                          # neither
+    lambda d: d["sources"][0].update(expected_min=-1),
+    lambda d: d["sources"][0].update(heartbeat=RATE["sources"][1]["heartbeat"]),
+    lambda d: d["sources"][0].update(form="rate"),
+    lambda d: d["sources"][1].pop("heartbeat"),                       # neither half
+    lambda d: d["sources"][1].update(window="PT1H"),
+    lambda d: d["sources"][1]["heartbeat"].pop("missing"),
+    lambda d: d["sources"][1]["heartbeat"].update(interval_seconds=60),
+    lambda d: d["sources"].append(copy.deepcopy(d["sources"][0])),   # duplicate class
+])
+def test_what_the_schema_refuses_is_refused(mutate):
+    with pytest.raises(ValueError):
+        parse_declaration(_broken(mutate))
+
+
+def _counter(counts):
+    return lambda source, since, until: [{"class": c, "count": n} for c, n in counts.items()]
+
+
+def test_a_stream_below_its_declared_rate_is_a_finding():
+    findings = evaluate([_sender()], _counter({"approval.issued": 1}))
+    assert len(findings) == 1
+    finding = findings[0].as_dict()
+    assert finding["kind"] == BELOW_CADENCE
+    assert (finding["observed"], finding["expected_min"]) == (1, 2)
+    assert "not proof of suppression" in finding["means"]
+
+
+def test_a_stream_at_its_declared_rate_is_not_a_finding():
+    assert evaluate([_sender()], _counter({"approval.issued": 2})) == []
+
+
+def test_an_undeclared_class_is_never_a_finding():
+    """Heartbeat-or-reconciliation classes are T04/T06's, not rate-monitored."""
+    findings = evaluate([_sender()], _counter({"approval.issued": 5}))
+    assert all(f.event_class != "approval.revocation" for f in findings)
+
+
+def test_a_wildcard_source_is_not_held_to_a_rate():
+    assert evaluate([_sender(sources=("*",))], _counter({})) == []
+
+
+def test_the_registration_carries_the_declaration():
+    identity = _identity_from({
+        "name": "approval-engine", "token": "t", "sources": ["approval-engine"],
+        "emission_cadence": RATE,
+    })
+    assert identity.evidence_declaration()["emission_cadence"]["declaration_id"] == (
+        "approval-engine.audit.v1"
+    )
+
+
+def test_an_invalid_declaration_refuses_the_registration():
+    with pytest.raises(ValueError):
+        _identity_from({
+            "name": "x", "token": "t", "sources": ["x"],
+            "emission_cadence": _broken(lambda d: d.pop("sources")),
+        })
+
+
+def test_the_overlay_may_not_touch_the_declaration(tmp_path):
+    base = {"name": "approval-engine", "token": "t", "sources": ["approval-engine"],
+            "emission_cadence": RATE}
+    scope = tmp_path / "scope.json"
+    scope.write_text(json.dumps([{"name": "approval-engine", "emission_cadence": RATE}]))
+    with pytest.raises(ValueError, match="emission_cadence"):
+        SenderRegistry.from_env({
+            "AUDIT_CORE_SENDERS": json.dumps([base]),
+            "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope),
+        })
+
+
+def test_the_findings_surface_reports_a_cadence_miss_and_clears_on_delivery(tmp_path):
+    operator = SenderIdentity(
+        name="operator", tokens=("operator",), sources=frozenset({"approval-engine"}),
+        tenants=frozenset({"*"}), may_write=False, may_read=True,
+    )
+    app = IngestionApplication(
+        SQLiteAuditBackend(str(tmp_path / "s.db")), SenderRegistry([_sender(), operator])
+    )
+
+    def misses():
+        status, body = invoke(app, "/v1/stream-findings", token="operator")
+        assert status.startswith("200")
+        return [f for f in body["stream_findings"] if f["kind"] == BELOW_CADENCE]
+
+    assert [f["observed"] for f in misses()] == [0]
+    for n in range(2):
+        assert emit(app, f"evt-{n}", "approval.issued")[0].startswith("202")
+    assert misses() == []