fix: deduplicate Codex log followers

README.md

@@ -156,6 +156,7 @@ agent-workflow logs --run <run-id> [--follow] [--tail <lines>]
 ```
 
 - **`--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 <lines>`** — number of lines to show from the end of the current log (default: 50).
 
 Examples:

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 <run-id> --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 <run-id> --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`

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<void> {
     return;
   }
 
-  // Tail the last N lines
+  let followLockPath: string | undefined;
-  const lines = readTailLines(logPath, opts.tail);
+  if (opts.follow) {
-  for (const line of lines) {
+    const codexThreadId = process.env.CODEX_THREAD_ID?.trim();
-    process.stdout.write(line + "\n");
+    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;
+  try {
-
+    // Tail the last N lines
-  // Follow mode: watch for new data and retry rotations
+    const lines = readTailLines(logPath, opts.tail);
-  let currentPath = logPath;
+    for (const line of lines) {
-  let lastByteOffset = fs.statSync(currentPath).size;
+      process.stdout.write(line + "\n");
-
-  // 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
+    if (!opts.follow) return;
-    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) {
+    // Follow mode: watch for new data and retry rotations
-      drainBytes(currentPath, lastByteOffset);
+    let currentPath = logPath;
-      currentPath = candidatePath;
+    let lastByteOffset = fs.statSync(currentPath).size;
-      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
+    // States that indicate execution has ceased (no more output expected)
-    try {
+    const terminalStates = new Set([
-      const stats = fs.statSync(currentPath);
+      "completed", "needs_human", "blocked", "budget_exhausted", "aborted",
-      if (stats.size > lastByteOffset) {
+      "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);
   }
 }
 

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";

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<LogsFollowLockRecord> = {};
+      try {
+        existing = JSON.parse(fs.readFileSync(lockPath, "utf8")) as Partial<LogsFollowLockRecord>;
+      } 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<LogsFollowLockRecord>;
+    if (record.pid === process.pid) fs.unlinkSync(lockPath);
+  } catch {
+    // The lock may already have been cleaned up after an interrupted process.
+  }
+}
+
 // ---------------------------------------------------------------------------
 // Run ID
 // ---------------------------------------------------------------------------