V48 Gate 3: Universal source-safe content-withholding filter for pipeline telemetry

sourceSafeStreamEvent withholds llm:{prompt,input,output,reasoningOutput,judgmentOutput,parsedOutput} content from every persisted/streamed pipeline event, leaving only a content-withheld summary. Applied at the execution_events persistence chokepoint so the formal agent-generics LLM substeps — which store full prompt/response content — stream lawfully (rawPromptVisible/rawProviderResponseVisible stay false) for ALL pipelines. Prerequisite for AssetPacksSynthesis on the real primitives.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: geraldarthurdavis

packages/pipelines-generics/src/streaming/pipeline-stream-integration.ts

@@ -26,9 +26,65 @@ export interface PipelineStreamConfig {
   structuredToDatabase?: boolean;
 }
 
+// Source-safety law (V48): pipeline telemetry must never serialize raw prompts
+// or provider responses (rawPromptVisible/rawProviderResponseVisible=false). The
+// formal LLM substeps store full prompt/response content under the `llm`
+// namespace, which auto-streams via Execution.store. This withholds that content
+// from any persisted/streamed event, leaving only a content-withheld summary —
+// applied universally to every pipeline stream event (not per-pipeline).
+const SOURCE_SAFE_LLM_CONTENT_KEYS = new Set([
+  'prompt',
+  'input',
+  'output',
+  'reasoningOutput',
+  'judgmentOutput',
+  'parsedOutput',
+]);
+
+export function sourceSafeStreamEvent(event: any): any {
+  if (!event || typeof event !== 'object') return event;
+  const namespace = (event as any).namespace;
+  const key = (event as any).key;
+  if (namespace !== 'llm' || !SOURCE_SAFE_LLM_CONTENT_KEYS.has(String(key))) {
+    return event;
+  }
+  const data =
+    (event as any).data && typeof (event as any).data === 'object'
+      ? ((event as any).data as Record<string, any>)
+      : {};
+  const contentChars =
+    typeof (event as any).data === 'string'
+      ? (event as any).data.length
+      : typeof data.content === 'string'
+        ? data.content.length
+        : typeof data.prompt === 'string'
+          ? data.prompt.length
+          : null;
+  const state =
+    (event as any).executionState && typeof (event as any).executionState === 'object'
+      ? (event as any).executionState
+      : {};
+  return {
+    ...event,
+    message: '[content withheld — source-safe]',
+    data: {
+      contentWithheld: true,
+      sourceSafetyClass: 'source_safe',
+      stage: key,
+      generation: data.generation ?? state.generation ?? null,
+      provider: data.provider ?? null,
+      model: data.model ?? null,
+      contentChars,
+      phase: data.phase ?? state.phase ?? null,
+      agent: data.agent ?? state.agent ?? null,
+      step: data.step ?? state.step ?? null,
+    },
+  };
+}
+
 /**
  * Wire up a pipeline execution with streaming
- * 
+ *
  * This registers a stream manager with the execution so that
  * all storage operations emit stream events automatically.
  */
@@ -87,7 +143,7 @@ export function enablePipelineStreaming(
         await eventsModel.create({
           run_id: config.runId,
           event_type: event.type,
-          event_data: event,
+          event_data: sourceSafeStreamEvent(event),
           created_at: new Date().toISOString(),
         });
       } catch (error) {