/** * @file Lane storage and lifecycle. A lane is a durable unit of parallel agent * work — one working directory, many sessions over time — so the dashboard can * show a pipeline that survives session restarts. This module owns every SQL * statement touching the `lanes` table, resolves an incoming hook's `cwd` onto a * lane, records stage transitions (with `stage_since` semantics), and classifies * liveness the way Shipyard does: a silent watcher is dead, a silent idle lane * is merely at rest. * @author Nguyễn Ngọc Trí Vĩ */ const { db } = require("../db"); const { listPipelines, getPipeline, phaseIdx, stageRecords, nodeStates, progressPct, } = require("./pipelines"); const DEAD_SEC = Number(process.env.LANE_DEAD_SEC || 300); /** * How long a detection holds the forward-only floor. * * Five minutes, not thirty: a real session cycles implement -> tests -> ship -> * implement -> tests within one sitting, and a thirty-minute hold pinned the * lane at the furthest stage it ever touched — one push left it reading `ship` * while the agent was demonstrably back to running tests. Five minutes is still * far longer than a burst of tool calls, so the anti-flap property (a Read right * after an Edit must not drag the lane back to `plan`) is unaffected. * * Read per call, not once at load, so a test and an operator can change it * without a restart. */ function detectionTtlMs() { const raw = Number(process.env.DETECTION_TTL_MS); return Number.isFinite(raw) && raw > 0 ? raw : 300_000; } /** True when `iso` is absent, unparseable, or older than the TTL. An unknown * age cannot be proven fresh, so it counts as stale. */ function detectionIsStale(iso) { if (!iso) return true; const at = Date.parse(iso); if (!Number.isFinite(at)) return true; return Date.now() - at > detectionTtlMs(); } /** Stages whose whole job is to wait — silence here means the loop died. */ const WATCH_STAGE_RE = /watch|poll/i; /** * Fields a client may change through `PATCH /api/lanes/:id`. * * `kind`, `source_repo`, `slug`, `base_branch`, `slot` and `ports` are * deliberately ABSENT: they are provisioning-time facts, and `kind` is check 1 of * the destroy guard. A client that could flip `kind` to "managed" at runtime could * point the guard at a directory the user owns; a client that could set `slot` * could move every slot-derived runtime fact (the ports a lane binds, and later * the database name a drop targets) onto another lane's resources. Provisioning * writes them through setProvisioningFacts instead. */ const PATCHABLE = new Set([ "title", "branch", "pipeline", "status", "gate_decision", "ci_status", "needs_action", "needs_action_session", "links", "notes", "session_id", "run_id", ]); /** Provisioning-time facts, writable only by this module's internal setter. */ const PROVISIONING_FIELDS = new Set([ "kind", "source_repo", "base_branch", "slug", "slot", "ports", ]); const nowIso = () => new Date().toISOString(); const VALID_KINDS = new Set(["adopted", "managed"]); function validateKind(kind) { if (!VALID_KINDS.has(kind)) { throw Object.assign(new Error(`unknown kind: ${kind}`), { code: "EBADKIND" }); } } const VALID_TOOLS = new Set(["claude", "codex", "antigravity"]); function validateTool(tool) { if (!VALID_TOOLS.has(tool)) { throw Object.assign(new Error(`unknown tool: ${tool}`), { code: "EBADTOOL" }); } } /** * `getPipeline` falls back to the default template for an unknown id — correct * when READING (a lane must always render something), wrong when WRITING: a * typo'd id would be accepted, stored, and then silently draw the default map * forever. Reject it at the write, where the caller can still be told. */ function validatePipeline(id) { if (!listPipelines().some((p) => p.id === id)) { throw Object.assign(new Error(`unknown pipeline: ${id}`), { code: "EBADPIPELINE" }); } } function hydrate(row) { if (!row) return null; let stages = {}; let links = {}; let ports = {}; try { stages = JSON.parse(row.stages || "{}"); } catch { /* corrupt blob -> empty */ } try { links = JSON.parse(row.links || "{}"); } catch { /* corrupt blob -> empty */ } try { ports = JSON.parse(row.ports || "{}"); } catch { /* corrupt blob -> empty */ } return { ...row, stages, links, ports }; } /** * A lane's display name, as a column value. The title column is NOT NULL TEXT * and the dashboard's rename field writes it straight through, so it is * trimmed and capped — a pasted file is not a lane name. `null` clears the * title rather than violating the NOT NULL, because clearing is legitimate: * every render is `lane.title || lane.cwd`. A non-scalar is bad input and is * rejected the same way every other patch value is (see `bindable`). * * @param {unknown} value - Whatever the caller offered as a title. * @returns {string} A safe column value. * @throws {Error} code `EBADFIELD` when the value is not a scalar. */ const TITLE_MAX = 200; function normalizeTitle(value) { if (value === null || value === undefined) return ""; const t = typeof value; if (t !== "string" && t !== "number" && t !== "bigint") { throw Object.assign(new Error("title must be a string"), { code: "EBADFIELD" }); } return String(value).trim().slice(0, TITLE_MAX); } /** * The path the OS calls this directory: macOS hands a session `/private/tmp/x` * for a typed `/tmp/x`, and a case-insensitive volume accepts `~/Code` for * `~/code`, so a lane stored under the typed string would never bind a * session. A path that does not exist yet (a worktree lane is created before * its directory) keeps the string it was given. */ function canonicalCwd(cwd) { // A relative path must stay relative, so createLane still refuses it. if (!cwd.startsWith("/")) return cwd; const trimmed = cwd.replace(/\/+$/, ""); // A path that does not exist yet resolves through its deepest existing // ancestor, so /link/new and /real/new still name the same lane. const path = require("node:path"); const tail = []; for (let dir = trimmed; ; dir = path.dirname(dir)) { try { return path.join(require("node:fs").realpathSync.native(dir || "/"), ...tail); } catch { if (dir === "/" || !dir) return trimmed; tail.unshift(path.basename(dir)); } } } function createLane({ title = "", cwd, branch = null, pipeline = "default", kind = "adopted", source_repo = null, base_branch = null, slug = null, tool = "claude", router_profile = null, } = {}) { if (!cwd || typeof cwd !== "string" || !cwd.startsWith("/")) { throw Object.assign(new Error("cwd must be an absolute path"), { code: "EBADCWD" }); } validateKind(kind); validatePipeline(pipeline); validateTool(tool); const info = db .prepare( "INSERT INTO lanes (title, cwd, branch, pipeline, kind, source_repo, base_branch, slug, tool, router_profile, stage_since) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" ) .run( normalizeTitle(title), canonicalCwd(cwd), branch, pipeline, kind, source_repo, base_branch, slug, tool, router_profile || null, nowIso() ); return getLane(info.lastInsertRowid); } function listLanes() { return db.prepare("SELECT * FROM lanes ORDER BY id ASC").all().map(hydrate); } function getLane(id) { return hydrate(db.prepare("SELECT * FROM lanes WHERE id = ?").get(id)); } /** * A patch value, as something better-sqlite3 will actually bind. * * It binds numbers, strings, bigints, buffers and null — and THROWS on * anything else, including the `true` and the `{}` a hand-rolled PATCH can * send for any free-text lane column. That throw is indistinguishable from a * server fault by the time it reaches Express, so the route answered 500 to * what is plainly bad input. Rejecting here gives every PATCHABLE field the * one rule, rather than a special case per field the UI happens to expose. * * @param {string} key - The column being patched, for the error message. * @param {unknown} value - Whatever the caller sent for it. * @returns {string|number|bigint|null} A bindable value. * @throws {Error} code `EBADFIELD` when the value is not a scalar. */ function bindable(key, value) { if (key === "title") return normalizeTitle(value); if (value === null || value === undefined) return null; const t = typeof value; if (t === "string" || t === "number" || t === "bigint") return value; throw Object.assign(new Error(`${key} must be a string, a number or null`), { code: "EBADFIELD", }); } function updateLane(id, patch = {}) { // Validate kind before building the UPDATE if it's being set if ("kind" in patch && patch.kind !== null && patch.kind !== undefined) { validateKind(patch.kind); } if ("pipeline" in patch && patch.pipeline !== null && patch.pipeline !== undefined) { validatePipeline(patch.pipeline); } const cols = []; const vals = []; for (const [k, v] of Object.entries(patch)) { if (!PATCHABLE.has(k)) continue; cols.push(`${k} = ?`); if (k === "links" && v !== null && typeof v === "object") vals.push(JSON.stringify(v)); else vals.push(bindable(k, v)); } if (cols.length) { cols.push("updated_at = ?"); vals.push(nowIso(), id); db.prepare(`UPDATE lanes SET ${cols.join(", ")} WHERE id = ?`).run(...vals); } return getLane(id); } /** * Record which AI Router profile a lane runs against. Deliberately its own * function rather than a `PATCHABLE` field: the router is also written into the * lane's config on disk (`server/lib/router-apply.js`), so the column may only * move together with that write — a bare PATCH would leave the row claiming a * router the lane's settings file has never heard of. * * @param {number} id - The lane id. * @param {string|null} profile - A profile name, or null/"native" for the * vendor's own API. */ function setRouterProfile(id, profile) { const value = !profile || profile === "native" ? null : String(profile); db.prepare("UPDATE lanes SET router_profile = ?, updated_at = ? WHERE id = ?").run( value, nowIso(), id ); return getLane(id); } /** * Every lane attached to one AI Router profile. Editing or deleting a profile * has to reach the lanes that copied it (`server/lib/router-apply.js`), and this * is how those lanes are found. * * @param {string} profile - A profile name. `native` matches nothing: it is * stored as NULL, and a lane on the vendor's own API has nothing to update. */ function lanesByRouter(profile) { if (!profile || profile === "native") return []; return db .prepare("SELECT * FROM lanes WHERE router_profile = ? ORDER BY id ASC") .all(String(profile)) .map(hydrate); } /** * Write provisioning-time facts that `PATCH /api/lanes/:id` must never reach: * `base_branch`, resolved after `git worktree add` succeeds, and the runtime * allocation (`slot`, `ports`) the lane earns on its first boot. Server-internal: * no route passes user input here. * * @param {number} id - The lane id. * @param {object} facts - Subset of PROVISIONING_FIELDS to write. */ function setProvisioningFacts(id, facts = {}) { const cols = []; const vals = []; for (const [k, v] of Object.entries(facts)) { if (!PROVISIONING_FIELDS.has(k)) continue; if (k === "kind") validateKind(v); cols.push(`${k} = ?`); vals.push(k === "ports" && typeof v === "object" ? JSON.stringify(v) : v); } if (cols.length) { cols.push("updated_at = ?"); vals.push(nowIso(), id); db.prepare(`UPDATE lanes SET ${cols.join(", ")} WHERE id = ?`).run(...vals); } return getLane(id); } function deleteLane(id) { return db.prepare("DELETE FROM lanes WHERE id = ?").run(id).changes > 0; } /** Slots currently held by a lane, ascending. Input to the slot allocator. */ function usedSlots() { return db .prepare("SELECT slot FROM lanes WHERE slot IS NOT NULL ORDER BY slot ASC") .all() .map((row) => row.slot); } /** * Every port already recorded by another lane. * * A live listener is not the only claim on a port: a lane whose stack is * currently down still owns the number it booted on, and handing that number to * a second lane would make the two fight the moment the first comes back up. * Deriving ports from `base + slot` keeps lanes of ONE repo apart on its own, but * two repos with different `PORT_BASE_*` values can still land on the same * number — so the allocator subtracts this set as well as what is listening. * * @param {number} [excludeLaneId] - Lane being allocated for; its own reservation is not a conflict. * @returns {Set} */ function reservedPorts(excludeLaneId = null) { const taken = new Set(); for (const row of db.prepare("SELECT id, ports FROM lanes").all()) { if (excludeLaneId !== null && Number(row.id) === Number(excludeLaneId)) continue; let ports; try { ports = JSON.parse(row.ports || "{}"); } catch { continue; // corrupt blob claims nothing } for (const port of Object.values(ports)) { if (Number.isInteger(port)) taken.add(port); } } return taken; } /** * Record a stage transition. `stage_since` moves ONLY when the stage value * actually changes, so the UI's time-on-phase is real; a re-report of the same * stage (a heartbeat, an added note) leaves it alone. * * A real transition also clears `detected_stage`/`detected_signal`/`detected_at`. * Inference tracks progress relative to whatever the agent last declared; once * the agent declares again, any older detection is either stale (a prior task's * leftover, e.g. `tests` from earlier work bleeding into a fresh `plan`) or * redundant (the agent's own claim now covers it). Left in place it would both * paint stale progress in the UI AND — because recordDetection is forward-only * — silently reject every real detection for the new stage until the old one * ages past DETECTION_TTL_MS. * * A transition BACKWARDS additionally drops the `stages` records of every node * ahead of the new one, for the same reason and with the same consequence in * the UI. Those records are the previous pass's — a lane that ran through * `done` and is then declared back at `plan` for a new request would otherwise * render `plan` as `current` while `tests`/`review`/`done` still read `done`, * carrying the *old* work's evidence, and `progress` (computed from the * declared index alone) would contradict the map it sits next to. A gate ahead * of where the agent now says it is has not been passed for this pass; it has * to be re-earned. Records under a key matching no node in the pipeline are * left untouched — their position is unknown, so they cannot be proven ahead. */ /** * Append one row to the stage log — the history `lanes.stages` cannot keep, * because that blob is the current pass and gets overwritten and truncated. * * The run a row belongs to is decided here rather than stored on the lane, so * there is one rule and no column to fall out of sync. A new run starts when * the lane is DECLARED backwards (the next feature begins at plan) or when the * last event sat on the pipeline's final node (the previous feature finished). * Any other move continues the run in flight. * * Only a declaration may open a run by going backwards. Detection bounces by * design — one feature reads `implement → tests → implement` as the session * edits, runs the suite and edits again — so counting every backwards detection * as a new run recorded a single feature as five or six of them, and `runs` is * the denominator the Workflows page reads every pass rate against. A pass ends * where the session says it ends. Reaching the final node still closes the run * for a detection too: work after `done` is new work however it was observed. * * Never throws: a lane's stage must still advance if this log cannot be * written — the dashboard losing a statistic beats an agent's pipeline dying. */ function appendStageEvent(lane, nodeId, { source, result, evidence } = {}) { try { const pipeline = getPipeline(lane.pipeline); const last = db .prepare("SELECT node_id, run_seq FROM lane_stage_events WHERE lane_id = ? ORDER BY id DESC") .get(lane.id); let runSeq = 1; if (last) { const lastIdx = phaseIdx(pipeline, last.node_id); const nextIdx = phaseIdx(pipeline, nodeId); const declared = source !== "detected"; const wentBack = declared && lastIdx > -1 && nextIdx > -1 && nextIdx < lastIdx; const finished = lastIdx > -1 && lastIdx === pipeline.nodes.length - 1 && nextIdx !== lastIdx; runSeq = last.run_seq + (wentBack || finished ? 1 : 0); } db.prepare( `INSERT INTO lane_stage_events (lane_id, pipeline, run_seq, node_id, source, result, evidence, at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` ).run( lane.id, lane.pipeline, runSeq, nodeId, source, result || null, evidence ? 1 : 0, nowIso() ); } catch { /* the log is a report, never a gate on the lane's own state */ } } function setStage(id, { stage, status, evidence, note, result } = {}) { const lane = getLane(id); if (!lane) throw Object.assign(new Error(`no lane ${id}`), { code: "ENOLANE" }); const next = stage || lane.stage; const changed = next !== lane.stage; const stages = { ...lane.stages }; const pipeline = getPipeline(lane.pipeline); const nextIdx = phaseIdx(pipeline, next); const wentBack = changed && nextIdx > -1 && nextIdx < phaseIdx(pipeline, lane.stage); if (wentBack) { for (const key of Object.keys(stages)) { const idx = phaseIdx(pipeline, key); if (idx > nextIdx) delete stages[key]; } } const prev = stages[next] || {}; stages[next] = { enteredAt: !changed && prev.enteredAt ? prev.enteredAt : nowIso(), evidence: evidence !== undefined ? evidence : prev.evidence || null, result: result !== undefined ? result : prev.result || null, }; db.prepare( `UPDATE lanes SET stage = ?, stage_since = ?, status = ?, stages = ?, notes = ?, updated_at = ?${ changed ? ", detected_stage = NULL, detected_signal = NULL, detected_at = NULL" : "" } WHERE id = ?` ).run( next, changed ? nowIso() : lane.stage_since || nowIso(), status || lane.status, JSON.stringify(stages), note !== undefined ? note : lane.notes, nowIso(), id ); appendStageEvent(lane, next, { source: "declared", result: stages[next].result, evidence: stages[next].evidence, }); return getLane(id); } /** * Record an inferred stage from the hook stream. Inference is never evidence * — this writes only `detected_stage`/`detected_signal`/`detected_at`, never * `stage` (the declared stage), so a lane's declared meaning never changes. * * Writes only when BOTH hold: * - forward-only: the detection's node index is strictly greater than the * current `detected_stage`'s index (reading a file after editing it must * 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) — 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`. * * @param {number} id - The lane id. * @param {{nodeId: string, signal: string}} detection - From stage-detect.detect(). * @returns {{written: boolean, reason?: string}} */ function recordDetection(id, { nodeId, signal } = {}) { const lane = getLane(id); if (!lane) throw Object.assign(new Error(`no lane ${id}`), { code: "ENOLANE" }); const pipeline = getPipeline(lane.pipeline); const nodeIdx = phaseIdx(pipeline, nodeId); if (nodeIdx === -1) return { written: false, reason: "unknown-node" }; // Forward-only holds only while the standing detection is fresh. Once it has // aged past the TTL the agent has almost certainly moved on to different // work, so a stale `ship` must not pin the lane forever. Declared-wins below // is NOT relaxed by staleness - an agent's own claim never expires. const detectedIdx = detectionIsStale(lane.detected_at) ? -1 : 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); 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 = ?" ).run(nodeId, signal || null, nowIso(), id); appendStageEvent(lane, nodeId, { source: "detected" }); return { written: true }; } /** * Reset a lane to a blank slate. The detection columns are cleared with the * declared ones on purpose: a kept `detected_stage` would both paint inferred * progress for a tree where nothing has happened AND permanently kill detection * for that lane, because recordDetection is forward-only — a stale `ship` can * never be advanced past. */ function clearLane(id) { // Opt-in: only a lane that has activated a feature has anything to archive. // Requiring the module here (not at file top) avoids a require cycle — // lane-features.js itself requires this file for lanesLib.getLane/setStage. require("./lane-features").archiveActiveFeature(id); db.prepare( `UPDATE lanes SET stage = 'idle', stage_since = ?, status = 'idle', gate_decision = NULL, ci_status = NULL, needs_action = NULL, stages = '{}', notes = NULL, run_id = NULL, detected_stage = NULL, detected_signal = NULL, detected_at = NULL, active_feature_id = NULL, updated_at = ? WHERE id = ?` ).run(nowIso(), nowIso(), id); return getLane(id); } /** * A provisioning task exists only in the server process that created it. On a * new boot, any lane still marked provisioning was interrupted before it could * report a terminal result, so expose it as a removable failure instead. * * @returns {number} Number of interrupted lanes recovered. */ function recoverInterruptedProvisioning() { return db .prepare( "UPDATE lanes SET status = 'failed', notes = ?, updated_at = ? WHERE status = 'provisioning'" ) .run("Provisioning was interrupted by a server restart.", nowIso()).changes; } /** * Longest path-boundary prefix match. `/tmp/wt` must NOT capture * `/tmp/wt-sibling`, and a nested lane must beat its parent. * * `tool`, when given, narrows the candidates to that lane's tool before the * prefix comparison — needed now that one cwd can back both a `claude` lane * and a `codex` lane (see `idx_lanes_cwd_tool`). Omitted, it matches any tool, * which stays correct for callers that already know they hold the lane's own * cwd (there's only ever one match for that exact path per tool, and this * caller shape doesn't care which). */ function resolveLaneByCwd(cwd, tool) { if (!cwd || typeof cwd !== "string") return null; const target = cwd.replace(/\/+$/, ""); let best = null; for (const lane of listLanes()) { if (tool && lane.tool !== tool) continue; const base = lane.cwd.replace(/\/+$/, ""); if (target === base || target.startsWith(`${base}/`)) { if (!best || base.length > best.cwd.length) best = lane; } } return best; } /** Absolute-looking path tokens a tool's input mentions. Bash carries them * inside `command` (`cd /path && ...`), editors carry one in `file_path`. */ function absolutePathsIn(toolInput) { if (!toolInput || typeof toolInput !== "object") return []; const out = []; for (const key of ["file_path", "path", "command", "notebook_path"]) { const value = toolInput[key]; if (typeof value !== "string" || !value) continue; // Quotes and shell operators are separators, not part of a path. for (const token of value.split(/[\s'"`;&|()<>]+/)) { if (token.startsWith("/") && token.length > 1) out.push(token); } } return out; } /** * Which lane should be credited for a tool event. * * A hook's `cwd` is the SESSION's directory, not the directory the command * actually ran in. Measured on a real install: 325 of 400 events carried the * session's cwd while the edits and test runs happened in another repo reached * with `cd && ...`, so the lane doing the work detected nothing and the * lane the terminal started in absorbed all of it. * * So prefer a lane named by the tool's own input — the file being edited, the * directory a command changed into — and fall back to the session's cwd when * the input names no other lane. Deepest match wins, same as resolveLaneByCwd. * * Only stage inference uses this. `session_id` and `needs_action` stay on the * session's own lane, because those genuinely are session-scoped facts. * * Hooks-only, and hooks are only ever emitted by the Claude Code CLI (Codex * has no equivalent hook mechanism here) — so both lookups are pinned to the * `claude` lane at each path, never the `codex` one that might share it. */ function resolveLaneForWork(sessionCwd, toolInput) { let best = null; for (const candidate of absolutePathsIn(toolInput)) { const lane = resolveLaneByCwd(candidate, "claude"); if (lane && (!best || lane.cwd.length > best.cwd.length)) best = lane; } return best || resolveLaneByCwd(sessionCwd, "claude"); } function classifyLiveness({ status, stage, ageSec }, deadSec = DEAD_SEC) { const expectLive = status === "running" || status === "provisioning" || WATCH_STAGE_RE.test(stage || ""); if (!expectLive) return "idle"; if (ageSec !== null && ageSec !== undefined && ageSec > deadSec) return "dead"; return "active"; } /** * Annotate nodeStates() with `detected: boolean` — true for the detected node * itself and for any node before it that carries no declaration. Never flips * a node to `done`: detection only ever adds this flag alongside whatever * state nodeStates() already computed from the declared stage, which is the * only path to `done`. * * The `current` node is never flagged, even when it has no `stages` entry under * its own id: declaring by ALIAS (`agenthub stage coding` → the `implement` node) * keys `stages` by the raw declared string, so the node the agent says it is on * would otherwise render as an inference instead of the blue `current` ring. * Past nodes go through `stageRecords`, which resolves those alias keys — a * node the agent DECLARED must never be painted as merely detected. * * A detection older than the TTL is treated the same as no detection: the TTL * in recordDetection only stops it from holding the forward-only floor for * FUTURE writes, it never clears the row, so without this check a lane that * went quiet would keep painting amber nodes from activity that is no longer * current — the exact stale-progress illusion detection exists to avoid. */ function withDetected(states, pipeline, lane) { if (detectionIsStale(lane.detected_at)) return states.map((n) => ({ ...n, detected: false })); const detectedIdx = phaseIdx(pipeline, lane.detected_stage); if (detectedIdx === -1) return states.map((n) => ({ ...n, detected: false })); const records = stageRecords(pipeline, lane.stages); return states.map((n, i) => ({ ...n, detected: i <= detectedIdx && !records.has(i) && n.state !== "current", })); } /** * The inferred stage a lane should be RENDERED at, or null when the lane has a * declaration of its own to render. * * A lane whose sessions never call `agenthub stage` — any repo that isn't wired to * AgentHub — has no declaration at all: `stage` sits on the DB sentinel `idle` and * `stages` stays `{}` forever, so its pipeline map drew 100% pending while * detection was quietly tracking every node the agent reached. Such a lane is * rendered from its detection instead. * * DISPLAY only, and inference still never becomes evidence: `stage` in the DB * is untouched (recordDetection remains the only writer of `detected_*`), the * payload's own `stage` field still reports the real sentinel, and every node * this paints is flagged `detected` — amber — so none of them can read as * `done`. The TTL is deliberately not applied here: for a lane with nothing * else to show, the last thing detected IS its last known progress, exactly * like a declared stage that also never expires. */ function inferredDisplayStage(lane) { if (!lane.detected_stage) return null; if (lane.stage && lane.stage !== "idle") return null; if (Object.keys(lane.stages || {}).length) return null; return lane.detected_stage; } function lanePayload(lane, ageSec = null) { const pipeline = getPipeline(lane.pipeline); const since = lane.stage_since ? Date.parse(lane.stage_since) : NaN; const inferred = inferredDisplayStage(lane); const inferredIdx = inferred ? phaseIdx(pipeline, inferred) : -1; const view = inferredIdx === -1 ? lane : { ...lane, stage: inferred }; const nodes = nodeStates(pipeline, view); return { ...lane, pipeline_name: pipeline.name, pipeline_skill: pipeline.skill || "", pipeline_nodes: inferredIdx === -1 ? withDetected(nodes, pipeline, lane) : nodes.map((n, i) => ({ ...n, detected: i <= inferredIdx })), progress: progressPct(pipeline, view), stage_seconds: Number.isNaN(since) ? null : Math.max(0, Math.round((Date.now() - since) / 1000)), last_event_seconds: ageSec, liveness: classifyLiveness({ status: lane.status, stage: lane.stage, ageSec }, DEAD_SEC), }; } /** * Build the LIKE pattern matching a lane's subdirectories, escaping the * characters LIKE treats as wildcards. * * CRITICAL: `_` is a single-character wildcard, and every managed lane directory * is named `__` — two literal underscores. Unescaped, a lane at * `/root/myrepo__feat-foo` also matched `/root/myrepoXXfeat-foo`, so a purge * deleted a sibling directory's sessions and the preflight count reported the * victims too: the confirmation was consistently wrong rather than detectably * wrong. `\` and `%` are escaped for the same reason. */ const SUBDIR_LIKE_ESCAPE = "\\"; function subdirLikePattern(cwd) { return `${cwd.replace(/[\\%_]/g, `${SUBDIR_LIKE_ESCAPE}$&`)}/%`; } /** * Find sessions that belong to a lane and may be purged: exact or subdirectory, * excluding the lane's bound session and any active sessions. Shared between * purgeLaneSessions (the deleter) and preflight counting, so the confirmation * dialog's numbers match what actually gets deleted. */ function purgeCandidateSessions(lane) { return db .prepare( `SELECT id FROM sessions WHERE (cwd = ? OR cwd LIKE ? ESCAPE '${SUBDIR_LIKE_ESCAPE}') AND id != ? AND status != 'active'` ) .all(lane.cwd, subdirLikePattern(lane.cwd), lane.session_id || ""); } /** * True when a lane owns at least one still-active session, which purge always * spares. Lives here beside purgeCandidateSessions so both derive their path * matching from the one escaped helper — preflight used to hand-write this * clause and inherited the unescaped-`_` bug with it. */ function hasActiveLaneSession(lane) { const row = db .prepare( `SELECT COUNT(*) AS count FROM sessions WHERE (cwd = ? OR cwd LIKE ? ESCAPE '${SUBDIR_LIKE_ESCAPE}') AND id != ? AND status = 'active'` ) .get(lane.cwd, subdirLikePattern(lane.cwd), lane.session_id || ""); return Boolean(row && row.count > 0); } /** * Delete all sessions associated with a lane, except the one bound to the lane * itself (lanes.session_id) and any active sessions. Deletes their events and * orphaned token_usage rows explicitly (token_usage has no FK to cascade). * Runs in a single transaction; on completion, runs db.pragma("optimize") * to update query statistics (never VACUUM, which locks the database). * * @param {number} laneId - The lane ID. * @returns {{sessions: number, events: number, tokenRows: number}} Count of deleted rows. */ function purgeLaneSessions(laneId) { const lane = getLane(laneId); if (!lane) throw Object.assign(new Error(`no lane ${laneId}`), { code: "ENOLANE" }); const result = { sessions: 0, events: 0, tokenRows: 0 }; db.transaction(() => { // Select sessions matching the lane's cwd (exact or subdir), excluding the // lane's bound session and any active sessions. const sessionsToDelete = purgeCandidateSessions(lane); // Delete events for those sessions result.events = db .prepare( `DELETE FROM events WHERE session_id IN (${sessionsToDelete.map(() => "?").join(",")})` ) .run(...sessionsToDelete.map((s) => s.id)).changes; // Delete orphaned token_usage rows (token_usage has no FK, so it won't cascade) result.tokenRows = db .prepare( `DELETE FROM token_usage WHERE session_id IN (${sessionsToDelete.map(() => "?").join(",")})` ) .run(...sessionsToDelete.map((s) => s.id)).changes; // Delete the sessions themselves result.sessions = db .prepare(`DELETE FROM sessions WHERE id IN (${sessionsToDelete.map(() => "?").join(",")})`) .run(...sessionsToDelete.map((s) => s.id)).changes; })(); db.pragma("optimize"); return result; } /** * Codex lanes carry no pipeline, but older builds inferred and advanced one for * them. Wipe what those builds left — the stage columns, the stage log and any * feature records — so nothing stale survives into the Workflows statistics or * resurfaces if a view ever reads it. Idempotent; run at every server start. * * @returns {number} how many codex lanes still carried pipeline data */ function clearCodexPipelines() { const CODEX = "SELECT id FROM lanes WHERE tool = 'codex'"; let changed = 0; db.transaction(() => { changed = db .prepare( `UPDATE lanes SET stage = 'idle', stage_since = NULL, stages = '{}', detected_stage = NULL, detected_signal = NULL, detected_at = NULL WHERE tool = 'codex' AND (stage <> 'idle' OR stages <> '{}' OR detected_stage IS NOT NULL OR id IN (SELECT lane_id FROM lane_stage_events) OR id IN (SELECT lane_id FROM lane_features))` ) .run().changes; db.prepare(`DELETE FROM lane_stage_events WHERE lane_id IN (${CODEX})`).run(); db.prepare(`DELETE FROM lane_features WHERE lane_id IN (${CODEX})`).run(); })(); return changed; } module.exports = { DEAD_SEC, VALID_TOOLS, detectionIsStale, canonicalCwd, createLane, listLanes, getLane, updateLane, deleteLane, setStage, clearCodexPipelines, setRouterProfile, lanesByRouter, recordDetection, clearLane, recoverInterruptedProvisioning, resolveLaneByCwd, resolveLaneForWork, classifyLiveness, lanePayload, purgeCandidateSessions, hasActiveLaneSession, purgeLaneSessions, setProvisioningFacts, usedSlots, reservedPorts, };