Claude-Code-Monitor/server/routes/lanes.js

/**
 * @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ĩ <vinnt@smartgift.vn>
 */

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;