From 71862e164719ceef38f99a6f60458a957de0b26e Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Mon, 1 Jun 2026 11:55:59 +0200 Subject: [PATCH 1/2] fix(cloud-agent-next): preserve wrapper cleanup backoff --- .../src/persistence/CloudAgentSession.ts | 12 +- services/cloud-agent-next/src/router.test.ts | 62 +---- .../cloud-agent-next/src/session-service.ts | 216 ------------------ .../session/deletion-lifecycle.test.ts | 32 +++ 4 files changed, 49 insertions(+), 273 deletions(-) diff --git a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts index 7d5cd06895..cc2090aa8f 100644 --- a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts +++ b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts @@ -93,6 +93,7 @@ import { clearWrapperRuntimeIdentity, getWrapperLease, getWrapperRuntimeState, + nextWrapperLeaseDeadline, } from '../session/wrapper-runtime-state.js'; import { getSessionMessageState, @@ -1399,13 +1400,20 @@ export class CloudAgentSession extends DurableObject { } } + private async schedulePhysicalWrapperCleanupRetry(): Promise { + const deadline = + nextWrapperLeaseDeadline(await getWrapperLease(this.ctx.storage)) ?? + Date.now() + REAPER_INTERVAL_MS_DEFAULT; + await this.ctx.storage.setAlarm(deadline); + } + private async finalizeSessionDeletion( reason: 'explicit' | 'retention-expired' ): Promise { const metadata = await this.getMetadata(); if (!metadata) { if ((await getWrapperLease(this.ctx.storage)).state !== 'none') { - await this.scheduleAlarmAtOrBefore(Date.now() + 1_000); + await this.schedulePhysicalWrapperCleanupRetry(); return false; } } else { @@ -1414,7 +1422,7 @@ export class CloudAgentSession extends DurableObject { await supervisor.runMaintenance(Date.now()); if ((await getWrapperLease(this.ctx.storage)).state !== 'none') { if (reason === 'explicit') { - await this.scheduleAlarmAtOrBefore(Date.now() + 1_000); + await this.schedulePhysicalWrapperCleanupRetry(); } return false; } diff --git a/services/cloud-agent-next/src/router.test.ts b/services/cloud-agent-next/src/router.test.ts index f5b1e0217d..163ace1f3e 100644 --- a/services/cloud-agent-next/src/router.test.ts +++ b/services/cloud-agent-next/src/router.test.ts @@ -6,17 +6,13 @@ vi.mock('@cloudflare/sandbox', () => ({ getSandbox: vi.fn(), })); -const { - interruptMock, - buildContextMock, - getOrCreateSessionMock, - recordCloudAgentSessionFailureMock, -} = vi.hoisted(() => ({ - interruptMock: vi.fn(), - buildContextMock: vi.fn(), - getOrCreateSessionMock: vi.fn(), - recordCloudAgentSessionFailureMock: vi.fn(), -})); +const { buildContextMock, getOrCreateSessionMock, recordCloudAgentSessionFailureMock } = vi.hoisted( + () => ({ + buildContextMock: vi.fn(), + getOrCreateSessionMock: vi.fn(), + recordCloudAgentSessionFailureMock: vi.fn(), + }) +); const { getSandboxIdForSessionMock, metadataMock } = vi.hoisted(() => ({ getSandboxIdForSessionMock: vi.fn(), @@ -74,7 +70,6 @@ vi.mock('./session-service.js', () => ({ // eslint-disable-next-line @typescript-eslint/no-unsafe-return return metadataMock(); } - static interrupt = interruptMock; }, })); @@ -297,11 +292,6 @@ describe('router sessionId validation', () => { beforeEach(() => { vi.clearAllMocks(); - interruptMock.mockResolvedValue({ - success: true, - message: 'stopped', - processesFound: true, - }); buildContextMock.mockImplementation( ({ sandboxId, @@ -704,38 +694,9 @@ describe('router sessionId validation', () => { let caller: ReturnType; let cloudAgentSession: MockCAS; let mockSessionStub: MockSessionStub; - let mockSandbox: ReturnType; beforeEach(() => { vi.clearAllMocks(); - interruptMock.mockResolvedValue({ - success: true, - message: 'Interrupted execution using pkill', - processesFound: true, - }); - buildContextMock.mockImplementation( - ({ - sandboxId, - orgId, - userId, - sessionId, - }: { - sandboxId: string; - orgId: string | undefined; - userId: string; - sessionId: string; - }) => ({ - sandboxId, - orgId, - userId, - sessionId, - sessionHome: `/home/${sessionId}`, - workspacePath: `/workspace/${sessionId}`, - branchName: `session/${sessionId}`, - }) - ); - getOrCreateSessionMock.mockResolvedValue({ token: 'session' }); - mockSessionStub = { deleteSession: vi.fn().mockResolvedValue(undefined), markAsInterrupted: vi.fn().mockResolvedValue(undefined), @@ -778,13 +739,6 @@ describe('router sessionId validation', () => { }; cloudAgentSession = mockContext.env.CLOUD_AGENT_SESSION as unknown as MockCAS; - mockSandbox = {} as ReturnType; - vi.mocked(getSandbox).mockReturnValue(mockSandbox); - - vi.stubGlobal('scheduler', { - wait: vi.fn().mockResolvedValue(undefined), - }); - caller = appRouter.createCaller(mockContext); }); @@ -808,7 +762,6 @@ describe('router sessionId validation', () => { }); expect(mockSessionStub.interruptExecution).toHaveBeenCalled(); expect(getSandbox).not.toHaveBeenCalled(); - expect(interruptMock).not.toHaveBeenCalled(); }); it('short-circuits queued-only interrupts before creating a sandbox session', async () => { @@ -838,7 +791,6 @@ describe('router sessionId validation', () => { expect(mockSessionStub.markAsInterrupted).toHaveBeenCalled(); expect(mockSessionStub.interruptExecution).toHaveBeenCalled(); expect(getOrCreateSessionMock).not.toHaveBeenCalled(); - expect(interruptMock).not.toHaveBeenCalled(); expect(getSandbox).not.toHaveBeenCalled(); expect(cloudAgentSession.idFromName).toHaveBeenCalledWith(`test-user-123:${sessionId}`); }); diff --git a/services/cloud-agent-next/src/session-service.ts b/services/cloud-agent-next/src/session-service.ts index 18408fdd74..2a4b5e3525 100644 --- a/services/cloud-agent-next/src/session-service.ts +++ b/services/cloud-agent-next/src/session-service.ts @@ -5,7 +5,6 @@ import type { SandboxId, SessionContext, SessionId, - InterruptResult, } from './types.js'; import { generateSandboxId } from './sandbox-id.js'; import { normalizeKilocodeModel } from './persistence/model-utils.js'; @@ -2141,221 +2140,6 @@ export class SessionService { } } - /** - * Identifies and kills all kilocode processes running in a specific session's workspace. - * This allows clients to stop running executions in a session without deleting the session itself. - * - * @param usePkill - If true, uses `pkill -f` with sessionId pattern instead of sandbox.listProcesses/killProcess. - * This is a temporary workaround for environments where sandbox process APIs are unreliable. - */ - static async interrupt( - sandbox: SandboxInstance, - session: ExecutionSession, - sessionContext: SessionContext, - usePkill: boolean = false, - executionId?: string - ): Promise { - if (usePkill) { - return SessionService.interruptWithPkill(session, sessionContext, executionId); - } - return SessionService.interruptWithSandboxApi(sandbox, session, sessionContext); - } - - /** - * Interrupt using pkill -f with the sessionId as the pattern. - * This kills any process whose command line contains the sessionId. - */ - private static async interruptWithPkill( - session: ExecutionSession, - sessionContext: SessionContext, - executionId?: string - ): Promise { - const startTime = Date.now(); - const { sessionId } = sessionContext; - - try { - const attemptPkill = async (pattern: string, label: string) => { - logger.info('Interrupting session using pkill', { - sessionId, - label, - pattern, - }); - return session.exec(`pkill -f -- '${pattern}'`); - }; - - let execIdError: string | null = null; - - if (executionId) { - // Prefer the wrapper execution ID for v2 sessions. - // pkill -f matches against the full command line. - const execResult = await attemptPkill(`--execution-id=${executionId}`, 'executionId'); - if (execResult.exitCode === 0) { - return { - success: true, - message: 'Interrupted execution using pkill (executionId)', - processesFound: true, - }; - } - if (execResult.exitCode !== 1) { - execIdError = `pkill failed with exit code ${execResult.exitCode}: ${execResult.stderr}`; - logger.error('pkill command failed for executionId', { - sessionId, - executionId, - exitCode: execResult.exitCode, - stderr: execResult.stderr, - }); - } - } - - // Fall back to sessionId for legacy sessions. - const sessionResult = await attemptPkill(sessionId, 'sessionId'); - const elapsed = Date.now() - startTime; - - if (sessionResult.exitCode === 0) { - logger.info('pkill successfully killed processes', { - sessionId, - elapsedMs: elapsed, - }); - - return { - success: true, - message: execIdError - ? `Interrupted execution using pkill (sessionId fallback). ${execIdError}` - : 'Interrupted execution using pkill', - processesFound: true, - }; - } - if (sessionResult.exitCode === 1) { - logger.info('No matching processes found for pkill', { - sessionId, - elapsedMs: elapsed, - }); - - return { - success: true, - message: execIdError - ? `No running processes found for this session. ${execIdError}` - : 'No running processes found for this session', - processesFound: false, - }; - } - - logger.error('pkill command failed for sessionId', { - sessionId, - exitCode: sessionResult.exitCode, - stderr: sessionResult.stderr, - elapsedMs: elapsed, - }); - - return { - success: false, - message: execIdError - ? `${execIdError}; sessionId pkill failed with exit code ${sessionResult.exitCode}: ${sessionResult.stderr}` - : `pkill failed with exit code ${sessionResult.exitCode}: ${sessionResult.stderr}`, - processesFound: false, - }; - } catch (error) { - logger.error('Interrupt with pkill failed', { - sessionId, - error: error instanceof Error ? error.message : String(error), - }); - - throw error; - } - } - - /** - * Interrupt using sandbox.listProcesses and session.killProcess APIs. - * This is the original implementation that enumerates and kills processes individually. - */ - private static async interruptWithSandboxApi( - sandbox: SandboxInstance, - session: ExecutionSession, - sessionContext: SessionContext - ): Promise { - type ProcessInfo = { - id: string; - status: string; - command: string; - }; - - const startTime = Date.now(); - - try { - // List all processes in the sandbox - const processes = await sandbox.listProcesses(); - - // Filter for kilocode processes in this session's workspace - const targetProcesses = processes.filter((proc: ProcessInfo) => { - const isRunning = proc.status === 'running'; - const isKilocode = proc.command.includes('kilocode'); - const isInWorkspace = proc.command.includes(`--workspace=${sessionContext.workspacePath}`); - - return isRunning && isKilocode && isInWorkspace; - }); - - if (targetProcesses.length === 0) { - logger.info('No matching kilocode processes found to interrupt', { - sessionId: sessionContext.sessionId, - workspacePath: sessionContext.workspacePath, - }); - - return { - success: true, - message: 'No running kilocode processes found for this session', - processesFound: false, - }; - } - - // Kill each target process - const killed: string[] = []; - const failed: string[] = []; - - for (const proc of targetProcesses) { - try { - // Send SIGTERM for graceful termination (exit code 143) - // This allows the SSE stream to properly close with an expected exit code - await session.killProcess(proc.id, 'SIGTERM'); - killed.push(proc.id); - logger.info('Successfully killed process', { - processId: proc.id, - command: proc.command, - }); - } catch (error) { - failed.push(proc.id); - logger.error('Failed to kill process', { - processId: proc.id, - error: error instanceof Error ? error.message : String(error), - }); - } - } - - const elapsed = Date.now() - startTime; - logger.info('Interrupt operation completed', { - sessionId: sessionContext.sessionId, - killedCount: killed.length, - failedCount: failed.length, - elapsedMs: elapsed, - }); - - return { - success: killed.length > 0, - message: - killed.length > 0 - ? `Interrupted execution: killed ${killed.length} process(es)${failed.length > 0 ? `, ${failed.length} failed` : ''}` - : `Failed to kill any processes (${failed.length} attempts failed)`, - processesFound: true, - }; - } catch (error) { - logger.error('Interrupt operation failed', { - sessionId: sessionContext.sessionId, - error: error instanceof Error ? error.message : String(error), - }); - - throw error; - } - } - /** * Create a cli_sessions_v2 record via session-ingest RPC. * Called during session preparation so the DB record exists before execution. diff --git a/services/cloud-agent-next/test/integration/session/deletion-lifecycle.test.ts b/services/cloud-agent-next/test/integration/session/deletion-lifecycle.test.ts index d3440b4a32..3943befb4a 100644 --- a/services/cloud-agent-next/test/integration/session/deletion-lifecycle.test.ts +++ b/services/cloud-agent-next/test/integration/session/deletion-lifecycle.test.ts @@ -94,6 +94,38 @@ describe('session deletion physical cleanup', () => { }); }); + it('retains physical cleanup backoff while explicit deletion is pending', async () => { + const userId = 'user_delete_backoff'; + const sessionId = 'agent_delete_backoff'; + const stub = env.CLOUD_AGENT_SESSION.get( + env.CLOUD_AGENT_SESSION.idFromName(`${userId}:${sessionId}`) + ); + + const result = await runInDurableObject(stub, async instance => { + await registerReadySession(instance, { + sessionId, + userId, + prompt: 'delete backoff', + mode: 'code', + model: 'test-model', + }); + await establishOwnedWrapper(instance); + + await expect(instance.deleteSession()).rejects.toThrow( + 'Session deletion pending physical wrapper cleanup' + ); + const lease = await getWrapperLease(instance.ctx.storage); + const alarm = await instance.ctx.storage.getAlarm(); + await instance.ctx.storage.deleteAll(); + return { lease, alarm }; + }); + + expect(result.lease).toMatchObject({ state: 'stop_needed', attempts: 1 }); + expect(result.lease.state).toBe('stop_needed'); + if (result.lease.state !== 'stop_needed') throw new Error('Expected pending wrapper cleanup'); + expect(result.alarm).toBe(result.lease.nextAttemptAt); + }); + it('rejects new message admission while explicit deletion is pending', async () => { const userId = 'user_delete_reject_admission'; const sessionId = 'agent_delete_reject_admission'; From 96cf40722f9881cd54ae95904f4a0c14ab8bf1aa Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Mon, 1 Jun 2026 12:11:06 +0200 Subject: [PATCH 2/2] test(cloud-agent-next): remove redundant lease assertion --- .../test/integration/session/deletion-lifecycle.test.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/services/cloud-agent-next/test/integration/session/deletion-lifecycle.test.ts b/services/cloud-agent-next/test/integration/session/deletion-lifecycle.test.ts index 3943befb4a..f495e46777 100644 --- a/services/cloud-agent-next/test/integration/session/deletion-lifecycle.test.ts +++ b/services/cloud-agent-next/test/integration/session/deletion-lifecycle.test.ts @@ -121,7 +121,6 @@ describe('session deletion physical cleanup', () => { }); expect(result.lease).toMatchObject({ state: 'stop_needed', attempts: 1 }); - expect(result.lease.state).toBe('stop_needed'); if (result.lease.state !== 'stop_needed') throw new Error('Expected pending wrapper cleanup'); expect(result.alarm).toBe(result.lease.nextAttemptAt); });