From e6536da04b0e9347a90a6110b42a748507050f1c Mon Sep 17 00:00:00 2001 From: RKS Date: Sun, 13 Sep 2026 01:22:25 -0400 Subject: [PATCH] Batch Guardian log writes and drain them on shutdown Signed-off-by: RKS --- README.md | 7 +- reference-implementations/agt/README.md | 2 +- .../agt/packages/guardian/README.md | 28 +++- .../guardian/src/batched-log-writer.ts | 122 ++++++++++++++ .../guardian/src/envelope-log-sink.ts | 18 ++- .../agt/packages/guardian/src/main.ts | 7 + .../agt/packages/guardian/src/server.ts | 89 +++-------- .../guardian/test/batched-log-writer.test.ts | 149 ++++++++++++++++++ .../test/envelope-log-sink-wiring.test.ts | 16 +- .../guardian/test/envelope-log-sink.test.ts | 90 ++++++++--- .../guardian/test/log-shutdown.test.ts | 98 ++++++++++++ .../agt/packages/guardian/test/server.test.ts | 9 +- 12 files changed, 516 insertions(+), 119 deletions(-) create mode 100644 reference-implementations/agt/packages/guardian/src/batched-log-writer.ts create mode 100644 reference-implementations/agt/packages/guardian/test/batched-log-writer.test.ts create mode 100644 reference-implementations/agt/packages/guardian/test/log-shutdown.test.ts diff --git a/README.md b/README.md index f570da3..2c1a81b 100644 --- a/README.md +++ b/README.md @@ -289,10 +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 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 +The Guardian batches its envelope and session-context logs for asynchronous append, +with bounded queues and a drain on graceful shutdown. 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/). diff --git a/reference-implementations/agt/README.md b/reference-implementations/agt/README.md index bb69941..c1dde1f 100644 --- a/reference-implementations/agt/README.md +++ b/reference-implementations/agt/README.md @@ -370,7 +370,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; 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 511f979..5d5f5cf 100644 --- a/reference-implementations/agt/packages/guardian/README.md +++ b/reference-implementations/agt/packages/guardian/README.md @@ -34,7 +34,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. @@ -78,8 +78,32 @@ 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/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. + +Each sink admits at most 4 MiB of serialized UTF-8 data and 4,096 records, counting +writes already in flight. A write error or a record that would exceed either limit +disables that sink and reports once on stderr. 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()` stops accepting connections, waits for active requests, and +drains both logs. Repeated calls wait for the same shutdown. The standalone Guardian +does this on SIGINT and SIGTERM. A stuck request or disk write can delay shutdown; +there is no forced shutdown deadline. SIGKILL, a crash, or power loss can lose buffered +records. Completion means the stream finished appending, not that data was fsynced. + +This batches the reference implementation's existing logs. It does not add an +OpenTelemetry exporter or 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. @@ -90,4 +114,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`. +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..0805a4d --- /dev/null +++ b/reference-implementations/agt/packages/guardian/src/batched-log-writer.ts @@ -0,0 +1,122 @@ +// 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 = { + write(line: string): void; + close(): Promise; +}; + +export const NULL_LOG_WRITER: BatchedLogWriter = { + 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. A failed or full + * sink is disabled and reported once, as the existing log sinks require. */ +export function batchLogWrites( + stream: Writable, + onError: (error: unknown) => void, + { batchSize = 64, flushIntervalMs = 100, maxBufferedBytes = 4 * 1024 * 1024, maxBufferedRecords = 4096 } = {}, +): BatchedLogWriter { + let timer: ReturnType | undefined; + let batchCount = 0; + let pendingRecords = 0; + let closed = false; + let failed = false; + + 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 { + write(line) { + if (closed || failed) return; + try { + const bytes = Buffer.byteLength(line); + if (stream.writableLength + bytes > maxBufferedBytes || pendingRecords >= maxBufferedRecords) { + fail(new Error("log buffer limit exceeded; queued records may be lost")); + 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(); + // 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 }); + } catch (error) { + fail(error); + } + return { + 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" }), fail); + 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 a3dee19..0aef2fc 100644 --- a/reference-implementations/agt/packages/guardian/src/main.ts +++ b/reference-implementations/agt/packages/guardian/src/main.ts @@ -49,3 +49,10 @@ 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)`); + +// 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 9fd4e21..e279656 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 { appendContextEntry, @@ -334,59 +333,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, @@ -407,19 +353,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`) } : {}, + ); const server = Bun.serve({ hostname: hostname ?? LOOPBACK_ONLY, @@ -434,10 +377,14 @@ export async function startGuardian({ }, }); + let closing: Promise | undefined; return { url: `http://localhost:${server.port}${ACS_PATH}`, - async close() { - await server.stop(true); + close() { + return closing ??= (async () => { + await server.stop(); + await Promise.all([envelopeLog.close(), contextLog.close()]); + })(); }, }; } @@ -557,7 +504,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..3fc4738 --- /dev/null +++ b/reference-implementations/agt/packages/guardian/test/batched-log-writer.test.ts @@ -0,0 +1,149 @@ +// 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).toHaveLength(1); + expect(String(errors[0])).toContain("buffer limit exceeded"); + expect(stream.destroyed).toBe(true); + finishWrite?.(); + await writer.close(); + expect(errors).toHaveLength(1); + }); + + 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).toHaveLength(1); + expect(batches).toEqual([]); + }); + + 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"); + }); +}); 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/log-shutdown.test.ts b/reference-implementations/agt/packages/guardian/test/log-shutdown.test.ts new file mode 100644 index 0000000..0d652db --- /dev/null +++ b/reference-implementations/agt/packages/guardian/test/log-shutdown.test.ts @@ -0,0 +1,98 @@ +// 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"]) { + 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(); + 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) { + it(`the standalone Guardian drains its logs on ${signal}`, async () => { + const dir = mkdtempSync(join(tmpdir(), "acs-log-signal-")); + const child = Bun.spawn([process.execPath, "run", "packages/guardian/src/main.ts"], { + env: { + ...process.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); + 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]); + } 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 });