fix: properly clean up processes on every code path

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

@@ -142,7 +142,7 @@ interface ManagedSession {
   taskId: string;
   repoPath: string;
   agent: Agent;
-  connection: ClientSideConnection;
+  clientSideConnection: ClientSideConnection;
   channel: string;
   createdAt: number;
   lastActivityAt: number;
@@ -422,7 +422,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
         taskId,
         repoPath,
         agent,
-        connection,
+        clientSideConnection: connection,
         channel,
         createdAt: Date.now(),
         lastActivityAt: Date.now(),
@@ -438,6 +438,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(
@@ -518,7 +523,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,
       });
@@ -530,7 +535,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,
         });
@@ -565,7 +570,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,
       });
@@ -591,7 +596,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
     }
 
     try {
-      await session.connection.extMethod("session/setModel", {
+      await session.clientSideConnection.extMethod("session/setModel", {
         sessionId,
         modelId,
       });
@@ -609,7 +614,7 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
     }
 
     try {
-      await session.connection.extMethod("session/setMode", {
+      await session.clientSideConnection.extMethod("session/setMode", {
         sessionId,
         modeId,
       });
@@ -722,32 +727,15 @@ For git operations while detached:
   }
 
   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");
   }
 
@@ -807,22 +795,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;