test: validate Python and TypeScript collector fault cases

.github/workflows/collector-experiment.yml

@@ -0,0 +1,46 @@
+name: Collector fault experiment
+on:
+  workflow_dispatch:
+  pull_request:
+    paths:
+      - 'experiments/collector/**'
+      - '.github/workflows/collector-experiment.yml'
+permissions:
+  contents: read
+jobs:
+  collector:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
+        with:
+          python-version: '3.12'
+      - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 # v7.1.0
+        with:
+          node-version: '22'
+      - name: Install experiment dependencies and verify collector
+        run: |
+          python -m pip install -r experiments/collector/environment.txt
+          (cd packages/typescript && npm ci --ignore-scripts && npm run build)
+          (cd experiments/collector && npm ci --ignore-scripts)
+          curl -fsSL https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.163.2/otelcol_0.163.2_linux_amd64.tar.gz -o "$RUNNER_TEMP/otelcol.tar.gz"
+          echo "9fbf3101331c88aab4b7f4cfd7c788b252e91e193793e39bd2154971c87f88a7  $RUNNER_TEMP/otelcol.tar.gz" | sha256sum --check
+          tar -xzf "$RUNNER_TEMP/otelcol.tar.gz" -C "$RUNNER_TEMP" otelcol
+      - name: Run synthetic loopback delivery cases
+        env:
+          PYTHONPATH: src
+        run: python experiments/collector/run.py --collector "$RUNNER_TEMP/otelcol" --output collector-results
+      - name: Run observer controls and stress cases
+        env:
+          PYTHONPATH: src
+        run: |
+          python -m unittest discover -s experiments/collector -p 'test_*.py'
+          python experiments/collector/stress.py --collector "$RUNNER_TEMP/otelcol" --output collector-stress-results
+      - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.0
+        if: always()
+        with:
+          name: collector-experiment-evidence
+          path: |
+            collector-results
+            collector-stress-results

experiments/collector/README.md

@@ -0,0 +1,150 @@
+# Collector fault harness
+
+This experimental harness wires the real Python and TypeScript telemetry SDKs
+through a real OpenTelemetry Collector to a synthetic loopback receiver backed by
+SQLite. It is not production delivery evidence and is not a release gate yet.
+
+## Reproduction
+
+Use Linux amd64, Python 3.12 and Node >=22.15. From the repository root:
+
+```sh
+uv venv --python 3.12 ../collector-env
+uv pip install --python ../collector-env/bin/python -r experiments/collector/environment.txt
+(cd packages/typescript && npm ci --ignore-scripts && npm run build)
+(cd experiments/collector && npm ci --ignore-scripts)
+PYTHONPATH=src ../collector-env/bin/python experiments/collector/run.py --collector /absolute/path/to/otelcol --output ../collector-results-new
+```
+
+Use the official core collector v0.163.2 Linux amd64 release:
+https://github.com/open-telemetry/opentelemetry-collector-releases/releases/tag/v0.163.2
+
+Archive `otelcol_0.163.2_linux_amd64.tar.gz` SHA256:
+`9fbf3101331c88aab4b7f4cfd7c788b252e91e193793e39bd2154971c87f88a7`.
+The runner enforces binary SHA256
+`d4fa834fae73cb37e6058955a3833894d1ec815a1fc3529b0c2aaa99d8b22a2e`.
+Use a fresh output directory. Clear HTTP(S)/ALL proxy environment variables for
+the Python emitter if your environment exports them; child processes clear these.
+
+## Assertions validated on an Ubuntu CI host
+
+For both SDKs: healthy delivery, first-response 503, commit then lost ACK, partial
+rejection, duplicate event ID, and collector idle restart. SQLite stores local
+evidence, observed requests, and committed rows separately. The synthetic backend
+uses a unique constraint for idempotency; the test must not be interpreted as a
+collector exactly-once guarantee. Retry observations cover the whole pipeline,
+not a unique attribution to collector retry versus exporter retry. The restart
+case restarts an idle collector before emission; it does not test in-flight recovery.
+In these twelve baseline cases queues are disabled and assertions compare row counts.
+The eight stress cases below add queue pressure, reordering and full projection checks.
+
+## Validation and local host limitation
+
+Python compilation, JavaScript syntax, dependency installation, and the repository
+TypeScript build succeeded. The downloaded archive matched the release digest and
+`otelcol components` succeeded. The host then failed before receiver startup:
+
+```
+Error: failed to register process metrics: process does not exist
+```
+
+Setting `service.telemetry.metrics.level: none` did not bypass the error. Upstream
+collector v0.163.0 `service/service.go` calls `registerProcessMetrics` on Linux
+regardless of that setting; `service/internal/proctelemetry/process_telemetry.go`
+initializes gopsutil with `os.Getpid()`. No collector delivery case passed on the local sandbox host. The supported Ubuntu
+CI run below completed all twelve cases after correcting the SDK v3 processor
+constructor. No production endpoint or separately provisioned cloud resource was used.
+
+## Recorded supported-host run
+
+[Run 39665288508](https://github.com/agentrust-io/agentrust-telemetry/actions/runs/39665288508),
+head `42fa91f0194741a1a1c7102d08142e7925fc86cb`, Ubuntu 24.04,
+Python 3.12.15 and Node 22.23.3: all twelve cases passed. For both languages:
+
+| Case | Local evidence rows | Observed attempts | Backend commits |
+| --- | ---: | ---: | ---: |
+| Healthy | 1 | 1 | 1 |
+| First 503 | 1 | 2 | 1 |
+| Lost acknowledgement | 1 | 2 | 1 |
+| Partial rejection | 1 | 1 | 0 |
+| Duplicate event | 2 | 2 | 1 |
+| Idle restart | 1 | 1 | 1 |
+
+The run retains configs, collector/emitter logs, synthetic inputs, SQLite observer
+and per-case summary in the `collector-experiment-evidence` artifact. These
+measurements establish the listed synthetic pipeline behaviors only.
+
+## Added stress and fidelity checks (new head)
+
+`stress.py` adds four cases for each SDK: an eight-event healthy queue control,
+an eight-event blocked-consumer/one-slot queue saturation case, a collector kill
+between backend commit and acknowledgement followed by a fresh post-restart event,
+and three events emitted with descending source timestamps. The sending queue is
+in memory; this does not claim persistent queue recovery. The receiver retains
+complete normalized bodies, event names and nanosecond timestamps in SQLite.
+Every observed body is compared to its submitted event, and every non-saturation
+case must account for all committed event IDs. Saturation requires both positive
+commits and independently observed loss relative to local evidence.
+
+The TypeScript experiment emitter now passes the normalized event timestamp to
+the OTel SDK as an integer seconds/nanoseconds pair. This is experiment wiring,
+not a change to the reference SDK or a claim about all exporters/backends.
+
+Six local observer tests catch changed nested fields, numeric type changes,
+wrong event names/timestamps, missing observations, dropped local evidence,
+a non-descending input control and a vacuous saturation result. The pinned collector validates
+the generated configuration. Local collector startup still fails at process
+metrics registration before any delivery case. The supported-host run below
+validates the added cases; the twelve-case historical run above remains evidence
+only for its original head.
+
+## Recorded stress run
+
+[Run 40204045774](https://github.com/agentrust-io/agentrust-telemetry/actions/runs/40204045774),
+code head `02d5c64572ee8341f3a4d54c10e8a400a51d76f8`, Ubuntu,
+Python 3.12.15: all twelve baseline cases, six observer unit tests, and all eight
+stress cases passed. Each language produced the same counts:
+
+| Case | Local evidence | Observed attempts | Backend commits |
+| --- | ---: | ---: | ---: |
+| Healthy queue | 8 | 8 | 8 |
+| Saturation | 8 | 1 | 1 |
+| In-flight restart | 2 | 2 | 2 |
+| Reversed timestamp input | 3 | 3 | 3 |
+
+Every observed stress record retained its complete normalized body, event name
+and nanosecond timestamp. Saturation waits for the seeded request to reach the
+blocked receiver before submitting pressure. Collector 0.163.2 counts that
+in-flight request against this one-slot queue capacity, so the seeded event alone
+committed; the runner also requires the Collector's explicit queue-full log.
+These are observed counts for this run, not universal retry/delivery guarantees.
+
+The first two stress workflow attempts exposed harness assumptions about consumer
+scheduling and queue capacity. Both were corrected before the successful run.
+The artifact includes full inputs, configs, Collector/emitter logs, SQLite
+observations and both summary files. No runtime delivery policy was changed.
+
+## Transport-order control correction
+
+The documentation confirmation run 40204226853 observed all three TypeScript
+reversed-input events with intact bodies/timestamps, but their concurrent HTTP
+exports arrived in reverse order of emission. The original arrival-order
+assertion was too strong and failed. The corrected harness verifies descending
+input timestamps, the exact input event-ID sequence in local evidence, complete
+backend delivery and per-event projection fidelity. It records submitted and
+observed event-ID sequences separately, allowing transport reordering and
+retries. It does not claim an SDK or Collector arrival-order guarantee. The
+recorded twenty-case run above predates this observer correction.
+
+[Run 40204492721](https://github.com/agentrust-io/agentrust-telemetry/actions/runs/40204492721)
+confirmed the corrected observer contract on code head
+`3cf135ad1698d9157e99ff1396e8c1665bb0b4a3`: all twelve baseline cases,
+six observer unit tests and all eight stress cases passed on Ubuntu. This
+records evidence for that code head; this documentation update adds no new
+execution result. In the baseline `lost_ack` case, the receiver commits before
+dropping the acknowledgement, then a retry of the same event identity yields
+one backend commit across multiple attempts. That is synthetic backend
+deduplication, not an exactly-once Collector guarantee. The stress
+`inflight_restart` case commits before acknowledgement, kills the Collector,
+and emits a different event after restart; it does not replay the unacknowledged
+identity or establish persistent queue recovery.

experiments/collector/emit.mjs

@@ -0,0 +1,24 @@
+import fs from 'node:fs';
+import {LoggerProvider, SimpleLogRecordProcessor} from '@opentelemetry/sdk-logs';
+import {OTLPLogExporter} from '@opentelemetry/exporter-logs-otlp-http';
+import {SchemaValidator, TelemetryClient} from '../../packages/typescript/dist/index.js';
+const [endpoint, input, evidence, timeout = '4000'] = process.argv.slice(2);
+const provider = new LoggerProvider({processors: [new SimpleLogRecordProcessor({exporter: new OTLPLogExporter({url: endpoint, timeoutMillis: Number(timeout)})})]});
+const logger = provider.getLogger('agentrust-collector-acceptance');
+const client = new TelemetryClient(SchemaValidator.bundled(), {
+  evidenceSink: {append(event) {
+    const fd = fs.openSync(evidence, 'a');
+    try {fs.writeSync(fd, JSON.stringify(event) + '\n'); fs.fsyncSync(fd);} finally {fs.closeSync(fd);}
+  }},
+  logEmitter: {emit(record) {
+    const ns = BigInt(record.timestamp_ns);
+    logger.emit({body: record.body, eventName: record.event_name,
+      timestamp: [Number(ns / 1_000_000_000n), Number(ns % 1_000_000_000n)]});
+  }},
+});
+for (const event of JSON.parse(fs.readFileSync(input, 'utf8'))) {
+  const result = client.emit(event);
+  if (!result.accepted || !result.evidencePersisted || !result.logEmitted) throw new Error('Local acceptance failed');
+  await provider.forceFlush();
+}
+await provider.shutdown();

experiments/collector/environment.txt

@@ -0,0 +1,46 @@
+agentrust-trace==0.10.0
+annotated-types==0.8.0
+arrow==1.4.0
+attrs==26.1.0
+build==1.6.1
+certifi==2026.10.19
+cffi==2.1.1
+charset-normalizer==3.5.2
+cryptography==50.0.1
+fqdn==1.5.1
+googleapis-common-protos==1.75.5
+idna==3.19
+isoduration==20.11.0
+jsonpointer==3.1.1
+jsonschema==4.26.0
+jsonschema-specifications==2025.9.1
+opentelemetry-api==1.45.1
+opentelemetry-exporter-http-transport==0.66b1
+opentelemetry-exporter-otlp-common==0.66b1
+opentelemetry-exporter-otlp-proto-common==1.45.1
+opentelemetry-exporter-otlp-proto-http==1.45.1
+opentelemetry-proto==1.45.1
+opentelemetry-sdk==1.45.1
+opentelemetry-semantic-conventions==0.66b1
+packaging==26.3
+protobuf==7.36.2
+pycparser==3.0
+pydantic==2.13.5
+pydantic-core==2.46.5
+pyproject-hooks==1.2.0
+python-dateutil==2.9.0.post0
+pyyaml==6.0.3
+referencing==0.37.0
+requests==2.34.2
+rfc3339-validator==0.1.4
+rfc3986-validator==0.1.1
+rfc3987==1.3.8
+rfc8785==0.1.4
+rpds-py==2026.6.3
+six==1.17.0
+typing-extensions==4.16.0
+typing-inspection==0.4.4
+tzdata==2026.3
+uri-template==1.3.0
+urllib3==2.8.0
+webcolors==25.10.0

experiments/collector/fidelity.py

@@ -0,0 +1,49 @@
+"""Pure observer assertions; local acceptance, attempts and commits stay distinct."""
+import json
+
+
+def any_value(value):
+    kind = value.WhichOneof("value")
+    if kind == "kvlist_value":
+        return {item.key: any_value(item.value) for item in value.kvlist_value.values}
+    if kind == "array_value":
+        return [any_value(item) for item in value.array_value.values]
+    if kind in {"string_value", "bool_value", "int_value", "double_value"}:
+        return getattr(value, kind)
+    raise ValueError(f"unsupported normalized-event value {kind}")
+
+
+def check_projection(events, observations):
+    expected = {event["event_id"]: event for event in events}
+    if not observations:
+        raise AssertionError("no independent projection observations")
+    for observation in observations:
+        event = observation["body"]
+        original = expected[event["event_id"]]
+        if json.dumps(event, sort_keys=True, allow_nan=False) != json.dumps(original, sort_keys=True, allow_nan=False):
+            raise AssertionError("normalized event body changed")
+        if observation["event_name"] != original["event_type"]:
+            raise AssertionError("event name changed")
+        if observation["timestamp_ns"] != original["time_unix_nano"]:
+            raise AssertionError("event timestamp changed")
+
+
+def check_delivery(mode, events, evidence_count, attempts, commits):
+    ids = [event["event_id"] for event in events]
+    if evidence_count != len(events):
+        raise AssertionError("local evidence is missing")
+    if not set(commits).issubset(ids) or not set(attempts).issubset(ids):
+        raise AssertionError("observer saw an unsubmitted event")
+    if mode == "queue_saturation":
+        if not 0 < len(set(commits)) < len(events):
+            raise AssertionError("expected bounded queue loss was not independently observed")
+    elif set(commits) != set(ids):
+        raise AssertionError("a submitted event did not commit")
+    if mode == "reorder":
+        times = [int(event["time_unix_nano"]) for event in events]
+        if len(times) < 2 or not all(left > right for left, right in zip(times, times[1:])):
+            raise AssertionError("reorder control did not submit descending timestamps")
+        if set(attempts) != set(ids):
+            raise AssertionError("a reordered input was not independently observed")
+        # Concurrent SDK exports/HTTP requests need not preserve arrival order.
+        # check_projection binds every observed timestamp and body to its event ID.