diff --git a/README.md b/README.md index 8c68eed..9b654bc 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,14 @@ Retry executor resume after clearing an external session conflict. agent-workflow retry-execute --run ``` +If a failed Claude launch recorded a session ID that was never created, explicitly bind a known persisted Claude session from the same repository and frozen plan while retrying: + +```bash +agent-workflow retry-execute --run --session +``` + +The override is accepted only for Claude runs. The session JSONL must exist under `~/.claude/projects`, belong to the run repository, and contain the frozen plan title. Baseline, scope, and review-artifact gates still apply, and the old/new session IDs are appended to the workflow audit log. + ### `agent-workflow status` Display current workflow status. @@ -202,6 +210,8 @@ Create `agent-workflow.json` in your repository root: } ``` +`timeoutSeconds` is an inactivity timeout, not a total runtime limit. The host resets the timeout whenever the executor emits stdout/stderr progress, so an active task may run longer than the configured window. A continuously silent executor is terminated after the configured number of seconds. Agy receives a 24-hour internal print-mode ceiling while the host inactivity timeout remains the primary stuck-process guard. + - `defaultVcs` - Default VCS kind: `"auto"` (detect), `"git"`, `"svn"`, or `"none"`. Default is `"auto"`. ## Environment Variables @@ -213,8 +223,9 @@ The Agy adapter runs its non-interactive implementation sessions with `--dangerously-skip-permissions`. This is required because Agy cannot prompt for tool approval in `--print` mode. The workflow's frozen scope, Git gates, and executor safety prompt remain the authorization boundary. Its internal -`--print-timeout` is set from the workflow `timeoutSeconds` value so long tasks -do not fall back to Agy's five-minute default. +`--print-timeout` is set to a 24-hour internal ceiling so active tasks do not +fall back to Agy's five-minute default; the workflow inactivity timeout remains +the primary guard. ## State Machine diff --git a/skills/agent-workflow-host/SKILL.md b/skills/agent-workflow-host/SKILL.md index 61ab7ed..45579af 100644 --- a/skills/agent-workflow-host/SKILL.md +++ b/skills/agent-workflow-host/SKILL.md @@ -19,6 +19,7 @@ $agent-workflow-host 继续 $agent-workflow-host 查看 $agent-workflow-host 清理后重试 $agent-workflow-host 执行器重试 +$agent-workflow-host 执行器重试 使用会话 $agent-workflow-host 延长 添加 个周期 $agent-workflow-host 中止 原因：需求变化 $agent-workflow-host 日志 [--follow] @@ -103,6 +104,8 @@ Executor has been running for ~2m. Current activity: Editing src/parser.ts (PID Watch live: agent-workflow logs --run --follow ``` +`timeoutSeconds` is a consecutive inactivity timeout. Observable executor output refreshes the deadline, so do not treat total runtime beyond that value as a timeout while activity continues. + ### `awaiting_scope_resolution` 1. Confirm the Codex review bundle was not prepared and inspect every structured `scopeViolations` entry. @@ -152,6 +155,7 @@ After `fix`, repeat review and decision for the new cycle. Respect the configure - For `继续 `, read status first, then continue from `awaiting_review`, `awaiting_scope_resolution`, or `awaiting_host`. Read the persisted state and referenced cycle artifacts when earlier command output is no longer in context; do not reconstruct findings from memory. - For `清理后重试 `, run `agent-workflow retry-review --run ` after the approved scope artifacts have been removed. It must refuse when baseline drift or out-of-scope paths remain. - For `执行器重试 `, run `agent-workflow retry-execute --run ` after the external cause of an executor failure is cleared (e.g., session conflict resolved, transient error gone). It preserves failed attempt logs and reuses the same session handle. Initial-cycle retries continue from the original plan; later-cycle retries resume from the preceding decision. Refuses when baseline has drifted, scope violations exist, or the cycle already produced review artifacts. +- For `执行器重试 使用会话 `, run `agent-workflow retry-execute --run --session ` only when a Claude run's recorded session was never created or is unavailable and a known persisted replacement session contains the same frozen plan. The CLI must validate that the session belongs to the same repository and plan before recording the rebind in the audit log and retrying. Never edit `state.json` manually or guess a session ID. - For `延长 添加 个周期`, run `agent-workflow extend --run --additional-cycles ` to extend a `budget_exhausted` run. This requires: the run ended at `budget_exhausted`, has a valid session handle, the last decision was `fix`, baseline hasn't drifted, and scope is clean. Returns to `awaiting_host` state to continue the same executor session. - Treat `completed`, `needs_human`, `blocked`, and `aborted` as terminal. Report them instead of attempting another cycle. - `budget_exhausted` is terminal but can be extended with the extend command if all conditions are met. @@ -159,7 +163,7 @@ After `fix`, repeat review and decision for the new cycle. Respect the configure ## Failure Handling -- Missing session handle, executor failure, or baseline drift (Git HEAD change, SVN revision change) must stop as `blocked`; do not work around the gate. +- Missing session handle, executor failure, or baseline drift (Git HEAD change, SVN revision change) must stop as `blocked`; do not work around the gate. A missing Claude session may use the explicit validated session-rebind retry operation when a matching persisted session is known. - When a resume failed before producing cycle artifacts because the same session was busy, ask the user to close the owning executor process, then use `retry-execute`; never kill an unrelated process automatically. - Out-of-scope edits must enter `awaiting_scope_resolution`; inspect and remediate them through the same executor session. Repeated identical violations must stop as `needs_human`. - Same finding in consecutive reviews or disappear/reappear oscillation must stop as `needs_human` when the engine reports non-convergence. diff --git a/src/cli.ts b/src/cli.ts index c815e72..7bb3d0e 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -57,7 +57,7 @@ async function main() { break; case "retry-execute": - await retryExecuteWorkflow({ runId: args.runId }); + await retryExecuteWorkflow({ runId: args.runId, sessionId: args.sessionId }); break; case "decide": diff --git a/src/index.ts b/src/index.ts index 1be3d1a..9ea0f2f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -91,6 +91,7 @@ export type { export { createExecutorAdapter, probeExecutor, + validateClaudeSessionForWorkflow, } from "./workflow-executor.js"; export type { diff --git a/src/vcs-provider-git.test.ts b/src/vcs-provider-git.test.ts new file mode 100644 index 0000000..a7165a1 --- /dev/null +++ b/src/vcs-provider-git.test.ts @@ -0,0 +1,39 @@ +import { execFileSync } from "node:child_process"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { GitProvider } from "./vcs-provider.js"; + +describe("GitProvider raw path handling", () => { + let repoRoot: string; + const provider = new GitProvider(); + + beforeEach(() => { + repoRoot = fs.mkdtempSync(path.join(os.tmpdir(), "agent-workflow-git-paths-")); + execFileSync("git", ["init"], { cwd: repoRoot, stdio: "ignore" }); + execFileSync("git", ["config", "user.email", "test@example.com"], { cwd: repoRoot }); + execFileSync("git", ["config", "user.name", "Test User"], { cwd: repoRoot }); + fs.mkdirSync(path.join(repoRoot, "中文目录"), { recursive: true }); + fs.writeFileSync(path.join(repoRoot, "中文目录", "已跟踪.txt"), "before\n"); + execFileSync("git", ["add", "."], { cwd: repoRoot }); + execFileSync("git", ["commit", "-m", "initial"], { cwd: repoRoot, stdio: "ignore" }); + }); + + afterEach(() => { + fs.rmSync(repoRoot, { recursive: true, force: true }); + }); + + it("matches tracked and untracked Unicode paths against the frozen scope", () => { + const baseline = provider.captureBaseline(repoRoot); + fs.writeFileSync(path.join(repoRoot, "中文目录", "已跟踪.txt"), "after\n"); + fs.writeFileSync(path.join(repoRoot, "中文目录", "未跟踪 文件.txt"), "new\n"); + + expect(provider.checkFilesInScope(repoRoot, baseline, ["中文目录"])).toEqual([]); + expect(provider.checkFilesInScope(repoRoot, baseline, ["other"])).toEqual([ + "中文目录/已跟踪.txt", + "中文目录/未跟踪 文件.txt", + ]); + expect(provider.diffFromBaseline(repoRoot, baseline)).toContain("+new"); + }); +}); diff --git a/src/vcs-provider.ts b/src/vcs-provider.ts index 966c6cc..3702f54 100644 --- a/src/vcs-provider.ts +++ b/src/vcs-provider.ts @@ -22,6 +22,10 @@ export class VcsError extends Error { } } +function parseNullDelimitedPaths(output: string): string[] { + return output.split("\0").filter((entry) => entry.length > 0); +} + // --------------------------------------------------------------------------- // VCS Provider Interface // --------------------------------------------------------------------------- @@ -104,15 +108,12 @@ export class GitProvider implements VcsProvider { stdio: ["ignore", "pipe", "pipe"], }).trim(); - const untrackedStr = execFileSync("git", ["ls-files", "--others", "--exclude-standard"], { + const untrackedStr = execFileSync("git", ["ls-files", "-z", "--others", "--exclude-standard"], { cwd: repoRoot, encoding: "utf8", stdio: ["ignore", "pipe", "pipe"], }); - const untrackedFiles = untrackedStr - .split("\n") - .map((line) => line.trim()) - .filter((line) => line.length > 0); + const untrackedFiles = parseNullDelimitedPaths(untrackedStr); let finalDiff = trackedDiff; for (const file of untrackedFiles) { @@ -142,12 +143,12 @@ export class GitProvider implements VcsProvider { let diff: string; let untracked: string; try { - diff = execFileSync("git", ["diff", "--name-only", baseline.head], { + diff = execFileSync("git", ["diff", "--name-only", "-z", baseline.head], { cwd: repoRoot, encoding: "utf8", stdio: ["ignore", "pipe", "pipe"], }); - untracked = execFileSync("git", ["ls-files", "--others", "--exclude-standard"], { + untracked = execFileSync("git", ["ls-files", "-z", "--others", "--exclude-standard"], { cwd: repoRoot, encoding: "utf8", stdio: ["ignore", "pipe", "pipe"], @@ -156,10 +157,10 @@ export class GitProvider implements VcsProvider { throw new VcsError("Failed to check modified files."); } - const modifiedFiles = (diff + "\n" + untracked) - .split("\n") - .map((line) => line.trim()) - .filter((line) => line.length > 0); + const modifiedFiles = [ + ...parseNullDelimitedPaths(diff), + ...parseNullDelimitedPaths(untracked), + ]; const outOfScope: string[] = []; for (const file of modifiedFiles) { const inScope = scope.some((s) => { diff --git a/src/workflow-args.test.ts b/src/workflow-args.test.ts new file mode 100644 index 0000000..a477bcb --- /dev/null +++ b/src/workflow-args.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "vitest"; +import { parseWorkflowArgs } from "./workflow-args.js"; + +describe("workflow retry-execute arguments", () => { + it("accepts an explicit Claude session ID", () => { + expect(parseWorkflowArgs([ + "retry-execute", + "--run", "run-123", + "--session", "22778b47-e4f3-4645-84cd-1c6a74a912d8", + ])).toEqual({ + sub: "retry-execute", + runId: "run-123", + sessionId: "22778b47-e4f3-4645-84cd-1c6a74a912d8", + }); + }); + + it("preserves retry without a session override", () => { + expect(parseWorkflowArgs(["retry-execute", "--run", "run-123"])).toEqual({ + sub: "retry-execute", + runId: "run-123", + sessionId: undefined, + }); + }); +}); diff --git a/src/workflow-args.ts b/src/workflow-args.ts index 9f3f93a..0385bad 100644 --- a/src/workflow-args.ts +++ b/src/workflow-args.ts @@ -25,6 +25,7 @@ export interface WorkflowRetryReviewArgs { export interface WorkflowRetryExecuteArgs { sub: "retry-execute"; runId: string; + sessionId?: string; } export interface WorkflowDecideArgs { @@ -96,7 +97,7 @@ Usage: agent-workflow start --executor --plan [--vcs ] [--max-cycles ] agent-workflow review --run agent-workflow retry-review --run - agent-workflow retry-execute --run + agent-workflow retry-execute --run [--session ] agent-workflow decide --run --input agent-workflow extend --run --additional-cycles agent-workflow status --run [--json] @@ -113,6 +114,7 @@ Examples: agent-workflow review --run 2026-01-01T00-00-00_01_abc123 agent-workflow retry-review --run 2026-01-01T00-00-00_01_abc123 agent-workflow retry-execute --run 2026-01-01T00-00-00_01_abc123 + agent-workflow retry-execute --run 2026-01-01T00-00-00_01_abc123 --session 12345678-1234-1234-1234-123456789abc agent-workflow decide --run --input decision.json agent-workflow decide --run --input - # read decision from stdin agent-workflow extend --run --additional-cycles 3 @@ -204,10 +206,13 @@ export function parseWorkflowArgs(argv: string[]): WorkflowArgs { case "retry-execute": { let runId: string | undefined; + let sessionId: string | undefined; while (rest.length > 0) { const arg = rest.shift()!; if (arg === "--run") { runId = requireValue(arg, rest); + } else if (arg === "--session") { + sessionId = requireValue(arg, rest); } else if (arg === "-h" || arg === "--help") { workflowUsage(0); } else { @@ -215,7 +220,7 @@ export function parseWorkflowArgs(argv: string[]): WorkflowArgs { } } if (!runId) throw new WorkflowArgsError("--run is required for workflow retry-execute"); - return { sub: "retry-execute", runId }; + return { sub: "retry-execute", runId, sessionId }; } case "decide": { diff --git a/src/workflow-engine.ts b/src/workflow-engine.ts index 1c554fa..81a085c 100644 --- a/src/workflow-engine.ts +++ b/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 { 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); } diff --git a/src/workflow-executor.test.ts b/src/workflow-executor.test.ts index 641deb8..bb36ed9 100644 --- a/src/workflow-executor.test.ts +++ b/src/workflow-executor.test.ts @@ -162,12 +162,14 @@ describe("ClaudeAdapter", () => { }); const adapter = createExecutorAdapter("claude"); + let activityCount = 0; const result = await adapter.start({ repoRoot: tmpDir, plan: makeValidPlan(), runDir: tmpDir, cycleIndex: 1, config: { binary: fakeBin }, + onActivity: () => { activityCount += 1; }, }); const capturedArgs = fs.existsSync(argsFile) @@ -181,6 +183,7 @@ describe("ClaudeAdapter", () => { const sessionIdIdx = capturedArgs.indexOf("--session-id"); expect(sessionIdIdx).toBeGreaterThanOrEqual(0); expect(capturedArgs[sessionIdIdx + 1]).toMatch(/^[0-9a-f-]{36}$/i); + expect(activityCount).toBeGreaterThan(0); expect(result.sessionHandle?.kind).toBe("claude"); }); @@ -490,7 +493,7 @@ describe("AgyAdapter", () => { expect(capturedArgs).toContain("--mode"); expect(capturedArgs[capturedArgs.indexOf("--mode") + 1]).toBe("accept-edits"); expect(capturedArgs).toContain("--print-timeout"); - expect(capturedArgs[capturedArgs.indexOf("--print-timeout") + 1]).toBe("1800s"); + expect(capturedArgs[capturedArgs.indexOf("--print-timeout") + 1]).toBe("24h"); expect(capturedArgs).not.toContain("--"); expect(printIndex).toBe(capturedArgs.length - 2); expect(capturedArgs[printIndex + 1]).toContain("# Implementation Task: Test plan"); @@ -539,7 +542,7 @@ describe("AgyAdapter", () => { expect(capturedArgs).toContain("--conversation"); expect(capturedArgs).toContain("known-conv-id"); expect(capturedArgs).toContain("--dangerously-skip-permissions"); - expect(capturedArgs[capturedArgs.indexOf("--print-timeout") + 1]).toBe("900s"); + expect(capturedArgs[capturedArgs.indexOf("--print-timeout") + 1]).toBe("24h"); expect(capturedArgs[capturedArgs.indexOf("--print") + 1]).toContain("# Fix Request: Test plan"); }); diff --git a/src/workflow-executor.ts b/src/workflow-executor.ts index 4c3ad09..1fdca85 100644 --- a/src/workflow-executor.ts +++ b/src/workflow-executor.ts @@ -490,6 +490,7 @@ class ReasonixAdapter implements ExecutorAdapter { processGroup: true, logFilePath: logFile, onProgress: ({ pid, bytesLogged }) => { + opts.onActivity?.(); updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); }, }); @@ -541,6 +542,7 @@ class ReasonixAdapter implements ExecutorAdapter { processGroup: true, logFilePath: logFile, onProgress: ({ pid, bytesLogged }) => { + opts.onActivity?.(); updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); }, }); @@ -585,6 +587,68 @@ function extractClaudeSessionId(output: string): string | undefined { return m?.[1]; } +export function validateClaudeSessionForWorkflow( + sessionId: string, + repoRoot: string, + planTitle: string, +): string { + if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(sessionId)) { + throw new Error(`Invalid Claude session ID: ${sessionId}`); + } + + const projectsDir = path.join(process.env.HOME || os.homedir(), ".claude", "projects"); + if (!fs.existsSync(projectsDir)) { + throw new Error(`Claude projects directory not found: ${projectsDir}`); + } + + const candidates = fs.readdirSync(projectsDir, { withFileTypes: true }) + .filter((entry) => entry.isDirectory()) + .map((entry) => path.join(projectsDir, entry.name, `${sessionId}.jsonl`)) + .filter((candidate) => fs.existsSync(candidate)); + + if (candidates.length !== 1) { + throw new Error( + candidates.length === 0 + ? `Claude session file not found for ID: ${sessionId}` + : `Multiple Claude session files found for ID: ${sessionId}`, + ); + } + + const sessionFile = candidates[0]; + const fd = fs.openSync(sessionFile, "r"); + const maxBytes = 4 * 1024 * 1024; + const buffer = Buffer.alloc(maxBytes); + let content: string; + try { + const bytesRead = fs.readSync(fd, buffer, 0, maxBytes, 0); + content = buffer.toString("utf8", 0, bytesRead); + } finally { + fs.closeSync(fd); + } + + let repoMatches = false; + let sessionMatches = false; + for (const line of content.split("\n")) { + if (!line.trim()) continue; + try { + const entry = JSON.parse(line) as Record; + if (entry.sessionId === sessionId) sessionMatches = true; + if (entry.cwd === repoRoot) repoMatches = true; + } catch { + // Ignore a trailing partial line at the read boundary. + } + } + + if (!sessionMatches || !repoMatches) { + throw new Error(`Claude session ${sessionId} does not belong to repository ${repoRoot}`); + } + if (!content.includes(planTitle)) { + throw new Error(`Claude session ${sessionId} does not contain frozen plan title: ${planTitle}`); + } + + return sessionFile; +} + function extractClaudeFailureReason(output: string): string | undefined { for (const line of output.split("\n")) { const trimmed = line.trim(); @@ -654,6 +718,7 @@ class ClaudeAdapter implements ExecutorAdapter { processGroup: true, logFilePath: logFile, onProgress: ({ pid, bytesLogged }) => { + opts.onActivity?.(); updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); }, onStdoutData: (chunk: string) => { @@ -727,6 +792,7 @@ class ClaudeAdapter implements ExecutorAdapter { processGroup: true, logFilePath: logFile, onProgress: ({ pid, bytesLogged }) => { + opts.onActivity?.(); updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); }, onStdoutData: (chunk: string) => { @@ -837,7 +903,7 @@ function agyCommonArgs(config: ExecutorConfig, timeoutSeconds?: number): string[ return [ "--dangerously-skip-permissions", "--mode", "accept-edits", - ...(timeoutSeconds ? ["--print-timeout", `${timeoutSeconds}s`] : []), + ...(timeoutSeconds ? ["--print-timeout", "24h"] : []), ...(config.agent ? ["--agent", config.agent] : []), ...(config.model ? ["--model", config.model] : []), ]; @@ -882,6 +948,7 @@ class AgyAdapter implements ExecutorAdapter { processGroup: true, logFilePath: logFile, onProgress: ({ pid, bytesLogged }) => { + opts.onActivity?.(); updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); }, }); @@ -952,6 +1019,7 @@ class AgyAdapter implements ExecutorAdapter { processGroup: true, logFilePath: logFile, onProgress: ({ pid, bytesLogged }) => { + opts.onActivity?.(); updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); }, }); diff --git a/src/workflow-state.test.ts b/src/workflow-state.test.ts index 1e262af..49d80ba 100644 --- a/src/workflow-state.test.ts +++ b/src/workflow-state.test.ts @@ -102,6 +102,7 @@ describe("Retry-execute recovery", () => { let repoRoot: string; let stateRoot: string; let originalAgentDir: string | undefined; + let originalHome: string | undefined; beforeEach(() => { tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "agent-workflow-retry-test-")); @@ -116,11 +117,15 @@ describe("Retry-execute recovery", () => { execFileSync("git", ["commit", "-m", "init"], { cwd: repoRoot }); originalAgentDir = process.env.AGENT_WORKFLOW_DIR; process.env.AGENT_WORKFLOW_DIR = stateRoot; + originalHome = process.env.HOME; + process.env.HOME = tmpDir; }); afterEach(() => { if (originalAgentDir === undefined) delete process.env.AGENT_WORKFLOW_DIR; else process.env.AGENT_WORKFLOW_DIR = originalAgentDir; + if (originalHome === undefined) delete process.env.HOME; + else process.env.HOME = originalHome; fs.rmSync(tmpDir, { recursive: true, force: true }); }); @@ -261,6 +266,95 @@ echo '{"session_id":"fake-session-123","conversation_id":"fake-conv-456"}' expect(events.find((event) => event.kind === "executor_retried")?.data.recoveryMode) .toBe("initial_continuation"); }); + + it("rebinds a validated Claude session before retrying the failed cycle", async () => { + const { retryExecuteWorkflow } = await import("./workflow-engine.js"); + const { gitHead, initWorkflowState, readState, writeState } = await import("./workflow-state.js"); + const runId = "claude-rebind-123"; + const sessionId = "22778b47-e4f3-4645-84cd-1c6a74a912d8"; + const dir = runDir(repoRoot, runId); + const claudeProject = path.join(tmpDir, ".claude", "projects", "test-project"); + fs.mkdirSync(claudeProject, { recursive: true }); + fs.writeFileSync( + path.join(claudeProject, `${sessionId}.jsonl`), + `${JSON.stringify({ sessionId, cwd: repoRoot, message: { content: "# Implementation Task: Test" } })}\n`, + ); + + const fakeClaude = path.join(tmpDir, "fake-claude"); + fs.writeFileSync(fakeClaude, `#!/bin/bash +printf '{"type":"system","subtype":"init","session_id":"${sessionId}"}\\n' +printf '{"type":"result","result":"done","session_id":"${sessionId}"}\\n' +`, { mode: 0o755 }); + + const state = initWorkflowState({ + runId, + repoRoot, + baselineHead: gitHead(repoRoot), + plan: plan(), + executor: "claude", + maxCycles: 3, + }); + state.status = "blocked"; + state.stopDescription = "Executor failed to resume: missing session"; + state.currentCycle = 1; + state.cycles = [{ + cycleIndex: 1, + startedAt: new Date().toISOString(), + executorLogFile: "cycle-1-exec.log", + executorAttemptLogs: ["cycle-1-exec.log"], + }]; + state.sessionHandle = { kind: "claude", sessionId: "97a606e8-cfa3-4996-83a8-ddb7ece4234b" }; + state.executorConfig = { binary: fakeClaude }; + fs.mkdirSync(dir, { recursive: true }); + fs.writeFileSync(path.join(dir, "cycle-1-exec.log"), "initial attempt failed"); + writeState(dir, state); + + await retryExecuteWorkflow({ runId, sessionId }); + + const recovered = readState(dir)!; + expect(recovered.status).toBe("awaiting_review"); + expect(recovered.sessionHandle).toEqual({ kind: "claude", sessionId }); + const events = fs.readFileSync(path.join(dir, "events.jsonl"), "utf8") + .trim().split("\n").map((line) => JSON.parse(line)); + expect(events.find((event) => event.kind === "executor_session_rebound")?.data).toEqual({ + previousSessionId: "97a606e8-cfa3-4996-83a8-ddb7ece4234b", + sessionId, + }); + }); + + it("rejects a Claude session from another repository without changing state", async () => { + const { retryExecuteWorkflow } = await import("./workflow-engine.js"); + const { gitHead, initWorkflowState, readState, writeState } = await import("./workflow-state.js"); + const runId = "claude-rebind-reject"; + const sessionId = "22778b47-e4f3-4645-84cd-1c6a74a912d8"; + const dir = runDir(repoRoot, runId); + const claudeProject = path.join(tmpDir, ".claude", "projects", "other-project"); + fs.mkdirSync(claudeProject, { recursive: true }); + fs.writeFileSync( + path.join(claudeProject, `${sessionId}.jsonl`), + `${JSON.stringify({ sessionId, cwd: path.join(tmpDir, "other"), message: { content: "Test" } })}\n`, + ); + + const state = initWorkflowState({ + runId, + repoRoot, + baselineHead: gitHead(repoRoot), + plan: plan(), + executor: "claude", + maxCycles: 3, + }); + state.status = "blocked"; + state.stopDescription = "Executor failed to resume: missing session"; + state.currentCycle = 1; + state.cycles = [{ cycleIndex: 1, startedAt: new Date().toISOString() }]; + state.sessionHandle = { kind: "claude", sessionId: "97a606e8-cfa3-4996-83a8-ddb7ece4234b" }; + writeState(dir, state); + + await expect(retryExecuteWorkflow({ runId, sessionId })) + .rejects.toThrow(/does not belong to repository/); + expect(readState(dir)?.sessionHandle).toEqual(state.sessionHandle); + expect(fs.existsSync(path.join(dir, "events.jsonl"))).toBe(false); + }); }); // --------------------------------------------------------------------------- diff --git a/src/workflow-timeout.test.ts b/src/workflow-timeout.test.ts new file mode 100644 index 0000000..ef8b581 --- /dev/null +++ b/src/workflow-timeout.test.ts @@ -0,0 +1,30 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createManagedController } from "./workflow-engine.js"; + +describe("executor inactivity timeout", () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it("extends the deadline whenever executor activity is observed", () => { + vi.useFakeTimers(); + const controller = createManagedController(10); + + vi.advanceTimersByTime(9_000); + controller.markActivity(); + vi.advanceTimersByTime(9_000); + expect(controller.ac.signal.aborted).toBe(false); + + vi.advanceTimersByTime(1_001); + expect(controller.ac.signal.aborted).toBe(true); + controller.cleanup(); + }); + + it("does not abort after cleanup", () => { + vi.useFakeTimers(); + const controller = createManagedController(10); + controller.cleanup(); + vi.advanceTimersByTime(20_000); + expect(controller.ac.signal.aborted).toBe(false); + }); +}); diff --git a/src/workflow-types.ts b/src/workflow-types.ts index 5ded763..278eeca 100644 --- a/src/workflow-types.ts +++ b/src/workflow-types.ts @@ -206,6 +206,7 @@ export interface WorkflowState { vcsKind?: VcsKind; executor: ExecutorKind; executorConfig?: ExecutorConfig; + /** Consecutive executor inactivity allowed before the host aborts the attempt. */ timeoutSeconds?: number; enginePid?: number; plan: WorkflowPlanV1; @@ -373,6 +374,7 @@ export type WorkflowEventKind = | "cycle_started" | "executor_started" | "executor_retried" + | "executor_session_rebound" | "executor_completed" | "review_started" | "review_completed" @@ -431,9 +433,10 @@ export interface ExecutorStartOpts { /** Path to the attempt log file (created before spawn, adapter streams output to it). */ logFilePath?: string; config: ExecutorConfig; - /** Host timeout, forwarded to executors that enforce their own deadline. */ + /** Host inactivity timeout. */ timeoutSeconds?: number; signal?: AbortSignal; + onActivity?: () => void; vcsKind?: VcsKind; } @@ -448,8 +451,9 @@ export interface ExecutorResumeOpts { /** Path to the attempt log file (created before spawn, adapter streams output to it). */ logFilePath?: string; config: ExecutorConfig; - /** Host timeout, forwarded to executors that enforce their own deadline. */ + /** Host inactivity timeout. */ timeoutSeconds?: number; signal?: AbortSignal; + onActivity?: () => void; vcsKind?: VcsKind; }