fix(lanes): auto-detect and report stage for codex lanes (issue 11)

A lane on the 16-node `ship-feature` template showed an untouched pipeline
map and never declared a stage. Four independent causes, fixed at their
shared choke points:

1. Codex emits no hooks, so `touchLaneFromHook` — the only caller of
   `stage-detect.detect()`, and correctly pinned to `claude` — could never
   see a `codex` lane. New `server/lib/codex-ingest.js` tails Codex's own
   rollout journals (`$CODEX_HOME/sessions/**/rollout-*.jsonl`), maps its
   tool calls onto the hook-shaped `{tool_name, tool_input}` the matcher
   already understands, and feeds the existing detect path plus an
   idle/running status mirror under the same guard hooks use. Polled every
   `DASHBOARD_CODEX_POLL_MS` ms (default 15000, `0` disables).
2. `resolveLaneArg` matched on cwd alone, so with a claude and a codex lane
   at one directory (the composite `idx_lanes_cwd_tool`) a `ccam stage`
   call could drive the sibling lane's card. It now resolves by
   `--lane` → `CCAM_LANE_ID` → `--tool`/harness marker → owning lane, and
   errors naming both candidates rather than picking one. Every pane the
   dashboard starts carries `CCAM_LANE_ID` (`tmux new-session -e`, or an
   assignment prefix when typing into a live shell pane).
3. `recordDetection`'s declared-wins veto had no exit: a lane left at the
   pipeline's last node rejected every detection for the rest of its life.
   A declaration on the LAST node no longer vetoes — the pipeline is over
   there, so a later detection is new work. Nothing else is relaxed, and
   detection still never writes `stage` nor renders `done`.
4. The run console pre-filled the pipeline's Claude slash command into
   codex panes, which Codex reads as prose — so the skill that would have
   called `ccam stage` never ran. Claude lanes only now.

A codex lane's *declared* stage still depends on the session calling
`ccam stage`; `AGENTS.md` is the only lever CCAM has, so it now says so.

Verified: server 1345/1345, client 441/441, and end-to-end against a real
codex lane row (detected_stage/detected_signal set, declared stage
untouched, no node rendered done).
This commit is contained in:
2026-08-27 16:38:23 +07:00
parent b1734a492f
commit fd7f57929c
22 changed files with 2343 additions and 35 deletions
+369
View File
@@ -0,0 +1,369 @@
/**
* @file Codex stage ingestion. Codex emits no hooks, so the path that infers a
* lane's stage (server/routes/hooks.js → stage-detect.detect) can never see a
* `codex` lane — and it is deliberately pinned to `claude`, because a Claude
* hook must never touch a codex lane sharing the same directory. Codex's
* on-disk equivalent is the rollout journal it appends live, one file per
* thread:
*
* $CODEX_HOME/sessions/<YYYY>/<MM>/<DD>/rollout-<ts>-<threadId>.jsonl
* line 1: {"type":"session_meta","payload":{"cwd":…,"session_id":…}}
* then: {"type":"response_item","payload":{"type":"custom_tool_call",…}}
*
* This module tails those journals for the cwd of every `codex` lane, maps each
* tool-call payload onto the same `{tool_name, tool_input}` shape a Claude hook
* delivers, and feeds the EXISTING detect() + recordDetection() path — so a
* codex lane paints the same amber "detected" badge under the same rules
* (forward-only, declared-wins) and never renders `done`. Like
* workflow-ingest.js, everything here is fail-safe: a malformed or partial
* journal throws only locally and is skipped, and a poll pass never throws.
*
* Codex 0.149 also mirrors threads into version-suffixed SQLite files
* (`state_5.sqlite`, `thread_history_1.sqlite`), but every thread row still
* points at a live rollout file, so reading the JSONL costs no coupling to a
* schema the next Codex release renames. `rolloutFiles()` is the one function
* to re-point if that ever stops being true.
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
*/
const fs = require("node:fs");
const path = require("node:path");
const os = require("node:os");
// `*** Update File: <path>` lines inside an apply_patch heredoc.
const PATCH_FILE_RE = /^\*\*\* (Add|Update|Delete) File: (.+)$/gm;
const SHELL_NAMES = new Set(["exec", "exec_command", "shell", "local_shell", "container.exec"]);
const CODEX_ACTIVE_MS = 120_000; // a journal written within 2 min means the agent is working
const TAIL_BYTES = 64 * 1024; // first sight of a live file: read only its recent tail
const MAX_FILES = 40; // newest-first cap so a years-deep sessions/ tree can't stall a pass
const MAX_READ_BYTES = 4 * 1024 * 1024; // one pass never buffers more than this per journal
const MAX_DEPTH = 4; // sessions/<YYYY>/<MM>/<DD>/<file>
/** The `cmd` of a shell call, whether it arrived as JSON arguments or an exec script. */
function commandFromPayload(payload, text) {
try {
const parsed = JSON.parse(payload.arguments);
if (parsed && typeof parsed.cmd === "string") return parsed.cmd;
if (parsed && Array.isArray(parsed.command)) return parsed.command.join(" ");
} catch {
/* not JSON — an exec script; the detect rules match inside it verbatim */
}
return text;
}
/**
* Hook-shaped events for one rollout payload. A patch script yields one event
* per touched file (Write for a new file, Edit otherwise — the split the
* pipeline templates' detect rules assume); a shell call yields one Bash event.
* Anything else — reasoning, messages, tool OUTPUT, a web search — yields
* nothing, so an unknown Codex tool can never be mistaken for work.
*
* @param {object|null} payload - One rollout line's `payload` object.
* @returns {Array<{tool_name: string, tool_input: object}>}
*/
function eventsFromPayload(payload) {
if (!payload || typeof payload !== "object") return [];
if (payload.type !== "custom_tool_call" && payload.type !== "function_call") return [];
const name = typeof payload.name === "string" ? payload.name : "";
const text = typeof payload.input === "string" ? payload.input : String(payload.arguments || "");
if (name === "update_plan") return [{ tool_name: "TaskCreate", tool_input: {} }];
if (name === "spawn_agent") {
let taskName = "";
try {
taskName = JSON.parse(payload.arguments).task_name || "";
} catch {
/* encrypted or malformed arguments — the tool name alone still matters */
}
return [{ tool_name: "Agent", tool_input: { subagent_type: taskName } }];
}
if (!SHELL_NAMES.has(name)) return [];
// A patch heredoc reaches us inside a JS string literal, so its line breaks
// are usually a literal backslash-n rather than newlines (real rollouts carry
// both forms in the same script). Unescape before scanning or the file list
// comes back empty and a whole edit looks like a plain shell command.
const patched = [...text.replace(/\\n/g, "\n").matchAll(PATCH_FILE_RE)];
if (patched.length) {
return patched.map(([, verb, file]) => ({
tool_name: verb === "Add" ? "Write" : "Edit",
tool_input: { file_path: file.trim() },
}));
}
return [{ tool_name: "Bash", tool_input: { command: commandFromPayload(payload, text) } }];
}
function codexHome() {
return process.env.CODEX_HOME || path.join(os.homedir(), ".codex");
}
/** Every rollout journal under sessions/, newest mtime first, capped at MAX_FILES. */
function rolloutFiles() {
const out = [];
const walk = (dir, depth) => {
let entries = [];
try {
entries = fs.readdirSync(dir, { withFileTypes: true });
} catch {
return; // absent or unreadable — Codex may simply not be installed
}
for (const entry of entries) {
const full = path.join(dir, entry.name);
if (entry.isDirectory()) {
if (depth < MAX_DEPTH) walk(full, depth + 1);
} else if (entry.isFile() && /^rollout-.*\.jsonl$/.test(entry.name)) {
try {
const stat = fs.statSync(full);
out.push({ path: full, mtime: stat.mtimeMs, size: stat.size });
} catch {
/* vanished mid-scan */
}
}
}
};
walk(path.join(codexHome(), "sessions"), 0);
return out.sort((a, b) => b.mtime - a.mtime).slice(0, MAX_FILES);
}
const cwdCache = new Map(); // rollout path → thread cwd (written once, at thread creation)
/** The thread's working directory, from the session_meta on the file's first line. */
function rolloutCwd(file) {
if (cwdCache.has(file)) return cwdCache.get(file);
let cwd = null;
try {
const fd = fs.openSync(file, "r");
try {
const buf = Buffer.alloc(64 * 1024);
const read = fs.readSync(fd, buf, 0, buf.length, 0);
const head = buf.toString("utf8", 0, read).split("\n")[0];
cwd = JSON.parse(head).payload.cwd || null;
} finally {
fs.closeSync(fd);
}
} catch {
cwd = null;
}
if (typeof cwd !== "string" || !cwd) cwd = null;
cwdCache.set(file, cwd);
return cwd;
}
const offsets = new Map(); // rollout path → bytes already ingested
/** Test seam: forget every journal's ingest position (and cached cwd). */
function __resetOffsets() {
offsets.clear();
cwdCache.clear();
}
// Both maps are keyed by file path, so in a process that runs for months they
// would otherwise hold one dead entry per journal Codex ever wrote. Pruned only
// once well past the per-pass scan size, so ordinary churn keeps its positions
// and a journal that drops out of the scan behaves exactly as it does after a
// server restart: re-seen as new, and read only if it is being written to.
const MAX_TRACKED = MAX_FILES * 4;
function pruneTracking(seen) {
for (const map of [offsets, cwdCache]) {
if (map.size <= MAX_TRACKED) continue;
for (const key of map.keys()) if (!seen.has(key)) map.delete(key);
}
}
/** Read `[from, size)` of a file, or an empty buffer when it can't be read. */
function readRange(file, from, size) {
try {
const fd = fs.openSync(file, "r");
try {
const buf = Buffer.alloc(size - from);
const read = fs.readSync(fd, buf, 0, buf.length, from);
return buf.subarray(0, read);
} finally {
fs.closeSync(fd);
}
} catch {
return Buffer.alloc(0);
}
}
/**
* The COMPLETE lines in `[from, size)`, plus the offset to resume from.
*
* Two fragments have to be handled, and only one of them is dropped:
* - a tail read (`firstSight` with `from > 0`) starts mid-line, so its leading
* fragment is discarded;
* - a read that catches the writer mid-line ends with a fragment, so the
* returned offset stops at the last newline and that fragment is re-read
* whole on the next pass.
*
* Every non-tail read therefore starts exactly on a line boundary, where the
* first line is a WHOLE record. Dropping it there — the obvious "the offset
* always lands mid-line" shortcut — loses every append that arrives one line at
* a time, which is how a journal actually grows.
*
* All of it is done on the BUFFER, never on a decoded string: a tail offset can
* land inside a multi-byte UTF-8 sequence, where the decoder emits a 3-byte
* U+FFFD for 1-2 consumed bytes. Measuring a decoded slice with
* `Buffer.byteLength` would then move the offset off the real line boundary and
* every later record would fail to parse.
*/
const NEWLINE = 0x0a;
function completeLines(file, from, size, firstSight) {
let chunk = readRange(file, from, size);
let start = from;
if (firstSight && from > 0) {
const firstBreak = chunk.indexOf(NEWLINE);
if (firstBreak === -1) return { lines: [], next: size };
start += firstBreak + 1;
chunk = chunk.subarray(firstBreak + 1);
}
const lastBreak = chunk.lastIndexOf(NEWLINE);
if (lastBreak === -1) return { lines: [], next: start };
return {
lines: chunk.subarray(0, lastBreak).toString("utf8").split("\n"),
next: start + lastBreak + 1,
};
}
/**
* One poll pass. Reads only the bytes appended since the last pass; on the
* first sight of a file, only its recent tail, and nothing at all when the file
* is already idle — replaying a thread that finished days ago would paint
* stages from work nobody is doing.
*
* Detection uses the LAST matching event of the batch, mirroring "one hook, one
* detection"; recordDetection is forward-only, so ordering inside a batch can
* never drag a lane backwards.
*
* @param {{broadcast?: (laneId: number) => void}} [options]
* @returns {number} How many lanes actually changed (and were broadcast).
*/
function ingestCodexOnce({ broadcast } = {}) {
const lanesLib = require("./lanes");
const { detect } = require("./stage-detect");
const { getPipeline } = require("./pipelines");
let codexLanes = [];
try {
codexLanes = lanesLib.listLanes().filter((l) => l.tool === "codex");
} catch {
return 0;
}
if (!codexLanes.length) return 0;
const touched = new Set();
const files = rolloutFiles();
pruneTracking(new Set(files.map((f) => f.path)));
for (const file of files) {
const cwd = rolloutCwd(file.path);
if (!cwd) continue;
const lane = codexLanes
.filter((l) => cwd === l.cwd || cwd.startsWith(`${l.cwd}/`))
.sort((a, b) => b.cwd.length - a.cwd.length)[0];
if (!lane) continue;
const firstSight = !offsets.has(file.path);
const active = Date.now() - file.mtime < CODEX_ACTIVE_MS;
let from = firstSight ? Math.max(0, file.size - TAIL_BYTES) : offsets.get(file.path);
// A stored offset past EOF means the file was truncated or a new thread took
// the same path. Restart from a bounded tail, never from 0: reading a
// multi-gigabyte journal whole would allocate it all in one buffer.
let tailOnly = firstSight;
if (from > file.size || file.size - from > MAX_READ_BYTES) {
from = Math.max(0, file.size - TAIL_BYTES);
tailOnly = true;
}
let changed = false;
// A journal seen for the first time while already idle is history, not work:
// note where it ends and read nothing.
if (firstSight && !active) {
offsets.set(file.path, file.size);
} else if (file.size > from) {
const { lines, next } = completeLines(file.path, from, file.size, tailOnly);
offsets.set(file.path, next);
const pipeline = getPipeline(lane.pipeline);
let detection = null;
for (const raw of lines) {
if (!raw.trim()) continue;
let payload = null;
try {
payload = JSON.parse(raw).payload;
} catch {
continue; // not a record this module understands
}
for (const event of eventsFromPayload(payload)) {
const hit = detect(pipeline, event);
if (hit) detection = hit;
}
}
if (detection) {
try {
changed = lanesLib.recordDetection(lane.id, detection).written;
} catch {
/* lane deleted mid-pass, or sqlite busy — an inference is never worth throwing for */
}
}
}
// Status mirror, under the same guard hook ingestion uses: only idle↔running,
// and only for a lane the dashboard did not launch — provisioning and failed
// are lifecycle states this path must never stomp on. `updateLane` filters
// through PATCHABLE, which allows `status` and forbids `tool`.
const next = active ? "running" : "idle";
if (
!lane.run_id &&
(lane.status === "idle" || lane.status === "running") &&
next !== lane.status
) {
try {
lanesLib.updateLane(lane.id, { status: next });
changed = true;
} catch {
/* as above */
}
}
if (changed) touched.add(lane.id);
}
for (const id of touched) {
try {
if (broadcast) broadcast(id);
} catch {
/* a broadcast failure must not abort the pass */
}
}
return touched.size;
}
/**
* Unref'd interval so it never blocks shutdown; disable with
* DASHBOARD_CODEX_POLL_MS=0. Codex appends its journal continuously, so a 15s
* beat keeps a codex card within one node of live without ever re-reading what
* it has already ingested.
*/
function startCodexPoll(broadcast) {
const POLL_MS = process.env.DASHBOARD_CODEX_POLL_MS
? Number(process.env.DASHBOARD_CODEX_POLL_MS)
: 15_000;
if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return;
const timer = setInterval(() => {
try {
ingestCodexOnce({ broadcast });
} catch {
/* a poll pass never throws into the event loop */
}
}, POLL_MS);
if (timer.unref) timer.unref();
}
module.exports = {
eventsFromPayload,
rolloutCwd,
ingestCodexOnce,
startCodexPoll,
__resetOffsets,
};
+12 -2
View File
@@ -327,7 +327,8 @@ function setStage(id, { stage, status, evidence, note, result } = {}) {
* not drag a lane back to `plan`);
* - declared wins: the lane's DECLARED stage index is strictly less than
* the detection's (a lane already declared at `review` ignores an
* `implement` detection).
* `implement` detection) — UNLESS the declaration is the pipeline's last
* node, which means the pipeline is over and a new detection is new work.
* Otherwise touches nothing and reports why: `behind-detected`,
* `behind-declared`, or `unknown-node`.
*
@@ -351,8 +352,17 @@ function recordDetection(id, { nodeId, signal } = {}) {
: phaseIdx(pipeline, lane.detected_stage);
if (nodeIdx <= detectedIdx) return { written: false, reason: "behind-detected" };
// Declared wins, with ONE exception: a declaration sitting on the pipeline's
// LAST node means the pipeline is finished, so a later detection is new work
// rather than a stale rerun of what the claim already covers. Without it a
// lane left at `done` rejects every detection for the rest of its life, which
// reads to a user as "auto-detect is broken". Still paints only the amber
// detected badge — `stage` is untouched here, as everywhere in this function.
const declaredIdx = phaseIdx(pipeline, lane.stage);
if (declaredIdx >= nodeIdx) return { written: false, reason: "behind-declared" };
const declarationIsTerminal = declaredIdx === pipeline.nodes.length - 1;
if (!declarationIsTerminal && declaredIdx >= nodeIdx) {
return { written: false, reason: "behind-declared" };
}
db.prepare(
"UPDATE lanes SET detected_stage = ?, detected_signal = ?, detected_at = ? WHERE id = ?"
+8 -2
View File
@@ -216,8 +216,14 @@ function spawnRun(args) {
});
};
// Every pane the dashboard starts carries its lane id. A `ccam stage` call
// from inside it then resolves the lane exactly, instead of matching on cwd —
// which is ambiguous now that one directory can back both a `claude` lane and
// a `codex` lane (see bin/ccam.js's resolveLaneArg).
const env = { CCAM_LANE_ID: String(laneId) };
if (!tmux.hasSession(id)) {
tmux.newSession({ name: id, cwd, argv });
tmux.newSession({ name: id, cwd, argv, env });
record();
} else if (SHELL_COMMANDS.has(tmux.paneCommand(id) || "")) {
// The session exists but its pane is sitting at a bare shell prompt — a
@@ -226,7 +232,7 @@ function spawnRun(args) {
// `--resume` and an initial prompt would never be typed, while the API
// still answered 200. Run the argv in the pane the user already sees
// instead of erroring or opening a second session.
tmux.sendCommand(id, argv);
tmux.sendCommand(id, argv, env);
record();
}
// Pane is running something (a live `claude`, an editor, a build): adopt
+24 -4
View File
@@ -34,12 +34,24 @@ function hasSession(name) {
}
}
/** `-e NAME=value` pairs for tmux, in a stable order. */
function envFlags(env) {
const flags = [];
for (const [key, value] of Object.entries(env || {})) flags.push("-e", `${key}=${value}`);
return flags;
}
/**
* Create a detached tmux session running `argv` as the pane's command. Throws
* if tmux itself fails to start (caller decides how to surface that).
*
* `env` becomes tmux `-e NAME=value` pairs, which tmux puts in the pane's own
* environment (tmux >= 3.2). The dashboard uses it for CCAM_LANE_ID so a `ccam`
* call from inside the pane knows exactly which lane it belongs to instead of
* guessing from cwd — see resolveLaneArg in bin/ccam.js.
*/
function newSession({ name, cwd, argv }) {
execImpl(["new-session", "-d", "-s", name, "-c", cwd, "--", ...argv]);
function newSession({ name, cwd, argv, env }) {
execImpl(["new-session", "-d", "-s", name, "-c", cwd, ...envFlags(env), "--", ...argv]);
enableMouse(name);
}
@@ -78,9 +90,17 @@ function shellQuote(arg) {
* Type `argv` into an existing session's pane and press Enter. Only ever
* called when the pane sits at a shell prompt (see `paneCommand`); `-l` sends
* the string literally so no character is read as a tmux key name.
*
* `env` rides along as a shell assignment prefix on the SAME line — tmux's own
* `-e` applies only at session creation, and a shell that is already running
* takes its environment from the command line. The name is never quoted (bash
* only recognizes an assignment when the `=` is unquoted); the value is.
*/
function sendCommand(name, argv) {
execImpl(["send-keys", "-t", name, "-l", argv.map(shellQuote).join(" ")]);
function sendCommand(name, argv, env) {
const prefix = Object.entries(env || {})
.map(([key, value]) => `${key}=${shellQuote(value)} `)
.join("");
execImpl(["send-keys", "-t", name, "-l", prefix + argv.map(shellQuote).join(" ")]);
execImpl(["send-keys", "-t", name, "Enter"]);
}