feat: add durable evidence accumulator

LIMITATIONS.md

@@ -5,7 +5,9 @@ Current `0.1.0-dev` limitations:
 - The contract and SDK are experimental and may change incompatibly.
 - Only a Python reference SDK exists.
 - OTel span events are implemented; the structured-log path is an emitter protocol rather than a concrete OTel Logs adapter.
-- No metrics projector, evidence accumulator, TRACE finalizer, or AGT adapter is implemented.
+- No metrics projector, TRACE finalizer, or AGT adapter is implemented.
+- Evidence memory mode is not durable. Callback mode defines acknowledgement and
+  retry behavior but the adopter owns storage, idempotency, and recovery.
 - Propagation currently supports mutable string mappings; framework-specific HTTP,
   RPC, and messaging carrier adapters are not yet included.
 - The SDK validates declared metadata but cannot prove a producer's policy decision, identity, classification, token count, or cost is truthful.

README.md

@@ -94,6 +94,7 @@ python examples/manual_governance.py
 
 - [Architecture](docs/architecture.md)
 - [OpenTelemetry projection](docs/otel-projection.md)
+- [Evidence chain profile](docs/evidence-chain.md)
 - [Privacy](PRIVACY.md)
 - [Limitations](LIMITATIONS.md)
 - [Roadmap](ROADMAP.md)

ROADMAP.md

@@ -12,7 +12,7 @@
 
 - Concrete OTel Logs and metrics projectors.
 - Additional W3C propagation carrier adapters beyond mutable string mappings.
-- Durable evidence accumulator and official TRACE adapter.
+- Official TRACE adapter over sealed evidence snapshots.
 - AGT, OPA, Cedar, and generic approval adapters.
 - TypeScript SDK and mixed-language conformance.
 

docs/architecture.md

@@ -39,6 +39,18 @@ If OpenTelemetry is installed, the SDK uses the caller's current span and adds a
 
 A caller-owned structured-log emitter may receive a defensive deep copy of the validated event. Projection failures are reported independently in `EmitResult`.
 
+### Evidence accumulator
+
+The optional accumulator receives validated events before OTel or log projection.
+It assigns a monotonic sequence and chains canonical event bytes with SHA-256.
+Memory mode is development-only. Callback mode requires the adopter's callback
+to acknowledge durable commit and to be idempotent by `event_id`; failure aborts
+operational projection and leaves local accumulator state retryable.
+
+Sealing requires an explicit caller assessment of `complete`, `incomplete`, or
+`unknown`. An open run always reports `unknown`. This assessment is an input to
+future TRACE mapping, not proof generated by the accumulator.
+
 ## Correlation semantics
 
 - `run_id` is the durable execution correlation key.

docs/evidence-chain.md

@@ -0,0 +1,28 @@
+# Evidence chain profile
+
+Status: experimental `agentrust-json-v1` for contract `0.1.0-dev`.
+
+Each accepted event is validated and privacy-checked before entering the chain.
+Sequence numbers start at zero and represent acceptance order, not event time.
+
+For entry `n`, calculate:
+
+```text
+event_bytes = UTF-8(JSON(event, keys sorted, no insignificant whitespace,
+                        non-ASCII preserved, NaN and infinity rejected))
+previous = 32 zero bytes for entry 0, otherwise raw bytes of entry n-1 digest
+material = previous || uint64_big_endian(n) || event_bytes
+digest = lowercase_hex(SHA-256(material))
+```
+
+The profile name is explicit because this pre-alpha JSON serialization is not a
+claim of RFC 8785 conformance. A future contract may adopt a standard canonical
+form through a versioned profile; existing chains retain their original profile.
+
+Callback-mode writers must commit idempotently by `event_id` and return `True`
+only after durable acknowledgement. A false return or exception leaves local
+sequence and chain state unchanged. The callback receives an isolated copy.
+
+Sealing prevents further appends. `completeness` is explicitly supplied by the
+caller and must remain `unknown` unless the caller has an independent basis for
+asserting `complete` or `incomplete`.

src/agentrust_telemetry/__init__.py

@@ -2,7 +2,15 @@
 
 from .client import EmitResult, TelemetryClient
 from .context import ContextIds, active_context_ids
-from .errors import ContextMismatchError, EventValidationError, ProjectionError, PropagationError
+from .errors import (
+    ContextMismatchError,
+    EventValidationError,
+    EvidenceError,
+    EvidencePersistenceError,
+    ProjectionError,
+    PropagationError,
+)
+from .evidence import EvidenceAccumulator, EvidenceEntry, EvidenceSnapshot
 from .propagation import ExtractedContext, extract_context, inject_context
 from .validation import SchemaValidator
 
@@ -11,6 +19,11 @@
     "ContextMismatchError",
     "EmitResult",
     "EventValidationError",
+    "EvidenceAccumulator",
+    "EvidenceEntry",
+    "EvidenceError",
+    "EvidencePersistenceError",
+    "EvidenceSnapshot",
     "ExtractedContext",
     "ProjectionError",
     "PropagationError",

src/agentrust_telemetry/client.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from copy import deepcopy
 from dataclasses import dataclass
 from typing import Any, Callable, Protocol
 
@@ -20,13 +21,18 @@ class LogEmitter(Protocol):
     def emit(self, record: dict[str, Any]) -> None: ...
 
 
+class EvidenceSink(Protocol):
+    def append(self, event: dict[str, Any]) -> Any: ...
+
+
 @dataclass(frozen=True)
 class EmitResult:
     accepted: bool
     span_event_emitted: bool
     log_emitted: bool
     context: ContextIds | None
     projection_errors: tuple[str, ...] = ()
+    evidence_persisted: bool = False
 
 
 class TelemetryClient:
@@ -36,13 +42,20 @@ def __init__(
         *,
         span_resolver: Callable[[], SpanLike | None] | None = None,
         log_emitter: LogEmitter | None = None,
+        evidence_sink: EvidenceSink | None = None,
     ):
         self._validator = validator
         self._span_resolver = span_resolver
         self._log_emitter = log_emitter
+        self._evidence_sink = evidence_sink
 
     def emit(self, event: dict[str, Any]) -> EmitResult:
         self._validator.validate(event)
+        evidence_persisted = False
+        if self._evidence_sink is not None:
+            # Evidence is accepted before any best-effort operational projection.
+            self._evidence_sink.append(deepcopy(event))
+            evidence_persisted = True
         resolver = self._span_resolver or current_span
         span = resolver()
         context = active_context_ids(lambda: span)
@@ -73,4 +86,11 @@ def emit(self, event: dict[str, Any]) -> EmitResult:
             except Exception as exc:  # exporter implementations are external
                 errors.append(f"log projection failed: {type(exc).__name__}: {exc}")
 
-        return EmitResult(True, span_emitted, log_emitted, context, tuple(errors))
+        return EmitResult(
+            True,
+            span_emitted,
+            log_emitted,
+            context,
+            tuple(errors),
+            evidence_persisted,
+        )

src/agentrust_telemetry/errors.py

@@ -12,3 +12,11 @@ class ProjectionError(RuntimeError):
 
 class PropagationError(ValueError):
     """AgentTrust propagation metadata is missing or unsafe."""
+
+
+class EvidenceError(RuntimeError):
+    """Evidence could not be accepted without weakening its guarantees."""
+
+
+class EvidencePersistenceError(EvidenceError):
+    """The configured durable evidence callback did not acknowledge an entry."""

src/agentrust_telemetry/evidence.py

@@ -0,0 +1,167 @@
+"""Deterministic, pre-sampling evidence accumulation."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from copy import deepcopy
+from dataclasses import dataclass
+from threading import Lock
+from typing import Any, Callable, Literal
+
+from .errors import EvidenceError, EvidencePersistenceError
+from .validation import SchemaValidator
+
+
+Completeness = Literal["complete", "incomplete", "unknown"]
+CANONICALIZATION_PROFILE = "agentrust-json-v1"
+_GENESIS_DIGEST = bytes(32)
+
+
+@dataclass(frozen=True)
+class EvidenceEntry:
+    sequence: int
+    event_id: str
+    previous_digest: str | None
+    digest: str
+    event: dict[str, Any]
+
+
+@dataclass(frozen=True)
+class EvidenceSnapshot:
+    run_id: str
+    entries: tuple[EvidenceEntry, ...]
+    chain_digest: str | None
+    canonicalization_profile: str
+    completeness: Completeness
+    sealed: bool
+
+
+DurableAppend = Callable[[EvidenceEntry], bool]
+
+
+class EvidenceAccumulator:
+    """Accumulate one run's validated events before operational projection.
+
+    A durable callback must be idempotent by ``event_id`` and return ``True`` only
+    after the entry is durably committed. Callback failure leaves local state
+    unchanged so callers can retry the same event.
+    """
+
+    def __init__(
+        self,
+        run_id: str,
+        validator: SchemaValidator,
+        *,
+        durable_append: DurableAppend | None = None,
+        max_events: int = 10_000,
+    ) -> None:
+        if not isinstance(run_id, str) or not run_id:
+            raise EvidenceError("run_id must be a non-empty string")
+        if max_events < 1:
+            raise EvidenceError("max_events must be at least 1")
+        self._run_id = run_id
+        self._validator = validator
+        self._durable_append = durable_append
+        self._max_events = max_events
+        self._entries: list[EvidenceEntry] = []
+        self._event_ids: set[str] = set()
+        self._sealed = False
+        self._completeness: Completeness = "unknown"
+        self._lock = Lock()
+
+    @property
+    def mode(self) -> Literal["memory", "callback"]:
+        return "callback" if self._durable_append is not None else "memory"
+
+    def append(self, event: dict[str, Any]) -> EvidenceEntry:
+        """Validate, chain, durably acknowledge, then accept an event."""
+        self._validator.validate(event)
+        event_copy = deepcopy(event)
+        if event_copy["run_id"] != self._run_id:
+            raise EvidenceError("event run_id does not match accumulator run_id")
+
+        with self._lock:
+            if self._sealed:
+                raise EvidenceError("evidence run is sealed")
+            event_id = event_copy["event_id"]
+            if event_id in self._event_ids:
+                raise EvidenceError(f"duplicate event_id: {event_id}")
+            if len(self._entries) >= self._max_events:
+                raise EvidenceError(f"evidence run exceeds max_events={self._max_events}")
+
+            sequence = len(self._entries)
+            previous = self._entries[-1].digest if self._entries else None
+            try:
+                digest = _entry_digest(sequence, previous, event_copy)
+            except (TypeError, ValueError) as exc:
+                raise EvidenceError(
+                    f"event_id={event_id} cannot be canonicalized under "
+                    f"{CANONICALIZATION_PROFILE}"
+                ) from exc
+            entry = EvidenceEntry(sequence, event_id, previous, digest, event_copy)
+
+            if self._durable_append is not None:
+                try:
+                    acknowledged = self._durable_append(_copy_entry(entry))
+                except Exception as exc:
+                    raise EvidencePersistenceError(
+                        f"durable evidence callback failed for event_id={event_id}"
+                    ) from exc
+                if acknowledged is not True:
+                    raise EvidencePersistenceError(
+                        f"durable evidence callback did not acknowledge event_id={event_id}"
+                    )
+
+            self._entries.append(_copy_entry(entry))
+            self._event_ids.add(event_id)
+            return _copy_entry(entry)
+
+    def seal(self, *, completeness: Completeness) -> EvidenceSnapshot:
+        """Close the run with a caller-asserted completeness assessment."""
+        if completeness not in ("complete", "incomplete", "unknown"):
+            raise EvidenceError(f"unsupported completeness: {completeness!r}")
+        with self._lock:
+            if self._sealed:
+                raise EvidenceError("evidence run is already sealed")
+            self._sealed = True
+            self._completeness = completeness
+            return self._snapshot()
+
+    def snapshot(self) -> EvidenceSnapshot:
+        with self._lock:
+            return self._snapshot()
+
+    def _snapshot(self) -> EvidenceSnapshot:
+        entries = tuple(_copy_entry(item) for item in self._entries)
+        return EvidenceSnapshot(
+            run_id=self._run_id,
+            entries=entries,
+            chain_digest=entries[-1].digest if entries else None,
+            canonicalization_profile=CANONICALIZATION_PROFILE,
+            completeness=self._completeness if self._sealed else "unknown",
+            sealed=self._sealed,
+        )
+
+
+def _entry_digest(sequence: int, previous_digest: str | None, event: dict[str, Any]) -> str:
+    canonical = json.dumps(
+        event,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+        allow_nan=False,
+    ).encode("utf-8")
+    previous = bytes.fromhex(previous_digest) if previous_digest else _GENESIS_DIGEST
+    material = previous + sequence.to_bytes(8, "big") + canonical
+    return hashlib.sha256(material).hexdigest()
+
+
+def _copy_entry(entry: EvidenceEntry) -> EvidenceEntry:
+    return EvidenceEntry(
+        entry.sequence,
+        entry.event_id,
+        entry.previous_digest,
+        entry.digest,
+        deepcopy(entry.event),
+    )