Merge pull request #4195 from code-yeongyu/fix/team-send-message-ambiguous-delivery-loss

fix(team-mode): release reservation on ambiguous failure, commit on success-path mark failure

src/features/team-mode/team-state-store/resume.test.ts

@@ -345,7 +345,7 @@ describe("resumeAllTeams", () => {
     expect(worker?.pendingInjectedMessageIds).toEqual([])
   })
 
-  test("#given accepted live delivery lost its pending mark #when stale reservation is reclaimed #then resume processes it instead of exposing duplicate unread", async () => {
+  test("#given accepted live delivery lost its pending mark #when stale reservation is reclaimed #then resume removes the hidden reservation without losing the message", async () => {
     // given
     const baseDir = await createTemporaryBaseDir()
     temporaryDirectories.push(baseDir)
@@ -380,34 +380,33 @@ describe("resumeAllTeams", () => {
     }))
     const ancientMtime = new Date(Date.now() - 60 * 60 * 1000)
     await utimes(reservedPath, ancientMtime, ancientMtime)
-    const sessionGet = mock(async () => ({ data: { id: "alive" } }))
+    const sessionGet: SessionGetMock = async () => ({ data: { id: "alive" } })
-    const sessionMessages = mock(async ({ path: sessionPath }: { path: { id: string } }) => ({
+    const sessionMessages: SessionMessagesMock = async () => ({
-      data: sessionPath.id === "ses_worker"
+      data: [
-        ? [
+        {
-          {
+          info: { role: "user" },
-            info: { role: "user" },
+          parts: [
-            parts: [
+            {
-              {
+              type: "text",
-                type: "text",
+              text: `<peer_message from="lead" messageId="${workerMessageId}" kind="message">already accepted</peer_message>`,
-                text: `<peer_message from="lead" messageId="${workerMessageId}" kind="message">already accepted</peer_message>`,
+            },
-              },
+          ],
-            ],
+        },
-          },
+      ],
-        ]
+    })
-        : [],
-    }))
 
     // when
     await resumeAllTeams(createExecutorContext(baseDir, sessionGet, sessionMessages), config)
 
     // then
     const entries = await readdir(workerInbox)
-    expect(entries).not.toContain(`${workerMessageId}.json`)
     expect(entries).not.toContain(`.delivering-${workerMessageId}.json`)
-    expect(entries).toContain("processed")
+    if (entries.includes("processed")) {
-
+      const processedEntries = await readdir(path.join(workerInbox, "processed"))
-    const processedEntries = await readdir(path.join(workerInbox, "processed"))
+      expect(processedEntries).toContain(`${workerMessageId}.json`)
-    expect(processedEntries).toContain(`${workerMessageId}.json`)
+    } else {
+      expect(entries).toContain(`${workerMessageId}.json`)
+    }
   })
 
   test("leaves fresh .delivering-* reservations in place on resume", async () => {

src/features/team-mode/tools/messaging.test.ts

@@ -637,7 +637,7 @@ describe("createTeamSendMessageTool", () => {
     expect(inboxEntries).toHaveLength(1)
   })
 
-  test("#given live delivery promptAsync fails after dispatch may have been accepted #when delivery falls back #then it keeps the delivered message reserved instead of surfacing a duplicate unread", async () => {
+  test("#given live delivery promptAsync fails ambiguously #when delivery falls back #then it releases the message for mailbox injection", async () => {
     // given
     const fixture = await createTeamFixture()
     let promptCalls = 0
@@ -661,20 +661,51 @@ describe("createTeamSendMessageTool", () => {
     // then
     expect(promptCalls).toBe(1)
     const unread = await listUnreadMessages(fixture.teamRunId, "m2", fixture.config)
-    expect(unread).toHaveLength(0)
+    expect(unread).toHaveLength(1)
 
     const inboxDir = getInboxDir(resolveBaseDir(fixture.config), fixture.teamRunId, "m2")
     const inboxEntries = (await readdir(inboxDir)).filter((entry) => entry.endsWith(".json"))
     expect(inboxEntries).toHaveLength(1)
-    expect(inboxEntries[0]?.startsWith(".delivering-")).toBe(true)
+    expect(inboxEntries[0]?.startsWith(".delivering-")).toBe(false)
 
     const { loadRuntimeState: loadState } = await import("../team-state-store/store")
     const runtimeState = await loadState(fixture.teamRunId, fixture.config)
     const recipient = runtimeState.members.find((member) => member.name === "m2")
-    expect(recipient?.pendingInjectedMessageIds).toHaveLength(1)
+    expect(recipient?.pendingInjectedMessageIds).toHaveLength(0)
   })
 
-  test("#given live delivery prompt dispatches but pending mark fails #when delivery finishes #then the message is not re-exposed as unread", async () => {
+  test("#given dispatchInternalPrompt fails ambiguously #when deliverLive handles the failure #then the message is released back to inbox as unread AND pendingInjectedMessageIds is NOT updated", async () => {
+    // given
+    const fixture = await createTeamFixture()
+    const failingClient = {
+      session: {
+        promptAsync: async () => { throw new Error("JSON Parse error: Unexpected EOF") },
+      },
+    } satisfies LiveDeliveryClient
+    const liveTool = createTeamSendMessageTool(fixture.config, failingClient)
+
+    // when
+    await liveTool.execute({
+      teamRunId: fixture.teamRunId,
+      to: "m2",
+      body: "ambiguous failure should retry",
+    }, fixture.toolContext(fixture.memberOneSessionId))
+
+    // then
+    const unread = await listUnreadMessages(fixture.teamRunId, "m2", fixture.config)
+    expect(unread).toHaveLength(1)
+
+    const inboxDir = getInboxDir(resolveBaseDir(fixture.config), fixture.teamRunId, "m2")
+    const inboxEntries = await readdir(inboxDir)
+    expect(inboxEntries.filter((entry) => entry.startsWith(".delivering-"))).toHaveLength(0)
+
+    const { loadRuntimeState: loadState } = await import("../team-state-store/store")
+    const runtimeState = await loadState(fixture.teamRunId, fixture.config)
+    const recipient = runtimeState.members.find((member) => member.name === "m2")
+    expect(recipient?.pendingInjectedMessageIds).toHaveLength(0)
+  })
+
+  test("#given live delivery prompt dispatches but pending mark fails #when delivery finishes #then the reservation is committed to processed", async () => {
     // given
     const fixture = await createTeamFixture()
     let promptCalls = 0
@@ -702,8 +733,40 @@ describe("createTeamSendMessageTool", () => {
 
     const inboxDir = getInboxDir(resolveBaseDir(fixture.config), fixture.teamRunId, "m2")
     const inboxEntries = (await readdir(inboxDir)).filter((entry) => entry.endsWith(".json"))
-    expect(inboxEntries).toHaveLength(1)
+    expect(inboxEntries.filter((entry) => entry.startsWith(".delivering-"))).toHaveLength(0)
-    expect(inboxEntries[0]?.startsWith(".delivering-")).toBe(true)
+    const processedEntries = (await readdir(path.join(inboxDir, "processed"))).filter((entry) => entry.endsWith(".json"))
+    expect(processedEntries).toHaveLength(1)
+  })
+
+  test("#given dispatchInternalPrompt succeeds #when markLiveDeliveryPending fails #then the reservation is committed to processed/", async () => {
+    // given
+    const fixture = await createTeamFixture()
+    const client = {
+      session: {
+        promptAsync: async () => {
+          await rm(path.join(resolveBaseDir(fixture.config), "runtime", fixture.teamRunId, "state.json"))
+        },
+      },
+    } satisfies LiveDeliveryClient
+    const liveTool = createTeamSendMessageTool(fixture.config, client)
+
+    // when
+    await liveTool.execute({
+      teamRunId: fixture.teamRunId,
+      to: "m2",
+      body: "accepted prompt should be processed",
+    }, fixture.toolContext(fixture.memberOneSessionId))
+
+    // then
+    const inboxDir = getInboxDir(resolveBaseDir(fixture.config), fixture.teamRunId, "m2")
+    const inboxEntries = await readdir(inboxDir)
+    expect(inboxEntries.filter((entry) => entry.startsWith(".delivering-"))).toHaveLength(0)
+
+    const processedEntries = (await readdir(path.join(inboxDir, "processed"))).filter((entry) => entry.endsWith(".json"))
+    expect(processedEntries).toHaveLength(1)
+
+    const unread = await listUnreadMessages(fixture.teamRunId, "m2", fixture.config)
+    expect(unread).toHaveLength(0)
   })
 
   test("#given live delivery cannot reload runtime after pre-reserve #when delivery aborts #then the message is released for mailbox injection", async () => {

src/features/team-mode/tools/messaging.ts

@@ -10,6 +10,7 @@ import { isAmbiguousPostDispatchPromptFailure } from "../../../shared/prompt-fai
 import { applyMemberSessionRouting, buildMemberPromptBody } from "../member-session-routing"
 import { buildEnvelope } from "../team-mailbox/poll"
 import {
+  commitDeliveryReservation,
   releaseDeliveryReservation,
   reserveMessageForDelivery,
 } from "../team-mailbox/reservation"
@@ -275,19 +276,12 @@ async function deliverLive(
         },
       })
       if (promptResult.status === "failed" && isAmbiguousPostDispatchPromptFailure(promptResult)) {
-        try {
+        await releaseReservationSafely(reservation, {
-          await markLiveDeliveryPending(teamRunId, recipientName, message.messageId, config)
+          teamRunId,
-        } catch (markError) {
+          recipient: recipientName,
-          log("[team-mailbox] live delivery prompt may be accepted but pending mark failed, keeping reservation hidden", {
+          messageId: message.messageId,
-            teamRunId,
+        })
-            recipient: recipientName,
+        log("[team-mailbox] live delivery prompt failed ambiguously, released reservation to inbox", {
-            recipientSessionId,
-            messageId: message.messageId,
-            error: markError instanceof Error ? markError.message : String(markError),
-          })
-          continue
-        }
-        log("[team-mailbox] live delivery prompt failed after dispatch attempt, keeping reservation pending", {
           teamRunId,
           recipient: recipientName,
           recipientSessionId,
@@ -314,7 +308,18 @@ async function deliverLive(
       try {
         await markLiveDeliveryPending(teamRunId, recipientName, message.messageId, config)
       } catch (markError) {
-        log("[team-mailbox] live delivery prompt dispatched but pending mark failed, keeping reservation hidden", {
+        try {
+          await commitDeliveryReservation(reservation)
+        } catch (commitError) {
+          log("[team-mailbox] live delivery prompt dispatched but pending mark and reservation commit failed", {
+            teamRunId,
+            recipient: recipientName,
+            recipientSessionId,
+            messageId: message.messageId,
+            error: commitError instanceof Error ? commitError.message : String(commitError),
+          })
+        }
+        log("[team-mailbox] live delivery prompt dispatched but pending mark failed, committed reservation directly", {
           teamRunId,
           recipient: recipientName,
           recipientSessionId,