fix: harden workflow execution recovery

src/workflow-engine.ts

@@ -23,7 +23,7 @@ import { execFileSync } from "node:child_process";
 import { resolveWorkflowConfig, validateHostDecision, validateWorkflowPlan, WorkflowConfigError } from "./workflow-config.js";
 import { checkConvergence, decisionSignatures } from "./workflow-convergence.js";
 import type { FindingSignature } from "./workflow-convergence.js";
-import { createExecutorAdapter, probeExecutor } from "./workflow-executor.js";
+import { createExecutorAdapter, probeExecutor, validateClaudeSessionForWorkflow } from "./workflow-executor.js";
 import {
   WorkflowLockError,
   acquireLock,
@@ -78,18 +78,28 @@ process.on("SIGTERM", () => {
   }, 500).unref();
 });
 
-function createManagedController(timeoutSeconds?: number): { ac: AbortController; cleanup: () => void } {
+export function createManagedController(timeoutSeconds?: number): {
+  ac: AbortController;
+  markActivity: () => void;
+  cleanup: () => void;
+} {
   const ac = new AbortController();
   activeAbortControllers.add(ac);
   let timer: NodeJS.Timeout | undefined;
-  if (timeoutSeconds) {
+
+  const armTimer = () => {
+    if (!timeoutSeconds || ac.signal.aborted) return;
+    if (timer) clearTimeout(timer);
     timer = setTimeout(() => {
       ac.abort();
     }, timeoutSeconds * 1000);
     timer.unref();
-  }
+  };
+
+  armTimer();
   return {
     ac,
+    markActivity: armTimer,
     cleanup: () => {
       if (timer) clearTimeout(timer);
       activeAbortControllers.delete(ac);
@@ -509,7 +519,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
   appendEvent(dir, { kind: "executor_started", runId, timestamp: nowIso(), cycleIndex, data: { executor: config.executor } });
 
   const adapter = createExecutorAdapter(config.executor);
-  const { ac, cleanup } = createManagedController(state.timeoutSeconds);
+  const { ac, markActivity, cleanup } = createManagedController(state.timeoutSeconds);
 
   // Create attempt log file BEFORE spawning the executor
   const execLogPath = path.join(dir, `cycle-${cycleIndex}-exec.log`);
@@ -528,6 +538,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
       config: config.executorConfig,
       timeoutSeconds: state.timeoutSeconds,
       signal: ac.signal,
+      onActivity: markActivity,
       vcsKind,
     });
   } catch (err) {
@@ -941,7 +952,7 @@ async function resumeExecutorCycle(
   }
 
   const adapter = createExecutorAdapter(state.executor);
-  const { ac, cleanup } = createManagedController(state.timeoutSeconds);
+  const { ac, markActivity, cleanup } = createManagedController(state.timeoutSeconds);
 
   // Determine attempt index and create log file before spawning
   const priorLogs = cycleRecord.executorAttemptLogs
@@ -967,6 +978,7 @@ async function resumeExecutorCycle(
       config: state.executorConfig ?? {},
       timeoutSeconds: state.timeoutSeconds,
       signal: ac.signal,
+      onActivity: markActivity,
       vcsKind: state.vcsKind || "git",
     });
   } catch (err) {
@@ -1243,10 +1255,14 @@ function isProcessAlive(pid: number): boolean {
 
 export interface RetryExecuteWorkflowOpts {
   runId: string;
+  sessionId?: string;
 }
 
 export async function retryExecuteWorkflow(opts: RetryExecuteWorkflowOpts): Promise<void> {
   const { state, dir } = requireState(opts.runId);
+  if (opts.sessionId && state.executor !== "claude") {
+    throw new WorkflowEngineError("--session can only be used with a Claude executor run.", 4);
+  }
   const staleExecution = state.status === "executing"
     && state.enginePid !== undefined
     && !isProcessAlive(state.enginePid);
@@ -1266,7 +1282,7 @@ export async function retryExecuteWorkflow(opts: RetryExecuteWorkflowOpts): Prom
   if (!cycle || cycle.diffFile || cycle.hostReviewFile || cycle.decisionFile || cycle.scopeViolations) {
     throw new WorkflowEngineError("Cannot retry execution: the failed cycle already contains review artifacts.", 4);
   }
-  if (!state.sessionHandle) {
+  if (!state.sessionHandle && !opts.sessionId) {
     throw new WorkflowEngineError("Cannot retry execution: no executor session handle is available.", 4);
   }
   const recoveringInitialCycle = state.currentCycle === 1;
@@ -1311,6 +1327,14 @@ export async function retryExecuteWorkflow(opts: RetryExecuteWorkflowOpts): Prom
     );
   }
 
+  if (opts.sessionId) {
+    try {
+      validateClaudeSessionForWorkflow(opts.sessionId, state.repoRoot, state.plan.title);
+    } catch (err) {
+      throw new WorkflowEngineError(`Cannot rebind Claude session: ${(err as Error).message}`, 4);
+    }
+  }
+
   let acceptedFindings: string;
   if (recoveringInitialCycle) {
     acceptedFindings = "# Execution Recovery\n\nContinue the original implementation from the current working tree.";
@@ -1336,6 +1360,13 @@ export async function retryExecuteWorkflow(opts: RetryExecuteWorkflowOpts): Prom
     throw err;
   }
 
+  const previousSessionId = state.sessionHandle?.kind === "claude"
+    ? state.sessionHandle.sessionId
+    : undefined;
+  if (opts.sessionId) {
+    state.sessionHandle = { kind: "claude", sessionId: opts.sessionId };
+  }
+
   state.status = "executing";
   delete state.stopReason;
   delete state.stopDescription;
@@ -1343,6 +1374,16 @@ export async function retryExecuteWorkflow(opts: RetryExecuteWorkflowOpts): Prom
   state.updatedAt = nowIso();
   writeState(dir, state);
 
+  if (opts.sessionId) {
+    appendEvent(dir, {
+      kind: "executor_session_rebound",
+      runId: state.runId,
+      timestamp: state.updatedAt,
+      cycleIndex: state.currentCycle,
+      data: { previousSessionId, sessionId: opts.sessionId },
+    });
+  }
+
   // Determine recovery mode: stale_execution trumps initial_continuation
   let recoveryMode: string;
   if (staleExecution) {
@@ -1364,7 +1405,7 @@ export async function retryExecuteWorkflow(opts: RetryExecuteWorkflowOpts): Prom
     },
   });
 
-  process.stdout.write(`retrying executor cycle ${state.currentCycle} with the existing ${state.executor} session\n`);
+  process.stdout.write(`retrying executor cycle ${state.currentCycle} with the ${opts.sessionId ? "rebound" : "existing"} ${state.executor} session\n`);
   await resumeExecutorCycle(state, dir, cycle, acceptedFindings);
 }