feat(team-mode): stream message.part.delta events to member FIFO

code-yeongyu · code-yeongyu · commit 2b4a3915ccf8 · 2026-04-18T05:10:09.000+09:00
SDK v1 Event union does not include EventMessagePartDelta, but OpenCode >=1.2.0 emits message.part.delta for streaming reasoning / text chunks (see background-agent/manager.ts line 1007). Without this handling, long provider responses that arrive as incremental deltas never reach the tmux panes, making live streaming effectively broken for any non-trivial output. Extend the hook with a narrow custom union covering the runtime shape { sessionID, partID?, field?, delta } and write deltas straight to the member FIFO. Add unit tests that cover: - multiple deltas appending to the same FIFO - non-text fields (e.g., tool) being ignored Oracle review blocker (reported by oracle session verifying #3493).
diff --git a/src/hooks/team-session-streamer/hook.test.ts b/src/hooks/team-session-streamer/hook.test.ts
@@ -95,4 +95,77 @@ describe("createTeamSessionStreamer", () => {
     expect(listActiveTeams).toHaveBeenCalledTimes(1)
     expect(loadRuntimeState).toHaveBeenCalledTimes(1)
   })
+
+  test("streams incremental message.part.delta events to member fifo", async () => {
+    // given
+    const listActiveTeams = mock(async () => [{
+      teamRunId: "11111111-1111-4111-8111-111111111111",
+      teamName: "team-alpha",
+      status: "active",
+      memberCount: 1,
+      scope: "project" as const,
+    }])
+    const loadRuntimeState = mock(async () => createRuntimeState())
+    const config = TeamModeConfigSchema.parse({ enabled: true, tmux_visualization: true })
+    const streamer = createTeamSessionStreamer(config, { listActiveTeams, loadRuntimeState })
+
+    // when
+    await streamer.event({
+      event: {
+        type: "message.part.delta",
+        properties: {
+          sessionID: "member-session",
+          partID: "part-delta",
+          field: "text",
+          delta: "chunk-one ",
+        },
+      },
+    })
+    await streamer.event({
+      event: {
+        type: "message.part.delta",
+        properties: {
+          sessionID: "member-session",
+          partID: "part-delta",
+          field: "text",
+          delta: "chunk-two",
+        },
+      },
+    })
+
+    // then
+    expect(writeTeamSessionFifoMock).toHaveBeenCalledTimes(2)
+    expect(writeTeamSessionFifoMock).toHaveBeenNthCalledWith(1, "/tmp/omo-team/11111111-1111-4111-8111-111111111111/member-a.fifo", "chunk-one ")
+    expect(writeTeamSessionFifoMock).toHaveBeenNthCalledWith(2, "/tmp/omo-team/11111111-1111-4111-8111-111111111111/member-a.fifo", "chunk-two")
+  })
+
+  test("ignores delta events for non-text fields", async () => {
+    // given
+    const listActiveTeams = mock(async () => [{
+      teamRunId: "11111111-1111-4111-8111-111111111111",
+      teamName: "team-alpha",
+      status: "active",
+      memberCount: 1,
+      scope: "project" as const,
+    }])
+    const loadRuntimeState = mock(async () => createRuntimeState())
+    const config = TeamModeConfigSchema.parse({ enabled: true, tmux_visualization: true })
+    const streamer = createTeamSessionStreamer(config, { listActiveTeams, loadRuntimeState })
+
+    // when
+    await streamer.event({
+      event: {
+        type: "message.part.delta",
+        properties: {
+          sessionID: "member-session",
+          partID: "part-other",
+          field: "tool",
+          delta: "ignored",
+        },
+      },
+    })
+
+    // then
+    expect(writeTeamSessionFifoMock).not.toHaveBeenCalled()
+  })
 })
diff --git a/src/hooks/team-session-streamer/hook.ts b/src/hooks/team-session-streamer/hook.ts
@@ -1,3 +1,5 @@
+import type { Event, EventMessagePartUpdated, Part } from "@opencode-ai/sdk"
+
 import type { TeamModeConfig } from "../../config/schema/team-mode"
 import { getTeamMemberFifoPath } from "../../features/team-mode/team-layout-tmux/fifo-path"
 import * as teamStateStore from "../../features/team-mode/team-state-store"
@@ -6,106 +8,80 @@ import { writeTeamSessionFifo } from "./fifo-writer"
 
 type TeamStateStore = Pick<typeof teamStateStore, "listActiveTeams" | "loadRuntimeState">
 
+type MessagePartDeltaEvent = {
+  type: "message.part.delta"
+  properties: {
+    sessionID: string
+    partID?: string
+    field?: string
+    delta: string
+  }
+}
+
+type StreamEvent = Event | MessagePartDeltaEvent
+
 type TeamSessionStreamTarget = {
   teamRunId: string
   memberName: string
   fifoPath: string
 }
 
-type TeamSessionStreamEvent = {
-  type: string
-  properties?: unknown
-}
-
-type HookInput = {
-  event?: unknown
-}
-
-type HookImpl = {
-  event: (input: HookInput) => Promise<void>
-}
+type HookInput = { event: StreamEvent }
+type HookImpl = { event: (input: HookInput) => Promise<void> }
 
 const DROPPABLE_FIFO_ERROR_CODES = new Set(["ENXIO", "ENOENT", "EPIPE"])
 
-function isRecord(value: unknown): value is Record<string, unknown> {
-  return typeof value === "object" && value !== null
-}
-
 function isErrorWithCode(error: unknown): error is Error & { code: string } {
   return error instanceof Error && "code" in error && typeof error.code === "string"
 }
 
-function normalizeEvent(input: HookInput): TeamSessionStreamEvent | undefined {
-  if (!isRecord(input.event) || typeof input.event.type !== "string") return undefined
-  return {
-    type: input.event.type,
-    properties: input.event.properties,
-  }
+function extractCumulativeText(part: Part): string | undefined {
+  if ("text" in part && typeof part.text === "string") return part.text
+  return undefined
 }
 
-function getDeletedSessionID(properties: unknown): string | undefined {
-  if (!isRecord(properties) || !isRecord(properties.info)) return undefined
-  return typeof properties.info.id === "string" ? properties.info.id : undefined
-}
+function extractUpdateSegment(
+  event: EventMessagePartUpdated,
+  partTextByKey: Map<string, string>,
+): { sessionID: string; text: string } | undefined {
+  const part = event.properties.part
+  const delta = event.properties.delta
+  const partKey = `${part.sessionID}:${part.id}`
 
-function getRemovedPartRef(properties: unknown): { sessionID: string; partID: string } | undefined {
-  if (!isRecord(properties)) return undefined
-  return typeof properties.sessionID === "string" && typeof properties.partID === "string"
-    ? { sessionID: properties.sessionID, partID: properties.partID }
-    : undefined
-}
+  if (typeof delta === "string" && delta.length > 0) {
+    const previousText = partTextByKey.get(partKey) ?? ""
+    partTextByKey.set(partKey, previousText + delta)
+    return { sessionID: part.sessionID, text: delta }
+  }
 
-function getErroredSessionID(properties: unknown): string | undefined {
-  return isRecord(properties) && typeof properties.sessionID === "string" ? properties.sessionID : undefined
-}
+  const cumulativeText = extractCumulativeText(part)
+  if (cumulativeText === undefined) return undefined
 
-function getTextPart(properties: unknown): { sessionID: string; partID: string; text: string } | undefined {
-  if (!isRecord(properties) || !isRecord(properties.part)) {
-    return undefined
-  }
+  const previousText = partTextByKey.get(partKey) ?? ""
+  partTextByKey.set(partKey, cumulativeText)
 
-  const part = properties.part
-  const sessionID = typeof properties.sessionID === "string"
-    ? properties.sessionID
-    : typeof part.sessionID === "string"
-      ? part.sessionID
-      : undefined
-
-  if (!sessionID) return undefined
-  if (typeof part.id !== "string") return undefined
-  if (typeof part.text === "string") {
-    return { sessionID, partID: part.id, text: part.text }
-  }
-  if (typeof part.content === "string") {
-    return { sessionID, partID: part.id, text: part.content }
-  }
+  const appendedText = cumulativeText.startsWith(previousText)
+    ? cumulativeText.slice(previousText.length)
+    : cumulativeText
+  if (appendedText.length === 0) return undefined
+  return { sessionID: part.sessionID, text: appendedText }
 }
 
-function getUpdatedSegment(
-  properties: unknown,
+function extractDeltaSegment(
+  event: MessagePartDeltaEvent,
   partTextByKey: Map<string, string>,
 ): { sessionID: string; text: string } | undefined {
-  if (!isRecord(properties)) return undefined
-
-  const delta = typeof properties.delta === "string" ? properties.delta : undefined
-  const part = getTextPart(properties)
-  if (!part) return undefined
-  const partKey = `${part.sessionID}:${part.partID}`
+  const { sessionID, partID, field, delta } = event.properties
+  if (typeof delta !== "string" || delta.length === 0) return undefined
+  if (field !== undefined && field !== "text" && field !== "content") return undefined
 
-  if (delta && delta.length > 0) {
+  if (partID) {
+    const partKey = `${sessionID}:${partID}`
     const previousText = partTextByKey.get(partKey) ?? ""
     partTextByKey.set(partKey, previousText + delta)
-    return { sessionID: part.sessionID, text: delta }
   }
 
-  const previousText = partTextByKey.get(partKey) ?? ""
-  partTextByKey.set(partKey, part.text)
-  const appendedText = part.text.startsWith(previousText)
-    ? part.text.slice(previousText.length)
-    : part.text
-
-  if (appendedText.length === 0) return undefined
-  return { sessionID: part.sessionID, text: appendedText }
+  return { sessionID, text: delta }
 }
 
 function clearSessionPartState(sessionID: string, partTextByKey: Map<string, string>): void {
@@ -151,61 +127,71 @@ export function createTeamSessionStreamer(config: TeamModeConfig, stateStore: Te
     return undefined
   }
 
+  async function writeSegment(
+    sessionID: string,
+    text: string,
+  ): Promise<void> {
+    const target = await resolveStreamTarget(sessionID)
+    if (!target) return
+
+    try {
+      await writeTeamSessionFifo(target.fifoPath, text)
+    } catch (error) {
+      if (isErrorWithCode(error) && DROPPABLE_FIFO_ERROR_CODES.has(error.code)) {
+        if (error.code === "ENOENT") {
+          streamTargetsBySession.delete(sessionID)
+        }
+        return
+      }
+
+      log("team session streamer write failed", {
+        event: "team-mode-session-streamer-write-error",
+        teamRunId: target.teamRunId,
+        memberName: target.memberName,
+        sessionID,
+        fifoPath: target.fifoPath,
+        error: error instanceof Error ? error.message : String(error),
+      })
+    }
+  }
+
   return {
-    event: async (input: HookInput): Promise<void> => {
-      const event = normalizeEvent(input)
-      if (!event) return
+    event: async ({ event }: HookInput): Promise<void> => {
       if (!config.enabled || !config.tmux_visualization) return
 
       if (event.type === "session.deleted") {
-        const sessionID = getDeletedSessionID(event.properties)
-        if (!sessionID) return
+        const sessionID = event.properties.info.id
         streamTargetsBySession.delete(sessionID)
         clearSessionPartState(sessionID, partTextByKey)
         return
       }
 
       if (event.type === "message.part.removed") {
-        const removedPart = getRemovedPartRef(event.properties)
-        if (!removedPart) return
-        partTextByKey.delete(`${removedPart.sessionID}:${removedPart.partID}`)
+        partTextByKey.delete(`${event.properties.sessionID}:${event.properties.partID}`)
         return
       }
 
       if (event.type === "session.error") {
-        const sessionID = getErroredSessionID(event.properties)
-        if (!sessionID) return
-        streamTargetsBySession.delete(sessionID)
-        clearSessionPartState(sessionID, partTextByKey)
+        const sessionID = event.properties.sessionID
+        if (sessionID) {
+          streamTargetsBySession.delete(sessionID)
+          clearSessionPartState(sessionID, partTextByKey)
+        }
+        return
+      }
+
+      if (event.type === "message.part.delta") {
+        const segment = extractDeltaSegment(event, partTextByKey)
+        if (!segment) return
+        await writeSegment(segment.sessionID, segment.text)
         return
       }
 
       if (event.type !== "message.part.updated") return
-      const segment = getUpdatedSegment(event.properties, partTextByKey)
+      const segment = extractUpdateSegment(event, partTextByKey)
       if (!segment) return
 
-      const target = await resolveStreamTarget(segment.sessionID)
-      if (!target) return
-
-      try {
-        await writeTeamSessionFifo(target.fifoPath, segment.text)
-      } catch (error) {
-        if (isErrorWithCode(error) && DROPPABLE_FIFO_ERROR_CODES.has(error.code)) {
-          if (error.code === "ENOENT") {
-            streamTargetsBySession.delete(segment.sessionID)
-          }
-          return
-        }
-
-        log("team session streamer write failed", {
-          event: "team-mode-session-streamer-write-error",
-          teamRunId: target.teamRunId,
-          memberName: target.memberName,
-          sessionID: segment.sessionID,
-          fifoPath: target.fifoPath,
-          error: error instanceof Error ? error.message : String(error),
-        })
-      }
+      await writeSegment(segment.sessionID, segment.text)
     },
   }
 }
