Tag streaming enforce calls with stable streamId (0.13.1)

Streaming adapters generate one streamId per stream and inject it,
plus a slice index, into metadata on every per-chunk enforce call.
Lets cloud dashboards collapse N per-chunk audit rows into one
logical operation for reviewers. Additive — pure metadata, no
public API change.

Applies to pre-post-stream.ts (used by Anthropic, LangChain, Vercel
AI, etc.) and mastra-processor-stream.ts (Mastra's per-chunk API).

Author: scotty595

packages/governance/package.json

@@ -1,6 +1,6 @@
 {
   "name": "governance-sdk",
-  "version": "0.13.0",
+  "version": "0.13.1",
   "description": "AI Agent Governance for TypeScript — policy enforcement, scoring, compliance, and audit for AI agents",
   "type": "module",
   "main": "./dist/index.js",

packages/governance/src/plugins/mastra-processor-stream.ts

@@ -45,6 +45,45 @@ interface StreamState {
   slidingBuffer?: string;
   /** Chunk count in the rolling window (sliding mode). */
   slidingChunks?: number;
+  /**
+   * Stable id for this stream. Every per-chunk enforce call carries it
+   * in metadata so the cloud dashboard can collapse N audit rows into
+   * one logical operation. Lazily generated — first chunk seen.
+   */
+  streamId?: string;
+  /** Monotonically increasing slice number for per-chunk audit tags. */
+  streamSlice?: number;
+}
+
+/** Zero-dep UUID generator, same helper shape as supply-chain-cyclonedx. */
+function generateStreamId(): string {
+  if (typeof globalThis.crypto !== "undefined" && globalThis.crypto.randomUUID) {
+    return `str_${globalThis.crypto.randomUUID()}`;
+  }
+  const bytes = new Uint8Array(16);
+  if (typeof globalThis.crypto !== "undefined" && globalThis.crypto.getRandomValues) {
+    globalThis.crypto.getRandomValues(bytes);
+  } else {
+    for (let i = 0; i < 16; i++) bytes[i] = Math.floor(Math.random() * 256);
+  }
+  const hex = Array.from(bytes, (b) => b.toString(16).padStart(2, "0")).join("");
+  return `str_${hex}`;
+}
+
+/**
+ * Read (and lazily init) the streamId + slice index for this Mastra
+ * stream. Stored on args.state so subsequent chunks share the same id
+ * across the whole stream lifecycle. Caller mutates state.streamSlice++
+ * after consuming.
+ */
+function getOrCreateStreamMeta(
+  state: StreamState,
+): { streamId: string; slice: number } {
+  if (!state.streamId) state.streamId = generateStreamId();
+  if (state.streamSlice == null) state.streamSlice = 0;
+  const slice = state.streamSlice;
+  state.streamSlice = slice + 1;
+  return { streamId: state.streamId, slice };
 }
 
 // ─── Entry point ──────────────────────────────────────────────
@@ -100,9 +139,13 @@ async function handlePerChunk(
   agentLevel: number,
   callbacks: OutcomeCallbacks,
 ): Promise<MastraStreamChunk | null | undefined> {
+  const state = (args.state ?? {}) as StreamState;
+  const streamMeta = getOrCreateStreamMeta(state);
+  if (args.state && args.state !== state) Object.assign(args.state, state);
+
   const decision = await runScan(
     governance, chunkText, config, agentId, agentLevel, callbacks,
-    "mastra.stream:per-chunk",
+    "mastra.stream:per-chunk", streamMeta,
   );
   return applyDecisionToChunk(args, decision, chunkText, config);
 }
@@ -124,6 +167,7 @@ async function handleSliding(
   callbacks: OutcomeCallbacks,
 ): Promise<MastraStreamChunk | null | undefined> {
   const state = (args.state ?? {}) as StreamState;
+  const streamMeta = getOrCreateStreamMeta(state);
   const lookback = config.streamLookbackChunks ?? 2;
   const lookbackChars = config.streamLookbackChars;
 
@@ -132,7 +176,7 @@ async function handleSliding(
 
   const decision = await runScan(
     governance, nextBuffer, config, agentId, agentLevel, callbacks,
-    "mastra.stream:sliding",
+    "mastra.stream:sliding", streamMeta,
   );
 
   // Trim the buffer once it grows past the lookback budget so it doesn't
@@ -170,6 +214,7 @@ async function runScan(
   agentLevel: number,
   callbacks: OutcomeCallbacks,
   toolName: string,
+  streamMeta?: { streamId: string; slice: number },
 ): Promise<EnforcementDecision> {
   // enforcePostprocess throws on block / require_approval (via handleOutcome).
   // In Mastra's per-chunk API we can't throw — we must call args.abort().
@@ -180,7 +225,12 @@ async function runScan(
       agentId,
       agentName: config.agentName,
       agentLevel,
-      metadata: config.metadata,
+      metadata: {
+        ...(config.metadata ?? {}),
+        ...(streamMeta
+          ? { streamId: streamMeta.streamId, streamSlice: streamMeta.slice }
+          : {}),
+      },
       callbacks,
       toolName,
     });

packages/governance/src/plugins/pre-post-stream.ts

@@ -47,6 +47,46 @@ import type { OutcomeCallbacks } from "./outcome-handler.js";
 import { enforcePostprocess } from "./pre-post-enforce.js";
 import type { PrePostEnforceOptions } from "./pre-post-enforce.js";
 
+/**
+ * Generate a stable stream id. Every enforce() call for a single LLM
+ * stream carries the same id in metadata so the dashboard can collapse
+ * N per-chunk rows into one logical "operation" for reviewers. Zero-dep
+ * UUIDv4 fallback mirrors the one in supply-chain-cyclonedx.ts so the
+ * SDK's "no runtime deps" rule stays intact.
+ */
+function generateStreamId(): string {
+  if (typeof globalThis.crypto !== "undefined" && globalThis.crypto.randomUUID) {
+    return `str_${globalThis.crypto.randomUUID()}`;
+  }
+  const bytes = new Uint8Array(16);
+  if (typeof globalThis.crypto !== "undefined" && globalThis.crypto.getRandomValues) {
+    globalThis.crypto.getRandomValues(bytes);
+  } else {
+    for (let i = 0; i < 16; i++) bytes[i] = Math.floor(Math.random() * 256);
+  }
+  const hex = Array.from(bytes, (b) => b.toString(16).padStart(2, "0")).join("");
+  return `str_${hex}`;
+}
+
+/**
+ * Merge a streamId + slice index into the options.metadata for each
+ * per-chunk enforce call. Preserves any caller-supplied metadata.
+ */
+function withStreamMeta<O extends PrePostEnforceOptions>(
+  options: O,
+  streamId: string,
+  sliceIndex: number,
+): O {
+  return {
+    ...options,
+    metadata: {
+      ...(options.metadata ?? {}),
+      streamId,
+      streamSlice: sliceIndex,
+    },
+  };
+}
+
 // ─── Types ────────────────────────────────────────────────────
 
 export type StreamMode = "buffered" | "sliding" | "per-chunk";
@@ -88,16 +128,21 @@ export async function* enforcePostprocessStream<ChunkT>(
   options: StreamEnforceOptions<ChunkT>,
 ): AsyncIterable<ChunkT> {
   const mode: StreamMode = options.streamMode ?? "buffered";
+  // One id per stream — every enforce call this helper triggers carries
+  // it in metadata so the cloud dashboard can collapse repeated rows
+  // into a single logical operation. Buffered mode also tags its single
+  // call so buffered-vs-per-chunk audit shape is consistent.
+  const streamId = generateStreamId();
 
   if (mode === "buffered") {
-    yield* runBuffered(governance, source, options);
+    yield* runBuffered(governance, source, options, streamId);
     return;
   }
   if (mode === "per-chunk") {
-    yield* runPerChunk(governance, source, options);
+    yield* runPerChunk(governance, source, options, streamId);
     return;
   }
-  yield* runSliding(governance, source, options);
+  yield* runSliding(governance, source, options, streamId);
 }
 
 // ─── Buffered mode ────────────────────────────────────────────
@@ -106,6 +151,7 @@ async function* runBuffered<ChunkT>(
   governance: GovernanceInstance,
   source: AsyncIterable<ChunkT>,
   options: StreamEnforceOptions<ChunkT>,
+  streamId: string,
 ): AsyncIterable<ChunkT> {
   const chunks: ChunkT[] = [];
   const texts: string[] = [];
@@ -121,7 +167,7 @@ async function* runBuffered<ChunkT>(
   }
 
   const result = await enforcePostprocess(governance, combined, {
-    ...options,
+    ...withStreamMeta(options, streamId, 0),
     toolName: options.toolName ?? "stream:buffered",
   });
 
@@ -147,7 +193,9 @@ async function* runPerChunk<ChunkT>(
   governance: GovernanceInstance,
   source: AsyncIterable<ChunkT>,
   options: StreamEnforceOptions<ChunkT>,
+  streamId: string,
 ): AsyncIterable<ChunkT> {
+  let sliceIndex = 0;
   for await (const chunk of source) {
     const text = options.extractText(chunk);
     if (!text) {
@@ -156,7 +204,7 @@ async function* runPerChunk<ChunkT>(
     }
 
     const result = await enforcePostprocess(governance, text, {
-      ...options,
+      ...withStreamMeta(options, streamId, sliceIndex++),
       toolName: options.toolName ?? "stream:per-chunk",
     });
 
@@ -176,18 +224,20 @@ async function* runSliding<ChunkT>(
   governance: GovernanceInstance,
   source: AsyncIterable<ChunkT>,
   options: StreamEnforceOptions<ChunkT>,
+  streamId: string,
 ): AsyncIterable<ChunkT> {
   const window: ChunkT[] = [];
   const windowTexts: string[] = [];
   const lookbackChunks = options.streamLookbackChunks ?? 2;
   const lookbackChars = options.streamLookbackChars;
+  const sliceCounter = { value: 0 };
 
   for await (const chunk of source) {
     window.push(chunk);
     windowTexts.push(options.extractText(chunk));
 
     while (shouldFlush(window.length, windowTexts, lookbackChunks, lookbackChars)) {
-      yield* flushOldest(governance, window, windowTexts, options);
+      yield* flushOldest(governance, window, windowTexts, options, streamId, sliceCounter);
     }
   }
 
@@ -200,7 +250,7 @@ async function* runSliding<ChunkT>(
     return;
   }
   const result = await enforcePostprocess(governance, tailText, {
-    ...options,
+    ...withStreamMeta(options, streamId, sliceCounter.value++),
     toolName: options.toolName ?? "stream:sliding-tail",
   });
   if (result.text === tailText) {
@@ -234,6 +284,8 @@ async function* flushOldest<ChunkT>(
   window: ChunkT[],
   windowTexts: string[],
   options: StreamEnforceOptions<ChunkT>,
+  streamId: string,
+  sliceCounter: { value: number },
 ): AsyncIterable<ChunkT> {
   // Scan the full lookback window (oldest + lookback tail) before flushing
   // the oldest chunk. This gives the scanner context straddling boundaries.
@@ -246,7 +298,7 @@ async function* flushOldest<ChunkT>(
   }
 
   const result = await enforcePostprocess(governance, scanText, {
-    ...options,
+    ...withStreamMeta(options, streamId, sliceCounter.value++),
     toolName: options.toolName ?? "stream:sliding",
   });