Implement AUDIT-WP-0006 honest operational custody.

Postgres now reports custody_class=operational with a cited 30-day
recoverable window. Join ITC-CAP operations.audit at D4, publish the
interface card, and overlay user-engine tenants [*] from Git so an
ExternalSecret refresh cannot shrink it.

tests/test_backend_conformance.py

@@ -86,7 +86,11 @@ def digest(event: AuditEvent) -> str:
 def test_declares_a_retention_policy(backend):
     policy = backend.retention_policy
     assert policy.durable is True
-    assert policy.custody_class in ("development", "archive", "hot_search")
+    assert policy.custody_class in ("development", "operational", "archive", "hot_search")
+    if policy.custody_class == "operational":
+        assert policy.recoverable_days == 30
+        assert policy.recoverable_basis == "measured"
+        assert policy.recoverable_source
     # A backend claiming tamper evidence must also claim immutability;
     # the reverse is allowed.
     if policy.tamper_evidence:

tests/test_capability_case.py

@@ -0,0 +1,65 @@
+import json
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+RECORD = ROOT / "data" / "capability" / "audit-core-operational.json"
+CARD = ROOT / "docs" / "interface-card.yaml"
+SCHEMA = Path.home() / "info-tech-canon" / "infospace" / "schemas" / "interface-card.schema.yaml"
+
+
+def test_capability_record_exists_and_joins_operations_audit():
+    record = json.loads(RECORD.read_text())
+    assert any(item["capability"] == "operations.audit" for item in record["requires"])
+    assert any(item["capability"] == "data.archive" for item in record["requires"])
+    provisions = {item["capability"]: item for item in record["provisions"]}
+    assert "data.archive" not in provisions
+    audit = provisions["operations.audit"]
+    assert audit["maturity"] == "D4"
+    assert audit["provider"] == "audit-core.railiance01"
+    used = {item["capability"]: item["relation"] for item in audit["uses_provisions"]}
+    assert used["data.backup"] == "may_use"
+    assert used["security.secrets"] == "may_use"
+    unknown = [row for row in audit["consumes"] if row["basis"] == "unknown"]
+    assert unknown
+    assert all(row["quantity"]["value"] is None and row.get("gap") for row in unknown)
+
+
+def test_capability_review_against_live_catalog():
+    import sys
+
+    canon_src = Path.home() / "info-tech-canon" / "src"
+    if canon_src.is_dir() and str(canon_src) not in sys.path:
+        sys.path.insert(0, str(canon_src))
+    try:
+        from info_tech_canon.capability import review_path
+    except ImportError:
+        pytest.skip("info-tech-canon is not importable")
+    result = review_path(RECORD)
+    assert result["ok"] is True
+    by_cap = {item["capability"]: item for item in result["requirements"]}
+    assert by_cap["operations.audit"]["status"] == "met"
+    assert by_cap["data.archive"]["status"] == "unprovided"
+
+
+def test_interface_card_has_required_fields():
+    text = CARD.read_text()
+    assert "id: audit-core/interface-card" in text
+    assert "title:" in text
+    assert "consumer:" in text
+    assert "canon_surfaces:" in text
+    assert "Evidence" in text
+    assert "AuditRecord" in text
+    assert "data.archive-unprovided" in text
+    assert "no-rapp-yaml" in text
+
+
+def test_interface_card_validates_against_canon_schema():
+    if not SCHEMA.is_file():
+        pytest.skip("info-tech-canon interface-card schema not on disk")
+    yaml = pytest.importorskip("yaml")
+    jsonschema = pytest.importorskip("jsonschema")
+    schema = yaml.safe_load(SCHEMA.read_text())
+    card = yaml.safe_load(CARD.read_text())
+    jsonschema.Draft202012Validator(schema).validate(card)

tests/test_ingestion.py

@@ -4,7 +4,11 @@ import json
 import pytest
 
 from audit_core.ingestion import IngestionApplication
-from audit_core.interface import BackendUnavailableError, RetentionPolicy
+from audit_core.interface import (
+    BackendUnavailableError,
+    RetentionPolicy,
+    custody_class_satisfies,
+)
 from audit_core.mock_file_backend import MockFileAuditBackend
 from audit_core.sqlite_backend import SQLiteAuditBackend
 
@@ -408,10 +412,71 @@ def test_required_custody_class_refuses_a_development_backend(tmp_path):
     """Losing AUDIT_CORE_DATABASE_URL must fail to start, not silently
     downgrade custody to the development store."""
     backend = SQLiteAuditBackend(str(tmp_path / "dev.db"))
+    with pytest.raises(ValueError, match="does not meet the required"):
+        IngestionApplication(backend, "opaque", require_custody_class="operational")
     with pytest.raises(ValueError, match="does not meet the required"):
         IngestionApplication(backend, "opaque", require_custody_class="archive")
 
 
+def test_operational_and_archive_alias_for_one_deploy():
+    """A mixed rollout must start: new backend + old require, and the reverse."""
+
+    class _Operational(_BrokenBackend):
+        @property
+        def retention_policy(self):
+            return RetentionPolicy(
+                custody_class="operational",
+                retention_days=None,
+                immutable=True,
+                tamper_evidence=False,
+                durable=True,
+                recoverable_days=30,
+                recoverable_source="cited",
+                recoverable_basis="measured",
+            )
+
+    IngestionApplication(_Operational(), "opaque", require_custody_class="archive")
+    IngestionApplication(_Operational(), "opaque", require_custody_class="operational")
+
+
+def test_custody_class_alias_is_not_development():
+    assert custody_class_satisfies("operational", "archive")
+    assert custody_class_satisfies("archive", "operational")
+    assert not custody_class_satisfies("development", "operational")
+    assert not custody_class_satisfies("development", "archive")
+    assert custody_class_satisfies("development", "development")
+
+
+def test_readiness_reports_recovery_fields_for_operational_backend():
+    class _Operational(_BrokenBackend):
+        @property
+        def retention_policy(self):
+            return RetentionPolicy(
+                custody_class="operational",
+                retention_days=None,
+                immutable=True,
+                tamper_evidence=False,
+                durable=True,
+                recoverable_days=30,
+                recoverable_source="resource-control/data/capability/platform-audit-storage.json",
+                recoverable_basis="measured",
+            )
+
+        def health(self):
+            return None
+
+    status, body = invoke(
+        IngestionApplication(_Operational(), "opaque"),
+        None, path="/readyz", method="GET", body=b"",
+    )
+    assert status.startswith("200")
+    assert body["custody_class"] == "operational"
+    assert body["durable"] is True
+    assert body["recoverable_days"] == 30
+    assert body["recoverable_basis"] == "measured"
+    assert "platform-audit-storage" in body["recoverable_source"]
+
+
 def test_counters_track_each_outcome(tmp_path):
     app, _ = bound_app(tmp_path, may_read=True)
     invoke(app, event())                                   # accepted

tests/test_interface.py

@@ -8,6 +8,7 @@ from audit_core.interface import (
     EventValidationError,
     RetentionPolicy,
     SCHEMA_VERSION_V1ALPHA1,
+    custody_class_satisfies,
     validate_event,
 )
 from audit_core.mock_file_backend import MockFileAuditBackend
@@ -79,6 +80,29 @@ def test_mock_backend_retention_policy_none_when_cleanup_disabled():
     assert backend.retention_policy.retention_days is None
 
 
+def test_custody_class_satisfies_aliases_operational_and_archive():
+    assert custody_class_satisfies("operational", "archive") is True
+    assert custody_class_satisfies("archive", "operational") is True
+    assert custody_class_satisfies("operational", "development") is False
+
+
+def test_readiness_payload_includes_recovery_when_declared():
+    policy = RetentionPolicy(
+        custody_class="operational",
+        retention_days=None,
+        immutable=True,
+        tamper_evidence=False,
+        durable=True,
+        recoverable_days=30,
+        recoverable_source="cited",
+        recoverable_basis="measured",
+    )
+    body = policy.as_readiness()
+    assert body["status"] == "ok"
+    assert body["custody_class"] == "operational"
+    assert body["recoverable_days"] == 30
+
+
 def test_audit_event_record_uses_v1alpha1_schema():
     event = AuditEvent(
         source="audit-core",

tests/test_senders.py

@@ -0,0 +1,97 @@
+import json
+from pathlib import Path
+
+import pytest
+
+from audit_core.senders import SenderRegistry, WILDCARD
+
+ROOT = Path(__file__).resolve().parents[1]
+SCOPE_FILE = ROOT / "deploy" / "senders-scope.json"
+
+
+def test_declared_scope_keeps_user_engine_tenants_wildcard():
+    scope = json.loads(SCOPE_FILE.read_text())
+    user_engine = next(entry for entry in scope if entry["name"] == "user-engine")
+    assert user_engine["tenants"] == ["*"]
+    assert user_engine["sources"] == ["user-engine"]
+    assert user_engine["may_write"] is True
+    assert user_engine["may_read"] is False
+    assert "tokens" not in user_engine
+    assert "token" not in user_engine
+
+
+def test_scope_overlay_widens_narrow_secret_tenants(tmp_path):
+    secret = json.dumps(
+        [
+            {
+                "name": "user-engine",
+                "tokens": ["live-token"],
+                "sources": ["user-engine"],
+                "tenants": ["tenant:friendly:binky"],
+                "may_write": True,
+                "may_read": False,
+            }
+        ]
+    )
+    env = {
+        "AUDIT_CORE_SENDERS": secret,
+        "AUDIT_CORE_SENDERS_SCOPE_PATH": str(SCOPE_FILE),
+    }
+    registry = SenderRegistry.from_env(env)
+    identity = registry.authenticate("Bearer live-token")
+    assert identity is not None
+    assert WILDCARD in identity.tenants
+    assert identity.permits_tenant("tenant:other:x")
+    assert identity.tokens == ("live-token",)
+
+
+def test_scope_overlay_does_not_take_tokens_from_git():
+    overlay = json.dumps(
+        [
+            {
+                "name": "user-engine",
+                "tokens": ["must-not-be-used"],
+                "tenants": ["*"],
+            }
+        ]
+    )
+    secret = json.dumps(
+        [
+            {
+                "name": "user-engine",
+                "tokens": ["live-token"],
+                "sources": ["user-engine"],
+                "tenants": ["tenant:friendly:binky"],
+            }
+        ]
+    )
+    registry = SenderRegistry.from_env(
+        {"AUDIT_CORE_SENDERS": secret, "AUDIT_CORE_SENDERS_SCOPE": overlay}
+    )
+    assert registry.authenticate("Bearer must-not-be-used") is None
+    assert registry.authenticate("Bearer live-token") is not None
+
+
+def test_missing_overlay_leaves_secret_as_is():
+    secret = json.dumps(
+        [
+            {
+                "name": "user-engine",
+                "tokens": ["live-token"],
+                "sources": ["user-engine"],
+                "tenants": ["tenant:friendly:binky"],
+            }
+        ]
+    )
+    identity = SenderRegistry.from_env({"AUDIT_CORE_SENDERS": secret}).identities[0]
+    assert identity.tenants == frozenset({"tenant:friendly:binky"})
+
+
+def test_invalid_scope_overlay_is_a_startup_error():
+    secret = json.dumps(
+        [{"name": "user-engine", "tokens": ["t"], "sources": ["user-engine"]}]
+    )
+    with pytest.raises(ValueError, match="JSON list"):
+        SenderRegistry.from_env(
+            {"AUDIT_CORE_SENDERS": secret, "AUDIT_CORE_SENDERS_SCOPE": "{}"}
+        )