diff --git a/services/cloud-agent-next/AGENTS.md b/services/cloud-agent-next/AGENTS.md index ee032f752b..efd2812a4f 100644 --- a/services/cloud-agent-next/AGENTS.md +++ b/services/cloud-agent-next/AGENTS.md @@ -121,6 +121,8 @@ This pattern blocks API endpoints from running for external contributors who don - `SessionMessageState` owns lifecycle/outbox status, terminal effect accounting, and a named immutable `admissionSnapshot` only for post-pending replay validation and recovery; predecessor records normalize into partial `legacyAdmissionConstraints` and never fabricate missing immutable input. Terminal and accepted/sent effects are repairable from pending/alarm replay and events use deterministic uniqueness. - Wrapper handoff is currently at-least-once under ambiguous delivery failures: the wrapper forwards prompt/command submissions directly to Kilo and does not query Kilo to suppress or recover duplicate `messageId` submissions. Duplicate prompt/command processing is an accepted edge-case trade-off until Kilo provides an atomic submit-or-return-existing contract. - When accepted work has no pending residue and its fenced wrapper runtime/socket is gone, disconnect or liveness expiry first reconciles each accepted message against the DO's stored kilocode events (`getAssistantMessageForUserMessage`): positive terminal evidence (assistant `time.completed` or a terminal assistant error) settles the message as `idle_reconciliation`; anything else terminalizes as wrapper failure without redispatch. There is still no live authoritative Kilo terminal query for redispatch; adding one remains separate lifecycle capability work. +- Physical wrapper cleanup exhaustion (`WRAPPER_STOP_MAX_ATTEMPTS` reached) is fenced but recoverable: the lease re-observes the sandbox on a slow cadence (`WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS`) and releases to `none` only after a confirmed `absent` observation, so a wedged sandbox reaped later by the container runtime does not brick the session. Recovery is observation-only (`observeWrappersWithoutWaking`) and never issues another stop: the attempt budget and its rollback fence still hold, and the probe must not wake a stopped container to ask about a process that cannot outlive it. Background rechecks stop after `WRAPPER_CLEANUP_EXHAUSTED_RECHECK_WINDOW_MS` so an unrecoverable exhaustion stops re-arming the DO alarm; explicit sends still force one probe afterwards. +- A pending-message flush blocked on exhaustion forces one out-of-cadence recheck (`recoverExhaustedDeliveryBlock` → `recheckExhaustedCleanup`) because a user is actively waiting, then retries on the `WRAPPER_CLEANUP_EXHAUSTED` budget before failing closed. The retry budget is what keeps the two halves consistent — recovery takes minutes, so terminalizing on the first blocked attempt would discard messages a later probe would have delivered — and failing closed at the end of it is what keeps a message from sitting `queued` with no terminal signal. The flush failure code must stay authoritative: `INTERNAL` is treated as non-authoritative by `recordPendingFlushFailure` and would terminalize the message under whatever earlier cause it carried. - Callback delivery retry policy is paired with `wrangler.jsonc`: `CALLBACK_DELIVERY_MAX_ATTEMPTS` includes the initial attempt, and each Cloud Agent Next callback queue consumer must configure `max_retries` for the remaining redeliveries. - Queue/drain emits unfenced `MessageDeliveryRequest`; only `AgentRuntime` may allocate/reuse current identity and construct `FencedWrapperDispatchRequest` with complete `WrapperRunFence` for downstream dispatch. - Session creation selects an explicit `ProfileResolutionPolicy` at the handler boundary. Implicit repository/default profile resolution is limited to the closed set of approved session origins; omitted, unknown, and non-approved automation origins fail closed unless they supply an explicit profile id. diff --git a/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.ts b/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.ts index 81473636ec..0b5d330340 100644 --- a/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.ts +++ b/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.ts @@ -770,6 +770,12 @@ export class CloudflareAgentSandbox implements AgentSandbox { }); } + async observeWrappersWithoutWaking(): Promise { + const sandbox = await this.getSandbox(); + if ((await isSandboxContainerRunning(sandbox)) === false) return { status: 'absent' }; + return this.discoverSessionWrappers(); + } + private async observeTarget(_target: WrapperStopTarget): Promise { // The lease is session-scoped: confirming absence must account for every // physical wrapper carrying this logical session marker, including duplicates. diff --git a/services/cloud-agent-next/src/agent-sandbox/protocol.ts b/services/cloud-agent-next/src/agent-sandbox/protocol.ts index b60552d08a..c13a968f05 100644 --- a/services/cloud-agent-next/src/agent-sandbox/protocol.ts +++ b/services/cloud-agent-next/src/agent-sandbox/protocol.ts @@ -131,6 +131,14 @@ export type EnsuredWrapper = export type AgentSandbox = { ensureWrapper(request: EnsureWrapperRequest): Promise; discoverSessionWrappers(): Promise; + /** + * Observe session wrappers without booting a stopped container. A wrapper is + * a process and a process cannot outlive its container, so "container not + * running" is already proof of absence. Callers that only need to learn + * whether a wrapper survives — not to stop one — use this instead of + * `discoverSessionWrappers`, whose container fetch wakes the container. + */ + observeWrappersWithoutWaking(): Promise; stopWrappers(request: { target: WrapperStopTarget; attemptId: string; diff --git a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts index e6ae60ae79..57ccb4d88e 100644 --- a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts +++ b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts @@ -806,11 +806,28 @@ export class CloudAgentSession extends DurableObject { }; } }, + observeWrappers: async () => { + if (this.physicalWrapperObserver) return this.physicalWrapperObserver(); + if (this.orchestrator) return { status: 'absent' }; + const metadata = await this.getStoredMetadata(); + if (!metadata) { + return { status: 'inspection-failed', error: 'Session metadata unavailable' }; + } + if ( + getSandboxProvider(metadata) === 'cloudflare' && + !this.env.Sandbox && + !this.env.SandboxSmall + ) { + return { status: 'absent' }; + } + return createAgentSandbox(this.env, metadata).observeWrappersWithoutWaking(); + }, recordSharedSandboxFailover: routeKey => this.sharedSandboxFailoverRecorder ? this.sharedSandboxFailoverRecorder(routeKey) : recordSharedSandboxFailover(this.env.SHARED_SANDBOX_OVERRIDES, routeKey), requestAlarmAtOrBefore: deadline => this.scheduleAlarmAtOrBefore(deadline), + isSessionDeletionInProgress: () => this.hasDeletionIntent(), getSessionIdForLogs: () => this.sessionId, }); } @@ -859,6 +876,12 @@ export class CloudAgentSession extends DurableObject { const retryAt = nextWrapperCleanupDeadline(lease); return retryAt === undefined ? null : { kind: 'retryable', retryAt }; }, + // A blocked flush means a user is waiting; let the supervisor force one + // out-of-cadence recheck so a reaped sandbox releases the lease + // immediately instead of failing the message on the stale fence. + recoverExhaustedDeliveryBlock: async () => { + await this.getWrapperSupervisor().recheckExhaustedCleanup(); + }, deliver: plan => this.executeDirectly(plan), isDeliveryHeld: async () => isWrapperRunFinalizing(await getWrapperRuntimeState(this.ctx.storage)), diff --git a/services/cloud-agent-next/src/persistence/schemas.test.ts b/services/cloud-agent-next/src/persistence/schemas.test.ts index 1576a5de7e..2c3e97cd3a 100644 --- a/services/cloud-agent-next/src/persistence/schemas.test.ts +++ b/services/cloud-agent-next/src/persistence/schemas.test.ts @@ -4,6 +4,7 @@ import { ImagesSchema, MCPServerConfigSchema, MetadataSchema, + modelIdSchema, RuntimeAgentSchema, RuntimeSkillSchema, RuntimeSkillsSchema, @@ -999,6 +1000,30 @@ describe('MetadataSchema with runtimeSkills', () => { }); }); +describe('modelIdSchema', () => { + it('accepts standard provider/model IDs', () => { + expect(modelIdSchema.parse('anthropic/claude-sonnet-4-20250514')).toBe( + 'anthropic/claude-sonnet-4-20250514' + ); + expect(modelIdSchema.parse('inclusionai/ling-3.0-flash:free')).toBe( + 'inclusionai/ling-3.0-flash:free' + ); + }); + + it('accepts tilde-prefixed latest aliases', () => { + expect(modelIdSchema.parse('~x-ai/grok-latest')).toBe('~x-ai/grok-latest'); + expect(modelIdSchema.parse('~anthropic/claude-sonnet-latest')).toBe( + '~anthropic/claude-sonnet-latest' + ); + }); + + it('rejects whitespace and other unsafe characters', () => { + expect(modelIdSchema.safeParse('x ai/grok').success).toBe(false); + expect(modelIdSchema.safeParse('x;ai/grok').success).toBe(false); + expect(modelIdSchema.safeParse('').success).toBe(false); + }); +}); + describe('RuntimeAgentSchema', () => { it('accepts a well-formed custom slug', () => { const agent = { slug: 'reviewer', name: 'Reviewer', config: {} }; diff --git a/services/cloud-agent-next/src/persistence/schemas.ts b/services/cloud-agent-next/src/persistence/schemas.ts index 1f489534ce..8a5d697988 100644 --- a/services/cloud-agent-next/src/persistence/schemas.ts +++ b/services/cloud-agent-next/src/persistence/schemas.ts @@ -175,8 +175,8 @@ export const modelIdSchema = z .min(1, 'Model ID cannot be empty') .max(255, 'Model ID too long') .regex( - /^[a-zA-Z0-9._\-/:]+$/, - 'Model ID can only contain alphanumeric characters, dots, dashes, underscores, slashes, and colons' + /^[a-zA-Z0-9._\-/:~]+$/, + 'Model ID can only contain alphanumeric characters, dots, dashes, underscores, slashes, colons, and tildes' ); /** diff --git a/services/cloud-agent-next/src/session/pending-messages.ts b/services/cloud-agent-next/src/session/pending-messages.ts index 8a656f50ff..04a2bbcefc 100644 --- a/services/cloud-agent-next/src/session/pending-messages.ts +++ b/services/cloud-agent-next/src/session/pending-messages.ts @@ -29,6 +29,12 @@ const WORKSPACE_CAPACITY_RETRY_DELAYS_MS = [10_000, 30_000, 60_000] as const; // minute — give it a short backed-off budget instead of one generic // redelivery. const GIT_RATE_LIMIT_RETRY_DELAYS_MS = [15_000, 45_000] as const; +// Wrapper cleanup exhaustion fences delivery, but it is recoverable: the lease +// releases once the wedged wrapper is observably gone, and every flush attempt +// forces one observation. Give the message a few spaced attempts so a container +// reaped moments after exhaustion still delivers, then fail closed rather than +// leaving it queued for the whole background recheck window. +const CLEANUP_EXHAUSTED_RETRY_DELAYS_MS = [30_000, 60_000, 120_000] as const; // Other pending delivery failures currently get one redelivery after the initial failed attempt. const WARM_FOLLOWUP_RETRY_DELAYS_MS = [PENDING_FLUSH_RETRY_BASE_DELAY_MS] as const; const COLD_INIT_RETRY_DELAYS_MS = [PENDING_FLUSH_RETRY_BASE_DELAY_MS] as const; @@ -86,6 +92,7 @@ const PendingFlushFailureCodeSchema = z.enum([ 'KILO_SERVER_FAILED', 'WRAPPER_START_FAILED', 'WRAPPER_FINALIZING', + 'WRAPPER_CLEANUP_EXHAUSTED', 'SANDBOX_CAPABILITY_UNAVAILABLE', 'NOT_FOUND', 'BAD_REQUEST', @@ -498,7 +505,8 @@ export function shouldSkipPendingFlush(message: PendingSessionMessage, now: numb /** * Reset-eligible modes each start a fresh retry budget on entry (sandbox-connect * has a short reconnect budget; sandbox-capacity and git-rate-limit each have a - * longer backed-off budget for transient, self-clearing conditions). + * longer backed-off budget for transient, self-clearing conditions; cleanup- + * exhausted has its own recovery budget). * Alternating between them must NOT keep resetting, so callers only reset when * entering one of these from a non-reset-eligible state. */ @@ -508,6 +516,7 @@ function isResetEligibleFailure( ): boolean { return ( code === 'SANDBOX_CONNECT_FAILED' || + code === 'WRAPPER_CLEANUP_EXHAUSTED' || (code === 'WORKSPACE_SETUP_FAILED' && (subtype === 'sandbox_storage_full' || subtype === 'git_rate_limited')) ); @@ -523,6 +532,7 @@ export async function recordPendingFlushFailure( code?: | RetryableResultCode | PermanentDeliveryResultCode + | 'WRAPPER_CLEANUP_EXHAUSTED' | 'NOT_FOUND' | 'BAD_REQUEST' | 'INTERNAL' @@ -564,11 +574,11 @@ export async function recordPendingFlushFailure( ? options.safeFailureMessage : undefined; // Reset the attempt counter only when a message ENTERS a reset-eligible - // transient mode (sandbox-connect, sandbox-capacity, or git-rate-limit) from a - // state that is not itself reset-eligible, so each fresh sequence gets its - // full backoff budget. When failures alternate between reset-eligible modes - // the counter is NOT reset, so attempts accumulate and the message still - // exhausts instead of flapping between modes forever. + // transient mode (sandbox-connect, sandbox-capacity, git-rate-limit, or + // cleanup-exhausted) from a state that is not itself reset-eligible, so each + // fresh sequence gets its full backoff budget. When failures alternate between + // reset-eligible modes the counter is NOT reset, so attempts accumulate and + // the message still exhausts instead of flapping between modes forever. const attempts = isResetEligibleFailure(flushFailureCode, failureSubtype) && !isResetEligibleFailure(message.lastFlushFailureCode, message.lastFlushFailureSubtype) @@ -577,13 +587,15 @@ export async function recordPendingFlushFailure( const retryDelays = flushFailureCode === 'SANDBOX_CONNECT_FAILED' ? SANDBOX_CONNECT_RETRY_DELAYS_MS - : flushFailureCode === 'WORKSPACE_SETUP_FAILED' && failureSubtype === 'sandbox_storage_full' - ? WORKSPACE_CAPACITY_RETRY_DELAYS_MS - : flushFailureCode === 'WORKSPACE_SETUP_FAILED' && failureSubtype === 'git_rate_limited' - ? GIT_RATE_LIMIT_RETRY_DELAYS_MS - : options.policy === 'cold-init' - ? COLD_INIT_RETRY_DELAYS_MS - : WARM_FOLLOWUP_RETRY_DELAYS_MS; + : flushFailureCode === 'WRAPPER_CLEANUP_EXHAUSTED' + ? CLEANUP_EXHAUSTED_RETRY_DELAYS_MS + : flushFailureCode === 'WORKSPACE_SETUP_FAILED' && failureSubtype === 'sandbox_storage_full' + ? WORKSPACE_CAPACITY_RETRY_DELAYS_MS + : flushFailureCode === 'WORKSPACE_SETUP_FAILED' && failureSubtype === 'git_rate_limited' + ? GIT_RATE_LIMIT_RETRY_DELAYS_MS + : options.policy === 'cold-init' + ? COLD_INIT_RETRY_DELAYS_MS + : WARM_FOLLOWUP_RETRY_DELAYS_MS; const retryable = options.retryable ?? isRetryableFlushCode(flushFailureCode); const exhausted = !retryable || attempts > retryDelays.length; const retryDelay = retryDelays[attempts - 1]; @@ -615,6 +627,7 @@ function isRetryableFlushCode( code: | RetryableResultCode | PermanentDeliveryResultCode + | 'WRAPPER_CLEANUP_EXHAUSTED' | 'NOT_FOUND' | 'BAD_REQUEST' | 'INTERNAL' @@ -629,7 +642,8 @@ function isRetryableFlushCode( code === 'SANDBOX_CONNECT_FAILED' || code === 'WORKSPACE_SETUP_FAILED' || code === 'KILO_SERVER_FAILED' || - code === 'WRAPPER_START_FAILED' + code === 'WRAPPER_START_FAILED' || + code === 'WRAPPER_CLEANUP_EXHAUSTED' ); } export async function deletePendingSessionMessageByMessageId( diff --git a/services/cloud-agent-next/src/session/session-message-queue.test.ts b/services/cloud-agent-next/src/session/session-message-queue.test.ts index b207723552..ddf087e59b 100644 --- a/services/cloud-agent-next/src/session/session-message-queue.test.ts +++ b/services/cloud-agent-next/src/session/session-message-queue.test.ts @@ -129,6 +129,7 @@ function createQueueHarness(options?: { failTerminalizationOnce?: boolean; ensureAcceptedMessageEffects?: (messageId: string) => Promise; getDeliveryBlock?: () => Promise; + recoverExhaustedDeliveryBlock?: () => Promise; }) { const storage = options?.storage ?? createMemoryStorage(); const events: QueueEvent[] = []; @@ -165,6 +166,7 @@ function createQueueHarness(options?: { validateModeAgainstRuntimeAgents: () => null, getDeliveryContext: async () => (metadata ? createContext(metadata) : null), getDeliveryBlock: options?.getDeliveryBlock ?? (async () => null), + recoverExhaustedDeliveryBlock: options?.recoverExhaustedDeliveryBlock, deliver, ensureQueuedMessageEvent: event => { if (failQueuedEvent) { @@ -655,6 +657,135 @@ describe('SessionMessageQueue', () => { expect(pending?.lastFlushError).toBeUndefined(); }); + it('recovers an exhausted cleanup block and delivers when the forced recheck clears it', async () => { + let blocked = true; + const recover = vi.fn(async () => { + blocked = false; + }); + const harness = createQueueHarness({ + getDeliveryBlock: async () => (blocked ? { kind: 'exhausted' as const } : null), + recoverExhaustedDeliveryBlock: recover, + }); + await harness.queue.admitSubmittedMessage({ + userId: 'user_test' as UserId, + turn: { type: 'prompt', id: FIRST_MESSAGE_ID, prompt: 'recover and deliver' }, + }); + + const drain = await harness.queue.drainNextPendingMessage(); + + expect(recover).toHaveBeenCalledOnce(); + expect(harness.deliver).toHaveBeenCalledOnce(); + expect(harness.terminalizations).toHaveLength(0); + expect(drain).toEqual({ retryAt: undefined, remainingPendingCount: 0 }); + await expect(listPendingSessionMessages(harness.storage)).resolves.toHaveLength(0); + }); + + it('retries on its own budget when the forced cleanup recheck does not clear the block', async () => { + const now = 200_000; + const recover = vi.fn(async () => {}); + const harness = createQueueHarness({ + getDeliveryBlock: async () => ({ kind: 'exhausted' as const }), + recoverExhaustedDeliveryBlock: recover, + }); + await storePendingSessionMessage( + harness.storage, + createPendingSessionMessage({ + messageId: FIRST_MESSAGE_ID, + role: 'user', + content: 'recheck keeps block', + createdAt: 1, + }) + ); + const clock = vi.spyOn(Date, 'now').mockReturnValue(now); + + let drain; + try { + drain = await harness.queue.drainNextPendingMessage(); + } finally { + clock.mockRestore(); + } + + // Exhaustion is recoverable, so one blocked attempt must not be terminal: + // the container may be reaped before the retry budget runs out. + expect(recover).toHaveBeenCalledOnce(); + expect(harness.deliver).not.toHaveBeenCalled(); + expect(harness.terminalizations).toHaveLength(0); + expect(drain).toEqual({ retryAt: now + 30_000, remainingPendingCount: 1 }); + const [pending] = await listPendingSessionMessages(harness.storage); + expect(pending).toMatchObject({ + flushAttempts: 1, + lastFlushFailureCode: 'WRAPPER_CLEANUP_EXHAUSTED', + }); + }); + + it('delivers when a later forced recheck clears a block that earlier attempts hit', async () => { + let blocked = true; + const harness = createQueueHarness({ + getDeliveryBlock: async () => (blocked ? { kind: 'exhausted' as const } : null), + recoverExhaustedDeliveryBlock: async () => {}, + }); + await storePendingSessionMessage( + harness.storage, + createPendingSessionMessage({ + messageId: FIRST_MESSAGE_ID, + role: 'user', + content: 'reaped between attempts', + createdAt: 1, + flushAttempts: 2, + lastFlushError: 'Wrapper cleanup exhausted', + lastFlushFailureCode: 'WRAPPER_CLEANUP_EXHAUSTED', + }) + ); + + blocked = false; + const drain = await harness.queue.drainNextPendingMessage(); + + expect(harness.deliver).toHaveBeenCalledOnce(); + expect(harness.terminalizations).toHaveLength(0); + expect(drain).toEqual({ remainingPendingCount: 0 }); + }); + + it('fails closed once the cleanup-exhausted retry budget is spent', async () => { + const now = 200_000; + const harness = createQueueHarness({ + getDeliveryBlock: async () => ({ kind: 'exhausted' as const }), + recoverExhaustedDeliveryBlock: async () => {}, + }); + await storePendingSessionMessage( + harness.storage, + createPendingSessionMessage({ + messageId: FIRST_MESSAGE_ID, + role: 'user', + content: 'recheck never clears', + createdAt: 1, + flushAttempts: 3, + nextFlushAttemptAt: now, + lastFlushError: 'Wrapper cleanup exhausted', + lastFlushFailureCode: 'WRAPPER_CLEANUP_EXHAUSTED', + }) + ); + const clock = vi.spyOn(Date, 'now').mockReturnValue(now); + + let drain; + try { + drain = await harness.queue.drainNextPendingMessage(); + } finally { + clock.mockRestore(); + } + + expect(harness.deliver).not.toHaveBeenCalled(); + expect(harness.terminalizations).toHaveLength(1); + expect(harness.terminalizations[0]?.params).toMatchObject({ + kind: 'failed', + reason: 'exhausted', + failureStage: 'pre_dispatch', + failureCode: 'delivery_failure_unknown', + error: 'Wrapper cleanup exhausted: the unresponsive wrapper could not be stopped or observed', + }); + expect(drain).toEqual({ retryAt: undefined, remainingPendingCount: 0 }); + await expect(listPendingSessionMessages(harness.storage)).resolves.toHaveLength(0); + }); + it('fails the second sandbox attempt when wrapper cleanup remains blocked', async () => { const now = 200_000; const harness = createQueueHarness({ @@ -746,7 +877,7 @@ describe('SessionMessageQueue', () => { expect(pending?.lastFlushError).toBeUndefined(); }); - it('retains a queued message without scheduling another cleanup after quarantine', async () => { + it('retries a queued message when delivery reports exhausted cleanup', async () => { const harness = createQueueHarness({ deliver: async () => { throw new WrapperCleanupBlockedError({ kind: 'exhausted' }); @@ -758,12 +889,78 @@ describe('SessionMessageQueue', () => { }); const drain = await harness.queue.drainNextPendingMessage(); + + expect(harness.terminalizations).toHaveLength(0); + expect(drain.remainingPendingCount).toBe(1); + expect(drain.retryAt).toBeDefined(); const [pending] = await listPendingSessionMessages(harness.storage); + expect(pending).toMatchObject({ + flushAttempts: 1, + lastFlushFailureCode: 'WRAPPER_CLEANUP_EXHAUSTED', + }); + }); - expect(drain).toEqual({ retryAt: undefined, remainingPendingCount: 1 }); - expect(pending?.messageId).toBe(FIRST_MESSAGE_ID); - expect(pending?.flushAttempts).toBeUndefined(); - expect(pending?.lastFlushError).toBeUndefined(); + it('does not attempt delivery while cleanup is exhausted', async () => { + const harness = createQueueHarness({ + getDeliveryBlock: async () => ({ kind: 'exhausted' }), + }); + await storePendingSessionMessage( + harness.storage, + createPendingSessionMessage({ + messageId: FIRST_MESSAGE_ID, + role: 'user', + content: 'blocked by exhausted cleanup', + createdAt: 1, + }) + ); + + const drain = await harness.queue.drainNextPendingMessage(); + + expect(harness.deliver).not.toHaveBeenCalled(); + expect(harness.terminalizations).toHaveLength(0); + expect(drain.remainingPendingCount).toBe(1); + await expect(listPendingSessionMessages(harness.storage)).resolves.toHaveLength(1); + }); + + it('attributes the terminal failure to cleanup exhaustion, not the earlier sandbox failure', async () => { + const now = 200_000; + const harness = createQueueHarness({ + getDeliveryBlock: async () => ({ kind: 'exhausted' }), + }); + await storePendingSessionMessage( + harness.storage, + createPendingSessionMessage({ + messageId: FIRST_MESSAGE_ID, + role: 'user', + content: 'sandbox retries then exhausted cleanup', + createdAt: 100_000, + flushAttempts: 3, + nextFlushAttemptAt: now, + lastFlushError: 'Sandbox connection failed during wrapper discovery', + lastFlushFailureCode: 'SANDBOX_CONNECT_FAILED', + }) + ); + const clock = vi.spyOn(Date, 'now').mockReturnValue(now); + + let drain; + try { + // Sandbox-connect is itself reset-eligible, so the counter carries over + // and the fourth attempt spends the budget. + drain = await harness.queue.drainNextPendingMessage(); + } finally { + clock.mockRestore(); + } + + expect(drain).toEqual({ retryAt: undefined, remainingPendingCount: 0 }); + expect(harness.deliver).not.toHaveBeenCalled(); + expect(harness.terminalizations).toHaveLength(1); + expect(harness.terminalizations[0]?.params).toMatchObject({ + kind: 'failed', + failureStage: 'pre_dispatch', + failureCode: 'delivery_failure_unknown', + error: 'Wrapper cleanup exhausted: the unresponsive wrapper could not be stopped or observed', + }); + await expect(listPendingSessionMessages(harness.storage)).resolves.toHaveLength(0); }); it('fails the second sandbox attempt when delivery discovers blocked cleanup', async () => { diff --git a/services/cloud-agent-next/src/session/session-message-queue.ts b/services/cloud-agent-next/src/session/session-message-queue.ts index c91daa9a86..d9bbbfaf8e 100644 --- a/services/cloud-agent-next/src/session/session-message-queue.ts +++ b/services/cloud-agent-next/src/session/session-message-queue.ts @@ -139,6 +139,13 @@ export type SessionMessageQueueDependencies = { validateModeAgainstRuntimeAgents: (metadata: SessionMetadata, mode: string) => string | null; getDeliveryContext: () => Promise; getDeliveryBlock: () => Promise; + /** + * Attempt one out-of-cadence recovery of an exhausted wrapper-cleanup block + * while a user is actively waiting on the flush. Implementations must be a + * no-op for non-exhausted blocks; the flush re-reads `getDeliveryBlock` + * afterwards to learn whether the block cleared. + */ + recoverExhaustedDeliveryBlock?: () => Promise; deliver: (plan: MessageDeliveryRequest) => Promise; isDeliveryHeld?: () => Promise; ensureQueuedMessageEvent: (event: PersistedQueuedMessageEvent & { entityId: string }) => void; @@ -176,6 +183,52 @@ function toFailureResult( }; } +/** + * Record a blocked flush when wrapper cleanup is exhausted. The lease fences off + * re-allocation, but exhaustion is recoverable — the supervisor releases it once + * the wedged wrapper is observably gone, and each flush attempt forces one + * observation. So this retries on the cleanup-exhausted budget (a container + * reaped moments after exhaustion still delivers) and then fails closed, because + * skipping indefinitely would leave the message `queued` with no terminal signal. + * + * The code must be `WRAPPER_CLEANUP_EXHAUSTED` rather than a generic `INTERNAL`: + * `recordPendingFlushFailure` treats `INTERNAL` as non-authoritative and would + * keep whatever earlier cause the message carried, terminalizing it with a stale + * reason. + */ +async function recordExhaustedCleanupFlushFailure( + params: { + storage: SessionMessageQueueStorage; + now: number; + scheduleTerminalizationRepair?: () => Promise; + }, + message: PendingSessionMessage, + policy: PendingFlushPolicy, + totalCount: number +): Promise { + const failure = await recordPendingFlushFailure( + params.storage, + message, + 'Wrapper cleanup exhausted: the unresponsive wrapper could not be stopped or observed', + params.now, + { + policy, + code: 'WRAPPER_CLEANUP_EXHAUSTED', + scheduleTerminalizationRepair: params.scheduleTerminalizationRepair, + } + ); + logger + .withFields({ + messageId: message.messageId, + attempts: failure.attempts, + exhausted: failure.exhausted, + nextFlushAttemptAt: failure.nextFlushAttemptAt, + logTag: 'pending_flush_cleanup_exhausted', + }) + .warn('Queued message blocked by exhausted wrapper cleanup'); + return toFailureResult(failure, totalCount); +} + function isSandboxConnectionRetry(message: PendingSessionMessage): boolean { return ( message.lastFlushFailureCode === 'SANDBOX_CONNECT_FAILED' && (message.flushAttempts ?? 0) >= 1 @@ -202,6 +255,7 @@ function classifyDeliveryFailure(code: PendingFlushFailureCode | undefined): { return { failureStage: 'pre_dispatch', failureCode: 'invalid_delivery_request' }; case 'MODEL_MISSING': return { failureStage: 'pre_dispatch', failureCode: 'model_missing' }; + case 'WRAPPER_CLEANUP_EXHAUSTED': case 'SANDBOX_CAPABILITY_UNAVAILABLE': case 'WRAPPER_FINALIZING': case 'INTERNAL': @@ -294,6 +348,7 @@ export async function flushNextPendingSessionMessage(params: { now: number; getDeliveryContext: () => Promise; getDeliveryBlock?: SessionMessageQueueDependencies['getDeliveryBlock']; + recoverExhaustedDeliveryBlock?: SessionMessageQueueDependencies['recoverExhaustedDeliveryBlock']; validateModeAgainstRuntimeAgents: SessionMessageQueueDependencies['validateModeAgainstRuntimeAgents']; deliver: (plan: MessageDeliveryRequest) => Promise; isDeliveryHeld?: () => Promise; @@ -411,8 +466,19 @@ export async function flushNextPendingSessionMessage(params: { await params.repairQueuedMessageEffects?.(intent); - const deliveryBlock = await params.getDeliveryBlock?.(); + let deliveryBlock = await params.getDeliveryBlock?.(); + if (deliveryBlock?.kind === 'exhausted' && params.recoverExhaustedDeliveryBlock) { + // A user is actively waiting on this flush; the recheck cadence gate exists + // to bound background churn, not to delay an explicit send. Force one + // bounded inspection — if the wedged sandbox has been reaped since + // exhaustion, the lease releases and delivery proceeds below. + await params.recoverExhaustedDeliveryBlock(); + deliveryBlock = await params.getDeliveryBlock?.(); + } if (deliveryBlock) { + if (deliveryBlock.kind === 'exhausted') { + return recordExhaustedCleanupFlushFailure(params, message, policy, totalCount); + } if (isSandboxConnectionRetry(message)) { const failure = await recordPendingFlushFailure( params.storage, @@ -429,7 +495,7 @@ export async function flushNextPendingSessionMessage(params: { } return { type: 'skipped', - nextFlushAttemptAt: deliveryBlock.kind === 'retryable' ? deliveryBlock.retryAt : undefined, + nextFlushAttemptAt: deliveryBlock.retryAt, remainingCount: totalCount, }; } @@ -446,6 +512,9 @@ export async function flushNextPendingSessionMessage(params: { startResult = await params.deliver(plan); } catch (error) { if (error instanceof WrapperCleanupBlockedError) { + if (error.block.kind === 'exhausted') { + return recordExhaustedCleanupFlushFailure(params, message, policy, totalCount); + } if (isSandboxConnectionRetry(message)) { const failure = await recordPendingFlushFailure( params.storage, @@ -462,7 +531,7 @@ export async function flushNextPendingSessionMessage(params: { } return { type: 'skipped', - nextFlushAttemptAt: error.block.kind === 'retryable' ? error.block.retryAt : undefined, + nextFlushAttemptAt: error.block.retryAt, remainingCount: totalCount, }; } @@ -545,6 +614,7 @@ export function createSessionMessageQueue( validateModeAgainstRuntimeAgents, getDeliveryContext, getDeliveryBlock, + recoverExhaustedDeliveryBlock, deliver, isDeliveryHeld, ensureQueuedMessageEvent, @@ -922,6 +992,7 @@ export function createSessionMessageQueue( now, getDeliveryContext, getDeliveryBlock, + recoverExhaustedDeliveryBlock, validateModeAgainstRuntimeAgents, deliver, isDeliveryHeld, diff --git a/services/cloud-agent-next/src/session/wrapper-runtime-state.test.ts b/services/cloud-agent-next/src/session/wrapper-runtime-state.test.ts index 8d0f1012a4..8bb93b513a 100644 --- a/services/cloud-agent-next/src/session/wrapper-runtime-state.test.ts +++ b/services/cloud-agent-next/src/session/wrapper-runtime-state.test.ts @@ -8,6 +8,7 @@ import { getWrapperLease, getWrapperRuntimeState, hasCompleteWrapperRunMessageIndex, + isWrapperCleanupExhausted, isWrapperDeliveryHeld, markWrapperFinalizing, nextSandboxRecoveryDeadline, @@ -18,6 +19,9 @@ import { recordWrapperAcceptedMessage, reduceSandboxRecoveryState, reduceWrapperLease, + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_WINDOW_MS, + WRAPPER_STOP_MAX_ATTEMPTS, } from './wrapper-runtime-state.js'; type MemoryStorage = Pick & DurableObjectStorage; @@ -209,7 +213,7 @@ describe('WrapperLease', () => { expect(nextWrapperLeaseDeadline(retrying)).toBe(5_200); }); - it('keeps an exhausted cleanup quarantined without another deadline', () => { + it('keeps an exhausted cleanup quarantined between slow recheck deadlines', () => { const requested = reduceWrapperLease(emptyWrapperLease(), { type: 'request_stop', target: { kind: 'session' }, @@ -238,9 +242,10 @@ describe('WrapperLease', () => { attempts: 5, exhaustedAt: 300, lastError: 'inspection failed', + nextRecheckAt: 300 + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, }); - expect(nextWrapperCleanupDeadline(exhausted)).toBeUndefined(); - expect(nextWrapperLeaseDeadline(exhausted)).toBeUndefined(); + expect(nextWrapperCleanupDeadline(exhausted)).toBe(300 + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS); + expect(nextWrapperLeaseDeadline(exhausted)).toBe(300 + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS); expect(isWrapperDeliveryHeld(emptyWrapperRuntimeState(), exhausted)).toBe(true); expect( reduceWrapperLease(exhausted, { @@ -252,6 +257,104 @@ describe('WrapperLease', () => { ).toEqual(exhausted); }); + it('falls back to one recheck interval for exhausted leases without a recheck deadline', () => { + const exhausted = reduceWrapperLease(emptyWrapperLease(), { + type: 'request_stop', + target: { kind: 'session' }, + reason: 'observation-failed', + now: 100, + }); + if (exhausted.state !== 'stop_needed') throw new Error('Expected cleanup request'); + const legacyExhausted = { + ...exhausted, + attempts: WRAPPER_STOP_MAX_ATTEMPTS, + exhaustedAt: 5_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + }; + + expect(nextWrapperCleanupDeadline(legacyExhausted)).toBe( + 5_000 + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS + ); + }); + + it('drops the maintenance deadline once the exhausted recovery window closes', () => { + const exhausted = reduceWrapperLease(emptyWrapperLease(), { + type: 'request_stop', + target: { kind: 'session' }, + reason: 'observation-failed', + now: 100, + }); + if (exhausted.state !== 'stop_needed') throw new Error('Expected cleanup request'); + const exhaustedAt = 5_000; + const base = { + ...exhausted, + attempts: WRAPPER_STOP_MAX_ATTEMPTS, + exhaustedAt, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + }; + const lastInWindow = exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_WINDOW_MS; + + expect(nextWrapperCleanupDeadline({ ...base, nextRecheckAt: lastInWindow })).toBe(lastInWindow); + // Past the window the lease stops re-arming the alarm, so an unrecoverable + // exhaustion does not wake the DO every recheck interval forever. + expect( + nextWrapperCleanupDeadline({ ...base, nextRecheckAt: lastInWindow + 1 }) + ).toBeUndefined(); + expect(nextWrapperLeaseDeadline({ ...base, nextRecheckAt: lastInWindow + 1 })).toBeUndefined(); + }); + + it('schedules the next exhausted recheck and releases to none on confirmed absence', () => { + const requested = reduceWrapperLease(emptyWrapperLease(), { + type: 'request_stop', + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + now: 100, + }); + if (requested.state !== 'stop_needed') throw new Error('Expected cleanup request'); + const stopping = reduceWrapperLease( + { ...requested, attempts: 4 }, + { type: 'begin_stop_attempt', attemptId: 'attempt_fifth', now: 200, attemptDeadlineAt: 300 } + ); + const exhausted = reduceWrapperLease(stopping, { + type: 'cleanup_exhausted', + attemptId: 'attempt_fifth', + now: 300, + error: 'inspection failed', + }); + if (!isWrapperCleanupExhausted(exhausted)) throw new Error('Expected exhausted cleanup'); + + const rescheduled = reduceWrapperLease(exhausted, { + type: 'exhausted_recheck_scheduled', + now: 1_000, + }); + expect(rescheduled).toMatchObject({ + state: 'stop_needed', + exhaustedAt: 300, + attempts: 5, + nextRecheckAt: 1_000 + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + }); + + const released = reduceWrapperLease(rescheduled, { type: 'release_exhausted' }); + expect(released).toEqual({ state: 'none', nextInstanceGeneration: 1 }); + }); + + it('ignores exhausted recheck events for non-exhausted leases', () => { + const requested = reduceWrapperLease(emptyWrapperLease(), { + type: 'request_stop', + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + now: 100, + }); + + expect( + reduceWrapperLease(requested, { type: 'exhausted_recheck_scheduled', now: 200 }) + ).toEqual(requested); + expect(reduceWrapperLease(requested, { type: 'release_exhausted' })).toEqual(requested); + expect(reduceWrapperLease(emptyWrapperLease(), { type: 'release_exhausted' })).toEqual( + emptyWrapperLease() + ); + }); + it('counts only fresh list-processes timeouts and persists publication deadlines', () => { const routeKey = `usr-${'d'.repeat(48)}` as SandboxId; expect( @@ -463,7 +566,12 @@ describe('WrapperLease', () => { exhaustedAt: expect.any(Number), }); expect(isWrapperDeliveryHeld(emptyWrapperRuntimeState(), repaired)).toBe(true); - expect(nextWrapperCleanupDeadline(repaired)).toBeUndefined(); + // Quarantined leases predate nextRecheckAt; the deadline falls back to one + // recheck interval after exhaustion so the quarantine is recoverable. + const exhaustedAt = isWrapperCleanupExhausted(repaired) ? repaired.exhaustedAt : 0; + expect(nextWrapperCleanupDeadline(repaired)).toBe( + exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS + ); await expect(getWrapperLease(storage)).resolves.toEqual(repaired); }); }); diff --git a/services/cloud-agent-next/src/session/wrapper-runtime-state.ts b/services/cloud-agent-next/src/session/wrapper-runtime-state.ts index 1e7eda3254..cc4969ecbf 100644 --- a/services/cloud-agent-next/src/session/wrapper-runtime-state.ts +++ b/services/cloud-agent-next/src/session/wrapper-runtime-state.ts @@ -15,6 +15,26 @@ const WRAPPER_LEASE_KEY = 'wrapper_lease'; const SANDBOX_RECOVERY_STATE_KEY = 'sandbox_recovery_state'; const CLEANUP_EXHAUSTED_ROLLBACK_FENCE_MS = 100 * 365 * 24 * 60 * 60 * 1_000; export const WRAPPER_STOP_MAX_ATTEMPTS = 5; +/** + * How long an exhausted cleanup waits before re-observing the sandbox once. + * Exhaustion usually means the sandbox was too wedged to answer listProcesses; + * the container runtime later reaps it (e.g. activity_expired), after which a + * single re-observation sees `absent` and the lease can be released instead + * of fencing delivery forever. The cadence only bounds background churn from + * maintenance alarms — a flush blocked on an exhausted lease forces an + * immediate recheck because a user is actively waiting on it. + */ +export const WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS = 10 * 60 * 1_000; +/** + * How long after exhaustion background rechecks keep running. Exhaustion used + * to end all maintenance deadlines for the lease; rechecks re-arm the alarm, so + * they need their own horizon or every exhausted session wakes its DO every ten + * minutes for the rest of the session TTL. A wedged container that has not been + * reaped within the window is not going to be, and explicit sends still force a + * recheck afterwards. Leases exhausted before the window elapsed (including + * ones persisted before rechecks existed) get one recheck and then stop. + */ +export const WRAPPER_CLEANUP_EXHAUSTED_RECHECK_WINDOW_MS = 60 * 60 * 1_000; const wrapperInstanceLeaseSchema = z.object({ instanceId: z.string().min(1), @@ -103,6 +123,7 @@ const wrapperLeaseSchema = z attempts: z.number().int().nonnegative(), lastError: z.string().optional(), exhaustedAt: z.number().int().nonnegative().optional(), + nextRecheckAt: z.number().int().nonnegative().optional(), }), z.object({ state: z.literal('stopping'), @@ -167,7 +188,9 @@ export type WrapperLeaseEvent = | { type: 'stop_absent'; attemptId: string } | { type: 'stop_not_confirmed'; attemptId: string; retryAt: number; error: string } | { type: 'stop_attempt_expired'; attemptId: string; retryAt: number } - | { type: 'cleanup_exhausted'; attemptId?: string; now: number; error: string }; + | { type: 'cleanup_exhausted'; attemptId?: string; now: number; error: string } + | { type: 'exhausted_recheck_scheduled'; now: number } + | { type: 'release_exhausted' }; export const emptyWrapperLease = (): WrapperLease => ({ state: 'none', @@ -441,7 +464,14 @@ export function reduceWrapperLease(state: WrapperLease, event: WrapperLeaseEvent attempts: state.attempts, lastError: event.error, exhaustedAt: event.now, + nextRecheckAt: event.now + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, }; + case 'exhausted_recheck_scheduled': + if (!isWrapperCleanupExhausted(state)) return state; + return { ...state, nextRecheckAt: event.now + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS }; + case 'release_exhausted': + if (!isWrapperCleanupExhausted(state)) return state; + return { state: 'none', nextInstanceGeneration: state.nextInstanceGeneration }; } } @@ -451,8 +481,23 @@ export function isWrapperCleanupExhausted( return lease.state === 'stop_needed' && lease.exhaustedAt !== undefined; } +/** + * When the next background recheck of an exhausted cleanup is due, or undefined + * once the recovery window has closed. Single source of truth for both the + * maintenance alarm deadline and the supervisor's cadence gate. + */ +export function nextExhaustedWrapperRecheckAt( + lease: Extract & { exhaustedAt: number } +): number | undefined { + // Leases persisted before rechecks existed have no nextRecheckAt; fall back + // to one interval after exhaustion so they still get a recovery observation. + const recheckAt = lease.nextRecheckAt ?? lease.exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS; + if (recheckAt > lease.exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_WINDOW_MS) return undefined; + return recheckAt; +} + export function nextWrapperCleanupDeadline(lease: WrapperLease): number | undefined { - if (isWrapperCleanupExhausted(lease)) return undefined; + if (isWrapperCleanupExhausted(lease)) return nextExhaustedWrapperRecheckAt(lease); if (lease.state === 'stop_needed') return lease.nextAttemptAt; if (lease.state === 'stopping') return lease.attemptDeadlineAt; return undefined; diff --git a/services/cloud-agent-next/src/session/wrapper-supervisor.test.ts b/services/cloud-agent-next/src/session/wrapper-supervisor.test.ts index 6c588f0c2d..d47c719ff3 100644 --- a/services/cloud-agent-next/src/session/wrapper-supervisor.test.ts +++ b/services/cloud-agent-next/src/session/wrapper-supervisor.test.ts @@ -21,6 +21,9 @@ import { getSandboxRecoveryState, getWrapperLease, getWrapperRuntimeState, + putWrapperLease, + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_WINDOW_MS, } from './wrapper-runtime-state.js'; import type { LatestAssistantMessage } from './types.js'; import type { SandboxId } from '../types.js'; @@ -123,6 +126,7 @@ function createHarness( wrapperRunId: string ) => Promise; recordSharedSandboxFailover?: (routeKey: SandboxId) => Promise; + isSessionDeletionInProgress?: () => Promise; } ) { const getAssistantMessageForUserMessage = @@ -134,6 +138,7 @@ function createHarness( const sentPings: string[] = []; const stops: string[] = []; const stopWrappers = vi.fn().mockResolvedValue({ status: 'absent' }); + const observeWrappers = vi.fn().mockResolvedValue({ status: 'absent' }); const requestedAlarms: number[] = []; const currentMetadata = options?.metadata ?? createMetadata(); const settlementOutbox = createMessageSettlementOutbox({ @@ -179,10 +184,12 @@ function createHarness( ensureAcceptedMessageBeforeTerminal: options?.ensureAcceptedMessageBeforeTerminal ?? (async () => {}), stopWrappers, + observeWrappers, recordSharedSandboxFailover: routeKey => recordSharedSandboxFailover(routeKey), requestAlarmAtOrBefore: async deadline => { requestedAlarms.push(deadline); }, + isSessionDeletionInProgress: options?.isSessionDeletionInProgress, getSessionIdForLogs: () => currentMetadata.identity.sessionId, }); @@ -194,6 +201,7 @@ function createHarness( sentPings, stops, stopWrappers, + observeWrappers, requestedAlarms, requestPendingDrainIfNeeded, recordSharedSandboxFailover, @@ -202,6 +210,26 @@ function createHarness( }; } +function exhaustedLease(overrides: { + exhaustedAt: number; + nextRecheckAt?: number; +}): [string, unknown] { + return [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + requestedAt: 1_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + attempts: 5, + lastError: 'Wrapper process discovery timed out', + ...overrides, + }, + ]; +} + function liveRuntimeState(overrides?: Record): [string, unknown] { return [ 'wrapper_runtime_state', @@ -2620,7 +2648,11 @@ describe('WrapperSupervisor', () => { }); await harness.supervisor.runMaintenance(25_000); expect(publish).toHaveBeenCalledTimes(4); - await expect(harness.supervisor.nextMaintenanceDeadlines()).resolves.toEqual([]); + // Exhausted cleanup still carries its recheck deadline (exhaustedAt: 1 has + // no persisted nextRecheckAt, so the fallback interval applies). + await expect(harness.supervisor.nextMaintenanceDeadlines()).resolves.toEqual([ + 1 + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + ]); }); it('quarantines cleanup after five failed attempts', async () => { @@ -2670,10 +2702,11 @@ describe('WrapperSupervisor', () => { }); expect(harness.stopWrappers).toHaveBeenCalledTimes(5); await harness.supervisor.runMaintenance(now + 60_000); + // Still inside the recheck interval: no extra stop is attempted. expect(harness.stopWrappers).toHaveBeenCalledTimes(5); - await expect(harness.supervisor.nextMaintenanceDeadlines()).resolves.not.toContainEqual( - expect.any(Number) - ); + await expect(harness.supervisor.nextMaintenanceDeadlines()).resolves.toEqual([ + now + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + ]); }); it('quarantines the fifth cleanup attempt when its watchdog expires', async () => { @@ -2703,9 +2736,358 @@ describe('WrapperSupervisor', () => { exhaustedAt: 47_000, lastError: 'Stop attempt deadline expired', }); + await expect(harness.supervisor.nextMaintenanceDeadlines()).resolves.toEqual([ + 47_000 + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + ]); + }); + + it('releases an exhausted cleanup once a recheck confirms wrapper absence', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([ + [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + requestedAt: 1_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + attempts: 5, + lastError: 'Wrapper process discovery timed out', + exhaustedAt, + nextRecheckAt: exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + }, + ], + ]); + harness.observeWrappers.mockResolvedValue({ status: 'absent' }); + + await harness.supervisor.runMaintenance(exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS); + + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + // The attempt budget is spent and the rollback fence stands: recovery may + // observe the sandbox but must never issue another stop. + expect(harness.stopWrappers).not.toHaveBeenCalled(); + await expect(getWrapperLease(harness.storage)).resolves.toEqual({ + state: 'none', + nextInstanceGeneration: 2, + }); + expect(harness.requestPendingDrainIfNeeded).toHaveBeenCalledOnce(); + }); + + it('does not recheck an exhausted cleanup before its recheck deadline', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([ + [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + requestedAt: 1_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + attempts: 5, + exhaustedAt, + nextRecheckAt: exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + }, + ], + ]); + + await harness.supervisor.runMaintenance(exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS - 1); + + expect(harness.observeWrappers).not.toHaveBeenCalled(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ + state: 'stop_needed', + exhaustedAt, + }); + expect(harness.requestPendingDrainIfNeeded).not.toHaveBeenCalled(); + }); + + it('falls back to one interval for exhausted leases without a persisted recheck deadline', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([ + [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + requestedAt: 1_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + attempts: 5, + exhaustedAt, + }, + ], + ]); + harness.observeWrappers.mockResolvedValue({ status: 'absent' }); + + await harness.supervisor.runMaintenance(exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS - 1); + expect(harness.observeWrappers).not.toHaveBeenCalled(); + + await harness.supervisor.runMaintenance(exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS); + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ state: 'none' }); + }); + + it('keeps an exhausted cleanup fenced when the recheck still observes the wrapper', async () => { + const exhaustedAt = 10_000; + const recheckAt = exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS; + const harness = createHarness([ + [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + requestedAt: 1_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + attempts: 5, + exhaustedAt, + nextRecheckAt: recheckAt, + }, + ], + ]); + harness.observeWrappers.mockResolvedValue({ status: 'present', observed: [] }); + + await harness.supervisor.runMaintenance(recheckAt); + + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + expect(harness.stopWrappers).not.toHaveBeenCalled(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ + state: 'stop_needed', + exhaustedAt, + nextRecheckAt: recheckAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + }); + expect(harness.requestPendingDrainIfNeeded).not.toHaveBeenCalled(); + }); + + it('records sandbox inspection failures from an exhausted recheck without releasing', async () => { + const exhaustedAt = 10_000; + const recheckAt = exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS; + const harness = createHarness([ + [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + requestedAt: 1_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + attempts: 5, + exhaustedAt, + nextRecheckAt: recheckAt, + }, + ], + ]); + harness.observeWrappers.mockResolvedValue({ + status: 'inspection-failed', + error: 'Wrapper process discovery timed out', + reason: 'wrapper_discovery_list_processes_timeout', + }); + + await harness.supervisor.runMaintenance(recheckAt); + + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ + state: 'stop_needed', + exhaustedAt, + nextRecheckAt: recheckAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + }); + await expect(getSandboxRecoveryState(harness.storage)).resolves.toMatchObject({ + listProcessesTimeouts: 1, + }); + expect(harness.requestPendingDrainIfNeeded).not.toHaveBeenCalled(); + }); + + it('releases a legacy exhausted cleanup quarantined without a recheck deadline', async () => { + const harness = createHarness([ + [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'observation-failed', + requestedAt: 1, + nextAttemptAt: 3_153_600_000_001, + attempts: 5, + lastError: 'inspection failed', + exhaustedAt: 1, + }, + ], + ]); + harness.observeWrappers.mockResolvedValue({ status: 'absent' }); + + await harness.supervisor.runMaintenance(1 + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS); + + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ state: 'none' }); + expect(harness.requestPendingDrainIfNeeded).toHaveBeenCalledOnce(); + }); + + it('forces an exhausted cleanup recheck on demand even inside the cadence window', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([ + [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + requestedAt: 1_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + attempts: 5, + exhaustedAt, + nextRecheckAt: exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + }, + ], + ]); + harness.observeWrappers.mockResolvedValue({ status: 'absent' }); + + await harness.supervisor.recheckExhaustedCleanup(); + + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + expect(harness.stopWrappers).not.toHaveBeenCalled(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ state: 'none' }); + expect(harness.requestPendingDrainIfNeeded).toHaveBeenCalledOnce(); + }); + + it('keeps the lease fenced when the forced recheck still observes the wrapper', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([ + [ + 'wrapper_lease', + { + state: 'stop_needed', + nextInstanceGeneration: 2, + target: { kind: 'session' }, + reason: 'unhealthy-wrapper', + requestedAt: 1_000, + nextAttemptAt: Number.MAX_SAFE_INTEGER, + attempts: 5, + exhaustedAt, + nextRecheckAt: exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + }, + ], + ]); + harness.observeWrappers.mockResolvedValue({ status: 'present', observed: [] }); + + await harness.supervisor.recheckExhaustedCleanup(); + + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ + state: 'stop_needed', + exhaustedAt, + }); + expect(harness.requestPendingDrainIfNeeded).not.toHaveBeenCalled(); + }); + + it('does not force a recheck when the lease is not exhausted', async () => { + const harness = createHarness([OWNED_WRAPPER_LEASE]); + + await harness.supervisor.recheckExhaustedCleanup(); + + expect(harness.observeWrappers).not.toHaveBeenCalled(); + }); + + it('stops background rechecks once the recovery window closes', async () => { + const exhaustedAt = 10_000; + const closedAt = exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_WINDOW_MS; + const harness = createHarness([exhaustedLease({ exhaustedAt, nextRecheckAt: closedAt + 1 })]); + harness.observeWrappers.mockResolvedValue({ status: 'absent' }); + + await harness.supervisor.runMaintenance(closedAt + 60_000); + + expect(harness.observeWrappers).not.toHaveBeenCalled(); + // No deadline left means the DO stops waking for this lease, which is the + // behaviour exhaustion had before rechecks re-armed the alarm. await expect(harness.supervisor.nextMaintenanceDeadlines()).resolves.toEqual([]); }); + it('still forces a recheck after the recovery window closes', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([ + exhaustedLease({ + exhaustedAt, + nextRecheckAt: exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_WINDOW_MS + 1, + }), + ]); + harness.observeWrappers.mockResolvedValue({ status: 'absent' }); + + await harness.supervisor.recheckExhaustedCleanup(); + + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ state: 'none' }); + }); + + it('does not recheck while session deletion owns physical teardown', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([exhaustedLease({ exhaustedAt })], { + isSessionDeletionInProgress: async () => true, + }); + + await harness.supervisor.recheckExhaustedCleanup(); + + expect(harness.observeWrappers).not.toHaveBeenCalled(); + await expect(getWrapperLease(harness.storage)).resolves.toMatchObject({ + state: 'stop_needed', + exhaustedAt, + }); + }); + + it('does not release a lease that changed while the recheck was in flight', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([exhaustedLease({ exhaustedAt })]); + harness.observeWrappers.mockImplementation(async () => { + // A concurrent path released the lease and allocated a fresh wrapper. + await putWrapperLease(harness.storage, { + state: 'none', + nextInstanceGeneration: 3, + }); + return { status: 'absent' }; + }); + + await harness.supervisor.recheckExhaustedCleanup(); + + await expect(getWrapperLease(harness.storage)).resolves.toEqual({ + state: 'none', + nextInstanceGeneration: 3, + }); + expect(harness.requestPendingDrainIfNeeded).not.toHaveBeenCalled(); + }); + + it('runs one shared probe when maintenance and a forced recheck overlap', async () => { + const exhaustedAt = 10_000; + const harness = createHarness([ + exhaustedLease({ + exhaustedAt, + nextRecheckAt: exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS, + }), + ]); + let releaseProbe: () => void = () => {}; + harness.observeWrappers.mockImplementation(async () => { + await new Promise(resolve => { + releaseProbe = resolve; + }); + return { status: 'present', observed: [] }; + }); + + const maintenance = harness.supervisor.runMaintenance( + exhaustedAt + WRAPPER_CLEANUP_EXHAUSTED_RECHECK_MS + ); + await vi.waitFor(() => expect(harness.observeWrappers).toHaveBeenCalledOnce()); + const forced = harness.supervisor.recheckExhaustedCleanup(); + releaseProbe(); + await Promise.all([maintenance, forced]); + + expect(harness.observeWrappers).toHaveBeenCalledOnce(); + }); + it('retries thrown cleanup and does not issue a parallel stop during a valid watchdog', async () => { const harness = createHarness([ [ diff --git a/services/cloud-agent-next/src/session/wrapper-supervisor.ts b/services/cloud-agent-next/src/session/wrapper-supervisor.ts index 6ee11af746..31aee2f087 100644 --- a/services/cloud-agent-next/src/session/wrapper-supervisor.ts +++ b/services/cloud-agent-next/src/session/wrapper-supervisor.ts @@ -3,6 +3,7 @@ import { logger } from '../logger.js'; import type { SessionMetadata } from '../persistence/session-metadata.js'; import type { StopWrappersResult, + WrapperObservation, WrapperStopReason, WrapperStopTarget, } from '../agent-sandbox/protocol.js'; @@ -60,6 +61,7 @@ import { resetWrapperLivenessAfterReconnect, type WrapperConnectionFence, type WrapperRuntimeState, + nextExhaustedWrapperRecheckAt, WRAPPER_STOP_MAX_ATTEMPTS, } from './wrapper-runtime-state.js'; @@ -147,6 +149,7 @@ export type WrapperSupervisor = { onDisconnected(input: WrapperDisconnectedInput): Promise; onTerminalEvent(params: WrapperTerminalEvent): Promise; requestPhysicalWrapperStop(reason: WrapperStopReason, target?: WrapperStopTarget): Promise; + recheckExhaustedCleanup(): Promise; clearDisconnectGrace(): Promise; runMaintenance(now: number): Promise; nextMaintenanceDeadlines(): Promise; @@ -185,8 +188,14 @@ export type WrapperSupervisorDependencies = { attemptId: string; reason: WrapperStopReason; }) => Promise; + /** + * Observation-only probe used to recover an exhausted cleanup lease. Must not + * stop wrappers and must not wake a stopped container. + */ + observeWrappers?: () => Promise; recordSharedSandboxFailover: (routeKey: SandboxId) => Promise; requestAlarmAtOrBefore?: (deadline: number) => Promise; + isSessionDeletionInProgress?: () => Promise; getSessionIdForLogs: () => string | undefined; }; @@ -411,8 +420,10 @@ export function createWrapperSupervisor( clearInterruptRequest, ensureAcceptedMessageBeforeTerminal, stopWrappers, + observeWrappers, recordSharedSandboxFailover, requestAlarmAtOrBefore, + isSessionDeletionInProgress, getSessionIdForLogs, } = dependencies; @@ -663,7 +674,33 @@ export function createWrapperSupervisor( if (next !== current) { await putWrapperLease(storage, next); await requestAlarmAtOrBefore?.(now); + return; } + // The reducer only accepts request_stop from `none`/`owns_wrapper`, so a + // no-op means cleanup was already in flight (or exhausted). That is usually + // fine — the existing lease deadline drives reconciliation — but it must + // never be silent: a stuck-session investigation needs to see that the stop + // request was received and why it did not schedule new work. Only the + // exhausted case warns; a stop already in flight is the common benign path. + const exhausted = isWrapperCleanupExhausted(current); + const entry = logger.withFields({ + sessionId: getSessionIdForLogs(), + reason, + leaseState: current.state, + leaseReason: + current.state === 'stop_needed' || current.state === 'stopping' + ? current.reason + : undefined, + attempts: + current.state === 'stop_needed' || current.state === 'stopping' + ? current.attempts + : undefined, + exhausted, + logTag: 'wrapper_stop_request_noop', + }); + const message = 'Physical wrapper stop request left the existing cleanup lease unchanged'; + if (exhausted) entry.warn(message); + else entry.info(message); } /** @@ -1222,6 +1259,133 @@ export function createWrapperSupervisor( } } + /** + * An exhausted cleanup no longer retries stops, but the physical sandbox can + * still change underneath it — the container runtime reaps wedged containers + * (e.g. activity_expired), after which the wrapper is observably gone. + * Re-observe on a slow cadence and release the lease once absence is + * confirmed so the session can allocate a fresh wrapper for new messages. + * Without this, exhaustion fenced delivery for the rest of the session TTL. + * + * This is deliberately an observation, never a stop attempt: the exhausted + * lease has spent its `WRAPPER_STOP_MAX_ATTEMPTS` budget and its rollback + * fence says so, and the observation must not wake a stopped container just + * to ask whether a process that cannot outlive it is still there. + * + * Callers with a user actively waiting (pending-message flush) pass `force` + * to skip the cadence gate and the recovery window; an explicit send is worth + * one probe. + */ + async function recheckExhaustedPhysicalCleanup( + lease: Extract>, { state: 'stop_needed' }> & { + exhaustedAt: number; + }, + now: number, + options?: { force?: boolean } + ): Promise { + if (!observeWrappers) return; + // While the session is being deleted, wholesale sandbox deletion owns + // physical teardown; re-observing individual wrappers is pointless churn + // (and the sandbox may already be gone). Rechecks exist to recover sessions + // that still have future work. + if (await isSessionDeletionInProgress?.()) return; + if (!options?.force) { + const recheckAt = nextExhaustedWrapperRecheckAt(lease); + if (recheckAt === undefined || now < recheckAt) return; + } + + // Re-read after the awaits above: the lease we were handed may already have + // been released (and a fresh wrapper allocated) by a concurrent recheck, and + // the cadence write below must not resurrect it. + const current = await getWrapperLease(storage); + if (!isWrapperCleanupExhausted(current) || current.exhaustedAt !== lease.exhaustedAt) return; + // Rate-limit before probing so a hanging or throwing observation cannot + // hot-loop on every maintenance pass. + await putWrapperLease( + storage, + reduceWrapperLease(current, { type: 'exhausted_recheck_scheduled', now }) + ); + + logger + .withFields({ + sessionId: getSessionIdForLogs(), + reason: lease.reason, + target: lease.target, + attempts: lease.attempts, + requestedAt: lease.requestedAt, + exhaustedAt: lease.exhaustedAt, + forced: options?.force === true, + logTag: 'wrapper_cleanup_exhausted_recheck', + }) + .info('Rechecking exhausted wrapper cleanup'); + + let observation: WrapperObservation; + try { + observation = await observeWrappers(); + } catch (error) { + observation = { + status: 'inspection-failed', + error: error instanceof Error ? error.message : String(error), + }; + } + if (observation.status === 'inspection-failed') { + await recordSandboxInspectionFailure(storage, observation.reason); + } + if (observation.status !== 'absent') { + logger + .withFields({ + sessionId: getSessionIdForLogs(), + reason: lease.reason, + observation: observation.status, + logTag: 'wrapper_cleanup_exhausted_recheck', + }) + .info('Exhausted wrapper cleanup recheck did not confirm absence'); + return; + } + + const latest = await getWrapperLease(storage); + if (!isWrapperCleanupExhausted(latest) || latest.exhaustedAt !== lease.exhaustedAt) return; + const released = reduceWrapperLease(latest, { type: 'release_exhausted' }); + if (released.state !== 'none') return; + await putWrapperLease(storage, released); + await clearSettledSandboxRecovery(storage); + logger + .withFields({ + sessionId: getSessionIdForLogs(), + reason: lease.reason, + logTag: 'wrapper_cleanup_exhausted_recheck', + }) + .info('Released exhausted wrapper cleanup after confirmed absence'); + if (!isWrapperDeliveryHeld(await getWrapperRuntimeState(storage), released)) { + await sessionMessageQueue.requestPendingDrainIfNeeded(); + } + } + + let exhaustedRecheck: Promise | undefined; + + /** + * Single-flight wrapper: a maintenance alarm and a blocked flush can both + * reach the recheck at the same time, and one probe answers both. + */ + async function recheckExhaustedPhysicalCleanupOnce( + lease: Extract>, { state: 'stop_needed' }> & { + exhaustedAt: number; + }, + now: number, + options?: { force?: boolean } + ): Promise { + if (exhaustedRecheck) { + await exhaustedRecheck; + return; + } + exhaustedRecheck = recheckExhaustedPhysicalCleanup(lease, now, options); + try { + await exhaustedRecheck; + } finally { + exhaustedRecheck = undefined; + } + } + async function performSharedSandboxFailoverReconciliation(): Promise { const currentTime = Date.now(); let recovery = await getSandboxRecoveryState(storage); @@ -1313,7 +1477,10 @@ export function createWrapperSupervisor( async function reconcilePhysicalCleanup(now: number): Promise { if (!stopWrappers) return; let lease = await getWrapperLease(storage); - if (isWrapperCleanupExhausted(lease)) return; + if (isWrapperCleanupExhausted(lease)) { + await recheckExhaustedPhysicalCleanupOnce(lease, now); + return; + } if (lease.state === 'stop_needed' && lease.attempts >= WRAPPER_STOP_MAX_ATTEMPTS) { await exhaustPhysicalCleanup( lease, @@ -1658,6 +1825,17 @@ export function createWrapperSupervisor( return deadlines; } + /** + * Force one out-of-cadence recheck of an exhausted cleanup lease. Used when a + * pending-message flush is blocked on exhaustion and a user is actively + * waiting on the outcome. No-op unless the persisted lease is exhausted. + */ + async function recheckExhaustedCleanup(): Promise { + const lease = await getWrapperLease(storage); + if (!isWrapperCleanupExhausted(lease)) return; + await recheckExhaustedPhysicalCleanupOnce(lease, Date.now(), { force: true }); + } + return { checkReconnect, recordReconnectAccepted, @@ -1668,6 +1846,7 @@ export function createWrapperSupervisor( onDisconnected, onTerminalEvent, requestPhysicalWrapperStop, + recheckExhaustedCleanup, clearDisconnectGrace, runMaintenance, nextMaintenanceDeadlines, diff --git a/services/cloud-agent-next/src/terminal/access.test.ts b/services/cloud-agent-next/src/terminal/access.test.ts index b456e4555c..63e5f3ddb1 100644 --- a/services/cloud-agent-next/src/terminal/access.test.ts +++ b/services/cloud-agent-next/src/terminal/access.test.ts @@ -101,6 +101,7 @@ function sandboxWithTerminalResult( return { ensureWrapper: vi.fn(), discoverSessionWrappers: vi.fn(), + observeWrappersWithoutWaking: vi.fn(), stopWrappers: vi.fn(), probeHealth: vi.fn(), getRunningWrapper: vi.fn(),