AgentHub/server/lib/lanes.js

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

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",
  "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" });
  }
}

/**
 * `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 };
}

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);
  validatePipeline(pipeline);
  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);
  }
  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} = ?`);
    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:
 * `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<number>}
 */
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.
 */
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) {
  // 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.
 */
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 <other> && ...`, 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.
 * 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",
  }));
}

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_skill: pipeline.skill || "",
    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 `<repo>__<slug>` — 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,
  usedSlots,
  reservedPorts,
};