GenAI-Security-Project · rksharma-owg · Sep 13, 2026
@@ -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/).
 

@@ -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.

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

@@ -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);