V48 Gate 3 (implementation-only): raw LLM I/O sidecar lands under the real run id, not a literal 'run' directory

The sidecar writer resolved the run id via findUp('pipeline',
'correlationId') — a namespace nothing writes — so every run's wire I/O
piled into ~/.bitcode/logs/executions/run with a global counter, defeating
per-run separation of stall evidence. createStreamingExecution now stores
the run id under the canonical execution/correlationId identity key at the
root, and the diagnostics context resolves that key (root execution id as
fallback) from any depth.

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

Author: geraldarthurdavis

packages/agent-generics/src/diagnostics/instrumentation.ts

@@ -10,7 +10,11 @@ function getCtx(execution: Execution, sequence?: string) {
   const agentName = (execution as any).findUp?.('agent', 'name');
   const execId = (execution as any).id;
   const path = (execution as any).getPath?.() || [];
-  const correlationId = (execution as any).findUp?.('pipeline', 'correlationId');
+  // The run id lives at the streaming root: canonically under
+  // execution/correlationId, with the root execution id (== runId for
+  // createStreamingExecution roots) as the fallback.
+  const correlationId = (execution as any).findUp?.('execution', 'correlationId')
+    ?? (execution as any).getRoot?.()?.id;
   // Surface any provider/model already stored on ancestors (best-effort)
   let providerUp: string | undefined;
   let modelUp: string | undefined;

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

@@ -1,6 +1,6 @@
 // @ts-nocheck
 import { Execution } from '@bitcode/execution-generics';
-import { enablePipelineStreaming, sourceSafeStreamEvent } from '../../streaming/pipeline-stream-integration';
+import { createStreamingExecution, enablePipelineStreaming, sourceSafeStreamEvent } from '../../streaming/pipeline-stream-integration';
 
 // Mock ORM model so we can assert persistence without a real DB
 const createdEvents: any[] = [];
@@ -48,6 +48,22 @@ describe('pipeline-stream-integration', () => {
     expect(createdEvents[0].run_id).toBe('run-123');
     expect(createdEvents[0].event_type).toBeDefined();
   });
+
+  it('createStreamingExecution stores the run id as the canonical correlationId, resolvable from deep children', () => {
+    const exec = createStreamingExecution({
+      runId: 'run-correlation-1',
+      userId: 'user-1',
+      supabase: {} as any,
+      streamToDatabase: false,
+      streamToSSE: false,
+    });
+
+    expect(exec.get('execution', 'correlationId')).toBe('run-correlation-1');
+    // Substep diagnostics (raw LLM I/O sidecar directory naming) resolve the
+    // run id via findUp from arbitrarily deep child nodes.
+    const deepChild = exec.child('phase:discovery').child('agent:search').child('step:plan');
+    expect(deepChild.findUp('execution', 'correlationId')).toBe('run-correlation-1');
+  });
 });
 
 describe('pipeline-stream-integration — executions-row FK race (QA: "Failed to persist stream event" 23503)', () => {

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

@@ -666,7 +666,12 @@ export function createStreamingExecution(
   config: PipelineStreamConfig & { parent?: Execution }
 ): Execution {
       const execution = new Execution(config.runId, config.parent);
-  
+
+  // Canonical run identity: substep diagnostics (e.g. the raw LLM I/O
+  // sidecar directory) resolve the run id via findUp('execution',
+  // 'correlationId') from arbitrarily deep child nodes.
+  execution.store('execution', 'correlationId', config.runId);
+
   // Enable streaming
   enablePipelineStreaming(execution, config);