V48 Gate 3 (specification-implementation): relay raw live event_data so the telemetry log-line contract holds during streaming (F19)

geraldarthurdavis · claude · geraldarthurdavis · commit 4d8134a9ad56 · 2026-06-26T15:07:08.000-07:00
The F19 contract (only LLM calls + Tool uses become rows) held on reload but NOT during a live synthesis: fragments (`/Users/.../uapi`, `setup`, `try`, `setup:asset-pack-clone-vcs-repository-agent`, commit hash, `bit-engine`, `main`, `setup-plan`, `custom_pipeline`, ...) still rendered as rows.

Cause: usePipelineExecution has two paths. History relays raw `event_data` (the structured onStore event, namespace/key/executionState intact) — the activity-builder classifier keys off `namespace` to suppress fragments, so history was correct. The live SSE tail re-parsed each frame through parseStreamChunk, which flattened the event into a namespace-less `{type:'status', status:{message}}` shape; with the namespace stripped, the classifier's no-namespace-status rule kept every store as a fragment row. The live tail is the path active while a synthesis streams, so the fragments persisted on screen.

Fix (usePipelineExecution.ts): the live tail now relays raw `event_data` verbatim (already source-safe filtered server-side), identical in shape to the history path — no parseStreamChunk flattening. work-update frames still feed recordWorkUpdate. parseStreamChunk import removed (now unused here). The classifier therefore sees namespace/key/executionState live and suppresses fragments while keeping llm/output (generation) + tool result/error rows with full hierarchy.

Added a usePipelineExecution test asserting the live tail preserves namespace/key/executionState verbatim. Spec (formal log-line contract) + QA F19 updated to record that the contract depends on a consistent structured event shape live and on reload. uapi tsc 0; usePipelineExecution (3) + terminal/deposit/conversation/reading telemetry batch green; spec checker green.

Co-Authored-By: Claude Fable 5 &lt;noreply@anthropic.com&gt;
diff --git a/BITCODE_SPEC_V48_NOTES.md b/BITCODE_SPEC_V48_NOTES.md
@@ -217,7 +217,12 @@ Accepted V48 architecture law (decided 2026-06-25):
   null-separated row key; the renderer displays the text before the separator,
   keys details by the full key, and bypasses de-dup for uniquely-keyed rows. This
   is what removes the `try`/`setup-plan`/`thricified-generation`/path fragments
-  and stabilizes the pipeline↔UI payload contract.
+  and stabilizes the pipeline↔UI payload contract. The contract depends on the
+  client seeing the SAME structured event shape live as on reload: the live SSE
+  tail (`usePipelineExecution`) therefore relays raw `event_data` verbatim
+  (namespace/key/executionState intact) rather than re-parsing it — re-parsing
+  flattened events into namespace-less status frames, which defeated the
+  classifier and leaked every store as a fragment row during streaming.
 - **OTF (on-the-fly instructions) is eradicated entirely** as legacy residue:
   the dead telemetry types (`otf_instructions`/`otf_adherence`), the unused
   `setOnTheFly` prompt primitive, the live instruction-injection feature (the
diff --git a/BITCODE_V48_QA.md b/BITCODE_V48_QA.md
@@ -199,7 +199,8 @@ Track 3-4 scripts (BTD ledger, settlement, pack journaling) get added when those
 - Repair (2026-06-26, `terminal-run-activity.ts` + `pipeline-execution-log.tsx`):
   - The activity builder maintains a rolling `{phase, agent, step, failsafe, generation}` context updated from every event, and emits a row ONLY for LLM calls (`generation` / `llm:output`), Tool uses (`tool|tools:result|error`), and terminal/high-level signals (`completion`/`error`/no-namespace status). LLM rows stamp their own 5-field hierarchy; tool rows stamp Phase/Agent/Step from the rolling context + the tool name/args accumulated per tool-execution node.
   - Distinct calls that share withheld text are kept distinct via a unique null-separated row key (`TELEMETRY_ROW_KEY_SEP`); the renderer displays only the text before the separator, looks up details by the full key, and bypasses message de-dup for uniquely-keyed rows.
-- Verified: uapi tsc 0; F19 contract test (fragments suppressed, LLM row carries 5-pill hierarchy, tool row carries Phase/Agent/Step + tool) + 10-suite telemetry/terminal/deposit regression batch (38 tests) green.
+  - `usePipelineExecution.ts` (the second half — fragments persisted live) — the live SSE tail re-parsed each frame through `parseStreamChunk`, which flattened the structured onStore event into a namespace-less `{type:'status', status:{message}}` shape; the activity-builder classifier keys off `namespace` to suppress fragments, so namespace-stripped live events leaked every store as a row (the history path, which relays raw `event_data`, was already correct). The tail now relays raw `event_data` verbatim — identical in shape to history — so the contract holds during streaming, not just on reload.
+- Verified: uapi tsc 0; F19 activity-builder contract test + a `usePipelineExecution` test asserting the live tail preserves `namespace`/`key`/`executionState` + 10-suite telemetry/terminal/deposit regression batch — all green.
 
 ## Track 1 — Identity / Authentication / Auxillaries — COMPLETE 2026-06-12 (email deferred by decision; F2/F9 and legacy eradication queued for gates)
 
diff --git a/uapi/hooks/usePipelineExecution.ts b/uapi/hooks/usePipelineExecution.ts
@@ -1,5 +1,4 @@
 import { useState, useEffect, useRef, useCallback, useMemo } from 'react';
-import { parseStreamChunk } from '@/streaming/stream-parser';
 
 interface PipelineExecution {
   id: string;
@@ -82,60 +81,37 @@ export function usePipelineExecution(runId: string | null): UsePipelineExecution
           if (!reader) return;
           const decoder = new TextDecoder();
           let buffer = '';
+          let liveSeq = 0;
           for (;;) {
             const { done, value } = await reader.read();
             if (done) break;
             buffer += decoder.decode(value, { stream: true });
             const parts = buffer.split('\n\n');
             buffer = parts.pop() || '';
             for (const chunk of parts) {
-              const parsed = parseStreamChunk(chunk + '\n');
-              if (parsed.type === 'work-update' && parsed.update) {
-                recordWorkUpdate(parsed);
-                setEvents(prev => prev.concat([{ id: `${Date.now()}:work-update`, event: parsed, created_at: new Date().toISOString() }]));
-                continue;
-              }
-              if (parsed.status) {
-                const ev = { type: 'status', status: parsed.status };
-                setEvents(prev => prev.concat([{ id: `${Date.now()}:status`, event: ev, created_at: new Date().toISOString() }]));
-              }
-              if (parsed.text && !parsed.status && !parsed.error && !parsed.completion) {
-                const ev = { type: 'message', message: parsed.text } as any;
-                setEvents(prev => prev.concat([{ id: `${Date.now()}:msg`, event: ev, created_at: new Date().toISOString() }]));
-              }
-              if (parsed.error) {
-                const ev = { type: 'error', message: parsed.error } as any;
-                setEvents(prev => prev.concat([{ id: `${Date.now()}:err`, event: ev, created_at: new Date().toISOString() }]));
-              }
-              if ((parsed as any).type) {
-                const ev = (parsed as any).event || parsed;
-                if (ev?.type === 'work-update') {
-                  recordWorkUpdate(ev);
-                  setEvents(prev => prev.concat([{ id: `${Date.now()}:work-update`, event: ev, created_at: new Date().toISOString() }]));
-                  continue;
-                }
-                if (ev) {
-                  setEvents(prev => prev.concat([{ id: `${Date.now()}:ev`, event: ev, created_at: new Date().toISOString() }]));
-                }
-              }
+              // The live SSE tail relays raw `event_data` — the structured onStore
+              // event with namespace/key/executionState intact (already source-safe
+              // filtered server-side). Push it verbatim, identical in shape to the
+              // history path, so the activity builder classifies live events the
+              // same way and the formal log-line contract (F19: only LLM calls +
+              // Tool uses become rows) holds during streaming, not just on reload.
+              // Re-parsing through parseStreamChunk here used to flatten events into
+              // namespace-less `status`/`message` shapes, which leaked every
+              // intermediate store as a fragment row.
               const lines = chunk.split('\n').filter((l) => l.startsWith('data: ')).map((l) => l.substring(6));
               for (const line of lines) {
+                let payload: any;
                 try {
-                  const payload = JSON.parse(line);
-                  if (payload?.type === 'work-update') {
-                    recordWorkUpdate(payload);
-                  }
-                  if (
-                    payload &&
-                    (payload.type === 'pipeline' ||
-                      payload.type === 'phase' ||
-                      payload.type === 'agent' ||
-                      payload.type === 'completion' ||
-                      payload.type === 'error')
-                  ) {
-                    setEvents(prev => prev.concat([{ id: `${Date.now()}:ev`, event: payload, created_at: new Date().toISOString() }]));
-                  }
-                } catch {}
+                  payload = JSON.parse(line);
+                } catch {
+                  continue;
+                }
+                if (!payload || typeof payload !== 'object') continue;
+                if (payload.type === 'work-update') {
+                  recordWorkUpdate(payload);
+                }
+                const createdAt = typeof payload.timestamp === 'string' ? payload.timestamp : new Date().toISOString();
+                setEvents(prev => prev.concat([{ id: `live:${Date.now()}:${liveSeq++}`, event: payload, created_at: createdAt }]));
               }
             }
           }
diff --git a/uapi/tests/usePipelineExecution.test.tsx b/uapi/tests/usePipelineExecution.test.tsx
@@ -72,6 +72,58 @@ describe('usePipelineExecution', () => {
     expect((global.fetch as jest.Mock).mock.calls[0][0]).toBe('/api/executions/history/r1');
   });
 
+  it('relays raw live event_data with namespace/key intact (F19 fragment fix)', async () => {
+    const historyResponse = {
+      run: { id: 'r3', user_id: 'user-1', created_at: new Date().toISOString(), items: [], context: {} },
+      events: [],
+    };
+
+    // A namespaced fragment store and an llm/output generation, exactly as the
+    // server persists them. The live tail must push these verbatim (namespace
+    // preserved) rather than flattening them through parseStreamChunk.
+    const streamReader = {
+      read: jest
+        .fn()
+        .mockResolvedValueOnce({
+          done: false,
+          value: encoder.encode(
+            'data: {"type":"status","namespace":"step","key":"name","message":"try","executionState":{}}\n\n' +
+              'data: {"type":"generation","namespace":"llm","key":"output","message":"[content withheld — source-safe]","executionState":{"phase":"setup","agent":"A","step":"plan","failsafe":"prepare","generation":"reason"}}\n\n',
+          ),
+        })
+        .mockResolvedValueOnce({ done: true, value: undefined }),
+    };
+
+    global.fetch = jest.fn((request: RequestInfo) => {
+      const url = typeof request === 'string' ? request : (request as Request)?.url ?? '';
+      if (url.startsWith('/api/executions/history/')) {
+        return Promise.resolve({ ok: true, json: async () => historyResponse } as any);
+      }
+      if (url.startsWith('/api/executions/stream')) {
+        return Promise.resolve({
+          ok: true,
+          headers: { get: () => 'text/event-stream; charset=utf-8' },
+          body: { getReader: () => streamReader },
+        } as any);
+      }
+      throw new Error(`Unexpected fetch: ${url}`);
+    }) as any;
+
+    let latest: any;
+    render(<Harness runId="r3" onResult={(state) => (latest = state)} />);
+
+    await waitFor(() => expect(latest?.isLoading).toBe(false));
+    await waitFor(() =>
+      expect(latest.events.some((e: any) => e.event?.namespace === 'llm' && e.event?.key === 'output')).toBe(true),
+    );
+    const stepEvent = latest.events.find((e: any) => e.event?.namespace === 'step');
+    const llmEvent = latest.events.find((e: any) => e.event?.namespace === 'llm');
+    // Namespace/key survive — the activity builder can classify and suppress the
+    // `step/name=try` fragment while keeping the llm/output as a formal row.
+    expect(stepEvent?.event?.key).toBe('name');
+    expect(llmEvent?.event?.executionState).toMatchObject({ phase: 'setup', step: 'plan', generation: 'reason' });
+  });
+
   it('handles history fetch failure gracefully', async () => {
     global.fetch = jest.fn().mockResolvedValue({ ok: false, status: 500 });
 
