test(evidence): exercise delivery faults and lost acknowledgements

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+### Added
+
+- Shared Python/TypeScript delivery-fault tests for dropped, duplicated and
+  reordered projections and lost persistence acknowledgements, plus Python
+  OpenTelemetry sampling/export-failure coverage. These tests do not establish
+  collector interoperability or durable backend delivery.
+
 ### Changed
 
 - CI checks the existing four job-level OIDC grants across all workflow files,

compatibility/golden/delivery-faults.json

@@ -0,0 +1,11 @@
+{
+  "scope": "synthetic operational delivery; not a collector or durability claim",
+  "fixtures": ["policy-decision.json", "usage.json"],
+  "cases": [
+    {"mode": "healthy", "delivered_indices": [0, 1], "log_emitted": true, "error_count": 0},
+    {"mode": "drop", "delivered_indices": [], "log_emitted": true, "error_count": 0},
+    {"mode": "duplicate", "delivered_indices": [0, 0, 1, 1], "log_emitted": true, "error_count": 0},
+    {"mode": "reverse", "delivered_indices": [1, 0], "log_emitted": true, "error_count": 0},
+    {"mode": "raise", "delivered_indices": [], "log_emitted": false, "error_count": 1}
+  ]
+}

docs/delivery-fault-testing.md

@@ -0,0 +1,41 @@
+# Delivery-fault acceptance tests
+
+The Python and TypeScript tests read the same delivery matrix in
+`compatibility/golden/delivery-faults.json` and compare retained evidence with
+the existing independently pinned evidence-chain golden digests. Healthy,
+dropped, duplicated, reversed and throwing log delivery must leave evidence in
+its original acceptance order. Successful emitter return is not acknowledgement
+from a collector. Neither delivery success nor failure establishes completeness.
+
+Both SDKs also exercise a callback that records a commit but loses its first
+acknowledgement, either by returning false or raising. The local chain and
+operational projection remain untouched until the identical event is retried
+and acknowledged. An idempotent test store observes one logical entry, and a
+subsequent duplicate is refused before reaching the store or projection again.
+The store is a test double: no crash recovery or durable persistence is proved.
+
+Python additionally uses the real OpenTelemetry SDK with explicit ALWAYS_OFF
+and ALWAYS_ON samplers and an exporter returning FAILURE. The caller-owned
+provider is shut down without changing global providers. Evidence remains
+available even with no sampled span; an export failure after `add_event` is not
+retroactively visible in `EmitResult`. The TypeScript tests cover the shared
+SDK contract, not parity of real OpenTelemetry exporter implementations.
+
+Run with the repository's locked test dependencies:
+
+```sh
+python -m unittest discover -s tests -p test_delivery_faults.py -v
+cd packages/typescript
+npm ci --ignore-scripts
+npm run check
+```
+
+## Remaining collector acceptance
+
+These are executable SDK fault tests, not OTLP wire or collector interoperability
+results. A separate run must pin collector/exporter versions and configuration,
+retain receiver observations, and exercise retry, queue exhaustion, restart,
+out-of-order delivery and partial failure. Its oracle must separately count
+accepted evidence, attempted projection, collector receipt and durable backend
+acknowledgement. A successful flush or exporter return alone is insufficient.
+No production backend, account or network endpoint is required by these tests.

docs/evidence-chain.md

@@ -33,3 +33,7 @@ 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`.
+
+See [delivery-fault acceptance tests](delivery-fault-testing.md) for executable
+Python and TypeScript coverage of lost acknowledgements and best-effort
+projection failures, including the limits of those tests.

packages/typescript/test/delivery-faults.test.ts

@@ -0,0 +1,86 @@
+import assert from "node:assert/strict";
+import {readFileSync} from "node:fs";
+import test from "node:test";
+import {fileURLToPath} from "node:url";
+import {EvidenceAccumulator, EvidenceError, EvidencePersistenceError, SchemaValidator, TelemetryClient} from "../src/index.js";
+import type {EvidenceEntry, NormalizedEvent} from "../src/index.js";
+
+const root = fileURLToPath(new URL("../../../", import.meta.url));
+const read = (path: string) => JSON.parse(readFileSync(`${root}/${path}`, "utf8"));
+const fixture = (name: string): NormalizedEvent => read(`conformance/fixtures/valid/${name}`) as NormalizedEvent;
+const matrix = read("compatibility/golden/delivery-faults.json") as {
+  fixtures: string[];
+  cases: Array<{mode: string; delivered_indices: number[]; log_emitted: boolean; error_count: number}>;
+};
+const golden = read("compatibility/golden/evidence-chain.json") as {run_id: string; entries: Array<{digest: string}>};
+
+for (const scenario of matrix.cases) {
+  test(`delivery ${scenario.mode} preserves evidence and unknown completeness`, () => {
+    const observed: Array<Record<string, unknown>> = [];
+    const validator = SchemaValidator.bundled();
+    const evidence = new EvidenceAccumulator(golden.run_id, validator);
+    const client = new TelemetryClient(validator, {
+      evidenceSink: evidence,
+      logEmitter: {emit(record) {
+        if (scenario.mode === "raise") throw new Error("synthetic delivery failure");
+        if (scenario.mode === "drop") return;
+        if (scenario.mode === "reverse") observed.unshift(record);
+        else {
+          observed.push(record);
+          if (scenario.mode === "duplicate") observed.push(record);
+        }
+      }},
+    });
+    const events = matrix.fixtures.map(fixture);
+    for (const event of events) {
+      const result = client.emit(event);
+      assert.equal(result.accepted, true);
+      assert.equal(result.evidencePersisted, true);
+      assert.equal(result.logEmitted, scenario.log_emitted);
+      assert.equal(result.projectionErrors.length, scenario.error_count);
+    }
+    const snapshot = evidence.snapshot();
+    assert.deepEqual(snapshot.entries.map(e => e.digest), golden.entries.map(e => e.digest));
+    assert.deepEqual(observed.map(r => (r.body as NormalizedEvent).event_id),
+      scenario.delivered_indices.map(i => events[i]!.event_id));
+    assert.equal(snapshot.completeness, "unknown");
+    assert.equal(snapshot.sealed, false);
+    assert.equal(evidence.seal("unknown").completeness, "unknown");
+  });
+}
+
+for (const failure of ["false", "exception"]) {
+  test(`lost ${failure} acknowledgement retries without double projection`, () => {
+    // An idempotent adopter-store test double; not durable storage.
+    const committed = new Map<string, EvidenceEntry>();
+    const attempts: EvidenceEntry[] = [];
+    const projections: unknown[] = [];
+    const validator = SchemaValidator.bundled();
+    const event = fixture("policy-decision.json");
+    const evidence = new EvidenceAccumulator(event.run_id, validator, {durableAppend(entry) {
+      attempts.push(entry);
+      const old = committed.get(entry.eventId);
+      if (old) assert.deepEqual(entry, old);
+      else committed.set(entry.eventId, entry);
+      if (attempts.length === 1) {
+        if (failure === "exception") throw new Error("commit happened; acknowledgement lost");
+        return false;
+      }
+      return true;
+    }});
+    const client = new TelemetryClient(validator, {evidenceSink: evidence,
+      logEmitter: {emit(record) {projections.push(record);}}});
+    assert.throws(() => client.emit(event), EvidencePersistenceError);
+    assert.equal(committed.size, 1);
+    assert.deepEqual(evidence.snapshot().entries, []);
+    assert.deepEqual(projections, []);
+    assert.equal(client.emit(event).evidencePersisted, true);
+    assert.deepEqual(attempts[0], attempts[1]);
+    assert.equal(committed.size, 1);
+    assert.equal(evidence.snapshot().entries.length, 1);
+    assert.equal(projections.length, 1);
+    assert.throws(() => client.emit(event), EvidenceError);
+    assert.equal(attempts.length, 2);
+    assert.equal(projections.length, 1);
+  });
+}

tests/test_delivery_faults.py

@@ -0,0 +1,139 @@
+"""Operational delivery and evidence acceptance are different boundaries."""
+
+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 DeliveryFaultTests(unittest.TestCase):
+    def test_shared_delivery_matrix_preserves_evidence_and_unknown_completeness(self):
+        matrix = json.loads((ROOT / "compatibility/golden/delivery-faults.json").read_text())
+        golden = json.loads((ROOT / "compatibility/golden/evidence-chain.json").read_text())
+        for case in matrix["cases"]:
+            with self.subTest(mode=case["mode"]):
+                observed = []
+
+                class Emitter:
+                    def emit(self, record):
+                        if case["mode"] == "raise":
+                            raise RuntimeError("synthetic delivery failure")
+                        if case["mode"] == "drop":
+                            return
+                        if case["mode"] == "reverse":
+                            observed.insert(0, record)
+                        else:
+                            observed.append(record)
+                            if case["mode"] == "duplicate":
+                                observed.append(record)
+
+                validator = SchemaValidator.bundled()
+                evidence = EvidenceAccumulator(golden["run_id"], validator)
+                client = TelemetryClient(validator, span_resolver=lambda: None,
+                                         evidence_sink=evidence, log_emitter=Emitter())
+                events = [fixture(name) for name in matrix["fixtures"]]
+                for event in events:
+                    result = client.emit(event)
+                    self.assertTrue(result.accepted)
+                    self.assertTrue(result.evidence_persisted)
+                    self.assertEqual(result.log_emitted, case["log_emitted"])
+                    self.assertEqual(len(result.projection_errors), case["error_count"])
+                snapshot = evidence.snapshot()
+                self.assertEqual([e.digest for e in snapshot.entries],
+                                 [e["digest"] for e in golden["entries"]])
+                self.assertEqual([r["body"]["event_id"] for r in observed],
+                                 [events[i]["event_id"] for i in case["delivered_indices"]])
+                self.assertEqual(snapshot.completeness, "unknown")
+                self.assertFalse(snapshot.sealed)
+                self.assertEqual(evidence.seal(completeness="unknown").completeness, "unknown")
+
+    def test_committed_but_unacknowledged_event_retries_without_double_projection(self):
+        for failure in ("false", "exception"):
+            with self.subTest(failure=failure):
+                # Test double for an idempotent adopter store, not durable storage.
+                committed = {}
+                attempts = []
+                projections = []
+
+                def append(entry):
+                    attempts.append(entry)
+                    old = committed.setdefault(entry.event_id, entry)
+                    if old != entry:
+                        raise ValueError("conflicting retry")
+                    if len(attempts) == 1:
+                        if failure == "exception":
+                            raise TimeoutError("commit happened; acknowledgement lost")
+                        return False
+                    return True
+
+                class Emitter:
+                    def emit(self, record):
+                        projections.append(record)
+
+                validator = SchemaValidator.bundled()
+                event = fixture("policy-decision.json")
+                evidence = EvidenceAccumulator(event["run_id"], validator, durable_append=append)
+                client = TelemetryClient(validator, span_resolver=lambda: None,
+                                         evidence_sink=evidence, log_emitter=Emitter())
+                with self.assertRaises(EvidencePersistenceError):
+                    client.emit(event)
+                self.assertEqual(len(committed), 1)
+                self.assertEqual(evidence.snapshot().entries, ())
+                self.assertEqual(projections, [])
+                self.assertTrue(client.emit(event).evidence_persisted)
+                self.assertEqual(attempts[0], attempts[1])
+                self.assertEqual(len(committed), 1)
+                self.assertEqual(len(evidence.snapshot().entries), 1)
+                self.assertEqual(len(projections), 1)
+                with self.assertRaises(EvidenceError):
+                    client.emit(event)
+                self.assertEqual(len(attempts), 2)
+                self.assertEqual(len(projections), 1)
+
+    def test_real_otel_sampling_and_export_failure_do_not_prove_delivery(self):
+        from opentelemetry.sdk.trace import TracerProvider
+        from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter, SpanExportResult
+        from opentelemetry.sdk.trace.sampling import ALWAYS_OFF, ALWAYS_ON
+
+        for sampled in (False, True):
+            with self.subTest(sampled=sampled):
+                exported = []
+
+                class FailingExporter(SpanExporter):
+                    def export(self, spans):
+                        exported.extend(spans)
+                        return SpanExportResult.FAILURE
+
+                provider = TracerProvider(sampler=ALWAYS_ON if sampled else ALWAYS_OFF)
+                provider.add_span_processor(SimpleSpanProcessor(FailingExporter()))
+                try:
+                    validator = SchemaValidator.bundled()
+                    event = fixture("usage.json")
+                    event.pop("trace_id", None)
+                    event.pop("span_id", None)
+                    evidence = EvidenceAccumulator(event["run_id"], validator)
+                    with provider.get_tracer("delivery-test").start_as_current_span("step"):
+                        result = TelemetryClient(validator, evidence_sink=evidence).emit(event)
+                    self.assertTrue(result.evidence_persisted)
+                    self.assertEqual(len(evidence.snapshot().entries), 1)
+                    self.assertEqual(evidence.snapshot().completeness, "unknown")
+                    self.assertEqual(len(exported), int(sampled))
+                    if sampled:
+                        self.assertEqual(len(exported[0].events), 1)
+                    # add_event returning is not collector acknowledgement.
+                    self.assertTrue(result.span_event_emitted)
+                    self.assertEqual(result.projection_errors, ())
+                finally:
+                    provider.shutdown()