wip v28

Author: geraldarthurdavis

packages/pipelines-generics/src/streaming/__tests__/forkable-execution.test.ts

@@ -285,4 +285,54 @@ describe('enablePipelineStreaming + Execution emits DB events (snapshot stream)'
       tool_error: null,
     });
   });
+
+  it('creates correlation rows for stored tool status events outside an explicit agent span', async () => {
+    const supabase = new MockSupabase() as any;
+    const runId = '77777777-7777-4777-8777-777777777777';
+    const userId = '88888888-8888-4888-8888-888888888888';
+
+    const exec = new Execution(`exec-${runId}`);
+    const streamer = enablePipelineStreaming(exec as any, {
+      runId,
+      userId,
+      supabase,
+      structuredToDatabase: true,
+    });
+
+    await ExecutionStreamAdapter.emitEvent(exec.id, 'status' as any, {
+      namespace: 'tools',
+      key: 'result',
+      data: {
+        tool: 'bitcode.depository.search',
+        ok: true,
+        input: { read: { id: 'read-1' }, assetCount: 1 },
+        output: { resultState: 'worthy_fit', selectedCandidateCount: 1 },
+        phase: 'setup',
+        agent: 'setup:depository-search',
+        step: 'try',
+      },
+    });
+
+    await (streamer as any).flushStructuredWrites?.();
+
+    expect(supabase.tables.deliverable_pipeline_phase_delegations).toHaveLength(1);
+    expect(supabase.tables.deliverable_pipeline_phase_delegations[0]).toMatchObject({
+      phase_name: 'setup',
+      status: 'completed',
+    });
+    expect(supabase.tables.deliverable_pipeline_agent_steps).toHaveLength(1);
+    expect(supabase.tables.deliverable_pipeline_agent_steps[0]).toMatchObject({
+      phase_delegation_id: 'deliverable_pipeline_phase_delegations-1',
+      agent_name: 'setup:depository-search',
+      status: 'completed',
+    });
+    expect(supabase.tables.deliverable_pipeline_tool_executions).toHaveLength(1);
+    expect(supabase.tables.deliverable_pipeline_tool_executions[0]).toMatchObject({
+      agent_step_id: 'deliverable_pipeline_agent_steps-1',
+      tool_name: 'bitcode.depository.search',
+      tool_input: { read: { id: 'read-1' }, assetCount: 1 },
+      tool_output: { resultState: 'worthy_fit', selectedCandidateCount: 1 },
+      tool_error: null,
+    });
+  });
 });

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

@@ -259,6 +259,63 @@ export function enablePipelineStreaming(
       }, 'id', deliverableRunId).catch(() => {});
     };
 
+    const ensurePhaseDelegationForEvent = async (
+      phaseId: string | null | undefined,
+      phaseName: string | null,
+      now: string
+    ): Promise<string | null> => {
+      if (phaseId) return phaseId;
+      if (!phaseName || !(await ensureDeliverableRun())) return null;
+      const existing = phaseIdByName.get(phaseName);
+      if (existing) return existing;
+
+      const row: import('../types/db').DPPhaseDelegationInsert = {
+        run_id: deliverableRunId,
+        phase_name: phaseName,
+        started_at: now,
+        completed_at: now,
+        status: 'completed',
+        input_data: { inferredFrom: 'tool-execution-event' },
+        output_data: { inferredFrom: 'tool-execution-event' },
+      } as any;
+      const { data, error } = await insertRow('deliverable_pipeline_phase_delegations', row, true);
+      if (!error && data?.id) {
+        phaseIdByName.set(phaseName, data.id);
+        return data.id;
+      }
+      return null;
+    };
+
+    const ensureAgentStepForEvent = async (
+      phaseId: string | null,
+      phaseName: string | null,
+      agentName: string | null,
+      stepType: string | null,
+      now: string
+    ): Promise<string | null> => {
+      if (!(phaseId && agentName && stepType)) return null;
+      const key = agentStepKey(phaseId, phaseName, agentName, stepType);
+      const existing = agentStepMap.get(key);
+      if (existing) return existing;
+
+      const row: import('../types/db').DPAgentStepInsert = {
+        phase_delegation_id: phaseId,
+        agent_name: agentName,
+        step_type: stepType,
+        started_at: now,
+        completed_at: now,
+        status: 'completed',
+        input_data: { inferredFrom: 'tool-execution-event' },
+        output_data: { inferredFrom: 'tool-execution-event' },
+      } as any;
+      const { data, error } = await insertRow('deliverable_pipeline_agent_steps', row, true);
+      if (!error && data?.id) {
+        agentStepMap.set(key, data.id);
+        return data.id;
+      }
+      return null;
+    };
+
     const completeOpenHierarchyRows = async (now: string) => {
       if (!(await ensureDeliverableRun())) return;
       const { data: openPhases } = await (supabase as any)
@@ -299,9 +356,12 @@ export function enablePipelineStreaming(
 
         const structuredTool = normalizeToolExecutionEvent(event);
         if (structuredTool && await ensureDeliverableRun()) {
-          const phaseId = phaseState.currentPhaseId || (phaseName ? phaseIdByName.get(phaseName) : null);
+          let phaseId = phaseState.currentPhaseId || (phaseName ? phaseIdByName.get(phaseName) : null);
+          phaseId = await ensurePhaseDelegationForEvent(phaseId ?? null, phaseName, now);
           const key = agentStepKey(phaseId ?? null, phaseName, agentName, stepType);
-          const agentStepId = agentStepMap.get(key) || null;
+          const agentStepId =
+            agentStepMap.get(key) ||
+            await ensureAgentStepForEvent(phaseId ?? null, phaseName, agentName, stepType, now);
           const row: import('../types/db').DPToolExecInsert = {
             agent_step_id: agentStepId,
             substep_id: null,

packages/pipelines/asset-pack/src/__tests__/finish-delivery.test.ts

@@ -94,14 +94,20 @@ describe('finish pull-request delivery', () => {
     expect(exec.get('tools', 'vcs_create_branch:0')).toMatchObject({
       tool: 'vcs_create_branch',
       ok: true,
+      phase: 'finish',
+      agent: 'finish:asset-pack-delivery-agent',
     });
     expect(exec.get('tools', 'vcs_create_or_update_file:1')).toMatchObject({
       tool: 'vcs_create_or_update_file',
       ok: true,
+      phase: 'finish',
+      agent: 'finish:asset-pack-delivery-agent',
     });
     expect(exec.get('tools', 'vcs_create_pull_request:2')).toMatchObject({
       tool: 'vcs_create_pull_request',
       ok: true,
+      phase: 'finish',
+      agent: 'finish:asset-pack-delivery-agent',
     });
     expect(emitToolUsage).toHaveBeenCalledTimes(3);
     expect(emitToolUsage.mock.calls[1][3]).toMatchObject({

packages/pipelines/asset-pack/src/agents/finish/deliver-asset-pack-to-destination-agent.ts

@@ -411,6 +411,10 @@ async function runDeliveryTool<T>(
       ok: true,
       input: summarizedInput,
       output: summarizedOutput,
+      phase: 'finish',
+      agent: 'finish:asset-pack-delivery-agent',
+      step: 'try',
+      generation: 'tools_execution',
     });
     await emitToolUsage(execution, toolName, summarizedInput, summarizedOutput, null);
     return output;
@@ -422,6 +426,10 @@ async function runDeliveryTool<T>(
       ok: false,
       input: summarizedInput,
       error: { message },
+      phase: 'finish',
+      agent: 'finish:asset-pack-delivery-agent',
+      step: 'try',
+      generation: 'tools_execution',
     });
     await emitToolUsage(execution, toolName, summarizedInput, null, message);
     throw error;