Something went wrong. Try again.
Monorepo for Aesthetic.Computer aesthetic.computer
Something went wrong. Try again.
JavaScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366#!/usr/bin/env node// codex-session-watch.mjs — drive slab session state from a Codex rollout log.//// Codex has no per-turn hooks like Claude Code, but it streams every session// to ~/.codex/sessions/YYYY/MM/DD/rollout-<ts>-<uuid>.jsonl with clean turn// boundaries. We tail that file and flip the SAME state files the Claude hooks// use, so the menubar reducer stays agent-agnostic:// task_started → working (rewrite active marker, drop awaiting, touch running-tools)// task_complete → complete (write awaiting "turn complete", drop running-tools)//// Usage: node codex-session-watch.mjs <sessionId> <beginEpochSec> <wrapperPid> <tty> <cwd>// Launched (and killed) by codex-slab.sh. Exits when the wrapper pid dies.
import { readFile, writeFile, stat, unlink, utimes } from "node:fs/promises";import { realpathSync } from "node:fs";import { execFile } from "node:child_process";import { promisify } from "node:util";import { join, basename, resolve } from "node:path";import { fileURLToPath } from "node:url";import { homedir } from "node:os";
const [sid = "", _beginArg, wrapperArg, tty = "", cwd = ""] = process.argv.slice(2);const isMain = process.argv[1] && realpathSync(process.argv[1]) === fileURLToPath(import.meta.url);if (isMain && !sid) process.exit(1);const wrapperPid = Number(wrapperArg) || 0;
const SLAB_HOME = process.env.SLAB_HOME || join(homedir(), ".local", "share", "slab");const ACTIVE = join(SLAB_HOME, "state", "active-prompts", sid);const AWAITING = join(SLAB_HOME, "state", "awaiting-prompts", sid);const RUNNING = join(SLAB_HOME, "state", "running-tools", sid);const OPEN_IMAGES = join(SLAB_HOME, "state", "open-images");const SESSIONS = join(process.env.CODEX_HOME || join(homedir(), ".codex"), "sessions");const CODEX_ROOT = process.env.CODEX_HOME || join(homedir(), ".codex");const LOOPBOY_CONFIG = join(homedir(), ".config", "slab", "loopboy.json");const execFileAsync = promisify(execFile);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));const nowISO = () => new Date().toISOString().replace(/\.\d+Z$/, "Z");const wrapperAlive = () => { if (!wrapperPid) return true; try { process.kill(wrapperPid, 0); return true; } catch (error) { // Sandboxed launch contexts may forbid signalling an otherwise-live // sibling. EPERM proves the PID exists; ESRCH is the actual dead case. return error?.code === "EPERM"; }};
// Find the rollout file for THIS Codex process. Concurrent windows often share// cwd and start within the same second, so "newest file" cross-wires their// rocks. The wrapper and Codex are parent/child; Codex keeps its own rollout// open for writing, giving us an exact, resume-safe association via lsof.export function rolloutFromTitle(rows, title, cwd) { const matches = rows.filter(row => typeof row.name === "string" && row.name.length >= 4 && resolve(row.cwd) === resolve(cwd) && (title === row.name || title.endsWith(`${row.name} | ${basename(row.cwd)}`))); // Never guess between concurrent windows or similarly named threads. return matches.length === 1 ? matches[0].rollout_path : null;}
async function namedRollout() { if (!/^ttys[0-9]+$/.test(tty)) return null; try { // New Codex versions share a daemon: its open files belong to MANY TUIs. // Match this TTY's exact displayed thread name to local thread metadata. // Read names/paths only; no prompts, transcript text, or credentials. const { stdout: title } = await execFileAsync("/usr/bin/osascript", ["-e", ` if application "Terminal" is not running then return "" tell application "Terminal" repeat with w in windows if tty of selected tab of w is "/dev/${tty}" then return name of w end repeat end tell return ""`], { timeout: 2000 }); if (!title.trim()) return null; const { stdout } = await execFileAsync("/usr/bin/sqlite3", ["-readonly", "-json", join(CODEX_ROOT, "state_5.sqlite"), "select name,cwd,rollout_path from threads where archived=0 and name is not null"], { timeout: 2000 }); const path = rolloutFromTitle(JSON.parse(stdout || "[]"), title.trim(), cwd); return path?.startsWith(SESSIONS + "/") ? path : null; } catch { return null; }}
async function findRollout() { // Reinstalling Slab reattaches watchers to the same live wrappers. Once an // exact identity is known, prefer it even while Slab owns the window title. try { const marker = JSON.parse(await readFile(ACTIVE, "utf8")); const id = marker.provider_session_id || marker.codex_session_id; if (/^[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}$/i.test(id || "")) { const { stdout } = await execFileAsync("/usr/bin/sqlite3", ["-readonly", "-json", join(CODEX_ROOT, "state_5.sqlite"), `select rollout_path from threads where id='${id}'`], { timeout: 2000 }); const path = JSON.parse(stdout || "[]")[0]?.rollout_path; if (path?.startsWith(SESSIONS + "/") && (await stat(path)).isFile()) return path; } } catch {} // A resume picker can remain open indefinitely; keep waiting for selection. for (let i = 0; wrapperAlive(); i++) { const named = await namedRollout(); if (named) return named; try { // BSD/macOS ps has no Linux `-P <parent>` selector. Read its compact // PID/PPID table and select the wrapper's children ourselves. const { stdout: processes } = await execFileAsync( "/bin/ps", ["-axo", "pid=,ppid="]); const pairs = processes.split("\n").map((line) => line.trim().split(/\s+/)) .filter((parts) => parts.length >= 2) .map(([pid, ppid]) => ({ pid: Number(pid), ppid: Number(ppid) })) .filter(({ pid, ppid }) => Number.isFinite(pid) && Number.isFinite(ppid)); // npm's current Codex package launches a Node shim, which then launches // the native Rust binary that owns the rollout. Walk the whole process // subtree instead of inspecting only the wrapper's direct child. const descendants = new Set([wrapperPid]); let grew = true; while (grew) { grew = false; for (const { pid, ppid } of pairs) { if (descendants.has(ppid) && !descendants.has(pid)) { descendants.add(pid); grew = true; } } } const pids = [...descendants] .filter((pid) => pid !== wrapperPid && pid !== process.pid) .map(String); const rollouts = new Set(); for (const pid of pids) { let stdout; try { ({ stdout } = await execFileAsync("/usr/sbin/lsof", ["-Fn", "-p", pid], { timeout: 2000 })); } catch { continue; } for (const rollout of stdout.split("\n") .filter((line) => line.startsWith("n")) .map((line) => line.slice(1)) .filter((p) => p.startsWith(SESSIONS + "/") && p.includes("/rollout-") && p.endsWith(".jsonl"))) rollouts.add(rollout); } if (rollouts.size === 1) return [...rollouts][0]; } catch { // Codex may not have opened its rollout yet; retry below. } await sleep(i < 20 ? 500 : 5000); } return null;}
// Read the marker, merge fields, write it back (atomic-ish).async function updateMarker(patch) { // Keep enough launch metadata here to reconstruct a marker if an external // janitor removes it while Codex is still alive. Without this baseline the // next rollout event recreated only `{state, updated}`, losing the tty and // agent type that Slab needs to theme the terminal. let obj = { session_id: sid, cwd, subject: "codex session", summary: "codex", tty, agent_pid: wrapperPid, agent_type: "codex", state: "blank", }; try { Object.assign(obj, JSON.parse(await readFile(ACTIVE, "utf8"))); } catch {} Object.assign(obj, patch, { updated: nowISO() }); try { await writeFile(ACTIVE, JSON.stringify(obj)); } catch {}}const rm = async (p) => { try { await unlink(p); } catch {} };const touch = async (p) => { const t = new Date(); try { await writeFile(p, "", { flag: "a" }); await utimes(p, t, t); } catch {}};
// rollout-<timestamp>-<provider uuid>.jsonl → the stable Codex thread id.// Persisting this in the active marker lets Slab refresh/zzz a tracked wrapper// and resume the same provider conversation instead of opening a fresh one.function sessionIdFromRollout(file) { return String(file || "").match(/-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\.jsonl$/i)?.[1] || "";}
// Pull readable text out of a rollout content array (user prompt → subject).function textOf(payload) { const c = payload?.content; if (!Array.isArray(c)) return ""; return c.map((x) => x?.text || x?.input_text || "").join(" ").replace(/\s+/g, " ").trim();}function summarize(s) { const words = s.split(" ").filter(Boolean).slice(0, 7).join(" "); return words.length > 48 ? words.slice(0, 45) + "…" : (s.split(" ").length > 7 ? words + "…" : words);}
async function isLoopboySession() { try { const cfg = JSON.parse(await readFile(LOOPBOY_CONFIG, "utf8")); return Object.values(cfg?.loops || {}).some((loop) => loop?.sessionId === sid); } catch { return false; }}
async function onTurnStart(subject) { await rm(AWAITING); await touch(RUNNING); const patch = { state: "working" }; // Intake is a real phase: a heartbeat first reads the assigned thread. The // assistant's explicit WORKING output advances it once implementation starts. if (await isLoopboySession()) patch.loopboy_state = "reading"; if (subject) { patch.subject = subject.slice(0, 140); patch.summary = summarize(subject); } await updateMarker(patch);}async function onTurnComplete(lastMessage = "") { try { await writeFile(AWAITING, "turn complete\n"); } catch {} await rm(RUNNING); const patch = { state: "complete" }; if (await isLoopboySession()) { if (/\bRESPONDING\b/i.test(lastMessage) && !/\b(?:sent once|outbound appears|verified outbound)\b/i.test(lastMessage)) { patch.loopboy_state = "responding"; patch.loopboy_response = lastMessage.slice(0, 1200); } else { // Sent, verified, or otherwise finished: no active phase remains. patch.loopboy_state = "idle"; patch.loopboy_response = ""; } } await updateMarker(patch);}async function onAwaiting(msg) { try { await writeFile(AWAITING, (msg || "needs approval") + "\n"); } catch {} await rm(RUNNING);}
async function onVisualArtifact(path) { // ImageGroupPreview consumes this request on the menubar's next tick. // The newest generated artifact becomes the current review wall. try { await writeFile(OPEN_IMAGES, path + "\n"); } catch {} await onAwaiting("visual artifact ready for review");}
export function handleLine(line, ctx) { // Image generation returns a saved host path beside an inline bitmap. Match // only tool response items so quoted history cannot reopen stale artifacts. const isImageToolOutput = line.includes('"type":"response_item"') && line.includes('"role":"tool"'); const generated = isImageToolOutput ? line.match(/Generated images are saved[^\n]*? as (\/[^\s]+\.(?:png|jpe?g|webp))/i) : null; if (generated && !ctx.replaying) ctx.pending.push(() => onVisualArtifact(generated[1])); let obj; try { obj = JSON.parse(line); } catch { return; } const type = obj.type; const payload = obj.payload || obj; if (type === "response_item" && (payload.role === "user")) { const t = textOf(payload); if (t) { ctx.lastUser = t; // Current Codex writes task_started before the user response items. If // we only sample lastUser at task_started, a fresh rock stays named // simply "codex" for the entire first turn. Update the visible subject // when the actual user message arrives; later user items in the same // rollout naturally win over injected context blocks. if (!ctx.replaying) ctx.pending.push(() => updateMarker({ subject: t.slice(0, 140), summary: summarize(t), })); } return; } if (type === "response_item" && payload.role === "assistant") { if (ctx.replaying) return; const t = textOf(payload); if (/\bRESPONDING\b/i.test(t)) { ctx.pending.push(() => updateMarker({ loopboy_state: "responding", loopboy_response: t.slice(0, 1200), })); } else if (/\bREADING\b/i.test(t)) { ctx.pending.push(() => updateMarker({ loopboy_state: "reading" })); } else if (/\bWORKING\b/i.test(t)) { ctx.pending.push(() => updateMarker({ loopboy_state: "working" })); } return; } if (type === "event_msg") { const pt = payload.type || ""; const transition = fn => { if (ctx.replaying) ctx.pending = [fn]; else ctx.pending.push(fn); }; if (pt === "task_started" || pt === "user_turn") { ctx.turnActive = true; transition(() => onTurnStart(ctx.lastUser)); } else if (pt === "task_complete" || pt === "turn_complete") { ctx.turnActive = false; transition(() => onTurnComplete(payload.last_agent_message || "")); } else if (pt === "turn_aborted") { ctx.turnActive = false; transition(async () => { await rm(RUNNING); await rm(AWAITING); await updateMarker({ state: "interrupted" }); }); } else if (pt.includes("approval") || pt.includes("elicitation")) { ctx.turnActive = false; transition(() => onAwaiting("codex needs approval")); } }}
async function main() { const file = await findRollout(); if (!file) process.exit(0); try { const first = (await readFile(file, "utf8")).split("\n")[0]; const meta = JSON.parse(first); const providerId = meta?.type === "session_meta" ? meta?.payload?.id : null; if (providerId) await updateMarker({ codex_session_id: providerId, provider_session_id: providerId, transcript_path: file, }); } catch {} // Replay once from the beginning so a resumed Codex window immediately // inherits its real last state (usually complete) instead of sitting blank // or aging into interrupted until the user submits another prompt. After // that first pass `offset` makes this an ordinary incremental tail. let offset = 0; const ctx = { lastUser: "", pending: [], turnActive: false, lastHeartbeatAt: 0, replaying: true }; while (wrapperAlive()) { let size = offset; try { size = (await stat(file)).size; } catch { break; } if (size > offset) { let chunk = Buffer.alloc(0); try { chunk = await readFile(file); } catch { chunk = Buffer.alloc(0); } // Offsets from stat are bytes, not JavaScript UTF-16 character counts. // Slice the Buffer first so emoji/non-ASCII output can never skew the // tail boundary or cause an already-seen completion to be replayed. const tail = chunk.subarray(offset).toString("utf8"); offset = chunk.length; for (const line of tail.split("\n")) if (line.trim()) handleLine(line, ctx); if (ctx.replaying && ctx.lastUser) { await updateMarker({ subject: ctx.lastUser.slice(0, 140), summary: summarize(ctx.lastUser) }); } // Apply transitions in order; last one wins the visible state. for (const fn of ctx.pending) await fn(); ctx.pending = []; ctx.replaying = false; } // Codex can spend many minutes inside one tool call without appending a // new rollout event. Keep both marker channels fresh for the full active // turn so Slab's reducer does not misclassify visible progress as an // interrupted prompt. Completion/approval events clear turnActive first. const now = Date.now(); if (ctx.turnActive && now - ctx.lastHeartbeatAt >= 5000) { await touch(RUNNING); await updateMarker({ state: "working" }); ctx.lastHeartbeatAt = now; } await sleep(600); }}
if (isMain) main().catch((error) => { console.error(`codex-session-watch: ${error?.stack || error}`); process.exit(1);});