diff --git a/README.md b/README.md index 71be2b4..d17576c 100644 --- a/README.md +++ b/README.md @@ -289,9 +289,9 @@ 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 -session-context logs, and the host adapter writes its audit log, as synchronous file -appends on the decision path, with no batching. Trace export is +The Guardian batches its envelope and session-context logs for asynchronous append, +with bounded queues and a drain on graceful shutdown. The host adapter still writes +its audit log synchronously on the decision path. 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) diff --git a/reference-implementations/agt/README.md b/reference-implementations/agt/README.md index cdc6a86..3a3b89c 100644 --- a/reference-implementations/agt/README.md +++ b/reference-implementations/agt/README.md @@ -493,7 +493,7 @@ The suite has 68 test files. Most of them drive a real Guardian, the real host s These are known limits of the tree as it stands. None is fixed. - **Memory and disk grow without bound.** The Guardian's session store never evicts a session. The envelope log, the audit log and the session-context log have no rotation and no size cap. A long-lived Guardian grows until something else stops it. -- **Log writes sit on the decision path.** The Guardian writes two synchronous file appends per request in a single-threaded server. A slow disk blocks every in-flight decision. +- **Logs are buffered.** The Guardian batches envelope and session-context records for asynchronous append. Graceful shutdown drains them within a deadline; a timed-out drain, abrupt termination or a disabled sink can lose queued records. See the [Guardian's logging contract](packages/guardian/README.md#log-batching-and-shutdown). - **The wire is unauthenticated.** See [What this project is, and is not](#what-this-project-is-and-is-not). Loopback by default is the only protection. - **The egress gate has three measured soft spots.** A shell command whose destination the extractor cannot parse falls through to `allow`, not `deny`. With the shipped allowlist, any command whose text contains an off-list URL is denied, whether or not it reaches that URL. A `WebFetch` URL reaches AGT's gate with no parsing, so four userinfo-style URL shapes pass on the fetch route and fail on the shell route. Closing the last one means editing a `.rego` file, which this project does not do. - **An `ask` never carries `ask_details`.** The Guardian's decision type has no such field. Every `ask` fails the response schema. A `defer` would fail the same way if any AGT verdict mapped to it. diff --git a/reference-implementations/agt/packages/guardian/README.md b/reference-implementations/agt/packages/guardian/README.md index 4c604f7..32ca4c9 100644 --- a/reference-implementations/agt/packages/guardian/README.md +++ b/reference-implementations/agt/packages/guardian/README.md @@ -36,7 +36,7 @@ To see a decision without an agent client, pipe a hook payload into the Claude C ## What happens to each request -1. The raw envelope is appended to the envelope log, before anything else. +1. The raw envelope is serialized and queued for the envelope log, before validation. 2. The envelope is validated against `request-envelope.json`. A `steps/toolCallRequest` or `steps/toolCallResult` is also validated against its own payload schema. An invalid envelope on a `steps/*` method is answered with an honoured `deny`, not a bare error, so the host has a decision to act on. 3. The AGT intervention point for the method is resolved from `mapping.yaml`, and so is the argument the policy target is read from for the tool the payload names. 4. The AGT snapshot is assembled, one shape per gate, and evaluated through the bridge. @@ -169,6 +169,7 @@ 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/batched-log-writer.ts` | Bounded asynchronous JSONL batching and shutdown drain | | `src/otel-trace-exporter.ts` | The OpenTelemetry span exporter, assembled from the three files below | | `src/otel-mapping.ts` | Reads `otel-mapping.json`, and says which envelope field feeds each span attribute | | `src/otlp-file-exporter.ts` | The OTLP/JSON file writer | @@ -176,6 +177,37 @@ The package has two entry points. `guardian` exports the governance verbs listed | `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 | +## Log batching and shutdown + +Envelope and session-context logs keep their existing JSONL formats. Each sink releases +a batch after 64 records or 100 ms from the first queued record, then appends it +asynchronously. The timer schedules a flush; it does not guarantee a disk-completion +deadline. The Inspector can therefore see a decision after the HTTP response arrives. +No file is opened until a record arrives. + +New log directories are created with mode `0700`, and new files with mode `0600`, +subject to the process umask on POSIX systems. Existing directories and files keep +their permissions. On Windows, access remains governed by filesystem ACLs. + +Each sink admits at most 4 MiB of serialized UTF-8 data and 4,096 records, counting +writes already in flight. A record that would exceed either limit +is dropped and counted, while queued records are retained and later traffic is accepted +as space becomes available. The first drop and the total at shutdown are reported on +stderr. A real write error disables the sink and reports once; queued records may be lost. Logging +failure does not change the Guardian's policy decision, and the session store remains +authoritative. The host adapter's audit log is a separate writer and is unchanged. + +`await guardian.close()` refuses new ACS requests, waits for active requests, and +drains both logs within one negotiated request timeout (currently five seconds), then +shuts down the trace exporter, whose own deadline is three seconds. Repeated calls +wait for the same shutdown. The standalone Guardian does this on SIGINT and SIGTERM, +with or without trace export. A drain that exceeds its deadline is reported on stderr; +shutdown continues and pending requests or log records may be lost. SIGKILL, a crash, +or power loss can also lose buffered records. Finishing the append does not fsync it. + +This batches the reference implementation's existing logs alongside the trace exporter. +It does not claim ACS-Trace conformance. + ## The wire is not secured The endpoint has no authentication, no origin check and no request signing. It binds loopback by default, so reachability is the only access control. Widen the bind only for a Guardian that runs in its own container, and read the header of `src/server.ts` first. @@ -186,4 +218,4 @@ The endpoint has no authentication, no origin check and no request signing. It b bun test packages/guardian ``` -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`. +Most tests 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/src/batched-log-writer.ts b/reference-implementations/agt/packages/guardian/src/batched-log-writer.ts new file mode 100644 index 0000000..dc823c2 --- /dev/null +++ b/reference-implementations/agt/packages/guardian/src/batched-log-writer.ts @@ -0,0 +1,138 @@ +// SPDX-License-Identifier: Apache-2.0 +import { createWriteStream, mkdirSync } from "node:fs"; +import { dirname } from "node:path"; +import type { Writable } from "node:stream"; +import { finished } from "node:stream/promises"; + +export type BatchedLogWriter = { + readonly droppedRecords: number; + write(line: string): void; + close(): Promise; +}; + +export const NULL_LOG_WRITER: BatchedLogWriter = { + droppedRecords: 0, + write() {}, + async close() {}, +}; + +/** Batches complete JSONL records without waiting for disk on the decision path. + * Both queued and in-flight writes count toward the bounds. Overflow drops + * only the new record; a disk failure still disables the sink. */ +export function batchLogWrites( + stream: Writable, + onError: (error: unknown) => void, + { batchSize = 64, flushIntervalMs = 100, maxBufferedBytes = 4 * 1024 * 1024, maxBufferedRecords = 4096, onDrop = (_count: number) => {} } = {}, +): BatchedLogWriter { + let timer: ReturnType | undefined; + let batchCount = 0; + let pendingRecords = 0; + let closed = false; + let failed = false; + let droppedRecords = 0; + const reportDrops = (): void => { + try { + onDrop(droppedRecords); + } catch { + // Reporting must not break governance. + } + }; + + const clearTimer = (): void => { + clearTimeout(timer); + timer = undefined; + }; + const fail = (error: unknown): void => { + if (failed) return; + failed = true; + clearTimer(); + stream.destroy(); + try { + onError(error); + } catch { + // A reporter must not turn a logging failure into a policy failure. + } + }; + const completion = finished(stream).catch(fail); + + const flush = (): void => { + clearTimer(); + batchCount = 0; + stream.uncork(); + }; + + return { + get droppedRecords() { return droppedRecords; }, + write(line) { + if (closed || failed) return; + try { + const bytes = Buffer.byteLength(line); + if (stream.writableLength + bytes > maxBufferedBytes || pendingRecords >= maxBufferedRecords) { + droppedRecords += 1; + if (droppedRecords === 1) reportDrops(); + return; + } + if (batchCount === 0) stream.cork(); + pendingRecords += 1; + stream.write(line, () => { pendingRecords -= 1; }); + batchCount += 1; + if (batchCount >= batchSize) { + flush(); + } else if (timer === undefined) { + timer = setTimeout(flush, flushIntervalMs); + } + } catch (error) { + fail(error); + } + }, + close() { + if (!closed) { + closed = true; + clearTimer(); + if (droppedRecords > 0) reportDrops(); + // end() uncorks pending records and finishes all outstanding writes. + stream.end(); + } + return completion; + }, + }; +} + +export function createBatchedFileLog(path: string, onError: (error: unknown) => void): BatchedLogWriter { + let writer: BatchedLogWriter | undefined; + let disabled = false; + let closed = false; + const fail = (error: unknown): void => { + if (disabled) return; + disabled = true; + try { + onError(error); + } catch { + // Match asynchronous write failures: reporting cannot break governance. + } + }; + try { + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); + } catch (error) { + fail(error); + } + return { + get droppedRecords() { return writer?.droppedRecords ?? 0; }, + write(line) { + if (closed || disabled) return; + try { + // Keep the existing no-traffic behavior: no file until the first record. + writer ??= batchLogWrites(createWriteStream(path, { flags: "a", mode: 0o600 }), fail, { + onDrop: (count) => console.error(`log buffer overflow (${path}): ${count} records dropped; queued records retained`), + }); + writer.write(line); + } catch (error) { + fail(error); + } + }, + async close() { + closed = true; + await writer?.close(); + }, + }; +} diff --git a/reference-implementations/agt/packages/guardian/src/envelope-log-sink.ts b/reference-implementations/agt/packages/guardian/src/envelope-log-sink.ts index e83d80f..a79590b 100644 --- a/reference-implementations/agt/packages/guardian/src/envelope-log-sink.ts +++ b/reference-implementations/agt/packages/guardian/src/envelope-log-sink.ts @@ -18,6 +18,9 @@ * rather than a JSON value, costing the Inspector its pretty-printing and * the round-trip contract test; the accurate sentence is the better trade. * + * Records are serialized at arrival, then batched for asynchronous append. + * close() drains accepted records before a reader takes a final snapshot. + * * Total by construction. Every write is wrapped: a failure disables the sink * for the process lifetime, reports once, and is never propagated to the * caller. The sink sits on the decision path: an observability feature that @@ -25,8 +28,7 @@ * mode this module must never have. Observability degrades; governance does * not. */ -import { appendFileSync, mkdirSync } from "node:fs"; -import { dirname } from "node:path"; +import { createBatchedFileLog } from "./batched-log-writer.ts"; /** Which side of the exchange one envelope-log line recorded. */ export type EnvelopeLogDirection = "request" | "response"; @@ -53,6 +55,7 @@ export type EnvelopeLogEntry = { export type EnvelopeLogSink = { write(direction: EnvelopeLogDirection, envelope: unknown, method: string | null): void; readonly path: string | null; + close(): Promise; }; export type CreateEnvelopeLogSinkOptions = { @@ -69,6 +72,7 @@ export type CreateEnvelopeLogSinkOptions = { export const NULL_ENVELOPE_LOG_SINK: EnvelopeLogSink = { path: null, write(): void {}, + async close() {}, }; /** The JSON-RPC id, when it is a scalar. Both request and response envelopes @@ -92,6 +96,7 @@ export function createEnvelopeLogSink({ let disabled = false; const fail = (error: unknown): void => { + if (disabled) return; disabled = true; try { if (onError) { @@ -106,14 +111,11 @@ export function createEnvelopeLogSink({ } }; - try { - mkdirSync(dirname(path), { recursive: true }); - } catch (error) { - fail(error); - } + const writer = createBatchedFileLog(path, fail); return { path, + close: () => writer.close(), write(direction, envelope, method): void { if (disabled) { return; @@ -128,7 +130,7 @@ export function createEnvelopeLogSink({ envelope, }; const line = `${JSON.stringify(entry)}\n`; - appendFileSync(path, line); + writer.write(line); seq += 1; } catch (error) { fail(error); diff --git a/reference-implementations/agt/packages/guardian/src/main.ts b/reference-implementations/agt/packages/guardian/src/main.ts index 302c81b..1b51036 100644 --- a/reference-implementations/agt/packages/guardian/src/main.ts +++ b/reference-implementations/agt/packages/guardian/src/main.ts @@ -86,13 +86,11 @@ console.log(`Failure posture: ${posture} (override with ACS_ON_DECISION_FAILUR 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); } + +// Finish active decisions and drain both log streams before a normal shutdown. +const shutdown = (): void => { + void guardian.close().then(() => process.exit(0)); +}; +process.once("SIGINT", shutdown); +process.once("SIGTERM", shutdown); diff --git a/reference-implementations/agt/packages/guardian/src/server.ts b/reference-implementations/agt/packages/guardian/src/server.ts index 922c11b..03cf6e2 100644 --- a/reference-implementations/agt/packages/guardian/src/server.ts +++ b/reference-implementations/agt/packages/guardian/src/server.ts @@ -63,8 +63,6 @@ * -- a Guardian in its own container, say -- opts in explicitly through the * `hostname` option (main.ts reads ACS_GUARDIAN_HOST for it). */ -import { appendFileSync, mkdirSync } from "node:fs"; -import { dirname } from "node:path"; import { fileURLToPath } from "node:url"; import type { Annotator, PolicyBridge } from "agt-bridge"; import { createDeploymentBridge } from "./deployment-bridge.ts"; @@ -92,6 +90,7 @@ import { } from "./validate-envelope.ts"; import { checkResponse } from "./check-response.ts"; import { buildServerHello, type ServerHello } from "./handshake.ts"; +import { createBatchedFileLog, NULL_LOG_WRITER } from "./batched-log-writer.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 { @@ -346,59 +345,6 @@ export type StartGuardianOptions = { }; export type StartedGuardian = { url: string; close(): Promise }; -/** - * A total-by-construction `appendLine` for the session-context chain's - * optional JSONL projection -- shaped like `createEnvelopeLogSink`'s own - * write path (envelope-log-sink.ts: mkdirSync guarded once at construction, - * every write wrapped, disabled and reported once rather than thrown after - * the first failure), but not a call into that function. - * `EnvelopeLogSink.write(direction, envelope, method)` builds its own - * `EnvelopeLogEntry` (seq, recorded_at, direction, method, rpc_id, envelope) - * around whatever it is handed; `SessionContextStore`'s `appendLine` - * contract is one already-serialized JSON line with no wrapping object at - * all (`session-context-store.ts`: "Called once per appended entry with its - * JSON line, no trailing newline"). Routing the chain's lines through - * `createEnvelopeLogSink` would nest every session-context entry inside an - * unrelated `EnvelopeLogEntry` -- `direction: "request"` on a chain entry is - * meaningless, and the JSONL shape `test/session-context-roundtrip.test.ts` - * pins (`{session_id, seq, request_id, ...}` at the line's own top level) - * would break. The failure behaviour is duplicated because the invariant it - * upholds is the same one envelope-log-sink.ts states for the envelope log: - * a projection write must never be able to turn a governed tool call into a - * denied one. - */ -function createSessionContextLogAppender(path: string): (line: string) => void { - let disabled = false; - - const fail = (error: unknown): void => { - disabled = true; - try { - const message = error instanceof Error ? error.message : String(error); - console.error(`session context log disabled after failure (${path}): ${message}`); - } catch { - // Silently swallow any error from console.error, matching - // envelope-log-sink.ts's own guard -- total means total. - } - }; - - try { - mkdirSync(dirname(path), { recursive: true }); - } catch (error) { - fail(error); - } - - return (line: string): void => { - if (disabled) { - return; - } - try { - appendFileSync(path, `${line}\n`); - } catch (error) { - fail(error); - } - }; -} - export async function startGuardian({ port, hostname, @@ -422,19 +368,16 @@ export async function startGuardian({ const bridge = bridgeOverride ?? createDeploymentBridge(manifestPath, annotator); const mapping = loadMapping(mappingPath ?? MAPPING_PATH); const envelopeLog = envelopeLogPath ? createEnvelopeLogSink({ path: envelopeLogPath }) : NULL_ENVELOPE_LOG_SINK; - // The session-context store is always the in-memory one, or the caller's - // own override -- sessionContextLog never becomes an alternative backing - // store, only a JSONL projection of whichever store is in use, the same - // relationship envelopeLogPath has to the envelope log. The `??` below - // means the projection is wired up (and its directory created) only in - // the branch that actually constructs the default store -- an override in - // sessionContextStore short-circuits past both, per that option's own doc - // comment. - const sessionContextStore = - sessionContextStoreOverride ?? - createMemorySessionContextStore( - sessionContextLog ? { appendLine: createSessionContextLogAppender(sessionContextLog) } : {}, - ); + // The JSONL is a projection, not the store. An override owns its persistence. + const contextLog = sessionContextLog && !sessionContextStoreOverride + ? createBatchedFileLog(sessionContextLog, (error) => { + const message = error instanceof Error ? error.message : String(error); + console.error(`session context log disabled after failure (${sessionContextLog}): ${message}`); + }) + : NULL_LOG_WRITER; + const sessionContextStore = sessionContextStoreOverride ?? createMemorySessionContextStore( + sessionContextLog ? { appendLine: (line) => contextLog.write(`${line}\n`) } : {}, + ); // 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. @@ -455,6 +398,9 @@ export async function startGuardian({ }) : NULL_OTEL_TRACE_EXPORTER); + let closing: Promise | undefined; + let activeRequests = 0; + let requestsDrained: (() => void) | undefined; const server = Bun.serve({ hostname: hostname ?? LOOPBACK_ONLY, port, @@ -463,29 +409,61 @@ 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, traceExporter); - return Response.json(response); + if (closing) return new Response("Guardian is shutting down", { status: 503 }); + activeRequests += 1; + try { + const response = await handleAcsRequest(req, bridge, mapping, envelopeLog, onDecisionFailure, sessionContextStore, traceExporter); + return Response.json(response); + } finally { + activeRequests -= 1; + if (activeRequests === 0) requestsDrained?.(); + } }, }); return { url: `http://localhost:${server.port}${ACS_PATH}`, - async close() { - // The same close whether trace export is on or off: - // NULL_OTEL_TRACE_EXPORTER's shutdown resolves at once, so with export - // off the drain is all that remains. Bun.serve's stop(true) closes - // every connection at once, and on Bun 1.3.11 it resolves only when - // every in-flight handler has returned. A handler whose bridge never - // answers never returns, so the drain is 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(); + close() { + return closing ??= (async () => { + // Requests and log writes share the negotiated timeout. A stuck bridge + // or disk must not prevent the exporter from shutting down afterwards. + const pendingRequests = activeRequests === 0 + ? Promise.resolve() + : new Promise((resolve) => { requestsDrained = resolve; }); + let timer: ReturnType | undefined; + let timedOut = false; + const deadline = new Promise((resolve) => { + timer = setTimeout(() => { + timedOut = true; + try { + console.error(`Guardian shutdown drain exceeded ${hello.timeout_config.default_ms} ms; pending requests or log records may be lost`); + } catch { + // A broken stderr must not keep shutdown waiting. + } + resolve(); + }, hello.timeout_config.default_ms).unref(); + }); + try { + await Promise.race([ + (async () => { + // Bun 1.3.11 cannot upgrade an already-started graceful stop to + // stop(true), so wait for handlers before choosing how to stop. + await pendingRequests; + await server.stop(); + await Promise.all([envelopeLog.close(), contextLog.close()]); + })(), + deadline, + ]); + } finally { + clearTimeout(timer); + } + if (timedOut) { + void server.stop(true); + void envelopeLog.close(); + void contextLog.close(); + } + await traceExporter.shutdown(); + })(); }, }; } @@ -610,7 +588,7 @@ async function handleAcsRequest( } catch { // A reporting failure (an EPIPE on stderr, say) must not be able to // throw out of handleAcsRequest with nothing above it to catch it -- - // matching envelope-log-sink.ts / createSessionContextLogAppender's own + // matching envelope-log-sink.ts / createBatchedFileLog's own // guard: total means total, including the reporter that exists only to // report a different total component's own finding. } diff --git a/reference-implementations/agt/packages/guardian/test/batched-log-writer.test.ts b/reference-implementations/agt/packages/guardian/test/batched-log-writer.test.ts new file mode 100644 index 0000000..c4ae317 --- /dev/null +++ b/reference-implementations/agt/packages/guardian/test/batched-log-writer.test.ts @@ -0,0 +1,199 @@ +// SPDX-License-Identifier: Apache-2.0 +import { describe, expect, it } from "bun:test"; +import { Writable } from "node:stream"; +import { batchLogWrites } from "../src/batched-log-writer.ts"; + +function capture() { + const batches: string[][] = []; + const stream = new Writable({ + write(chunk, _encoding, done) { + batches.push([chunk.toString()]); + done(); + }, + writev(chunks, done) { + batches.push(chunks.map(({ chunk }) => chunk.toString())); + done(); + }, + }); + return { stream, batches }; +} + +describe("batched log writes", () => { + it("writes a burst as one batch, preserving Unicode and record order", async () => { + const { stream, batches } = capture(); + const errors: unknown[] = []; + const writer = batchLogWrites(stream, (error) => errors.push(error)); + try { + writer.write('{"id":1,"text":"日本語"}\n'); + writer.write('{"id":2}\n'); + expect(batches).toEqual([]); + await writer.close(); + expect(batches).toEqual([['{"id":1,"text":"日本語"}\n', '{"id":2}\n']]); + expect(errors).toEqual([]); + } finally { + await writer.close(); + } + }); + + it("flushes at the record threshold and drains the remaining batch on close", async () => { + const { stream, batches } = capture(); + const writer = batchLogWrites(stream, () => {}, { batchSize: 2 }); + try { + writer.write("a\n"); + writer.write("b\n"); + expect(batches).toEqual([["a\n", "b\n"]]); + writer.write("c\n"); + await writer.close(); + expect(batches).toEqual([["a\n", "b\n"], ["c\n"]]); + } finally { + await writer.close(); + } + }); + + it("flushes sparse traffic on the timer without another write or shutdown", async () => { + const wrote = Promise.withResolvers(); + const stream = new Writable({ + write(chunk, _encoding, done) { + wrote.resolve(chunk.toString()); + done(); + }, + }); + const writer = batchLogWrites(stream, wrote.reject, { flushIntervalMs: 5 }); + try { + writer.write("one\n"); + expect(await wrote.promise).toBe("one\n"); + } finally { + await writer.close(); + } + }); + + it("keeps later batches behind a pending write and waits for all of them on close", async () => { + const batches: string[][] = []; + let finishFirst: (() => void) | undefined; + const stream = new Writable({ + writev(chunks, done) { + batches.push(chunks.map(({ chunk }) => chunk.toString())); + if (batches.length === 1) finishFirst = done; + else done(); + }, + }); + const writer = batchLogWrites(stream, () => {}, { batchSize: 2 }); + try { + for (const line of ["a\n", "b\n", "c\n", "d\n"]) writer.write(line); + const closing = writer.close(); + let closed = false; + void closing.then(() => { closed = true; }); + await Promise.resolve(); + expect(closed).toBe(false); + expect(batches).toEqual([["a\n", "b\n"]]); + finishFirst?.(); + await closing; + expect(batches.flat()).toEqual(["a\n", "b\n", "c\n", "d\n"]); + expect(closed).toBe(true); + expect(writer.close()).toBe(closing); + writer.write("after close\n"); + expect(batches.flat()).toEqual(["a\n", "b\n", "c\n", "d\n"]); + } finally { + await writer.close(); + } + }); + + it("bounds UTF-8 bytes, including the write already in flight", async () => { + const errors: unknown[] = []; + let finishWrite: (() => void) | undefined; + const stream = new Writable({ write(_chunk, _encoding, done) { finishWrite = done; } }); + const writer = batchLogWrites(stream, (error) => errors.push(error), { + batchSize: 1, + maxBufferedBytes: 4, + }); + writer.write("é\n"); // Three bytes remain in flight. + writer.write("a\n"); + writer.write("b\n"); + expect(errors).toEqual([]); + expect(writer.droppedRecords).toBe(2); + expect(stream.destroyed).toBe(false); + finishWrite?.(); + await writer.close(); + expect(errors).toEqual([]); + }); + + it("bounds the number of queued records even when each is tiny", async () => { + const { stream, batches } = capture(); + const errors: unknown[] = []; + const writer = batchLogWrites(stream, (error) => errors.push(error), { maxBufferedRecords: 2 }); + writer.write("\n"); + writer.write("\n"); + writer.write("\n"); + await writer.close(); + expect(errors).toEqual([]); + expect(writer.droppedRecords).toBe(1); + expect(batches).toEqual([["\n", "\n"]]); + }); + + it("reports an asynchronous disk failure once and absorbs a throwing reporter", async () => { + const errors: unknown[] = []; + const stream = new Writable({ + write(_chunk, _encoding, done) { + queueMicrotask(() => done(new Error("disk full"))); + }, + }); + const writer = batchLogWrites(stream, (error) => { + errors.push(error); + throw new Error("reporter failed too"); + }); + expect(() => writer.write("one\n")).not.toThrow(); + await writer.close(); + expect(() => writer.write("two\n")).not.toThrow(); + expect(errors).toHaveLength(1); + expect(String(errors[0])).toContain("disk full"); + }); +}); + +it("keeps queued records and accepts later traffic after a full buffer", async () => { + const lines: string[] = []; + const second = Promise.withResolvers(); + let release: (() => void) | undefined; + const stream = new Writable({ + write(chunk, _encoding, done) { + lines.push(chunk.toString()); + if (lines.length === 1) release = done; + else { done(); second.resolve(); } + }, + }); + const errors: unknown[] = []; + const writer = batchLogWrites(stream, (error) => errors.push(error), { batchSize: 1, maxBufferedBytes: 4 }); + try { + writer.write("a\n"); + writer.write("b\n"); + writer.write("c\n"); + expect(stream.destroyed).toBe(false); + release?.(); + await second.promise; + writer.write("d\n"); + await writer.close(); + expect(lines).toEqual(["a\n", "b\n", "d\n"]); + expect(writer.droppedRecords).toBe(1); + expect(errors).toEqual([]); + } finally { + release?.(); + stream.destroy(); + await writer.close(); + } +}); + + +it("counts oversized drops, reports a bounded summary, and retains valid records", async () => { + const { stream, batches } = capture(); + const reports: number[] = []; + const writer = batchLogWrites(stream, () => { throw new Error("unexpected disk failure"); }, { + maxBufferedBytes: 2, + onDrop: (count) => { reports.push(count); throw new Error("reporter failed"); }, + }); + writer.write("large\n"); + writer.write("large again\n"); + writer.write("a\n"); + await writer.close(); + expect(writer.droppedRecords).toBe(2); + expect(reports).toEqual([1, 2]); + expect(batches.flat()).toEqual(["a\n"]); +}); diff --git a/reference-implementations/agt/packages/guardian/test/envelope-log-sink-wiring.test.ts b/reference-implementations/agt/packages/guardian/test/envelope-log-sink-wiring.test.ts index f21bbc4..c329fae 100644 --- a/reference-implementations/agt/packages/guardian/test/envelope-log-sink-wiring.test.ts +++ b/reference-implementations/agt/packages/guardian/test/envelope-log-sink-wiring.test.ts @@ -46,13 +46,13 @@ function readEntries(path: string): EnvelopeLogEntry[] { /** Non-recursive cleanup, as in envelope-log-sink.test.ts. */ async function withGuardian( logPathFor: (dir: string) => string, - run: (url: string, logPath: string) => Promise, + run: (url: string, logPath: string, close: () => Promise) => Promise, ): Promise { const dir = mkdtempSync(join(tmpdir(), "acs-envelope-log-wiring-")); const logPath = logPathFor(dir); const guardian = await startGuardian({ port: 0, manifestPath: "policy/manifest.yaml", envelopeLogPath: logPath }); try { - await run(guardian.url, logPath); + await run(guardian.url, logPath, () => guardian.close()); } finally { await guardian.close(); try { @@ -77,9 +77,10 @@ const logIn = (dir: string) => join(dir, "envelopes.jsonl"); describe("Guardian envelope log wiring", () => { it("records one request and one response per exchange, paired by rpc_id", async () => { - await withGuardian(logIn, async (url, logPath) => { + await withGuardian(logIn, async (url, logPath, close) => { await postRaw(url, JSON.stringify(toolCallEnvelope("rm -rf /", { id: 11 }))); + await close(); const entries = readEntries(logPath); expect(entries.length).toBe(2); expect(entries[0]?.direction).toBe("request"); @@ -93,9 +94,10 @@ describe("Guardian envelope log wiring", () => { }); it("records handshake/hello in both directions", async () => { - await withGuardian(logIn, async (url, logPath) => { + await withGuardian(logIn, async (url, logPath, close) => { await postRaw(url, JSON.stringify(makeEnvelope("handshake/hello", {}, { id: 42 }))); + await close(); const entries = readEntries(logPath); expect(entries.map((e) => e.direction)).toEqual(["request", "response"]); expect(entries.every((e) => e.method === "handshake/hello")).toBe(true); @@ -110,12 +112,13 @@ describe("Guardian envelope log wiring", () => { // recorded in both directions the same as any other response (asserted // below). it("records a schema-invalid steps/* request, then its deny decision", async () => { - await withGuardian(logIn, async (url, logPath) => { + await withGuardian(logIn, async (url, logPath, close) => { const bad = toolCallEnvelope("rm -rf /", { id: 12 }); delete (bad.params as Record).acs_version; await postRaw(url, JSON.stringify(bad)); + await close(); const entries = readEntries(logPath); expect(entries.length).toBe(2); expect(entries[0]?.direction).toBe("request"); @@ -130,9 +133,10 @@ describe("Guardian envelope log wiring", () => { }); it("records an unparseable body as a lone response with rpc_id null -- no request line to pair with", async () => { - await withGuardian(logIn, async (url, logPath) => { + await withGuardian(logIn, async (url, logPath, close) => { await postRaw(url, "{not json"); + await close(); const entries = readEntries(logPath); expect(entries.length).toBe(1); expect(entries[0]?.direction).toBe("response"); diff --git a/reference-implementations/agt/packages/guardian/test/envelope-log-sink.test.ts b/reference-implementations/agt/packages/guardian/test/envelope-log-sink.test.ts index ba642a0..22179f7 100644 --- a/reference-implementations/agt/packages/guardian/test/envelope-log-sink.test.ts +++ b/reference-implementations/agt/packages/guardian/test/envelope-log-sink.test.ts @@ -1,9 +1,9 @@ import { describe, expect, it } from "bun:test"; -import { mkdtempSync, readFileSync, rmdirSync, unlinkSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, readFileSync, rmdirSync, unlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { - createEnvelopeLogSink, + createEnvelopeLogSink as createSink, extractRpcId, NULL_ENVELOPE_LOG_SINK, type EnvelopeLogEntry, @@ -12,11 +12,19 @@ import { /** A temp directory per test. Cleanup is deliberately non-recursive -- * unlink the one file we created, then rmdir -- so a stray file makes the * test fail loudly instead of being silently blown away. */ -function withTempDir(run: (dir: string) => void): void { +async function withTempDir( + run: (dir: string, createEnvelopeLogSink: typeof createSink) => Promise, +): Promise { const dir = mkdtempSync(join(tmpdir(), "acs-envelope-log-")); + const sinks: ReturnType[] = []; try { - run(dir); + await run(dir, (options) => { + const sink = createSink(options); + sinks.push(sink); + return sink; + }); } finally { + await Promise.all(sinks.map((sink) => sink.close())); try { unlinkSync(join(dir, "envelopes.jsonl")); } catch { @@ -37,14 +45,27 @@ const REQUEST = { jsonrpc: "2.0", method: "steps/toolCallRequest", id: 7, params const RESPONSE = { jsonrpc: "2.0", id: 7, result: { decision: "deny" } }; describe("createEnvelopeLogSink -- the envelope log's JSONL format", () => { - it("writes one line per call, with a monotonic seq starting at 1", () => { - withTempDir((dir) => { + it("buffers a short burst instead of appending on the caller's stack", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { + const path = join(dir, "envelopes.jsonl"); + const sink = createEnvelopeLogSink({ path }); + sink.write("request", REQUEST, "steps/toolCallRequest"); + sink.write("response", RESPONSE, "steps/toolCallRequest"); + expect(existsSync(path) ? readFileSync(path, "utf8") : "").toBe(""); + await sink.close(); + expect(readEntries(path).map((entry) => entry.seq)).toEqual([1, 2]); + }); + }); + + it("writes one line per call, with a monotonic seq starting at 1", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const path = join(dir, "envelopes.jsonl"); const sink = createEnvelopeLogSink({ path }); sink.write("request", REQUEST, "steps/toolCallRequest"); sink.write("response", RESPONSE, "steps/toolCallRequest"); + await sink.close(); const entries = readEntries(path); expect(entries.map((e) => e.seq)).toEqual([1, 2]); expect(entries.map((e) => e.direction)).toEqual(["request", "response"]); @@ -55,43 +76,49 @@ describe("createEnvelopeLogSink -- the envelope log's JSONL format", () => { // nothing stripped, nothing reordered. This is not byte identity with the // wire: the sink is handed the already-parsed result of `await // req.json()`. - it("records the envelope unmodified -- no reformatting or stripping", () => { - withTempDir((dir) => { + it("records the envelope unmodified -- no reformatting or stripping", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const path = join(dir, "envelopes.jsonl"); - createEnvelopeLogSink({ path }).write("request", REQUEST, "steps/toolCallRequest"); + const sink = createEnvelopeLogSink({ path }); + sink.write("request", REQUEST, "steps/toolCallRequest"); + await sink.close(); expect(readEntries(path)[0]?.envelope).toEqual(REQUEST); }); }); - it("carries rpc_id from both directions, so the Inspector can pair them", () => { - withTempDir((dir) => { + it("carries rpc_id from both directions, so the Inspector can pair them", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const path = join(dir, "envelopes.jsonl"); const sink = createEnvelopeLogSink({ path }); sink.write("request", REQUEST, "steps/toolCallRequest"); sink.write("response", RESPONSE, "steps/toolCallRequest"); + await sink.close(); expect(readEntries(path).map((e) => e.rpc_id)).toEqual([7, 7]); }); }); - it("stamps recorded_at from the injected clock", () => { - withTempDir((dir) => { + it("stamps recorded_at from the injected clock", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const path = join(dir, "envelopes.jsonl"); const sink = createEnvelopeLogSink({ path, now: () => new Date("2026-08-09T12:04:31.221Z") }); sink.write("request", REQUEST, "steps/toolCallRequest"); + await sink.close(); expect(readEntries(path)[0]?.recorded_at).toBe("2026-08-09T12:04:31.221Z"); }); }); - it("records method as null when the caller cannot determine one", () => { - withTempDir((dir) => { + it("records method as null when the caller cannot determine one", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const path = join(dir, "envelopes.jsonl"); - createEnvelopeLogSink({ path }).write("response", { jsonrpc: "2.0", id: null, error: { code: -32700 } }, null); + const sink = createEnvelopeLogSink({ path }); + sink.write("response", { jsonrpc: "2.0", id: null, error: { code: -32700 } }, null); + await sink.close(); const entry = readEntries(path)[0]; expect(entry?.method).toBeNull(); expect(entry?.rpc_id).toBeNull(); @@ -100,8 +127,8 @@ describe("createEnvelopeLogSink -- the envelope log's JSONL format", () => { // The whole reason the sink is a module rather than three inline // appendFileSync calls. - it("never throws when the log path is unwritable, reports once, and goes quiet", () => { - withTempDir((dir) => { + it("never throws when the log path is unwritable, reports once, and goes quiet", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const blocker = join(dir, "envelopes.jsonl"); writeFileSync(blocker, ""); // A path *through* a regular file: mkdirSync and appendFileSync both @@ -116,8 +143,8 @@ describe("createEnvelopeLogSink -- the envelope log's JSONL format", () => { }); }); - it("never throws on an envelope JSON.stringify cannot serialize", () => { - withTempDir((dir) => { + it("never throws on an envelope JSON.stringify cannot serialize", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const path = join(dir, "envelopes.jsonl"); const errors: unknown[] = []; const sink = createEnvelopeLogSink({ path, onError: (error) => errors.push(error) }); @@ -129,8 +156,8 @@ describe("createEnvelopeLogSink -- the envelope log's JSONL format", () => { }); }); - it("never throws when onError itself throws at construction time", () => { - withTempDir((dir) => { + it("never throws when onError itself throws at construction time", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const blocker = join(dir, "envelopes.jsonl"); writeFileSync(blocker, ""); const path = join(blocker, "nested", "envelopes.jsonl"); @@ -146,8 +173,8 @@ describe("createEnvelopeLogSink -- the envelope log's JSONL format", () => { }); }); - it("never throws when onError itself throws at write time, and disables the sink", () => { - withTempDir((dir) => { + it("never throws when onError itself throws at write time, and disables the sink", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { const path = join(dir, "envelopes.jsonl"); const blocker = join(dir, "envelopes.jsonl"); writeFileSync(blocker, ""); @@ -165,6 +192,21 @@ describe("createEnvelopeLogSink -- the envelope log's JSONL format", () => { }); }); + it("reports once when serialization fails while a disk failure is pending", async () => { + await withTempDir(async (dir, createEnvelopeLogSink) => { + const errors: unknown[] = []; + // Opening a directory for append fails asynchronously, after write returns. + const sink = createEnvelopeLogSink({ path: dir, onError: (error) => errors.push(error) }); + sink.write("request", REQUEST, "steps/toolCallRequest"); + const circular: Record = {}; + circular.self = circular; + sink.write("request", circular, "steps/toolCallRequest"); + await sink.close(); + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(TypeError); + }); + }); + it("NULL_ENVELOPE_LOG_SINK writes nothing and never throws", () => { expect(() => NULL_ENVELOPE_LOG_SINK.write("request", REQUEST, "steps/toolCallRequest")).not.toThrow(); expect(NULL_ENVELOPE_LOG_SINK.path).toBeNull(); diff --git a/reference-implementations/agt/packages/guardian/test/file-log-permissions.test.ts b/reference-implementations/agt/packages/guardian/test/file-log-permissions.test.ts new file mode 100644 index 0000000..fee5ed3 --- /dev/null +++ b/reference-implementations/agt/packages/guardian/test/file-log-permissions.test.ts @@ -0,0 +1,60 @@ +// SPDX-License-Identifier: Apache-2.0 +import { describe, expect, it } from "bun:test"; +import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, rmdirSync, statSync, unlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createBatchedFileLog } from "../src/batched-log-writer.ts"; + +describe.skipIf(process.platform === "win32")("file log creation permissions", () => { + it("creates nested directories and a log with owner-only permissions under a permissive umask", async () => { + const root = mkdtempSync(join(tmpdir(), "acs-log-mode-")); + const parent = join(root, "logs"); + const dir = join(parent, "session"); + const path = join(dir, "log.jsonl"); + const previousUmask = process.umask(0); + const errors: unknown[] = []; + const writer = createBatchedFileLog(path, (error) => errors.push(error)); + try { + expect(existsSync(path)).toBe(false); + writer.write("{}\n"); + await writer.close(); + expect(errors).toEqual([]); + expect(readFileSync(path, "utf8")).toBe("{}\n"); + expect(statSync(parent).mode & 0o777).toBe(0o700); + expect(statSync(dir).mode & 0o777).toBe(0o700); + expect(statSync(path).mode & 0o777).toBe(0o600); + } finally { + await writer.close(); + process.umask(previousUmask); + if (existsSync(path)) unlinkSync(path); + rmdirSync(dir); + rmdirSync(parent); + rmdirSync(root); + } + }); + + it("appends without changing permissions on an existing directory or file", async () => { + const root = mkdtempSync(join(tmpdir(), "acs-log-existing-mode-")); + const dir = join(root, "logs"); + const path = join(dir, "log.jsonl"); + mkdirSync(dir); + writeFileSync(path, "first\n"); + chmodSync(dir, 0o750); + chmodSync(path, 0o640); + const errors: unknown[] = []; + const writer = createBatchedFileLog(path, (error) => errors.push(error)); + try { + writer.write("second\n"); + await writer.close(); + expect(errors).toEqual([]); + expect(readFileSync(path, "utf8")).toBe("first\nsecond\n"); + expect(statSync(dir).mode & 0o777).toBe(0o750); + expect(statSync(path).mode & 0o777).toBe(0o640); + } finally { + await writer.close(); + unlinkSync(path); + rmdirSync(dir); + rmdirSync(root); + } + }); +}); diff --git a/reference-implementations/agt/packages/guardian/test/log-shutdown.test.ts b/reference-implementations/agt/packages/guardian/test/log-shutdown.test.ts new file mode 100644 index 0000000..d68cc43 --- /dev/null +++ b/reference-implementations/agt/packages/guardian/test/log-shutdown.test.ts @@ -0,0 +1,119 @@ +// SPDX-License-Identifier: Apache-2.0 +import { expect, it } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync, rmdirSync, unlinkSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { startGuardian } from "../src/server.ts"; + +function request() { + return { + jsonrpc: "2.0", method: "steps/toolCallRequest", id: 1, + params: { + acs_version: "0.1.0", request_id: crypto.randomUUID(), timestamp: new Date().toISOString(), + metadata: { agent_id: "test", session_id: crypto.randomUUID() }, + payload: { tool: { name: "run_shell" }, arguments: { command: { value: "ls" } } }, + }, + }; +} + +function readLines(path: string): Record[] { + return readFileSync(path, "utf8").trim().split("\n").map((line) => JSON.parse(line)); +} + +function cleanup(dir: string): void { + for (const name of ["envelopes.jsonl", "context.jsonl", "traces.jsonl"]) { + const path = join(dir, name); + if (existsSync(path)) unlinkSync(path); + } + rmdirSync(dir); +} + +it("close waits for an active decision and drains its response and session record", async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-log-shutdown-")); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const guardian = await startGuardian({ + port: 0, manifestPath: "policy/manifest.yaml", + envelopeLogPath: join(dir, "envelopes.jsonl"), sessionContextLog: join(dir, "context.jsonl"), + bridge: { async evaluate() { entered.resolve(); await release.promise; return { decision: "allow" }; } }, + }); + try { + const response = fetch(guardian.url, { method: "POST", body: JSON.stringify(request()) }); + await entered.promise; + const closing = guardian.close(); + const refused = await fetch(guardian.url, { method: "POST", body: JSON.stringify(request()) }); + expect(refused.status).toBe(503); + await refused.arrayBuffer(); + let closed = false; + void closing.then(() => { closed = true; }); + await Promise.resolve(); + expect(closed).toBe(false); + release.resolve(); + expect(((await (await response).json()) as { result: { decision: string } }).result.decision).toBe("allow"); + await closing; + expect(guardian.close()).toBe(closing); + expect(readLines(join(dir, "envelopes.jsonl")).map((entry) => entry.direction)).toEqual(["request", "response"]); + expect(readLines(join(dir, "context.jsonl")).map((entry) => entry.seq)).toEqual([1]); + } finally { + release.resolve(); + await guardian.close(); + cleanup(dir); + } +}); + +for (const signal of ["SIGINT", "SIGTERM"] as const) { + for (const tracing of [false, true]) { + it(`the standalone Guardian drains its logs on ${signal} with trace export ${tracing ? "on" : "off"}`, async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-log-signal-")); + const env = { ...process.env }; + delete env.ACS_OTEL_FILE; + delete env.ACS_OTEL_ENDPOINT; + if (tracing) env.ACS_OTEL_FILE = join(dir, "traces.jsonl"); + const child = Bun.spawn([process.execPath, "run", "packages/guardian/src/main.ts"], { + env: { + ...env, ACS_GUARDIAN_PORT: "0", + ACS_ENVELOPE_LOG: join(dir, "envelopes.jsonl"), ACS_SESSION_CONTEXT_LOG: join(dir, "context.jsonl"), + }, + stdout: "pipe", stderr: "pipe", timeout: 3000, + }); + try { + const reader = child.stdout.getReader(); + const decoder = new TextDecoder(); + let output = ""; + let url: string | undefined; + try { + while (!url) { + const { done, value } = await reader.read(); + if (done) throw new Error(`Guardian exited before its startup banner: ${output}`); + output += decoder.decode(value, { stream: true }); + url = /Guardian listening at (http:\/\/localhost:\d+\/acs)\n/.exec(output)?.[1]; + } + } finally { + reader.releaseLock(); + } + const response = await fetch(url, { method: "POST", body: JSON.stringify(request()) }); + expect(((await response.json()) as { result: { decision: string } }).result.decision).toBe("allow"); + child.kill(signal); + expect(await child.exited).toBe(0); + const remaining = child.stdout.getReader(); + try { + while (true) { + const { done, value } = await remaining.read(); + if (done) break; + output += decoder.decode(value, { stream: true }); + } + } finally { + remaining.releaseLock(); + } + expect(output.includes("OTel traces:")).toBe(tracing); + expect(readLines(join(dir, "envelopes.jsonl")).map((entry) => entry.seq)).toEqual([1, 2]); + expect(readLines(join(dir, "context.jsonl")).map((entry) => entry.seq)).toEqual([1]); + if (tracing) expect(readLines(join(dir, "traces.jsonl"))).toHaveLength(1); + } finally { + if (child.exitCode === null) child.kill("SIGKILL"); + await child.exited; + cleanup(dir); + } + }); + } +} diff --git a/reference-implementations/agt/packages/guardian/test/server.test.ts b/reference-implementations/agt/packages/guardian/test/server.test.ts index 60f0be3..b9da00e 100644 --- a/reference-implementations/agt/packages/guardian/test/server.test.ts +++ b/reference-implementations/agt/packages/guardian/test/server.test.ts @@ -352,6 +352,7 @@ describe("startGuardian POST /acs -- denyOnInvalidEnvelope's boundary: what stay const response = await postAcs(guardian.url, bad); expect(response.result?.decision).toBe("deny"); + await guardian.close(); const lines = readFileSync(logPath, "utf8").trim().split("\n"); const entries = lines.map((line) => JSON.parse(line) as { direction: string; rpc_id: unknown }); expect(entries.map((e) => e.direction)).toEqual(["request", "response"]); @@ -554,7 +555,7 @@ const SCHEMALESS_SCRATCH_DIR = join(GUARDIAN_PKG, "tmp-schemaless-scratch"); * Deletions here are explicit per file (repo constraint: nothing recursive). */ async function withSchemalessGuardian( - body: (guardian: { url: string; logPath: string }) => Promise, + body: (guardian: { url: string; logPath: string; close(): Promise }) => Promise, ): Promise { const root = SCHEMALESS_SCRATCH_DIR; const srcDir = join(root, "src"); @@ -581,7 +582,7 @@ async function withSchemalessGuardian( }); guardian = started; - await body({ url: started.url, logPath }); + await body({ url: started.url, logPath, close: () => started.close() }); } finally { await guardian?.close(); for (const path of [logPath, ...copied.map((file) => join(srcDir, file))]) { @@ -876,9 +877,10 @@ describe("startGuardian POST /acs -- the outer net around dispatch", () => { }); it("records both the request and the response, so the envelope log has no unrecorded exit", async () => { - await withSchemalessGuardian(async ({ url, logPath }) => { + await withSchemalessGuardian(async ({ url, logPath, close }) => { await postAcs(url, toolCallEnvelope("ls -la", { id: 11 })); + await close(); const lines = readFileSync(logPath, "utf8").trim().split("\n"); const entries = lines.map((line) => JSON.parse(line) as { direction: string; rpc_id: unknown }); expect(entries.map((e) => e.direction)).toEqual(["request", "response"]); @@ -1372,6 +1374,7 @@ describe("session state end to end", () => { // ever appended would still pass a single-post version of this test. await postStep(guardian, toolCallRequest({ session_id: sessionId, request_id: requestId1 })); await postStep(guardian, toolCallRequest({ session_id: sessionId, request_id: requestId2 })); + await guardian.close(); const lines = readFileSync(path, "utf8").trim().split("\n"); expect(lines).toHaveLength(2); expect(JSON.parse(lines[0]!)).toMatchObject({ session_id: sessionId, seq: 1, request_id: requestId1 }); diff --git a/reference-implementations/agt/packages/guardian/test/shutdown-deadline.test.ts b/reference-implementations/agt/packages/guardian/test/shutdown-deadline.test.ts new file mode 100644 index 0000000..30877aa --- /dev/null +++ b/reference-implementations/agt/packages/guardian/test/shutdown-deadline.test.ts @@ -0,0 +1,86 @@ +// SPDX-License-Identifier: Apache-2.0 +import { expect, it, spyOn } from "bun:test"; +import * as fs from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Writable } from "node:stream"; +import { NULL_OTEL_TRACE_EXPORTER } from "../src/otel-trace-exporter.ts"; +import { buildServerHello } from "../src/handshake.ts"; +import { startGuardian } from "../src/server.ts"; + +const timeoutMs = buildServerHello().timeout_config.default_ms; + +function request() { + return { + jsonrpc: "2.0", method: "steps/toolCallRequest", id: 1, + params: { + acs_version: "0.1.0", request_id: crypto.randomUUID(), timestamp: new Date().toISOString(), + metadata: { agent_id: "test", session_id: crypto.randomUUID() }, + payload: { tool: { name: "run_shell" }, arguments: { command: { value: "ls" } } }, + }, + }; +} + +it("closes within the negotiated deadline when a decision never finishes, without trace export", async () => { + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const guardian = await startGuardian({ + port: 0, manifestPath: "policy/manifest.yaml", + bridge: { async evaluate() { entered.resolve(); await release.promise; return { decision: "allow" }; } }, + }); + const response = fetch(guardian.url, { method: "POST", body: JSON.stringify(request()) }).catch(() => undefined); + try { + await entered.promise; + const closing = guardian.close(); + expect(guardian.close()).toBe(closing); + let closed = false; + void closing.then(() => { closed = true; }); + await Bun.sleep(timeoutMs + 500); + expect(closed).toBe(true); + } finally { + release.resolve(); + await response; + await guardian.close(); + } +}, timeoutMs + 5000); + +it("bounds both log drains by one deadline before shutting down the trace exporter", async () => { + const dir = fs.mkdtempSync(join(tmpdir(), "acs-log-deadline-")); + const releases: (() => void)[] = []; + const streams: Writable[] = []; + const create = spyOn(fs, "createWriteStream").mockImplementation(() => { + const stream = new Writable({ write(_chunk, _encoding, done) { releases.push(done); } }); + streams.push(stream); + return stream as fs.WriteStream; + }); + let shutdowns = 0; + const guardian = await startGuardian({ + port: 0, manifestPath: "policy/manifest.yaml", + envelopeLogPath: join(dir, "envelopes.jsonl"), sessionContextLog: join(dir, "context.jsonl"), + traceExporter: { ...NULL_OTEL_TRACE_EXPORTER, async shutdown() { shutdowns += 1; } }, + bridge: { async evaluate() { return { decision: "allow" }; } }, + }); + try { + const response = await fetch(guardian.url, { method: "POST", body: JSON.stringify(request()) }); + expect(response.ok).toBe(true); + await response.arrayBuffer(); + const closing = guardian.close(); + let closed = false; + void closing.then(() => { closed = true; }); + await Bun.sleep(10); + expect(closed).toBe(false); + expect(shutdowns).toBe(0); + await Bun.sleep(timeoutMs + 500); + expect(streams).toHaveLength(2); + expect(streams.every((stream) => stream.writableEnded && !stream.writableFinished)).toBe(true); + expect(closed).toBe(true); + expect(shutdowns).toBe(1); + expect(guardian.close()).toBe(closing); + } finally { + create.mockRestore(); + for (const stream of streams) stream.destroy(); + for (const release of releases) release(); + await guardian.close(); + fs.rmdirSync(dir); + } +}, timeoutMs + 5000);