diff --git a/README.md b/README.md index 78b5e1a..0bd9457 100644 --- a/README.md +++ b/README.md @@ -156,6 +156,7 @@ agent-workflow logs --run [--follow] [--tail ] ``` - **`--follow` / `-f`** — continuously follow new output. Automatically switches to new retry attempt logs when a retry begins. +- When `CODEX_THREAD_ID` is set, only one `--follow` process for a given Run is allowed per Codex thread. A duplicate follower is rejected without affecting the executor; stale locks from dead followers are cleaned automatically. Different threads and manual viewers without `CODEX_THREAD_ID` remain allowed. - **`--tail `** — number of lines to show from the end of the current log (default: 50). Examples: diff --git a/skills/agent-workflow-host/SKILL.md b/skills/agent-workflow-host/SKILL.md index c076d3c..77ce8c2 100644 --- a/skills/agent-workflow-host/SKILL.md +++ b/skills/agent-workflow-host/SKILL.md @@ -105,11 +105,13 @@ Executor has been running for ~2m. Current activity: Editing src/parser.ts (PID - 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: +- Invite the user once after the Run starts to watch live output independently; do not repeat this command in every progress update: ``` Watch live: agent-workflow logs --run --follow ``` +Start that `logs --follow` command at most once for a given Run in the current Codex thread. If it reports that a follower is already active, do not retry the same command; use `agent-workflow status --run --json` for progress instead. + `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` diff --git a/src/workflow-engine.ts b/src/workflow-engine.ts index e3ebddb..3d4b303 100644 --- a/src/workflow-engine.ts +++ b/src/workflow-engine.ts @@ -48,9 +48,12 @@ import { findRunDir, generateRunId, initWorkflowState, + acquireLogsFollowLock, + LogsFollowLockError, nowIso, readState, readProgressSidecar, + releaseLogsFollowLock, releaseLock, runDir as makeRunDir, writeState, @@ -1944,70 +1947,89 @@ export async function logsWorkflow(opts: LogsWorkflowOpts): Promise { return; } - // Tail the last N lines - const lines = readTailLines(logPath, opts.tail); - for (const line of lines) { - process.stdout.write(line + "\n"); + let followLockPath: string | undefined; + if (opts.follow) { + const codexThreadId = process.env.CODEX_THREAD_ID?.trim(); + if (codexThreadId) { + try { + followLockPath = acquireLogsFollowLock(dir, codexThreadId); + } catch (err) { + if (err instanceof LogsFollowLockError) { + throw new WorkflowEngineError(err.message, 4); + } + throw err; + } + } } - 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; + try { + // Tail the last N lines + const lines = readTailLines(logPath, opts.tail); + for (const line of lines) { + process.stdout.write(line + "\n"); } - // 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 (!opts.follow) return; - 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; - } + // Follow mode: watch for new data and retry rotations + let currentPath = logPath; + let lastByteOffset = fs.statSync(currentPath).size; - // Drain new bytes since lastByteOffset using byte-accurate reading - try { - const stats = fs.statSync(currentPath); - if (stats.size > lastByteOffset) { + // 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); - lastByteOffset = stats.size; + return; } - } catch { - // File may have been deleted or rotated - } - await sleepMs(500); + // 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); + } + } finally { + if (followLockPath) releaseLogsFollowLock(followLockPath); } } diff --git a/src/workflow-state.test.ts b/src/workflow-state.test.ts index fde0f72..5ada9d8 100644 --- a/src/workflow-state.test.ts +++ b/src/workflow-state.test.ts @@ -737,6 +737,135 @@ describe("logsWorkflow", () => { expect(output).toContain("initial log content"); }); + it("rejects a second follower from the same Codex thread", async () => { + const { runDir, writeState, initWorkflowState, gitHead, readState } = await import("./workflow-state.js"); + const runId = "logs-follow-dedupe-004"; + 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 previousThread = process.env.CODEX_THREAD_ID; + process.env.CODEX_THREAD_ID = "thread-for-follower-dedupe"; + 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 firstFollower = logsWorkflow({ runId, follow: true, tail: 5 }); + try { + await expect(logsWorkflow({ runId, follow: true, tail: 5 })) + .rejects.toThrow(/already active for this Codex thread/); + const curState = readState(dir)!; + curState.status = "awaiting_review"; + writeState(dir, curState); + await firstFollower; + expect(chunks.join("")).toContain("initial log content"); + } finally { + process.stdout.write = origWrite; + if (previousThread === undefined) delete process.env.CODEX_THREAD_ID; + else process.env.CODEX_THREAD_ID = previousThread; + } + }); + + it("allows different Codex threads to hold followers for the same run", async () => { + const { + acquireLogsFollowLock, + logsFollowLockPath, + releaseLogsFollowLock, + runDir, + } = await import("./workflow-state.js"); + const dir = runDir(repoRoot, "logs-follow-different-threads-005"); + fs.mkdirSync(dir, { recursive: true }); + + const first = acquireLogsFollowLock(dir, "thread-a"); + const second = acquireLogsFollowLock(dir, "thread-b"); + try { + expect(first).toBe(logsFollowLockPath(dir, "thread-a")); + expect(second).toBe(logsFollowLockPath(dir, "thread-b")); + expect(first).not.toBe(second); + } finally { + releaseLogsFollowLock(first); + releaseLogsFollowLock(second); + } + expect(fs.existsSync(first)).toBe(false); + expect(fs.existsSync(second)).toBe(false); + }); + + it("reclaims a follower lock whose process is dead", async () => { + const { + acquireLogsFollowLock, + logsFollowLockPath, + releaseLogsFollowLock, + runDir, + } = await import("./workflow-state.js"); + const dir = runDir(repoRoot, "logs-follow-stale-006"); + fs.mkdirSync(dir, { recursive: true }); + const lockPath = logsFollowLockPath(dir, "stale-thread"); + fs.writeFileSync(lockPath, JSON.stringify({ + pid: 999_999_999, + codexThreadId: "stale-thread", + startedAt: new Date(0).toISOString(), + })); + + const acquired = acquireLogsFollowLock(dir, "stale-thread"); + expect(acquired).toBe(lockPath); + expect(JSON.parse(fs.readFileSync(lockPath, "utf8")).pid).toBe(process.pid); + releaseLogsFollowLock(acquired); + expect(fs.existsSync(lockPath)).toBe(false); + }); + + it("keeps manual follow behavior unchanged without CODEX_THREAD_ID", async () => { + const { runDir, writeState, initWorkflowState, gitHead, readState } = await import("./workflow-state.js"); + const runId = "logs-follow-manual-007"; + 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"), "manual viewer\n"); + + const previousThread = process.env.CODEX_THREAD_ID; + delete process.env.CODEX_THREAD_ID; + const { logsWorkflow } = await import("./workflow-engine.js"); + const origWrite = process.stdout.write.bind(process.stdout); + process.stdout.write = () => true; + const first = logsWorkflow({ runId, follow: true, tail: 5 }); + const second = logsWorkflow({ runId, follow: true, tail: 5 }); + setTimeout(() => { + const curState = readState(dir)!; + curState.status = "awaiting_review"; + writeState(dir, curState); + }, 100); + try { + await Promise.all([first, second]); + expect(fs.readdirSync(dir).some((name) => name.startsWith(".logs-follow-"))).toBe(false); + } finally { + process.stdout.write = origWrite; + if (previousThread === undefined) delete process.env.CODEX_THREAD_ID; + else process.env.CODEX_THREAD_ID = previousThread; + } + }); + 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"; diff --git a/src/workflow-state.ts b/src/workflow-state.ts index 015e306..114de45 100644 --- a/src/workflow-state.ts +++ b/src/workflow-state.ts @@ -72,6 +72,101 @@ export function cycleDecisionFile(dir: string, cycleIndex: number): string { return path.join(dir, `cycle-${cycleIndex}-decision.json`); } +/** Runtime lock used to deduplicate logs --follow within one Codex thread. */ +export function logsFollowLockPath(dir: string, codexThreadId: string): string { + const threadHash = crypto.createHash("sha256").update(codexThreadId).digest("hex").slice(0, 16); + return path.join(dir, `.logs-follow-${threadHash}.lock`); +} + +export class LogsFollowLockError extends Error { + constructor(message: string) { + super(message); + this.name = "LogsFollowLockError"; + } +} + +interface LogsFollowLockRecord { + pid: number; + codexThreadId: string; + startedAt: string; +} + +function processIsAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (err) { + return (err as NodeJS.ErrnoException).code === "EPERM"; + } +} + +/** Acquire a per-thread log follower lock, removing locks from dead processes. */ +export function acquireLogsFollowLock(dir: string, codexThreadId: string): string { + const lockPath = logsFollowLockPath(dir, codexThreadId); + fs.mkdirSync(dir, { recursive: true }); + const record: LogsFollowLockRecord = { + pid: process.pid, + codexThreadId, + startedAt: nowIso(), + }; + + for (let attempt = 0; attempt < 2; attempt += 1) { + try { + const fd = fs.openSync(lockPath, "wx", 0o600); + try { + fs.writeFileSync(fd, JSON.stringify(record), "utf8"); + } finally { + fs.closeSync(fd); + } + return lockPath; + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err; + let existing: Partial = {}; + try { + existing = JSON.parse(fs.readFileSync(lockPath, "utf8")) as Partial; + } catch { + // The owner may still be writing a newly created lock. + } + if (typeof existing.pid === "number" && processIsAlive(existing.pid)) { + throw new LogsFollowLockError( + `A logs --follow process is already active for this Codex thread (PID ${existing.pid}). ` + + "Use the existing follower or agent-workflow status instead of starting another one.", + ); + } + if (typeof existing.pid !== "number") { + try { + if (Date.now() - fs.statSync(lockPath).mtimeMs < 5000) { + throw new LogsFollowLockError( + "A logs --follow process is already starting for this Codex thread. " + + "Use agent-workflow status instead of starting another one.", + ); + } + } catch (statErr) { + if (statErr instanceof LogsFollowLockError) throw statErr; + if ((statErr as NodeJS.ErrnoException).code !== "ENOENT") throw statErr; + } + } + try { + fs.unlinkSync(lockPath); + } catch (unlinkErr) { + if ((unlinkErr as NodeJS.ErrnoException).code !== "ENOENT") throw unlinkErr; + } + } + } + + throw new LogsFollowLockError("Could not acquire the logs --follow lock."); +} + +/** Release a follower lock only when it belongs to this process. */ +export function releaseLogsFollowLock(lockPath: string): void { + try { + const record = JSON.parse(fs.readFileSync(lockPath, "utf8")) as Partial; + if (record.pid === process.pid) fs.unlinkSync(lockPath); + } catch { + // The lock may already have been cleaned up after an interrupted process. + } +} + // --------------------------------------------------------------------------- // Run ID // ---------------------------------------------------------------------------