fix(trace-opencode): support opencode 1.14.x delta event stream

src/delta-synthesis.ts

@@ -0,0 +1,84 @@
+/**
+ * Shared synthesis of an LLM span from streamed deltas (opencode 1.14.x).
+ *
+ * Used when `message.updated` is not delivered to plugins, so the existing
+ * LLM-span path never fires. The fallback accumulates `message.part.delta`
+ * payloads into a per-message buffer; on `session.idle` we synthesize one
+ * LLM span per turn from the accumulated text.
+ *
+ * Both event-processor.ts (test-driving EventProcessor) and tracing.ts
+ * (production hooks) need this logic — extracted here so they can't drift.
+ */
+
+import type { SpanData } from "./client"
+import { msToSeconds } from "./clock"
+
+/**
+ * Structural type matching the fields the helper reads. Both SessionState
+ * shapes (in event-processor.ts and tracing.ts) satisfy this — no nominal
+ * import needed.
+ */
+export interface DeltaSynthesizableState {
+  currentTurnSpanId?: string
+  effectiveRootSpanId: string
+  systemPrompt?: string
+  currentInput?: string
+  currentTurnStartTime?: number
+  currentProviderID?: string
+  currentModelID?: string
+  deltaAccumulatedOutput: Map<string, string>
+}
+
+export function joinAccumulatedDeltas(state: DeltaSynthesizableState): string {
+  return Array.from(state.deltaAccumulatedOutput.values()).join("")
+}
+
+/**
+ * Build a synthesized LLM span for the current turn, or return `undefined`
+ * if there's nothing to synthesize (no accumulated text, or no turn span to
+ * attach to). Tokens are intentionally omitted (not zeroed) so the Braintrust
+ * dashboard doesn't render misleading "0 tokens" — that information is
+ * unrecoverable from the delta stream.
+ */
+export function buildSynthesizedLlmSpan(
+  state: DeltaSynthesizableState,
+  endSeconds: number,
+): SpanData | undefined {
+  const accumulated = joinAccumulatedDeltas(state)
+  if (!accumulated) return undefined
+  if (!state.currentTurnSpanId) return undefined
+
+  const providerID = state.currentProviderID ?? "unknown"
+  const modelID = state.currentModelID ?? "unknown"
+  const modelName = `${providerID}/${modelID}`
+
+  const llmInput: Array<Record<string, unknown>> = []
+  if (state.systemPrompt) llmInput.push({ role: "system", content: state.systemPrompt })
+  if (state.currentInput) llmInput.push({ role: "user", content: state.currentInput })
+
+  const id = crypto.randomUUID()
+  return {
+    id,
+    span_id: id,
+    root_span_id: state.effectiveRootSpanId,
+    span_parents: [state.currentTurnSpanId],
+    input: llmInput.length > 0 ? llmInput : undefined,
+    output: [{ role: "assistant", content: accumulated }],
+    metrics: {
+      start:
+        state.currentTurnStartTime !== undefined
+          ? msToSeconds(state.currentTurnStartTime)
+          : undefined,
+      end: endSeconds,
+    },
+    metadata: {
+      model: modelID,
+      provider: providerID,
+      synthesized_from: "session.idle",
+    },
+    span_attributes: {
+      name: modelName,
+      type: "llm",
+    },
+  }
+}

src/event-processor.ts

@@ -9,6 +9,7 @@ import type { Event } from "@opencode-ai/sdk"
 import type { SpanData } from "./client"
 import type { Clock } from "./clock"
 import { msToSeconds, wallClock } from "./clock"
+import { buildSynthesizedLlmSpan, joinAccumulatedDeltas } from "./delta-synthesis"
 import type { SpanSink } from "./span-sink"
 
 // Generate a UUID
@@ -44,6 +45,13 @@ interface SessionState {
   >
   llmReasoningParts: Map<string, string> // messageId -> reasoning/thinking text
   processedLlmMessages: Set<string>
+  // opencode 1.14.x fallback: model captured from chat.message hook (used when
+  // message.updated is never delivered), accumulated delta text per messageID,
+  // and a per-turn guard so we don't double-emit when both event paths fire.
+  currentProviderID?: string
+  currentModelID?: string
+  llmSpanEmittedForCurrentTurn?: boolean
+  deltaAccumulatedOutput: Map<string, string>
   // Tool span tracking
   toolStartTimes: Map<string, number>
   toolCallMessageIds: Map<string, string> // callID -> messageId (to look up reasoning)
@@ -94,6 +102,9 @@ export class EventProcessor {
       await this.handleSessionCreated(info)
     } else if (event.type === "message.part.updated") {
       await this.handleMessagePartUpdated(props)
+    } else if ((event.type as string) === "message.part.delta") {
+      // opencode 1.14+ event, not yet in the SDK's typed Event union.
+      await this.handleMessagePartDelta(props)
     } else if (event.type === "message.updated") {
       await this.handleMessageUpdated(props)
     } else if (event.type === "session.idle") {
@@ -143,6 +154,13 @@ export class EventProcessor {
     state.currentTurnSpanId = generateUUID()
     state.currentOutput = undefined
     state.currentInput = userMessage
+    state.deltaAccumulatedOutput.clear()
+    state.llmSpanEmittedForCurrentTurn = false
+
+    // Reset unconditionally so a turn without a model object doesn't inherit
+    // the previous turn's identity in the synthesized LLM span fallback.
+    state.currentProviderID = model?.providerID
+    state.currentModelID = model?.modelID
 
     const now = this.clock.now()
     const nowSeconds = msToSeconds(now)
@@ -225,6 +243,7 @@ export class EventProcessor {
       llmToolCalls: new Map(),
       llmReasoningParts: new Map(),
       processedLlmMessages: new Set(),
+      deltaAccumulatedOutput: new Map(),
       toolStartTimes: new Map(),
       toolCallMessageIds: new Map(),
       toolCallArgs: new Map(),
@@ -385,6 +404,7 @@ export class EventProcessor {
           llmToolCalls: new Map(),
           llmReasoningParts: new Map(),
           processedLlmMessages: new Set(),
+          deltaAccumulatedOutput: new Map(),
           toolStartTimes: new Map(),
           toolCallMessageIds: new Map(),
           toolCallArgs: new Map(),
@@ -447,6 +467,7 @@ export class EventProcessor {
       llmToolCalls: new Map(),
       llmReasoningParts: new Map(),
       processedLlmMessages: new Set(),
+      deltaAccumulatedOutput: new Map(),
       toolStartTimes: new Map(),
       toolCallMessageIds: new Map(),
       toolCallArgs: new Map(),
@@ -560,6 +581,31 @@ export class EventProcessor {
     }
   }
 
+  /**
+   * Accumulate streamed deltas from opencode 1.14.x. The event payload is
+   * { sessionID, messageID, partID, field, delta }. The `field` is always
+   * "text" for both text and reasoning parts, so we cannot distinguish them
+   * here; we concatenate every delta into one buffer keyed by messageID.
+   * The synthesized LLM span on session.idle reads from this buffer.
+   */
+  private async handleMessagePartDelta(props: Record<string, unknown>): Promise<void> {
+    const sessionID = props.sessionID as string
+    const messageID = props.messageID as string
+    const field = props.field as string
+    const delta = props.delta as string
+
+    if (!sessionID || !messageID || !delta) return
+    // Only accumulate text-field deltas. Other fields (e.g., a future
+    // `tool_input` for streamed tool-call args) must not corrupt assistant text.
+    if (field !== "text") return
+
+    const state = this.sessionStates.get(sessionID)
+    if (!state) return
+
+    const prev = state.deltaAccumulatedOutput.get(messageID) ?? ""
+    state.deltaAccumulatedOutput.set(messageID, prev + delta)
+  }
+
   private async handleMessageUpdated(props: Record<string, unknown>): Promise<void> {
     const messageInfo = props.info as Record<string, unknown> | undefined
     if (!messageInfo) {
@@ -678,10 +724,27 @@ export class EventProcessor {
       },
     }
 
+    // Set the guard BEFORE awaiting so a failed insert doesn't leave the flag
+    // false and let session.idle synthesize a duplicate span.
+    state.llmSpanEmittedForCurrentTurn = true
     await this.spanSink.insertSpan(llmSpan)
     this.log("Created LLM span", { messageId, modelName, tokens: totalTokens })
   }
 
+  private async synthesizeLlmSpanFromDeltas(
+    state: SessionState,
+    endSeconds: number,
+  ): Promise<void> {
+    const span = buildSynthesizedLlmSpan(state, endSeconds)
+    if (!span) return
+    // Set the guard BEFORE awaiting (see handleMessageUpdated for rationale).
+    state.llmSpanEmittedForCurrentTurn = true
+    await this.spanSink.insertSpan(span)
+    this.log("Synthesized LLM span from deltas", {
+      modelName: span.span_attributes?.name,
+    })
+  }
+
   private async handleSessionIdle(sessionID: string | undefined): Promise<void> {
     if (!sessionID) {
       this.log("session.idle but no session ID found")
@@ -695,13 +758,23 @@ export class EventProcessor {
       const now = this.clock.nowSeconds()
       const isChildSession = !!state.parentSessionId
 
+      // opencode 1.14.x fallback: if no message.updated landed for this turn
+      // but we accumulated streamed deltas, synthesize an LLM span before
+      // closing the turn so the trace isn't empty.
+      if (state.currentTurnSpanId && !state.llmSpanEmittedForCurrentTurn) {
+        await this.synthesizeLlmSpanFromDeltas(state, now)
+      }
+
       // Close current turn span if exists
       if (state.currentTurnSpanId) {
+        // currentOutput may still be empty when only deltas arrived; fall back to
+        // the accumulated delta buffer so the turn span shows the assistant text.
+        const fallbackOutput = state.currentOutput || joinAccumulatedDeltas(state) || undefined
         const turnSpan: SpanData = {
           id: state.currentTurnSpanId,
           span_id: state.currentTurnSpanId,
           root_span_id: state.effectiveRootSpanId,
-          output: state.currentOutput || undefined,
+          output: fallbackOutput,
           metrics: {
             end: now,
           },
@@ -712,6 +785,8 @@ export class EventProcessor {
         state.currentInput = undefined
         state.currentOutput = undefined
         state.currentTurnStartTime = undefined
+        state.deltaAccumulatedOutput.clear()
+        state.llmSpanEmittedForCurrentTurn = false
         this.log("Turn span closed", { sessionKey, turnNumber: state.turnNumber })
       }
 

src/test-helpers.ts

@@ -290,6 +290,28 @@ export function toolCallCompletedPart(
   }
 }
 
+/**
+ * Create a message.part.delta event (opencode 1.14.x event stream).
+ * Not yet in @opencode-ai/sdk's typed Event union, so we cast through unknown.
+ */
+export function messagePartDelta(
+  sessionID: string,
+  messageID: string,
+  delta: string,
+  options?: { partID?: string; field?: string },
+): Event {
+  return {
+    type: "message.part.delta",
+    properties: {
+      sessionID,
+      messageID,
+      partID: options?.partID ?? `prt_text_${messageID}`,
+      field: options?.field ?? "text",
+      delta,
+    },
+  } as unknown as Event
+}
+
 /**
  * Create message.updated event for assistant message completion
  */