/** * @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 { getPipeline, phaseIdx, 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` and `base_branch` 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. Provisioning writes them through * setProvisioningFacts instead. */ const PATCHABLE = new Set([ "title", "branch", "pipeline", "status", "gate_decision", "ci_status", "needs_action", "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"]); 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" }); } } function hydrate(row) { if (!row) return null; let stages = {}; let links = {}; try { stages = JSON.parse(row.stages || "{}"); } catch { /* corrupt blob -> empty */ } try { links = JSON.parse(row.links || "{}"); } catch { /* corrupt blob -> empty */ } return { ...row, stages, links }; } function createLane({ title = "", cwd, branch = null, pipeline = "default", kind = "adopted", source_repo = null, base_branch = null, slug = null, } = {}) { if (!cwd || typeof cwd !== "string" || !cwd.startsWith("/")) { throw Object.assign(new Error("cwd must be an absolute path"), { code: "EBADCWD" }); } validateKind(kind); const info = db .prepare( "INSERT INTO lanes (title, cwd, branch, pipeline, kind, source_repo, base_branch, slug, stage_since) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)" ) .run( title, cwd.replace(/\/+$/, ""), branch, pipeline, kind, source_repo, base_branch, slug, 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)); } 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); } const cols = []; const vals = []; for (const [k, v] of Object.entries(patch)) { if (!PATCHABLE.has(k)) continue; cols.push(`${k} = ?`); vals.push(k === "links" && 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); } /** * Write provisioning-time facts that `PATCH /api/lanes/:id` must never reach — * today only `base_branch`, resolved after `git worktree add` succeeds. 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(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; } /** * 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. */ 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 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 ); 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). * 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" }; const declaredIdx = phaseIdx(pipeline, lane.stage); if (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); 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) { 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, 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. */ function resolveLaneByCwd(cwd) { if (!cwd || typeof cwd !== "string") return null; const target = cwd.replace(/\/+$/, ""); let best = null; for (const lane of listLanes()) { 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. */ function resolveLaneForWork(sessionCwd, toolInput) { let best = null; for (const candidate of absolutePathsIn(toolInput)) { const lane = resolveLaneByCwd(candidate); if (lane && (!best || lane.cwd.length > best.cwd.length)) best = lane; } return best || resolveLaneByCwd(sessionCwd); } 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 (`ccam 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. */ function withDetected(states, pipeline, lane) { const detectedIdx = phaseIdx(pipeline, lane.detected_stage); if (detectedIdx === -1) return states.map((n) => ({ ...n, detected: false })); const stages = lane.stages || {}; return states.map((n, i) => ({ ...n, detected: i <= detectedIdx && !stages[n.id] && n.state !== "current", })); } function lanePayload(lane, ageSec = null) { const pipeline = getPipeline(lane.pipeline); const since = lane.stage_since ? Date.parse(lane.stage_since) : NaN; return { ...lane, pipeline_name: pipeline.name, pipeline_nodes: withDetected(nodeStates(pipeline, lane), pipeline, lane), progress: progressPct(pipeline, lane), 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; } module.exports = { DEAD_SEC, createLane, listLanes, getLane, updateLane, deleteLane, setStage, recordDetection, clearLane, recoverInterruptedProvisioning, resolveLaneByCwd, resolveLaneForWork, classifyLiveness, lanePayload, purgeCandidateSessions, hasActiveLaneSession, purgeLaneSessions, setProvisioningFacts, };