diff --git a/LIMITATIONS.md b/LIMITATIONS.md index d0baf31..f881e71 100644 --- a/LIMITATIONS.md +++ b/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. diff --git a/README.md b/README.md index 1296bd8..b45842c 100644 --- a/README.md +++ b/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) diff --git a/ROADMAP.md b/ROADMAP.md index d5a0e11..4764d7f 100644 --- a/ROADMAP.md +++ b/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. diff --git a/docs/architecture.md b/docs/architecture.md index c6b79b2..db1fb07 100644 --- a/docs/architecture.md +++ b/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. diff --git a/docs/evidence-chain.md b/docs/evidence-chain.md new file mode 100644 index 0000000..2e02740 --- /dev/null +++ b/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`. diff --git a/src/agentrust_telemetry/__init__.py b/src/agentrust_telemetry/__init__.py index e4049af..7df76ff 100644 --- a/src/agentrust_telemetry/__init__.py +++ b/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", diff --git a/src/agentrust_telemetry/client.py b/src/agentrust_telemetry/client.py index 7e32b65..94e9100 100644 --- a/src/agentrust_telemetry/client.py +++ b/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,6 +21,10 @@ 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 @@ -27,6 +32,7 @@ class EmitResult: 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, + ) diff --git a/src/agentrust_telemetry/errors.py b/src/agentrust_telemetry/errors.py index eff8e47..bb6bfe7 100644 --- a/src/agentrust_telemetry/errors.py +++ b/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.""" diff --git a/src/agentrust_telemetry/evidence.py b/src/agentrust_telemetry/evidence.py new file mode 100644 index 0000000..8e464d5 --- /dev/null +++ b/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), + ) diff --git a/tests/test_evidence.py b/tests/test_evidence.py new file mode 100644 index 0000000..d2b87d3 --- /dev/null +++ b/tests/test_evidence.py @@ -0,0 +1,172 @@ +import json +import sys +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) + +from agentrust_telemetry import ( # noqa: E402 + EvidenceAccumulator, + EvidenceError, + EvidencePersistenceError, + SchemaValidator, + TelemetryClient, +) + + +def fixture(name): + return json.loads((ROOT / "conformance" / "fixtures" / "valid" / name).read_text()) + + +class RecordingSpan: + def __init__(self): + self.events = [] + + def get_span_context(self): + return type("Context", (), {"is_valid": False})() + + def add_event(self, name, attributes, timestamp): + self.events.append(name) + + +class EvidenceTests(unittest.TestCase): + def setUp(self): + self.validator = SchemaValidator(ROOT / "spec" / "schema") + + def test_chain_is_ordered_and_has_stable_golden_digest(self): + accumulator = EvidenceAccumulator("run-governed-sdlc-001", self.validator) + first = accumulator.append(fixture("policy-decision.json")) + second = accumulator.append(fixture("usage.json")) + + self.assertEqual(first.sequence, 0) + self.assertIsNone(first.previous_digest) + self.assertEqual(second.sequence, 1) + self.assertEqual(second.previous_digest, first.digest) + self.assertEqual( + second.digest, + "e6a03fca3d030c9c0295251fe5feaa467f32df5b3a8ec6b070dd179f3156fc10", + ) + + def test_durable_callback_acknowledges_before_local_acceptance(self): + durable = [] + + def append(entry): + durable.append(entry) + return True + + accumulator = EvidenceAccumulator( + "run-governed-sdlc-001", self.validator, durable_append=append + ) + accepted = accumulator.append(fixture("policy-decision.json")) + self.assertEqual(accumulator.mode, "callback") + self.assertEqual(durable, [accepted]) + self.assertEqual(accumulator.snapshot().entries, (accepted,)) + + def test_failed_durable_ack_is_fail_closed_and_retryable(self): + acknowledgements = iter([False, True]) + accumulator = EvidenceAccumulator( + "run-governed-sdlc-001", + self.validator, + durable_append=lambda entry: next(acknowledgements), + ) + event = fixture("policy-decision.json") + with self.assertRaises(EvidencePersistenceError): + accumulator.append(event) + self.assertEqual(accumulator.snapshot().entries, ()) + self.assertEqual(accumulator.append(event).sequence, 0) + + def test_client_does_not_project_when_evidence_fails(self): + accumulator = EvidenceAccumulator( + "run-governed-sdlc-001", + self.validator, + durable_append=lambda entry: False, + ) + span = RecordingSpan() + with self.assertRaises(EvidencePersistenceError): + TelemetryClient( + self.validator, + span_resolver=lambda: span, + evidence_sink=accumulator, + ).emit(fixture("policy-decision.json")) + self.assertEqual(span.events, []) + + def test_client_reports_evidence_persistence(self): + accumulator = EvidenceAccumulator("run-governed-sdlc-001", self.validator) + result = TelemetryClient( + self.validator, span_resolver=lambda: None, evidence_sink=accumulator + ).emit(fixture("policy-decision.json")) + self.assertTrue(result.evidence_persisted) + self.assertEqual(len(accumulator.snapshot().entries), 1) + + def test_client_isolates_operational_event_from_evidence_sink(self): + class MutatingSink: + def append(self, event): + event["run_id"] = "mutated-by-sink" + + event = fixture("usage.json") + TelemetryClient( + self.validator, span_resolver=lambda: None, evidence_sink=MutatingSink() + ).emit(event) + self.assertEqual(event["run_id"], "run-governed-sdlc-001") + + def test_non_finite_number_cannot_enter_digest_chain(self): + accumulator = EvidenceAccumulator("run-governed-sdlc-001", self.validator) + event = fixture("usage.json") + event["cost"]["amount"] = float("nan") + with self.assertRaisesRegex(EvidenceError, "cannot be canonicalized"): + accumulator.append(event) + self.assertEqual(accumulator.snapshot().entries, ()) + + def test_rejects_cross_run_duplicate_overflow_and_append_after_seal(self): + accumulator = EvidenceAccumulator( + "run-governed-sdlc-001", self.validator, max_events=1 + ) + event = fixture("policy-decision.json") + wrong_run = fixture("usage.json") + wrong_run["run_id"] = "another-run" + with self.assertRaisesRegex(EvidenceError, "does not match"): + accumulator.append(wrong_run) + accumulator.append(event) + with self.assertRaisesRegex(EvidenceError, "duplicate"): + accumulator.append(event) + + full = fixture("usage.json") + with self.assertRaisesRegex(EvidenceError, "max_events"): + accumulator.append(full) + snapshot = accumulator.seal(completeness="incomplete") + self.assertTrue(snapshot.sealed) + self.assertEqual(snapshot.completeness, "incomplete") + with self.assertRaisesRegex(EvidenceError, "sealed"): + accumulator.append(full) + + def test_snapshot_is_defensive_and_unsealed_never_claims_complete(self): + accumulator = EvidenceAccumulator("run-governed-sdlc-001", self.validator) + accumulator.append(fixture("usage.json")) + snapshot = accumulator.snapshot() + snapshot.entries[0].event["run_id"] = "mutated" + fresh = accumulator.snapshot() + self.assertEqual(fresh.completeness, "unknown") + self.assertEqual(fresh.entries[0].event["run_id"], "run-governed-sdlc-001") + + def test_callback_and_returned_entry_cannot_mutate_retained_evidence(self): + callback_entries = [] + + def append(entry): + callback_entries.append(entry) + entry.event["run_id"] = "callback-mutated" + return True + + accumulator = EvidenceAccumulator( + "run-governed-sdlc-001", self.validator, durable_append=append + ) + returned = accumulator.append(fixture("usage.json")) + returned.event["run_id"] = "caller-mutated" + retained = accumulator.snapshot().entries[0] + self.assertEqual(retained.event["run_id"], "run-governed-sdlc-001") + self.assertNotEqual(callback_entries[0].event["run_id"], retained.event["run_id"]) + + +if __name__ == "__main__": + unittest.main()