/** * @file Express router for lanes — the durable per-working-directory unit of * parallel agent work. Read endpoints join each lane with its most recent event * timestamp so liveness can be computed without a separate heartbeat, and every * mutation re-broadcasts the lane over the existing WebSocket as `lane_update`. * Orchestration is deliberately absent: the driving Claude session declares its * own stage (`POST /:id/stage`); the dashboard never guesses a transition. * @author Nguyễn Ngọc Trí Vĩ */ const { Router } = require("express"); const fs = require("node:fs"); const path = require("node:path"); const { db } = require("../db"); const lanesLib = require("../lib/lanes"); const { listPipelines } = require("../lib/pipelines"); const { broadcast } = require("../websocket"); const runs = require("../lib/run-spawner"); const { sameOriginGuard } = require("./run"); const { preflight } = require("../lib/lane-preflight"); const { LANES_ROOT, addWorktree, gitFacts, isGitRepo, listBranches, removeWorktree, resetWorktree, resolveBase, slugify, } = require("../lib/worktree"); const { withLaneLock } = require("../lib/lane-lock"); const router = Router(); const MAX_WORKTREE_DIRECTORY_ATTEMPTS = 50; /** Seconds since this lane's session last emitted an event; null if never. */ function lastEventAge(lane) { if (!lane.session_id) return null; const row = db .prepare("SELECT MAX(created_at) AS last FROM events WHERE session_id = ?") .get(lane.session_id); if (!row || !row.last) return null; const t = Date.parse(row.last); return Number.isNaN(t) ? null : Math.max(0, Math.round((Date.now() - t) / 1000)); } function payload(lane) { return lanesLib.lanePayload(lane, lastEventAge(lane)); } /** Push the current state of one lane to every connected client. */ function broadcastLane(id) { const lane = lanesLib.getLane(id); if (lane) broadcast("lane_update", { lane: payload(lane) }); } /** * Release the lane holding a run that has just finished. Registered as a * callback because the spawner must not require this router back: it is * already required FROM here, and broadcastLane needs this file's payload(). * * No lane lock: the read, the guard and the write are one synchronous * better-sqlite3 sequence with no `await` between them, so nothing can * interleave. Matching run_id is what keeps a lane that has already moved on to * a different run untouched. */ runs.setRunExitHandler(({ runId }) => { const lane = lanesLib.listLanes().find((l) => l.run_id === runId); if (!lane) return; lanesLib.updateLane(lane.id, { run_id: null, status: "idle" }); broadcastLane(lane.id); }); router.get("/", (_req, res) => { const lanes = lanesLib.listLanes().map(payload); res.json({ lanes, counts: { total: lanes.length, running: lanes.filter((l) => l.status === "running").length, needs_you: lanes.filter((l) => l.needs_action).length, dead: lanes.filter((l) => l.liveness === "dead").length, }, }); }); // Registered before "/:id" so the literal path is not swallowed by the param. router.get("/pipelines", (_req, res) => res.json({ pipelines: listPipelines() })); /** * Local branches of a candidate source repo, for the "Add lane" picker: pick * a repo, then pick which branch to fork the new worktree from, instead of * typing a branch name and hoping it exists. Same validation as `/worktree` * (below), since a repo this can't resolve branches for can't be provisioned * from either. Read-only, so no same-origin guard. */ router.get("/branches", async (req, res) => { const sourceRepo = typeof req.query.repo === "string" ? req.query.repo : ""; if (!sourceRepo || !path.isAbsolute(sourceRepo) || !fs.existsSync(sourceRepo)) { return res.status(400).json({ error: { code: "EBADSOURCEREPO", message: "repo must be an existing absolute path" }, }); } if (!(await isGitRepo(sourceRepo))) { return res.status(400).json({ error: { code: "EBADSOURCEREPO", message: "repo is not a git repository" }, }); } try { const { branches, current } = await listBranches(sourceRepo); res.json({ branches, current }); } catch (err) { res.status(500).json({ error: { code: err.code, message: err.message, git: err.git } }); } }); /** * Idempotent "which lane owns this directory?" — the Workspace page opens on a * cwd, not on a lane id, so it needs one lane to exist for that cwd without * ever creating a duplicate. Also registered before "/:id". */ router.post("/ensure", sameOriginGuard, (req, res) => { const body = req.body || {}; const owner = lanesLib.resolveLaneByCwd(body.cwd); if (owner) return res.json({ lane: payload(owner), created: false }); try { const lane = lanesLib.createLane({ cwd: body.cwd, title: body.title || "" }); broadcastLane(lane.id); return res.status(201).json({ lane: payload(lane), created: true }); } catch (err) { if (err.code === "EBADCWD") { return res.status(400).json({ error: { code: err.code, message: err.message } }); } // The cwd UNIQUE constraint is the arbiter: someone else won the race, so // re-read and return THEIR lane rather than reporting a conflict. if (err.code === "SQLITE_CONSTRAINT_UNIQUE" || String(err.message).includes("UNIQUE")) { const winner = lanesLib.resolveLaneByCwd(body.cwd); if (winner) return res.json({ lane: payload(winner), created: false }); } return res.status(500).json({ error: { code: err.code, message: err.message } }); } }); router.get("/:id", (req, res) => { const lane = lanesLib.getLane(req.params.id); if (!lane) return res.status(404).json({ error: { code: "ENOLANE", message: "lane not found" } }); res.json({ lane: payload(lane) }); }); router.post("/", sameOriginGuard, (req, res) => { try { const lane = lanesLib.createLane(req.body || {}); broadcastLane(lane.id); res.status(201).json({ lane: payload(lane) }); } catch (err) { if (err.code === "EBADCWD") { return res.status(400).json({ error: { code: err.code, message: err.message } }); } if (err.code === "SQLITE_CONSTRAINT_UNIQUE" || String(err.message).includes("UNIQUE")) { return res .status(409) .json({ error: { code: "EDUPCWD", message: "a lane already owns that cwd" } }); } res.status(500).json({ error: { message: err.message } }); } }); router.patch("/:id", sameOriginGuard, (req, res) => { if (!lanesLib.getLane(req.params.id)) { return res.status(404).json({ error: { code: "ENOLANE", message: "lane not found" } }); } let lane; try { lane = lanesLib.updateLane(req.params.id, req.body || {}); } catch (err) { // A bad `kind` is invalid input, not a server fault — every sibling route // answers 400 here, so this one must too instead of throwing into Express. if (err.code === "EBADKIND") { return res.status(400).json({ error: { code: err.code, message: err.message } }); } return res.status(500).json({ error: { code: err.code, message: err.message } }); } broadcastLane(lane.id); res.json({ lane: payload(lane) }); }); router.post("/:id/stage", sameOriginGuard, (req, res) => { if (!lanesLib.getLane(req.params.id)) { return res.status(404).json({ error: { code: "ENOLANE", message: "lane not found" } }); } const lane = lanesLib.setStage(req.params.id, req.body || {}); broadcastLane(lane.id); res.json({ lane: payload(lane) }); }); router.get("/:id/preflight", async (req, res) => { const lane = lanesLib.getLane(req.params.id); if (!lane) return res.status(404).json({ error: { code: "ENOLANE", message: "lane not found" } }); const action = String(req.query.action || ""); if (!["reset", "remove", "purge"].includes(action)) { return res .status(400) .json({ error: { code: "EBADACTION", message: `unknown action ${action}` } }); } try { res.json(await preflight(lane, action)); } catch (err) { res.status(500).json({ error: { code: err.code, message: err.message } }); } }); /** * A lane's working-copy facts: branch, short HEAD, that commit's subject, and * the uncommitted counts. Read-only, so no same-origin guard — that guard * exists for the destructive actions. * * Deliberately NOT part of `GET /api/lanes`: this shells out to git three * times, and that payload is polled and re-broadcast on every hook-driven * lane_update. Any failure — no such directory, not a repo, git itself * erroring — is reported as `available: false` rather than a 500, because a * lane pointing at a plain directory is a normal state, not a fault. * * The `/:id/:action` catch-all below cannot shadow this one — that route is a * POST and Express matches on method as well as path. Verified by moving this * registration after it: the suite stayed green. */ router.get("/:id/git", async (req, res) => { const lane = lanesLib.getLane(req.params.id); if (!lane) return res.status(404).json({ error: { code: "ENOLANE", message: "lane not found" } }); try { res.json({ available: true, ...(await gitFacts(lane.cwd)) }); } catch { res.json({ available: false }); } }); /** * Create a dashboard-managed worktree lane. Git work happens after the 202 * response because provisioning a large repository can take seconds. */ router.post("/worktree", sameOriginGuard, async (req, res) => { const body = req.body || {}; const sourceRepo = body.sourceRepo; if ( typeof sourceRepo !== "string" || !path.isAbsolute(sourceRepo) || !fs.existsSync(sourceRepo) || !(await isGitRepo(sourceRepo)) ) { return res.status(400).json({ error: { code: "EBADSOURCEREPO", message: "sourceRepo must be an existing absolute git repository", }, }); } let slug; try { slug = slugify(body.slug || body.title); } catch (err) { return res.status(400).json({ error: { code: err.code, message: err.message } }); } const resolvedSourceRepo = path.resolve(sourceRepo); const repoName = path.basename(resolvedSourceRepo); const originalSlug = slug; let dir = path.join(LANES_ROOT, `${repoName}__${slug}`); let attempts = 0; while (fs.existsSync(dir) && attempts < MAX_WORKTREE_DIRECTORY_ATTEMPTS) { attempts += 1; const suffix = attempts + 1; slug = `${originalSlug}-${suffix}`; dir = path.join(LANES_ROOT, `${repoName}__${slug}`); } if (fs.existsSync(dir)) { return res.status(409).json({ error: { code: "EWORKTREEDIRCOLLISION", message: "could not allocate a unique worktree directory after 50 attempts", }, }); } const branchPrefix = process.env.LANE_BRANCH_PREFIX || "feat/"; const branch = `${branchPrefix}${slug}`; let lane; try { lane = lanesLib.createLane({ title: body.title || "", cwd: dir, branch, kind: "managed", source_repo: resolvedSourceRepo, base_branch: body.base || null, slug, }); lane = lanesLib.updateLane(lane.id, { status: "provisioning" }); } catch (err) { if (err.code === "SQLITE_CONSTRAINT_UNIQUE" || String(err.message).includes("UNIQUE")) { return res .status(409) .json({ error: { code: "EDUPCWD", message: "a lane already owns that cwd" } }); } return res.status(500).json({ error: { code: err.code, message: err.message } }); } res.status(202).json({ lane: payload(lane) }); void withLaneLock(lane.id, async () => { try { const baseBranch = await resolveBase( resolvedSourceRepo, body.base || process.env.LANE_BASE_BRANCH || "main" ); await addWorktree({ sourceRepo: resolvedSourceRepo, dir, branch, base: baseBranch }); // base_branch is a provisioning fact, not a patchable field — see PATCHABLE. lanesLib.setProvisioningFacts(lane.id, { base_branch: baseBranch }); lanesLib.updateLane(lane.id, { status: "idle", notes: null }); } catch (err) { lanesLib.updateLane(lane.id, { status: "failed", notes: err.git?.stderr || err.message, }); } broadcastLane(lane.id); }); }); const ACTIONS = new Set(["start", "stop", "message", "clear", "reset", "remove", "purge"]); // The modes the spawner accepts, same as POST /api/run. const RUN_MODES = new Set(["headless", "conversation"]); const DESTRUCTIVE_ACTIONS = new Set(["reset", "remove", "purge"]); const RUN_EXIT_POLL_MS = 50; // killRun escalates from SIGTERM to SIGKILL after five seconds. Leave enough // time for that escalation and for Node to receive the child's real exit. const RUN_EXIT_TIMEOUT_MS = 7500; function lifecycleError(code, message) { return Object.assign(new Error(message), { code }); } function expectedFields(action) { return action === "purge" ? ["sessions", "events", "tokenRows"] : ["head", "dirty", "untracked", "unpushed"]; } /** Refuse a destructive action when the facts shown in its preflight have moved. */ function assertExpectedPreflight(action, current, expected) { const fields = expectedFields(action); if ( !expected || typeof expected !== "object" || Array.isArray(expected) || fields.some((field) => !Object.hasOwn(expected, field)) ) { throw lifecycleError( "EEXPECT", `${action} requires a complete expect object with: ${fields.join(", ")}` ); } const changed = fields.some((field) => current[field] !== expected[field]); if (changed) { const err = lifecycleError("ESTALE", "lane state changed since preflight"); err.expected = expected; err.current = current; throw err; } } function wait(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } /** Kill a lane run and wait for the child's real `exit` event before touching its cwd. */ async function stopLaneRun(lane) { if (!lane.run_id) return; try { runs.killRun(lane.run_id); } catch { /* a concurrently completed run is already safe */ } const deadline = Date.now() + RUN_EXIT_TIMEOUT_MS; let run = runs.getRun(lane.run_id); while (run && !run.actualExitedAt) { if (Date.now() >= deadline) { throw lifecycleError( "ERUNTIMEOUT", `lane run ${lane.run_id} did not exit within ${RUN_EXIT_TIMEOUT_MS / 1000} seconds` ); } await wait(RUN_EXIT_POLL_MS); run = runs.getRun(lane.run_id); } lanesLib.updateLane(lane.id, { run_id: null }); } function sendLifecycleError(res, err) { if (["ENOTMANAGED", "EOUTSIDEROOT", "ENOTWORKTREE", "EEXPECT"].includes(err.code)) { return res.status(400).json({ error: { code: err.code, message: err.message } }); } if (["ESTALE", "EUNPUSHED"].includes(err.code)) { return res.status(409).json({ error: { code: err.code, message: err.message, ...(err.code === "ESTALE" ? { expected: err.expected, current: err.current } : {}), }, }); } if (err.git) { return res.status(500).json({ error: { code: err.code, message: err.message, stderr: err.git.stderr }, }); } return res.status(500).json({ error: { code: err.code, message: err.message } }); } /** * Lane control. Deliberately thin: every action maps onto one existing * run-spawner call. There is no queue, no chaining, no gate evaluation — the * dashboard drives a lane, it does not orchestrate a pipeline. */ router.post("/:id/:action", sameOriginGuard, async (req, res) => { const { action } = req.params; if (!ACTIONS.has(action)) { return res .status(400) .json({ error: { code: "EBADACTION", message: `unknown action ${action}` } }); } const lane = lanesLib.getLane(req.params.id); if (!lane) return res.status(404).json({ error: { code: "ENOLANE", message: "lane not found" } }); const body = req.body || {}; if (DESTRUCTIVE_ACTIONS.has(action)) { if (body.confirm !== true) { return res.status(400).json({ error: { code: "ECONFIRM", message: `${action} requires confirm: true` }, }); } try { const result = await withLaneLock(lane.id, async () => { const lockedLane = lanesLib.getLane(lane.id); if (!lockedLane) throw lifecycleError("ENOLANE", "lane not found"); await stopLaneRun(lockedLane); const current = lanesLib.getLane(lane.id); const facts = await preflight(current, action); assertExpectedPreflight(action, facts, body.expect); if ( (action === "reset" || (action === "remove" && current.kind === "managed")) && facts.unpushed > 0 && body.force !== true ) { throw lifecycleError( "EUNPUSHED", `${action} requires force: true when commits are unpushed` ); } if (action === "reset") { await resetWorktree(current); return { lane: lanesLib.clearLane(current.id) }; } if (action === "remove") { // Forgetting an adopted lane only removes dashboard metadata. The // filesystem destroy guard is deliberately reached only for managed // worktrees, where removal can actually touch a directory. if (current.kind === "managed") await removeWorktree(current); lanesLib.deleteLane(current.id); return { removed: current.id }; } return { purged: lanesLib.purgeLaneSessions(current.id) }; }); if (action === "remove") { broadcast("lane_update", { removed: result.removed }); return res.json({ ok: true }); } if (action === "purge") { broadcastLane(lane.id); return res.json({ ok: true, purged: result.purged }); } broadcastLane(lane.id); return res.json({ lane: payload(result.lane) }); } catch (err) { if (err.code === "ENOLANE") { return res.status(404).json({ error: { code: err.code, message: err.message } }); } return sendLifecycleError(res, err); } } try { switch (action) { case "start": { // Same two modes POST /api/run accepts. Unlike that route, an unknown // value is refused rather than silently coerced to a conversation. if (body.mode != null && !RUN_MODES.has(body.mode)) { return res.status(400).json({ error: { code: "EBADMODE", message: `mode must be one of: headless, conversation` }, }); } // Overwriting run_id while its child is alive orphans that child: a later // reset would kill and await only the RECORDED run, then `git clean -fd` // the directory the orphan is still writing into — the exact hazard // actualExitedAt exists to close. Stop the first run before starting a // second. The check and the spawn happen under the per-lane lock so that // atomicity is guaranteed rather than an accident of this code having no // `await` between them — a future edit that adds one must not reopen the // race. const outcome = await withLaneLock(lane.id, async () => { const current = lanesLib.getLane(lane.id); // A concurrent `remove` could have deleted the row while this request // waited for the lock — the lock makes that race visible instead of // spawning a run for a lane that no longer exists. if (!current) return { missing: true }; const live = current.run_id ? runs.getRun(current.run_id) : null; if (live && (live.status === "spawning" || live.status === "running")) { return { conflict: true }; } const handle = runs.spawnRun({ mode: body.mode || "conversation", laneId: current.id, prompt: body.prompt || "", cwd: current.cwd, model: body.model, permissionMode: body.permissionMode, effort: body.effort, resumeSessionId: body.resumeSessionId || (body.resume ? current.session_id : undefined), }); lanesLib.updateLane(current.id, { run_id: handle.id, status: "running" }); return {}; }); if (outcome.missing) { return res.status(404).json({ error: { code: "ENOLANE", message: "lane not found" } }); } if (outcome.conflict) { return res .status(409) .json({ error: { code: "ERUNLIVE", message: "lane already has a live run" } }); } break; } case "stop": { // A lane with no live run is already stopped — say so, don't 500. if (lane.run_id) { try { runs.killRun(lane.run_id); } catch { /* already gone */ } } lanesLib.updateLane(lane.id, { status: "idle", run_id: null }); break; } case "message": { if (!lane.run_id) { return res .status(409) .json({ error: { code: "ENORUN", message: "lane has no live run" } }); } // Check that the recorded run is actually live (spawning or running). // If a run finished recently, its run_id is still recorded but sendInput // would throw ENOTRUNNING. Return 409 so the client knows it's not a server error. const run = runs.getRun(lane.run_id); if (!run || (run.status !== "spawning" && run.status !== "running")) { return res .status(409) .json({ error: { code: "ENORUN", message: "lane has no live run" } }); } runs.sendInput(lane.run_id, String(body.text || "")); lanesLib.updateLane(lane.id, { needs_action: null }); break; } case "clear": lanesLib.clearLane(lane.id); break; default: break; } } catch (err) { return res.status(500).json({ error: { code: err.code, message: err.message } }); } broadcastLane(lane.id); res.json({ lane: payload(lanesLib.getLane(lane.id)) }); }); router.delete("/:id", sameOriginGuard, (req, res) => { const ok = lanesLib.deleteLane(req.params.id); if (!ok) return res.status(404).json({ error: { code: "ENOLANE", message: "lane not found" } }); broadcast("lane_update", { removed: Number(req.params.id) }); res.json({ ok: true }); }); module.exports = router; module.exports.broadcastLane = broadcastLane;