diff --git a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts index 80cb30892..23392f67c 100644 --- a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts +++ b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts @@ -16,6 +16,7 @@ import { markFastAgentInferenceRetryNoticeInterruption, reconcileExpiredFastAgentInferenceRetryNotices, reconcileFastAgentInferenceRetryNotices, + renewFastSessionRespondingLease, } from '../fast-agent-conversation-repository'; import { FAST_AGENT_REACTION_INPUT_TYPE } from '../fast-agent-conversation'; import { hasFastAgentSession } from '../fast-agent-session'; @@ -980,4 +981,56 @@ describe('Fast conversation repository', () => { }); } }); + + it('renews only a live responding lease', async () => { + const user = await createUser(); + const session = await fastAgentConversationRepository.getOrCreate({ + userId: user.id, + conversation: { + surface: 'web', + workspaceId: user.id, + conversationId: crypto.randomUUID(), + }, + }); + const readLease = async () => { + const [row] = await db + .select({ respondingUntil: sessions.respondingUntil }) + .from(sessions) + .where(eq(sessions.fastConversationId, session.id)); + return row?.respondingUntil ?? null; + }; + + // A cleared lease is fenced out: a stale renewal cannot resurrect it. + await db + .update(sessions) + .set({ respondingUntil: null }) + .where(eq(sessions.fastConversationId, session.id)); + await expect(renewFastSessionRespondingLease(session.id)).resolves.toBe( + false, + ); + await expect(readLease()).resolves.toBeNull(); + + // An expired lease is fenced out and left untouched. + const expired = new Date(Date.now() - 1_000); + await db + .update(sessions) + .set({ respondingUntil: expired }) + .where(eq(sessions.fastConversationId, session.id)); + await expect(renewFastSessionRespondingLease(session.id)).resolves.toBe( + false, + ); + await expect(readLease()).resolves.toEqual(expired); + + // A live lease is extended. + const live = new Date(Date.now() + 60_000); + await db + .update(sessions) + .set({ respondingUntil: live }) + .where(eq(sessions.fastConversationId, session.id)); + await expect(renewFastSessionRespondingLease(session.id)).resolves.toBe( + true, + ); + const renewed = await readLease(); + expect(renewed?.getTime()).toBeGreaterThan(live.getTime()); + }); }); 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 ba7ddf4fb..ce6849afc 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 @@ -29,6 +29,7 @@ const mocks = vi.hoisted(() => ({ revokeMcpCapabilities: vi.fn(), reconcileRetryNotices: vi.fn(), markRetryNoticeInterruption: vi.fn(), + renewRespondingLease: vi.fn(), getUnifiedSession: vi.fn(), touchSessionActivity: vi.fn(), getSessionForTask: vi.fn(), @@ -93,6 +94,7 @@ vi.mock('../fast-agent-conversation-repository', () => ({ reconcileFastAgentInferenceRetryNotices: mocks.reconcileRetryNotices, markFastAgentInferenceRetryNoticeInterruption: mocks.markRetryNoticeInterruption, + renewFastSessionRespondingLease: mocks.renewRespondingLease, })); vi.mock('../../router', () => ({ @@ -237,6 +239,7 @@ import { FastAgentProcessShutdownError, FastAgentTurnLockLostError, } from '../fast-agent-turn-lock'; +import { FAST_RESPONDING_LEASE_RENEW_MS } from '../fast-agent-constants'; const baseParams = { question: 'What does this service do?', @@ -379,6 +382,7 @@ describe('answerFastAgentQuestion native OpenCode tools', () => { mocks.upsertMessage.mockResolvedValue({ initialHumanTurn: true }); mocks.reconcileRetryNotices.mockResolvedValue(0); mocks.markRetryNoticeInterruption.mockResolvedValue(undefined); + mocks.renewRespondingLease.mockResolvedValue(true); mocks.getActiveTasks.mockResolvedValue([]); mocks.getEnvironments.mockResolvedValue([ { @@ -2608,6 +2612,161 @@ describe('answerFastAgentQuestion native OpenCode tools', () => { } }); + it('renews the responding lease on wall clock while a long turn executes', async () => { + vi.useFakeTimers(); + mocks.getUnifiedSession.mockResolvedValue({ id: 'session-1' }); + let finishInference: (() => void) | undefined; + mocks.generateText.mockImplementation( + () => + new Promise((resolve) => { + finishInference = () => resolve('All done.'); + }), + ); + + try { + const answer = answerFastAgentQuestion({ + ...baseParams, + adapter: callbacks(), + }); + await vi.advanceTimersByTimeAsync(0); + expect(mocks.touchSessionActivity).toHaveBeenCalledOnce(); + expect(mocks.renewRespondingLease).not.toHaveBeenCalled(); + + // No assistant message persists during this stretch; only the + // wall-clock renewal keeps the lease ahead of the reconciler. + await vi.advanceTimersByTimeAsync(FAST_RESPONDING_LEASE_RENEW_MS); + expect(mocks.renewRespondingLease).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(FAST_RESPONDING_LEASE_RENEW_MS); + expect(mocks.renewRespondingLease).toHaveBeenCalledTimes(2); + expect(mocks.renewRespondingLease).toHaveBeenCalledWith('conversation-1'); + + finishInference?.(); + await vi.advanceTimersByTimeAsync(1); + await answer; + + // Settling the turn clears the lease and stops the renewal timer. + expect(mocks.touchSessionActivity).toHaveBeenLastCalledWith( + expect.anything(), + 'session-1', + expect.any(Number), + { respondingUntil: null }, + ); + await vi.advanceTimersByTimeAsync(FAST_RESPONDING_LEASE_RENEW_MS * 2); + expect(mocks.renewRespondingLease).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + + it('stops renewing once ownership is lost, even for a queued renewal', async () => { + vi.useFakeTimers(); + const controller = new AbortController(); + const lockLost = new FastAgentTurnLockLostError(); + mocks.getUnifiedSession.mockResolvedValue({ id: 'session-1' }); + let releaseRenewal: (() => void) | undefined; + mocks.renewRespondingLease.mockImplementation(async () => { + // The first renewal stalls mid-write; a second tick queues behind it. + await new Promise((resolve) => { + releaseRenewal = resolve; + }); + return true; + }); + mocks.generateText.mockImplementation( + (_params: unknown, _session: unknown, options: { signal: AbortSignal }) => + new Promise((_resolve, reject) => { + options.signal.addEventListener( + 'abort', + () => reject(options.signal.reason), + { once: true }, + ); + }), + ); + + try { + const answer = answerFastAgentQuestion({ + ...baseParams, + adapter: callbacks(), + signal: controller.signal, + }); + const rejection = expect(answer).rejects.toBe(lockLost); + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(FAST_RESPONDING_LEASE_RENEW_MS); + expect(mocks.renewRespondingLease).toHaveBeenCalledTimes(1); + expect(releaseRenewal).toBeDefined(); + await vi.advanceTimersByTimeAsync(FAST_RESPONDING_LEASE_RENEW_MS); + + controller.abort(lockLost); + releaseRenewal?.(); + await vi.advanceTimersByTimeAsync(1); + await rejection; + + // The queued second renewal saw the lost ownership and never ran, and + // the fenced-off owner did not clear the lease. + expect(mocks.renewRespondingLease).toHaveBeenCalledTimes(1); + expect(mocks.touchSessionActivity).toHaveBeenCalledOnce(); + } finally { + vi.useRealTimers(); + } + }); + + it('waits out an in-flight lease renewal before settling the lease', async () => { + vi.useFakeTimers(); + mocks.getUnifiedSession.mockResolvedValue({ id: 'session-1' }); + let releaseRenewal: (() => void) | undefined; + mocks.renewRespondingLease.mockImplementation(async () => { + // The wall-clock renewal stalls mid-write. + await new Promise((resolve) => { + releaseRenewal = resolve; + }); + return true; + }); + let finishInference: (() => void) | undefined; + mocks.generateText.mockImplementation( + () => + new Promise((resolve) => { + finishInference = () => resolve('All done.'); + }), + ); + + try { + const answer = answerFastAgentQuestion({ + ...baseParams, + adapter: callbacks(), + }); + let settled = false; + void answer.finally(() => { + settled = true; + }); + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(FAST_RESPONDING_LEASE_RENEW_MS); + expect(releaseRenewal).toBeDefined(); + + finishInference?.(); + await vi.advanceTimersByTimeAsync(1); + // Settlement must wait for the stalled renewal so the terminal lease + // write cannot be overwritten by the stale extension. + expect(settled).toBe(false); + const settleWrites = () => + mocks.touchSessionActivity.mock.calls.filter( + ([, , , update]) => update?.respondingUntil === null, + ); + expect(settleWrites()).toHaveLength(0); + + releaseRenewal?.(); + await vi.advanceTimersByTimeAsync(1); + await answer; + expect(settleWrites()).toHaveLength(1); + expect(mocks.touchSessionActivity).toHaveBeenLastCalledWith( + expect.anything(), + 'session-1', + expect.any(Number), + { respondingUntil: null }, + ); + } finally { + vi.useRealTimers(); + } + }); + it('posts a terminal closeout when API shutdown interrupts silent retry backoff', async () => { const controller = new AbortController(); const shutdown = new FastAgentProcessShutdownError('SIGTERM'); diff --git a/packages/cloud-agents/src/server/fast-agent/fast-agent-constants.ts b/packages/cloud-agents/src/server/fast-agent/fast-agent-constants.ts index ce9b2952e..8d1561dcc 100644 --- a/packages/cloud-agents/src/server/fast-agent/fast-agent-constants.ts +++ b/packages/cloud-agents/src/server/fast-agent/fast-agent-constants.ts @@ -4,6 +4,11 @@ export const FAST_AGENT_MODEL_ROLE = 'orchestration' as const; // responses, short enough that a crashed turn self-heals the session status. // Streaming touch points re-extend it so long turns keep the lease fresh. export const FAST_RESPONDING_LEASE_MS = 15 * 60 * 1000; +// Assistant-message persists also extend the lease, but a turn can spend +// longer than the lease inside tool calls or a streaming stretch without +// persisting one; the time-based renewal keeps the lease fresh for exactly +// as long as the turn is actually executing. +export const FAST_RESPONDING_LEASE_RENEW_MS = FAST_RESPONDING_LEASE_MS / 3; export const FAST_AGENT_GITHUB_MCP_PATH = '/api/mcp-routing/github'; export const FAST_AGENT_TASKS_API_PATH = '/api/mcp/tasks'; export const FAST_AGENT_ENVIRONMENTS_API_PATH = '/api/mcp/environments'; 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 c82992554..f2ced7c5b 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 @@ -10,6 +10,8 @@ import { advanceSessionNotifiedCursor, advanceSessionReadCursor, getSessionForFastConversation, + gt, + isNotNull, isNull, lt, or, @@ -207,6 +209,33 @@ export async function markFastAgentInferenceRetryNoticeInterruption( return stamped.length > 0; } +/** + * Extend the responding lease with the fence in the statement itself: only a + * lease that is still live is extended, so a stale renewal from an owner that + * lost the conversation mid-write can never resurrect a lease a settlement + * or successor already cleared. No read precedes the write, which removes + * the check-then-write window entirely. Returns whether a lease was renewed. + */ +export async function renewFastSessionRespondingLease( + fastConversationId: string, +): Promise { + const renewed = await db + .update(sessions) + .set({ + respondingUntil: new Date(Date.now() + FAST_RESPONDING_LEASE_MS), + updatedAt: new Date(), + }) + .where( + and( + eq(sessions.fastConversationId, fastConversationId), + isNotNull(sessions.respondingUntil), + gt(sessions.respondingUntil, new Date()), + ), + ) + .returning({ id: sessions.id }); + return renewed.length > 0; +} + export async function reconcileExpiredFastAgentInferenceRetryNotices( limit = 100, ): Promise { 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 976e4bc8f..55a625e6b 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 @@ -61,6 +61,7 @@ import { getAvailableEnvironments, type RoutableEnvironment } from '../router'; import { FAST_AGENT_MODEL_ROLE, FAST_RESPONDING_LEASE_MS, + FAST_RESPONDING_LEASE_RENEW_MS, } from './fast-agent-constants'; import { buildFastAgentSystemPrompt } from './fast-agent-prompt'; import { @@ -95,6 +96,7 @@ import { INTERRUPTED_INFERENCE_RETRY_MESSAGE, markFastAgentInferenceRetryNoticeInterruption, reconcileFastAgentInferenceRetryNotices, + renewFastSessionRespondingLease, RESTARTED_ACTIVE_TURN_MESSAGE, type FastAgentInterruptionReason, } from './fast-agent-conversation-repository'; @@ -258,9 +260,13 @@ async function markFastAgentHumanFollowUpDelivered(id: string): Promise { async function setFastSessionResponding( fastConversationId: string, responding: boolean, + /** Re-checked after the session lookup, immediately before the write, so + * an owner fenced off mid-lookup cannot extend a successor's lease. */ + isOwnershipCurrent?: () => boolean, ): Promise { const session = await getSessionForFastConversation(db, fastConversationId); if (!session) return; + if (isOwnershipCurrent && !isOwnershipCurrent()) return; await touchSessionActivity(db, session.id, Math.floor(Date.now() / 1000), { respondingUntil: responding ? new Date(Date.now() + FAST_RESPONDING_LEASE_MS) @@ -1054,6 +1060,12 @@ export async function answerFastAgentQuestion({ let nativeSteer: NonTaskOpenCodeNativeSteer | undefined; let activeToolExecutions = 0; let humanSteerPollTimer: ReturnType | undefined; + let respondingLeaseRenewalTimer: ReturnType | undefined; + // Renewals chain onto this promise so settlement can await the in-flight + // write before recording the terminal lease state; a fire-and-forget tick + // could otherwise commit after the settle write and leave an idle Session + // marked responding for another lease. + let respondingLeaseRenewal: Promise = Promise.resolve(); let activeHumanSteerPoll = Promise.resolve(); const injectedHumanFollowUpIds = new Set(); const humanFollowUpTurnSeqs = new Map(); @@ -1553,11 +1565,38 @@ export async function answerFastAgentQuestion({ `[Fast Agent] Failed to reconcile interrupted inference retry notices: ${formatErrorForLog(error)}`, ); }); - await setFastSessionResponding(session.id, true).catch((error) => { + await setFastSessionResponding( + session.id, + true, + () => !signal?.aborted, + ).catch((error) => { console.warn( `[sessions] Failed to mark Fast Session active: ${formatErrorForLog(error)}`, ); }); + // Assistant-message persists extend the lease as a side effect, but a + // turn can spend longer than the lease inside tool calls or a streaming + // stretch without persisting one, and the expired-lease reconciler would + // then stamp its live retry notice as interrupted. Renew on wall clock + // for as long as this owner is executing; the tick stops renewing the + // moment ownership is aborted so a fenced-off owner cannot extend a + // successor's lease. + respondingLeaseRenewalTimer = setInterval(() => { + if (signal?.aborted) return; + respondingLeaseRenewal = respondingLeaseRenewal.then(async () => { + // The abort check is only a cheap short-circuit; correctness comes + // from the renewal statement itself, which extends the lease only + // where it is still live, so a stale write cannot resurrect a lease + // a settlement or successor already cleared. + if (signal?.aborted) return; + await renewFastSessionRespondingLease(session.id).catch((error) => { + console.warn( + `[sessions] Failed to renew Fast Session responding lease: ${formatErrorForLog(error)}`, + ); + }); + }); + }, FAST_RESPONDING_LEASE_RENEW_MS); + respondingLeaseRenewalTimer.unref(); durableOpenCodeSessionId = session.openCodeSessionId; activeOpenCodeSessionId = session.openCodeSessionId; diagnostics.setCanonicalConversationId(session.id); @@ -3136,6 +3175,11 @@ export async function answerFastAgentQuestion({ } return lastVisibleMessage || message; } finally { + if (respondingLeaseRenewalTimer) clearInterval(respondingLeaseRenewalTimer); + respondingLeaseRenewalTimer = undefined; + // Wait out any renewal already in flight so the terminal lease write + // below cannot be overwritten by a stale extension. + await respondingLeaseRenewal; signal?.removeEventListener('abort', stopHumanSteerPolling); stopHumanSteerPolling(); await activeHumanSteerPoll;