feat: per-change-block hunk actions (#50), codex lanes without a pipeline, macOS LaunchAgent
- git (#50): stage-hunk/discard-hunk take an optional `block` — one run of consecutive +/- lines inside an @@ hunk. `onlyBlock` (server/lib/git-diff.js) cuts git's fresh hunk down to it, folding the other blocks into the side being checked; LaneDiffView hangs one review bar per block (unified + split) and its arrows walk blocks. - git exclude: entries AgentHub writes for a lane go to the COMMON git dir with the lane's path prefix (server/lib/git-exclude.js, used by router-apply and lane-mcp) — a linked worktree's own info/exclude is never read by git, and a subdir lane has no .git. MCP sync no longer fails with ENOTDIR on a worktree lane or creates a stray .git in a non-repo lane. - codex lanes carry no pipeline: codex-ingest mirrors only idle/running, codex-nudge is removed, agents install writes nothing for codex, the Workspace hides stage/pipeline/feature pickers/PipelineMap for them, and every server start clears leftover codex stage data and the old AGENTS.md stage block. - Changes panel: a whole file opens for reading and enters the editor on a click on its text (caret at the clicked line), not automatically. - macOS: LaunchAgent start-at-login (scripts/launchd-service.js, picked by scripts/service-manager.js). - Adds the AgentHub introduction slides and their image-embedding script.
This commit is contained in:
+22
-222
@@ -1,26 +1,16 @@
|
||||
/**
|
||||
* @file Codex stage ingestion. Codex emits no hooks, so the path that infers a
|
||||
* lane's stage (server/routes/hooks.js → stage-detect.detect) can never see a
|
||||
* `codex` lane — and it is deliberately pinned to `claude`, because a Claude
|
||||
* hook must never touch a codex lane sharing the same directory. Codex's
|
||||
* on-disk equivalent is the rollout journal it appends live, one file per
|
||||
* thread:
|
||||
* @file Codex lane status mirror. Codex emits no hooks, so nothing else can
|
||||
* tell the dashboard that a `codex` lane's agent is working. Its on-disk
|
||||
* signal is the rollout journal it appends live, one file per thread:
|
||||
*
|
||||
* $CODEX_HOME/sessions/<YYYY>/<MM>/<DD>/rollout-<ts>-<threadId>.jsonl
|
||||
* line 1: {"type":"session_meta","payload":{"cwd":…,"session_id":…}}
|
||||
* then: {"type":"response_item","payload":{"type":"custom_tool_call",…}}
|
||||
*
|
||||
* This module tails those journals for the cwd of every `codex` lane, maps each
|
||||
* tool-call payload onto the same `{tool_name, tool_input}` shape a Claude hook
|
||||
* delivers, and feeds the EXISTING detect() + recordDetection() path — so a
|
||||
* codex lane paints the same amber "detected" badge under the same rules
|
||||
* (forward-only, declared-wins) and never renders `done`. A written detection
|
||||
* then ALSO advances the lane's declared `stage` to that node, without
|
||||
* evidence: codex lanes are the one place where nothing else can move the
|
||||
* pipeline map, so the node renders `passed-no-evidence` (amber) rather than
|
||||
* `done`, and codex-nudge.js asks the session for the real evidence. Like
|
||||
* workflow-ingest.js, everything here is fail-safe: a malformed or partial
|
||||
* journal throws only locally and is skipped, and a poll pass never throws.
|
||||
* This module maps each journal to the `codex` lane owning its cwd and mirrors
|
||||
* idle↔running from how recently the journal was written. It reads nothing but
|
||||
* the first line (for the cwd) and never touches a lane's stage — codex lanes
|
||||
* carry no pipeline. Fail-safe like workflow-ingest.js: a malformed journal is
|
||||
* skipped, and a poll pass never throws.
|
||||
*
|
||||
* Codex 0.149 also mirrors threads into version-suffixed SQLite files
|
||||
* (`state_5.sqlite`, `thread_history_1.sqlite`), but every thread row still
|
||||
@@ -35,70 +25,10 @@ const path = require("node:path");
|
||||
const os = require("node:os");
|
||||
const { getCodexHome } = require("./codex-home");
|
||||
|
||||
// `*** Update File: <path>` lines inside an apply_patch heredoc.
|
||||
const PATCH_FILE_RE = /^\*\*\* (Add|Update|Delete) File: (.+)$/gm;
|
||||
const SHELL_NAMES = new Set(["exec", "exec_command", "shell", "local_shell", "container.exec"]);
|
||||
|
||||
const CODEX_ACTIVE_MS = 120_000; // a journal written within 2 min means the agent is working
|
||||
const TAIL_BYTES = 64 * 1024; // first sight of a live file: read only its recent tail
|
||||
const MAX_FILES = 40; // newest-first cap so a years-deep sessions/ tree can't stall a pass
|
||||
const MAX_READ_BYTES = 4 * 1024 * 1024; // one pass never buffers more than this per journal
|
||||
const MAX_DEPTH = 4; // sessions/<YYYY>/<MM>/<DD>/<file>
|
||||
|
||||
/** The `cmd` of a shell call, whether it arrived as JSON arguments or an exec script. */
|
||||
function commandFromPayload(payload, text) {
|
||||
try {
|
||||
const parsed = JSON.parse(payload.arguments);
|
||||
if (parsed && typeof parsed.cmd === "string") return parsed.cmd;
|
||||
if (parsed && Array.isArray(parsed.command)) return parsed.command.join(" ");
|
||||
} catch {
|
||||
/* not JSON — an exec script; the detect rules match inside it verbatim */
|
||||
}
|
||||
return text;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hook-shaped events for one rollout payload. A patch script yields one event
|
||||
* per touched file (Write for a new file, Edit otherwise — the split the
|
||||
* pipeline templates' detect rules assume); a shell call yields one Bash event.
|
||||
* Anything else — reasoning, messages, tool OUTPUT, a web search — yields
|
||||
* nothing, so an unknown Codex tool can never be mistaken for work.
|
||||
*
|
||||
* @param {object|null} payload - One rollout line's `payload` object.
|
||||
* @returns {Array<{tool_name: string, tool_input: object}>}
|
||||
*/
|
||||
function eventsFromPayload(payload) {
|
||||
if (!payload || typeof payload !== "object") return [];
|
||||
if (payload.type !== "custom_tool_call" && payload.type !== "function_call") return [];
|
||||
const name = typeof payload.name === "string" ? payload.name : "";
|
||||
const text = typeof payload.input === "string" ? payload.input : String(payload.arguments || "");
|
||||
|
||||
if (name === "update_plan") return [{ tool_name: "TaskCreate", tool_input: {} }];
|
||||
if (name === "spawn_agent") {
|
||||
let taskName = "";
|
||||
try {
|
||||
taskName = JSON.parse(payload.arguments).task_name || "";
|
||||
} catch {
|
||||
/* encrypted or malformed arguments — the tool name alone still matters */
|
||||
}
|
||||
return [{ tool_name: "Agent", tool_input: { subagent_type: taskName } }];
|
||||
}
|
||||
if (!SHELL_NAMES.has(name)) return [];
|
||||
|
||||
// A patch heredoc reaches us inside a JS string literal, so its line breaks
|
||||
// are usually a literal backslash-n rather than newlines (real rollouts carry
|
||||
// both forms in the same script). Unescape before scanning or the file list
|
||||
// comes back empty and a whole edit looks like a plain shell command.
|
||||
const patched = [...text.replace(/\\n/g, "\n").matchAll(PATCH_FILE_RE)];
|
||||
if (patched.length) {
|
||||
return patched.map(([, verb, file]) => ({
|
||||
tool_name: verb === "Add" ? "Write" : "Edit",
|
||||
tool_input: { file_path: file.trim() },
|
||||
}));
|
||||
}
|
||||
return [{ tool_name: "Bash", tool_input: { command: commandFromPayload(payload, text) } }];
|
||||
}
|
||||
|
||||
/** Every rollout journal under sessions/, newest mtime first, capped at MAX_FILES. */
|
||||
function rolloutFiles() {
|
||||
const out = [];
|
||||
@@ -151,101 +81,29 @@ function rolloutCwd(file) {
|
||||
return cwd;
|
||||
}
|
||||
|
||||
const offsets = new Map(); // rollout path → bytes already ingested
|
||||
|
||||
/** Test seam: forget every journal's ingest position (and cached cwd). */
|
||||
function __resetOffsets() {
|
||||
offsets.clear();
|
||||
/** Test seam: forget every cached cwd. */
|
||||
function __resetCache() {
|
||||
cwdCache.clear();
|
||||
}
|
||||
|
||||
// Both maps are keyed by file path, so in a process that runs for months they
|
||||
// would otherwise hold one dead entry per journal Codex ever wrote. Pruned only
|
||||
// once well past the per-pass scan size, so ordinary churn keeps its positions
|
||||
// and a journal that drops out of the scan behaves exactly as it does after a
|
||||
// server restart: re-seen as new, and read only if it is being written to.
|
||||
// Keyed by file path, so in a process that runs for months it would otherwise
|
||||
// hold one dead entry per journal Codex ever wrote. Pruned only once well past
|
||||
// the per-pass scan size, so ordinary churn keeps its entries.
|
||||
const MAX_TRACKED = MAX_FILES * 4;
|
||||
|
||||
function pruneTracking(seen) {
|
||||
for (const map of [offsets, cwdCache]) {
|
||||
if (map.size <= MAX_TRACKED) continue;
|
||||
for (const key of map.keys()) if (!seen.has(key)) map.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
/** Read `[from, size)` of a file, or an empty buffer when it can't be read. */
|
||||
function readRange(file, from, size) {
|
||||
try {
|
||||
const fd = fs.openSync(file, "r");
|
||||
try {
|
||||
const buf = Buffer.alloc(size - from);
|
||||
const read = fs.readSync(fd, buf, 0, buf.length, from);
|
||||
return buf.subarray(0, read);
|
||||
} finally {
|
||||
fs.closeSync(fd);
|
||||
}
|
||||
} catch {
|
||||
return Buffer.alloc(0);
|
||||
}
|
||||
if (cwdCache.size <= MAX_TRACKED) return;
|
||||
for (const key of cwdCache.keys()) if (!seen.has(key)) cwdCache.delete(key);
|
||||
}
|
||||
|
||||
/**
|
||||
* The COMPLETE lines in `[from, size)`, plus the offset to resume from.
|
||||
*
|
||||
* Two fragments have to be handled, and only one of them is dropped:
|
||||
* - a tail read (`firstSight` with `from > 0`) starts mid-line, so its leading
|
||||
* fragment is discarded;
|
||||
* - a read that catches the writer mid-line ends with a fragment, so the
|
||||
* returned offset stops at the last newline and that fragment is re-read
|
||||
* whole on the next pass.
|
||||
*
|
||||
* Every non-tail read therefore starts exactly on a line boundary, where the
|
||||
* first line is a WHOLE record. Dropping it there — the obvious "the offset
|
||||
* always lands mid-line" shortcut — loses every append that arrives one line at
|
||||
* a time, which is how a journal actually grows.
|
||||
*
|
||||
* All of it is done on the BUFFER, never on a decoded string: a tail offset can
|
||||
* land inside a multi-byte UTF-8 sequence, where the decoder emits a 3-byte
|
||||
* U+FFFD for 1-2 consumed bytes. Measuring a decoded slice with
|
||||
* `Buffer.byteLength` would then move the offset off the real line boundary and
|
||||
* every later record would fail to parse.
|
||||
*/
|
||||
const NEWLINE = 0x0a;
|
||||
|
||||
function completeLines(file, from, size, firstSight) {
|
||||
let chunk = readRange(file, from, size);
|
||||
let start = from;
|
||||
if (firstSight && from > 0) {
|
||||
const firstBreak = chunk.indexOf(NEWLINE);
|
||||
if (firstBreak === -1) return { lines: [], next: size };
|
||||
start += firstBreak + 1;
|
||||
chunk = chunk.subarray(firstBreak + 1);
|
||||
}
|
||||
const lastBreak = chunk.lastIndexOf(NEWLINE);
|
||||
if (lastBreak === -1) return { lines: [], next: start };
|
||||
return {
|
||||
lines: chunk.subarray(0, lastBreak).toString("utf8").split("\n"),
|
||||
next: start + lastBreak + 1,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* One poll pass. Reads only the bytes appended since the last pass; on the
|
||||
* first sight of a file, only its recent tail, and nothing at all when the file
|
||||
* is already idle — replaying a thread that finished days ago would paint
|
||||
* stages from work nobody is doing.
|
||||
*
|
||||
* Detection uses the LAST matching event of the batch, mirroring "one hook, one
|
||||
* detection"; recordDetection is forward-only, so ordering inside a batch can
|
||||
* never drag a lane backwards.
|
||||
* One poll pass: mirror each codex lane's idle↔running from its journals.
|
||||
*
|
||||
* @param {{broadcast?: (laneId: number) => void}} [options]
|
||||
* @returns {number} How many lanes actually changed (and were broadcast).
|
||||
*/
|
||||
function ingestCodexOnce({ broadcast } = {}) {
|
||||
const lanesLib = require("./lanes");
|
||||
const { detect } = require("./stage-detect");
|
||||
const { getPipeline } = require("./pipelines");
|
||||
|
||||
let codexLanes = [];
|
||||
try {
|
||||
@@ -266,67 +124,12 @@ function ingestCodexOnce({ broadcast } = {}) {
|
||||
.sort((a, b) => b.cwd.length - a.cwd.length)[0];
|
||||
if (!lane) continue;
|
||||
|
||||
const firstSight = !offsets.has(file.path);
|
||||
// An mtime in the FUTURE (clock stepped back, NTP correction, a restore that
|
||||
// preserved timestamps) makes the raw difference negative, which reads as
|
||||
// "written milliseconds ago" and would let a finished thread's journal
|
||||
// replay as live work. A file that claims to be from the future is not
|
||||
// evidence of an agent working now, so treat it as idle.
|
||||
// "written milliseconds ago". A file that claims to be from the future is
|
||||
// not evidence of an agent working now, so treat it as idle.
|
||||
const age = Date.now() - file.mtime;
|
||||
const active = age >= 0 && age < CODEX_ACTIVE_MS;
|
||||
let from = firstSight ? Math.max(0, file.size - TAIL_BYTES) : offsets.get(file.path);
|
||||
// A stored offset past EOF means the file was truncated or a new thread took
|
||||
// the same path. Restart from a bounded tail, never from 0: reading a
|
||||
// multi-gigabyte journal whole would allocate it all in one buffer.
|
||||
let tailOnly = firstSight;
|
||||
if (from > file.size || file.size - from > MAX_READ_BYTES) {
|
||||
from = Math.max(0, file.size - TAIL_BYTES);
|
||||
tailOnly = true;
|
||||
}
|
||||
|
||||
let changed = false;
|
||||
// A journal seen for the first time while already idle is history, not work:
|
||||
// note where it ends and read nothing.
|
||||
if (firstSight && !active) {
|
||||
offsets.set(file.path, file.size);
|
||||
} else if (file.size > from) {
|
||||
const { lines, next } = completeLines(file.path, from, file.size, tailOnly);
|
||||
offsets.set(file.path, next);
|
||||
const pipeline = getPipeline(lane.pipeline);
|
||||
let detection = null;
|
||||
for (const raw of lines) {
|
||||
if (!raw.trim()) continue;
|
||||
let payload = null;
|
||||
try {
|
||||
payload = JSON.parse(raw).payload;
|
||||
} catch {
|
||||
continue; // not a record this module understands
|
||||
}
|
||||
for (const event of eventsFromPayload(payload)) {
|
||||
const hit = detect(pipeline, event);
|
||||
if (hit) detection = hit;
|
||||
}
|
||||
}
|
||||
if (detection) {
|
||||
try {
|
||||
changed = lanesLib.recordDetection(lane.id, detection).written;
|
||||
// Auto-advance: a codex session has no hook and rarely calls `agenthub
|
||||
// stage` on its own, so without this its card sits at `idle` forever
|
||||
// while the amber badge is the only thing that ever moves. Gated on
|
||||
// recordDetection having WRITTEN, so it inherits that function's
|
||||
// forward-only + declared-wins + TTL guards for free. Deliberately
|
||||
// passes NO evidence: the node renders `passed-no-evidence` (amber),
|
||||
// never `done` — detection sees a tool call, never its outcome. The
|
||||
// last node has no detect rules in any template, so this can never
|
||||
// auto-declare a lane `done`.
|
||||
if (changed) {
|
||||
lanesLib.setStage(lane.id, { stage: detection.nodeId, keepDetection: true });
|
||||
}
|
||||
} catch {
|
||||
/* lane deleted mid-pass, or sqlite busy — an inference is never worth throwing for */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Status mirror, under the same guard hook ingestion uses: only idle↔running,
|
||||
// and only for a lane the dashboard did not launch — provisioning and failed
|
||||
@@ -340,12 +143,11 @@ function ingestCodexOnce({ broadcast } = {}) {
|
||||
) {
|
||||
try {
|
||||
lanesLib.updateLane(lane.id, { status: next });
|
||||
changed = true;
|
||||
touched.add(lane.id);
|
||||
} catch {
|
||||
/* as above */
|
||||
/* lane deleted mid-pass, or sqlite busy — a status mirror is never worth throwing for */
|
||||
}
|
||||
}
|
||||
if (changed) touched.add(lane.id);
|
||||
}
|
||||
|
||||
for (const id of touched) {
|
||||
@@ -361,8 +163,7 @@ function ingestCodexOnce({ broadcast } = {}) {
|
||||
/**
|
||||
* Unref'd interval so it never blocks shutdown; disable with
|
||||
* DASHBOARD_CODEX_POLL_MS=0. Codex appends its journal continuously, so a 15s
|
||||
* beat keeps a codex card within one node of live without ever re-reading what
|
||||
* it has already ingested.
|
||||
* beat keeps a codex card's running/idle within one beat of live.
|
||||
*/
|
||||
function startCodexPoll(broadcast) {
|
||||
const POLL_MS = process.env.DASHBOARD_CODEX_POLL_MS
|
||||
@@ -380,9 +181,8 @@ function startCodexPoll(broadcast) {
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
eventsFromPayload,
|
||||
rolloutCwd,
|
||||
ingestCodexOnce,
|
||||
startCodexPoll,
|
||||
__resetOffsets,
|
||||
__resetCache,
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user