diff --git a/CHANGELOG.md b/CHANGELOG.md index da2c0ae..60cb757 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/compatibility/golden/delivery-faults.json b/compatibility/golden/delivery-faults.json new file mode 100644 index 0000000..f88d926 --- /dev/null +++ b/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} + ] +} diff --git a/docs/delivery-fault-testing.md b/docs/delivery-fault-testing.md new file mode 100644 index 0000000..e0c8195 --- /dev/null +++ b/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. diff --git a/docs/evidence-chain.md b/docs/evidence-chain.md index 43dff25..436db88 100644 --- a/docs/evidence-chain.md +++ b/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. diff --git a/packages/typescript/test/delivery-faults.test.ts b/packages/typescript/test/delivery-faults.test.ts new file mode 100644 index 0000000..3e6f8de --- /dev/null +++ b/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> = []; + 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(); + 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); + }); +} diff --git a/tests/test_delivery_faults.py b/tests/test_delivery_faults.py new file mode 100644 index 0000000..920e8aa --- /dev/null +++ b/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()