From c9d76d8f5882ac93db1afbd59b228244488b03db Mon Sep 17 00:00:00 2001 From: liujing Date: Fri, 17 Jul 2026 00:48:52 +0800 Subject: [PATCH] feat: add live executor observability --- README.md | 32 +++ skills/agent-workflow-host/SKILL.md | 24 +- src/child-process.test.ts | 133 ++++++++++ src/child-process.ts | 82 +++++- src/cli.ts | 9 + src/index.ts | 10 + src/workflow-args.ts | 35 ++- src/workflow-engine.ts | 254 +++++++++++++++++- src/workflow-executor.test.ts | 194 +++++++++++++- src/workflow-executor.ts | 314 +++++++++++++++++++++- src/workflow-state.test.ts | 396 ++++++++++++++++++++++++++++ src/workflow-state.ts | 44 +++- src/workflow-types.ts | 47 ++++ 13 files changed, 1546 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index dbd1c8c..80e2e1d 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,38 @@ Display current workflow status. agent-workflow status --run [--json] ``` +When the workflow is executing, additional live fields are shown: +- **Executor PID** — process ID of the running executor +- **Elapsed** — running time +- **Activity** — current operation being performed +- **Last activity** — timestamp of most recent output +- **Log bytes** — bytes written to the attempt log so far +- **Live log** — path to the streaming log file + +### `agent-workflow logs` + +Tail and follow live executor output for a running workflow. + +```bash +agent-workflow logs --run [--follow] [--tail ] +``` + +- **`--follow` / `-f`** — continuously follow new output. Automatically switches to new retry attempt logs when a retry begins. +- **`--tail `** — number of lines to show from the end of the current log (default: 50). + +Examples: + +```bash +# Show the last 50 lines of the current attempt log +agent-workflow logs --run + +# Follow new output (like tail -f) +agent-workflow logs --run --follow + +# Show the last 200 lines +agent-workflow logs --run --tail 200 +``` + ### `agent-workflow abort` Terminate an active workflow. diff --git a/skills/agent-workflow-host/SKILL.md b/skills/agent-workflow-host/SKILL.md index 64a7db7..c42c1c7 100644 --- a/skills/agent-workflow-host/SKILL.md +++ b/skills/agent-workflow-host/SKILL.md @@ -18,7 +18,9 @@ $agent-workflow-host 继续 $agent-workflow-host 查看 $agent-workflow-host 清理后重试 $agent-workflow-host 执行器重试 +$agent-workflow-host 执行器重试 $agent-workflow-host 中止 原因：需求变化 +$agent-workflow-host 日志 [--follow] ``` - Treat an explicit `执行` invocation as authorization to execute the latest clearly bounded plan in the conversation. @@ -59,12 +61,17 @@ $agent-workflow-host 中止 原因：需求变化 5. Require non-empty repository-relative scope entries and at least one verification command. Use argument arrays; never encode shell pipelines, redirections, glob expansion, `&&`, or `;` as one command. 6. Store input JSON in a `mktemp -d` directory outside the repository. The workflow's durable state is the audit record; remove temporary input files after a terminal result. -7. Start the run and record the emitted run ID: +7. Start the run and record the emitted run ID, then immediately print a logs-follow command the user can run independently: ```bash agent-workflow start --plan --executor ``` +After starting, announce: +```text +Run started. Monitor live output: agent-workflow logs --run --follow +``` + Honor an explicit `max-cycles` argument. Otherwise use project configuration and workflow defaults. ## Drive The State Machine @@ -78,6 +85,21 @@ Continue until a terminal status. Do not stop after merely proposing a decision. 3. Review for correctness, regressions, security, and missing tests. Assign stable IDs such as `codex--1` to concrete issues. 4. Independently run every `verificationCommands` entry from the repository root and retain concise evidence. +### While waiting for an executor + +While `agent-workflow status --run --json` reports `status: "executing"`, provide a concise progress update approximately every 20-30 seconds using the `activitySummary` and `executionElapsed` fields from the status JSON or from `agent-workflow status --run --json`. Example: + +``` +Executor has been running for ~2m. Current activity: Editing src/parser.ts (PID 12345). +``` + +- Do not emit raw JSON, empty heartbeat spam, or repeat identical summaries. +- Mention stage, error summaries, or notable file changes when available. +- Always invite the user to watch live output independently: + ``` + Watch live: agent-workflow logs --run --follow + ``` + ### `awaiting_scope_resolution` 1. Confirm the Codex review bundle was not prepared and inspect every structured `scopeViolations` entry. diff --git a/src/child-process.test.ts b/src/child-process.test.ts index 0a1dca6..9a1c582 100644 --- a/src/child-process.test.ts +++ b/src/child-process.test.ts @@ -1,4 +1,6 @@ import assert from "node:assert/strict"; +import fs from "node:fs"; +import path from "node:path"; import { Writable } from "node:stream"; import { test } from "vitest"; import { spawnBufferedChild, spawnStreamingChild } from "./child-process.js"; @@ -117,3 +119,134 @@ test("spawnStreamingChild escalates ignored SIGTERM to SIGKILL after the abort g assert.equal(child.signal, "SIGKILL"); assert.match(chunks.join("") + child.stdout, /ready/); }); + +// --------------------------------------------------------------------------- +// logFilePath tee (real-time observability) +// --------------------------------------------------------------------------- + +test("logFilePath: stdout is readable from the log file before the child exits", async () => { + const tmpDir = fs.mkdtempSync("/tmp/agent-workflow-tee-test-"); + try { + const logFile = path.join(tmpDir, "exec.log"); + const { stream } = collectSink(); + const script = [ + "process.stdout.write('pre-delay\\n');", + "setTimeout(() => { process.stdout.write('post-delay\\n'); }, 200);", + ].join(""); + + const childPromise = spawnStreamingChild(process.execPath, ["-e", script], { + stdoutSink: stream, + stderrSink: stream, + logFilePath: logFile, + }); + + // Before the child finishes, the log file should already have 'pre-delay' + await new Promise((resolve, reject) => { + const check = () => { + try { + if (fs.existsSync(logFile) && fs.readFileSync(logFile, "utf8").includes("pre-delay")) return resolve(); + setTimeout(check, 10); + } catch { setTimeout(check, 10); } + }; + setTimeout(() => reject(new Error("timed out waiting for pre-delay in log")), 2000); + check(); + }); + + await childPromise; + + // After completion, the log file contains all output + const fullLog = fs.readFileSync(logFile, "utf8"); + assert.match(fullLog, /pre-delay/); + assert.match(fullLog, /post-delay/); + } finally { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } +}); + +test("logFilePath: stderr tee is also captured in the log file", async () => { + const tmpDir = fs.mkdtempSync("/tmp/agent-workflow-tee-test-"); + try { + const logFile = path.join(tmpDir, "exec.log"); + const { stream } = collectSink(); + const r = await spawnStreamingChild( + process.execPath, + ["-e", "process.stderr.write('stderr-msg\\n'); process.stdout.write('stdout-msg\\n');"], + { stdoutSink: stream, stderrSink: stream, logFilePath: logFile }, + ); + assert.equal(r.status, 0); + const logContent = fs.readFileSync(logFile, "utf8"); + assert.match(logContent, /stderr-msg/); + assert.match(logContent, /stdout-msg/); + } finally { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } +}); + +test("logFilePath: Chinese UTF-8 is preserved correctly", async () => { + const tmpDir = fs.mkdtempSync("/tmp/agent-workflow-utf8-test-"); + try { + const logFile = path.join(tmpDir, "exec.log"); + const { stream } = collectSink(); + const r = await spawnStreamingChild( + process.execPath, + ["-e", "console.log('你好世界'); console.log('日本語'); console.log('🚀');"], + { stdoutSink: stream, stderrSink: stream, logFilePath: logFile }, + ); + assert.equal(r.status, 0); + const logContent = fs.readFileSync(logFile, "utf8"); + assert.match(logContent, /你好世界/); + assert.match(logContent, /日本語/); + assert.match(logContent, /🚀/); + } finally { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } +}); + +test("logFilePath: onProgress reports pid and bytesLogged", async () => { + const tmpDir = fs.mkdtempSync("/tmp/agent-workflow-progress-test-"); + try { + const logFile = path.join(tmpDir, "exec.log"); + const { stream } = collectSink(); + const progressUpdates: Array<{ pid?: number; bytesLogged: number }> = []; + const r = await spawnStreamingChild( + process.execPath, + ["-e", "process.stdout.write('hello world');"], + { + stdoutSink: stream, + stderrSink: stream, + logFilePath: logFile, + onProgress: (info) => { progressUpdates.push(info); }, + }, + ); + assert.equal(r.status, 0); + assert.ok(progressUpdates.length >= 1); + assert.ok(progressUpdates.some((p) => p.bytesLogged > 0)); + if (progressUpdates[0].pid) assert.ok(progressUpdates[0].pid! > 0); + } finally { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } +}); + +test("logFilePath: write stream error resolves spawnStreamingChild with result.error", async () => { + const tmpDir = fs.mkdtempSync("/tmp/agent-workflow-log-err-test-"); + try { + // Parent does not exist → ENOENT on first write + const badLogFile = path.join(tmpDir, "nonexistent", "exec.log"); + + const { stream } = collectSink(); + const r = await spawnStreamingChild( + process.execPath, + ["-e", "console.log('hello');"], + { + stdoutSink: stream, + stderrSink: stream, + logFilePath: badLogFile, + }, + ); + assert.equal(r.status, 0); + assert.ok(r.error !== undefined, "expected result.error from log stream failure"); + assert.match(String(r.error), /ENOENT|no such file|open/); + } finally { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } +}); diff --git a/src/child-process.ts b/src/child-process.ts index d68e1df..4e3b3bf 100644 --- a/src/child-process.ts +++ b/src/child-process.ts @@ -1,3 +1,4 @@ +import fs from "node:fs"; import { spawn, spawnSync, type SpawnOptions } from "node:child_process"; export type ChildRunResult = { @@ -46,6 +47,12 @@ export function spawnBufferedChild( export const DEFAULT_ABORT_KILL_GRACE_MS = 2_000; +/** Progress info emitted during output streaming. */ +export interface StreamProgress { + pid?: number; + bytesLogged: number; +} + export async function spawnStreamingChild( command: string, argv: string[], @@ -55,6 +62,12 @@ export async function spawnStreamingChild( maxCaptureChars?: number; /** After SIGTERM, escalate to SIGKILL if the process is still alive. */ abortKillGraceMs?: number; + /** Path to a durable log file that receives a tee of both stdout and stderr. */ + logFilePath?: string; + /** Periodic progress callback (throttled internally). */ + onProgress?: (info: StreamProgress) => void; + /** Live callback for each stdout chunk (for real-time parsing, e.g. Claude JSON lines). */ + onStdoutData?: (chunk: string) => void; }, ): Promise { const stdoutSink = options.stdoutSink ?? process.stdout; @@ -67,8 +80,6 @@ export async function spawnStreamingChild( cwd: options.cwd, env: options.env, stdio: ["ignore", "pipe", "pipe"], - // Preserve terminal signal delivery for ordinary streaming children. - // Workflow callers opt into a group so their abort signal reaches executor descendants. detached: options.processGroup === true && process.platform !== "win32", }; const child = spawn(command, argv, spawnOpts); @@ -76,13 +87,57 @@ export async function spawnStreamingChild( let stderr = ""; let settled = false; let killTimer: NodeJS.Timeout | undefined; + let bytesLogged = 0; + let lastProgressMs = 0; + const PROGRESS_THROTTLE_MS = 1000; + + // Open durable log file if requested + const logStream = options.logFilePath + ? fs.createWriteStream(options.logFilePath, { flags: "a" }) + : null; + let logStreamSettled = false; + let logStreamError: Error | undefined; + if (logStream) { + logStream.on("error", (err) => { + logStreamError = err; + // If finish() hasn't run yet, the stream is already destroyed; + // flag it so finish() knows not to wait for events that won't fire. + logStreamSettled = true; + }); + } const finish = (result: ChildRunResult) => { if (settled) return; settled = true; if (killTimer) clearTimeout(killTimer); options.signal?.removeEventListener("abort", abort); - resolve(result); + if (logStream) { + // Attach handlers BEFORE calling end so we never miss an event. + if (!logStreamSettled) { + logStream.on("finish", () => { + logStreamSettled = true; + if (logStreamError && !result.error) result.error = logStreamError; + resolve(result); + }); + logStream.on("error", (err) => { + if (!settled) { + logStreamSettled = true; + if (!result.error) result.error = err; + resolve(result); + } + }); + } + logStream.end(); + // If the stream was already destroyed by an earlier error, end() may + // not emit finish or error — resolve immediately. + if (logStreamSettled) { + if (logStreamError && !result.error) result.error = logStreamError; + resolve(result); + } + } else { + if (logStreamError && !result.error) result.error = logStreamError; + resolve(result); + } }; const sendSignal = (signal: NodeJS.Signals) => { @@ -117,18 +172,39 @@ export async function spawnStreamingChild( if (options.signal?.aborted) abort(); else options.signal?.addEventListener("abort", abort, { once: true }); + callOnProgress(true); + child.stdout?.setEncoding("utf8"); child.stderr?.setEncoding("utf8"); child.stdout?.on("data", (chunk: string) => { stdout = appendBoundedTail(stdout, chunk, maxCaptureChars); stdoutSink.write(chunk); + options.onStdoutData?.(chunk); + if (logStream) { + bytesLogged += Buffer.byteLength(chunk, "utf8"); + logStream.write(chunk); + } + callOnProgress(); }); child.stderr?.on("data", (chunk: string) => { stderr = appendBoundedTail(stderr, chunk, maxCaptureChars); stderrSink.write(chunk); + if (logStream) { + bytesLogged += Buffer.byteLength(chunk, "utf8"); + logStream.write(chunk); + } + callOnProgress(); }); + function callOnProgress(skipThrottle = false): void { + const now = Date.now(); + if (skipThrottle || now - lastProgressMs >= PROGRESS_THROTTLE_MS) { + if (!skipThrottle) lastProgressMs = now; + options.onProgress?.({ pid: child.pid ?? undefined, bytesLogged }); + } + } + child.on("error", (error) => { finish({ status: null, diff --git a/src/cli.ts b/src/cli.ts index da552ee..b84a160 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -16,6 +16,7 @@ import { decideWorkflow, statusWorkflow, abortWorkflow, + logsWorkflow, retryReviewWorkflow, retryExecuteWorkflow, WorkflowEngineError, @@ -72,6 +73,14 @@ async function main() { abortWorkflow({ runId: args.runId, reason: args.reason }); break; + case "logs": + await logsWorkflow({ + runId: args.runId, + follow: args.follow, + tail: args.tail, + }); + break; + default: { const _exhaustive: never = args; throw new Error(`Unknown subcommand: ${(_exhaustive as any).sub}`); diff --git a/src/index.ts b/src/index.ts index 42d5ee3..30f1cac 100644 --- a/src/index.ts +++ b/src/index.ts @@ -10,6 +10,7 @@ export { decideWorkflow, statusWorkflow, abortWorkflow, + logsWorkflow, retryReviewWorkflow, retryExecuteWorkflow, WorkflowEngineError, @@ -22,6 +23,7 @@ export type { DecideWorkflowOpts, StatusWorkflowOpts, AbortWorkflowOpts, + LogsWorkflowOpts, RetryReviewWorkflowOpts, RetryExecuteWorkflowOpts, } from "./workflow-engine.js"; @@ -66,6 +68,11 @@ export { nowIso, initWorkflowState, findRunDir, + writeProgressSidecar, + readProgressSidecar, + progressSidecarPath, + attemptLogPath, + nextAttemptIndex, } from "./workflow-state.js"; export { @@ -88,6 +95,7 @@ export type { ExecutorResult, ExecutorStartOpts, ExecutorResumeOpts, + ProgressSidecar, } from "./workflow-types.js"; export { @@ -105,6 +113,7 @@ export type { WorkflowDecideArgs, WorkflowStatusArgs, WorkflowAbortArgs, + WorkflowLogsArgs, } from "./workflow-args.js"; export type { @@ -149,4 +158,5 @@ export { export type { ChildRunResult, + StreamProgress, } from "./child-process.js"; diff --git a/src/workflow-args.ts b/src/workflow-args.ts index eba0fee..88f5a7c 100644 --- a/src/workflow-args.ts +++ b/src/workflow-args.ts @@ -2,7 +2,7 @@ * Argument parsing for agent-workflow subcommands. */ -export type WorkflowSubcommand = "start" | "review" | "retry-review" | "retry-execute" | "decide" | "status" | "abort"; +export type WorkflowSubcommand = "start" | "review" | "retry-review" | "retry-execute" | "decide" | "status" | "abort" | "logs"; export interface WorkflowStartArgs { sub: "start"; @@ -44,6 +44,13 @@ export interface WorkflowAbortArgs { reason: string; } +export interface WorkflowLogsArgs { + sub: "logs"; + runId: string; + follow: boolean; + tail: number; +} + export type WorkflowArgs = | WorkflowStartArgs | WorkflowReviewArgs @@ -51,7 +58,8 @@ export type WorkflowArgs = | WorkflowRetryExecuteArgs | WorkflowDecideArgs | WorkflowStatusArgs - | WorkflowAbortArgs; + | WorkflowAbortArgs + | WorkflowLogsArgs; export class WorkflowArgsError extends Error { constructor(message: string) { @@ -84,6 +92,7 @@ Usage: agent-workflow decide --run --input agent-workflow status --run [--json] agent-workflow abort --run --reason + agent-workflow logs --run [--follow] [--tail ] Outputs AGENT_WORKFLOW_META_JSON: to stderr (set AGENT_WORKFLOW_META_STDOUT=1 for stdout). @@ -252,6 +261,28 @@ export function parseWorkflowArgs(argv: string[]): WorkflowArgs { return { sub: "abort", runId, reason }; } + case "logs": { + let runId: string | undefined; + let follow = false; + let tail = 50; + while (rest.length > 0) { + const arg = rest.shift()!; + if (arg === "--run") { + runId = requireValue(arg, rest); + } else if (arg === "--follow" || arg === "-f") { + follow = true; + } else if (arg === "--tail") { + tail = parsePositiveInt(arg, requireValue(arg, rest)); + } else if (arg === "-h" || arg === "--help") { + workflowUsage(0); + } else { + throw new WorkflowArgsError(`Unknown option: ${arg}`); + } + } + if (!runId) throw new WorkflowArgsError("--run is required for workflow logs"); + return { sub: "logs", runId, follow, tail }; + } + default: throw new WorkflowArgsError(`Unknown workflow subcommand: ${sub}\n\n${WORKFLOW_USAGE}`); } diff --git a/src/workflow-engine.ts b/src/workflow-engine.ts index 6dc7916..c3bdb98 100644 --- a/src/workflow-engine.ts +++ b/src/workflow-engine.ts @@ -41,6 +41,7 @@ import { initWorkflowState, nowIso, readState, + readProgressSidecar, releaseLock, runDir as makeRunDir, writeState, @@ -179,20 +180,51 @@ function transitionToTerminal( releaseLock(state.repoRoot, state.runId); } +/** + * Register an attempt log path in the cycle record BEFORE spawning the executor + * so that a crash during execution still leaves a discoverable log file. + */ +function registerAttemptLog(dir: string, cycleRecord: CycleRecord, logFilePath: string): void { + const priorLogs = cycleRecord.executorAttemptLogs + ?? (cycleRecord.executorLogFile ? [cycleRecord.executorLogFile] : []); + const relativeLog = path.relative(dir, logFilePath); + cycleRecord.executorAttemptLogs = [...priorLogs, relativeLog]; + cycleRecord.executorLogFile = relativeLog; +} + +/** + * Finalize an attempt after the executor finishes: set exit code and report slice. + * The log path is already registered (by registerAttemptLog), so we never re-derive it. + */ function persistExecutorAttempt( dir: string, cycleRecord: CycleRecord, execResult: ExecutorResult, + logFilePath?: string, ): void { - const priorLogs = cycleRecord.executorAttemptLogs - ?? (cycleRecord.executorLogFile ? [cycleRecord.executorLogFile] : []); - const attemptIndex = priorLogs.length; + // If the log file was created and registered before spawning (even if empty), + // keep it and don't overwrite. Empty logs are valid (e.g. immediate exit). + if (logFilePath && fs.existsSync(logFilePath)) { + const relativeLog = path.relative(dir, logFilePath); + cycleRecord.executorLogFile = relativeLog; + cycleRecord.executorExitCode = execResult.exitCode ?? undefined; + cycleRecord.executorReport = execResult.report.slice(0, 2000); + if (!cycleRecord.executorAttemptLogs?.includes(relativeLog)) { + cycleRecord.executorAttemptLogs = [...(cycleRecord.executorAttemptLogs ?? []), relativeLog]; + } + return; + } + + // Fallback: write the full report if logFilePath wasn't provided (synchronous adapter or legacy path) + const attemptIndex = (cycleRecord.executorAttemptLogs?.length ?? 0); const execLog = attemptIndex === 0 ? cycleExecLogFile(dir, cycleRecord.cycleIndex) : path.join(dir, `cycle-${cycleRecord.cycleIndex}-exec-retry-${attemptIndex}.log`); fs.writeFileSync(execLog, execResult.report, "utf8"); const relativeLog = path.relative(dir, execLog); - cycleRecord.executorAttemptLogs = [...priorLogs, relativeLog]; + if (!cycleRecord.executorAttemptLogs?.includes(relativeLog)) { + cycleRecord.executorAttemptLogs = [...(cycleRecord.executorAttemptLogs ?? []), relativeLog]; + } cycleRecord.executorLogFile = relativeLog; cycleRecord.executorExitCode = execResult.exitCode ?? undefined; cycleRecord.executorReport = execResult.report.slice(0, 2000); @@ -318,6 +350,13 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s const adapter = createExecutorAdapter(config.executor); const { ac, cleanup } = createManagedController(state.timeoutSeconds); + + // Create attempt log file BEFORE spawning the executor + const execLogPath = path.join(dir, `cycle-${cycleIndex}-exec.log`); + fs.writeFileSync(execLogPath, "", "utf8"); + registerAttemptLog(dir, cycleRecord, execLogPath); + writeState(dir, state); // persist so retry can discover the log + let execResult; try { execResult = await adapter.start({ @@ -325,6 +364,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s plan: typedPlan, runDir: dir, cycleIndex, + logFilePath: execLogPath, config: config.executorConfig, timeoutSeconds: state.timeoutSeconds, signal: ac.signal, @@ -336,7 +376,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s } cleanup(); - persistExecutorAttempt(dir, cycleRecord, execResult); + persistExecutorAttempt(dir, cycleRecord, execResult, execLogPath); if (execResult.sessionHandle) { state.sessionHandle = execResult.sessionHandle; if (execResult.sessionHandle.kind === "agy") { @@ -716,6 +756,18 @@ async function resumeExecutorCycle( const adapter = createExecutorAdapter(state.executor); const { ac, cleanup } = createManagedController(state.timeoutSeconds); + + // Determine attempt index and create log file before spawning + const priorLogs = cycleRecord.executorAttemptLogs + ?? (cycleRecord.executorLogFile ? [cycleRecord.executorLogFile] : []); + const attemptIndex = priorLogs.length; + const execLogPath = attemptIndex === 0 + ? path.join(dir, `cycle-${cycleIndex}-exec.log`) + : path.join(dir, `cycle-${cycleIndex}-exec-retry-${attemptIndex}.log`); + fs.writeFileSync(execLogPath, "", "utf8"); + registerAttemptLog(dir, cycleRecord, execLogPath); + writeState(dir, state); // persist so retry can discover the log + let execResult; try { execResult = await adapter.resume({ @@ -725,6 +777,7 @@ async function resumeExecutorCycle( handle: state.sessionHandle, runDir: dir, cycleIndex, + logFilePath: execLogPath, config: state.executorConfig ?? {}, timeoutSeconds: state.timeoutSeconds, signal: ac.signal, @@ -742,7 +795,7 @@ async function resumeExecutorCycle( } cleanup(); - persistExecutorAttempt(dir, cycleRecord, execResult); + persistExecutorAttempt(dir, cycleRecord, execResult, execLogPath); if (execResult.sessionHandle) { state.sessionHandle = execResult.sessionHandle; if (execResult.sessionHandle.kind === "agy") { @@ -1088,6 +1141,26 @@ export function statusWorkflow(opts: StatusWorkflowOpts): void { const { state, dir } = requireState(opts.runId); const cycle = state.cycles.find((c) => c.cycleIndex === state.currentCycle); + // Read live progress sidecar for real-time observability + const sidecar = readProgressSidecar(dir); + const liveLogFile = cycle?.executorLogFile + ? path.join(dir, cycle.executorLogFile) + : sidecar?.logFilePath + ? path.join(dir, sidecar.logFilePath) + : undefined; + let executionElapsed: string | undefined; + if (sidecar?.executionDurationMs !== undefined) { + const ms = sidecar.executionDurationMs; + const secs = Math.floor(ms / 1000); + if (secs < 60) executionElapsed = `${secs}s`; + else executionElapsed = `${Math.floor(secs / 60)}m ${secs % 60}s`; + } else if (sidecar?.startedAt) { + const ms = Date.now() - new Date(sidecar.startedAt).getTime(); + const secs = Math.floor(ms / 1000); + if (secs < 60) executionElapsed = `${secs}s`; + else executionElapsed = `${Math.floor(secs / 60)}m ${secs % 60}s`; + } + const output: WorkflowStatusOutput = { runId: state.runId, status: state.status, @@ -1108,6 +1181,12 @@ export function statusWorkflow(opts: StatusWorkflowOpts): void { repoRoot: state.repoRoot, createdAt: state.createdAt, updatedAt: state.updatedAt, + liveLogFile, + executionElapsed, + lastActivity: sidecar?.lastActivityAt, + logBytes: sidecar?.logBytes, + executorPid: sidecar?.pid, + activitySummary: sidecar?.activitySummary, }; if (opts.json) { @@ -1133,6 +1212,14 @@ export function statusWorkflow(opts: StatusWorkflowOpts): void { lines.push(p("Session handle", kind + extra)); } + // Live execution fields + if (output.executorPid) lines.push(p("Executor PID", String(output.executorPid))); + if (output.executionElapsed) lines.push(p("Elapsed", output.executionElapsed)); + if (output.activitySummary) lines.push(p("Activity", output.activitySummary)); + if (output.lastActivity) lines.push(p("Last activity", output.lastActivity)); + if (output.logBytes !== undefined) lines.push(p("Log bytes", `${output.logBytes}`)); + if (output.liveLogFile) lines.push(p("Live log", output.liveLogFile)); + if (cycle?.hostReviewFile) lines.push(p("Host review", path.join(dir, cycle.hostReviewFile))); if (state.status === "awaiting_scope_resolution" && cycle?.scopeViolations?.length) { lines.push(p("Scope violations", String(cycle.scopeViolations.length))); @@ -1183,3 +1270,158 @@ export function abortWorkflow(opts: AbortWorkflowOpts): void { process.stdout.write(`workflow aborted: ${opts.runId}\nreason: ${opts.reason}\n`); emitWorkflowMeta({ runId: opts.runId, status: "aborted", reason: opts.reason }); } + +// --------------------------------------------------------------------------- +// workflow logs +// --------------------------------------------------------------------------- + +export interface LogsWorkflowOpts { + runId: string; + follow: boolean; + tail: number; + repoRoot?: string; +} + +export async function logsWorkflow(opts: LogsWorkflowOpts): Promise { + const { state, dir } = requireState(opts.runId); + + // Resolve current log path: sidecar is freshest during execution + const sidecar = readProgressSidecar(dir); + let logPath: string | undefined; + if (sidecar?.logFilePath && fs.existsSync(path.join(dir, sidecar.logFilePath))) { + logPath = path.join(dir, sidecar.logFilePath); + } else { + const cycle = state.cycles.find((c) => c.cycleIndex === state.currentCycle); + if (cycle?.executorLogFile) { + logPath = path.join(dir, cycle.executorLogFile); + } + } + + // If no log file and state is executing, wait briefly for it to appear + const isExecuting = state.status === "executing"; + if (!logPath && isExecuting) { + for (let i = 0; i < 30; i++) { + await sleepMs(500); + const sc = readProgressSidecar(dir); + if (sc?.logFilePath && fs.existsSync(path.join(dir, sc.logFilePath))) { + logPath = path.join(dir, sc.logFilePath); + break; + } + const updatedState = readState(dir); + if (updatedState) { + const uc = updatedState.cycles.find((c) => c.cycleIndex === updatedState.currentCycle); + if (uc?.executorLogFile) { + logPath = path.join(dir, uc.executorLogFile); + break; + } + } + } + } + + if (!logPath || !fs.existsSync(logPath)) { + process.stdout.write(`No log file found for run '${opts.runId}'.\n`); + return; + } + + // Tail the last N lines + const lines = readTailLines(logPath, opts.tail); + for (const line of lines) { + process.stdout.write(line + "\n"); + } + + if (!opts.follow) return; + + // Follow mode: watch for new data and retry rotations + let currentPath = logPath; + let lastByteOffset = fs.statSync(currentPath).size; + + // States that indicate execution has ceased (no more output expected) + const terminalStates = new Set([ + "completed", "needs_human", "blocked", "budget_exhausted", "aborted", + "awaiting_review", "awaiting_host", "awaiting_scope_resolution", + ]); + + while (true) { + const curState = readState(dir); + if (!curState) break; + + // Exit if the workflow has stopped executing + if (terminalStates.has(curState.status)) { + drainBytes(currentPath, lastByteOffset); + return; + } + + // Check for log rotation to a newer attempt + const sc = readProgressSidecar(dir); + let candidatePath: string | undefined; + if (sc?.logFilePath && fs.existsSync(path.join(dir, sc.logFilePath))) { + candidatePath = path.join(dir, sc.logFilePath); + } else { + const curCycle = curState.cycles.find((c) => c.cycleIndex === curState.currentCycle); + if (curCycle?.executorLogFile) { + candidatePath = path.join(dir, curCycle.executorLogFile); + } + } + + if (candidatePath && candidatePath !== currentPath) { + drainBytes(currentPath, lastByteOffset); + currentPath = candidatePath; + lastByteOffset = fs.statSync(currentPath).size; + process.stdout.write(`\n--- log rotated to ${path.basename(currentPath)} ---\n`); + const headLines = readTailLines(currentPath, Math.min(opts.tail, 10)); + for (const line of headLines) { + process.stdout.write(line + "\n"); + } + continue; + } + + // Drain new bytes since lastByteOffset using byte-accurate reading + try { + const stats = fs.statSync(currentPath); + if (stats.size > lastByteOffset) { + drainBytes(currentPath, lastByteOffset); + lastByteOffset = stats.size; + } + } catch { + // File may have been deleted or rotated + } + + await sleepMs(500); + } +} + +function sleepMs(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function readTailLines(filePath: string, n: number): string[] { + try { + const content = fs.readFileSync(filePath, "utf8"); + const lines = content.split("\n"); + return lines.slice(Math.max(0, lines.length - n)); + } catch { + return []; + } +} + +/** + * Drain all bytes starting at fromByteOffset using proper byte-level I/O. + */ +function drainBytes(filePath: string, fromByteOffset: number): void { + try { + const fd = fs.openSync(filePath, "r"); + try { + const stats = fs.fstatSync(fd); + const remaining = stats.size - fromByteOffset; + if (remaining > 0) { + const buf = Buffer.alloc(remaining); + fs.readSync(fd, buf, 0, remaining, fromByteOffset); + process.stdout.write(buf.toString("utf8")); + } + } finally { + fs.closeSync(fd); + } + } catch { + // ignore + } +} diff --git a/src/workflow-executor.test.ts b/src/workflow-executor.test.ts index 9b24cbc..eba6cb1 100644 --- a/src/workflow-executor.test.ts +++ b/src/workflow-executor.test.ts @@ -8,7 +8,7 @@ * without running real AI agents. */ -import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { describe, it, expect, beforeAll, beforeEach, afterEach } from "vitest"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; @@ -218,6 +218,198 @@ describe("ClaudeAdapter", () => { }); }); +// --------------------------------------------------------------------------- +// Claude stream-json activity derivation (pure unit tests via exported function) +// --------------------------------------------------------------------------- + +describe("deriveClaudeActivity (stream-json parsing)", () => { + let parseLine: (line: string) => string | undefined; + + beforeAll(async () => { + const mod = await import("./workflow-executor.js"); + parseLine = mod.parseClaudeActivityLine; + }); + + it("extracts tool_use from a type=assistant record with message.content and file_path", () => { + const line = `{"type":"assistant","message":{"content":[{"type":"thinking","thinking":"I'll read the file first"},{"type":"tool_use","name":"Read","input":{"file_path":"src/foo.ts"}}]}}`; + expect(parseLine(line)).toBe("Reading src/foo.ts"); + }); + + it("extracts tool_use from a type=assistant record with command", () => { + const line = `{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Bash","input":{"command":"npm test"}}]}}`; + expect(parseLine(line)).toBe("Running: npm test"); + }); + + it("does NOT surface thinking content in activity summary", () => { + const line = `{"type":"assistant","message":{"content":[{"type":"thinking","thinking":"secret reasoning"},{"type":"tool_use","name":"Read","input":{"file_path":"src/ts"}}]}}`; + const result = parseLine(line); + expect(result).toBe("Reading src/ts"); + expect(result).not.toContain("secret reasoning"); + // A line with only thinking should return undefined + const thinkingOnly = `{"type":"assistant","message":{"content":[{"type":"thinking","thinking":"hidden"}]}}`; + expect(parseLine(thinkingOnly)).toBeUndefined(); + }); + + it("extracts capitalized tool names: Write, Edit, Glob, Grep", () => { + expect(parseLine(`{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Write","input":{"file_path":"new.ts"}}]}}`)).toBe("Writing new.ts"); + expect(parseLine(`{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Edit","input":{"file_path":"edit.ts"}}]}}`)).toBe("Editing edit.ts"); + expect(parseLine(`{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Glob"}]}}`)).toBe("Searching files"); + expect(parseLine(`{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Grep","input":{"pattern":"foo"}}]}}`)).toBe("Searching code"); + }); + + it("extracts type=system subtype=init as session", () => { + expect(parseLine(`{"type":"system","subtype":"init","session_id":"abc123"}`)).toBe("Creating session"); + }); + + it("extracts is_error result as error", () => { + expect(parseLine(`{"type":"result","is_error":true,"result":"Rate limit exceeded","session_id":"abc"}`)).toBe("Error: Rate limit exceeded"); + }); + + it("extracts type=result at top level", () => { + expect(parseLine(`{"type":"result","result":"Tool ran without output"}`)).toBe("Result: Tool ran without output"); + }); + + it("extracts type=error at top level", () => { + expect(parseLine(`{"type":"error","error":"Permission denied"}`)).toBe("Error: Permission denied"); + }); + + it("returns undefined for non-JSON lines", () => { + expect(parseLine("implemented")).toBeUndefined(); + expect(parseLine("")).toBeUndefined(); + }); + + it("returns undefined for unknown types", () => { + expect(parseLine(`{"type":"ping"}`)).toBeUndefined(); + }); +}); + +// --------------------------------------------------------------------------- +// End-to-end fake Claude stream-json fixture (adapter start and resume) +// --------------------------------------------------------------------------- + +describe("ClaudeAdapter stream-json fixture", () => { + let tmpDir: string; + let previousHome: string | undefined; + + beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "agent-workflow-claude-fixture-")); + previousHome = process.env.HOME; + process.env.HOME = tmpDir; + }); + + afterEach(() => { + if (previousHome === undefined) delete process.env.HOME; + else process.env.HOME = previousHome; + cleanupDir(tmpDir); + }); + + /** Create a fake Claude binary that outputs stream-json lines. */ + function makeFakeClaude(lines: string[], exitCode = 0): string { + const scriptPath = path.join(tmpDir, "fake-claude.sh"); + const escaped = lines.map((l) => l.replace(/'/g, "'\\''")).join("\\n"); + fs.writeFileSync(scriptPath, `#!/bin/bash +printf '${escaped}\\n' +exit ${exitCode} +`, { mode: 0o755 }); + return scriptPath; + } + + it("start: emits --output-format stream-json, captures session, has tool activity in sidecar", async () => { + const argsFile = path.join(tmpDir, "args.txt"); + const fakeBin = makeFakeClaude([ + '{"type":"system","subtype":"init","session_id":"fixture-session-456"}', + '{"type":"assistant","message":{"content":[{"type":"thinking","thinking":"thinking text hidden"},{"type":"tool_use","name":"Read","input":{"file_path":"src/index.ts"}}]}}', + '{"type":"result","result":"ok"}', + ]); + const trackingBin = path.join(tmpDir, "tracking-claude.sh"); + fs.writeFileSync(trackingBin, `#!/bin/bash +printf '%s\\0' "$@" >> "${argsFile}" +"${fakeBin}" "$@" +`, { mode: 0o755 }); + + const adapter = createExecutorAdapter("claude"); + const logFilePath = path.join(tmpDir, "claude-exec.log"); + const result = await adapter.start({ + repoRoot: tmpDir, + plan: makeValidPlan(), + runDir: tmpDir, + cycleIndex: 1, + logFilePath, + config: { binary: trackingBin }, + }); + + // 1) --output-format stream-json was passed + const capturedArgs = fs.existsSync(argsFile) ? fs.readFileSync(argsFile, "utf8") : ""; + expect(capturedArgs).toContain("--output-format"); + expect(capturedArgs).toContain("stream-json"); + + // 2) Exact session preservation + if (result.sessionHandle?.kind === "claude") { + expect(result.sessionHandle.sessionId).toBe("fixture-session-456"); + } + + // 3) Tool activity in the completed sidecar (no thinking text) + const { readProgressSidecar } = await import("./workflow-state.js"); + const sidecar = readProgressSidecar(tmpDir); + expect(sidecar).toBeDefined(); + expect(sidecar!.activitySummary).toContain("Reading"); + expect(sidecar!.activitySummary).not.toContain("thinking text hidden"); + expect(sidecar!.phase).toBe("completed"); + expect(sidecar!.executor).toBe("claude"); + }); + + it("start with error: extracts failure reason and marks sidecar failed", async () => { + const fakeBin = makeFakeClaude([ + '{"type":"system","subtype":"init","session_id":"err-session-789"}', + '{"type":"result","is_error":true,"result":"Rate limit exceeded","session_id":"err-session-789"}', + ], 1); + + const adapter = createExecutorAdapter("claude"); + const logFilePath = path.join(tmpDir, "claude-error.log"); + const result = await adapter.start({ + repoRoot: tmpDir, + plan: makeValidPlan(), + runDir: tmpDir, + cycleIndex: 1, + logFilePath, + config: { binary: fakeBin }, + }); + + expect(result.failed).toBe(true); + expect(result.failureReason).toContain("Rate limit exceeded"); + + const { readProgressSidecar } = await import("./workflow-state.js"); + const sidecar = readProgressSidecar(tmpDir); + expect(sidecar).toBeDefined(); + expect(sidecar!.phase).toBe("failed"); + }); + + it("resume: preserves session ID from the handle", async () => { + const fakeBin = makeFakeClaude([ + '{"type":"system","subtype":"init","session_id":"resumed-session-999"}', + '{"type":"result","result":"completed"}', + ]); + + const adapter = createExecutorAdapter("claude"); + const logFilePath = path.join(tmpDir, "claude-resume.log"); + const result = await adapter.resume({ + repoRoot: tmpDir, + plan: makeValidPlan(), + acceptedFindings: "Fix the bug", + handle: { kind: "claude", sessionId: "resumed-session-999" }, + runDir: tmpDir, + cycleIndex: 2, + logFilePath, + config: { binary: fakeBin }, + }); + + if (result.sessionHandle?.kind === "claude") { + expect(result.sessionHandle.sessionId).toBe("resumed-session-999"); + } + expect(result.failed).toBe(false); + }); +}); + // --------------------------------------------------------------------------- // Agy adapter tests // --------------------------------------------------------------------------- diff --git a/src/workflow-executor.ts b/src/workflow-executor.ts index 9f35ff0..e4e49d4 100644 --- a/src/workflow-executor.ts +++ b/src/workflow-executor.ts @@ -7,6 +7,7 @@ * - Return ExecutorResult with session handle for future resume */ +import { randomUUID } from "node:crypto"; import { execFileSync, spawnSync } from "node:child_process"; import fs from "node:fs"; import path from "node:path"; @@ -22,9 +23,15 @@ import type { ExecutorResult, ExecutorSessionHandle, ExecutorStartOpts, + ProgressSidecar, ReasonixSessionHandle, } from "./workflow-types.js"; -import { nowIso } from "./workflow-state.js"; +import { + nowIso, + writeProgressSidecar, + readProgressSidecar, + attemptLogPath, +} from "./workflow-state.js"; // --------------------------------------------------------------------------- // Safety prompt footer (appended to every executor instruction) @@ -45,6 +52,166 @@ const SAFETY_CONSTRAINTS = ` - If you encounter an error you cannot resolve, stop and report it clearly. `.trimStart(); +// --------------------------------------------------------------------------- +// Progress sidecar helpers +// --------------------------------------------------------------------------- + +function startProgressSidecar( + dir: string, + executor: "reasonix" | "claude" | "agy", + cycleIndex: number, + attemptIndex: number, + logFilePath: string | undefined, + intervalMs = 5000, +): NodeJS.Timeout { + const sidecar: ProgressSidecar = { + executor, + cycleIndex, + attemptIndex, + phase: "executing", + startedAt: nowIso(), + lastActivityAt: nowIso(), + logFilePath: logFilePath ? path.relative(dir, logFilePath) : undefined, + logBytes: 0, + activitySummary: "Starting...", + updatedAt: nowIso(), + }; + writeProgressSidecar(dir, sidecar); + + return setInterval(() => { + const current = readProgressSidecar(dir); + if (!current) return; + const now = nowIso(); + current.updatedAt = now; + if (current.logFilePath) { + try { + current.logBytes = fs.statSync(path.join(dir, current.logFilePath)).size; + } catch { + // file may not exist yet + } + } + writeProgressSidecar(dir, current); + }, intervalMs); +} + +function updateSidecar(dir: string, updates: Partial): void { + const current = readProgressSidecar(dir); + if (!current) return; + Object.assign(current, updates, { updatedAt: nowIso() }); + writeProgressSidecar(dir, current); +} + +function stopProgressSidecar(dir: string, timer: NodeJS.Timeout, failed?: boolean): void { + clearInterval(timer); + const current = readProgressSidecar(dir); + if (!current) return; + current.phase = failed ? "failed" : "completed"; + const elapsed = Date.now() - new Date(current.startedAt).getTime(); + current.executionDurationMs = elapsed; + current.updatedAt = nowIso(); + if (current.logFilePath) { + try { + current.logBytes = fs.statSync(path.join(dir, current.logFilePath)).size; + } catch { + // ignore + } + } + writeProgressSidecar(dir, current); +} + +// -------------------------------------------------------------------------- +// Claude stream-json activity derivation +// --------------------------------------------------------------------------- + +function deriveClaudeActivity(line: string): string | undefined { + if (!line.trimStart().startsWith("{")) return undefined; + try { + const parsed = JSON.parse(line) as Record; + const type = String(parsed.type ?? ""); + + // Real Claude Code stream-json envelope: + // type=system, subtype=init, session_id + // type=assistant, message:{content:[{type:"tool_use",...}]} + // type=result, is_error?, result, session_id + + if (type === "assistant") { + const message = parsed.message as Record | undefined; + const content = message?.content as unknown[] | undefined; + if (!Array.isArray(content) || content.length === 0) return undefined; + for (const item of content) { + if (typeof item !== "object" || item === null) continue; + const block = item as Record; + const blockType = String(block.type ?? ""); + if (blockType === "thinking") continue; // DO NOT surface thinking + if (blockType === "tool_use") { + return deriveToolUse(block); + } + } + return undefined; + } + + // Direct tool_use at top level (legacy) + if (type === "tool_use") { + return deriveToolUse(parsed); + } + + // System init / result / error at top level + switch (type) { + case "system": { + if (parsed.subtype === "init") return "Creating session"; + return undefined; + } + case "result": { + // is_error result → error + if (parsed.is_error === true) { + const err = String(parsed.result ?? "").slice(0, 80); + return err ? `Error: ${err}` : "Error occurred"; + } + const result = String(parsed.result ?? "").slice(0, 60); + return result ? `Result: ${result}` : "Processing"; + } + case "error": { + const err = String(parsed.error ?? "").slice(0, 80); + return err ? `Error: ${err}` : "Error occurred"; + } + default: return undefined; + } + } catch { + return undefined; + } +} + +/** Derive a concise activity string from a tool_use content block. */ +function deriveToolUse(block: Record): string | undefined { + const name = String(block.name ?? ""); + const input = block.input as Record | undefined; + const target = (input?.file_path ?? input?.path ?? input?.file ?? input?.filePath ?? "") as string; + const lowerName = name.toLowerCase(); + switch (lowerName) { + case "read": + case "read_file": return target ? `Reading ${target}` : "Reading file"; + case "write": + case "create": return target ? `Writing ${target}` : "Writing file"; + case "edit": + case "edit_file": + case "str_replace_editor": return target ? `Editing ${target}` : "Editing file"; + case "bash": + case "command": { + const cmd = String(input?.command ?? "").slice(0, 80); + return cmd ? `Running: ${cmd}` : "Running command"; + } + case "glob": return "Searching files"; + case "grep": + case "search": return "Searching code"; + case "web_fetch": + case "web": return "Fetching URL"; + case "subagent": + case "agent": + case "dispatch_agent": return "Running sub-agent"; + default: return `${name}...`; + } +} + // --------------------------------------------------------------------------- // Prompt builders // --------------------------------------------------------------------------- @@ -147,6 +314,13 @@ function resolveExecutable(config: ExecutorConfig, defaultBin: string): string { return config.binary || defaultBin; } +function deriveAttemptIndex(logFilePath: string | undefined): number { + if (!logFilePath) return 0; + const base = path.basename(logFilePath); + const match = base.match(/retry-(\d+)\.log$/); + return match ? parseInt(match[1], 10) : 0; +} + // --------------------------------------------------------------------------- // Reasonix adapter // --------------------------------------------------------------------------- @@ -291,12 +465,22 @@ class ReasonixAdapter implements ExecutorAdapter { const args: string[] = ["run", "--dir", opts.repoRoot, prompt]; const startMs = Date.now(); + const logFile = opts.logFilePath; + const attemptIndex = deriveAttemptIndex(logFile); + const sidecarTimer = startProgressSidecar(opts.runDir, "reasonix", opts.cycleIndex, attemptIndex, logFile); + const child = await spawnStreamingChild(bin, args, { cwd: opts.repoRoot, env: executorEnv(), signal: opts.signal, processGroup: true, + logFilePath: logFile, + onProgress: ({ pid, bytesLogged }) => { + updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); + }, }); + const reasonixFailed = child.status !== 0 || !!child.error; + stopProgressSidecar(opts.runDir, sidecarTimer, reasonixFailed); const durationMs = Date.now() - startMs; const newSession = detectNewSessionFile(beforeSnapshot, snapshotDirs, opts.repoRoot); @@ -305,7 +489,7 @@ class ReasonixAdapter implements ExecutorAdapter { : undefined; const report = child.stdout + (child.stderr ? `\n[stderr]\n${child.stderr}` : ""); - const failed = child.status !== 0 || !!child.error; + const failed = reasonixFailed; return { exitCode: child.status, @@ -332,12 +516,22 @@ class ReasonixAdapter implements ExecutorAdapter { ]; const startMs = Date.now(); + const logFile = opts.logFilePath; + const attemptIndex = deriveAttemptIndex(logFile); + const sidecarTimer = startProgressSidecar(opts.runDir, "reasonix", opts.cycleIndex, attemptIndex, logFile); + const child = await spawnStreamingChild(bin, args, { cwd: opts.repoRoot, env: executorEnv(), signal: opts.signal, processGroup: true, + logFilePath: logFile, + onProgress: ({ pid, bytesLogged }) => { + updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); + }, }); + const reasonixFailed = child.status !== 0 || !!child.error; + stopProgressSidecar(opts.runDir, sidecarTimer, reasonixFailed); const durationMs = Date.now() - startMs; const report = child.stdout + (child.stderr ? `\n[stderr]\n${child.stderr}` : ""); @@ -346,7 +540,7 @@ class ReasonixAdapter implements ExecutorAdapter { report, durationMs, sessionHandle: handle, // Reasonix handle stays the same file - failed: child.status !== 0 || !!child.error, + failed: reasonixFailed, ...(child.error ? { failureReason: child.error.message } : {}), }; } @@ -390,6 +584,10 @@ function extractClaudeFailureReason(output: string): string | undefined { if (parsed.is_error === true && typeof parsed.result === "string" && parsed.result.trim()) { return parsed.result.trim(); } + // Handle stream-json error format: {"type":"error","error":"Rate limit exceeded"} + if (parsed.type === "error" && typeof parsed.error === "string" && parsed.error.trim()) { + return parsed.error.trim(); + } } catch { // Ignore non-JSON output lines. } @@ -397,8 +595,6 @@ function extractClaudeFailureReason(output: string): string | undefined { return undefined; } -import { randomUUID } from "node:crypto"; - class ClaudeAdapter implements ExecutorAdapter { readonly kind: ExecutorKind = "claude"; @@ -423,19 +619,50 @@ class ClaudeAdapter implements ExecutorAdapter { "-p", "--safe-mode", "--permission-mode", "bypassPermissions", - "--output-format", "json", + "--output-format", "stream-json", "--session-id", sessionId, ...(opts.config.model ? ["--model", opts.config.model] : []), prompt, ]; const startMs = Date.now(); + const logFile = opts.logFilePath; + const attemptIndex = deriveAttemptIndex(logFile); + const sidecarTimer = startProgressSidecar(opts.runDir, "claude", opts.cycleIndex, attemptIndex, logFile); + + let lineBuffer = ""; + const child = await spawnStreamingChild(bin, args, { cwd: opts.repoRoot, env: executorEnv(), signal: opts.signal, processGroup: true, + logFilePath: logFile, + onProgress: ({ pid, bytesLogged }) => { + updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); + }, + onStdoutData: (chunk: string) => { + lineBuffer += chunk; + while (true) { + const nl = lineBuffer.indexOf("\n"); + if (nl < 0) break; + const line = lineBuffer.slice(0, nl); + lineBuffer = lineBuffer.slice(nl + 1); + const activity = deriveClaudeActivity(line); + if (activity) { + // Only surface tool actions in activitySummary, not result/error + const isToolActivity = !activity.startsWith("Result:") && !activity.startsWith("Error:"); + if (isToolActivity) { + updateSidecar(opts.runDir, { activitySummary: activity, lastActivityAt: nowIso() }); + } else { + updateSidecar(opts.runDir, { lastActivityAt: nowIso() }); + } + } + } + }, }); + const claudeFailed = child.status !== 0 || !!child.error; + stopProgressSidecar(opts.runDir, sidecarTimer, claudeFailed); const durationMs = Date.now() - startMs; const capturedId = extractClaudeSessionId(child.stdout) ?? sessionId; @@ -464,19 +691,50 @@ class ClaudeAdapter implements ExecutorAdapter { "-p", "--safe-mode", "--permission-mode", "bypassPermissions", - "--output-format", "json", + "--output-format", "stream-json", "--resume", handle.sessionId, ...(opts.config.model ? ["--model", opts.config.model] : []), prompt, ]; const startMs = Date.now(); + const logFile = opts.logFilePath; + const attemptIndex = deriveAttemptIndex(logFile); + const sidecarTimer = startProgressSidecar(opts.runDir, "claude", opts.cycleIndex, attemptIndex, logFile); + + let lineBuffer = ""; + const child = await spawnStreamingChild(bin, args, { cwd: opts.repoRoot, env: executorEnv(), signal: opts.signal, processGroup: true, + logFilePath: logFile, + onProgress: ({ pid, bytesLogged }) => { + updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); + }, + onStdoutData: (chunk: string) => { + lineBuffer += chunk; + while (true) { + const nl = lineBuffer.indexOf("\n"); + if (nl < 0) break; + const line = lineBuffer.slice(0, nl); + lineBuffer = lineBuffer.slice(nl + 1); + const activity = deriveClaudeActivity(line); + if (activity) { + // Only surface tool actions in activitySummary, not result/error + const isToolActivity = !activity.startsWith("Result:") && !activity.startsWith("Error:"); + if (isToolActivity) { + updateSidecar(opts.runDir, { activitySummary: activity, lastActivityAt: nowIso() }); + } else { + updateSidecar(opts.runDir, { lastActivityAt: nowIso() }); + } + } + } + }, }); + const claudeFailed = child.status !== 0 || !!child.error; + stopProgressSidecar(opts.runDir, sidecarTimer, claudeFailed); const durationMs = Date.now() - startMs; const report = child.stdout + (child.stderr ? `\n[stderr]\n${child.stderr}` : ""); @@ -587,26 +845,36 @@ class AgyAdapter implements ExecutorAdapter { async start(opts: ExecutorStartOpts): Promise { const bin = resolveExecutable(opts.config, "agy"); const prompt = buildStartPrompt(opts); - const logFile = path.join(opts.runDir, `cycle-${opts.cycleIndex}-agy.log`); + const agyLogFile = path.join(opts.runDir, `cycle-${opts.cycleIndex}-agy.log`); const previousWorkspaceConversationId = getAgyWorkspaceConversationId(opts.repoRoot); const args: string[] = [ ...agyCommonArgs(opts.config, opts.timeoutSeconds), - "--log-file", logFile, + "--log-file", agyLogFile, "--print", prompt, ]; const startMs = Date.now(); + const logFile = opts.logFilePath; + const attemptIndex = deriveAttemptIndex(logFile); + const sidecarTimer = startProgressSidecar(opts.runDir, "agy", opts.cycleIndex, attemptIndex, logFile); + const child = await spawnStreamingChild(bin, args, { cwd: opts.repoRoot, env: executorEnv(), signal: opts.signal, processGroup: true, + logFilePath: logFile, + onProgress: ({ pid, bytesLogged }) => { + updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); + }, }); + const agyFailed = child.status !== 0 || !!child.error || !!extractAgyFailureReason(child.stdout, child.stderr); + stopProgressSidecar(opts.runDir, sidecarTimer, agyFailed); const durationMs = Date.now() - startMs; const conversationId = extractAgyConversationId(child.stdout) - ?? extractAgyConversationIdFromLog(logFile) + ?? extractAgyConversationIdFromLog(agyLogFile) ?? (() => { const currentId = getAgyWorkspaceConversationId(opts.repoRoot); return currentId !== previousWorkspaceConversationId ? currentId : undefined; @@ -634,7 +902,7 @@ class AgyAdapter implements ExecutorAdapter { const handle = opts.handle as AgySessionHandle; const bin = resolveExecutable(opts.config, "agy"); const prompt = buildFixPrompt(opts); - const logFile = path.join(opts.runDir, `cycle-${opts.cycleIndex}-agy.log`); + const agyLogFile = path.join(opts.runDir, `cycle-${opts.cycleIndex}-agy.log`); const conversationId = handle.conversationId ?? getAgyWorkspaceConversationId(opts.repoRoot); @@ -643,7 +911,7 @@ class AgyAdapter implements ExecutorAdapter { args = [ ...agyCommonArgs(opts.config, opts.timeoutSeconds), "--conversation", conversationId, - "--log-file", logFile, + "--log-file", agyLogFile, "--print", prompt, ]; } else { @@ -651,24 +919,34 @@ class AgyAdapter implements ExecutorAdapter { args = [ ...agyCommonArgs(opts.config, opts.timeoutSeconds), "--continue", - "--log-file", logFile, + "--log-file", agyLogFile, "--print", prompt, ]; } const startMs = Date.now(); + const logFile = opts.logFilePath; + const attemptIndex = deriveAttemptIndex(logFile); + const sidecarTimer = startProgressSidecar(opts.runDir, "agy", opts.cycleIndex, attemptIndex, logFile); + const child = await spawnStreamingChild(bin, args, { cwd: opts.repoRoot, env: executorEnv(), signal: opts.signal, processGroup: true, + logFilePath: logFile, + onProgress: ({ pid, bytesLogged }) => { + updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() }); + }, }); + const agyFailed = child.status !== 0 || !!child.error || !!extractAgyFailureReason(child.stdout, child.stderr); + stopProgressSidecar(opts.runDir, sidecarTimer, agyFailed); const durationMs = Date.now() - startMs; const report = child.stdout + (child.stderr ? `\n[stderr]\n${child.stderr}` : ""); const capturedConversationId = conversationId ?? extractAgyConversationId(child.stdout) - ?? extractAgyConversationIdFromLog(logFile); + ?? extractAgyConversationIdFromLog(agyLogFile); const updatedHandle: AgySessionHandle = { kind: "agy", ...(capturedConversationId ? { conversationId: capturedConversationId } : {}), @@ -723,3 +1001,11 @@ export async function probeExecutor(kind: ExecutorKind, config: ExecutorConfig = ); } } + +/** + * Derive a concise human-readable activity summary from a Claude Code stream-json line. + * Exported for testing. Returns undefined when the line does not describe observable activity. + */ +export function parseClaudeActivityLine(line: string): string | undefined { + return deriveClaudeActivity(line); +} diff --git a/src/workflow-state.test.ts b/src/workflow-state.test.ts index d10aa7a..1e262af 100644 --- a/src/workflow-state.test.ts +++ b/src/workflow-state.test.ts @@ -262,3 +262,399 @@ echo '{"session_id":"fake-session-123","conversation_id":"fake-conv-456"}' .toBe("initial_continuation"); }); }); + +// --------------------------------------------------------------------------- +// Progress sidecar +// --------------------------------------------------------------------------- + +describe("Progress sidecar", () => { + let tmpDir: string; + + beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "agent-workflow-sidecar-test-")); + }); + + afterEach(() => { + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + it("writes and reads progress sidecar", async () => { + const { writeProgressSidecar, readProgressSidecar } = + await import("./workflow-state.js"); + const sidecar = { + executor: "reasonix" as const, + cycleIndex: 1, + attemptIndex: 0, + phase: "executing", + startedAt: new Date().toISOString(), + lastActivityAt: new Date().toISOString(), + logFilePath: "cycle-1-exec.log", + logBytes: 1024, + activitySummary: "Running build", + updatedAt: new Date().toISOString(), + }; + writeProgressSidecar(tmpDir, sidecar); + const read = readProgressSidecar(tmpDir); + expect(read).toBeDefined(); + expect(read!.executor).toBe("reasonix"); + expect(read!.phase).toBe("executing"); + expect(read!.logBytes).toBe(1024); + expect(read!.activitySummary).toBe("Running build"); + }); + + it("returns undefined when sidecar file does not exist", async () => { + const { readProgressSidecar } = await import("./workflow-state.js"); + expect(readProgressSidecar(tmpDir)).toBeUndefined(); + }); + + it("reads attempt log path for initial attempt and retry", async () => { + const { attemptLogPath } = await import("./workflow-state.js"); + const dir = tmpDir; + expect(path.basename(attemptLogPath(dir, 1, 0))).toBe("cycle-1-exec.log"); + expect(path.basename(attemptLogPath(dir, 2, 0))).toBe("cycle-2-exec.log"); + expect(path.basename(attemptLogPath(dir, 1, 1))).toBe("cycle-1-exec-retry-1.log"); + expect(path.basename(attemptLogPath(dir, 1, 3))).toBe("cycle-1-exec-retry-3.log"); + }); + + it("nextAttemptIndex finds the next free index", async () => { + const { nextAttemptIndex, attemptLogPath } = await import("./workflow-state.js"); + fs.writeFileSync(attemptLogPath(tmpDir, 1, 0), "initial"); + expect(nextAttemptIndex(tmpDir, 1)).toBe(1); + fs.writeFileSync(attemptLogPath(tmpDir, 1, 1), "retry1"); + expect(nextAttemptIndex(tmpDir, 1)).toBe(2); + fs.writeFileSync(attemptLogPath(tmpDir, 1, 2), "retry2"); + expect(nextAttemptIndex(tmpDir, 1)).toBe(3); + // Non-existent cycle returns 0 + expect(nextAttemptIndex(tmpDir, 9)).toBe(0); + }); + + it("sidecar preserves Chinese UTF-8 in activitySummary", async () => { + const { writeProgressSidecar, readProgressSidecar } = + await import("./workflow-state.js"); + writeProgressSidecar(tmpDir, { + executor: "claude", + cycleIndex: 1, + attemptIndex: 0, + phase: "executing", + startedAt: new Date().toISOString(), + lastActivityAt: new Date().toISOString(), + logFilePath: "cycle-1-exec.log", + logBytes: 42, + activitySummary: "读取文件 src/index.ts", + updatedAt: new Date().toISOString(), + }); + const read = readProgressSidecar(tmpDir); + expect(read).toBeDefined(); + expect(read!.activitySummary).toBe("读取文件 src/index.ts"); + }); +}); + +// --------------------------------------------------------------------------- +// Status live fields and sidecar integration +// --------------------------------------------------------------------------- + +describe("Status and sidecar live fields", () => { + let tmpDir: string; + let stateRoot: string; + let repoRoot: string; + let originalAgentDir: string | undefined; + + beforeEach(async () => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "agent-workflow-status-test-")); + repoRoot = path.join(tmpDir, "repo"); + stateRoot = path.join(tmpDir, "workflows"); + fs.mkdirSync(repoRoot, { recursive: true }); + const { execFileSync } = await import("node:child_process"); + 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.writeFileSync(path.join(repoRoot, "test.txt"), "test"); + execFileSync("git", ["add", "."], { cwd: repoRoot }); + execFileSync("git", ["commit", "-m", "init"], { cwd: repoRoot }); + originalAgentDir = process.env.AGENT_WORKFLOW_DIR; + process.env.AGENT_WORKFLOW_DIR = stateRoot; + }); + + afterEach(() => { + if (originalAgentDir === undefined) delete process.env.AGENT_WORKFLOW_DIR; + else process.env.AGENT_WORKFLOW_DIR = originalAgentDir; + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + it("statusWorkflow includes live fields from sidecar when executing", async () => { + const { runDir, writeState, initWorkflowState, gitHead, writeProgressSidecar, readState } + = await import("./workflow-state.js"); + const runId = "status-live-fields-001"; + const dir = runDir(repoRoot, runId); + fs.mkdirSync(dir, { recursive: true }); + const state = initWorkflowState({ + runId, + repoRoot, + baselineHead: gitHead(repoRoot), + plan: { version: "1", title: "Test", planMarkdown: "Test", scope: ["src/"], acceptanceCriteria: ["Ok"], verificationCommands: [["echo"]] }, + executor: "claude", + maxCycles: 3, + }); + state.status = "executing"; + state.currentCycle = 1; + state.cycles = [{ cycleIndex: 1, startedAt: new Date().toISOString() }]; + writeState(dir, state); + + // Write a sidecar as if an adapter was running + writeProgressSidecar(dir, { + executor: "claude", + cycleIndex: 1, + attemptIndex: 0, + pid: 12345, + phase: "executing", + startedAt: new Date(Date.now() - 120_000).toISOString(), // 2 min ago + lastActivityAt: new Date().toISOString(), + logFilePath: "cycle-1-exec.log", + logBytes: 4096, + activitySummary: "Reading src/index.ts", + updatedAt: new Date().toISOString(), + }); + fs.writeFileSync(path.join(dir, "cycle-1-exec.log"), "log content"); + + const { statusWorkflow } = await import("./workflow-engine.js"); + const chunks: string[] = []; + const origWrite = process.stdout.write.bind(process.stdout); + process.stdout.write = (chunk: unknown) => { chunks.push(String(chunk)); return true; }; + try { + statusWorkflow({ runId, json: true }); + } finally { + process.stdout.write = origWrite; + } + const output = JSON.parse(chunks.join("")); + expect(output.liveLogFile).toContain("cycle-1-exec.log"); + expect(output.executionElapsed).toBeDefined(); + expect(output.lastActivity).toBeDefined(); + expect(output.logBytes).toBe(4096); + expect(output.executorPid).toBe(12345); + expect(output.activitySummary).toBe("Reading src/index.ts"); + }); + + it("status live fields absent when no sidecar", async () => { + const { runDir, writeState, initWorkflowState, gitHead } = await import("./workflow-state.js"); + const runId = "status-no-sidecar-002"; + const dir = runDir(repoRoot, runId); + fs.mkdirSync(dir, { recursive: true }); + const state = initWorkflowState({ + runId, + repoRoot, + baselineHead: gitHead(repoRoot), + plan: { version: "1", title: "Test", planMarkdown: "Test", scope: ["src/"], acceptanceCriteria: ["Ok"], verificationCommands: [["echo"]] }, + executor: "reasonix", + maxCycles: 3, + }); + state.status = "completed"; + state.currentCycle = 1; + state.cycles = [{ cycleIndex: 1, startedAt: new Date().toISOString() }]; + writeState(dir, state); + + const { statusWorkflow } = await import("./workflow-engine.js"); + const chunks: string[] = []; + const origWrite = process.stdout.write.bind(process.stdout); + process.stdout.write = (chunk: unknown) => { chunks.push(String(chunk)); return true; }; + try { + statusWorkflow({ runId, json: true }); + } finally { + process.stdout.write = origWrite; + } + const output = JSON.parse(chunks.join("")); + expect(output.liveLogFile).toBeUndefined(); + expect(output.executionElapsed).toBeUndefined(); + expect(output.activitySummary).toBeUndefined(); + }); +}); + +// --------------------------------------------------------------------------- +// logsWorkflow follow mode +// --------------------------------------------------------------------------- + +describe("logsWorkflow", () => { + let tmpDir: string; + let stateRoot: string; + let repoRoot: string; + let originalAgentDir: string | undefined; + + beforeEach(async () => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "agent-workflow-logs-test-")); + repoRoot = path.join(tmpDir, "repo"); + stateRoot = path.join(tmpDir, "workflows"); + fs.mkdirSync(repoRoot, { recursive: true }); + const { execFileSync } = await import("node:child_process"); + 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.writeFileSync(path.join(repoRoot, "test.txt"), "test"); + execFileSync("git", ["add", "."], { cwd: repoRoot }); + execFileSync("git", ["commit", "-m", "init"], { cwd: repoRoot }); + originalAgentDir = process.env.AGENT_WORKFLOW_DIR; + process.env.AGENT_WORKFLOW_DIR = stateRoot; + }); + + afterEach(() => { + if (originalAgentDir === undefined) delete process.env.AGENT_WORKFLOW_DIR; + else process.env.AGENT_WORKFLOW_DIR = originalAgentDir; + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + it("tail mode prints Chinese UTF-8 lines without duplication", async () => { + const { runDir, writeState, initWorkflowState, gitHead, readState } = await import("./workflow-state.js"); + const runId = "logs-tail-utf8-001"; + const dir = runDir(repoRoot, runId); + fs.mkdirSync(dir, { recursive: true }); + const state = initWorkflowState({ + runId, + repoRoot, + baselineHead: gitHead(repoRoot), + plan: { version: "1", title: "Test", planMarkdown: "Test", scope: ["src/"], acceptanceCriteria: ["Ok"], verificationCommands: [["echo"]] }, + executor: "claude", + maxCycles: 3, + }); + state.status = "executing"; + state.currentCycle = 1; + state.cycles = [{ cycleIndex: 1, startedAt: new Date().toISOString(), executorLogFile: "cycle-1-exec.log" }]; + writeState(dir, state); + + // Write Chinese UTF-8 content to the log file + const logContent = "line1\n读取文件 src/index.ts\n日本語\n🚀\nlast line"; + fs.writeFileSync(path.join(dir, "cycle-1-exec.log"), logContent); + + const { logsWorkflow } = await import("./workflow-engine.js"); + const chunks: string[] = []; + const origWrite = process.stdout.write.bind(process.stdout); + process.stdout.write = (chunk: unknown) => { chunks.push(String(chunk)); return true; }; + try { + await logsWorkflow({ runId, follow: false, tail: 100 }); + } finally { + process.stdout.write = origWrite; + } + const output = chunks.join(""); + expect(output).toContain("读取文件 src/index.ts"); + expect(output).toContain("日本語"); + expect(output).toContain("🚀"); + // No duplication + expect(output.split("读取文件 src/index.ts").length - 1).toBe(1); + }); + + it("follow mode exits when state changes from executing to awaiting_review", async () => { + const { runDir, writeState, initWorkflowState, gitHead, readState } = await import("./workflow-state.js"); + const runId = "logs-follow-exit-002"; + const dir = runDir(repoRoot, runId); + fs.mkdirSync(dir, { recursive: true }); + const state = initWorkflowState({ + runId, + repoRoot, + baselineHead: gitHead(repoRoot), + plan: { version: "1", title: "Test", planMarkdown: "Test", scope: ["src/"], acceptanceCriteria: ["Ok"], verificationCommands: [["echo"]] }, + executor: "claude", + maxCycles: 3, + }); + state.status = "executing"; + state.currentCycle = 1; + state.cycles = [{ cycleIndex: 1, startedAt: new Date().toISOString(), executorLogFile: "cycle-1-exec.log" }]; + writeState(dir, state); + fs.writeFileSync(path.join(dir, "cycle-1-exec.log"), "initial log content\n"); + + const { logsWorkflow } = await import("./workflow-engine.js"); + const chunks: string[] = []; + const origWrite = process.stdout.write.bind(process.stdout); + process.stdout.write = (chunk: unknown) => { chunks.push(String(chunk)); return true; }; + + // Start logsWorkflow in follow mode, change state after a short delay + const logsPromise = logsWorkflow({ runId, follow: true, tail: 5 }); + setTimeout(() => { + const curState = readState(dir)!; + curState.status = "awaiting_review"; + writeState(dir, curState); + }, 200); + + try { + await logsPromise; + } finally { + process.stdout.write = origWrite; + } + const output = chunks.join(""); + expect(output).toContain("initial log content"); + }); + + it("follow mode switches to retry log when sidecar changes and emits new content", async () => { + const { runDir, writeState, initWorkflowState, gitHead, writeProgressSidecar, readState } = await import("./workflow-state.js"); + const runId = "logs-rotation-003"; + const dir = runDir(repoRoot, runId); + fs.mkdirSync(dir, { recursive: true }); + const state = initWorkflowState({ + runId, + repoRoot, + baselineHead: gitHead(repoRoot), + plan: { version: "1", title: "Test", planMarkdown: "Test", scope: ["src/"], acceptanceCriteria: ["Ok"], verificationCommands: [["echo"]] }, + executor: "claude", + maxCycles: 3, + }); + state.status = "executing"; + state.currentCycle = 1; + state.cycles = [{ cycleIndex: 1, startedAt: new Date().toISOString(), executorLogFile: "cycle-1-exec.log" }]; + writeState(dir, state); + // Create initial log + fs.writeFileSync(path.join(dir, "cycle-1-exec.log"), "initial log\n"); + // Write a sidecar pointing to the original log + writeProgressSidecar(dir, { + executor: "claude", + cycleIndex: 1, + attemptIndex: 0, + phase: "executing", + startedAt: new Date().toISOString(), + lastActivityAt: new Date().toISOString(), + logFilePath: "cycle-1-exec.log", + logBytes: 12, + activitySummary: "Running", + updatedAt: new Date().toISOString(), + }); + + const { logsWorkflow } = await import("./workflow-engine.js"); + const chunks: string[] = []; + const origWrite = process.stdout.write.bind(process.stdout); + process.stdout.write = (chunk: unknown) => { chunks.push(String(chunk)); return true; }; + + const logsPromise = logsWorkflow({ runId, follow: true, tail: 5 }); + setTimeout(() => { + // Create retry log and update cycle record + fs.writeFileSync(path.join(dir, "cycle-1-exec-retry-1.log"), "retry content\n你好世界\n"); + const curState = readState(dir)!; + const cycle = curState.cycles[0]!; + cycle.executorLogFile = "cycle-1-exec-retry-1.log"; + cycle.executorAttemptLogs = ["cycle-1-exec.log", "cycle-1-exec-retry-1.log"]; + writeState(dir, curState); + // Update sidecar to point to retry log + writeProgressSidecar(dir, { + executor: "claude", + cycleIndex: 1, + attemptIndex: 1, + phase: "executing", + startedAt: new Date().toISOString(), + lastActivityAt: new Date().toISOString(), + logFilePath: "cycle-1-exec-retry-1.log", + logBytes: 30, + activitySummary: "Retrying", + updatedAt: new Date().toISOString(), + }); + }, 200); + setTimeout(() => { + const curState = readState(dir)!; + curState.status = "awaiting_review"; + writeState(dir, curState); + }, 600); + + try { + await logsPromise; + } finally { + process.stdout.write = origWrite; + } + const output = chunks.join(""); + expect(output).toContain("initial log"); + expect(output).toContain("retry content"); + expect(output).toContain("你好世界"); + }); +}); diff --git a/src/workflow-state.ts b/src/workflow-state.ts index 64e4839..f703c97 100644 --- a/src/workflow-state.ts +++ b/src/workflow-state.ts @@ -18,7 +18,7 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { execFileSync, spawnSync } from "node:child_process"; -import type { WorkflowEvent, WorkflowState } from "./workflow-types.js"; +import type { ProgressSidecar, WorkflowEvent, WorkflowState } from "./workflow-types.js"; // --------------------------------------------------------------------------- // Paths @@ -340,3 +340,45 @@ export function findRunDir(runId: string, repoRoot?: string): string | undefined } return undefined; } + +// --------------------------------------------------------------------------- +// Progress sidecar +// --------------------------------------------------------------------------- + +/** Path to the progress sidecar JSON file within a run directory. */ +export function progressSidecarPath(dir: string): string { + return path.join(dir, ".progress.json"); +} + +/** Write the progress sidecar atomically. */ +export function writeProgressSidecar(dir: string, progress: ProgressSidecar): void { + const filePath = progressSidecarPath(dir); + const tmp = `${filePath}.tmp.${process.pid}`; + fs.mkdirSync(dir, { recursive: true }); + fs.writeFileSync(tmp, JSON.stringify(progress), "utf8"); + fs.renameSync(tmp, filePath); +} + +/** Read the progress sidecar. Returns undefined if absent or corrupt. */ +export function readProgressSidecar(dir: string): ProgressSidecar | undefined { + const filePath = progressSidecarPath(dir); + if (!fs.existsSync(filePath)) return undefined; + try { + return JSON.parse(fs.readFileSync(filePath, "utf8")) as ProgressSidecar; + } catch { + return undefined; + } +} + +/** Compute the absolute log path for a given cycle and attempt index. */ +export function attemptLogPath(dir: string, cycleIndex: number, attemptIndex: number): string { + if (attemptIndex === 0) return cycleExecLogFile(dir, cycleIndex); + return path.join(dir, `cycle-${cycleIndex}-exec-retry-${attemptIndex}.log`); +} + +/** Compute the next attempt index for a cycle by inspecting existing log files. */ +export function nextAttemptIndex(dir: string, cycleIndex: number): number { + let idx = 0; + while (fs.existsSync(attemptLogPath(dir, cycleIndex, idx))) idx++; + return idx; +} diff --git a/src/workflow-types.ts b/src/workflow-types.ts index cc36292..1630227 100644 --- a/src/workflow-types.ts +++ b/src/workflow-types.ts @@ -15,6 +15,37 @@ export const EXECUTOR_KINDS = ["reasonix", "claude", "agy"] as const; export type ExecutorKind = (typeof EXECUTOR_KINDS)[number]; +// --------------------------------------------------------------------------- +// Progress sidecar — lightweight real-time observability +// --------------------------------------------------------------------------- + +export interface ProgressSidecar { + executor: ExecutorKind; + cycleIndex: number; + attemptIndex: number; + /** Process ID of the executor when available. */ + pid?: number; + /** Phase: "starting" | "executing" | "completed" | "failed" | "waiting_review" */ + phase: string; + /** ISO timestamp when the executor was started. */ + startedAt: string; + /** ISO timestamp of the most recent log output. */ + lastActivityAt: string; + /** Path to the current attempt log (relative to run dir). */ + logFilePath?: string; + /** Total bytes written to the attempt log. */ + logBytes: number; + /** Concise human-readable summary of current activity. */ + activitySummary: string; + /** ISO timestamp when this sidecar was last updated. */ + updatedAt: string; + /** Duration in ms from start to stop (available after phase is completed/failed). */ + executionDurationMs?: number; +} + +/** Fields that can be updated on a sidecar without reading it first. */ +export type ProgressSidecarUpdate = Partial>; + // --------------------------------------------------------------------------- // Plan (input from Codex) // --------------------------------------------------------------------------- @@ -268,6 +299,18 @@ export interface WorkflowStatusOutput { repoRoot: string; createdAt: string; updatedAt: string; + /** Live log path for the current or most recent attempt. */ + liveLogFile?: string; + /** Elapsed time since the current cycle's executor started (ISO 8601 duration). */ + executionElapsed?: string; + /** ISO timestamp of the most recent executor log output. */ + lastActivity?: string; + /** Total bytes written to the live attempt log. */ + logBytes?: number; + /** Process ID of the active executor, if available. */ + executorPid?: number; + /** Real-time activity summary from the progress sidecar. */ + activitySummary?: string; } // --------------------------------------------------------------------------- @@ -333,6 +376,8 @@ export interface ExecutorStartOpts { plan: WorkflowPlanV1; runDir: string; cycleIndex: number; + /** 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. */ timeoutSeconds?: number; @@ -347,6 +392,8 @@ export interface ExecutorResumeOpts { handle: ExecutorSessionHandle; runDir: string; cycleIndex: number; + /** 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. */ timeoutSeconds?: number;