diff --git a/README.md b/README.md index 7b728514..1e42c727 100644 --- a/README.md +++ b/README.md @@ -290,11 +290,17 @@ harness measures the wire, not the language behind it. ### Harden it for production Two gaps are named and open, not vague. The Guardian writes its envelope and audit logs -as synchronous file appends on the decision path, with no batching, so trace spans need -a batched writer before a Trace-pillar claim would hold up under load. No OpenTelemetry -collector or exporter exists yet. The conformance harness only measures which -attributes a consumer *could* emit from the envelopes, not that anything ships them. -Start in [`reference-implementations/agt/packages/`](reference-implementations/agt/packages/). +as synchronous file appends on the decision path, with no batching. Trace export is +partial: set `ACS_OTEL_FILE` or `ACS_OTEL_ENDPOINT` and the Guardian exports OTLP spans +for the two live methods, `steps/toolCallRequest` and `steps/toolCallResult`, as the +[Guardian README](reference-implementations/agt/packages/guardian/README.md#export-traces) +describes. + +That exporter does not make the reference ACS-Trace conformant, and this tree claims no +Trace profile. It covers two methods, emits no OCSF, and leaves out a required attribute +when the request omits it. The conformance harness still measures which attributes a +consumer *could* emit from the wire, not what the exporter ships. Start in +[`reference-implementations/agt/packages/`](reference-implementations/agt/packages/). ### Go deeper diff --git a/reference-implementations/agt/bun.lock b/reference-implementations/agt/bun.lock index cc2dc5a0..5058e1e1 100644 --- a/reference-implementations/agt/bun.lock +++ b/reference-implementations/agt/bun.lock @@ -57,6 +57,12 @@ "name": "guardian", "version": "0.1.0", "dependencies": { + "@opentelemetry/api": "1.9.1", + "@opentelemetry/core": "2.12.0", + "@opentelemetry/exporter-trace-otlp-http": "0.223.0", + "@opentelemetry/otlp-transformer": "0.223.0", + "@opentelemetry/resources": "2.12.0", + "@opentelemetry/sdk-trace-base": "2.12.0", "agt-bridge": "workspace:*", "ajv": "8.20.0", "ajv-formats": "3.0.1", @@ -93,6 +99,30 @@ "@opencode-ai/sdk": ["@opencode-ai/sdk@1.18.15", "", { "dependencies": { "cross-spawn": "7.0.6" } }, "sha512-8sfo9nGiVwesAZW9Wqkvynyn7w4wYaHx1O9qOHpYL65+Bs2XUpHP3kBbZe58gjQydFgk6I74kUF7sqCiPu2arQ=="], + "@opentelemetry/api": ["@opentelemetry/api@1.9.1", "", {}, "sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q=="], + + "@opentelemetry/api-logs": ["@opentelemetry/api-logs@0.223.0", "", { "dependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-Oepm6etVPFCn0VWB1t833iOTgz1eYBDUPxcrOR/XvwzD1+Ag9VzSPSrrKAFNJ/haatj+8CXbG47pZXvdNQKDJw=="], + + "@opentelemetry/core": ["@opentelemetry/core@2.12.0", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-1HsSAuvT4/my0QrXWsyXFdWjKaPeCHsismKZgEEIb8NK13LP6DrMRlsUbvD/0F9mDWK3xDom9AMv3IscxgOnOQ=="], + + "@opentelemetry/exporter-trace-otlp-http": ["@opentelemetry/exporter-trace-otlp-http@0.223.0", "", { "dependencies": { "@opentelemetry/otlp-exporter-base": "0.223.0", "@opentelemetry/otlp-transformer": "0.223.0", "@opentelemetry/sdk-trace": "2.12.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-N8NWZaoQuF9e307S6xoVoI1lvxOOFkmBnynD42r02CPCfxoC95cJExAKBFrTD4C5iuhqthbzOZJN/W29ez4o/A=="], + + "@opentelemetry/otlp-exporter-base": ["@opentelemetry/otlp-exporter-base@0.223.0", "", { "dependencies": { "@opentelemetry/core": "2.12.0", "@opentelemetry/otlp-transformer": "0.223.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-HlJMrJXjOqdQeoYNDHEvFTpaY2p9uWDK1GPl5AZ4wI5easzfoppTduOv47M8eiF+4ew42vfI/Xvt7THYgESnRw=="], + + "@opentelemetry/otlp-transformer": ["@opentelemetry/otlp-transformer@0.223.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.223.0", "@opentelemetry/core": "2.12.0", "@opentelemetry/resources": "2.12.0", "@opentelemetry/sdk-logs": "0.223.0", "@opentelemetry/sdk-metrics": "2.12.0", "@opentelemetry/sdk-trace": "2.12.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-oxmmPL5/MbSsO8K588V9n/XLIe7FZTK30Fo+/U/lAnt0ZIK/hoPM03n2WpSTZVbUBGTUE7aLyxnAzh7nE2PD4Q=="], + + "@opentelemetry/resources": ["@opentelemetry/resources@2.12.0", "", { "dependencies": { "@opentelemetry/core": "2.12.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-+rSRoOdln6NhXTVcBdHSlZxJDNzwxadeweK2ltZO3tVtfp6aklABc06xclDk6ovRR+EiGgIcSbh/6NgjFhnb3A=="], + + "@opentelemetry/sdk-logs": ["@opentelemetry/sdk-logs@0.223.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.223.0", "@opentelemetry/core": "2.12.0", "@opentelemetry/resources": "2.12.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.4.0 <1.10.0" } }, "sha512-ApiAfUiUh6fmJt1GbsBuoeDn8D8IL42UvGyPUr8o1Z89ZeIiOCE1rZ78KkIUjEX3OV8Oorl9ez5Qjo43ZynDyw=="], + + "@opentelemetry/sdk-metrics": ["@opentelemetry/sdk-metrics@2.12.0", "", { "dependencies": { "@opentelemetry/core": "2.12.0", "@opentelemetry/resources": "2.12.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.9.0 <1.10.0" } }, "sha512-mVMaEHwCUJuY3S4Esn+nT0yQA3qCEGh+SOPJsXygU0Gl453WcsDNjtEa8EfdewMwoLyr1tlofB4fXytVpiXuow=="], + + "@opentelemetry/sdk-trace": ["@opentelemetry/sdk-trace@2.12.0", "", { "dependencies": { "@opentelemetry/core": "2.12.0", "@opentelemetry/resources": "2.12.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-ImVuRQ6faOsTi1Wna88s2avgrrf/iDgXPigIeynkUhjstMuoQUDOJXgkez2j1mlU4btjXcXBGa+q7qKFYR3GsA=="], + + "@opentelemetry/sdk-trace-base": ["@opentelemetry/sdk-trace-base@2.12.0", "", { "dependencies": { "@opentelemetry/core": "2.12.0", "@opentelemetry/resources": "2.12.0", "@opentelemetry/sdk-trace": "2.12.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-JOERgZCxRM06FEJ5kEWxlSRChah6MmXDx3KwCruqEv/iEyLkTokCSemf2y1w8oIHtfrr8EVTpzebq/31vn2Cpg=="], + + "@opentelemetry/semantic-conventions": ["@opentelemetry/semantic-conventions@1.43.0", "", {}, "sha512-eSYWTm620tTk45EKSedaUL8MFYI8hW164hIXsgIHyxu3VobUB3fFCu5t0hQby6OoWRPsG1KkKUG2M5UadiLiVg=="], + "@standard-schema/spec": ["@standard-schema/spec@1.1.0", "", {}, "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w=="], "@types/node": ["@types/node@26.2.0", "", { "dependencies": { "undici-types": "~8.3.0" } }, "sha512-5IviulTZeRNp2vAJ514cc/HUlY5nZ9fCbq9DMyC52BrhFZACo3nI0R7qBxhQmo/d27NFe96ur/b7Wwxklda+kg=="], diff --git a/reference-implementations/agt/otel-collector.yaml b/reference-implementations/agt/otel-collector.yaml new file mode 100644 index 00000000..533bbb24 --- /dev/null +++ b/reference-implementations/agt/otel-collector.yaml @@ -0,0 +1,31 @@ +# An OpenTelemetry Collector that receives the Guardian's spans and writes +# them as OTLP/JSON, one export request per line -- the same shape +# ACS_OTEL_FILE writes directly. Run it with the contrib distribution, which +# is where the `file` exporter lives: +# otelcol-contrib --config otel-collector.yaml +# then start the Guardian with ACS_OTEL_ENDPOINT=http://127.0.0.1:4318/v1/traces +# Tested with otelcol-contrib 0.162.0. In a container, run it as your own +# user (--user "$(id -u):$(id -g)") so it can write into .acs/. +receivers: + otlp: + protocols: + http: + endpoint: 127.0.0.1:4318 + +exporters: + file: + path: .acs/otel-collector-traces.jsonl + # A fresh checkout has no .acs/ yet, and without this the exporter + # refuses to start. + create_directory: true + # basic, not detailed: detailed prints every attribute, reasoning text + # and provenance source ids included, to a stdout that often ends up in a + # log system nobody planned for. + debug: + verbosity: basic + +service: + pipelines: + traces: + receivers: [otlp] + exporters: [file, debug] diff --git a/reference-implementations/agt/packages/guardian/README.md b/reference-implementations/agt/packages/guardian/README.md index 511f979b..a46a56f3 100644 --- a/reference-implementations/agt/packages/guardian/README.md +++ b/reference-implementations/agt/packages/guardian/README.md @@ -29,6 +29,8 @@ It constructs the AGT runtime once, at startup, against `policy/manifest.yaml`, | `ACS_MANIFEST_PATH` | `policy/manifest.yaml` | The AGT manifest. `policy/manifest.drift.yaml` is the second manifest, used to reach the `warn` verdict | | `ACS_ENVELOPE_LOG` | `.acs/envelopes.jsonl` | Where every envelope is recorded | | `ACS_SESSION_CONTEXT_LOG` | `.acs/session-context.jsonl` | Where each session's hash chain is written | +| `ACS_OTEL_FILE` | unset | Writes one OpenTelemetry span per decided step as OTLP/JSON, one export request per line. Unset means no trace export | +| `ACS_OTEL_ENDPOINT` | unset | Sends the same spans to an OTLP/HTTP collector, for example `http://127.0.0.1:4318/v1/traces` | To see a decision without an agent client, pipe a hook payload into the Claude Code shim while the Guardian runs. The [tree README](../../README.md#drive-one-hook-by-hand) shows the command and its output. @@ -46,6 +48,53 @@ To see a decision without an agent client, pipe a hook payload into the Claude C Every throw on the request path is caught. Nothing escapes the fetch handler, because an unhandled rejection would produce an HTML error page, the host's client would fail to parse it, and Claude Code would read the failed hook as "never fired" and proceed ungoverned. +## Export traces + +Set `ACS_OTEL_FILE`, `ACS_OTEL_ENDPOINT`, or both. With neither set, no exporter exists and the Guardian runs exactly as before. This is an exporter for the two live methods, not an ACS-Trace conformance claim. + +```bash +ACS_OTEL_FILE=.acs/otel-traces.jsonl bun run guardian +``` + +Each decided step becomes one span, named by [`trace/otel-mapping.json`](../../../../specification/v0.1.0/trace/otel-mapping.json): `gen_ai.tool.call` for `steps/toolCallRequest` and `gen_ai.tool.result` for `steps/toolCallResult`. The span opens when the request is recorded and closes when the response is. It carries each attribute the mapping lists for it when the envelope supplies a value. An attribute the envelope omits, such as `acs.capability`, is absent from the span. + +Every span also carries `acs.session.id`, and `acs.request_id` whenever the envelope has one, so a session can be rebuilt from the trace alone. When an argument or output carries a Provenance object, the span carries `acs.provenance.origin` as an array, one element per Provenance object in wire order. `acs.provenance.source_id` is the matching array, with `""` where an object has no `source_id`, and it is left out when no object has one. Both stop at 64 objects. `acs.provenance.lineage_depth` is not emitted: `provenance.json` defines no such field. + +Every string attribute is cut at 4096 characters, a long `acs.reasoning` included, so one large request cannot grow the export queue without bound. The sampler and span limits are set in code, so an `OTEL_TRACES_SAMPLER` or `OTEL_SPAN_*` variable in the environment does not change what is exported. + +The decision rides on the span as an `acs.decision` event. The event always carries `acs.decision`, carries `acs.reasoning` when the response has one, and carries `acs.reason_codes`, `acs.policy.ids`, `acs.policy.versions`, `acs.policy.rule_ids`, `acs.cited_provenance_ids` and `acs.chain_hash` only when the response carries them. `acs.evaluator` appears when the Guardian's response carries `metadata.evaluator`. This Guardian sends it once #229 is merged. + +This Guardian does not send `chain_hash`, `cited_provenance_ids`, or `policy_version` on its responses yet, so those attributes are absent from its spans today. The three policy arrays always have the same length as `policy_references`, with `""` where a reference has no value. `handshake/hello`, methods this Guardian does not dispatch, and unparseable bodies produce no span. + +A step that got no decision is recorded as an `acs.decision.failure` event on the same span, with `acs.failure.class`, `acs.failure.posture` and `acs.failure.reported_by`, plus the error code, the timeout, the elapsed time and the request size when known. The Guardian emits it for an error it answered instead of a decision, for a decision it finished after the negotiated timeout, and for a step still open when it shut down. It carries the error code but never the error text, because that text can quote the request. A replay rejection (`-32005`) gets no event, because its pair belongs to the earlier request. The converter below produces the same event, reported by the observed agent, from the host adapter's audit log. + +The file holds one OTLP `ExportTraceServiceRequest` per line, the shape the Collector's file exporter writes, with `service.name` set to `acs-reference-guardian`. Spans are batched and flushed when the Guardian closes on SIGINT or SIGTERM. A SIGKILL loses the last batch, and so does an output that has not finished three seconds into shutdown: the Guardian reports it on stderr and exits instead of waiting. [`otel-collector.yaml`](../../otel-collector.yaml) at the root of this tree receives the endpoint output and writes the same file shape. + +Export never touches a decision. Each output fails on its own: the first failed write or send disables that output for the rest of the process and prints one line on stderr, the other output keeps running, and the response goes out unchanged. + +Use synthetic sessions for lab runs. The envelope log holds raw envelopes before any redaction, a span's `acs.reasoning` repeats the Guardian's explanation of them, and `acs.provenance.source_id` goes out as the host sent it: by `provenance.json` it can be a URL or a file path, so a signed URL or a home-directory path would too ([#71](https://github.com/GenAI-Security-Project/agent-control-standard/issues/71)). + +### Convert the host audit log + +The failures only the host sees, a Guardian it could not reach or one that answered too late, are in the host adapter's audit log, not in the Guardian's file. `audit-to-otlp.ts` turns that log into the same OTLP/JSON shape: one span per posture entry, with an `acs.decision.failure` event whose `acs.failure.reported_by` is `observed_agent`. Each span carries the same `acs.session.id` and `acs.request_id` as the Guardian's span for that request, so the two files join. It does not carry `gen_ai.tool.name`, because the audit log does not record the tool. + +```bash +bun run packages/guardian/src/audit-to-otlp.ts .acs/audit.jsonl .acs/otel-audit.jsonl +``` + +Run it after the Guardian has stopped. Two writers on one file can interleave lines, so give the converter a file of its own or wait. Running it twice over the same log writes every span twice. + +`acs.failure.posture` is the deployment's posture, not what became of the step. A refused request shows `fail_open` under a `proceed` deployment even though the host blocked it. The detail sentence says so, and the error code is in `acs.failure.error_code`. The detail is always a sentence the converter writes, never the host's message, which quotes the Guardian's error text. + +Entries with `outcome: "ungoverned"` are skipped: those are steps no gate asked about, which the failure classes do not cover. Also skipped, and counted on stdout: + +- entries whose method has no `step_to_span` entry; +- entries with no method, by failure kind, because the host never built a request for them; +- replay rejections; +- lines it cannot read. + +It prints how many spans reached the file and exits non-zero when that is fewer than it built. + ## Use it in-process The tests and the conformance harness start a Guardian in the same process. @@ -58,7 +107,7 @@ const guardian = await startGuardian({ port: 0, manifestPath: "policy/manifest.y await guardian.close(); ``` -`StartGuardianOptions` also accepts `hostname`, `envelopeLogPath`, `sessionContextLog` and `onDecisionFailure`, which mirror the environment variables above, and four options for tests: `mappingPath`, `annotator`, `bridge` and `sessionContextStore`. Each is documented where it is declared, in `src/server.ts`. +`StartGuardianOptions` also accepts `hostname`, `envelopeLogPath`, `sessionContextLog`, `otelTraceFile`, `otelTraceEndpoint` and `onDecisionFailure`, which mirror the environment variables above, and five options for tests: `mappingPath`, `annotator`, `bridge`, `sessionContextStore` and `traceExporter`. Each is documented where it is declared, in `src/server.ts`. The package has two entry points. `guardian` exports the governance verbs listed in `src/index.ts`. `guardian/deployment` exports `createDeploymentBridge`, the one function that builds the deployment's bridge with the egress annotator wired in. The conformance harness imports the second so it measures the bridge the Guardian ships, not a replica. @@ -78,6 +127,8 @@ The package has two entry points. `guardian` exports the governance verbs listed | `src/deny-on-invalid-envelope.ts` | Turns a schema failure on a `steps/*` method into an honoured `deny` | | `src/session-context-store.ts`, `src/session-context.ts`, `src/ifc-labels.ts` | The per-session hash chain, provenance, and information-flow labels | | `src/envelope-log-sink.ts` | The envelope log writer | +| `src/otel-trace-exporter.ts` | The OpenTelemetry span exporter and its OTLP/JSON file writer | +| `src/audit-to-otlp.ts` | Converts the host adapter's audit log into the same OTLP/JSON file | | `src/acs-result.ts` | The decision result type | ## The wire is not secured @@ -90,4 +141,4 @@ The endpoint has no authentication, no origin check and no request signing. It b bun test packages/guardian ``` -Twelve files. Most start a real Guardian on port 0 and drive real envelopes through the pinned policy bundle. One test copies `src/` one directory deeper into a fixed scratch directory, so the relative schema path resolves to nothing. It proves that a missing schema directory becomes a recorded JSON-RPC error and never an HTML page. The scratch directories are gitignored and removed in a `finally`. +Thirteen files. Most start a real Guardian on port 0 and drive real envelopes through the pinned policy bundle. One test copies `src/` one directory deeper into a fixed scratch directory, so the relative schema path resolves to nothing. It proves that a missing schema directory becomes a recorded JSON-RPC error and never an HTML page. The scratch directories are gitignored and removed in a `finally`. diff --git a/reference-implementations/agt/packages/guardian/package.json b/reference-implementations/agt/packages/guardian/package.json index f5cd92dd..49e89c88 100644 --- a/reference-implementations/agt/packages/guardian/package.json +++ b/reference-implementations/agt/packages/guardian/package.json @@ -9,6 +9,12 @@ "./package.json": "./package.json" }, "dependencies": { + "@opentelemetry/api": "1.9.1", + "@opentelemetry/core": "2.12.0", + "@opentelemetry/exporter-trace-otlp-http": "0.223.0", + "@opentelemetry/otlp-transformer": "0.223.0", + "@opentelemetry/resources": "2.12.0", + "@opentelemetry/sdk-trace-base": "2.12.0", "agt-bridge": "workspace:*", "ajv": "8.20.0", "ajv-formats": "3.0.1" diff --git a/reference-implementations/agt/packages/guardian/src/audit-to-otlp.ts b/reference-implementations/agt/packages/guardian/src/audit-to-otlp.ts new file mode 100644 index 00000000..90ded720 --- /dev/null +++ b/reference-implementations/agt/packages/guardian/src/audit-to-otlp.ts @@ -0,0 +1,322 @@ +/** + * audit-to-otlp turns the host adapter's audit log into the same OTLP/JSON + * file the Guardian's trace exporter writes, so a reader of that file sees + * every decision failure, not only the ones the Guardian could see itself. + * + * The Guardian's exporter records the failures it can see: an error it + * answered, a decision it finished late, a step still open at shutdown. The + * host adapter sees the rest -- a Guardian it could not reach, one that + * never answered in time -- and #37 makes it record each one in its own + * audit log (`.acs/audit.jsonl`, packages/host-adapter/src/audit-sink.ts) + * and surface it on Trace when it can. The adapter runs once per hook in a + * fresh process and should not carry the OpenTelemetry SDK, so the surfacing + * happens here, afterwards, from the log. + * + * bun run packages/guardian/src/audit-to-otlp.ts + * + * Each posture-resolved entry (`outcome` of `proceeded` or `blocked`) + * becomes one span named by otel-mapping.json for the entry's method, timed + * at `recorded_at`, carrying the same `acs.session.id` and `acs.request_id` + * the Guardian's own spans carry, and one `acs.decision.failure` event with + * `acs.failure.reported_by` set to `observed_agent`. The audit entry carries + * no elapsed time, timeout or payload size, so those "when known" attributes + * are left out. + * + * Skipped, and counted in the summary: + * - an `ungoverned` entry: a step no gate asked about, not a decision failure; + * - an entry with no method: the host never built a request (a broken + * hookmap, say), so there is no span name and no request id to carry -- + * counted by its failure kind, because a fail-open among these is still a + * bypass someone has to see; + * - an entry whose method the mapping names no span for; + * - a REPLAY_DETECTED error, whose pair belongs to the earlier request (#228); + * - a line that is not JSON, not an audit entry, or has no posture. + * + * Reads the log with plain JSON.parse and loose field reads, importing + * nothing from the host adapter: the file is the contract, the same way the + * Inspector reads it. Run it after the Guardian has stopped. Two writers on + * one file can interleave lines, and running it twice over the same log + * writes every span twice. + */ +import { createHash } from "node:crypto"; +import { readFileSync } from "node:fs"; +import { SpanStatusCode, type Attributes } from "@opentelemetry/api"; +import { ExportResultCode, type ExportResult } from "@opentelemetry/core"; +import { resourceFromAttributes } from "@opentelemetry/resources"; +import { + AlwaysOnSampler, + BasicTracerProvider, + SimpleSpanProcessor, + type ReadableSpan, + type SpanExporter, +} from "@opentelemetry/sdk-trace-base"; +import { + createFileSpanExporter, + DECISION_FAILURE_EVENT, + decisionFailureAttributes, + GUARDIAN_ERROR_STATUS, + loadOtelMapping, + OTEL_SERVICE_NAME, + type FailureClass, +} from "./otel-trace-exporter.ts"; + +/** + * The host's `failure.kind` (packages/host-adapter/src/failure-kinds.ts) + * to #37's `acs.failure.class`. + * + * `transport` is a request that never reached the Guardian, which is what + * `unreachable` names. `error_without_decision` and `refused` both mean an + * error came back instead of a decision (§6.4), so both are `guardian_error`; + * the code goes in `acs.failure.error_code`. `host_configuration` is the host + * failing before any request existed. `decision_unrenderable` is a decision + * that did arrive and that this host could not express, which none of the + * other classes describe, so it falls to `other` with its own detail. + * `unknown`, and any kind this table does not know, is `other`, and a rising + * `other` count is the signal the list needs a row. + */ +export const FAILURE_CLASS_BY_KIND: Readonly> = { + timeout: "timeout", + transport: "unreachable", + error_without_decision: "guardian_error", + refused: "guardian_error", + host_configuration: "observed_agent_error", +}; + +export function failureClassOf(kind: unknown): FailureClass { + return (typeof kind === "string" ? FAILURE_CLASS_BY_KIND[kind] : undefined) ?? "other"; +} + +/** + * `acs.failure.detail` for each host kind: a sentence this converter writes, + * never the host's `failure.message`. That message quotes the Guardian's + * error text, and #228 forbids copying bytes from a response or an error into + * the trace. The sentence is also the only place the host's kind survives, + * since the class folds some kinds together. A kind this table does not + * know gets the generic sentence; the kind string itself is never copied. + */ +export const FAILURE_DETAIL_BY_KIND: Readonly> = { + timeout: "No decision arrived before the host's timeout.", + transport: "The host could not reach the Guardian.", + error_without_decision: GUARDIAN_ERROR_STATUS, + refused: "The Guardian refused the request, and this host blocked the step.", + host_configuration: "The host could not build a request for this hook.", + decision_unrenderable: "A decision arrived, and this host could not express it in its hook output.", +}; +const GENERIC_DETAIL = "The host recorded a failure this converter does not classify."; + +/** The JSON-RPC code, parsed out of the two message shapes the host writes + * (packages/host-adapter/src/failure-posture.ts). Only the integer is + * taken; nothing else in the message is copied. */ +const ERROR_CODE_IN_MESSAGE = /^guardian (?:refused the envelope with|returned) error (-?\d+):/; +const REPLAY_DETECTED_CODE = -32005; + +/** + * The host adapter's `toSessionUuid` (packages/host-adapter/src/build-envelope.ts), + * copied because host-adapter already depends on this package, so importing + * it would be a cycle. The Guardian's spans carry this derived UUID, not the + * host's raw session id, so a converter span has to carry it too or the two + * files cannot be joined. The root test pins the two as equal. + */ +const SESSION_ID_NAMESPACE = "11c79ebd-8469-4b4e-8de0-72674add484c"; +const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +export function sessionUuid(rawSessionId: string): string { + if (UUID_RE.test(rawSessionId)) { + return rawSessionId.toLowerCase(); + } + const namespaceBytes = Buffer.from(SESSION_ID_NAMESPACE.replace(/-/g, ""), "hex"); + const hash = createHash("sha1") + .update(Buffer.concat([namespaceBytes, Buffer.from(rawSessionId, "utf8")])) + .digest(); + hash[6] = (hash[6]! & 0x0f) | 0x50; // version 5 + hash[8] = (hash[8]! & 0x3f) | 0x80; // RFC 4122 variant + const hex = hash.subarray(0, 16).toString("hex"); + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20, 32)}`; +} + +export type ConversionSummary = { + /** Spans built from the log. */ + converted: number; + /** Spans the file exporter confirmed written. Differs from `converted` + * only when an export failed, and then the command exits non-zero. */ + written: number; + /** Posture entries whose method the mapping names no span for. */ + skippedUnmapped: number; + /** Posture entries with no method, counted by failure kind. */ + skippedNoMethod: Record; + /** REPLAY_DETECTED errors. */ + skippedReplay: number; + /** Steps no gate asked about. */ + skippedUngoverned: number; + /** Lines that were not JSON, not an audit entry, or had no posture. */ + skippedUnreadable: number; +}; + +type JsonObject = Record; + +function readObject(value: unknown): JsonObject | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) ? (value as JsonObject) : undefined; +} + +function readString(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +/** `recorded_at` as a Date, or undefined for a value Date cannot read, in + * which case the span is timed at now rather than at a nonsense instant. */ +function readInstant(value: unknown): Date | undefined { + const text = readString(value); + if (text === undefined) return undefined; + const instant = new Date(text); + return Number.isNaN(instant.getTime()) ? undefined : instant; +} + +/** Counts the spans the wrapped exporter reports written, so the summary + * says what reached the file and not what was handed to the SDK. */ +function countingExporter(inner: SpanExporter): { exporter: SpanExporter; written: () => number } { + let written = 0; + return { + written: () => written, + exporter: { + export(spans: ReadableSpan[], done: (result: ExportResult) => void): void { + inner.export(spans, (result) => { + if (result.code === ExportResultCode.SUCCESS) { + written += spans.length; + } + done(result); + }); + }, + shutdown: () => inner.shutdown(), + forceFlush: () => inner.forceFlush?.() ?? Promise.resolve(), + }, + }; +} + +export async function convertAuditLog(auditPath: string, otelPath: string, mappingPath?: string): Promise { + const mapping = loadOtelMapping(mappingPath); + const output = countingExporter(createFileSpanExporter(otelPath)); + // One span per export, not batched: a batch processor's bounded queue + // overflowed inside this synchronous loop and dropped spans silently. + // Sampler set explicitly, so no OTEL_TRACES_SAMPLER can drop them either. + const provider = new BasicTracerProvider({ + resource: resourceFromAttributes({ "service.name": OTEL_SERVICE_NAME }), + sampler: new AlwaysOnSampler(), + spanProcessors: [new SimpleSpanProcessor(output.exporter)], + }); + const tracer = provider.getTracer(OTEL_SERVICE_NAME); + const summary: ConversionSummary = { + converted: 0, + written: 0, + skippedUnmapped: 0, + skippedNoMethod: {}, + skippedReplay: 0, + skippedUngoverned: 0, + skippedUnreadable: 0, + }; + + for (const line of readFileSync(auditPath, "utf8").split("\n")) { + if (line.trim() === "") continue; + let entry: JsonObject | undefined; + try { + entry = readObject(JSON.parse(line)); + } catch { + entry = undefined; + } + if (entry === undefined) { + summary.skippedUnreadable += 1; + continue; + } + const outcome = readString(entry.outcome); + if (outcome === "ungoverned") { + summary.skippedUngoverned += 1; + continue; + } + // acs.failure.posture is the deployment's posture (#228), which the host + // records on every posture entry. Inferring it from `outcome` would be + // wrong for a refusal, which is blocked whatever the posture said. + const posture = readString(entry.posture); + if ((outcome !== "proceeded" && outcome !== "blocked") || (posture !== "proceed" && posture !== "deny")) { + summary.skippedUnreadable += 1; + continue; + } + const failure = readObject(entry.failure) ?? {}; + const kind = readString(failure.kind); + const method = readString(entry.method); + if (method === undefined) { + const key = kind !== undefined && kind in FAILURE_DETAIL_BY_KIND ? kind : "other"; + summary.skippedNoMethod[key] = (summary.skippedNoMethod[key] ?? 0) + 1; + continue; + } + const spanName = mapping.stepToSpan[method]?.span_name; + if (spanName === undefined) { + summary.skippedUnmapped += 1; + continue; + } + const codeText = ERROR_CODE_IN_MESSAGE.exec(readString(failure.message) ?? "")?.[1]; + const errorCode = codeText === undefined ? undefined : Number(codeText); + if (errorCode === REPLAY_DETECTED_CODE) { + summary.skippedReplay += 1; + continue; + } + + const instant = readInstant(entry.recorded_at); + const detail = (kind !== undefined ? FAILURE_DETAIL_BY_KIND[kind] : undefined) ?? GENERIC_DETAIL; + const attributes: Attributes = {}; + const sessionId = readString(entry.session_id); + if (sessionId !== undefined) { + attributes["acs.session.id"] = sessionUuid(sessionId); + } + // The host builds every envelope with its JSON-RPC id equal to + // params.request_id (build-envelope.ts), so the entry's rpc_id is the + // request id the Guardian's span for the same request carries. + const requestId = readString(entry.rpc_id); + if (requestId !== undefined) { + attributes["acs.request_id"] = requestId; + } + // The entry is one instant, so the span is too: it opens and closes at + // recorded_at. The host, not the Guardian, is the reporter. + const span = tracer.startSpan(spanName, { attributes, root: true, startTime: instant }); + span.addEvent( + DECISION_FAILURE_EVENT, + decisionFailureAttributes({ + class: failureClassOf(kind), + posture: posture === "deny" ? "fail_closed" : "fail_open", + reportedBy: "observed_agent", + detail, + errorCode, + }), + instant, + ); + span.setStatus({ code: SpanStatusCode.ERROR, message: detail }); + span.end(instant); + summary.converted += 1; + } + + // Every export has settled once forceFlush resolves, so the count is final + // here; shutdown then releases the processor. + await provider.forceFlush(); + summary.written = output.written(); + await provider.shutdown(); + return summary; +} + +if (import.meta.main) { + const [auditPath, otelPath] = Bun.argv.slice(2); + if (auditPath === undefined || otelPath === undefined) { + console.error("usage: bun run packages/guardian/src/audit-to-otlp.ts "); + process.exit(2); + } + const summary = await convertAuditLog(auditPath, otelPath); + const noMethod = Object.entries(summary.skippedNoMethod) + .map(([kind, count]) => `${count} ${kind}`) + .join(", "); + console.log( + `wrote ${summary.written} of ${summary.converted} span(s) to ${otelPath}; skipped ${summary.skippedUnmapped} with no mapping entry, ` + + `${noMethod === "" ? 0 : noMethod} with no method, ${summary.skippedReplay} replay, ` + + `${summary.skippedUngoverned} ungoverned, ${summary.skippedUnreadable} unreadable`, + ); + if (summary.written !== summary.converted) { + console.error(`only ${summary.written} of ${summary.converted} span(s) reached ${otelPath}`); + process.exit(1); + } +} diff --git a/reference-implementations/agt/packages/guardian/src/main.ts b/reference-implementations/agt/packages/guardian/src/main.ts index a3dee199..08cea99a 100644 --- a/reference-implementations/agt/packages/guardian/src/main.ts +++ b/reference-implementations/agt/packages/guardian/src/main.ts @@ -31,6 +31,10 @@ const hostname = process.env.ACS_GUARDIAN_HOST; const manifestPath = process.env.ACS_MANIFEST_PATH ?? DEFAULT_MANIFEST_PATH; const envelopeLogPath = process.env.ACS_ENVELOPE_LOG ?? DEFAULT_ENVELOPE_LOG; const sessionContextLog = process.env.ACS_SESSION_CONTEXT_LOG ?? DEFAULT_SESSION_CONTEXT_LOG; +// Trace export is off unless one of these names an output. Read here and +// passed in, like ACS_ENVELOPE_LOG: server.ts reads no env itself. +const otelTraceFile = process.env.ACS_OTEL_FILE; +const otelTraceEndpoint = process.env.ACS_OTEL_ENDPOINT; // Read and validate the posture BEFORE starting the server. Not the audit // sink's path: that file is the host's, written by the hook, not by @@ -44,8 +48,33 @@ const sessionContextLog = process.env.ACS_SESSION_CONTEXT_LOG ?? DEFAULT_SESSION // healthy start immediately followed by one. const posture = buildServerHello().on_decision_failure; -const guardian = await startGuardian({ port, hostname, manifestPath, envelopeLogPath, sessionContextLog }); +const guardian = await startGuardian({ + port, + hostname, + manifestPath, + envelopeLogPath, + sessionContextLog, + otelTraceFile, + otelTraceEndpoint, +}); console.log(`Guardian listening at ${guardian.url}`); console.log(`Envelope log: ${envelopeLogPath}`); console.log(`Session context log: ${sessionContextLog}`); console.log(`Failure posture: ${posture} (override with ACS_ON_DECISION_FAILURE=deny)`); + +// Only when export is on. The four lines above are what README.md captures +// and test/readme-captures.test.ts holds them to the line, so a fifth line +// appears only for a Guardian the quickstart does not run. +const traceOutputs = [otelTraceFile, otelTraceEndpoint].filter((target): target is string => target !== undefined); +if (traceOutputs.length > 0) { + console.log(`OTel traces: ${traceOutputs.join(", ")}`); + // The last batch of spans reaches its output only if the processor is + // flushed before the process goes, so a signal closes the Guardian rather + // than ending it mid-batch. + const stop = async (): Promise => { + await guardian.close(); + process.exit(0); + }; + process.once("SIGINT", stop); + process.once("SIGTERM", stop); +} diff --git a/reference-implementations/agt/packages/guardian/src/otel-trace-exporter.ts b/reference-implementations/agt/packages/guardian/src/otel-trace-exporter.ts new file mode 100644 index 00000000..6a10917d --- /dev/null +++ b/reference-implementations/agt/packages/guardian/src/otel-trace-exporter.ts @@ -0,0 +1,909 @@ +/** + * createOtelTraceExporter emits one OpenTelemetry span per ACS step this + * Guardian decides, so a reviewer can rebuild a session offline from a trace + * file without reading the envelope log. Span names, the attributes each + * span must and may carry, and the event a decision is recorded as all come + * from `specification/v0.1.0/trace/otel-mapping.json`, read once at + * construction. A rename in that mapping needs no change here. + * + * What the mapping does not say is which envelope field feeds which + * attribute. That is the one piece of knowledge this module adds, in the + * four source tables below, written once by hand. An attribute the mapping + * names and no table knows is not emitted, and nothing is invented to fill a + * slot the wire did not carry -- a required attribute the envelope lacks is + * simply absent, which is itself a fact worth seeing. + * + * Span lifetime follows the envelope log. `start` is called where the + * request is recorded and hands back the span; `end` is called with that + * same handle where the response is recorded, at the same two call sites in + * server.ts. The handle is what pairs them, never a lookup: a JSON-RPC id is + * the client's, two sessions can reuse one at the same moment, and nothing + * keyed on the id could tell their responses apart. Only a method that has a + * `step_to_span` entry AND that this Guardian dispatches becomes a span; + * `handshake/hello`, a method the Guardian does not dispatch, and an + * unparseable body produce none, and `start` answers null for them. + * + * A step that got no decision is recorded too, as the `acs.decision.failure` + * event #37 defines, for the failures this Guardian can see itself: an + * error it answered instead of a decision, a decision it finished after the + * host's negotiated timeout (so the host had already applied its posture), + * and a step still open when the Guardian shut down. The host adapter's + * audit log holds the failures only the host sees; audit-to-otlp.ts turns + * those into the same event with `reported_by` set to the observed agent. + * + * Off by default. With no file and no endpoint configured, startGuardian is + * handed NULL_OTEL_TRACE_EXPORTER and behaves exactly as before. + * + * Total by construction, as envelope-log-sink.ts is. Every entry point is + * wrapped, and nothing propagates into or delays the decision path. Each + * output fails on its own: the first failed or throwing export disables + * that output alone, reports once under its label, and reaches the + * processor as a success, so a collector that is down, restarting or + * answering 4xx cannot reject a flush and take the file down with it -- the + * lab reads the file. A broken mapping or a bug inside start/end still + * disables everything, once. Ended spans go to one BatchSpanProcessor per + * output and the file is appended asynchronously, so serialisation and I/O + * happen off the request's own turn. Observability degrades; governance + * does not. + */ +import { mkdirSync, readFileSync } from "node:fs"; +import { appendFile } from "node:fs/promises"; +import { dirname } from "node:path"; +import { fileURLToPath } from "node:url"; +import { SpanStatusCode, type Attributes, type AttributeValue, type Span, type Tracer } from "@opentelemetry/api"; +import { ExportResultCode } from "@opentelemetry/core"; +import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http"; +import { JsonTraceSerializer } from "@opentelemetry/otlp-transformer"; +import { resourceFromAttributes } from "@opentelemetry/resources"; +import { AlwaysOnSampler, BasicTracerProvider, BatchSpanProcessor, type SpanExporter } from "@opentelemetry/sdk-trace-base"; + +/** How long the OTLP/HTTP exporter keeps retrying one batch against an + * endpoint that does not answer before it reports the failure. */ +const ENDPOINT_TIMEOUT_MS = 2000; + +/** How long one batch export, and the exporter's shutdown, may take before + * it is abandoned and reported. A stalled output (a hung network mount, a + * collector that never answers) must not hold the process open: the SDK's + * own default is 30 s, longer than many supervisors wait before a SIGKILL. */ +const EXPORT_TIMEOUT_MS = ENDPOINT_TIMEOUT_MS + 1000; + +/** The longest string attribute a span carries. Every attribute value can + * come from the request, and an uncapped value lets one large request grow + * the Guardian's memory in the export queue; a Guardian that runs out of + * memory fails open under the default posture. `acs.reasoning` longer than + * this is cut to it. */ +const ATTRIBUTE_VALUE_LENGTH_LIMIT = 4096; + +/** The most Provenance objects one step span describes. The length limit + * above bounds each string, not how many a column holds. Both columns are + * cut at the same index, so the i-th element of each still describes the + * same object. */ +const PROVENANCE_COLUMN_LIMIT = 64; + +/** JSON-RPC REPLAY_DETECTED. The pair on a replayed request belongs to the + * earlier request, so #228 forbids a failure event for it: one would mark + * that earlier, legitimate decision as not enforced. */ +const REPLAY_DETECTED_CODE = -32005; + +/** The `service.name` every span carries. The literal key rather than the + * semantic-conventions package: one constant does not earn a dependency. */ +export const OTEL_SERVICE_NAME = "acs-reference-guardian"; + +// The repository's own mapping, five directories up from this file -- the +// same relationship SCHEMA_ROOT in validate-envelope.ts relies on. +const OTEL_MAPPING_PATH = fileURLToPath( + new URL("../../../../../specification/v0.1.0/trace/otel-mapping.json", import.meta.url), +); + +/** + * `acs.decision.failure` is not in otel-mapping.json yet; #228 adds it next + * to `decision_event`. Until then its names live here, in one place, the way + * DECISION_DETAIL_SOURCES holds the decision extras: the event name, its + * attribute keys, and the values the decision on #37 fixed for the class, + * the posture and the reporter. When the mapping gains the event, read the + * name from it the way `decision_event` is read. + */ +export const DECISION_FAILURE_EVENT = "acs.decision.failure"; + +/** The span status for a step answered with an error instead of a decision. + * One sentence for both writers, the Guardian here and the audit converter, + * so a query on status text finds both sides of the same failure. */ +export const GUARDIAN_ERROR_STATUS = "The Guardian returned an error instead of a decision."; + +/** The event's attribute keys: the three required ones first, then the + * ones carried when known. */ +export const FAILURE_ATTRIBUTES = { + class: "acs.failure.class", + posture: "acs.failure.posture", + reportedBy: "acs.failure.reported_by", + detail: "acs.failure.detail", + errorCode: "acs.failure.error_code", + timeoutMs: "acs.failure.timeout_ms", + elapsedMs: "acs.failure.elapsed_ms", + payloadBytes: "acs.failure.payload_bytes", +} as const; + +export type FailureClass = "timeout" | "unreachable" | "malformed_response" | "guardian_error" | "observed_agent_error" | "other"; +export type FailurePosture = "fail_open" | "fail_closed"; +export type FailureReporter = "guardian" | "observed_agent"; + +/** One decision failure, as the event records it. `detail` is required + * when the class is `other`; the rest travel when known. */ +export type DecisionFailure = { + class: FailureClass; + posture: FailurePosture; + reportedBy: FailureReporter; + detail?: string; + errorCode?: number; + timeoutMs?: number; + elapsedMs?: number; + payloadBytes?: number; +}; + +/** The event's attributes: the required three always, the others when + * present. One function for both reporters, so the exporter below and the + * audit-log converter (audit-to-otlp.ts) build the same event. */ +export function decisionFailureAttributes(failure: DecisionFailure): Attributes { + const attributes: Attributes = { + [FAILURE_ATTRIBUTES.class]: failure.class, + [FAILURE_ATTRIBUTES.posture]: failure.posture, + [FAILURE_ATTRIBUTES.reportedBy]: failure.reportedBy, + }; + if (failure.detail !== undefined) attributes[FAILURE_ATTRIBUTES.detail] = failure.detail; + if (failure.errorCode !== undefined) attributes[FAILURE_ATTRIBUTES.errorCode] = failure.errorCode; + if (failure.timeoutMs !== undefined) attributes[FAILURE_ATTRIBUTES.timeoutMs] = failure.timeoutMs; + if (failure.elapsedMs !== undefined) attributes[FAILURE_ATTRIBUTES.elapsedMs] = failure.elapsedMs; + if (failure.payloadBytes !== undefined) attributes[FAILURE_ATTRIBUTES.payloadBytes] = failure.payloadBytes; + return attributes; +} + +/** One `step_to_span` entry. `conditional_attributes` is the list #227 + * adds, for attributes the wire carries only sometimes (`acs.capability` + * among them); read the same way as the other two. */ +type StepSpanEntry = { + span_name: string; + required_attributes: string[]; + conditional_attributes?: string[]; + optional_attributes?: string[]; +}; + +/** + * Methods this Guardian evaluates that have no `step_to_span` entry in + * otel-mapping.json yet, and why. A method here gets no span. The test + * fails for an evaluated method that is neither mapped nor listed here, so + * a new method cannot quietly go untraced: it is either mapped or named + * with its reason. + */ +export const METHODS_WITHOUT_SPAN: Readonly> = {}; + +/** otel-mapping.json is a JSON Schema whose data lives inside `default` + * keys -- the same three places `packages/conformance/src/trace-pillar.ts` + * reads. This type names only those. */ +type OtelMappingSchema = { + properties: { + step_to_span: { default: Record }; + decision_event: { + properties: { + event_name: { const: string }; + required_attributes: { default: string[] }; + conditional_attributes: { default: string[] }; + }; + }; + provenance_attributes: { + properties: { + required: { default: string[] }; + optional: { default: string[] }; + }; + }; + }; +}; + +export type OtelMapping = { + stepToSpan: Record; + decisionEventName: string; + /** The event's required attributes first, then its conditional ones. Both + * are read the same way -- emitted when a value is present -- so one list + * serves. */ + decisionAttributes: string[]; + /** Required first, then optional -- carried on a step span whose payload + * has a Provenance object attached, as the mapping's own text says. */ + provenanceAttributes: string[]; +}; + +export function loadOtelMapping(path: string = OTEL_MAPPING_PATH): OtelMapping { + const schema = JSON.parse(readFileSync(path, "utf8")) as OtelMappingSchema; + const decision = schema.properties.decision_event.properties; + const provenance = schema.properties.provenance_attributes.properties; + return { + stepToSpan: schema.properties.step_to_span.default, + decisionEventName: decision.event_name.const, + decisionAttributes: [...decision.required_attributes.default, ...decision.conditional_attributes.default], + provenanceAttributes: [...provenance.required.default, ...provenance.optional.default], + }; +} + +type JsonObject = Record; + +function asObject(value: unknown): JsonObject | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) ? (value as JsonObject) : undefined; +} + +function asString(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +function asNumber(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) ? value : undefined; +} + +/** A non-empty array of strings, or nothing. An empty array is carried by + * the wire (`policy_references: []` on an envelope-invalid deny) but says + * nothing a reader can use, so it is not emitted. */ +function asStrings(value: unknown): string[] | undefined { + if (!Array.isArray(value) || value.length === 0) return undefined; + return value.every((item) => typeof item === "string") ? (value as string[]) : undefined; +} + +/** + * One column of `policy_references`, in the wire's own order: the same + * length as `policy_references`, with `""` where a reference lacks the + * field. That is the rule agreed on #228, so a reader lines the three + * columns up by position and never loses the values the other references + * did carry. Absent only when there are no references at all. + */ +function policyColumn(references: unknown, field: string): string[] | undefined { + if (!Array.isArray(references) || references.length === 0) return undefined; + return references.map((reference) => asString(asObject(reference)?.[field]) ?? ""); +} + +/** The parts of a request envelope the source tables read. Read off the + * parsed JSON rather than the validated envelope, because the span starts + * before validation runs and an envelope that fails the schema is still a + * step this session took. */ +export type RequestParts = { params: JsonObject; metadata: JsonObject; payload: JsonObject; tool: JsonObject }; + +function requestParts(envelope: unknown): RequestParts { + const params = asObject(asObject(envelope)?.params) ?? {}; + const metadata = asObject(params.metadata) ?? {}; + const payload = asObject(params.payload) ?? {}; + const tool = asObject(payload.tool) ?? {}; + return { params, metadata, payload, tool }; +} + +/** The decision half of a success response: the `result` and its optional + * `metadata`. `undefined` for a JSON-RPC error response, which carries no + * decision. */ +export type ResultParts = { result: JsonObject; metadata: JsonObject }; + +function resultParts(response: unknown): ResultParts | undefined { + const result = asObject(asObject(response)?.result); + if (result === undefined) return undefined; + return { result, metadata: asObject(result.metadata) ?? {} }; +} + +export type Source = (parts: Parts) => AttributeValue | undefined; + +/** + * Attribute name -> where it is read from on the request envelope, for every + * attribute `step_to_span` can name on a span this Guardian emits. The + * mapping decides WHICH of these a span carries; this table only says HOW + * each one is read. Paths follow the hook payload schemas + * (`hooks/tool-call-request.json`, `hooks/tool-call-result.json`) and + * `request-envelope.json`'s `AcsParams.metadata`. + */ +export const STEP_ATTRIBUTE_SOURCES: Readonly>> = { + "acs.session.id": (r) => asString(r.metadata.session_id), + "acs.agent.id": (r) => asString(r.metadata.agent_id), + "gen_ai.tool.name": (r) => asString(r.tool.name), + "acs.tool.provider": (r) => asString(r.tool.provider), + "acs.tool.version": (r) => asString(r.tool.version), + "acs.capability": (r) => asString(r.payload.capability), + "acs.operation": (r) => asString(r.payload.operation), + "acs.exit_status": (r) => asString(r.payload.exit_status), + "acs.duration_ms": (r) => asNumber(r.payload.duration_ms), + "acs.request_id_ref": (r) => asString(r.payload.request_id_ref), +}; + +/** + * Carried on every span, whatever the mapping lists for it, so a reader can + * rebuild a session and pair each span with its two envelope-log lines. + * `acs.session.id` is the mapping's own name for the session. + * `acs.request_id` is this module's name for `params.request_id`, the ACS + * correlation id the response echoes; the mapping has no attribute for the + * step's own request id (its `acs.request_id_ref` names the request a + * RESULT refers back to, which is a different fact). + */ +const CORRELATION_ATTRIBUTE_SOURCES: Record> = { + "acs.session.id": (r) => asString(r.metadata.session_id), + "acs.request_id": (r) => asString(r.params.request_id), +}; + +/** + * The `acs.decision` event's attributes, for every name the mapping's + * `decision_event` lists. Paths follow `response-envelope.json`'s + * `AcsResult`: `decision` and `reasoning` are direct members; the other four + * sit under `metadata`. `acs.evaluator` is read off the wire like the rest + * and never filled in here: the value this Guardian would send is + * `deterministic`, but it has to come from the Guardian's response (#229 + * puts it there), or the trace says something the wire never said and a + * reader cannot tell the two apart. + */ +export const DECISION_ATTRIBUTE_SOURCES: Readonly>> = { + "acs.decision": (d) => asString(d.result.decision), + "acs.evaluator": (d) => asString(d.metadata.evaluator), + "acs.reasoning": (d) => asString(d.result.reasoning), + "acs.confidence": (d) => asNumber(d.metadata.confidence), + "acs.evaluator_version": (d) => asString(d.metadata.evaluator_version), + "acs.model_id": (d) => asString(d.metadata.model_id), + // The decision facts #228 adds to the mapping. Each is emitted only when + // the response carries it; `acs.chain_hash` in particular appears only once + // a Guardian puts the chain head on the wire. + "acs.reason_codes": (d) => asStrings(d.result.reason_codes), + "acs.policy.ids": (d) => policyColumn(d.result.policy_references, "policy_id"), + "acs.policy.versions": (d) => policyColumn(d.result.policy_references, "policy_version"), + "acs.policy.rule_ids": (d) => policyColumn(d.result.policy_references, "rule_id"), + "acs.cited_provenance_ids": (d) => asStrings(d.result.cited_provenance_ids), + "acs.chain_hash": (d) => asString(d.result.chain_hash), +}; + +/** + * The decision facts emitted whether or not the mapping lists them yet. + * Before #228 the mapping does not name them; after it, it does. Reading + * this list beside the mapping's own makes the exporter correct under both, + * so this PR and #228 can land in either order. Once #228 is in, the mapping + * alone covers them and this list can go. + */ +const DECISION_DETAIL_NAMES = [ + "acs.reason_codes", + "acs.policy.ids", + "acs.policy.versions", + "acs.policy.rule_ids", + "acs.cited_provenance_ids", + "acs.chain_hash", +] as const; + +/** + * Attribute name -> which field of a Provenance object (`provenance.json`) + * it reads. Unlike the tables above, each of these becomes a COLUMN: one + * array per attribute, one element per Provenance object the payload + * carries, in wire order. A span has one value per attribute key, and a + * payload can carry several Provenance objects -- one per argument -- so an + * array is the only shape that keeps every object visible; the i-th element + * of each column describes the i-th object, and `acs.provenance.origin` is + * always complete because provenance.json requires `origin`. + * + * `acs.provenance.lineage_depth` is in the mapping's optional list and has + * no entry here. provenance.json defines no lineage_depth field, so there is + * nothing on the wire to read it from, and deriving one (from the length of + * `derived_from`, say) would be a value nobody sent. + */ +const PROVENANCE_ATTRIBUTE_SOURCES: Record string | undefined> = { + "acs.provenance.origin": (p) => asString(p.origin), + "acs.provenance.source_id": (p) => asString(p.source_id), +}; + +/** The Provenance objects a payload carries, in wire order: one per + * argument that has one, in the order the parsed object holds them, then one + * per output that has one, in array order. "Parsed" order rather than the + * bytes' order because `JSON.parse` has already hoisted integer-like + * argument names ahead of the rest (envelope-log-sink.ts notes `{"0": ...}` + * is a real shape), so this is the order every reader of the envelope log + * sees too. The two never meet on one payload -- the request gate carries + * `arguments`, the result gate `outputs` -- and an argument or output + * without one contributes nothing. */ +function provenanceObjects(payload: JsonObject): JsonObject[] { + const carriers: unknown[] = [ + ...Object.values(asObject(payload.arguments) ?? {}), + ...(Array.isArray(payload.outputs) ? payload.outputs : []), + ]; + const found: JsonObject[] = []; + for (const carrier of carriers) { + if (found.length === PROVENANCE_COLUMN_LIMIT) { + break; + } + const provenance = asObject(asObject(carrier)?.provenance); + if (provenance !== undefined) { + found.push(provenance); + } + } + return found; +} + +/** One column over every Provenance object, in order, padded with `""` + * where an object lacks the field -- the same rule policyColumn applies, for + * the same reason: every column is the length of `acs.provenance.origin`, so + * the i-th element of each describes the i-th object, and a `source_id` one + * argument carries is never hidden by another argument that has none. */ +function provenanceColumn(objects: JsonObject[], read: (provenance: JsonObject) => string | undefined): string[] { + return objects.map((object) => read(object) ?? ""); +} + +function collect(into: Attributes, names: Iterable, sources: Record>, parts: Parts): void { + for (const name of names) { + const value = sources[name]?.(parts); + if (value !== undefined) { + into[name] = value; + } + } +} + +/** + * The span this module started for one step, handed to the caller so the + * response can close the same span and no other. Opaque: a caller holds it + * between the two calls and reads nothing off it. Beside the span it keeps + * what a failure event needs at the end: the method, for its per-method + * timeout; when it started, for `acs.failure.elapsed_ms`; and the request + * body's size, when the caller measured one. + */ +export type StepSpan = { + readonly span: Span; + readonly method: string; + readonly startedAt: number; + readonly payloadBytes?: number; +}; + +/** The Guardian-side writer role, called at the envelope log's own two + * points: `start` where the request is recorded, `end` where the response + * is, with the handle `start` returned. */ +export type OtelTraceExporter = { + /** Null when this request produces no span -- `end` then does nothing. + * `payloadBytes` is the request body's size, for the failure event. */ + start(envelope: unknown, method: string | null, payloadBytes?: number): StepSpan | null; + end(stepSpan: StepSpan | null, response: unknown): void; + /** Pushes every ended span through to the exporters now rather than on + * the batch timer. Tests call it; shutdown does too. */ + flush(): Promise; + /** Flushes, then releases the processors. A second call waits for the + * first rather than returning at once: SIGINT followed by SIGTERM is two + * calls, and the second must not let the process exit mid-flush. */ + shutdown(): Promise; +}; + +/** The exporter a Guardian gets when neither output is configured: the + * composition root disabling trace export gets a value of the same type, + * not a special case. */ +export const NULL_OTEL_TRACE_EXPORTER: OtelTraceExporter = { + start(): null { + return null; + }, + end(): void {}, + async flush(): Promise {}, + async shutdown(): Promise {}, +}; + +export type CreateOtelTraceExporterOptions = { + /** OTLP/JSON file, one ExportTraceServiceRequest per line. */ + filePath?: string; + /** OTLP/HTTP traces endpoint, e.g. `http://127.0.0.1:4318/v1/traces`. */ + endpoint?: string; + /** A span exporter used beside (or instead of) the two above. Exists so + * tests can read spans back in-process; not meant for production use. */ + spanExporter?: SpanExporter; + /** The methods this Guardian dispatches -- the ServerHello's own list. + * A method outside it produces no span even when the mapping names one. */ + methodsEvaluated: readonly string[]; + /** The ServerHello's `timeout_config`: how long the host waits for a + * decision before it applies its posture. A decision that finishes after + * that is one the host never acted on. Omitted means no timeout is known + * and no step is reported as late. */ + timeoutConfig?: { default_ms: number; per_method_ms?: Record }; + /** The posture the host applies when no decision arrives, in the trace + * vocabulary: `fail_open` for a ServerHello declaring `proceed`, + * `fail_closed` for `deny`. Defaults to `fail_open`, the spec's own default + * posture (handshake.json), for an exporter built without a ServerHello. */ + posture?: "fail_open" | "fail_closed"; + /** Overrides the mapping file. Defaults to the repository's own. */ + mappingPath?: string; + /** How long one export, and shutdown, may take before it is abandoned and + * reported. Defaults to EXPORT_TIMEOUT_MS; tests pass a short one. */ + exportTimeoutMs?: number; + /** Called at most once per output, on that output's first failure, and + * at most once more for a failure of the exporter as a whole (a mapping + * that will not load, a bug inside start or end). Defaults to one stderr + * line each. */ + onError?: (error: unknown) => void; +}; + +function asError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + +/** + * Writes OTLP/JSON, one `ExportTraceServiceRequest` per line -- the shape + * the OpenTelemetry Collector's file exporter writes -- serialised by the + * official JsonTraceSerializer rather than by hand. + * + * The append is asynchronous: this exporter runs on the same event loop the + * Guardian answers requests on, and a synchronous write would hold every + * in-flight decision for the duration of the disk I/O. Writes are chained on + * one promise so lines land in export order even when two batches are in + * flight, and `forceFlush` and `shutdown` await the chain so a close never + * leaves a batch unwritten. The chain itself never rejects: one failed write + * is reported through its own batch's callback and cannot poison every write + * after it. + * + * 0o700 on the directory and 0o600 on the file: the file holds reasoning + * text and provenance source ids, so other local users should not read it + * (#71 is the same problem in the envelope log). The mode applies when the + * exporter creates the path; an existing file keeps whatever it had. + */ +export function createFileSpanExporter(path: string): SpanExporter { + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); + let chain: Promise = Promise.resolve(); + return { + export(spans, resultCallback): void { + let line: string; + try { + const bytes = JsonTraceSerializer.serializeRequest(spans); + if (bytes === undefined) { + throw new Error("JsonTraceSerializer produced no request"); + } + line = `${new TextDecoder().decode(bytes)}\n`; + } catch (error) { + resultCallback({ code: ExportResultCode.FAILED, error: asError(error) }); + return; + } + const write = chain.then(() => appendFile(path, line, { mode: 0o600 })); + chain = write.then( + () => undefined, + () => undefined, + ); + write.then( + () => resultCallback({ code: ExportResultCode.SUCCESS }), + (error: unknown) => resultCallback({ code: ExportResultCode.FAILED, error: asError(error) }), + ); + }, + async shutdown(): Promise { + await chain; + }, + async forceFlush(): Promise { + await chain; + }, + }; +} + +/** One place spans go, with the name its failure is reported under: the + * file's path, the endpoint's URL, or "span exporter" for an injected one. */ +type Output = { label: string; exporter: SpanExporter }; + +/** + * Wraps one output so it can fail without taking any other output with it. + * + * The first failed result, thrown export, or rejected flush marks this output + * dead and reports it once, under its label; every later batch for it is + * dropped. The processor is answered with SUCCESS either way, deliberately: + * a FAILED result makes the processor's forceFlush reject, the provider's + * forceFlush and shutdown reject with it, and a collector that is merely down + * would then end export to the file as well. The failure is this module's + * finding, reported here, not the processor's. The SDK's own route for a + * failed result is its global error handler, silent unless a diag logger is + * installed, which is why the report happens here at all. + */ +function isolated({ label, exporter }: Output, report: (label: string, error: unknown) => void): SpanExporter { + let dead = false; + const die = (error: unknown): void => { + if (dead) { + return; + } + dead = true; + report(label, error); + }; + return { + export(spans, resultCallback): void { + if (dead) { + resultCallback({ code: ExportResultCode.SUCCESS }); + return; + } + try { + exporter.export(spans, (result) => { + if (result.code === ExportResultCode.FAILED) { + die(result.error ?? new Error("span export failed")); + } + resultCallback({ code: ExportResultCode.SUCCESS }); + }); + } catch (error) { + die(error); + resultCallback({ code: ExportResultCode.SUCCESS }); + } + }, + async shutdown(): Promise { + // Shut the output down whether or not it died: a dead HTTP output + // still holds its keep-alive agent until its exporter releases it. + try { + await exporter.shutdown(); + } catch (error) { + die(error); + } + }, + async forceFlush(): Promise { + if (dead) { + return; + } + try { + await exporter.forceFlush?.(); + } catch (error) { + die(error); + } + }, + }; +} + +export function createOtelTraceExporter({ + filePath, + endpoint, + spanExporter, + methodsEvaluated, + timeoutConfig, + posture = "fail_open", + mappingPath, + exportTimeoutMs = EXPORT_TIMEOUT_MS, + onError, +}: CreateOtelTraceExporterOptions): OtelTraceExporter { + /** One report, to the caller's hook or to stderr, never thrown: the + * reporter is the one component here that exists only to report, and a + * throw out of it would be the failure it is meant to describe. */ + const report = (what: string, error: unknown): void => { + try { + if (onError) { + onError(error); + return; + } + const message = error instanceof Error ? error.message : String(error); + console.error(`${what} disabled after failure: ${message}`); + } catch { + // Silently swallow any error from the callback or console.error + // to maintain the total-by-construction guarantee + } + }; + + // The exporter as a whole: a mapping that will not load, a provider that + // will not build, a bug inside start or end. Once, then quiet. + let disabled = false; + const fail = (error: unknown): void => { + if (disabled) { + return; + } + disabled = true; + report("otel trace exporter", error); + }; + + // One output: its own first failure, under its own label, and only it. + const reportOutput = (label: string, error: unknown): void => report(`otel trace export to ${label}`, error); + + // Each output is constructed on its own, so an unwritable file path costs + // the file and not the endpoint, and the other way round. + const outputs: Output[] = []; + const construct = (label: string, build: () => SpanExporter): void => { + try { + outputs.push({ label, exporter: build() }); + } catch (error) { + reportOutput(label, error); + } + }; + if (filePath !== undefined) { + construct(filePath, () => createFileSpanExporter(filePath)); + } + if (endpoint !== undefined) { + // A refused connection is retryable to this exporter, so a collector that + // is down costs its whole timeout before the failure is reported. Two + // seconds rather than the 10 s default: long enough for a collector that + // is restarting, short enough that a flush or a shutdown behind a dead + // one does not hold the process for ten. + construct(endpoint, () => new OTLPTraceExporter({ url: endpoint, timeoutMillis: ENDPOINT_TIMEOUT_MS })); + } + if (spanExporter !== undefined) { + construct("span exporter", () => spanExporter); + } + if (outputs.length === 0) { + return NULL_OTEL_TRACE_EXPORTER; + } + + let provider: BasicTracerProvider | undefined; + let tracer: Tracer | undefined; + let mapping: OtelMapping | undefined; + try { + mapping = loadOtelMapping(mappingPath); + provider = new BasicTracerProvider({ + resource: resourceFromAttributes({ "service.name": OTEL_SERVICE_NAME }), + // Set explicitly so no OTEL_* environment variable can sample spans + // away or drop decision events: the SDK reads OTEL_TRACES_SAMPLER and + // the span limits from the environment for anything left unset. + sampler: new AlwaysOnSampler(), + spanLimits: { + attributeValueLengthLimit: ATTRIBUTE_VALUE_LENGTH_LIMIT, + attributeCountLimit: 128, + eventCountLimit: 128, + attributePerEventCountLimit: 128, + linkCountLimit: 128, + attributePerLinkCountLimit: 128, + }, + spanProcessors: outputs.map( + (output) => new BatchSpanProcessor(isolated(output, reportOutput), { exportTimeoutMillis: exportTimeoutMs }), + ), + }); + tracer = provider.getTracer(OTEL_SERVICE_NAME); + } catch (error) { + fail(error); + } + + const handled = new Set(methodsEvaluated); + + // Started and not yet ended, so shutdown can close what no response ever + // reached: a bridge that never answers, or a signal during traffic. An + // open span is never exported, and without this the step left no trace. + const openSpans = new Set(); + + /** The failure event's attributes for one step, as this Guardian reports + * them: its own posture, elapsed time since the span started, and the + * request's size when known. */ + function failureAttributes(stepSpan: StepSpan, failureClass: FailureClass, known: Partial = {}): Attributes { + return decisionFailureAttributes({ + class: failureClass, + posture, + reportedBy: "guardian", + elapsedMs: Math.round(performance.now() - stepSpan.startedAt), + payloadBytes: stepSpan.payloadBytes, + ...known, + }); + } + + function begin(envelope: unknown, method: string | null, payloadBytes: number | undefined, tracer: Tracer, mapping: OtelMapping): StepSpan | null { + if (method === null || !handled.has(method)) { + return null; + } + const entry = mapping.stepToSpan[method]; + if (entry === undefined) { + return null; + } + const parts = requestParts(envelope); + const attributes: Attributes = {}; + collect(attributes, Object.keys(CORRELATION_ATTRIBUTE_SOURCES), CORRELATION_ATTRIBUTE_SOURCES, parts); + collect( + attributes, + [...entry.required_attributes, ...(entry.conditional_attributes ?? []), ...(entry.optional_attributes ?? [])], + STEP_ATTRIBUTE_SOURCES, + parts, + ); + const provenance = provenanceObjects(parts.payload); + if (provenance.length > 0) { + for (const name of mapping.provenanceAttributes) { + const read = PROVENANCE_ATTRIBUTE_SOURCES[name]; + // A column is emitted only when at least one object carries the + // field (#228). `origin` is required, so it always is; an + // all-padding `source_id` column would say something nobody sent. + if (read !== undefined && provenance.some((object) => read(object) !== undefined)) { + attributes[name] = provenanceColumn(provenance, read); + } + } + } + const stepSpan: StepSpan = { + span: tracer.startSpan(entry.span_name, { attributes, root: true }), + method, + startedAt: performance.now(), + payloadBytes, + }; + openSpans.add(stepSpan); + return stepSpan; + } + + function finish(stepSpan: StepSpan, envelope: unknown, mapping: OtelMapping): void { + const { span } = stepSpan; + const decision = resultParts(envelope); + if (decision === undefined) { + // A JSON-RPC error: the step was answered but not decided. The first + // of the two failures a Guardian can see itself (#37). Only the code is + // carried: the message can quote the request (a schema error names its + // argument keys), and #228 forbids copying response or error bytes into + // the trace, so the status message is a fixed sentence too. + const error = asObject(asObject(envelope)?.error); + const errorCode = asNumber(error?.code); + span.setStatus({ code: SpanStatusCode.ERROR, message: GUARDIAN_ERROR_STATUS }); + if (errorCode !== REPLAY_DETECTED_CODE) { + span.addEvent(DECISION_FAILURE_EVENT, failureAttributes(stepSpan, "guardian_error", { errorCode })); + } + } else { + const attributes: Attributes = {}; + collect(attributes, new Set([...mapping.decisionAttributes, ...DECISION_DETAIL_NAMES]), DECISION_ATTRIBUTE_SOURCES, decision); + if (attributes["acs.decision"] !== undefined) { + span.addEvent(mapping.decisionEventName, attributes); + } + // The second: a decision that finished after the host stopped waiting. + // The host applied its posture at the timeout, so the decision above + // probably never took effect, and a reader has to be told. + const timeoutMs = timeoutConfig?.per_method_ms?.[stepSpan.method] ?? timeoutConfig?.default_ms; + if (timeoutMs !== undefined && performance.now() - stepSpan.startedAt > timeoutMs) { + span.addEvent(DECISION_FAILURE_EVENT, failureAttributes(stepSpan, "timeout", { timeoutMs })); + } + } + span.end(); + } + + let stopped = false; + // The one shutdown, started by the first call and returned to every call + // after it, so a second signal waits for the flush the first began. + let shutdownPromise: Promise | undefined; + + return { + start(envelope, method, payloadBytes): StepSpan | null { + if (disabled || stopped || tracer === undefined || mapping === undefined) { + return null; + } + try { + return begin(envelope, method, payloadBytes, tracer, mapping); + } catch (error) { + fail(error); + return null; + } + }, + end(stepSpan, response): void { + if (stepSpan === null) { + return; + } + // Removed first, whatever happens next: a span the caller is ending + // is no longer one shutdown has to close. + openSpans.delete(stepSpan); + if (disabled || stopped || mapping === undefined) { + return; + } + try { + finish(stepSpan, response, mapping); + } catch (error) { + fail(error); + } + }, + async flush(): Promise { + if (provider === undefined || stopped) { + return; + } + try { + await provider.forceFlush(); + } catch (error) { + fail(error); + } + }, + shutdown(): Promise { + shutdownPromise ??= (async () => { + if (provider === undefined) { + return; + } + stopped = true; + // Steps still open: a bridge that never answered, or a signal during + // traffic. Closed here, as the third failure this Guardian can see, + // so the step is exported at all. + for (const stepSpan of openSpans) { + try { + stepSpan.span.addEvent( + DECISION_FAILURE_EVENT, + failureAttributes(stepSpan, "other", { detail: "the Guardian shut down before it answered this step" }), + ); + stepSpan.span.setStatus({ code: SpanStatusCode.ERROR, message: "no response before shutdown" }); + stepSpan.span.end(); + } catch (error) { + fail(error); + } + } + openSpans.clear(); + // Flushes every processor, then releases them -- bounded, because a + // stalled output would otherwise hold the process open. The SDK's own + // export timeout covers a batch in flight; this covers an exporter + // whose shutdown never returns. + let timer: ReturnType | undefined; + const deadline = new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), exportTimeoutMs); + timer.unref?.(); + }); + try { + const outcome = await Promise.race([provider.shutdown().then(() => "done" as const), deadline]); + if (outcome === "timeout") { + report("otel trace exporter shutdown", new Error(`did not finish within ${exportTimeoutMs} ms; unexported spans were dropped`)); + } + } catch (error) { + fail(error); + } finally { + clearTimeout(timer); + } + })(); + return shutdownPromise; + }, + }; +} diff --git a/reference-implementations/agt/packages/guardian/src/server.ts b/reference-implementations/agt/packages/guardian/src/server.ts index 9fd4e21f..a20d10a8 100644 --- a/reference-implementations/agt/packages/guardian/src/server.ts +++ b/reference-implementations/agt/packages/guardian/src/server.ts @@ -93,6 +93,7 @@ import { import { checkResponse } from "./check-response.ts"; import { buildServerHello, type ServerHello } from "./handshake.ts"; import { createEnvelopeLogSink, NULL_ENVELOPE_LOG_SINK, type EnvelopeLogSink } from "./envelope-log-sink.ts"; +import { createOtelTraceExporter, NULL_OTEL_TRACE_EXPORTER, type OtelTraceExporter } from "./otel-trace-exporter.ts"; import { appendContextEntry, createMemorySessionContextStore, @@ -331,6 +332,17 @@ export type StartGuardianOptions = { * `sessionContextLog` when both are supplied. */ sessionContextStore?: SessionContextStore; + /** OTLP/JSON trace file, one export request per line. Omitted means no + * span is written. `packages/guardian/src/main.ts` threads ACS_OTEL_FILE + * into this; otel-trace-exporter.ts says what a span carries. */ + otelTraceFile?: string; + /** OTLP/HTTP traces endpoint. Omitted means no span is sent. main.ts + * threads ACS_OTEL_ENDPOINT into this. */ + otelTraceEndpoint?: string; + /** Overrides the trace exporter entirely, bypassing both paths above. + * Exists so a test can read spans back in-process or stand in one that + * fails. Not meant for production use. */ + traceExporter?: OtelTraceExporter; }; export type StartedGuardian = { url: string; close(): Promise }; @@ -398,6 +410,9 @@ export async function startGuardian({ bridge: bridgeOverride, sessionContextLog, sessionContextStore: sessionContextStoreOverride, + otelTraceFile, + otelTraceEndpoint, + traceExporter: traceExporterOverride, }: StartGuardianOptions): Promise { // Construct the bridge once at boot, not per request -- and through the // shared recipe rather than assembling one here, so this Guardian and every @@ -420,6 +435,25 @@ export async function startGuardian({ createMemorySessionContextStore( sessionContextLog ? { appendLine: createSessionContextLogAppender(sessionContextLog) } : {}, ); + // Off unless an output is named -- the same relationship envelopeLogPath + // has to the envelope log. The override wins over both paths, as the other + // test seams above do. + // The same ServerHello the handshake sends, read once here, so the trace + // agrees with what this Guardian told the host: which methods it + // evaluates, how long the host waits, and which posture it applies when + // no decision arrives. The timeout also bounds close() below. + const hello = buildServerHello(onDecisionFailure === undefined ? undefined : { ACS_ON_DECISION_FAILURE: onDecisionFailure }); + const traceExporter = + traceExporterOverride ?? + (otelTraceFile !== undefined || otelTraceEndpoint !== undefined + ? createOtelTraceExporter({ + filePath: otelTraceFile, + endpoint: otelTraceEndpoint, + methodsEvaluated: hello.methods_evaluated, + timeoutConfig: hello.timeout_config, + posture: hello.on_decision_failure === "deny" ? "fail_closed" : "fail_open", + }) + : NULL_OTEL_TRACE_EXPORTER); const server = Bun.serve({ hostname: hostname ?? LOOPBACK_ONLY, @@ -429,7 +463,7 @@ export async function startGuardian({ if (req.method !== "POST" || pathname !== ACS_PATH) { return new Response("Not Found", { status: 404 }); } - const response = await handleAcsRequest(req, bridge, mapping, envelopeLog, onDecisionFailure, sessionContextStore); + const response = await handleAcsRequest(req, bridge, mapping, envelopeLog, onDecisionFailure, sessionContextStore, traceExporter); return Response.json(response); }, }); @@ -437,7 +471,25 @@ export async function startGuardian({ return { url: `http://localhost:${server.port}${ACS_PATH}`, async close() { - await server.stop(true); + if (traceExporter === NULL_OTEL_TRACE_EXPORTER) { + // No exporter, nothing to close behind the drain: the close every + // Guardian had before trace export existed. + await server.stop(true); + return; + } + // Bun.serve's stop(true) closes every connection at once but resolves + // only when every in-flight handler has returned -- and a handler whose + // bridge never answers never returns. The drain is therefore bounded by + // the negotiated timeout: past it, no host is still waiting for the + // answer. The timer is unref'd so a prompt stop does not leave it + // holding the process open. The exporter shuts down after the drain, + // so a step still open then is closed as the failure it is rather + // than lost. + const deadline = new Promise((resolve) => { + setTimeout(resolve, hello.timeout_config.default_ms).unref(); + }); + await Promise.race([server.stop(true), deadline]); + await traceExporter.shutdown(); }, }; } @@ -469,7 +521,10 @@ export async function startGuardian({ * MAX_REQUEST_BODY_BYTES). * * The sink itself is total (see envelope-log-sink.ts): these two calls cannot - * throw, so they cannot turn a governed tool call into an ungoverned one. + * throw, so they cannot turn a governed tool call into an ungoverned one. The + * trace exporter's `start` and `end` sit beside them: the handle `start` + * returns is what `end` closes, so a span can only ever close on its own + * response, and both are total for the same reason (otel-trace-exporter.ts). */ async function handleAcsRequest( req: Request, @@ -481,6 +536,7 @@ async function handleAcsRequest( envelopeLog: EnvelopeLogSink, onDecisionFailure: "proceed" | "deny" | undefined, sessionContextStore: SessionContextStore, + traceExporter: OtelTraceExporter, ): Promise { const body = await readCappedBody(req); if (body.withinLimit === false) { @@ -518,6 +574,7 @@ async function handleAcsRequest( // visible to the Inspector rather than invisible. const method = extractMethod(raw); envelopeLog.write("request", raw, method); + const stepSpan = traceExporter.start(raw, method, Buffer.byteLength(body.text)); let response: JsonRpcSuccess | JsonRpcFailure; try { @@ -563,6 +620,7 @@ async function handleAcsRequest( } envelopeLog.write("response", response, method); + traceExporter.end(stepSpan, response); return response; } diff --git a/reference-implementations/agt/packages/guardian/test/otel-trace-exporter.test.ts b/reference-implementations/agt/packages/guardian/test/otel-trace-exporter.test.ts new file mode 100644 index 00000000..b3ca0da5 --- /dev/null +++ b/reference-implementations/agt/packages/guardian/test/otel-trace-exporter.test.ts @@ -0,0 +1,1238 @@ +import { describe, expect, it, spyOn } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync, rmdirSync, statSync, unlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { SpanStatusCode } from "@opentelemetry/api"; +import { InMemorySpanExporter, type ReadableSpan, type SpanExporter } from "@opentelemetry/sdk-trace-base"; +import type { PolicyBridge } from "agt-bridge"; +import { startGuardian, type GuardianSnapshot } from "../src/index.ts"; +import { FAILURE_DETAIL_BY_KIND } from "../src/audit-to-otlp.ts"; +import { buildServerHello } from "../src/handshake.ts"; +import { + createFileSpanExporter, + createOtelTraceExporter, + DECISION_ATTRIBUTE_SOURCES, + loadOtelMapping, + METHODS_WITHOUT_SPAN, + NULL_OTEL_TRACE_EXPORTER, + OTEL_SERVICE_NAME, + STEP_ATTRIBUTE_SOURCES, + type CreateOtelTraceExporterOptions, + type OtelTraceExporter, +} from "../src/otel-trace-exporter.ts"; + +/** The live mapping, read the same way the exporter reads it, so every + * expected name below comes from the file and not from a literal that could + * drift from it. */ +const MAPPING = loadOtelMapping(); +/** The methods this Guardian evaluates, from the ServerHello it sends -- + * the same source server.ts builds the exporter from. */ +const METHODS_EVALUATED = buildServerHello().methods_evaluated; +const TOOL_CALL_SPAN = MAPPING.stepToSpan["steps/toolCallRequest"]!; + +function makeEnvelope( + method: string, + payload: Record, + overrides: { id?: number | null; requestId?: string; sessionId?: string } = {}, +): Record { + const { id = 1, requestId = crypto.randomUUID(), sessionId = crypto.randomUUID() } = overrides; + return { + jsonrpc: "2.0", + method, + ...(id === null ? {} : { id }), + params: { + acs_version: "0.1.0", + request_id: requestId, + timestamp: new Date().toISOString(), + metadata: { agent_id: "agent-1", session_id: sessionId }, + payload, + }, + }; +} + +/** Carries `capability`, which hooks/tool-call-request.json declares + * optional and the mapping requires on the span -- so a span built from this + * envelope has every required attribute the mapping names. */ +function toolCallEnvelope(command: string, overrides: { id?: number | null; requestId?: string; sessionId?: string } = {}) { + return makeEnvelope( + "steps/toolCallRequest", + { tool: { name: "run_shell" }, capability: "process.exec", arguments: { command: { value: command } } }, + overrides, + ); +} + +/** A Provenance object as provenance.json defines it: `provenance_id` and + * `origin` required, everything else optional. */ +function provenance(origin: string, extra: Record = {}): Record { + return { provenance_id: crypto.randomUUID(), origin, ...extra }; +} + +function decisionResponse(id: number, result: Record): Record { + return { jsonrpc: "2.0", id, result: { type: "final", acs_version: "0.1.0", request_id: "r-1", ...result } }; +} + +const DENY = { + decision: "deny", + reasoning: "blocked", + reason_codes: ["dangerous_command"], + policy_references: [{ policy_id: "agt_stock", rule_id: "deny-rm-rf" }], +}; + +function memoryExporter(options: Partial = {}): { + exporter: OtelTraceExporter; + spans(): ReadableSpan[]; +} { + const memory = new InMemorySpanExporter(); + const exporter = createOtelTraceExporter({ spanExporter: memory, methodsEvaluated: METHODS_EVALUATED, ...options }); + return { exporter, spans: () => memory.getFinishedSpans() }; +} + +/** A span exporter that fails on every export, the way an unwritable file or + * an unreachable collector would. */ +function throwingSpanExporter(): SpanExporter { + return { + export(): void { + throw new Error("exporter threw"); + }, + async shutdown(): Promise {}, + }; +} + +async function postRaw(url: string, body: unknown): Promise> { + const res = await fetch(url, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) }); + return (await res.json()) as Record; +} + +const baseOptions = { port: 0, manifestPath: "policy/manifest.yaml" } as const; + +/** The OTLP/JSON `ExportTraceServiceRequest` shape, as far as these tests + * read it: resource attributes, and span names and events. */ +type OtlpAttribute = { key: string; value: { stringValue?: string; intValue?: string | number } }; +type OtlpEvent = { name: string; attributes?: OtlpAttribute[] }; +type OtlpSpan = { name: string; attributes?: OtlpAttribute[]; events?: OtlpEvent[]; status?: { code?: number } }; +type OtlpRequest = { + resourceSpans: { + resource: { attributes: OtlpAttribute[] }; + scopeSpans: { spans: OtlpSpan[] }[]; + }[]; +}; + +function otlpSpans(requests: OtlpRequest[]): OtlpSpan[] { + return requests.flatMap((r) => r.resourceSpans.flatMap((rs) => rs.scopeSpans.flatMap((ss) => ss.spans))); +} + +/** One string attribute off an OTLP/JSON span or event, or undefined. */ +function otlpAttribute(carrier: OtlpSpan | OtlpEvent | undefined, key: string): string | undefined { + return carrier?.attributes?.find((attribute) => attribute.key === key)?.value.stringValue; +} + +/** Every line of an OTLP/JSON file, parsed. */ +function otlpFile(path: string): OtlpRequest[] { + return readFileSync(path, "utf8") + .split("\n") + .filter((line) => line.trim() !== "") + .map((line) => JSON.parse(line) as OtlpRequest); +} + +describe("createOtelTraceExporter -- what one exchange becomes", () => { + it("names the span from the live mapping and carries every required attribute the wire supplied", async () => { + const { exporter, spans } = memoryExporter(); + const sessionId = crypto.randomUUID(); + const requestId = crypto.randomUUID(); + const request = toolCallEnvelope("rm -rf /", { id: 7, requestId, sessionId }); + + const step = exporter.start(request, "steps/toolCallRequest"); + exporter.end(step, decisionResponse(7, DENY)); + await exporter.flush(); + + const [span] = spans(); + expect(spans()).toHaveLength(1); + expect(span?.name).toBe(TOOL_CALL_SPAN.span_name); + for (const attribute of TOOL_CALL_SPAN.required_attributes) { + expect({ attribute, present: span?.attributes[attribute] !== undefined }).toEqual({ attribute, present: true }); + } + expect(span?.attributes["gen_ai.tool.name"]).toBe("run_shell"); + expect(span?.attributes["acs.session.id"]).toBe(sessionId); + expect(span?.attributes["acs.request_id"]).toBe(requestId); + expect(span?.resource.attributes["service.name"]).toBe(OTEL_SERVICE_NAME); + }); + + it("records the decision as the mapping's event on that span, with acs.decision and no evaluator the response did not send", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end(exporter.start(toolCallEnvelope("rm -rf /", { id: 7 }), "steps/toolCallRequest"), decisionResponse(7, DENY)); + await exporter.flush(); + + const [event] = spans()[0]?.events ?? []; + expect(event?.name).toBe(MAPPING.decisionEventName); + expect(event?.attributes?.["acs.decision"]).toBe("deny"); + // The response carried no metadata.evaluator, so the event carries no + // acs.evaluator: the trace never says what the wire did not. + expect(event?.attributes?.["acs.evaluator"]).toBeUndefined(); + expect(event?.attributes?.["acs.reasoning"]).toBe("blocked"); + expect(event?.attributes?.["acs.reason_codes"]).toEqual(["dangerous_command"]); + expect(event?.attributes?.["acs.policy.ids"]).toEqual(["agt_stock"]); + expect(event?.attributes?.["acs.policy.rule_ids"]).toEqual(["deny-rm-rf"]); + // The reference carried no version: the column is still there, one "" + // per reference, so the three columns always line up (#228). + expect(event?.attributes?.["acs.policy.versions"]).toEqual([""]); + }); + + it("takes acs.evaluator from the response when it carries one", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end( + exporter.start(toolCallEnvelope("ls", { id: 7 }), "steps/toolCallRequest"), + decisionResponse(7, { decision: "allow", metadata: { evaluator: "agent", model_id: "m-1" } }), + ); + await exporter.flush(); + + const attributes = spans()[0]?.events[0]?.attributes; + expect(attributes?.["acs.evaluator"]).toBe("agent"); + expect(attributes?.["acs.model_id"]).toBe("m-1"); + }); + + it("lines up policy columns in wire order, with \"\" where a reference lacks the field", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end( + exporter.start(toolCallEnvelope("ls", { id: 7 }), "steps/toolCallRequest"), + decisionResponse(7, { + decision: "deny", + reasoning: "x", + policy_references: [ + { policy_id: "p1", policy_version: "v1", rule_id: "r1" }, + { policy_id: "p2", rule_id: "r2" }, + ], + }), + ); + await exporter.flush(); + + const attributes = spans()[0]?.events[0]?.attributes; + expect(attributes?.["acs.policy.ids"]).toEqual(["p1", "p2"]); + expect(attributes?.["acs.policy.rule_ids"]).toEqual(["r1", "r2"]); + expect(attributes?.["acs.policy.versions"]).toEqual(["v1", ""]); + }); + + it("pads a missing rule_id with \"\" in its own position", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end( + exporter.start(toolCallEnvelope("ls", { id: 7 }), "steps/toolCallRequest"), + decisionResponse(7, { + decision: "deny", + reasoning: "x", + policy_references: [ + { policy_id: "p1", rule_id: "r1" }, + { policy_id: "p2" }, + { policy_id: "p3", rule_id: "r3" }, + ], + }), + ); + await exporter.flush(); + + const attributes = spans()[0]?.events[0]?.attributes; + expect(attributes?.["acs.policy.rule_ids"]).toEqual(["r1", "", "r3"]); + }); + + // The #228 rule, as the maintainer's own case states it: every policy + // column is emitted whenever policy_references is non-empty, all the same + // length, "" where a reference has no value. + it("emits all three policy columns at the length of policy_references", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end( + exporter.start(toolCallEnvelope("ls", { id: 7 }), "steps/toolCallRequest"), + decisionResponse(7, { + decision: "deny", + reasoning: "x", + policy_references: [{ policy_id: "a", rule_id: "r" }, { policy_id: "b" }], + }), + ); + await exporter.flush(); + + const attributes = spans()[0]?.events[0]?.attributes; + expect(attributes?.["acs.policy.ids"]).toEqual(["a", "b"]); + expect(attributes?.["acs.policy.rule_ids"]).toEqual(["r", ""]); + expect(attributes?.["acs.policy.versions"]).toEqual(["", ""]); + }); + + it("carries acs.chain_hash only when the response does", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 1 }), "steps/toolCallRequest"), decisionResponse(1, { decision: "allow", chain_hash: "a".repeat(64) })); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 2 }), "steps/toolCallRequest"), decisionResponse(2, { decision: "allow" })); + await exporter.flush(); + + const [withHash, withoutHash] = spans(); + expect(withHash?.events[0]?.attributes?.["acs.chain_hash"]).toBe("a".repeat(64)); + expect(withoutHash?.events[0]?.attributes?.["acs.chain_hash"]).toBeUndefined(); + }); + + it("emits no span for a method the mapping does not name, nor for one this Guardian does not dispatch", async () => { + const { exporter, spans } = memoryExporter(); + // handshake/hello has no step_to_span entry. steps/sessionStart has one, + // and is not in the ServerHello's methods_evaluated. + const hello = exporter.start(makeEnvelope("handshake/hello", {}, { id: 1 }), "handshake/hello"); + const sessionStart = exporter.start(makeEnvelope("steps/sessionStart", {}, { id: 2 }), "steps/sessionStart"); + expect(hello).toBeNull(); + expect(sessionStart).toBeNull(); + exporter.end(hello, { jsonrpc: "2.0", id: 1, result: { negotiated_version: "0.1.0" } }); + exporter.end(sessionStart, { jsonrpc: "2.0", id: 2, error: { code: -32011, message: "not dispatched" } }); + await exporter.flush(); + + expect(spans()).toHaveLength(0); + }); + + // The first of the two failures a Guardian can see itself (#37): it + // answered an error instead of a decision. No acs.decision event, since + // there was no decision; one acs.decision.failure event, with the code. + it("ends a span answered with a JSON-RPC error in error status, with a guardian_error failure event and no decision event", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end( + exporter.start(toolCallEnvelope("ls", { id: 9 }), "steps/toolCallRequest"), + { jsonrpc: "2.0", id: 9, error: { code: -32020, message: "evaluation failed" } }, + ); + await exporter.flush(); + + const [span] = spans(); + expect(span?.status.code).toBe(SpanStatusCode.ERROR); + expect(span?.status.message).toBe(FAILURE_DETAIL_BY_KIND.error_without_decision); + expect(span?.events.map((event) => event.name)).toEqual(["acs.decision.failure"]); + const failure = span?.events[0]?.attributes; + expect(failure?.["acs.failure.class"]).toBe("guardian_error"); + expect(failure?.["acs.failure.error_code"]).toBe(-32020); + expect(failure?.["acs.failure.detail"]).toBeUndefined(); + expect(failure?.["acs.failure.reported_by"]).toBe("guardian"); + expect(failure?.["acs.failure.posture"]).toBe("fail_open"); + expect(typeof failure?.["acs.failure.elapsed_ms"]).toBe("number"); + }); + + // The pairing is the handle, not the id: an envelope with no JSON-RPC id + // at all still gets its span, closed by its own response. + it("pairs request and response by handle, so a request with no JSON-RPC id still gets its span", async () => { + const { exporter, spans } = memoryExporter(); + const step = exporter.start(toolCallEnvelope("ls", { id: null }), "steps/toolCallRequest"); + expect(step).not.toBeNull(); + exporter.end(step, { jsonrpc: "2.0", id: null, error: { code: -32010, message: "invalid" } }); + await exporter.flush(); + + expect(spans()).toHaveLength(1); + expect(spans()[0]?.status.code).toBe(SpanStatusCode.ERROR); + }); + + // The whole reason the exporter is total: a broken output must cost + // observability, never a decision. + it("never throws when the span exporter throws, reports once, and goes quiet", async () => { + const errors: unknown[] = []; + const exporter = createOtelTraceExporter({ + spanExporter: throwingSpanExporter(), + methodsEvaluated: METHODS_EVALUATED, + onError: (error) => errors.push(error), + }); + + const step = exporter.start(toolCallEnvelope("ls", { id: 1 }), "steps/toolCallRequest"); + expect(() => exporter.end(step, decisionResponse(1, DENY))).not.toThrow(); + await exporter.flush(); + expect(() => exporter.start(toolCallEnvelope("ls", { id: 2 }), "steps/toolCallRequest")).not.toThrow(); + await exporter.flush(); + await exporter.shutdown(); + + expect(errors).toHaveLength(1); + }); + + it("never throws when the file path is unwritable, and reports once", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-broken-")); + const blocker = join(dir, "blocker"); + await Bun.write(blocker, ""); + const errors: unknown[] = []; + try { + // A path *through* a regular file: mkdirSync fails with ENOTDIR, + // deterministically, on every platform. + const exporter = createOtelTraceExporter({ + filePath: join(blocker, "nested", "traces.jsonl"), + methodsEvaluated: METHODS_EVALUATED, + onError: (error) => errors.push(error), + }); + expect(exporter.start(toolCallEnvelope("ls", { id: 1 }), "steps/toolCallRequest")).toBeNull(); + await exporter.shutdown(); + expect(errors).toHaveLength(1); + } finally { + unlinkSync(blocker); + rmdirSync(dir); + } + }); + + // The lab reads the file. A collector that is down, restarting or + // answering 4xx must cost the endpoint and nothing else: every span still + // reaches the file, and the endpoint is reported once. + it("keeps writing the file when the HTTP collector rejects every export, and reports that output once", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-isolated-")); + const path = join(dir, "traces.jsonl"); + const collector = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: () => new Response("bad request", { status: 400 }) }); + const errors: unknown[] = []; + const exporter = createOtelTraceExporter({ + filePath: path, + endpoint: `http://127.0.0.1:${collector.port}/v1/traces`, + methodsEvaluated: METHODS_EVALUATED, + onError: (error) => errors.push(error), + }); + try { + for (const id of [1, 2, 3]) { + exporter.end(exporter.start(toolCallEnvelope("ls", { id }), "steps/toolCallRequest"), decisionResponse(id, { decision: "allow" })); + await exporter.flush(); + } + await exporter.shutdown(); + + expect(otlpSpans(otlpFile(path)).map((span) => span.name)).toEqual([ + TOOL_CALL_SPAN.span_name, + TOOL_CALL_SPAN.span_name, + TOOL_CALL_SPAN.span_name, + ]); + expect(errors).toHaveLength(1); + } finally { + collector.stop(true); + unlinkSync(path); + rmdirSync(dir); + } + }); + + it("sends one OTLP/JSON request over HTTP whose span is named by the mapping", async () => { + const bodies: OtlpRequest[] = []; + const collector = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(req) { + bodies.push((await req.json()) as OtlpRequest); + return new Response("{}", { status: 200, headers: { "content-type": "application/json" } }); + }, + }); + const exporter = createOtelTraceExporter({ endpoint: `http://127.0.0.1:${collector.port}/v1/traces`, methodsEvaluated: METHODS_EVALUATED }); + try { + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 1 }), "steps/toolCallRequest"), decisionResponse(1, { decision: "allow" })); + await exporter.flush(); + await exporter.shutdown(); + + expect(bodies).toHaveLength(1); + expect(otlpSpans(bodies).map((span) => span.name)).toEqual([TOOL_CALL_SPAN.span_name]); + } finally { + collector.stop(true); + } + }); + + it("carries provenance facts as one column per attribute, one element per Provenance object, in argument order", async () => { + const { exporter, spans } = memoryExporter(); + const request = makeEnvelope( + "steps/toolCallRequest", + { + tool: { name: "run_shell" }, + capability: "process.exec", + arguments: { + command: { value: "ls", provenance: provenance("user_input", { source_id: "user-1" }) }, + cwd: { value: "/tmp", provenance: provenance("tool_output", { source_id: "run_shell" }) }, + // No provenance: contributes no element to any column. + flags: { value: "-la" }, + }, + }, + { id: 1 }, + ); + exporter.end(exporter.start(request, "steps/toolCallRequest"), decisionResponse(1, { decision: "allow" })); + await exporter.flush(); + + const attributes = spans()[0]?.attributes; + expect(attributes?.["acs.provenance.origin"]).toEqual(["user_input", "tool_output"]); + expect(attributes?.["acs.provenance.source_id"]).toEqual(["user-1", "run_shell"]); + }); + + it("pads acs.provenance.source_id to the length of acs.provenance.origin with \"\"", async () => { + const { exporter, spans } = memoryExporter(); + const request = makeEnvelope( + "steps/toolCallRequest", + { + tool: { name: "run_shell" }, + arguments: { + command: { value: "ls", provenance: provenance("user_input", { source_id: "user-1" }) }, + cwd: { value: "/tmp", provenance: provenance("system") }, + }, + }, + { id: 1 }, + ); + exporter.end(exporter.start(request, "steps/toolCallRequest"), decisionResponse(1, { decision: "allow" })); + await exporter.flush(); + + const attributes = spans()[0]?.attributes; + expect(attributes?.["acs.provenance.origin"]).toEqual(["user_input", "system"]); + expect(attributes?.["acs.provenance.source_id"]).toEqual(["user-1", ""]); + }); + + it("carries no provenance attribute when nothing in the payload has one", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 1 }), "steps/toolCallRequest"), decisionResponse(1, { decision: "allow" })); + await exporter.flush(); + + const attributes = spans()[0]?.attributes ?? {}; + expect(Object.keys(attributes).filter((key) => key.startsWith("acs.provenance."))).toEqual([]); + }); + + it("reads output provenance on the result gate the same way, in array order", async () => { + const { exporter, spans } = memoryExporter(); + const request = makeEnvelope( + "steps/toolCallResult", + { + tool: { name: "Bash" }, + exit_status: "success", + outputs: [{ value: "a", provenance: provenance("tool_output", { source_id: "Bash" }) }, { value: "b" }], + }, + { id: 1 }, + ); + exporter.end(exporter.start(request, "steps/toolCallResult"), decisionResponse(1, { decision: "allow" })); + await exporter.flush(); + + const [span] = spans(); + expect(span?.name).toBe(MAPPING.stepToSpan["steps/toolCallResult"]!.span_name); + expect(span?.attributes["acs.provenance.origin"]).toEqual(["tool_output"]); + expect(span?.attributes["acs.provenance.source_id"]).toEqual(["Bash"]); + }); + + // The file exporter on its own: two exports handed over back to back, + // without waiting between them, land as two lines in export order. The + // append is asynchronous, so order is a property the chain has to keep. + it("writes two exports as two lines, in export order", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-file-order-")); + const path = join(dir, "traces.jsonl"); + const { exporter, spans } = memoryExporter(); + const first = crypto.randomUUID(); + const second = crypto.randomUUID(); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 1, requestId: first }), "steps/toolCallRequest"), decisionResponse(1, { decision: "allow" })); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 2, requestId: second }), "steps/toolCallRequest"), decisionResponse(2, { decision: "allow" })); + await exporter.flush(); + const [spanA, spanB] = spans(); + try { + const file = createFileSpanExporter(path); + const results = await Promise.all([ + new Promise((resolve) => file.export([spanA!], (result) => resolve(result.code))), + new Promise((resolve) => file.export([spanB!], (result) => resolve(result.code))), + ]); + await file.shutdown(); + + expect(results).toEqual([0, 0]); + const lines = otlpFile(path).map((request) => otlpSpans([request]).map((span) => otlpAttribute(span, "acs.request_id"))); + expect(lines).toEqual([[first], [second]]); + } finally { + await exporter.shutdown(); + unlinkSync(path); + rmdirSync(dir); + } + }); + + // The file holds reasoning text and provenance source ids (#71), so it is + // created readable by its owner only, and so is the directory the + // exporter makes for it. + it("creates the file 0600 and its directory 0700", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-file-mode-")); + const nested = join(dir, "nested"); + const path = join(nested, "traces.jsonl"); + const exporter = createOtelTraceExporter({ filePath: path, methodsEvaluated: METHODS_EVALUATED }); + try { + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 1 }), "steps/toolCallRequest"), decisionResponse(1, { decision: "allow" })); + await exporter.shutdown(); + + expect((statSync(nested).mode & 0o777).toString(8)).toBe("700"); + expect((statSync(path).mode & 0o777).toString(8)).toBe("600"); + } finally { + unlinkSync(path); + rmdirSync(nested); + rmdirSync(dir); + } + }); + + // SIGINT followed by SIGTERM is two calls to close(). The second must + // wait for the first's flush to finish, not return at once and let the + // process exit in the middle of it. + it("makes a second shutdown() call wait for the first", async () => { + let shutDown = false; + const slow: SpanExporter = { + export(_spans, resultCallback): void { + resultCallback({ code: 0 }); + }, + async shutdown(): Promise { + await Bun.sleep(150); + shutDown = true; + }, + }; + const exporter = createOtelTraceExporter({ spanExporter: slow, methodsEvaluated: METHODS_EVALUATED }); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 1 }), "steps/toolCallRequest"), decisionResponse(1, { decision: "allow" })); + + const first = exporter.shutdown(); + const second = exporter.shutdown(); + await second; + + expect(shutDown).toBe(true); + await first; + }); + + // #227 moves acs.capability into a per-span conditional_attributes list. + // A small mapping of this test's own proves the exporter reads that list; + // the real-mapping test below proves the sources cover whatever the real + // file lists. + it("emits an attribute a step_to_span entry lists under conditional_attributes", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-conditional-")); + const mappingPath = join(dir, "otel-mapping.json"); + writeFileSync( + mappingPath, + JSON.stringify({ + properties: { + step_to_span: { + default: { + "steps/toolCallRequest": { + span_name: "test.tool.call", + required_attributes: ["gen_ai.tool.name"], + conditional_attributes: ["acs.capability"], + }, + }, + }, + decision_event: { + properties: { + event_name: { const: "acs.decision" }, + required_attributes: { default: ["acs.decision"] }, + conditional_attributes: { default: [] }, + }, + }, + provenance_attributes: { properties: { required: { default: [] }, optional: { default: [] } } }, + }, + }), + ); + try { + const { exporter, spans } = memoryExporter({ mappingPath }); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 1 }), "steps/toolCallRequest"), decisionResponse(1, { decision: "allow" })); + await exporter.flush(); + + const [span] = spans(); + expect(span?.name).toBe("test.tool.call"); + expect(span?.attributes["gen_ai.tool.name"]).toBe("run_shell"); + expect(span?.attributes["acs.capability"]).toBe("process.exec"); + } finally { + unlinkSync(mappingPath); + rmdirSync(dir); + } + }); + + // Against the real mapping, not a fixture: a name the mapping lists for + // an evaluated method, in any of its three lists, must have a source here, + // or that attribute quietly stops appearing when the mapping changes -- + // which is what #227's conditional_attributes would have done. + it("has a source for every attribute the real mapping lists on an evaluated method, and for every decision attribute", () => { + for (const method of METHODS_EVALUATED) { + const entry = MAPPING.stepToSpan[method]; + if (entry === undefined) { + continue; + } + const listed = [...entry.required_attributes, ...(entry.conditional_attributes ?? []), ...(entry.optional_attributes ?? [])]; + for (const attribute of listed) { + expect({ method, attribute, known: attribute in STEP_ATTRIBUTE_SOURCES }).toEqual({ method, attribute, known: true }); + } + } + for (const attribute of MAPPING.decisionAttributes) { + expect({ attribute, known: attribute in DECISION_ATTRIBUTE_SOURCES }).toEqual({ attribute, known: true }); + } + }); + + it("maps every method the ServerHello evaluates, or names it in METHODS_WITHOUT_SPAN with a reason", () => { + for (const method of METHODS_EVALUATED) { + const mapped = MAPPING.stepToSpan[method] !== undefined; + const listed = typeof METHODS_WITHOUT_SPAN[method] === "string" && METHODS_WITHOUT_SPAN[method].length > 0; + expect({ method, mappedOrListed: mapped || listed }).toEqual({ method, mappedOrListed: true }); + } + }); + + it("NULL_OTEL_TRACE_EXPORTER writes nothing and never throws", async () => { + expect(NULL_OTEL_TRACE_EXPORTER.start(toolCallEnvelope("ls"), "steps/toolCallRequest")).toBeNull(); + expect(() => NULL_OTEL_TRACE_EXPORTER.end(null, decisionResponse(1, DENY))).not.toThrow(); + await NULL_OTEL_TRACE_EXPORTER.flush(); + await NULL_OTEL_TRACE_EXPORTER.shutdown(); + }); +}); + +describe("Guardian trace export wiring", () => { + it("emits one span per decided step, carrying the session and the decision event", async () => { + const { exporter, spans } = memoryExporter(); + const guardian = await startGuardian({ ...baseOptions, traceExporter: exporter }); + const sessionId = crypto.randomUUID(); + const requestId = crypto.randomUUID(); + try { + await postRaw(guardian.url, toolCallEnvelope("rm -rf /", { id: 11, requestId, sessionId })); + await exporter.flush(); + + const [span] = spans(); + expect(spans()).toHaveLength(1); + expect(span?.name).toBe(TOOL_CALL_SPAN.span_name); + expect(span?.attributes["acs.session.id"]).toBe(sessionId); + expect(span?.attributes["acs.request_id"]).toBe(requestId); + expect(span?.attributes["gen_ai.tool.name"]).toBe("run_shell"); + const [event] = span?.events ?? []; + expect(event?.name).toBe(MAPPING.decisionEventName); + expect(event?.attributes?.["acs.decision"]).toBe("deny"); + // This Guardian sends metadata.evaluator once #229 lands; until then + // the event carries none, rather than one the exporter made up. + expect(event?.attributes?.["acs.evaluator"]).toBeUndefined(); + expect(typeof event?.attributes?.["acs.reasoning"]).toBe("string"); + expect(Array.isArray(event?.attributes?.["acs.reason_codes"])).toBe(true); + expect(Array.isArray(event?.attributes?.["acs.policy.ids"])).toBe(true); + } finally { + await guardian.close(); + } + }); + + // Hosts run hooks in parallel and number their requests independently, so + // two sessions can be in flight with the same JSON-RPC id. Each must end + // its own span with its own decision: the deny for the destructive call, + // the allow for the benign one. + // Every span a real Guardian emits is named by the mapping: one of its + // step_to_span span_name values, never a name this module made up. + it("names every emitted span with a span_name from step_to_span", async () => { + const { exporter, spans } = memoryExporter(); + const guardian = await startGuardian({ ...baseOptions, traceExporter: exporter }); + try { + await postRaw(guardian.url, toolCallEnvelope("ls -la", { id: 1 })); + await postRaw( + guardian.url, + makeEnvelope("steps/toolCallResult", { tool: { name: "Bash" }, exit_status: "success", outputs: [{ value: "ok" }] }, { id: 2 }), + ); + await exporter.flush(); + + const names = new Set(Object.values(MAPPING.stepToSpan).map((entry) => entry.span_name)); + expect(spans()).toHaveLength(2); + for (const span of spans()) { + expect({ span: span.name, mapped: names.has(span.name) }).toEqual({ span: span.name, mapped: true }); + } + } finally { + await guardian.close(); + } + }); + + it("keeps two concurrent sessions apart when they reuse one JSON-RPC id", async () => { + const { exporter, spans } = memoryExporter(); + const guardian = await startGuardian({ ...baseOptions, traceExporter: exporter }); + const denied = { requestId: crypto.randomUUID(), sessionId: crypto.randomUUID() }; + const allowed = { requestId: crypto.randomUUID(), sessionId: crypto.randomUUID() }; + try { + await Promise.all([ + postRaw(guardian.url, toolCallEnvelope("rm -rf /", { id: 1, ...denied })), + postRaw(guardian.url, toolCallEnvelope("ls -la", { id: 1, ...allowed })), + ]); + await exporter.flush(); + + expect(spans()).toHaveLength(2); + const bySession = new Map(spans().map((span) => [span.attributes["acs.session.id"], span])); + const deniedSpan = bySession.get(denied.sessionId); + const allowedSpan = bySession.get(allowed.sessionId); + expect(deniedSpan?.attributes["acs.request_id"]).toBe(denied.requestId); + expect(allowedSpan?.attributes["acs.request_id"]).toBe(allowed.requestId); + expect(deniedSpan?.events).toHaveLength(1); + expect(allowedSpan?.events).toHaveLength(1); + expect(deniedSpan?.events[0]?.attributes?.["acs.decision"]).toBe("deny"); + expect(allowedSpan?.events[0]?.attributes?.["acs.decision"]).toBe("allow"); + } finally { + await guardian.close(); + } + }); + + // `capability` is optional on the wire (#227) and required on the span by + // the mapping. The exporter emits what the envelope supplies and invents + // nothing: no capability, no acs.capability, and the span is otherwise whole. + it("emits a span without acs.capability for a request that carries no capability, and nothing breaks", async () => { + const { exporter, spans } = memoryExporter(); + const guardian = await startGuardian({ ...baseOptions, traceExporter: exporter }); + try { + const response = await postRaw( + guardian.url, + makeEnvelope("steps/toolCallRequest", { tool: { name: "run_shell" }, arguments: { command: { value: "ls -la" } } }, { id: 1 }), + ); + await exporter.flush(); + + expect((response.result as { decision?: string }).decision).toBe("allow"); + const [span] = spans(); + expect(span?.name).toBe(TOOL_CALL_SPAN.span_name); + expect(span?.attributes["gen_ai.tool.name"]).toBe("run_shell"); + expect(span?.attributes["acs.capability"]).toBeUndefined(); + expect(span?.events[0]?.attributes?.["acs.decision"]).toBe("allow"); + } finally { + await guardian.close(); + } + }); + + it("emits no span for handshake/hello or for a method this Guardian does not dispatch", async () => { + const { exporter, spans } = memoryExporter(); + const guardian = await startGuardian({ ...baseOptions, traceExporter: exporter }); + try { + await postRaw(guardian.url, makeEnvelope("handshake/hello", {}, { id: 42 })); + await postRaw(guardian.url, makeEnvelope("steps/sessionStart", {}, { id: 43 })); + await exporter.flush(); + + expect(spans()).toHaveLength(0); + } finally { + await guardian.close(); + } + }); + + // The span opens before validation, so the envelope that fails the schema + // is visible in the trace too -- as a deny, and with no chain hash, because + // the step never reached the chain. + it("gives a schema-invalid steps/* envelope a span with a deny event and no acs.chain_hash", async () => { + const { exporter, spans } = memoryExporter(); + const guardian = await startGuardian({ ...baseOptions, traceExporter: exporter }); + try { + const bad = toolCallEnvelope("rm -rf /", { id: 12 }); + delete (bad.params as Record).acs_version; + await postRaw(guardian.url, bad); + await exporter.flush(); + + const [span] = spans(); + expect(spans()).toHaveLength(1); + const attributes = span?.events[0]?.attributes; + expect(attributes?.["acs.decision"]).toBe("deny"); + expect(attributes?.["acs.reason_codes"]).toEqual(["envelope_invalid"]); + expect(attributes?.["acs.chain_hash"]).toBeUndefined(); + } finally { + await guardian.close(); + } + }); + + it("carries the provenance a real envelope's argument attaches", async () => { + const { exporter, spans } = memoryExporter(); + const guardian = await startGuardian({ ...baseOptions, traceExporter: exporter }); + try { + const envelope = makeEnvelope( + "steps/toolCallRequest", + { + tool: { name: "run_shell" }, + arguments: { command: { value: "ls -la", provenance: provenance("user_input", { source_id: "user-12345" }) } }, + }, + { id: 31 }, + ); + const response = await postRaw(guardian.url, envelope); + await exporter.flush(); + + expect((response.result as { decision?: string }).decision).toBe("allow"); + const [span] = spans(); + expect(span?.attributes["acs.provenance.origin"]).toEqual(["user_input"]); + expect(span?.attributes["acs.provenance.source_id"]).toEqual(["user-12345"]); + } finally { + await guardian.close(); + } + }); + + // End to end: the exporter is on the decision path, so this is the test + // that says a broken exporter cannot change what the host receives. + it("answers identically with a failing exporter and with export off, and reports once", async () => { + const errorSpy = spyOn(console, "error").mockImplementation(() => {}); + const exporter = createOtelTraceExporter({ spanExporter: throwingSpanExporter(), methodsEvaluated: METHODS_EVALUATED }); + const exporting = await startGuardian({ ...baseOptions, traceExporter: exporter }); + const silent = await startGuardian(baseOptions); + try { + const requestId = crypto.randomUUID(); + const sessionId = crypto.randomUUID(); + const envelope = toolCallEnvelope("rm -rf /", { id: 5, requestId, sessionId }); + + const withExporter = await postRaw(exporting.url, envelope); + await exporter.flush(); + const withoutExporter = await postRaw(silent.url, envelope); + + expect(withExporter).toEqual(withoutExporter); + expect((withExporter.result as { decision?: string }).decision).toBe("deny"); + expect(errorSpy).toHaveBeenCalledTimes(1); + } finally { + errorSpy.mockRestore(); + await exporting.close(); + await silent.close(); + } + }); + + it("adds no failure event to a decision that arrives in time", async () => { + const memory = new InMemorySpanExporter(); + const exporter = createOtelTraceExporter({ + spanExporter: memory, + methodsEvaluated: METHODS_EVALUATED, + timeoutConfig: { default_ms: 5000 }, + posture: "fail_open", + }); + const guardian = await startGuardian({ ...baseOptions, traceExporter: exporter }); + try { + await postRaw(guardian.url, toolCallEnvelope("ls -la", { id: 1 })); + await exporter.flush(); + + const [span] = memory.getFinishedSpans(); + expect(span?.events.map((event) => event.name)).toEqual([MAPPING.decisionEventName]); + } finally { + await guardian.close(); + } + }); + + // The second failure a Guardian can see itself: a decision that finished + // after the host stopped waiting. The host applied its posture at the + // timeout, so the decision on the span probably never took effect, and a + // reader has to be told. A 50 ms timeout against a 200 ms bridge keeps + // the test fast and the ordering certain. + it("adds a timeout failure event when the decision finishes after the negotiated timeout", async () => { + const memory = new InMemorySpanExporter(); + const exporter = createOtelTraceExporter({ + spanExporter: memory, + methodsEvaluated: METHODS_EVALUATED, + timeoutConfig: { default_ms: 50 }, + posture: "fail_open", + }); + const slow: PolicyBridge = { + async evaluate() { + await Bun.sleep(200); + return { decision: "allow" }; + }, + }; + const guardian = await startGuardian({ ...baseOptions, bridge: slow, traceExporter: exporter }); + try { + await postRaw(guardian.url, toolCallEnvelope("ls -la", { id: 1 })); + await exporter.flush(); + + const [span] = memory.getFinishedSpans(); + expect(span?.events.map((event) => event.name)).toEqual([MAPPING.decisionEventName, "acs.decision.failure"]); + const failure = span?.events[1]?.attributes; + expect(failure?.["acs.failure.class"]).toBe("timeout"); + expect(failure?.["acs.failure.timeout_ms"]).toBe(50); + expect(failure?.["acs.failure.elapsed_ms"]).toBeGreaterThanOrEqual(200); + expect(failure?.["acs.failure.posture"]).toBe("fail_open"); + expect(failure?.["acs.failure.reported_by"]).toBe("guardian"); + expect(failure?.["acs.failure.payload_bytes"]).toBeGreaterThan(0); + } finally { + await guardian.close(); + } + }); + + // A bridge that never answers leaves the span open, and an open span is + // never exported: the step would leave no trace at all. Shutdown closes + // it with an `other` failure event that says why. close() drains for the + // negotiated timeout (5 s in the ServerHello) before it gives up on the + // handler, which is what bounds this test. + it("closes a step still open at shutdown with an other failure event and ERROR status", async () => { + // The file, not an in-memory exporter: close() shuts the provider down, + // and InMemorySpanExporter forgets its spans when it is shut down. + const dir = mkdtempSync(join(tmpdir(), "acs-otel-open-at-shutdown-")); + const path = join(dir, "traces.jsonl"); + const exporter = createOtelTraceExporter({ filePath: path, methodsEvaluated: METHODS_EVALUATED, posture: "fail_open" }); + let reached: () => void = () => {}; + const reachedBridge = new Promise((resolve) => { + reached = resolve; + }); + const silent: PolicyBridge = { + evaluate() { + reached(); + return new Promise(() => {}); + }, + }; + const guardian = await startGuardian({ ...baseOptions, bridge: silent, traceExporter: exporter }); + const pending = postRaw(guardian.url, toolCallEnvelope("ls -la", { id: 1 })).catch(() => undefined); + try { + await reachedBridge; + const closeStarted = performance.now(); + await guardian.close(); + const closeMs = performance.now() - closeStarted; + // The request is answered by the server going away, never by a + // decision; bounded so a connection that lingers cannot hang the test. + await Promise.race([pending, Bun.sleep(2000)]); + + const spans = otlpSpans(otlpFile(path)); + expect(spans).toHaveLength(1); + expect(spans[0]?.status?.code).toBe(2); + expect(spans[0]?.events?.map((event) => event.name)).toEqual(["acs.decision.failure"]); + const failure = spans[0]?.events?.[0]; + expect(otlpAttribute(failure, "acs.failure.class")).toBe("other"); + expect(typeof otlpAttribute(failure, "acs.failure.detail")).toBe("string"); + expect(otlpAttribute(failure, "acs.failure.reported_by")).toBe("guardian"); + expect(closeMs).toBeLessThan(10000); + } finally { + unlinkSync(path); + rmdirSync(dir); + } + }, 15_000); + + // server.ts builds the exporter from the ServerHello, so the posture on a + // failure event is the one this Guardian declared to the host. An envelope + // with neither an id nor a request_id earns a deny that cannot be + // addressed to it, so the Guardian answers a bare JSON-RPC error: a + // guardian_error on the span, through the file path that production uses. + it("reports the ServerHello's posture on a failure event: a Guardian declaring deny says fail_closed", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-posture-")); + const path = join(dir, "traces.jsonl"); + const guardian = await startGuardian({ ...baseOptions, onDecisionFailure: "deny", otelTraceFile: path }); + try { + const response = await postRaw(guardian.url, { jsonrpc: "2.0", method: "steps/toolCallRequest", params: {} }); + expect(response.error).toBeDefined(); + } finally { + await guardian.close(); + } + try { + const [span] = otlpSpans(otlpFile(path)); + const failure = span?.events?.find((event) => event.name === "acs.decision.failure"); + expect(otlpAttribute(failure, "acs.failure.class")).toBe("guardian_error"); + expect(otlpAttribute(failure, "acs.failure.posture")).toBe("fail_closed"); + expect(otlpAttribute(failure, "acs.failure.reported_by")).toBe("guardian"); + } finally { + unlinkSync(path); + rmdirSync(dir); + } + }); + + // A collector that is down must cost the endpoint and nothing else. The + // real OTLPTraceExporter against a closed port, not the throwing stub: a + // real collector fails late, after the exporter's retry budget (a refused + // connection is retryable in this exporter), and the stub never walked + // that path. Port 1 is closed on any normal machine. + it("keeps the file running after the HTTP endpoint dies, answers as with export off, and closes in bounded time", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-dead-endpoint-")); + const path = join(dir, "traces.jsonl"); + const errors: unknown[] = []; + const exporter = createOtelTraceExporter({ + filePath: path, + endpoint: "http://127.0.0.1:1/v1/traces", + methodsEvaluated: METHODS_EVALUATED, + onError: (error) => errors.push(error), + }); + const exporting = await startGuardian({ ...baseOptions, traceExporter: exporter }); + const silent = await startGuardian(baseOptions); + let exportingClosed = false; + try { + // Decisions are the same as with export off, before the endpoint dies. + for (const envelope of [toolCallEnvelope("rm -rf /", { id: 1 }), toolCallEnvelope("ls -la", { id: 2 })]) { + expect(await postRaw(exporting.url, envelope)).toEqual(await postRaw(silent.url, envelope)); + } + // flush() returns once every output has answered, and the dead + // endpoint answers only when it gives up -- so this is where its + // death is reported, and how long that took. + const flushStarted = performance.now(); + await exporter.flush(); + const flushMs = performance.now() - flushStarted; + expect(errors).toHaveLength(1); + + // And the same again, after the endpoint died. + const after = [crypto.randomUUID(), crypto.randomUUID()]; + for (const [index, requestId] of after.entries()) { + const envelope = toolCallEnvelope("ls -la", { id: 10 + index, requestId }); + expect(await postRaw(exporting.url, envelope)).toEqual(await postRaw(silent.url, envelope)); + } + const closeStarted = performance.now(); + await exporting.close(); + exportingClosed = true; + const closeMs = performance.now() - closeStarted; + + // The file holds every span, the two sent after the endpoint failed + // included -- the assertion that fails when one dead output stops the + // others. + const requestIds = otlpSpans(otlpFile(path)).map((span) => otlpAttribute(span, "acs.request_id")); + expect(requestIds).toHaveLength(4); + for (const requestId of after) { + expect(requestIds).toContain(requestId); + } + expect(errors).toHaveLength(1); + // Bounded: the endpoint gives up within its own short timeout, not the + // exporter's 10 s default, and closing never waits that budget again. + expect(flushMs).toBeLessThan(5000); + expect(closeMs).toBeLessThan(5000); + } finally { + if (!exportingClosed) { + await exporting.close(); + } + await silent.close(); + unlinkSync(path); + rmdirSync(dir); + } + }, 20_000); + + it("writes nothing when neither output is configured", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-off-")); + const path = join(dir, "traces.jsonl"); + const guardian = await startGuardian(baseOptions); + try { + await postRaw(guardian.url, toolCallEnvelope("ls -la")); + expect(existsSync(path)).toBe(false); + } finally { + await guardian.close(); + rmdirSync(dir); + } + }); + + it("flushes the file on close: every line is OTLP/JSON holding the expected spans", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-otel-file-")); + const path = join(dir, "traces.jsonl"); + const guardian = await startGuardian({ ...baseOptions, otelTraceFile: path }); + try { + await postRaw(guardian.url, toolCallEnvelope("rm -rf /", { id: 21 })); + await postRaw(guardian.url, toolCallEnvelope("ls -la", { id: 22 })); + } finally { + // The spans are batched; close is what flushes them. + await guardian.close(); + } + try { + expect(existsSync(path)).toBe(true); + const requests = otlpFile(path); + expect(requests.length).toBeGreaterThan(0); + const spans = otlpSpans(requests); + expect(spans.map((s) => s.name)).toEqual([TOOL_CALL_SPAN.span_name, TOOL_CALL_SPAN.span_name]); + expect(spans.every((s) => s.events?.some((e) => e.name === MAPPING.decisionEventName))).toBe(true); + const serviceName = requests[0]?.resourceSpans[0]?.resource.attributes.find((a) => a.key === "service.name"); + expect(serviceName?.value.stringValue).toBe(OTEL_SERVICE_NAME); + } finally { + unlinkSync(path); + rmdirSync(dir); + } + }); +}); + +// Review fixes on PR #244. Each test names the finding it pins. +describe("createOtelTraceExporter -- review fixes", () => { + const SENTINEL = "SENTINEL-/Users/alice/.secrets/token"; + + // #228: acs.failure.detail MUST be text the emitter composes itself, and + // never bytes copied from a response or an error. The span status is held + // to the same rule, since it lands in the same backend. + it("copies no JSON-RPC error text into the failure event or the span status, and keeps the code", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 3 }), "steps/toolCallRequest"), { + jsonrpc: "2.0", + id: 3, + error: { code: -32020, message: `evaluation failed: ${SENTINEL}` }, + }); + await exporter.flush(); + + const [span] = spans(); + expect(JSON.stringify(span?.events)).not.toContain("SENTINEL"); + expect(span?.status.message ?? "").not.toContain("SENTINEL"); + expect(span?.events[0]?.attributes?.["acs.failure.error_code"]).toBe(-32020); + expect(span?.events[0]?.attributes?.["acs.failure.detail"]).toBeUndefined(); + }); + + // #228: a Guardian MUST NOT emit the failure event for a request it rejects + // as a replay, because that pair belongs to the earlier request. + it("emits no failure event for REPLAY_DETECTED (-32005)", async () => { + const { exporter, spans } = memoryExporter(); + exporter.end(exporter.start(toolCallEnvelope("ls", { id: 4 }), "steps/toolCallRequest"), { + jsonrpc: "2.0", + id: 4, + error: { code: -32005, message: "replay" }, + }); + await exporter.flush(); + + expect(spans()[0]?.events.map((event) => event.name)).toEqual([]); + }); + + // #228: acs.provenance.source_id is emitted only when at least one attached + // Provenance object has a source_id. + it("omits acs.provenance.source_id when no Provenance object carries one", async () => { + const { exporter, spans } = memoryExporter(); + const envelope = makeEnvelope("steps/toolCallRequest", { + tool: { name: "run_shell" }, + arguments: { a: { value: "x", provenance: provenance("tool_output") }, b: { value: "y", provenance: provenance("user_input") } }, + }); + exporter.end(exporter.start(envelope, "steps/toolCallRequest"), decisionResponse(1, DENY)); + await exporter.flush(); + + const attributes = spans()[0]?.attributes; + expect(attributes?.["acs.provenance.origin"]).toEqual(["tool_output", "user_input"]); + expect(attributes?.["acs.provenance.source_id"]).toBeUndefined(); + }); + + // Unbounded attribute values let one large request grow the Guardian's + // memory without limit; a crash then fails open under the default posture. + // Driven through start() directly, because the server's body cap would + // refuse a request this large before any span existed. + it("caps attribute values at 4096 characters and provenance columns at 64 entries", async () => { + const { exporter, spans } = memoryExporter(); + const args: Record = {}; + for (let i = 0; i < 200; i += 1) { + args[`a${i}`] = { value: i, provenance: provenance("tool_output", { source_id: `s${i}` }) }; + } + const envelope = makeEnvelope( + "steps/toolCallRequest", + { tool: { name: "run_shell" }, arguments: args }, + { sessionId: "x".repeat(10_000) }, + ); + exporter.end(exporter.start(envelope, "steps/toolCallRequest"), decisionResponse(1, DENY)); + await exporter.flush(); + + expect(spans()).toHaveLength(1); + const attributes = spans()[0]?.attributes; + expect(String(attributes?.["acs.session.id"]).length).toBeLessThanOrEqual(4096); + expect((attributes?.["acs.provenance.origin"] as string[]).length).toBe(64); + expect((attributes?.["acs.provenance.source_id"] as string[]).length).toBe(64); + }); + + // A machine-wide OTEL_TRACES_SAMPLER or OTEL_SPAN_EVENT_COUNT_LIMIT must + // not silently drop the Guardian's spans or its decision events. + it("ignores sampler and event-limit environment variables", async () => { + const saved = { sampler: process.env.OTEL_TRACES_SAMPLER, events: process.env.OTEL_SPAN_EVENT_COUNT_LIMIT }; + process.env.OTEL_TRACES_SAMPLER = "always_off"; + process.env.OTEL_SPAN_EVENT_COUNT_LIMIT = "0"; + try { + const { exporter, spans } = memoryExporter(); + exporter.end(exporter.start(toolCallEnvelope("ls"), "steps/toolCallRequest"), decisionResponse(1, DENY)); + await exporter.flush(); + + expect(spans()).toHaveLength(1); + expect(spans()[0]?.events.map((event) => event.name)).toEqual([MAPPING.decisionEventName]); + } finally { + for (const [key, value] of [["OTEL_TRACES_SAMPLER", saved.sampler], ["OTEL_SPAN_EVENT_COUNT_LIMIT", saved.events]] as const) { + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + } + } + }); + + // A stalled output must not hold shutdown, and with it the process, open. + it("bounds shutdown when an output never finishes, and reports it", async () => { + const stuck: SpanExporter = { + export(): void {}, + shutdown(): Promise { + return new Promise(() => {}); + }, + }; + const errors: unknown[] = []; + const exporter = createOtelTraceExporter({ + spanExporter: stuck, + methodsEvaluated: METHODS_EVALUATED, + exportTimeoutMs: 100, + onError: (error) => errors.push(error), + }); + exporter.end(exporter.start(toolCallEnvelope("ls"), "steps/toolCallRequest"), decisionResponse(1, DENY)); + + const started = performance.now(); + await exporter.shutdown(); + expect(performance.now() - started).toBeLessThan(2000); + expect(errors.length).toBeGreaterThan(0); + }); + + // #228 moves the six decision extras into decision_event's own list. The + // exporter has to emit them under both mappings, so #244 and #228 can land + // in either order. + it("emits the decision extras whether the mapping lists them or not", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-mapping-228-")); + const path = join(dir, "otel-mapping.json"); + try { + const schema = JSON.parse( + readFileSync(new URL("../../../../../specification/v0.1.0/trace/otel-mapping.json", import.meta.url), "utf8"), + ) as { properties: { decision_event: { properties: { conditional_attributes: { default: string[] } } } } }; + const extras = ["acs.reason_codes", "acs.policy.ids", "acs.policy.versions", "acs.policy.rule_ids", "acs.cited_provenance_ids", "acs.chain_hash"]; + const listed = schema.properties.decision_event.properties.conditional_attributes; + listed.default = [...listed.default.filter((name) => !extras.includes(name)), ...extras]; + writeFileSync(path, JSON.stringify(schema)); + + for (const attribute of loadOtelMapping(path).decisionAttributes) { + expect({ attribute, known: attribute in DECISION_ATTRIBUTE_SOURCES }).toEqual({ attribute, known: true }); + } + for (const mappingPath of [undefined, path]) { + const { exporter, spans } = memoryExporter({ mappingPath }); + exporter.end( + exporter.start(toolCallEnvelope("ls"), "steps/toolCallRequest"), + decisionResponse(1, { ...DENY, reason_codes: ["blocked"], chain_hash: "a".repeat(64) }), + ); + await exporter.flush(); + const event = spans()[0]?.events.find((e) => e.name === MAPPING.decisionEventName); + expect(event?.attributes?.["acs.reason_codes"]).toEqual(["blocked"]); + expect(event?.attributes?.["acs.chain_hash"]).toBe("a".repeat(64)); + } + } finally { + unlinkSync(path); + rmdirSync(dir); + } + }); +}); diff --git a/reference-implementations/agt/packages/host-adapter/test/build-envelope.test.ts b/reference-implementations/agt/packages/host-adapter/test/build-envelope.test.ts index c85ad649..de58ac5c 100644 --- a/reference-implementations/agt/packages/host-adapter/test/build-envelope.test.ts +++ b/reference-implementations/agt/packages/host-adapter/test/build-envelope.test.ts @@ -151,6 +151,15 @@ describe("buildEnvelope", () => { expect(envelope.params.metadata.session_id).toBe(realUuid); }); + // The audit log records the envelope's JSON-RPC id as `rpc_id`, and the + // trace converter (guardian/src/audit-to-otlp.ts) carries it as + // acs.request_id to join the host's failure spans to the Guardian's. That + // join holds only while the two ids are the same value. + it("uses the request_id as the JSON-RPC id", () => { + const envelope = buildEnvelope("PreToolUse", preToolUsePayload, hookmap); + expect(envelope.id).toBe(envelope.params.request_id); + }); + it("toSessionUuid is deterministic and produces a valid uuid for a non-uuid input", () => { expect(toSessionUuid("abc123")).toMatch(UUID_RE); expect(toSessionUuid("abc123")).toBe(toSessionUuid("abc123")); diff --git a/reference-implementations/agt/test/audit-to-otlp.test.ts b/reference-implementations/agt/test/audit-to-otlp.test.ts new file mode 100644 index 00000000..b898ede5 --- /dev/null +++ b/reference-implementations/agt/test/audit-to-otlp.test.ts @@ -0,0 +1,333 @@ +import { describe, expect, it } from "bun:test"; +import { appendFileSync, existsSync, mkdtempSync, readFileSync, rmdirSync, unlinkSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { auditUngovernedStep, createAuditSink, createGuardianClient, governStep, toSessionUuid, type Hookmap } from "host-adapter"; +import { + convertAuditLog, + FAILURE_CLASS_BY_KIND, + FAILURE_DETAIL_BY_KIND, + failureClassOf, + sessionUuid as converterSessionUuid, +} from "../packages/guardian/src/audit-to-otlp.ts"; +import { loadOtelMapping } from "../packages/guardian/src/otel-trace-exporter.ts"; + +/** + * The converter's contract, measured with the Guardian down. The host + * adapter records the failures only it can see (#37: a Guardian it could + * not reach, one that answered too late) in its own audit log, and the + * converter turns that log into the same OTLP/JSON file the Guardian's + * exporter writes. Here, because it uses two packages: the adapter writes + * the log for real, against a port nothing listens on, and the converter + * in the guardian package reads it back. + * + * The same deliberately synthetic hookmap packages/host-adapter/test/govern-step.test.ts + * uses, naming no real host's fields. + */ +const ON_STEP = { + acs_method: "steps/toolCallRequest", + tool_name: "$.tool_name", + arguments: "$.tool_input", + decisions: { + allow: { output: { outcome: { value: "go" }, note: { from: "reasoning", type: "string" } } }, + deny: { output: { outcome: { value: "stop" }, note: { from: "reasoning", type: "string" } } }, + }, +} as const; + +const hookmap: Hookmap = { + host: "test-host", + hooks: { + OnStep: ON_STEP, + // The same gate scoped to one tool, so a step on any other tool is one + // the gate declines to ask about -- an `ungoverned` entry. + OnScopedStep: { ...ON_STEP, tools: ["Bash"] }, + }, +}; + +const payload = { session_id: "sess-1", tool_name: "Bash", tool_input: { command: "ls -la" } }; + +type OtlpAttribute = { key: string; value: { stringValue?: string } }; +type OtlpEvent = { name: string; attributes?: OtlpAttribute[] }; +type OtlpSpan = { name: string; attributes?: OtlpAttribute[]; events?: OtlpEvent[] }; +type OtlpRequest = { resourceSpans: { scopeSpans: { spans: OtlpSpan[] }[] }[] }; + +function spansIn(path: string): OtlpSpan[] { + if (!existsSync(path)) { + return []; + } + return readFileSync(path, "utf8") + .split("\n") + .filter((line) => line.trim() !== "") + .flatMap((line) => (JSON.parse(line) as OtlpRequest).resourceSpans.flatMap((rs) => rs.scopeSpans.flatMap((ss) => ss.spans))); +} + +function attribute(carrier: OtlpSpan | OtlpEvent | undefined, key: string): string | undefined { + return carrier?.attributes?.find((a) => a.key === key)?.value.stringValue; +} + +type AuditLine = { session_id: string; method: string | null; outcome: string; failure?: { kind: string; message: string } }; + +function auditLines(path: string): AuditLine[] { + return readFileSync(path, "utf8") + .split("\n") + .filter((line) => line.trim() !== "") + .map((line) => JSON.parse(line) as AuditLine); +} + +/** Non-recursive cleanup: the files a test can create, by name, then the directory. */ +function withScratch(run: (auditPath: string, otelPath: string) => Promise): Promise { + const dir = mkdtempSync(join(tmpdir(), "acs-audit-to-otlp-")); + const auditPath = join(dir, "audit.jsonl"); + const otelPath = join(dir, "otel.jsonl"); + return run(auditPath, otelPath).finally(() => { + for (const path of [auditPath, otelPath]) { + if (existsSync(path)) { + unlinkSync(path); + } + } + rmdirSync(dir); + }); +} + +describe("audit-to-otlp: the host's decision failures become acs.decision.failure spans", () => { + // Port 1 is closed on any normal machine, so the request never reaches a + // Guardian and the adapter records a `transport` failure resolved by its + // `proceed` posture. That one entry is the converter's whole input here. + it("turns the transport entry a dead Guardian leaves into one unreachable span, reported by the observed agent, fail_open", async () => { + await withScratch(async (auditPath, otelPath) => { + const audit = createAuditSink({ path: auditPath }); + await governStep({ + hookEventName: "OnStep", + payload, + hookmap, + guardian: createGuardianClient("http://127.0.0.1:1/acs"), + session: { config: { timeout_config: { default_ms: 1000 }, on_decision_failure: "proceed" }, failure: undefined }, + sessionId: "sess-1", + audit, + }); + const [entry] = auditLines(auditPath); + expect(entry?.outcome).toBe("proceeded"); + expect(entry?.failure?.kind).toBe("transport"); + + const summary = await convertAuditLog(auditPath, otelPath); + + expect(summary).toEqual({ + converted: 1, + written: 1, + skippedUnmapped: 0, + skippedNoMethod: {}, + skippedReplay: 0, + skippedUngoverned: 0, + skippedUnreadable: 0, + }); + const spans = spansIn(otelPath); + expect(spans).toHaveLength(1); + expect(spans[0]?.name).toBe(loadOtelMapping().stepToSpan["steps/toolCallRequest"]!.span_name); + expect(attribute(spans[0], "acs.session.id")).toBe(toSessionUuid(entry!.session_id)); + const failure = spans[0]?.events?.find((event) => event.name === "acs.decision.failure"); + expect(attribute(failure, "acs.failure.class")).toBe("unreachable"); + expect(attribute(failure, "acs.failure.reported_by")).toBe("observed_agent"); + expect(attribute(failure, "acs.failure.posture")).toBe("fail_open"); + expect(attribute(failure, "acs.failure.detail")).toBe(FAILURE_DETAIL_BY_KIND.transport); + }); + }); + + it("skips ungoverned entries, and counts unmapped methods apart from entries with no method", async () => { + await withScratch(async (auditPath, otelPath) => { + const audit = createAuditSink({ path: auditPath }); + // A step the scoped gate declined to ask about: not a decision failure, + // so none of #37's classes describes it. + auditUngovernedStep({ hookmap, hookEventName: "OnScopedStep", tool: "WebFetch", sessionId: "sess-1", audit }); + // Two posture entries the mapping has no span for: a method with no + // step_to_span entry, and an entry with no method at all. + const recorded_at = new Date().toISOString(); + appendFileSync( + auditPath, + `${JSON.stringify({ seq: 2, recorded_at, session_id: "sess-1", method: "protocols/MCP/tools/call", rpc_id: 2, outcome: "proceeded", posture: "proceed", posture_source: "negotiated", failure: { kind: "timeout", message: "no decision within 1000ms" } })}\n`, + ); + appendFileSync( + auditPath, + `${JSON.stringify({ seq: 3, recorded_at, session_id: "sess-1", method: null, rpc_id: null, outcome: "blocked", posture: "deny", posture_source: "default", failure: { kind: "host_configuration", message: "no hookmap entry for this hook" } })}\n`, + ); + expect(auditLines(auditPath).map((line) => line.outcome)).toEqual(["ungoverned", "proceeded", "blocked"]); + + const summary = await convertAuditLog(auditPath, otelPath); + + expect(summary).toEqual({ + converted: 0, + written: 0, + skippedUnmapped: 1, + skippedNoMethod: { host_configuration: 1 }, + skippedReplay: 0, + skippedUngoverned: 1, + skippedUnreadable: 0, + }); + expect(spansIn(otelPath)).toHaveLength(0); + }); + }); + + it("maps every host failure kind to a class from #37, and anything else to other", () => { + expect(FAILURE_CLASS_BY_KIND).toEqual({ + timeout: "timeout", + transport: "unreachable", + error_without_decision: "guardian_error", + refused: "guardian_error", + host_configuration: "observed_agent_error", + }); + // A decision did arrive, so observed_agent_error does not fit (#244 review). + expect(failureClassOf("decision_unrenderable")).toBe("other"); + expect(failureClassOf("unknown")).toBe("other"); + expect(failureClassOf(undefined)).toBe("other"); + }); +}); + +// Review fixes on PR #244. Each test names the finding it pins. +describe("audit-to-otlp: review fixes", () => { + const SENTINEL = "SENTINEL-ignore-previous-instructions"; + + function entryLine(fields: Record): string { + return `${JSON.stringify({ + seq: 1, + recorded_at: new Date().toISOString(), + session_id: "sess-1", + method: "steps/toolCallRequest", + rpc_id: "3f6c1e2a-8b4d-4c1e-9a7f-2d5b6e8c9a01", + outcome: "proceeded", + posture: "proceed", + posture_source: "negotiated", + failure: { kind: "transport", message: "connect ECONNREFUSED 127.0.0.1:1" }, + ...fields, + })}\n`; + } + + function numberAttribute(carrier: OtlpEvent | undefined, key: string): number | undefined { + const value = (carrier?.attributes?.find((a) => a.key === key)?.value as { intValue?: string | number } | undefined)?.intValue; + return value === undefined ? undefined : Number(value); + } + + // #228: posture is the deployment's, and detail is text the emitter writes. + it("reports a refusal under a proceed posture as guardian_error and fail_open, keeps the code, and copies no message text", async () => { + await withScratch(async (auditPath, otelPath) => { + appendFileSync( + auditPath, + entryLine({ outcome: "blocked", failure: { kind: "refused", message: `guardian refused the envelope with error -32003: ${SENTINEL}` } }), + ); + + await convertAuditLog(auditPath, otelPath); + + const raw = readFileSync(otelPath, "utf8"); + expect(raw).not.toContain("SENTINEL"); + const failure = spansIn(otelPath)[0]?.events?.find((event) => event.name === "acs.decision.failure"); + expect(attribute(failure, "acs.failure.class")).toBe("guardian_error"); + expect(attribute(failure, "acs.failure.posture")).toBe("fail_open"); + expect(numberAttribute(failure, "acs.failure.error_code")).toBe(-32003); + expect(attribute(failure, "acs.failure.detail")).toBe(FAILURE_DETAIL_BY_KIND.refused); + }); + }); + + it("files decision_unrenderable under other, with a detail", async () => { + await withScratch(async (auditPath, otelPath) => { + appendFileSync(auditPath, entryLine({ failure: { kind: "decision_unrenderable", message: `no hookmap entry: ${SENTINEL}` } })); + + await convertAuditLog(auditPath, otelPath); + + const failure = spansIn(otelPath)[0]?.events?.find((event) => event.name === "acs.decision.failure"); + expect(attribute(failure, "acs.failure.class")).toBe("other"); + expect(attribute(failure, "acs.failure.detail")).toBe(FAILURE_DETAIL_BY_KIND.decision_unrenderable); + }); + }); + + // The batch processor's queue used to overflow inside the synchronous + // loop: 5000 entries wrote 2560 spans while the summary claimed 5000. + it("writes every span for a large audit log, and the summary counts what was written", async () => { + await withScratch(async (auditPath, otelPath) => { + let lines = ""; + for (let i = 0; i < 5000; i += 1) { + lines += entryLine({ seq: i + 1, rpc_id: crypto.randomUUID() }); + } + appendFileSync(auditPath, lines); + + const summary = await convertAuditLog(auditPath, otelPath); + + expect(summary.converted).toBe(5000); + expect(summary.written).toBe(5000); + expect(spansIn(otelPath)).toHaveLength(5000); + }); + }); + + // The Guardian's spans carry the derived UUID and the request id; a + // converter span has to carry the same pair, or #228's pair rules can't + // join the two files. + it("carries the same acs.session.id and acs.request_id the Guardian's spans carry", async () => { + await withScratch(async (auditPath, otelPath) => { + const rpcId = crypto.randomUUID(); + appendFileSync(auditPath, entryLine({ rpc_id: rpcId })); + + await convertAuditLog(auditPath, otelPath); + + const [span] = spansIn(otelPath); + expect(attribute(span, "acs.session.id")).toBe(toSessionUuid("sess-1")); + expect(attribute(span, "acs.request_id")).toBe(rpcId); + }); + }); + + it("derives the session UUID exactly as the host adapter does", () => { + for (const raw of ["sess-1", "ses_opencode_abc", "6B1F2C9E-4D7A-4E3B-8A51-9C0D2E7F1A36", "セッション-é", ""]) { + expect({ raw, uuid: converterSessionUuid(raw) }).toEqual({ raw, uuid: toSessionUuid(raw) }); + } + }); + + it("counts an entry with no method by its failure kind, apart from unmapped methods", async () => { + await withScratch(async (auditPath, otelPath) => { + appendFileSync(auditPath, entryLine({ method: null, rpc_id: null, outcome: "blocked", posture: "deny", failure: { kind: "host_configuration", message: "x" } })); + appendFileSync(auditPath, entryLine({ method: "protocols/MCP/tools/call" })); + + const summary = await convertAuditLog(auditPath, otelPath); + + expect(summary.skippedNoMethod).toEqual({ host_configuration: 1 }); + expect(summary.skippedUnmapped).toBe(1); + expect(summary.written).toBe(0); + }); + }); + + // The posture field has been on every posture entry since the first audit + // format, so a missing one is a hand-edited log. Inferring it from outcome + // is the bug this PR fixes, so the entry is skipped and counted instead. + it("skips and counts an entry with no posture rather than guessing one", async () => { + await withScratch(async (auditPath, otelPath) => { + appendFileSync(auditPath, entryLine({ posture: undefined })); + + const summary = await convertAuditLog(auditPath, otelPath); + + expect(summary.skippedUnreadable).toBe(1); + expect(summary.written).toBe(0); + }); + }); + + // #228: no failure event for a replay, because the pair is the earlier request's. + it("skips and counts a REPLAY_DETECTED error", async () => { + await withScratch(async (auditPath, otelPath) => { + appendFileSync(auditPath, entryLine({ outcome: "blocked", failure: { kind: "refused", message: "guardian refused the envelope with error -32005: replay" } })); + + const summary = await convertAuditLog(auditPath, otelPath); + + expect(summary.skippedReplay).toBe(1); + expect(summary.written).toBe(0); + }); + }); + + it("ignores sampler environment variables", async () => { + const saved = process.env.OTEL_TRACES_SAMPLER; + process.env.OTEL_TRACES_SAMPLER = "always_off"; + try { + await withScratch(async (auditPath, otelPath) => { + appendFileSync(auditPath, entryLine({})); + const summary = await convertAuditLog(auditPath, otelPath); + expect(summary.written).toBe(1); + }); + } finally { + if (saved === undefined) delete process.env.OTEL_TRACES_SAMPLER; + else process.env.OTEL_TRACES_SAMPLER = saved; + } + }); +});