From f5eeee5c838e192834e1a47252d6aa90aad54bf1 Mon Sep 17 00:00:00 2001 From: Matt Rubens <2600+mrubens@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:19:26 -0400 Subject: [PATCH 1/2] [Fix] Tell the user when a restart cuts off a Fast turn that cannot resume Every admitted turn resumes on the next process, so #2156 removed the restart notice. A turn whose durable admission write failed has no row to resume from, and it ended in silence with the Session left showing responding. That turn now posts the recorded restart closeout again. Queue-delivered follow-ups stay quiet because the queue re-runs them, and platform events keep their silent handling; setup platform events are re-run by their own scheduler and a closeout would mark them complete. --- .../fast-restart-closeout-unrecoverable.md | 5 ++ apps/docs/fast-sessions.mdx | 2 +- .../src/trpc/commands/fast-sessions/index.ts | 6 +- .../__tests__/fast-agent-service.test.ts | 74 +++++++++++++++---- .../fast-agent-conversation-repository.ts | 8 ++ .../server/fast-agent/fast-agent-service.ts | 43 ++++++++++- 6 files changed, 121 insertions(+), 17 deletions(-) create mode 100644 .changeset/fast-restart-closeout-unrecoverable.md diff --git a/.changeset/fast-restart-closeout-unrecoverable.md b/.changeset/fast-restart-closeout-unrecoverable.md new file mode 100644 index 000000000..04836d85e --- /dev/null +++ b/.changeset/fast-restart-closeout-unrecoverable.md @@ -0,0 +1,5 @@ +--- +'@roomote/web': patch +--- + +A Fast turn that a restart cuts off with no way to resume it tells the user so again. Every admitted turn resumes on the next process, so the restart notice was removed; but a turn whose durable admission write failed has no row to resume from, and it was ending in silence with the Session left showing "responding". That turn now posts the recorded closeout "Roomote restarted while working on this request. Please send it again." Turns the queue delivers stay quiet, since the queue re-runs them itself, and platform events keep their existing silent handling. diff --git a/apps/docs/fast-sessions.mdx b/apps/docs/fast-sessions.mdx index 88c4c6062..d5e054c90 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. In the rare case that a turn could not be recorded durably before it started, a restart ends it with a message asking 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..971c72f7e 100644 --- a/apps/web/src/trpc/commands/fast-sessions/index.ts +++ b/apps/web/src/trpc/commands/fast-sessions/index.ts @@ -138,8 +138,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: 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..ae411a8e0 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,26 +4925,29 @@ 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('tells the user to resend when API shutdown cuts off a turn that has no row to resume from', async () => { + // No durable row (the admission write failed): 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 timeout = abortOnFirstTimer(controller, shutdown); mocks.generateText.mockRejectedValue(new Error('TypeError: fetch failed')); try { @@ -4955,15 +4960,58 @@ describe('answerFastAgentQuestion native OpenCode tools', () => { ).rejects.toBe(shutdown); expect(mocks.generateText).toHaveBeenCalledOnce(); - expect(postReply).not.toHaveBeenCalled(); + 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.filter((message) => - message.eventId.startsWith('100.2:assistant:'), + persisted.some( + (message) => + message.eventId.startsWith('100.2:tool:') && + (message.payload as { toolName?: string } | undefined)?.toolName === + 'send_chat_reply', ), - ).toHaveLength(0); + ).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-service.ts b/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts index dfced2e7f..40bff22e3 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,17 @@ export async function answerFastAgentQuestion({ durableTurnReplayable && Boolean(durableAdmission) && (shutdownInterrupted || lockOwnershipLost); + // A restart with nothing to resume the turn: no inline row (the + // admission write failed), and not a queue-delivered row, which the + // queue re-runs on its own. A platform event stays silent as it does + // on every other error; its scheduler re-runs it when it never + // completed. Only here does the user have to send the request again, + // so only here is it worth saying so. + const restartedWithoutRecovery = + shutdownInterrupted && + !durableAdmission && + !currentDurableHumanFollowUpEventId && + !platformEvent; console.error( `[Fast Agent] Turn interrupted (reason=${interruptionReason}, conversation=${canonicalConversationId ?? 'unknown'}, retryNoticeVisible=${Boolean(inferenceRetryReply)}, resumable=${resumable}, error=${formatErrorForLog(terminalError)})`, ); @@ -4630,12 +4642,41 @@ export async function answerFastAgentQuestion({ 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()) { + // 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 && From 6a75a270a8ec85e91ec4deb6394b8831291e0336 Mon Sep 17 00:00:00 2001 From: Matt Rubens <2600+mrubens@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:27:34 -0400 Subject: [PATCH 2/2] Admit an unrecordable Fast turn late and resume it instead of asking the user to resend When a shutdown interrupts a turn that has no durable row, the handler's requestLateDurableAdmission hook persists the row now, releases its claim, and wakes the queue, so the next process resumes the turn. The restart closeout is posted only when that late admission fails too. --- .../fast-restart-closeout-unrecoverable.md | 2 +- apps/api/src/handlers/discord/fast-agent.ts | 9 ++- .../slack/events/fast-agent-reaction.ts | 15 +++- .../src/handlers/slack/events/fast-agent.ts | 9 ++- apps/docs/fast-sessions.mdx | 2 +- .../src/trpc/commands/fast-sessions/index.ts | 25 +++++-- .../__tests__/fast-agent-service.test.ts | 50 +++++++++++-- .../fast-agent/fast-agent-conversation.ts | 8 +++ .../server/fast-agent/fast-agent-service.ts | 45 ++++++++---- packages/sdk/src/server/index.ts | 1 + .../lib/fast-agent-human-follow-up.test.ts | 72 +++++++++++++++++++ .../server/lib/fast-agent-human-follow-up.ts | 32 +++++++++ .../server/lib/fast-agent-surface-reply.ts | 12 +++- 13 files changed, 250 insertions(+), 32 deletions(-) diff --git a/.changeset/fast-restart-closeout-unrecoverable.md b/.changeset/fast-restart-closeout-unrecoverable.md index 04836d85e..70bf6a991 100644 --- a/.changeset/fast-restart-closeout-unrecoverable.md +++ b/.changeset/fast-restart-closeout-unrecoverable.md @@ -2,4 +2,4 @@ '@roomote/web': patch --- -A Fast turn that a restart cuts off with no way to resume it tells the user so again. Every admitted turn resumes on the next process, so the restart notice was removed; but a turn whose durable admission write failed has no row to resume from, and it was ending in silence with the Session left showing "responding". That turn now posts the recorded closeout "Roomote restarted while working on this request. Please send it again." Turns the queue delivers stay quiet, since the queue re-runs them itself, and platform events keep their existing silent handling. +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 d5e054c90..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. In the rare case that a turn could not be recorded durably before it started, a restart ends it with a message asking you to send the request again. +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 971c72f7e..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, @@ -260,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', @@ -283,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 = () => @@ -347,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 ae411a8e0..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 @@ -4941,12 +4941,14 @@ describe('answerFastAgentQuestion native OpenCode tools', () => { }) as typeof setTimeout); } - it('tells the user to resend when API shutdown cuts off a turn that has no row to resume from', async () => { - // No durable row (the admission write failed): nothing will re-run this - // turn, so the honest outcome is a recorded restart closeout. + 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')); @@ -4954,11 +4956,51 @@ describe('answerFastAgentQuestion native OpenCode tools', () => { await expect( answerFastAgentQuestion({ ...baseParams, - adapter: callbacks({ postReply }), + adapter: callbacks({ postReply, requestLateDurableAdmission }), + signal: controller.signal, + }), + ).rejects.toBe(shutdown); + + expect(requestLateDurableAdmission).toHaveBeenCalledOnce(); + expect(postReply).not.toHaveBeenCalled(); + const persisted = mocks.upsertMessage.mock.calls.map( + ([input]) => input.message, + ); + expect( + persisted.some( + (message) => + (message.metadata as { purpose?: string } | undefined)?.purpose === + 'closeout', + ), + ).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', 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 40bff22e3..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 @@ -4585,17 +4585,18 @@ export async function answerFastAgentQuestion({ durableTurnReplayable && Boolean(durableAdmission) && (shutdownInterrupted || lockOwnershipLost); - // A restart with nothing to resume the turn: no inline row (the - // admission write failed), and not a queue-delivered row, which the - // queue re-runs on its own. A platform event stays silent as it does - // on every other error; its scheduler re-runs it when it never - // completed. Only here does the user have to send the request again, - // so only here is it worth saying so. - const restartedWithoutRecovery = + // 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)})`, ); @@ -4622,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. @@ -4631,14 +4648,14 @@ 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', @@ -4651,8 +4668,8 @@ export async function answerFastAgentQuestion({ interruptionReason, ); } else if (restartedWithoutRecovery && !isInstructionClosed()) { - // Recorded like every other closeout, so the transcript shows the - // turn ended here and why. + // 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, 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, },