Merge pull request #444 from subsy/devin/1788434576-fix-agent-exit-completion

fix(agents): complete an agent execution once the process exits

Author: subsy

src/plugins/agents/base.ts

@@ -10,7 +10,7 @@ import { tmpdir, platform } from 'node:os';
 import { isAbsolute, join } from 'node:path';
 
 /** Debug log helper - writes to file to avoid TUI interference */
-function debugLog(msg: string): void {
+export function debugLog(msg: string): void {
   // Always log during debugging phase (TODO: restore RALPH_DEBUG check later)
   try {
     const logPath = join(tmpdir(), 'ralph-agent-debug.log');
@@ -175,6 +175,10 @@ interface RunningExecution {
   reject: (error: Error) => void;
   timeoutId?: ReturnType<typeof setTimeout>;
   options?: AgentExecuteOptions;
+  drainFallbackId?: ReturnType<typeof setTimeout>;
+  exitCode?: number | null;
+  exitSignal?: NodeJS.Signals | null;
+  exitAt?: number;
 }
 
 /**
@@ -201,6 +205,12 @@ const MAX_EXECUTION_STREAM_CHARS = 2_000_000;
  */
 const STREAM_TRUNCATED_PREFIX = '[...agent output truncated in memory...]\n';
 
+// Balances capturing late descendant output against delaying completed agents.
+export const STDIO_DRAIN_GRACE_PERIOD_MS = 2000;
+
+// Output arriving this long after exit is attributed to a descendant, not the agent.
+export const STDIO_DRAIN_MAX_WAIT_MS = 10_000;
+
 /**
  * Append chunk data while enforcing an in-memory size cap.
  * Keeps the most recent content (tail) to preserve completion markers near the end.
@@ -713,6 +723,9 @@ export abstract class BaseAgentPlugin implements AgentPlugin {
           MAX_EXECUTION_STREAM_CHARS
         );
         options?.onStdout?.(text);
+        if (execution.drainFallbackId !== undefined) {
+          this.scheduleDrainFallback(execution);
+        }
       });
 
       // Handle stderr
@@ -724,6 +737,9 @@ export abstract class BaseAgentPlugin implements AgentPlugin {
           MAX_EXECUTION_STREAM_CHARS
         );
         options?.onStderr?.(text);
+        if (execution.drainFallbackId !== undefined) {
+          this.scheduleDrainFallback(execution);
+        }
       });
 
       // Handle process error
@@ -738,18 +754,8 @@ export abstract class BaseAgentPlugin implements AgentPlugin {
           debugLog(`[DEBUG] Process close: code=${code}, signal=${signal}, execId=${executionId}`);
         }
 
-        // Determine status
-        let status: AgentExecutionStatus;
-        if (execution.interrupted) {
-          status = 'interrupted';
-        } else if (signal === 'SIGTERM' || signal === 'SIGKILL') {
-          status = execution.timeoutId ? 'timeout' : 'interrupted';
-        } else if (code === 0) {
-          status = 'completed';
-        } else {
-          status = 'failed';
-        }
-
+        this.clearDrainFallback(execution);
+        const status = this.deriveExitStatus(execution, code, signal);
         this.completeExecution(executionId, status, code ?? undefined);
       });
 
@@ -758,8 +764,10 @@ export abstract class BaseAgentPlugin implements AgentPlugin {
         if (process.env.RALPH_DEBUG) {
           debugLog(`[DEBUG] Process exit: code=${code}, signal=${signal}, execId=${executionId}`);
         }
-        // Note: We don't call completeExecution here to avoid double-completion
-        // 'close' should fire after 'exit' once stdio streams are closed
+        execution.exitCode = code;
+        execution.exitSignal = signal;
+        execution.exitAt = Date.now();
+        this.scheduleDrainFallback(execution);
       });
 
       // Set up timeout if specified
@@ -895,6 +903,71 @@ export abstract class BaseAgentPlugin implements AgentPlugin {
     };
   }
 
+  /**
+   * Derive the final execution status from process termination details.
+   */
+  private deriveExitStatus(
+    execution: RunningExecution,
+    code: number | null,
+    signal: NodeJS.Signals | null
+  ): AgentExecutionStatus {
+    if (execution.interrupted) {
+      return 'interrupted';
+    }
+    if (signal === 'SIGTERM' || signal === 'SIGKILL') {
+      return execution.timeoutId ? 'timeout' : 'interrupted';
+    }
+    return code === 0 ? 'completed' : 'failed';
+  }
+
+  /**
+   * Cancel a pending fallback timer for an execution.
+   */
+  private clearDrainFallback(execution: RunningExecution): void {
+    if (execution.drainFallbackId !== undefined) {
+      clearTimeout(execution.drainFallbackId);
+      execution.drainFallbackId = undefined;
+    }
+  }
+
+  /**
+   * Schedule bounded completion after an exited process's stdio goes quiet.
+   */
+  private scheduleDrainFallback(execution: RunningExecution): void {
+    this.clearDrainFallback(execution);
+
+    const remainingBudget =
+      (execution.exitAt ?? Date.now()) + STDIO_DRAIN_MAX_WAIT_MS - Date.now();
+    const delay = Math.min(STDIO_DRAIN_GRACE_PERIOD_MS, remainingBudget);
+    const finalize = (): void => {
+      execution.drainFallbackId = undefined;
+      if (!this.executions.has(execution.executionId)) {
+        return;
+      }
+
+      if (process.env.RALPH_DEBUG) {
+        debugLog(
+          `[DEBUG] Process stdio stayed open after drain: code=${execution.exitCode}, ` +
+            `signal=${execution.exitSignal}, execId=${execution.executionId}`
+        );
+      }
+      const status = this.deriveExitStatus(
+        execution,
+        execution.exitCode ?? null,
+        execution.exitSignal ?? null
+      );
+      this.completeExecution(execution.executionId, status, execution.exitCode ?? undefined);
+      execution.process.stdout?.destroy();
+      execution.process.stderr?.destroy();
+    };
+
+    if (delay <= 0) {
+      finalize();
+      return;
+    }
+    execution.drainFallbackId = setTimeout(finalize, delay);
+  }
+
   /**
    * Complete an execution and resolve its promise.
    */
@@ -920,6 +993,7 @@ export abstract class BaseAgentPlugin implements AgentPlugin {
     if (execution.timeoutId) {
       clearTimeout(execution.timeoutId);
     }
+    this.clearDrainFallback(execution);
 
     const endedAt = new Date();
     const durationMs = endedAt.getTime() - execution.startedAt.getTime();

src/plugins/agents/droid/index.ts

@@ -6,7 +6,13 @@
 import { spawn } from 'node:child_process';
 import { randomUUID } from 'node:crypto';
 import { platform } from 'node:os';
-import { BaseAgentPlugin, quoteForWindowsShell } from '../base.js';
+import {
+  BaseAgentPlugin,
+  debugLog,
+  quoteForWindowsShell,
+  STDIO_DRAIN_GRACE_PERIOD_MS,
+  STDIO_DRAIN_MAX_WAIT_MS,
+} from '../base.js';
 import type {
   AgentPluginMeta,
   AgentPluginFactory,
@@ -230,24 +236,63 @@ export class DroidAgentPlugin extends BaseAgentPlugin {
     let stdout = '';
     let stderr = '';
     let interrupted = false;
+    let completed = false;
     let timeoutId: ReturnType<typeof setTimeout> | undefined;
+    let drainFallbackId: ReturnType<typeof setTimeout> | undefined;
+    let exitCode: number | null = null;
+    let exitSignal: NodeJS.Signals | null = null;
+    let exitAt: number | undefined;
 
     options?.onStart?.(executionId);
 
+    const clearDrainFallback = (): void => {
+      if (drainFallbackId !== undefined) {
+        clearTimeout(drainFallbackId);
+        drainFallbackId = undefined;
+      }
+    };
+
+    const deriveExitStatus = (
+      code: number | null,
+      signal: NodeJS.Signals | null
+    ): AgentExecutionStatus => {
+      if (interrupted) {
+        return 'interrupted';
+      }
+      if (signal === 'SIGTERM' || signal === 'SIGKILL') {
+        return timeoutId ? 'timeout' : 'interrupted';
+      }
+      if (code === 0) {
+        return 'completed';
+      }
+      return 'failed';
+    };
+
     proc.stdout?.on('data', (data: Buffer) => {
       const text = data.toString();
       stdout += text;
       options?.onStdout?.(text);
+      if (drainFallbackId !== undefined) {
+        scheduleDrainFallback();
+      }
     });
 
     proc.stderr?.on('data', (data: Buffer) => {
       const text = data.toString();
       stderr += text;
       options?.onStderr?.(text);
+      if (drainFallbackId !== undefined) {
+        scheduleDrainFallback();
+      }
     });
 
     const complete = (status: AgentExecutionStatus, exitCode?: number, error?: string) => {
+      if (completed) {
+        return;
+      }
+      completed = true;
       if (timeoutId) clearTimeout(timeoutId);
+      clearDrainFallback();
       const endedAt = new Date();
       resolvePromise!({
         executionId,
@@ -263,22 +308,49 @@ export class DroidAgentPlugin extends BaseAgentPlugin {
       });
     };
 
+    const scheduleDrainFallback = (): void => {
+      if (completed) {
+        return;
+      }
+      clearDrainFallback();
+      const remainingBudget = (exitAt ?? Date.now()) + STDIO_DRAIN_MAX_WAIT_MS - Date.now();
+      const delay = Math.min(STDIO_DRAIN_GRACE_PERIOD_MS, remainingBudget);
+      const finalize = (): void => {
+        drainFallbackId = undefined;
+        if (completed) {
+          return;
+        }
+        if (process.env.RALPH_DEBUG) {
+          debugLog(
+            `[DEBUG] Process stdio stayed open after drain: code=${exitCode}, ` +
+              `signal=${exitSignal}, execId=${executionId}`
+          );
+        }
+        complete(deriveExitStatus(exitCode, exitSignal), exitCode ?? undefined);
+        proc.stdout?.destroy();
+        proc.stderr?.destroy();
+      };
+
+      if (delay <= 0) {
+        finalize();
+        return;
+      }
+      drainFallbackId = setTimeout(finalize, delay);
+    };
+
     proc.on('error', (error) => {
       complete('failed', undefined, error.message);
     });
 
     proc.on('close', (code, signal) => {
-      let status: AgentExecutionStatus;
-      if (interrupted) {
-        status = 'interrupted';
-      } else if (signal === 'SIGTERM' || signal === 'SIGKILL') {
-        status = timeoutId ? 'timeout' : 'interrupted';
-      } else if (code === 0) {
-        status = 'completed';
-      } else {
-        status = 'failed';
-      }
-      complete(status, code ?? undefined);
+      complete(deriveExitStatus(code, signal), code ?? undefined);
+    });
+
+    proc.on('exit', (code, signal) => {
+      exitCode = code;
+      exitSignal = signal;
+      exitAt = Date.now();
+      scheduleDrainFallback();
     });
 
     if (timeout > 0) {