fix: properly clean up processes on every code path (#580)

When sessions were recreated, e.g. after token refresh, old agent processes were not properly being cleaned up.
This (probably) causes zombie processes such as in #523.

Author: jonathanlab

apps/twig/src/main/services/agent/service.ts

@@ -169,7 +169,7 @@ interface ManagedSession {
   taskId: string;
   repoPath: string;
   agent: Agent;
-  connection: ClientSideConnection;
+  clientSideConnection: ClientSideConnection;
   channel: string;
   createdAt: number;
   lastActivityAt: number;
@@ -449,7 +449,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
         taskId,
         repoPath,
         agent,
-        connection,
+        clientSideConnection: connection,
         channel,
         createdAt: Date.now(),
         lastActivityAt: Date.now(),
@@ -465,6 +465,11 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
       }
       return session;
     } catch (err) {
+      try {
+        await agent.cleanup();
+      } catch {
+        log.debug("Agent cleanup failed during error handling", { taskRunId });
+      }
       this.cleanupMockNodeEnvironment(mockNodeDir);
       if (!isRetry && isAuthError(err)) {
         log.warn(
@@ -545,7 +550,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
     session.promptPending = true;
 
     try {
-      const result = await session.connection.prompt({
+      const result = await session.clientSideConnection.prompt({
         sessionId,
         prompt: finalPrompt,
       });
@@ -557,7 +562,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
       if (isAuthError(err)) {
         log.warn("Auth error during prompt, recreating session", { sessionId });
         session = await this.recreateSession(sessionId);
-        const result = await session.connection.prompt({
+        const result = await session.clientSideConnection.prompt({
           sessionId,
           prompt: finalPrompt,
         });
@@ -592,7 +597,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
     if (!session) return false;
 
     try {
-      await session.connection.cancel({
+      await session.clientSideConnection.cancel({
         sessionId,
         _meta: reason ? { interruptReason: reason } : undefined,
       });
@@ -618,7 +623,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
     }
 
     try {
-      await session.connection.extMethod("session/setModel", {
+      await session.clientSideConnection.extMethod("session/setModel", {
         sessionId,
         modelId,
       });
@@ -636,7 +641,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
     }
 
     try {
-      await session.connection.extMethod("session/setMode", {
+      await session.clientSideConnection.extMethod("session/setMode", {
         sessionId,
         modeId,
       });
@@ -750,32 +755,15 @@ For git operations while detached:
 
   @preDestroy()
   async cleanupAll(): Promise<void> {
+    const sessionIds = Array.from(this.sessions.keys());
     log.info("Cleaning up all agent sessions", {
-      sessionCount: this.sessions.size,
+      sessionCount: sessionIds.length,
     });
 
-    for (const [taskRunId, session] of this.sessions) {
-      // Step 1: Send ACP cancel notification for any ongoing prompt turns
-      try {
-        if (!session.connection.signal.aborted) {
-          await session.connection.cancel({ sessionId: taskRunId });
-          log.info("Sent ACP cancel for session", { taskRunId });
-        }
-      } catch (err) {
-        log.warn("Failed to send ACP cancel", { taskRunId, error: err });
-      }
-
-      // Step 2: Cleanup agent connection (closes streams, aborts subprocess)
-      try {
-        await session.agent.cleanup();
-      } catch (err) {
-        log.warn("Failed to cleanup agent", { taskRunId, error: err });
-      }
-
-      this.cleanupMockNodeEnvironment(session.mockNodeDir);
+    for (const taskRunId of sessionIds) {
+      await this.cleanupSession(taskRunId);
     }
 
-    this.sessions.clear();
     log.info("All agent sessions cleaned up");
   }
 
@@ -835,22 +823,11 @@ For git operations while detached:
   private async cleanupSession(taskRunId: string): Promise<void> {
     const session = this.sessions.get(taskRunId);
     if (session) {
-      // Cancel any ongoing operations
-      try {
-        if (!session.connection.signal.aborted) {
-          await session.connection.cancel({ sessionId: taskRunId });
-        }
-      } catch {
-        // Ignore cancel errors
-      }
-
-      // Cleanup agent (closes streams, aborts subprocess)
       try {
         await session.agent.cleanup();
       } catch {
-        // Ignore cleanup errors
+        log.debug("Agent cleanup failed", { taskRunId });
       }
-
       this.cleanupMockNodeEnvironment(session.mockNodeDir);
       this.sessions.delete(taskRunId);
     }

packages/agent/src/adapters/acp-connection.ts

@@ -92,7 +92,7 @@ export function createAcpConnection(
       logger.info("Cleaning up ACP connection");
 
       if (agent) {
-        agent.closeAllSessions();
+        await agent.closeAllSessions();
       }
 
       // Then close the streams to properly terminate the ACP connection

packages/agent/src/adapters/base-acp-agent.ts

@@ -1,6 +1,14 @@
 import type {
+  Agent,
   AgentSideConnection,
   AuthenticateRequest,
+  CancelNotification,
+  InitializeRequest,
+  InitializeResponse,
+  NewSessionRequest,
+  NewSessionResponse,
+  PromptRequest,
+  PromptResponse,
   ReadTextFileRequest,
   ReadTextFileResponse,
   SessionNotification,
@@ -15,7 +23,7 @@ export interface BaseSession extends SessionState {
   interruptReason?: string;
 }
 
-export abstract class BaseAcpAgent {
+export abstract class BaseAcpAgent implements Agent {
   abstract readonly adapterName: string;
   protected sessions: { [key: string]: BaseSession } = {};
   client: AgentSideConnection;
@@ -27,13 +35,19 @@ export abstract class BaseAcpAgent {
     this.logger = new Logger({ debug: true, prefix: "[BaseAcpAgent]" });
   }
 
-  closeAllSessions(): void {
+  abstract initialize(request: InitializeRequest): Promise<InitializeResponse>;
+  abstract newSession(params: NewSessionRequest): Promise<NewSessionResponse>;
+  abstract prompt(params: PromptRequest): Promise<PromptResponse>;
+  abstract cancel(params: CancelNotification): Promise<void>;
+
+  async closeAllSessions(): Promise<void> {
     for (const [sessionId, session] of Object.entries(this.sessions)) {
       try {
+        await this.cancel({ sessionId });
         session.abortController.abort();
-        this.logger.info("Aborted session", { sessionId });
+        this.logger.info("Closed session", { sessionId });
       } catch (err) {
-        this.logger.warn("Failed to abort session", { sessionId, error: err });
+        this.logger.warn("Failed to close session", { sessionId, error: err });
       }
     }
     this.sessions = {};

packages/agent/src/adapters/claude/agent.ts

@@ -2,7 +2,6 @@ import * as fs from "node:fs";
 import * as os from "node:os";
 import * as path from "node:path";
 import {
-  type Agent,
   type AgentSideConnection,
   type AuthenticateRequest,
   type AvailableCommand,
@@ -146,7 +145,7 @@ async function getAvailableSlashCommands(
     );
 }
 
-export class ClaudeAcpAgent extends BaseAcpAgent implements Agent {
+export class ClaudeAcpAgent extends BaseAcpAgent {
   readonly adapterName = "claude";
   declare sessions: { [key: string]: Session };
   toolUseCache: ToolUseCache;