Batch Guardian log writes and drain them on shutdown

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)

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.

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,13 +169,45 @@ 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 |
 | `src/loopback-proxy.ts` | Keeps a loopback trace endpoint off an HTTP proxy, applied by `src/main.ts` |
 | `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`.

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<void>;
+};
+
+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<typeof setTimeout> | 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();
+    },
+  };
+}

reference-implementations/agt/packages/guardian/src/envelope-log-sink.ts

@@ -18,15 +18,17 @@
  * 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
  * can turn a governed tool call into an ungoverned one is the one failure
  * 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<void>;
 };
 
 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);

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<void> => {
-    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);