diff --git a/.changeset/fast-restart-closeout-unrecoverable.md b/.changeset/fast-restart-closeout-unrecoverable.md new file mode 100644 index 000000000..70bf6a991 --- /dev/null +++ b/.changeset/fast-restart-closeout-unrecoverable.md @@ -0,0 +1,5 @@ +--- +'@roomote/web': patch +--- + +A Fast turn whose durable admission write failed before it ran no longer ends in silence when a restart cuts it off. Every other turn already resumes on the next process; this one had no row to resume from, so the turn disappeared and the Session stayed marked responding. The turn is now admitted late at the moment of interruption and handed straight to the queue, which resumes it the same way. Only if that late admission also fails does the turn post "Roomote restarted while working on this request. Please send it again." Queue-delivered follow-ups stay quiet because the queue re-runs them itself, and platform events keep their existing handling. diff --git a/apps/api/src/handlers/discord/fast-agent.ts b/apps/api/src/handlers/discord/fast-agent.ts index 8fbf66e15..c4979e0a3 100644 --- a/apps/api/src/handlers/discord/fast-agent.ts +++ b/apps/api/src/handlers/discord/fast-agent.ts @@ -29,6 +29,7 @@ import { import { admitFastAgentHumanFollowUp, createFastAgentConversationArtifact, + handOffFastAgentInterruptedTurn, persistFastAgentInlineHumanTurn, recordFastAgentConversationMessageBestEffort, resolveUserMcpServerConfigs, @@ -374,7 +375,13 @@ export async function processDiscordFastAgentMessage( retryAt, ), } - : {}), + : { + requestLateDurableAdmission: () => + handOffFastAgentInterruptedTurn({ + parent: { sessionId: session.id, conversation }, + event: humanFollowUpEvent, + }), + }), resolveMcpServerConfigs: () => resolveUserMcpServerConfigs({ userId: input.senderUserId, diff --git a/apps/api/src/handlers/slack/events/fast-agent-reaction.ts b/apps/api/src/handlers/slack/events/fast-agent-reaction.ts index 5e93f496a..a3d7601b9 100644 --- a/apps/api/src/handlers/slack/events/fast-agent-reaction.ts +++ b/apps/api/src/handlers/slack/events/fast-agent-reaction.ts @@ -15,6 +15,7 @@ import { import { buildFastAgentArtifactCreator, findFastAgentSessionForProviderMessage, + handOffFastAgentInterruptedTurn, persistFastAgentInlineHumanTurn, recordFastAgentConversationMessageBestEffort, resolveUserMcpServerConfigs, @@ -93,7 +94,9 @@ async function processFastAgentReaction(params: { // claim before it runs, so an interruption hands it to the queue, which // resumes it with the same reaction input instead of asking the user to // react again. - const durableTurn = await persistFastAgentInlineHumanTurn({ + const reactionAdmission: Parameters< + typeof persistFastAgentInlineHumanTurn + >[0] = { parent: { sessionId: session.id, conversation }, event: { type: 'human_follow_up', @@ -107,7 +110,10 @@ async function processFastAgentReaction(params: { : {}), input: { type: 'reaction', externalInput: reactionInput }, }, - }).catch((error) => { + }; + const durableTurn = await persistFastAgentInlineHumanTurn( + reactionAdmission, + ).catch((error) => { console.error( `[SlackWebhook] Failed to persist Fast reaction turn admission: ${error instanceof Error ? error.message : String(error)}`, ); @@ -159,7 +165,10 @@ async function processFastAgentReaction(params: { retryAt, ), } - : {}), + : { + requestLateDurableAdmission: () => + handOffFastAgentInterruptedTurn(reactionAdmission), + }), createArtifact: buildFastAgentArtifactCreator(session.id), activity: createFastAgentSlackSessionActivity({ slack: context.slack, diff --git a/apps/api/src/handlers/slack/events/fast-agent.ts b/apps/api/src/handlers/slack/events/fast-agent.ts index d7d466592..2466a0715 100644 --- a/apps/api/src/handlers/slack/events/fast-agent.ts +++ b/apps/api/src/handlers/slack/events/fast-agent.ts @@ -25,6 +25,7 @@ import { appendAttachmentTextsToPromptText } from '@roomote/cloud-agents'; import { admitFastAgentHumanFollowUp, createFastAgentConversationArtifact, + handOffFastAgentInterruptedTurn, persistFastAgentInlineHumanTurn, wakeFastAgentParentEventAt, wakeFastAgentParentEventNow, @@ -357,7 +358,13 @@ export async function processFastAgentMessage(params: { retryAt, ), } - : {}), + : { + requestLateDurableAdmission: () => + handOffFastAgentInterruptedTurn({ + parent: { sessionId: session.id, conversation }, + event: humanFollowUpEvent, + }), + }), activity: createFastAgentSlackSessionActivity({ slack, workspaceId: teamId, diff --git a/apps/docs/fast-sessions.mdx b/apps/docs/fast-sessions.mdx index 88c4c6062..72836a087 100644 --- a/apps/docs/fast-sessions.mdx +++ b/apps/docs/fast-sessions.mdx @@ -118,7 +118,7 @@ for a separate turn under that participant's identity. This keeps shared conversations ordered without losing instructions that arrive during active work. -Fast turns also survive API or worker interruptions. Roomote durably admits the turn before acknowledging it, records every action the turn takes as it happens, and resumes unfinished work on another process. The resumed run receives the transcript of its earlier attempt, up to the point it was cut off, so it continues from there instead of repeating actions or asking you to send the request again. If the cut lands on the final reply itself, the resumed run finishes from the transcript: a reply that was recorded is not posted again, and a reply the process died while posting goes out once more without another model request. This covers turns started by a typed message, by an emoji reaction, and by platform events such as a setup kickoff. Provider-side retry waits are honored the same way: short waits keep the turn in place, and longer waits park it durably until the scheduled time. +Fast turns also survive API or worker interruptions. Roomote durably admits the turn before acknowledging it, records every action the turn takes as it happens, and resumes unfinished work on another process. The resumed run receives the transcript of its earlier attempt, up to the point it was cut off, so it continues from there instead of repeating actions or asking you to send the request again. If the cut lands on the final reply itself, the resumed run finishes from the transcript: a reply that was recorded is not posted again, and a reply the process died while posting goes out once more without another model request. This covers turns started by a typed message, by an emoji reaction, and by platform events such as a setup kickoff. Provider-side retry waits are honored the same way: short waits keep the turn in place, and longer waits park it durably until the scheduled time. A turn that could not be recorded durably before it started is recorded at the moment a restart interrupts it and resumed the same way; only if that also fails does Roomote ask you to send the request again. ### Message Suggestions diff --git a/apps/web/src/trpc/commands/fast-sessions/index.ts b/apps/web/src/trpc/commands/fast-sessions/index.ts index 621e523df..49cc6d904 100644 --- a/apps/web/src/trpc/commands/fast-sessions/index.ts +++ b/apps/web/src/trpc/commands/fast-sessions/index.ts @@ -19,6 +19,7 @@ import { buildFastAgentArtifactCreator, buildFastAgentSurfaceReplyDelivery, createFastAgentSessionArtifact, + handOffFastAgentInterruptedTurn, persistFastAgentInlineHumanTurn, resolveUserMcpServerConfigs, wakeFastAgentParentEventAt, @@ -138,8 +139,10 @@ type WebFastAgentTurnInput = { /** Deterministic turn ID override. Canonical event IDs derive from it, so a * fixed value lets a turn be claimed idempotently across retries. */ currentMessageId?: string; - /** Fast conversation id for durable admission of a human turn. Platform - * turns (kickoffs, artifact builds) omit it and stay non-replayable. */ + /** Fast conversation id for durable admission. Every turn passes it; + * only setup-session turns (adapter extensions, setup snapshot) are still + * process-bound, and their scheduler re-runs them when they never + * completed. */ durableSessionId?: string; /** Skip the turn if this exact canonical event row already exists when the * turn acquires its lock. This is the atomic claim for the setup kickoff: @@ -258,9 +261,11 @@ async function runWebFastAgentTurn({ // events ride the same row with their framing recorded; the ones that // need adapter extensions or a setup snapshot cannot be rebuilt by the // queue and stay process-bound. - const durableTurn = + const durableAdmissionRequest: + | Parameters[0] + | null = durableSessionId && !adapterExtensions && !setupSnapshot - ? await persistFastAgentInlineHumanTurn({ + ? { parent: { sessionId: durableSessionId, conversation }, event: { type: 'human_follow_up', @@ -281,13 +286,18 @@ async function runWebFastAgentTurn({ : {}), ...(setupSession ? { setupSession: true } : {}), }, - }).catch((error) => { + } + : null; + const durableTurn = durableAdmissionRequest + ? await persistFastAgentInlineHumanTurn(durableAdmissionRequest).catch( + (error) => { console.error( `[Fast Web] Failed to persist turn admission: ${formatErrorForLog(error)}`, ); return null; - }) - : null; + }, + ) + : null; if (durableTurn && durableSessionId) { release.durableRowId = durableTurn.id; release.durableResume = () => @@ -345,7 +355,12 @@ async function runWebFastAgentTurn({ retryAt, ), } - : {}), + : durableAdmissionRequest + ? { + requestLateDurableAdmission: () => + handOffFastAgentInterruptedTurn(durableAdmissionRequest), + } + : {}), ...delivery.adapter, ...adapterExtensions, }, diff --git a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-service.test.ts b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-service.test.ts index b576f1ef3..ccbc44099 100644 --- a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-service.test.ts +++ b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-service.test.ts @@ -103,6 +103,8 @@ vi.mock('../fast-agent-session', () => ({ vi.mock('../fast-agent-conversation-repository', () => ({ INTERRUPTED_INFERENCE_RETRY_MESSAGE: 'The inference retry was interrupted before it completed. Please send the request again.', + RESTARTED_ACTIVE_TURN_MESSAGE: + 'Roomote restarted while working on this request. Please send it again.', reconcileFastAgentInferenceRetryNotices: mocks.reconcileRetryNotices, markFastAgentInferenceRetryNoticeInterruption: mocks.markRetryNoticeInterruption, @@ -4923,47 +4925,135 @@ describe('answerFastAgentQuestion native OpenCode tools', () => { }); }); - it('posts nothing when API shutdown cuts off an unadmitted turn during silent retry backoff', async () => { - // No durable row (the admission write failed) and no visible retry - // notice to correct: there is no restart notice any more, so the turn - // ends without a user-facing message. - const controller = new AbortController(); - const shutdown = new FastAgentProcessShutdownError('SIGTERM'); - const postReply = vi.fn().mockResolvedValue({ messageId: 'closeout-1' }); + function abortOnFirstTimer(controller: AbortController, reason: Error) { const originalSetTimeout = globalThis.setTimeout; let shouldAbort = true; - const timeout = vi.spyOn(globalThis, 'setTimeout').mockImplementation((( + return vi.spyOn(globalThis, 'setTimeout').mockImplementation((( callback: () => void, ) => { return originalSetTimeout(() => { callback(); if (shouldAbort) { shouldAbort = false; - controller.abort(shutdown); + controller.abort(reason); } }, 0); }) as typeof setTimeout); + } + + it('admits a turn late and hands it to the queue when API shutdown cuts it off without a row', async () => { + // The admission write failed before the turn ran. Rather than asking the + // user to resend, the turn is persisted now and handed to the queue, + // which resumes it on the next process; nothing is posted. + const controller = new AbortController(); + const shutdown = new FastAgentProcessShutdownError('SIGTERM'); + const postReply = vi.fn().mockResolvedValue({ messageId: 'closeout-1' }); + const requestLateDurableAdmission = vi.fn().mockResolvedValue(true); + const timeout = abortOnFirstTimer(controller, shutdown); mocks.generateText.mockRejectedValue(new Error('TypeError: fetch failed')); try { await expect( answerFastAgentQuestion({ ...baseParams, - adapter: callbacks({ postReply }), + adapter: callbacks({ postReply, requestLateDurableAdmission }), signal: controller.signal, }), ).rejects.toBe(shutdown); - expect(mocks.generateText).toHaveBeenCalledOnce(); + expect(requestLateDurableAdmission).toHaveBeenCalledOnce(); expect(postReply).not.toHaveBeenCalled(); const persisted = mocks.upsertMessage.mock.calls.map( ([input]) => input.message, ); expect( - persisted.filter((message) => - message.eventId.startsWith('100.2:assistant:'), + persisted.some( + (message) => + (message.metadata as { purpose?: string } | undefined)?.purpose === + 'closeout', ), - ).toHaveLength(0); + ).toBe(false); + } finally { + timeout.mockRestore(); + } + }); + + it('tells the user to resend only when a turn has no row and cannot be admitted late', async () => { + // Nothing will re-run this turn, so the honest outcome is a recorded + // restart closeout. + const controller = new AbortController(); + const shutdown = new FastAgentProcessShutdownError('SIGTERM'); + const postReply = vi.fn().mockResolvedValue({ messageId: 'closeout-1' }); + const requestLateDurableAdmission = vi + .fn() + .mockRejectedValue(new Error('db offline')); + const timeout = abortOnFirstTimer(controller, shutdown); + mocks.generateText.mockRejectedValue(new Error('TypeError: fetch failed')); + + try { + await expect( + answerFastAgentQuestion({ + ...baseParams, + adapter: callbacks({ postReply, requestLateDurableAdmission }), + signal: controller.signal, + }), + ).rejects.toBe(shutdown); + + expect(requestLateDurableAdmission).toHaveBeenCalledOnce(); + + expect(mocks.generateText).toHaveBeenCalledOnce(); + expect(postReply).toHaveBeenCalledWith({ + purpose: 'closeout', + message: + 'Roomote restarted while working on this request. Please send it again.', + }); + const persisted = mocks.upsertMessage.mock.calls.map( + ([input]) => input.message, + ); + const closeout = persisted.find( + (message) => + message.eventId.startsWith('100.2:assistant:') && + (message.metadata as { purpose?: string } | undefined)?.purpose === + 'closeout', + ); + expect(closeout?.metadata).toMatchObject({ + interruptionReason: 'api_shutdown', + }); + // The closeout is recorded as a call and a result like any other. + expect( + persisted.some( + (message) => + message.eventId.startsWith('100.2:tool:') && + (message.payload as { toolName?: string } | undefined)?.toolName === + 'send_chat_reply', + ), + ).toBe(true); + } finally { + timeout.mockRestore(); + } + }); + + it('posts nothing when API shutdown cuts off a queue-delivered follow-up the queue will re-run', async () => { + const controller = new AbortController(); + const shutdown = new FastAgentProcessShutdownError('SIGTERM'); + const postReply = vi.fn().mockResolvedValue({ messageId: 'closeout-1' }); + const timeout = abortOnFirstTimer(controller, shutdown); + mocks.generateText.mockRejectedValue(new Error('TypeError: fetch failed')); + + try { + await expect( + answerFastAgentQuestion({ + ...baseParams, + adapter: callbacks({ postReply }), + signal: controller.signal, + currentDurableHumanFollowUpEventId: '100.2', + }), + ).rejects.toBe(shutdown); + + expect(postReply).not.toHaveBeenCalled(); + const persisted = mocks.upsertMessage.mock.calls.map( + ([input]) => input.message, + ); expect( persisted.some( (message) => diff --git a/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts b/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts index b4f61c2d6..215399587 100644 --- a/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts +++ b/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts @@ -62,6 +62,14 @@ export type FastAgentMessageUpsertResult = { initialHumanTurn: boolean; }; +/** + * Closeout for a turn a restart cut off with no way to resume it: the turn + * had no durable row (its admission write failed), so nothing will re-run + * it and the user has to ask again. Every admitted turn resumes instead. + */ +export const RESTARTED_ACTIVE_TURN_MESSAGE = + 'Roomote restarted while working on this request. Please send it again.'; + export const INTERRUPTED_INFERENCE_RETRY_MESSAGE = 'The inference retry was interrupted before it completed. Please send the request again.'; diff --git a/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation.ts b/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation.ts index 8af832c86..1631d8697 100644 --- a/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation.ts +++ b/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation.ts @@ -208,6 +208,14 @@ export type FastAgentTurnAdapter = { * wait for the next sweep. Best effort. */ requestDurableResume?: () => Promise; + /** + * Called when a shutdown interrupts a turn that has no durable row (its + * admission write failed before it ran). Persists the row now and hands + * it to the queue so the turn resumes on the next process instead of + * asking the user to send it again. Resolves true when the hand-off + * landed; false means nothing will re-run the turn. + */ + requestLateDurableAdmission?: () => Promise; /** * Called when a turn has parked itself for a durable inference * retry; schedules the queue wakeup for `retryAt` so the retry does not diff --git a/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts b/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts index dfced2e7f..3be4e2a47 100644 --- a/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts +++ b/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts @@ -111,6 +111,7 @@ import { buildFastAgentExplicitSkillInvocationContext } from './fast-agent-skill import { findFastAgentUnresolvedRequest, INTERRUPTED_INFERENCE_RETRY_MESSAGE, + RESTARTED_ACTIVE_TURN_MESSAGE, findFastAgentActiveInferenceRetryNotice, markFastAgentDurableTurnDelivered, markFastAgentInferenceRetryNoticeInterruption, @@ -4584,6 +4585,18 @@ export async function answerFastAgentQuestion({ durableTurnReplayable && Boolean(durableAdmission) && (shutdownInterrupted || lockOwnershipLost); + // A restart with no row to resume the turn from: the admission write + // failed before it ran. A queue-delivered row is not this case (the + // queue re-runs it on its own), nor is a platform event (its + // scheduler re-runs it when it never completed). The turn is admitted + // late and handed to the queue; only if that fails too is the user + // asked to send it again. + const restartedWithoutRow = + shutdownInterrupted && + !durableAdmission && + !currentDurableHumanFollowUpEventId && + !platformEvent; + let handedOffLate = false; console.error( `[Fast Agent] Turn interrupted (reason=${interruptionReason}, conversation=${canonicalConversationId ?? 'unknown'}, retryNoticeVisible=${Boolean(inferenceRetryReply)}, resumable=${resumable}, error=${formatErrorForLog(terminalError)})`, ); @@ -4610,6 +4623,22 @@ export async function answerFastAgentQuestion({ ); }); } + if (restartedWithoutRow && adapter.requestLateDurableAdmission) { + handedOffLate = await adapter + .requestLateDurableAdmission() + .catch((admitError) => { + console.warn( + `[Fast Agent] Late durable admission failed: ${formatErrorForLog(admitError)}`, + ); + return false; + }); + if (handedOffLate) { + console.info( + `[Fast Agent] Turn ${turnId} admitted late and handed to the queue after a restart (conversation=${canonicalConversationId ?? 'unknown'}).`, + ); + } + } + const restartedWithoutRecovery = restartedWithoutRow && !handedOffLate; // A terminal interruption closeout is only safe once the row can no // longer be re-run; if that revocation did not land, post nothing // and let recovery own the outcome. @@ -4619,23 +4648,52 @@ export async function answerFastAgentQuestion({ : await revokeDurableTurnReplay( `Turn interrupted without replay (${interruptionReason}).`, ); - if (!terminalCloseoutAllowed) { - // Resumable turns and unrevoked rows fall through to the rethrow - // below without a user-facing closeout. + if (!terminalCloseoutAllowed || handedOffLate) { + // Resumable turns, late-admitted turns, and unrevoked rows fall + // through to the rethrow below without a user-facing closeout; a + // visible retry notice stays for the resumed run to inherit. } else if (!lockOwnershipLost && inferenceRetryReply) { // A visible retry notice must not stay up claiming a retry that // will never come: a deliberately cancelled turn, or the rare turn - // whose admission write failed and so has no row to resume from. - // Every admitted turn resumes instead and never reaches here. + // that has no row and could not be admitted late either. await replaceInferenceRetryReply( { purpose: 'closeout', - message: INTERRUPTED_INFERENCE_RETRY_MESSAGE, + message: restartedWithoutRecovery + ? RESTARTED_ACTIVE_TURN_MESSAGE + : INTERRUPTED_INFERENCE_RETRY_MESSAGE, }, true, undefined, interruptionReason, ); + } else if (restartedWithoutRecovery && !isInstructionClosed()) { + // Nothing will re-run this turn. Recorded like every other + // closeout, so the transcript shows the turn ended here and why. + const reply = { + purpose: 'closeout' as const, + message: RESTARTED_ACTIVE_TURN_MESSAGE, + }; + try { + await postRecordedSystemCloseout(reply.message, async () => { + const posted = + (await surfaceReplyStream.deliver(reply)) ?? + (await adapter.postReply(reply)); + diagnostics.recordVisibleReply(); + await persistAssistantReply({ + reply, + event: allocateCanonicalEvent( + `assistant:${nextAssistantOrdinal++}`, + ), + platformMessageId: posted?.messageId, + interruptionReason, + }); + }); + } catch (postError) { + console.error( + `[Fast Agent] Failed to post restart closeout: ${formatErrorForLog(postError)}`, + ); + } } else if ( lockOwnershipLost && canonicalConversationId && diff --git a/packages/sdk/src/server/index.ts b/packages/sdk/src/server/index.ts index d34a0330f..0a4c6146f 100644 --- a/packages/sdk/src/server/index.ts +++ b/packages/sdk/src/server/index.ts @@ -223,6 +223,7 @@ export { } from './lib/fast-agent-parent-event-queue'; export { admitFastAgentHumanFollowUp, + handOffFastAgentInterruptedTurn, persistFastAgentInlineHumanTurn, type FastAgentDurableTurn, type FastAgentHumanFollowUpAdmission, diff --git a/packages/sdk/src/server/lib/fast-agent-human-follow-up.test.ts b/packages/sdk/src/server/lib/fast-agent-human-follow-up.test.ts index 1b137a96b..90de836d7 100644 --- a/packages/sdk/src/server/lib/fast-agent-human-follow-up.test.ts +++ b/packages/sdk/src/server/lib/fast-agent-human-follow-up.test.ts @@ -1,5 +1,7 @@ const mocks = vi.hoisted(() => ({ acquireTurnLock: vi.fn(), + releaseDurableClaim: vi.fn(), + wakeNow: vi.fn(), enqueueParentEvent: vi.fn(), updateWhere: vi.fn(), insertOnConflict: vi.fn(), @@ -9,6 +11,7 @@ const mocks = vi.hoisted(() => ({ vi.mock('@roomote/cloud-agents/server', () => ({ acquireFastAgentTurnLock: mocks.acquireTurnLock, + releaseFastAgentDurableTurnClaim: mocks.releaseDurableClaim, FAST_AGENT_DURABLE_TURN_CLAIM_MS: 15 * 60 * 1000, })); @@ -45,11 +48,13 @@ vi.mock('@roomote/db/server', () => ({ vi.mock('./fast-agent-parent-event-queue', () => ({ enqueueFastAgentParentEvent: mocks.enqueueParentEvent, + wakeFastAgentParentEventNow: mocks.wakeNow, buildFastAgentParentEventKey: vi.fn(() => 'stable-event-key'), })); import { admitFastAgentHumanFollowUp, + handOffFastAgentInterruptedTurn, persistFastAgentInlineHumanTurn, } from './fast-agent-human-follow-up'; @@ -147,6 +152,73 @@ describe('persistFastAgentInlineHumanTurn', () => { }); }); +describe('handOffFastAgentInterruptedTurn', () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.updateWhere.mockResolvedValue(undefined); + mocks.insertOnConflict.mockReturnValue({ + returning: mocks.insertReturning, + }); + mocks.insertReturning.mockResolvedValue([{ id: 'row-1' }]); + mocks.releaseDurableClaim.mockResolvedValue(true); + mocks.wakeNow.mockResolvedValue(undefined); + }); + + it('persists the row, releases the claim, and wakes the queue', async () => { + mocks.findFirst.mockResolvedValue({ + id: 'row-1', + admission: 'inline', + deliveredAt: null, + discardedAt: null, + }); + + await expect( + handOffFastAgentInterruptedTurn({ parent, event }), + ).resolves.toBe(true); + expect(mocks.insertOnConflict).toHaveBeenCalledOnce(); + expect(mocks.releaseDurableClaim).toHaveBeenCalledWith('row-1'); + expect(mocks.wakeNow).toHaveBeenCalledWith({ + conversationId: parent.sessionId, + eventKey: 'stable-event-key', + }); + }); + + it('still reports the hand-off when only the wakeup fails', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + mocks.findFirst.mockResolvedValue({ + id: 'row-1', + admission: 'inline', + deliveredAt: null, + discardedAt: null, + }); + mocks.wakeNow.mockRejectedValue(new Error('redis down')); + + try { + // The row is pending with no claim, so the recovery sweep picks it up. + await expect( + handOffFastAgentInterruptedTurn({ parent, event }), + ).resolves.toBe(true); + expect(mocks.releaseDurableClaim).toHaveBeenCalledWith('row-1'); + } finally { + warn.mockRestore(); + } + }); + + it('reports no hand-off when the same message already settled', async () => { + mocks.findFirst.mockResolvedValue({ + id: 'row-1', + deliveredAt: new Date(), + discardedAt: null, + }); + + await expect( + handOffFastAgentInterruptedTurn({ parent, event }), + ).resolves.toBe(false); + expect(mocks.releaseDurableClaim).not.toHaveBeenCalled(); + expect(mocks.wakeNow).not.toHaveBeenCalled(); + }); +}); + describe('admitFastAgentHumanFollowUp', () => { beforeEach(() => { vi.clearAllMocks(); diff --git a/packages/sdk/src/server/lib/fast-agent-human-follow-up.ts b/packages/sdk/src/server/lib/fast-agent-human-follow-up.ts index d0a63af17..6d7d6f84a 100644 --- a/packages/sdk/src/server/lib/fast-agent-human-follow-up.ts +++ b/packages/sdk/src/server/lib/fast-agent-human-follow-up.ts @@ -1,6 +1,7 @@ import { acquireFastAgentTurnLock, FAST_AGENT_DURABLE_TURN_CLAIM_MS, + releaseFastAgentDurableTurnClaim, type FastAgentTurnLockHandle, } from '@roomote/cloud-agents/server'; import { @@ -19,6 +20,7 @@ import type { import { buildFastAgentParentEventKey, enqueueFastAgentParentEvent, + wakeFastAgentParentEventNow, } from './fast-agent-parent-event-queue'; export type FastAgentDurableTurn = { @@ -122,6 +124,36 @@ export async function persistFastAgentInlineHumanTurn(params: { }); } +/** + * Late durable admission for a turn a shutdown interrupted after its + * admission write had failed. The turn already ran under this process with + * its prompt and actions recorded by turn id, so persisting the row now and + * handing it straight to the queue lets the next process resume it exactly + * as it would a turn admitted up front. Returns false when no row could be + * persisted (or the same message already settled), in which case nothing + * will re-run the turn. + */ +export async function handOffFastAgentInterruptedTurn(params: { + parent: FastAgentParent; + event: FastAgentHumanFollowUpEvent; +}): Promise { + const durable = await persistFastAgentInlineHumanTurn(params); + if (!durable) return false; + // The row is persisted under this process's claim; this process is going + // away, so release it at once and ask the queue not to wait for a sweep. + await releaseFastAgentDurableTurnClaim(durable.id); + await wakeFastAgentParentEventNow({ + conversationId: params.parent.sessionId, + eventKey: durable.eventKey, + }).catch((error) => { + // The recovery sweep recreates the wakeup within its interval. + console.warn( + `[Fast Agent] Late-admitted turn ${durable.id} persisted, but its wakeup failed: ${error instanceof Error ? error.message : String(error)}`, + ); + }); + return true; +} + /** * Start a human turn immediately when the conversation is idle. While another * Fast generation owns the turn lock, durably admit the message for native diff --git a/packages/sdk/src/server/lib/fast-agent-surface-reply.ts b/packages/sdk/src/server/lib/fast-agent-surface-reply.ts index d343fc028..17a0d0c7e 100644 --- a/packages/sdk/src/server/lib/fast-agent-surface-reply.ts +++ b/packages/sdk/src/server/lib/fast-agent-surface-reply.ts @@ -49,6 +49,7 @@ import { } from './fast-agent-reply-replacement'; import { admitFastAgentHumanFollowUp, + handOffFastAgentInterruptedTurn, persistFastAgentInlineHumanTurn, } from './fast-agent-human-follow-up'; import { @@ -769,7 +770,16 @@ async function runFastAgentSurfaceReply( retryAt, ), } - : {}), + : { + requestLateDurableAdmission: () => + handOffFastAgentInterruptedTurn({ + parent: { + sessionId: params.sessionId, + conversation: delivery.conversation, + }, + event: buildSurfaceHumanFollowUpEvent(params), + }), + }), createArtifact: buildFastAgentArtifactCreator(params.sessionId), ...delivery.adapter, },