feat: reuse executor sessions within Codex tasks

2026-07-19 21:51:31 +08:00
parent 717d4344ee
commit 71b8025666
13 changed files with 702 additions and 13 deletions
@@ -27,8 +27,16 @@ import {
  createExecutorAdapter,
  findReasonixSessionForWorkflow,
  probeExecutor,
+  validateClaudeSessionForRepository,
  validateClaudeSessionForWorkflow,
+  validateReasonixSessionForRepository,
 } from "./workflow-executor.js";
+import {
+  getOrCreateTaskBinding,
+  readTaskBinding,
+  saveTaskExecutorBinding,
+  WorkflowTaskBindingError,
+} from "./workflow-task-binding.js";
 import {
  WorkflowLockError,
  acquireLock,
@@ -56,7 +64,9 @@ import {
 } from "./vcs-provider.js";
 import type {
  CycleRecord,
+  ExecutorKind,
  ExecutorResult,
+  ExecutorSessionHandle,
  HostDecisionV1,
  HostReviewBundleV1,
  ScopeViolation,
@@ -259,8 +269,51 @@ export interface StartWorkflowOpts {
  executor?: string;
  vcs?: string;
  maxCycles?: number;
+  /** Start a fresh task binding for this Codex thread and repository. */
+  newTask?: boolean;
+  /** Overrides CODEX_THREAD_ID for programmatic callers and tests. */
+  codexThreadId?: string;
  cwd?: string;
 }
+
+function validateTaskSessionHandle(
+  executor: ExecutorKind,
+  handle: ExecutorSessionHandle,
+  repoRoot: string,
+): void {
+  if (handle.kind !== executor) {
+    throw new WorkflowTaskBindingError(`Task binding contains a ${handle.kind} handle for executor ${executor}.`);
+  }
+  switch (handle.kind) {
+    case "reasonix":
+      validateReasonixSessionForRepository(handle.sessionFilePath, repoRoot);
+      return;
+    case "claude":
+      validateClaudeSessionForRepository(handle.sessionId, repoRoot);
+      return;
+    case "agy":
+      if (handle.degraded || !handle.conversationId || !/^[0-9a-f-]{36}$/i.test(handle.conversationId)) {
+        throw new WorkflowTaskBindingError("Task binding contains an Agy session without a reliable conversation ID.");
+      }
+  }
+}
+
+function persistTaskSession(state: WorkflowState, handle: ExecutorSessionHandle): void {
+  if (!state.taskId || !state.codexThreadId) return;
+  try {
+    saveTaskExecutorBinding({
+      taskId: state.taskId,
+      codexThreadId: state.codexThreadId,
+      repoRoot: state.repoRoot,
+      executor: state.executor,
+      handle,
+      runId: state.runId,
+      planTitle: state.plan.title,
+    });
+  } catch (err) {
+    process.stderr.write(`Warning: failed to update task executor binding: ${(err as Error).message}\n`);
+  }
+}
 /** Resolve VCS selection from candidates and configs - exported for testing */
 export async function resolveVcsSelection(
  cwd: string,
@@ -425,6 +478,26 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
  });

  const vcsProvider = createVcsProvider(vcsKind);
+  const codexThreadId = (opts.codexThreadId ?? process.env.CODEX_THREAD_ID)?.trim();
+  if (opts.newTask && !codexThreadId) {
+    throw new WorkflowEngineError("--new-task requires CODEX_THREAD_ID to identify the owning Codex session.", 4);
+  }
+
+  let existingTaskBinding;
+  let taskSessionHandle: ExecutorSessionHandle | undefined;
+  if (codexThreadId && !opts.newTask) {
+    try {
+      existingTaskBinding = readTaskBinding(repoRoot, codexThreadId);
+      taskSessionHandle = existingTaskBinding?.executors[config.executor]?.handle;
+      if (taskSessionHandle) validateTaskSessionHandle(config.executor, taskSessionHandle, repoRoot);
+    } catch (err) {
+      throw new WorkflowEngineError(
+        `Cannot reuse the current task's ${config.executor} session: ${(err as Error).message} ` +
+        "Start with --new-task only when this is intentionally a new task.",
+        4,
+      );
+    }
+  }

  // 4. Probe executor
  await probeExecutor(config.executor, config.executorConfig);
@@ -490,18 +563,46 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
  state.timeoutSeconds = config.timeoutSeconds;
  state.enginePid = process.pid;
  state.status = "executing";
+  if (codexThreadId) {
+    try {
+      const binding = opts.newTask || !existingTaskBinding
+        ? getOrCreateTaskBinding({
+          repoRoot,
+          codexThreadId,
+          planTitle: typedPlan.title,
+          newTask: opts.newTask,
+        })
+        : existingTaskBinding;
+      state.taskId = binding.taskId;
+      state.codexThreadId = codexThreadId;
+      state.taskSessionReused = !!taskSessionHandle;
+      if (taskSessionHandle) state.sessionHandle = taskSessionHandle;
+    } catch (err) {
+      releaseLock(repoRoot, runId);
+      throw new WorkflowEngineError(`Cannot initialize Codex task binding: ${(err as Error).message}`, 4);
+    }
+  }
  writeState(dir, state);
  appendEvent(dir, {
    kind: "workflow_started",
    runId,
    timestamp: state.createdAt,
-    data: { executor: config.executor, maxCycles: config.maxCycles, vcsKind, vcsBaseline },
+    data: {
+      executor: config.executor,
+      maxCycles: config.maxCycles,
+      vcsKind,
+      vcsBaseline,
+      ...(state.taskId ? { taskId: state.taskId, taskSessionReused: state.taskSessionReused } : {}),
+    },
  });

  process.stdout.write(`workflow started: ${runId}\n`);
  process.stdout.write(`  vcs:         ${vcsKind}\n`);
  process.stdout.write(`  executor:    ${config.executor}\n`);
  process.stdout.write(`  max-cycles:  ${config.maxCycles}\n`);
+  if (state.taskId) {
+    process.stdout.write(`  task:        ${state.taskId}${state.taskSessionReused ? " (session reused)" : ""}\n`);
+  }
  if (vcsBaseline.kind === "git") {
    process.stdout.write(`  baseline:    ${vcsBaseline.head}\n`);
  } else if (vcsBaseline.kind === "svn") {
@@ -521,7 +622,22 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
  writeState(dir, state);

  appendEvent(dir, { kind: "cycle_started", runId, timestamp: nowIso(), cycleIndex });
-  appendEvent(dir, { kind: "executor_started", runId, timestamp: nowIso(), cycleIndex, data: { executor: config.executor } });
+  appendEvent(dir, {
+    kind: "executor_started",
+    runId,
+    timestamp: nowIso(),
+    cycleIndex,
+    data: { executor: config.executor, sessionMode: taskSessionHandle ? "task_reuse" : "new" },
+  });
+  if (taskSessionHandle) {
+    appendEvent(dir, {
+      kind: "executor_session_reused",
+      runId,
+      timestamp: nowIso(),
+      cycleIndex,
+      data: { executor: config.executor, taskId: state.taskId },
+    });
+  }

  const adapter = createExecutorAdapter(config.executor);
  const { ac, markActivity, cleanup } = createManagedController(state.timeoutSeconds);
@@ -534,7 +650,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s

  let execResult;
  try {
-    execResult = await adapter.start({
+    const commonOpts = {
      repoRoot,
      plan: typedPlan,
      runDir: dir,
@@ -545,7 +661,15 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
      signal: ac.signal,
      onActivity: markActivity,
      vcsKind,
-    });
+    };
+    execResult = taskSessionHandle
+      ? await adapter.resume({
+        ...commonOpts,
+        handle: taskSessionHandle,
+        acceptedFindings: "",
+        purpose: "task_plan",
+      })
+      : await adapter.start(commonOpts);
  } catch (err) {
    cleanup();
    transitionToTerminal(state, dir, "blocked", `Executor failed to start: ${(err as Error).message}`);
@@ -556,6 +680,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
  persistExecutorAttempt(dir, cycleRecord, execResult, execLogPath);
  if (execResult.sessionHandle) {
    state.sessionHandle = execResult.sessionHandle;
+    persistTaskSession(state, execResult.sessionHandle);
    if (execResult.sessionHandle.kind === "agy") {
      state.agyDegraded = execResult.sessionHandle.degraded;
    }
@@ -1002,6 +1127,7 @@ async function resumeExecutorCycle(
  persistExecutorAttempt(dir, cycleRecord, execResult, execLogPath);
  if (execResult.sessionHandle) {
    state.sessionHandle = execResult.sessionHandle;
+    persistTaskSession(state, execResult.sessionHandle);
    if (execResult.sessionHandle.kind === "agy") {
      state.agyDegraded = execResult.sessionHandle.degraded;
    }
@@ -1649,6 +1775,8 @@ export function statusWorkflow(opts: StatusWorkflowOpts): void {
    maxCycles: state.maxCycles,
    remainingCycles: Math.max(0, state.maxCycles - state.currentCycle),
    sessionHandleKind: state.sessionHandle?.kind,
+    taskId: state.taskId,
+    taskSessionReused: state.taskSessionReused,
    degradedResume: state.agyDegraded,
    runDir: dir,
    diffFile: cycle?.diffFile ? path.join(dir, cycle.diffFile) : undefined,
@@ -1693,6 +1821,9 @@ export function statusWorkflow(opts: StatusWorkflowOpts): void {
    const extra = kind === "agy" && state.agyDegraded ? " [degraded-continue]" : "";
    lines.push(p("Session handle", kind + extra));
  }
+  if (state.taskId) {
+    lines.push(p("Task", `${state.taskId}${state.taskSessionReused ? " [session reused]" : ""}`));
+  }

  // Live execution fields
  if (output.executorPid) lines.push(p("Executor PID", String(output.executorPid)));