fix(team-mode): close peer message delivery races

2026-05-19 18:23:57 +09:00
parent bcea4a9d28
commit b2918fd4db
13 changed files with 492 additions and 47 deletions
@@ -711,6 +711,46 @@ describe("dispatchInternalPrompt shared gate behavior", () => {
    expect(promptCalls).toBe(0)
  })

+  test("#given latest assistant turn has unknown finish #when an internal promptAsync is requested #then no prompt is sent", async () => {
+    // given
+    let promptCalls = 0
+    const client = {
+      session: {
+        status: async () => ({ data: { ses_unknown_finish: { type: "idle" } } }),
+        messages: async () => ({
+          data: [
+            {
+              info: { id: "msg_user", role: "user" },
+              parts: [{ type: "text", text: "run work" }],
+            },
+            {
+              info: { id: "msg_assistant", role: "assistant", finish: "unknown" },
+              parts: [{ type: "reasoning", text: "still resolving" }],
+            },
+          ],
+        }),
+        promptAsync: async () => {
+          promptCalls += 1
+        },
+      },
+    }
+
+    // when
+    const result = await dispatchInternalPrompt({
+      mode: "async",
+      client,
+      sessionID: "ses_unknown_finish",
+      input: { path: { id: "ses_unknown_finish" }, body: { parts: [] } },
+      source: "test:unknown-finish",
+      settleMs: 0,
+      postDispatchHoldMs: 0,
+    })
+
+    // then
+    expect(result.status).toBe("queued")
+    expect(promptCalls).toBe(0)
+  })
+
  test("#given internal user tail follows an assistant waiting on tools #when an internal promptAsync is requested #then no prompt is sent", async () => {
    // given
    let promptCalls = 0
@@ -567,6 +567,48 @@ describe("createTeamIdleWakeHint", () => {
    expect(processedEntries).toContain(`${messageId}.json`)
  })

+  test("#given stale idle while pending live delivery is still busy #when idle wake runs #then it keeps the reservation pending", async () => {
+    // given
+    const baseDir = await createTemporaryBaseDir()
+    const config = createConfig(baseDir)
+    const teamRunId = randomUUID()
+    const messageId = randomUUID()
+    await seedRuntimeState(createRuntimeState(teamRunId, [messageId]), config)
+    await seedReservedUnreadMessage(teamRunId, config, messageId, "live delivery body", 100)
+
+    const ackSpy = spyOn(ackModule, "ackMessages")
+    const promptAsyncSpy = mock(async (_input: WakeHintPromptInput) => ({}))
+    const handler = createTeamIdleWakeHint({
+      directory: "/tmp/project",
+      client: {
+        session: {
+          promptAsync: promptAsyncSpy,
+          status: async () => ({ data: { "member-session": { type: "busy" } } }),
+        },
+      },
+    }, config, { idleSettleMs: 0 })
+
+    // when
+    await handler({
+      event: {
+        type: "session.idle",
+        properties: { sessionID: "member-session" },
+      },
+    })
+
+    // then
+    expect(ackSpy).not.toHaveBeenCalled()
+    expect(promptAsyncSpy).not.toHaveBeenCalled()
+
+    const runtimeState = await loadRuntimeState(teamRunId, config)
+    expect(runtimeState.members[0]?.pendingInjectedMessageIds).toEqual([messageId])
+
+    const inboxDir = getInboxDir(resolveBaseDir(config), teamRunId, "worker")
+    const inboxEntries = await readdir(inboxDir)
+    expect(inboxEntries).toContain(`.delivering-${messageId}.json`)
+    expect(inboxEntries).not.toContain(`${messageId}.json`)
+  })
+
  test("#given a pending live-delivery ack and later unread message #when member idles after the live reply #then it wakes the member for the unread message", async () => {
    // given
    const baseDir = await createTemporaryBaseDir()
@@ -10,6 +10,7 @@ import { loadRuntimeState, transitionRuntimeState } from "../../features/team-mo
 import { resolveSessionEventID } from "../../shared/event-session-id"
 import { isAmbiguousPromptDispatchFailure } from "../../shared/prompt-failure-classifier"
 import { log } from "../../shared/logger"
+import { isSessionActive, settleAfterSessionIdle } from "../../shared/session-idle-settle"
 import { dispatchInternalPrompt, isInternalPromptDispatchAccepted } from "../shared/prompt-async-gate"

 type PromptAsyncInput = {
@@ -73,6 +74,20 @@ export function createTeamIdleWakeHint(ctx: TeamIdleWakeHintContext, config: Tea

      const pendingInjectedMessageIds = [...memberEntry.pendingInjectedMessageIds]
      if (pendingInjectedMessageIds.length > 0) {
+        if (typeof ctx.client.session.status === "function") {
+          await settleAfterSessionIdle(options?.idleSettleMs ?? 0)
+          if (await isSessionActive(ctx.client, sessionID)) {
+            log("team idle pending ack skipped while session remains active", {
+              event: "team-mode-idle-pending-ack-active",
+              teamRunId: runtimeState.teamRunId,
+              memberName: memberEntry.name,
+              sessionID,
+              pendingCount: pendingInjectedMessageIds.length,
+            })
+            return
+          }
+        }
+
        await ackMessages(runtimeState.teamRunId, memberEntry.name, pendingInjectedMessageIds, config)
        await transitionRuntimeState(runtimeState.teamRunId, (currentRuntimeState) => ({
          ...currentRuntimeState,
@@ -226,4 +226,89 @@ describe("createTeamMemberErrorHandler", () => {
    expect(inboxEntries).not.toContain(`.delivering-${messageId}.json`)
    expect(inboxEntries).not.toContain("processed")
  })
+
+  test("#given session.error arrives while OpenCode still reports busy #when pending live delivery exists #then it does not requeue a duplicate peer message", async () => {
+    // given
+    const baseDir = await createTemporaryBaseDir()
+    const config = createConfig(baseDir)
+    const teamRunId = randomUUID()
+    const messageId = randomUUID()
+    await seedRuntimeState(createRuntimeStateWithPendingMessage(teamRunId, messageId), config)
+    await seedReservedMessage(teamRunId, config, messageId)
+    const handler = createTeamMemberErrorHandler(config, {
+      settleMs: 0,
+      client: {
+        session: {
+          status: async () => ({ data: { "member-session": { type: "busy" } } }),
+        },
+      },
+    })
+
+    // when
+    await handler({
+      event: {
+        type: "session.error",
+        properties: { sessionID: "member-session", error: new Error("transient provider error") },
+      },
+    })
+
+    // then
+    const runtimeState = await loadRuntimeState(teamRunId, config)
+    expect(runtimeState.members[0]?.status).toBe("running")
+    expect(runtimeState.members[0]?.pendingInjectedMessageIds).toEqual([messageId])
+
+    const inboxEntries = await readdir(getInboxDir(resolveBaseDir(config), teamRunId, "worker"))
+    expect(inboxEntries).toContain(`.delivering-${messageId}.json`)
+    expect(inboxEntries).not.toContain(`${messageId}.json`)
+    expect(inboxEntries).not.toContain("processed")
+  })
+
+  test("#given session.error arrives after peer message reached history #when pending live delivery exists #then it does not requeue a duplicate peer message", async () => {
+    // given
+    const baseDir = await createTemporaryBaseDir()
+    const config = createConfig(baseDir)
+    const teamRunId = randomUUID()
+    const messageId = randomUUID()
+    await seedRuntimeState(createRuntimeStateWithPendingMessage(teamRunId, messageId), config)
+    await seedReservedMessage(teamRunId, config, messageId)
+    const handler = createTeamMemberErrorHandler(config, {
+      settleMs: 0,
+      client: {
+        session: {
+          status: async () => ({ data: { "member-session": { type: "idle" } } }),
+          messages: async () => ({
+            data: [
+              {
+                info: { role: "user" },
+                parts: [
+                  {
+                    type: "text",
+                    text: `<peer_message from="lead" messageId="${messageId}" kind="message">pending live delivery</peer_message>`,
+                  },
+                ],
+              },
+            ],
+          }),
+        },
+      },
+    })
+
+    // when
+    await handler({
+      event: {
+        type: "session.error",
+        properties: { sessionID: "member-session", error: new Error("late session.error after accepted prompt") },
+      },
+    })
+
+    // then
+    const runtimeState = await loadRuntimeState(teamRunId, config)
+    expect(runtimeState.members[0]?.status).toBe("running")
+    expect(runtimeState.members[0]?.pendingInjectedMessageIds).toEqual([messageId])
+
+    const inboxEntries = await readdir(getInboxDir(resolveBaseDir(config), teamRunId, "worker"))
+    expect(inboxEntries).toContain(`.delivering-${messageId}.json`)
+    expect(inboxEntries).not.toContain(`${messageId}.json`)
+    expect(inboxEntries).not.toContain("processed")
+  })
 })
@@ -7,9 +7,23 @@ import {
 import { loadRuntimeState, transitionRuntimeState } from "../../features/team-mode/team-state-store/store"
 import { resolveSessionEventID } from "../../shared/event-session-id"
 import { log } from "../../shared/logger"
+import {
+  DEFAULT_SESSION_IDLE_SETTLE_MS,
+  isSessionActive,
+  settleAfterSessionIdle,
+} from "../../shared/session-idle-settle"

 type HookInput = { event: { type: string; properties?: unknown } }
 export type HookImpl = (input: HookInput) => Promise<void>
+type TeamMemberErrorHandlerDeps = {
+  client?: {
+    session?: {
+      status?: () => Promise<unknown>
+      messages?: (input: { path: { id: string } }) => Promise<unknown>
+    }
+  }
+  settleMs?: number
+}

 function getErroredSessionID(properties: unknown): string | undefined {
  return resolveSessionEventID(properties)
@@ -31,7 +45,73 @@ async function requeuePendingLiveDeliveries(
  }
 }

-export function createTeamMemberErrorHandler(config: TeamModeConfig): HookImpl {
+async function shouldKeepPendingLiveDeliveries(
+  deps: TeamMemberErrorHandlerDeps,
+  sessionID: string,
+): Promise<boolean> {
+  if (typeof deps.client?.session?.status !== "function") {
+    return false
+  }
+
+  await settleAfterSessionIdle(deps.settleMs ?? DEFAULT_SESSION_IDLE_SETTLE_MS)
+  return await isSessionActive(deps.client, sessionID)
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null
+}
+
+function getMessagesData(response: unknown): unknown[] {
+  if (isRecord(response) && Array.isArray(response.data)) {
+    return response.data
+  }
+
+  return Array.isArray(response) ? response : []
+}
+
+function valueContainsAnyMessageId(value: unknown, messageIds: ReadonlySet<string>): boolean {
+  if (typeof value === "string") {
+    return [...messageIds].some((messageId) => value.includes(messageId))
+  }
+
+  if (Array.isArray(value)) {
+    return value.some((entry) => valueContainsAnyMessageId(entry, messageIds))
+  }
+
+  if (isRecord(value)) {
+    return Object.values(value).some((entry) => valueContainsAnyMessageId(entry, messageIds))
+  }
+
+  return false
+}
+
+async function sessionHistoryContainsPendingMessage(
+  deps: TeamMemberErrorHandlerDeps,
+  sessionID: string,
+  messageIds: readonly string[],
+): Promise<boolean> {
+  if (messageIds.length === 0 || typeof deps.client?.session?.messages !== "function") {
+    return false
+  }
+
+  try {
+    const response = await deps.client.session.messages({ path: { id: sessionID } })
+    const pendingMessageIds = new Set(messageIds)
+    return getMessagesData(response).some((message) => valueContainsAnyMessageId(message, pendingMessageIds))
+  } catch (error) {
+    log("team member session history check failed", {
+      event: "team-mode-member-error-history-check-failed",
+      sessionID,
+      error: error instanceof Error ? error.message : String(error),
+    })
+    return false
+  }
+}
+
+export function createTeamMemberErrorHandler(
+  config: TeamModeConfig,
+  deps: TeamMemberErrorHandlerDeps = {},
+): HookImpl {
  return async ({ event }: HookInput): Promise<void> => {
    if (event.type !== "session.error") return

@@ -47,6 +127,29 @@ export function createTeamMemberErrorHandler(config: TeamModeConfig): HookImpl {
      const runtimeState = await loadRuntimeState(runtimeMember.teamRunId, config)
      const memberEntry = runtimeState.members.find((member) => member.name === runtimeMember.memberName)
      const pendingInjectedMessageIds = memberEntry?.pendingInjectedMessageIds ?? []
+      if (await shouldKeepPendingLiveDeliveries(deps, erroredSessionID)) {
+        log("team member session error ignored while session remains active", {
+          event: "team-mode-member-error-active",
+          teamRunId: runtimeState.teamRunId,
+          teamName: runtimeState.teamName,
+          memberName: runtimeMember.memberName,
+          sessionID: erroredSessionID,
+          pendingCount: pendingInjectedMessageIds.length,
+        })
+        return
+      }
+      if (await sessionHistoryContainsPendingMessage(deps, erroredSessionID, pendingInjectedMessageIds)) {
+        log("team member session error ignored after pending peer message reached history", {
+          event: "team-mode-member-error-peer-message-accepted",
+          teamRunId: runtimeState.teamRunId,
+          teamName: runtimeState.teamName,
+          memberName: runtimeMember.memberName,
+          sessionID: erroredSessionID,
+          pendingCount: pendingInjectedMessageIds.length,
+        })
+        return
+      }
+
      await requeuePendingLiveDeliveries(
        runtimeState.teamRunId,
        runtimeMember.memberName,