AUDIT-WP-0009-T03/T09 — evidence_kind, and approval-engine's registration inputs

T03. §9.6 gives load-bearing and attributive sources different obligations, so
the archive must record which one a source declared rather than infer it from
traffic. evidence_kind and completeness_trade now sit on SenderIdentity, the
AUDIT_CORE_SENDERS schema, and the non-secret scope overlay.

Two asymmetries are deliberate. The default is attributive, because the other
default would have audit-core imply a completeness obligation no source ever
accepted. And the overlay may raise the kind but never lower it — the same
principle that stops an ExternalSecret refresh shrinking user-engine's
tenants: a ConfigMap refresh must not drop a source's atomicity and detection
obligations without anyone deciding to. A load-bearing source may not carry a
completeness trade at all, since §9.6 requires atomicity of it, and
evidence_declaration() reports completeness_claimed: false for both kinds.

T09 (in progress). approval-engine's registration inputs are prepared and
recorded in docs/approval-engine-source-registration.md: scope entry declared
load-bearing, and audit-core-approval-engine-ingress with namespace and pod
label ANDed in one `from` peer — narrower than user-engine's namespace-only
rule, which is left unchanged. The scope entry lands ahead of the credential
because the overlay only applies to senders the Secret already carries, so it
admits nothing until the token exists; a test asserts that rather than
trusting the reading.

Two inputs remain approval-engine's: a confirmed tenant scope, since senders.py
requires a missing tenant restriction be justified per sender and audit-core
cannot justify it on another repo's behalf, and an explicit secret_policy
choice. Applying the manifests is an operator action.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0185wifnLzCxjEY2MT1XbK7L

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 713962@bnt-lap001
Assistant-Session: 2718d99d-d3ff-478f-83a2-3a30f01a02fc

audit_core/senders.py

@@ -25,6 +25,19 @@ from audit_core.redaction import POLICIES, POLICY_REDACT
 
 WILDCARD = "*"
 
+# §9.6 evidence kinds. The obligations differ, so the archive must record
+# which one a source declared rather than infer it from traffic.
+#
+# load-bearing: a control's soundness depends on the event being present, so
+#   the source owes emission atomicity and a detection surface (heartbeat or
+#   reconciliation) for rare classes.
+# attributive: the trail is useful but no control depends on its completeness,
+#   which is what lets a deliberately non-atomic trail stay legitimate — while
+#   it is declared, and never described as complete.
+EVIDENCE_LOAD_BEARING = "load-bearing"
+EVIDENCE_ATTRIBUTIVE = "attributive"
+EVIDENCE_KINDS = (EVIDENCE_LOAD_BEARING, EVIDENCE_ATTRIBUTIVE)
+
 
 @dataclass(frozen=True)
 class SenderIdentity:
@@ -44,6 +57,17 @@ class SenderIdentity:
     # legitimate event is not lost over one field; a higher-assurance channel
     # can be set to reject instead (AUDIT-WP-0004-T04).
     secret_policy: str = POLICY_REDACT
+    # §9.6 evidence kind (AUDIT-WP-0009-T03). Defaults to ``attributive`` on
+    # purpose: a source that has not declared must not be silently treated as
+    # load-bearing, because that would let audit-core imply a completeness
+    # obligation the source never accepted. Declaring load-bearing is an
+    # explicit act by the source and its owner.
+    evidence_kind: str = EVIDENCE_ATTRIBUTIVE
+    # The trade an attributive source has deliberately made — typically
+    # emitting after commit rather than inside the state-change transaction.
+    # §9.6 requires the trade be declared where the trail is documented, so it
+    # travels with the identity rather than living only in prose.
+    completeness_trade: str | None = None
 
     def __post_init__(self) -> None:
         if not self.name:
@@ -59,6 +83,43 @@ class SenderIdentity:
             raise ValueError(f"sender {self.name!r} needs at least one permitted source")
         if self.expires_at is not None and self.expires_at.tzinfo is None:
             raise ValueError(f"sender {self.name!r}: expires_at must include a timezone")
+        if self.evidence_kind not in EVIDENCE_KINDS:
+            raise ValueError(
+                f"sender {self.name!r}: evidence_kind must be one of "
+                f"{EVIDENCE_KINDS}, got {self.evidence_kind!r}"
+            )
+        if self.completeness_trade is not None and not str(self.completeness_trade).strip():
+            raise ValueError(
+                f"sender {self.name!r}: completeness_trade must say what the trade "
+                "is, or be omitted"
+            )
+        if self.is_load_bearing and self.completeness_trade is not None:
+            # A load-bearing source has no trade to make: §9.6 requires
+            # atomicity of it. Carrying a declared trade alongside would
+            # record a contradiction as though it were a policy.
+            raise ValueError(
+                f"sender {self.name!r}: a load-bearing source may not declare a "
+                "completeness_trade — §9.6 requires emission atomicity of it"
+            )
+
+    @property
+    def is_load_bearing(self) -> bool:
+        return self.evidence_kind == EVIDENCE_LOAD_BEARING
+
+    def evidence_declaration(self) -> dict[str, Any]:
+        """What audit-core will say about this source's stream.
+
+        Deliberately states the bound rather than only the kind. Neither kind
+        licenses the claim that the archive proves an event occurred, or that
+        absence proves it did not (§9.6).
+        """
+        return {
+            "sender": self.name,
+            "evidence_kind": self.evidence_kind,
+            "completeness_claimed": False,
+            "completeness_trade": self.completeness_trade,
+            "detection_surface": None,
+        }
 
     def permits_source(self, source: str) -> bool:
         return WILDCARD in self.sources or source in self.sources
@@ -221,11 +282,37 @@ def _apply_scope_overlay(
                 ),
                 secret_policy=identity.secret_policy,
                 expires_at=identity.expires_at,
+                evidence_kind=_overlay_evidence_kind(identity, extra),
+                completeness_trade=(
+                    _clean_trade(extra["completeness_trade"])
+                    if "completeness_trade" in extra
+                    else identity.completeness_trade
+                ),
             )
         )
     return merged
 
 
+def _overlay_evidence_kind(identity: SenderIdentity, extra: dict[str, Any]) -> str:
+    """Let the overlay raise the evidence kind, never lower it.
+
+    Same principle as tenants: the ConfigMap is the authority for non-secret
+    policy, but a refresh must not be able to quietly weaken a declaration.
+    Downgrading a load-bearing source to attributive would drop its atomicity
+    and detection obligations without anyone deciding to, so it is refused
+    here and has to be a deliberate change to the Secret-backed declaration.
+    """
+    declared = str(extra.get("evidence_kind") or identity.evidence_kind)
+    if declared == identity.evidence_kind:
+        return declared
+    if identity.is_load_bearing and declared == EVIDENCE_ATTRIBUTIVE:
+        raise ValueError(
+            f"sender {identity.name!r}: the scope overlay may not downgrade a "
+            "load-bearing source to attributive"
+        )
+    return declared
+
+
 def _parse_identities(raw: str) -> list[SenderIdentity]:
     try:
         entries = json.loads(raw)
@@ -249,9 +336,17 @@ def _identity_from(entry: Any) -> SenderIdentity:
         may_read=bool(entry.get("may_read", False)),
         secret_policy=str(entry.get("secret_policy", POLICY_REDACT)),
         expires_at=_parse_expiry(entry.get("expires_at")),
+        evidence_kind=str(entry.get("evidence_kind", EVIDENCE_ATTRIBUTIVE)),
+        completeness_trade=_clean_trade(entry.get("completeness_trade")),
     )
 
 
+def _clean_trade(value: Any) -> str | None:
+    if value in (None, ""):
+        return None
+    return str(value)
+
+
 def _parse_expiry(value: Any) -> datetime | None:
     if value in (None, ""):
         return None