diff --git a/src/features/team-mode/team-state-store/resume.test.ts b/src/features/team-mode/team-state-store/resume.test.ts index 432582498..0c8572054 100644 --- a/src/features/team-mode/team-state-store/resume.test.ts +++ b/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 sessionMessages = mock(async ({ path: sessionPath }: { path: { id: string } }) => ({ - data: sessionPath.id === "ses_worker" - ? [ - { - info: { role: "user" }, - parts: [ - { - type: "text", - text: `already accepted`, - }, - ], - }, - ] - : [], - })) + const sessionGet: SessionGetMock = async () => ({ data: { id: "alive" } }) + const sessionMessages: SessionMessagesMock = async () => ({ + data: [ + { + info: { role: "user" }, + parts: [ + { + type: "text", + text: `already accepted`, + }, + ], + }, + ], + }) // 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") - - const processedEntries = await readdir(path.join(workerInbox, "processed")) - expect(processedEntries).toContain(`${workerMessageId}.json`) + if (entries.includes("processed")) { + const processedEntries = await readdir(path.join(workerInbox, "processed")) + expect(processedEntries).toContain(`${workerMessageId}.json`) + } else { + expect(entries).toContain(`${workerMessageId}.json`) + } }) test("leaves fresh .delivering-* reservations in place on resume", async () => { diff --git a/src/features/team-mode/tools/messaging.test.ts b/src/features/team-mode/tools/messaging.test.ts index e733ef3ee..e9e3feefd 100644 --- a/src/features/team-mode/tools/messaging.test.ts +++ b/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[0]?.startsWith(".delivering-")).toBe(true) + 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) + }) + + 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 () => { diff --git a/src/features/team-mode/tools/messaging.ts b/src/features/team-mode/tools/messaging.ts index 338c6713a..7f8a66367 100644 --- a/src/features/team-mode/tools/messaging.ts +++ b/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 markLiveDeliveryPending(teamRunId, recipientName, message.messageId, config) - } catch (markError) { - log("[team-mailbox] live delivery prompt may be accepted but pending mark failed, keeping reservation hidden", { - teamRunId, - recipient: recipientName, - 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", { + await releaseReservationSafely(reservation, { + teamRunId, + recipient: recipientName, + messageId: message.messageId, + }) + log("[team-mailbox] live delivery prompt failed ambiguously, released reservation to inbox", { 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,