Something went wrong. Try again.
presents omp as codex for use with chatgpt remote
Something went wrong. Try again.
53 kB · 1263 lines
JavaScript
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264import fs from "node:fs/promises";import os from "node:os";import path from "node:path";import { atomicWriteJson, commandActionsForOmpTool, compactWholeFileUnifiedDiff, debug, decodeCursor, encodeCursor, formatToolCommand, paginateArray, readJson, safePreview, unixSeconds } from "./util.mjs";import { makeThread, normalizePatchChangeKind } from "./protocol.mjs";import { completeHistoricalToolItem, inspectOmpSessionActivity, listOmpSessionsForCwd, listAllOmpSessions, parseOmpSessionFile, defaultSessionsDir } from "./omp-sessions.mjs";import { findPrimeSession, listAllPrimeSessions, listPrimeSessionsForCwd } from "./prime-sessions.mjs";const SCHEMA_VERSION = 1;
function processIsAlive(pid) { if (!Number.isInteger(pid) || pid <= 0) return false; try { process.kill(pid, 0); return true; } catch { return false; }}
function usesOmpSessionProjection(backend) { return backend?.kind !== "prime-rpc" || backend?.external === true;}
function defaultState() { return { schemaVersion: SCHEMA_VERSION, threads: {}, order: [] };}
function storedModelId(thread) { const model = thread?.model; const raw = typeof model === "object" ? (model.id || model.model || model.selector) : model; const selector = thread?.modelSelector || (typeof model === "object" ? model.selector : null); if (selector && raw === selector && String(raw).includes("/")) { return String(raw).slice(String(raw).indexOf("/") + 1); } return raw || null;}
function normalizeStoredItem(item, cwd) { if (!item) return item; if (item.type === "fileChange" && Array.isArray(item.changes)) { // iOS decodes the change kind as an enum object, a bare "update" string // from an older projection makes it fail the whole thread return { ...item, changes: item.changes.map((change) => ({ ...change, kind: normalizePatchChangeKind(change), diff: change?.diff ? compactWholeFileUnifiedDiff(change.path || change.file || "file", change.diff) : change?.diff, })), }; } if (item.type !== "commandExecution") return item; let normalized = item; if ( typeof item.command === "string" && /^read\s+/i.test(item.command) && (item.command.includes("?") || /\.(?:sqlite3?|db3?):/i.test(item.command)) ) { const target = item.command.replace(/^read\s+/i, ""); const command = formatToolCommand("read", { path: target }); normalized = { ...item, command, commandActions: commandActionsForOmpTool("read", { path: target }, command, cwd), }; } if ((normalized.status === "completed" || normalized.status === "failed") && (normalized.durationMs == null || Number.isNaN(normalized.durationMs))) { normalized = { ...normalized, durationMs: 50 }; } return normalized;}
/** * Bumped whenever historical projection shape changes (per-file edit diffs, * `ask` cards, device write commands, answers the live stream dropped, ...). A * thread stored under an older version gets one full re-parse so already- * projected turns pick the new shape up; the append-only `_parsedSize` cursor * alone never revisits consumed lines. */const PROJECTION_REPAIR_VERSION = 17;
// old bridge versions stored every edit/write as a generic tool callconst LEGACY_GENERIC_FILE_TOOL = /^(?:edit|write|file|apply[_-]?patch|ast[_-]?edit)$/i;
function normalizeStoredThread(t) { if (!t) return t; const rawStatus = typeof t.status === "string" ? { type: t.status } : (t.status || { type: "idle" }); const status = rawStatus.type === "active" && !Array.isArray(rawStatus.activeFlags) ? { ...rawStatus, activeFlags: [] } : rawStatus; return { id: t.id, extra: t.extra ?? null, sessionId: t.sessionId || t.id, forkedFromId: t.forkedFromId ?? null, parentThreadId: t.parentThreadId ?? null, preview: t.preview || "", ephemeral: Boolean(t.ephemeral), section: t.section && typeof t.section === "object" ? t.section : null, sectionEnteredAt: t.sectionEnteredAt ?? null, projectId: t.projectId ?? null, historyMode: t.historyMode || "legacy", model: storedModelId(t), modelProvider: t.modelProvider || (typeof t.model === "object" ? t.model.provider : null) || "openai", modelSelector: t.modelSelector || (typeof t.model === "object" ? t.model.selector || null : null), reasoningEffort: t.reasoningEffort || null, createdAt: t.createdAt || unixSeconds(), updatedAt: t.updatedAt || t.createdAt || unixSeconds(), recencyAt: t.recencyAt || t.updatedAt || t.createdAt || unixSeconds(), status, path: t.path ?? null, cwd: path.resolve(t.cwd || process.cwd()), cliVersion: t.cliVersion || "0.145.0", source: t.source || "appServer", canAcceptDirectInput: t.canAcceptDirectInput ?? true, threadSource: t.threadSource ?? null, agentNickname: t.agentNickname ?? null, agentRole: t.agentRole ?? null, gitInfo: t.gitInfo ?? null, workspace: t.workspace ?? null, name: t.name ?? null, goal: t.goal && typeof t.goal === "object" ? structuredClone(t.goal) : null, turns: (Array.isArray(t.turns) ? t.turns : []).map((turn) => { if (!turn || !Array.isArray(turn.items)) return turn; return { ...turn, items: turn.items.map((item) => normalizeStoredItem(item, t.cwd)), }; }), };}
export function turnPromptText(turn) { if (!turn) return ""; const parts = []; for (const item of turn?.items || []) { if (item?.type !== "userMessage") continue; if (Array.isArray(item.content)) { for (const part of item.content) { if (typeof part === "string") { parts.push(part); } else if (part?.type === "text" && part.text != null) { parts.push(String(part.text)); } } } else if (typeof item.content === "string") { parts.push(item.content); } else if (item.text != null) { parts.push(String(item.text)); } } return parts.join("\n");}
export function canonicalizePromptText(prompt) { if (prompt === null || prompt === undefined) return ""; return String(prompt).replace(/\r\n/g, "\n").trim();}
export function extractGoalPayload(prompt) { if (!prompt) return null; const normalized = String(prompt).replace(/\r\n/g, "\n").trim(); const goalMatch = normalized.match(/^\s*<thread-goal(?:\s+[^>]*)?>[\s\S]*?<\/thread-goal>\s*([\s\S]*)$/i); if (!goalMatch) return null; let payload = goalMatch[1] || ""; payload = payload.replace(/^\s*Continue working toward this persisted thread goal while directly handling the user's message below\.?\s*/i, ""); return payload.trim();}
export function promptMatches(candidatePrompt, targetPrompt) { const normCandidate = canonicalizePromptText(candidatePrompt); const normTarget = canonicalizePromptText(targetPrompt); if (!normCandidate || !normTarget) return false; if (normCandidate === normTarget) return true;
const candidatePayload = extractGoalPayload(candidatePrompt); const targetPayload = extractGoalPayload(targetPrompt);
if (candidatePayload !== null && candidatePayload.length > 0) { if (candidatePayload === normTarget) return true; if (targetPayload !== null && candidatePayload === targetPayload) return true; } if (targetPayload !== null && targetPayload.length > 0) { if (targetPayload === normCandidate) return true; } return false;}
export function appendNewParsedTurns(storedTurns, parsedTurns, reconciliation = null) { if (parsedTurns.length <= storedTurns.length) return storedTurns; const prefixMatches = storedTurns.every( (turn, index) => promptMatches(turnPromptText(turn), turnPromptText(parsedTurns[index])) || (reconciliation && (reconciliation.turnId === turn?.id || (reconciliation.prompt && promptMatches(reconciliation.prompt, turnPromptText(parsedTurns[index]))))) ); return prefixMatches ? [...storedTurns, ...parsedTurns.slice(storedTurns.length)] : storedTurns;}
function mergeProjectedText(existing, parsed) { const current = String(existing || ""); const incoming = String(parsed || ""); if (!current) return incoming; if (!incoming || current === incoming || current.startsWith(incoming)) return current; if (incoming.startsWith(current)) return incoming; return `${current}\n\n${incoming}`;}
function mergeProjectedTurn(stored, parsed) { stored.items ||= []; for (const parsedItem of parsed?.items || []) { if (parsedItem?.type === "userMessage") continue; let existing = parsedItem?.id ? stored.items.find((item) => item?.id === parsedItem.id) : null; if (!existing && parsedItem?.type === "agentMessage") { existing = stored.items.find((item) => item?.type === "agentMessage" && (String(parsedItem.text || "").startsWith(String(item.text || "")) || String(item.text || "").startsWith(String(parsedItem.text || "")))); } if (!existing && parsedItem?.type === "agentMessage") { existing = stored.items.find((item) => item?.type === "agentMessage" && item.status === "inProgress"); } if (!existing && parsedItem?.type === "reasoning") { existing = stored.items.find((item) => item?.type === "reasoning"); } if (!existing) { stored.items.push(parsedItem); continue; } if (existing.type === "agentMessage" && parsedItem.type === "agentMessage") { const parsedIsTerminal = parsedItem.status === "completed" || parsedItem.status === "failed"; existing.text = parsedIsTerminal ? String(parsedItem.text || "") : mergeProjectedText(existing.text, parsedItem.text); } else if (existing.type === "reasoning" && parsedItem.type === "reasoning") { // the durable omp turn owns the one summary card and clears legacy raw copies existing.summary = Array.isArray(parsedItem.summary) ? [...parsedItem.summary] : []; existing.content = Array.isArray(parsedItem.content) ? [...parsedItem.content] : []; stored.items = stored.items.filter((item) => item?.type !== "reasoning" || item === existing); } else if (existing.type === "dynamicToolCall" && parsedItem.type !== "dynamicToolCall" && LEGACY_GENERIC_FILE_TOOL.test(String(existing.tool || ""))) { // an edit/write only lands here when it was projected by an older // bridge, so the parsed card wins stored.items[stored.items.indexOf(existing)] = parsedItem; continue; } else if (parsedItem.type === "userInputResponse" && existing.type !== "userInputResponse") { // Legacy ask projections were generic tool calls; the durable log now // yields the native ask card, so the parsed shape wins. stored.items[stored.items.indexOf(existing)] = parsedItem; continue; } else if (existing.type === "userInputResponse" && parsedItem.type === "userInputResponse") { if (Array.isArray(parsedItem.questions) && parsedItem.questions.length) existing.questions = parsedItem.questions; if (parsedItem.answers && Object.keys(parsedItem.answers).length) existing.answers = parsedItem.answers; // A durable card is always resolved; `false` would render a picker whose // submit target died with the in-flight request. existing.completed = true; } else if (existing.type === "fileChange" && parsedItem.type === "fileChange") { if (Array.isArray(parsedItem.changes) && parsedItem.changes.length) { existing.changes = parsedItem.changes; } } else { if ("aggregatedOutput" in parsedItem) { const parsedIsTerminal = parsedItem.status === "completed" || parsedItem.status === "failed"; existing.aggregatedOutput = parsedIsTerminal ? String(parsedItem.aggregatedOutput || "") : mergeProjectedText(existing.aggregatedOutput, parsedItem.aggregatedOutput); } if (Array.isArray(parsedItem.contentItems) && parsedItem.contentItems.length) { existing.contentItems = parsedItem.contentItems; } } for (const field of ["status", "exitCode", "durationMs", "success", "result", "error", "completedAt"]) { if (parsedItem[field] !== undefined && parsedItem[field] !== null) existing[field] = parsedItem[field]; } } stored.status = parsed.status || stored.status; stored.error = parsed.error ?? stored.error ?? null; stored.completedAt = parsed.completedAt || stored.completedAt; if (parsed.tokenUsage) stored.tokenUsage = parsed.tokenUsage; return stored;}
function applyParsedToolResults(turns, toolResults) { if (!Array.isArray(toolResults) || toolResults.length === 0) return; for (const result of toolResults) { const toolCallId = result?.toolCallId; if (!toolCallId) continue; let matchingItem = null; for (let turnIndex = turns.length - 1; turnIndex >= 0 && !matchingItem; turnIndex -= 1) { const items = Array.isArray(turns[turnIndex]?.items) ? turns[turnIndex].items : []; for (let itemIndex = items.length - 1; itemIndex >= 0; itemIndex -= 1) { if (items[itemIndex]?.id === toolCallId) { matchingItem = items[itemIndex]; break; } } } if (matchingItem) completeHistoricalToolItem(matchingItem, result); }}
// a live turn can stay open across two OMP turns, so the same call ends up in// the local blob and again in the turn the parse claims it forexport function dropRehomedItems(turns = [], claims = newParsedClaims()) { for (const turn of turns) { if (!Array.isArray(turn?.items)) continue; const owned = claims.byTurn.get(turn); const ownedAnswers = claims.answersByTurn.get(turn); const kept = new Set(); turn.items = turn.items.filter((item) => { if (!item?.id) return true; if (claims.all.has(item.id) && !owned?.has(item.id)) return false; const answer = agentMessageClaim(item); if (answer && claims.answers.has(answer) && !ownedAnswers?.has(answer)) return false; // one message id can carry both a reasoning and an agentMessage item, // and short call ids do get reused across a long session const key = `${item.id}\u0000${item.type}`; if (kept.has(key)) return false; kept.add(key); return true; }); } return turns;}
function agentMessageClaim(item) { if (item?.type !== "agentMessage") return null; const text = String(item.text || ""); return text.trim() ? text : null;}
function newParsedClaims() { return { all: new Set(), byTurn: new Map(), answers: new Set(), answersByTurn: new Map() };}
function recordParsedClaims(claims, storedTurn, parsedTurn) { let owned = claims.byTurn.get(storedTurn); if (!owned) { owned = new Set(); claims.byTurn.set(storedTurn, owned); } let ownedAnswers = claims.answersByTurn.get(storedTurn); if (!ownedAnswers) { ownedAnswers = new Set(); claims.answersByTurn.set(storedTurn, ownedAnswers); } for (const item of parsedTurn?.items || []) { if (item?.id) { claims.all.add(item.id); owned.add(item.id); } const answer = agentMessageClaim(item); if (answer) { claims.answers.add(answer); ownedAnswers.add(answer); } }}
export function appendParsedTurnSuffix(storedTurns, parsedTurns, reconciliation = null) { const merged = storedTurns.slice(); const claims = newParsedClaims(); for (const parsed of parsedTurns) { const parsedPrompt = turnPromptText(parsed); const last = merged.at(-1); const targetTurn = reconciliation?.turnId ? merged.find((turn) => turn?.id === reconciliation.turnId) || last : last; const pendingMatch = Boolean( parsedPrompt && targetTurn && reconciliation && (!reconciliation.turnId || reconciliation.turnId === targetTurn.id) && !reconciliation.matched && ( (reconciliation.prompt && promptMatches(parsedPrompt, reconciliation.prompt)) || promptMatches(parsedPrompt, turnPromptText(targetTurn)) ) ); const isStandaloneCompaction = (parsed?.items || []).some((item) => item?.type === "contextCompaction");
if (pendingMatch) { reconciliation.matched = true; mergeProjectedTurn(targetTurn, parsed); recordParsedClaims(claims, targetTurn, parsed); continue; } if (!parsedPrompt && last && !isStandaloneCompaction) { mergeProjectedTurn(last, parsed); recordParsedClaims(claims, last, parsed); continue; } merged.push(parsed); recordParsedClaims(claims, parsed, parsed); } return dropRehomedItems(merged, claims);}
export function reconcileParsedTurns(storedTurns, parsedTurns, reconciliation = null) { if (!Array.isArray(storedTurns) || storedTurns.length === 0) { return (parsedTurns || []).slice(); } const merged = storedTurns.slice(); const parsedList = Array.isArray(parsedTurns) ? parsedTurns : []; const minLen = Math.min(merged.length, parsedList.length); const claims = newParsedClaims(); let prefixMatches = true;
for (let i = 0; i < minLen; i += 1) { const storedTurn = merged[i]; const parsedTurn = parsedList[i]; const storedPrompt = turnPromptText(storedTurn); const parsedPrompt = turnPromptText(parsedTurn); const isPendingTurn = Boolean( reconciliation && !reconciliation.matched && (!reconciliation.turnId || reconciliation.turnId === storedTurn?.id) && parsedPrompt && ( (reconciliation.prompt && promptMatches(parsedPrompt, reconciliation.prompt)) || promptMatches(parsedPrompt, storedPrompt) ) );
if (promptMatches(storedPrompt, parsedPrompt) || isPendingTurn) { if (isPendingTurn && reconciliation) { reconciliation.matched = true; } mergeProjectedTurn(storedTurn, parsedTurn); recordParsedClaims(claims, storedTurn, parsedTurn); } else { prefixMatches = false; break; } }
if (!prefixMatches) { return parsedList.slice(); }
if (parsedList.length > merged.length) { for (let i = merged.length; i < parsedList.length; i += 1) { const parsedTurn = parsedList[i]; const parsedPrompt = turnPromptText(parsedTurn); const last = merged.at(-1); const targetTurn = reconciliation?.turnId ? merged.find((turn) => turn?.id === reconciliation.turnId) || last : last; const isPendingMatch = Boolean( parsedPrompt && targetTurn && reconciliation && (!reconciliation.turnId || reconciliation.turnId === targetTurn.id) && !reconciliation.matched && ( (reconciliation.prompt && promptMatches(parsedPrompt, reconciliation.prompt)) || promptMatches(parsedPrompt, turnPromptText(targetTurn)) ) ); if (isPendingMatch) { reconciliation.matched = true; mergeProjectedTurn(targetTurn, parsedTurn); recordParsedClaims(claims, targetTurn, parsedTurn); } else { merged.push(parsedTurn); recordParsedClaims(claims, parsedTurn, parsedTurn); } } }
return dropRehomedItems(merged, claims);}
function repairDuplicateThreadAliases(state) { let modified = false; const canonicalByAlias = new Map(); for (const [key, record] of Object.entries(state.threads || {})) { const canonicalId = record.thread?.id; if (!canonicalId || key === canonicalId) continue; canonicalByAlias.set(key, canonicalId); const canonical = state.threads[canonicalId]; if (!canonical) { state.threads[canonicalId] = record; } else { const score = (candidate) => { const thread = candidate?.thread || {}; return [ Number(thread.updatedAt || thread.recencyAt || 0), Array.isArray(thread.turns) ? thread.turns.length : 0, candidate?.backend?.sessionPath || candidate?.backend?.resume?.sessionPath ? 1 : 0, ]; }; const currentScore = score(canonical); const aliasScore = score(record); const aliasWins = aliasScore[0] > currentScore[0] || (aliasScore[0] === currentScore[0] && aliasScore[1] > currentScore[1]) || (aliasScore[0] === currentScore[0] && aliasScore[1] === currentScore[1] && aliasScore[2] > currentScore[2]); if (aliasWins) state.threads[canonicalId] = record; } delete state.threads[key]; modified = true; } const resolveAlias = (threadId) => { const seen = new Set(); let resolved = threadId; while (canonicalByAlias.has(resolved) && !seen.has(resolved)) { seen.add(resolved); resolved = canonicalByAlias.get(resolved); } return resolved; }; for (const record of Object.values(state.threads || {})) { const thread = record.thread; if (!thread) continue; if (thread.forkedFromId && canonicalByAlias.has(thread.forkedFromId)) { thread.forkedFromId = resolveAlias(thread.forkedFromId); modified = true; } if (thread.parentThreadId && canonicalByAlias.has(thread.parentThreadId)) { thread.parentThreadId = resolveAlias(thread.parentThreadId); modified = true; } } if (canonicalByAlias.size > 0) { state.order = Array.from(new Set( (state.order || []).map((threadId) => canonicalByAlias.get(threadId) || threadId) )).filter((threadId) => state.threads[threadId]); } return modified;}
function repairDanglingForkMetadata(state) { const entries = Object.entries(state.threads || {}); const knownIds = new Set(); for (const [key, record] of entries) { for (const value of [ key, record.thread?.id, record.thread?.sessionId, record.backend?.sessionId, record.backend?.resume?.sessionId, ]) { if (value) knownIds.add(value); } }
let modified = false; for (const [key, record] of entries) { const thread = record.thread; if (!thread) continue; const ownIds = new Set([ key, thread.id, thread.sessionId, record.backend?.sessionId, record.backend?.resume?.sessionId, ].filter(Boolean)); const forkedFromId = thread.forkedFromId ?? null; const parentThreadId = thread.parentThreadId ?? null; const invalidFork = forkedFromId && (ownIds.has(forkedFromId) || !knownIds.has(forkedFromId)); const invalidParent = parentThreadId && (ownIds.has(parentThreadId) || !knownIds.has(parentThreadId)); if (!invalidFork && !invalidParent) continue; if (invalidFork) thread.forkedFromId = null; if (invalidParent) thread.parentThreadId = null; if (thread.source === "fork" && thread.forkedFromId == null && thread.parentThreadId == null) { thread.source = "appServer"; } modified = true; } return modified;}
/** * Which copy of a thread wins when a second process wrote the file. * * Only called for threads we have no unwritten change to, so the disk copy is * normally the newer one - including when it carries an edit made in the same * second. A copy that lost turns is the one case where our own snapshot is * demonstrably further along. */function foreignRecordWins(mine, theirs) { const mineTurns = mine?.thread?.turns?.length || 0; const theirTurns = theirs?.thread?.turns?.length || 0; if (theirTurns !== mineTurns) return theirTurns > mineTurns; return Number(theirs?.thread?.updatedAt || 0) >= Number(mine?.thread?.updatedAt || 0);}
export class StateStore { #file; #sessionsDir; #state = null; #loadPromise = null; #writeChain = Promise.resolve(); #activeThreadIds = new Set(); #dirtyThreadIds = new Set(); #tombstones = new Set(); #fileStamp = null;
constructor( file = process.env.OMP_CODEX_STATE_FILE || path.join(os.homedir(), ".omp", "codex-bridge", "state.json"), sessionsDir = null ) { this.#file = file; this.#sessionsDir = sessionsDir; } get file() { return this.#file; } markThreadActive(threadId) { this.#activeThreadIds.add(threadId); } markThreadInactive(threadId) { this.#activeThreadIds.delete(threadId); }
async load() { if (this.#state) return this.#state; if (this.#loadPromise) return this.#loadPromise; this.#loadPromise = (async () => { try { const loaded = await readJson(this.#file, null); this.#state = loaded && typeof loaded === "object" ? { ...defaultState(), ...loaded } : defaultState(); let modified = false; for (const [id, r] of Object.entries(this.#state.threads || {})) { const t = r.thread?.name || r.thread?.preview || ""; if (t === "OMP Session" || t.startsWith("You are a helpful assistant") || t === "cli" || t === "test") { delete this.#state.threads[id]; this.#state.order = this.#state.order.filter((oId) => oId !== id); modified = true; } } if (this.#state.threads?.["thread_8fa6c13610174ea090cf0a7dcd1302ea"]?.archived) { this.#state.threads["thread_8fa6c13610174ea090cf0a7dcd1302ea"].archived = false; if (this.#state.threads["thread_8fa6c13610174ea090cf0a7dcd1302ea"].thread) { this.#state.threads["thread_8fa6c13610174ea090cf0a7dcd1302ea"].thread.archived = false; } modified = true; } if (repairDuplicateThreadAliases(this.#state)) modified = true; if (repairDanglingForkMetadata(this.#state)) modified = true; if (modified) { await atomicWriteJson(this.#file, this.#state).catch(() => {}); } await this.#stampFile(); return this.#state; } finally { this.#loadPromise = null; } })(); return this.#loadPromise; }
async #stampFile() { try { const stat = await fs.stat(this.#file); this.#fileStamp = { mtimeMs: stat.mtimeMs, size: stat.size }; } catch { this.#fileStamp = null; } }
/** * Take over records a second process wrote since our last write. * * One server should own this file, but an older Desktop-spawned stdio server * or a standalone run writes its own whole snapshot too, and whoever writes * last would otherwise delete every thread the other one created - the fork * ChatGPT just made included. Threads we are driving or have unwritten edits * for stay ours; for everything else the file is the better source, so we * merge before overwriting it. */ async #adoptForeignWrites() { let stat = null; try { stat = await fs.stat(this.#file); } catch { return; } if (this.#fileStamp && stat.mtimeMs === this.#fileStamp.mtimeMs && stat.size === this.#fileStamp.size) return; const disk = await readJson(this.#file, null); const foreign = disk?.threads; if (!foreign || typeof foreign !== "object") return; const adopted = []; for (const [threadId, record] of Object.entries(foreign)) { if (!record?.thread?.id || this.#tombstones.has(threadId)) continue; const mine = this.#state.threads[threadId]; if (mine && (this.#activeThreadIds.has(threadId) || this.#dirtyThreadIds.has(threadId) || !foreignRecordWins(mine, record))) continue; this.#state.threads[threadId] = record; adopted.push(threadId); } if (!adopted.length) return; const diskOrder = Array.isArray(disk.order) ? disk.order : []; this.#state.order = [...new Set([...this.#state.order, ...diskOrder])] .filter((threadId) => this.#state.threads[threadId]); debug("adopted threads from a concurrent writer", adopted.length); }
async flush() { await this.load(); this.#writeChain = this.#writeChain .catch(() => {}) .then(async () => { await this.#adoptForeignWrites(); const written = [...this.#dirtyThreadIds]; await atomicWriteJson(this.#file, this.#state); for (const threadId of written) this.#dirtyThreadIds.delete(threadId); await this.#stampFile(); }); return this.#writeChain; }
async createThread(options) { const state = await this.load(); const thread = makeThread(options); state.threads[thread.id] = { thread, backend: {}, approvalCache: {}, archived: false }; state.order = [thread.id, ...state.order.filter((id) => id !== thread.id)]; this.#dirtyThreadIds.add(thread.id); this.#tombstones.delete(thread.id); await this.flush(); return structuredClone(thread); }
async getRecord(threadId) { const state = await this.load(); let existing = state.threads[threadId]; if (!existing) { existing = Object.values(state.threads).find( (r) => r.thread?.sessionId === threadId || r.backend?.sessionId === threadId || r.backend?.resume?.sessionId === threadId ); } if (existing) { const ownedByThisStore = this.#activeThreadIds.has(existing.thread.id) || this.#activeThreadIds.has(threadId); const recoveringOwnedTurn = existing.thread?.status?.type === "active" && ownedByThisStore; const sessionPath = existing.backend?.resume?.sessionPath || existing.backend?.sessionPath; const projectOmpSession = usesOmpSessionProjection(existing.backend); let sessionChangedBeforeSync = false; if (existing.thread?.status?.type === "active") { const activePid = Number(existing._activePid || 0); const externalActiveUntil = activePid ? 0 : Number(existing._externalActiveUntil || 0); if (ownedByThisStore) return existing; if (externalActiveUntil > 0 && externalActiveUntil <= Date.now()) { delete existing._externalActiveUntil; existing.thread.status = { type: "idle" }; } else if (!sessionPath) { const updatedAt = Number(existing.thread.updatedAt || existing.thread.recencyAt || 0); const ownerAlive = processIsAlive(activePid); const recentlyActive = updatedAt > 0 && unixSeconds() - updatedAt <= 60; if ((activePid && ownerAlive && recentlyActive) || (!activePid && recentlyActive)) { return existing; } existing._activePid = null; existing.thread.status = { type: "idle" }; } } if (sessionPath && projectOmpSession) { try { const stat = await fs.stat(sessionPath).catch(() => null); const sessionMoved = Boolean( stat && (stat.size !== existing._parsedSize || stat.mtimeMs !== existing._parsedMtime || !existing._parsed) ); const needsProjectionRepair = existing._projectionRepairVersion !== PROJECTION_REPAIR_VERSION; if (stat && (sessionMoved || needsProjectionRepair)) { sessionChangedBeforeSync = sessionMoved; const appendOnly = Boolean( !needsProjectionRepair && existing._parsed && Number(existing._parsedSize) > 0 && stat.size > Number(existing._parsedSize) ); const nativeSessionId = existing.backend?.resume?.sessionId || existing.backend?.sessionId || existing.thread?.sessionId || existing.thread?.id; const parsedResult = await parseOmpSessionFile(sessionPath, appendOnly ? { startOffset: existing._parsedSize, sessionInfo: { id: nativeSessionId, cwd: existing.thread?.cwd || process.cwd(), timestamp: existing.thread?.createdAt ? new Date(Number(existing.thread.createdAt) * 1000).toISOString() : null, title: existing.thread?.name || null, }, title: existing.thread?.name || null, preview: existing.thread?.preview || "", activeModel: existing.thread?.modelSelector || existing.thread?.model?.selector || null, activeThinking: existing.thread?.reasoningEffort || null, createdAt: existing.thread?.createdAt || 0, } : {}); const parsed = parsedResult.thread; if (parsed) { const storedTurns = Array.isArray(existing.thread.turns) ? existing.thread.turns : []; const reconciliation = existing._pendingTurnReconciliation || null; if (appendOnly) { existing.thread.turns = appendParsedTurnSuffix(storedTurns, parsed.turns || [], reconciliation); if (reconciliation?.matched) delete existing._pendingTurnReconciliation; } else { const isNativeOmpThread = existing.thread.id === parsed.id; if (storedTurns.length === 0 || isNativeOmpThread) { existing.thread.turns = parsed.turns || []; } else { existing.thread.turns = reconcileParsedTurns(storedTurns, parsed.turns || [], reconciliation); } if (reconciliation?.matched) { delete existing._pendingTurnReconciliation; } else if (isNativeOmpThread || storedTurns.length === 0) { delete existing._pendingTurnReconciliation; } } applyParsedToolResults(existing.thread.turns, parsedResult.toolResults); if (parsed.name && parsed.name !== "OMP Session" && (!existing.thread.name || existing.thread.name === "OMP Session")) { existing.thread.name = parsed.name; } if (parsed.preview && parsed.preview !== "OMP Session") existing.thread.preview = parsed.preview; const preserveBackendModel = Boolean(existing.backend?.backendTag && existing.thread.model); if (parsed.model && !preserveBackendModel) existing.thread.model = parsed.model; if (parsed.modelProvider) existing.thread.modelProvider = parsed.modelProvider; if (parsed.modelSelector && !preserveBackendModel) existing.thread.modelSelector = parsed.modelSelector; if (parsed.reasoningEffort) existing.thread.reasoningEffort = parsed.reasoningEffort; existing.thread.updatedAt = parsed.updatedAt || existing.thread.updatedAt; existing.thread.recencyAt = parsed.recencyAt || existing.thread.recencyAt; const ownedByThisStore = this.#activeThreadIds.has(existing.thread.id) || this.#activeThreadIds.has(threadId); if (!ownedByThisStore && !recoveringOwnedTurn) { existing.thread.status = parsed.status || { type: "idle" }; if (parsedResult.activityPending && parsedResult.externalActiveUntil > Date.now()) { existing._externalActiveUntil = parsedResult.externalActiveUntil; } else { delete existing._externalActiveUntil; } } existing._parsed = true; existing._projectionRepairVersion = PROJECTION_REPAIR_VERSION; existing._parsedMtime = parsedResult.parsedMtime ?? stat.mtimeMs; existing._parsedSize = parsedResult.parsedSize ?? stat.size; await this.flush(); } } } catch (error) { debug("Failed to synchronize OMP session", sessionPath, error); } } if (sessionPath && projectOmpSession && !ownedByThisStore) { const sessionStat = await fs.stat(sessionPath).catch(() => null); const pending = existing._pendingTurnReconciliation; const ownerPid = Number(existing._activePid || 0); const ownerAlive = processIsAlive(ownerPid); const sessionChanged = sessionChangedBeforeSync; const updatedAt = Number(existing.thread.updatedAt || existing.thread.recencyAt || 0); const awaitingFirstWrite = ownerAlive && pending && sessionStat && Number(pending.parsedSize ?? -1) >= sessionStat.size && updatedAt > 0 && unixSeconds() - updatedAt <= 60; if (ownerPid && !ownerAlive && !sessionChanged) { existing.thread.status = { type: "idle" }; existing._activePid = null; } else if (awaitingFirstWrite) { existing.thread.status = { type: "active", activeFlags: [] }; } else { const activity = await inspectOmpSessionActivity(sessionPath); existing.thread.status = activity.status; if (activity.pending && activity.activeUntil > Date.now()) { existing._externalActiveUntil = activity.activeUntil; } else { delete existing._externalActiveUntil; existing._activePid = null; } } } return existing; } const prime = await findPrimeSession(threadId).catch(() => null); if (prime?.file) { const stat = await fs.stat(prime.file).catch(() => null); const { thread } = await parseOmpSessionFile(prime.file); const record = { thread, backend: { kind: "prime-rpc", external: true, activeSessionId: prime.activeSessionId || null, sessionId: thread.id, sessionPath: prime.file, resume: { sessionId: thread.id, sessionPath: prime.file }, }, approvalCache: {}, archived: false, _parsed: true, _projectionRepairVersion: PROJECTION_REPAIR_VERSION, _parsedMtime: stat?.mtimeMs ?? null, _parsedSize: stat?.size ?? null, }; state.threads[thread.id] = record; if (!state.order.includes(thread.id)) state.order.push(thread.id); await this.flush(); return record; } try { const baseDir = this.#sessionsDir || defaultSessionsDir(); const entries = await fs.readdir(baseDir, { withFileTypes: true }).catch(() => []); const dirNames = entries.filter((e) => e.isDirectory()).map((e) => e.name); const matchPromises = dirNames.map(async (dirName) => { const dirPath = path.join(baseDir, dirName); const files = await fs.readdir(dirPath, { withFileTypes: true }).catch(() => []); const match = files.find((f) => f.isFile() && f.name.endsWith(`_${threadId}.jsonl`)); return match ? path.join(dirPath, match.name) : null; }); const matchedPaths = (await Promise.all(matchPromises)).filter(Boolean); if (matchedPaths.length > 0) { const filePath = matchedPaths[0]; const stat = await fs.stat(filePath).catch(() => null); const { thread } = await parseOmpSessionFile(filePath); const record = { thread, backend: { resume: { sessionId: thread.id, sessionPath: filePath } }, approvalCache: {}, archived: false, _parsed: true, _parsedMtime: stat?.mtimeMs ?? null, _parsedSize: stat?.size ?? null, }; state.threads[thread.id] = record; if (!state.order.includes(thread.id)) state.order.push(thread.id); await this.flush(); return record; } } catch {} return null; }
async getThread(threadId) { const record = await this.getRecord(threadId); return record ? normalizeStoredThread(record.thread) : null; } async getExternalThreadStatus(threadId) { const state = await this.load(); const record = state.threads[threadId] || Object.values(state.threads).find( (candidate) => candidate.thread?.id === threadId || candidate.thread?.sessionId === threadId || candidate.backend?.sessionId === threadId || candidate.backend?.resume?.sessionId === threadId ); if (!record) return null; if (this.#activeThreadIds.has(threadId) || this.#activeThreadIds.has(record.thread?.id)) { return normalizeStoredThread(record.thread).status; } const sessionPath = record.backend?.resume?.sessionPath || record.backend?.sessionPath; if (!usesOmpSessionProjection(record.backend)) return normalizeStoredThread(record.thread).status; if (!sessionPath) return normalizeStoredThread(record.thread).status; const activity = await inspectOmpSessionActivity(sessionPath); record.thread.status = activity.status; if (activity.pending && activity.activeUntil > Date.now()) { record._externalActiveUntil = activity.activeUntil; } else { delete record._externalActiveUntil; } return normalizeStoredThread(record.thread).status; } async updateThread(threadId, updater, { flush = true, clone = true } = {}) { const record = await this.getRecord(threadId); if (!record) return null; await updater(record.thread, record); record.thread.updatedAt = unixSeconds(); record.thread.recencyAt = record.thread.updatedAt; const state = await this.load(); const canonicalId = record.thread?.id || threadId; state.threads[canonicalId] = record; if (threadId !== canonicalId) delete state.threads[threadId]; if (threadId !== canonicalId) { for (const candidate of Object.values(state.threads)) { if (candidate.thread?.forkedFromId === threadId) candidate.thread.forkedFromId = canonicalId; if (candidate.thread?.parentThreadId === threadId) candidate.thread.parentThreadId = canonicalId; } } state.order = [canonicalId, ...state.order.filter((id) => id !== canonicalId && id !== threadId)]; this.#dirtyThreadIds.add(canonicalId); if (threadId !== canonicalId) this.#tombstones.add(threadId); if (flush) await this.flush(); return clone ? structuredClone(record.thread) : record.thread; }
async setBackendState(threadId, backend) { const record = await this.getRecord(threadId); if (!record) return null; const nextBackend = { ...record.backend, ...backend }; if (JSON.stringify(nextBackend) === JSON.stringify(record.backend)) return null; return this.updateThread(threadId, (_thread, current) => { current.backend = nextBackend; }, { clone: false }); }
async listThreads({ cursor = null, limit = 50, archived = false, cwd = null, sortKey = "recency_at", sortDirection = "desc" } = {}) { const state = await this.load(); const cap = Math.max(1, Math.min(200, Number(limit || 50))); const cwdFilterList = Array.isArray(cwd) ? cwd.map((c) => path.resolve(c)) : cwd ? [path.resolve(cwd)] : null;
// Discover native OMP and Prime sessions. let nativeSessions = []; if (cwdFilterList && cwdFilterList.length > 0) { const results = await Promise.all(cwdFilterList.flatMap((c) => [ listOmpSessionsForCwd(c, {}, this.#sessionsDir).catch(() => ({ sessions: [] })), listPrimeSessionsForCwd(c).catch(() => ({ sessions: [] })), ])); nativeSessions = results.flatMap((r) => r.sessions); } else { const [omp, prime] = await Promise.all([ listAllOmpSessions({}, this.#sessionsDir).catch(() => ({ sessions: [] })), listAllPrimeSessions().catch(() => ({ sessions: [] })), ]); nativeSessions = [...omp.sessions, ...prime.sessions]; }
const knownSessionIds = new Set(); const knownSessionPaths = new Set(); const recordsBySessionId = new Map(); const recordsBySessionPath = new Map();
for (const [id, rec] of Object.entries(state.threads)) { knownSessionIds.add(id); if (rec.thread?.id) knownSessionIds.add(rec.thread.id); if (rec.thread?.sessionId) knownSessionIds.add(rec.thread.sessionId); if (rec.backend?.sessionId) knownSessionIds.add(rec.backend.sessionId); if (rec.backend?.resume?.sessionId) knownSessionIds.add(rec.backend.resume.sessionId);
if (rec.backend?.sessionPath) knownSessionPaths.add(path.resolve(rec.backend.sessionPath)); if (rec.backend?.resume?.sessionPath) knownSessionPaths.add(path.resolve(rec.backend.resume.sessionPath)); for (const sessionId of [ id, rec.thread?.id, rec.thread?.sessionId, rec.backend?.sessionId, rec.backend?.resume?.sessionId, ].filter(Boolean)) { recordsBySessionId.set(sessionId, rec); } for (const sessionPath of [ rec.backend?.sessionPath, rec.backend?.resume?.sessionPath, ].filter(Boolean)) { recordsBySessionPath.set(path.resolve(sessionPath), rec); } }
const pendingBridgeThreads = Object.values(state.threads) .map((record) => record.thread && !record.backend?.sessionId && !record.backend?.resume?.sessionId && record.thread.status?.type === "active" ? { cwd: path.resolve(record.thread.cwd), createdAt: Number(record.thread.createdAt || 0) } : null) .filter(Boolean);
const threadsById = new Map();
// 1. Add all state.threads (bridge state has precedence for active threads / renamed titles / archived status) for (const [id, record] of Object.entries(state.threads)) { threadsById.set(id, record); }
// 2. Add unmigrated native sessions from disk that aren't already managed by bridge. for (const s of nativeSessions) { const managedRecord = recordsBySessionId.get(s.id) || (s.file ? recordsBySessionPath.get(path.resolve(s.file)) : null); if (managedRecord) { const locallyActive = this.#activeThreadIds.has(managedRecord.thread?.id); if (s.backendKind === "prime-rpc") { managedRecord.backend = { ...managedRecord.backend, kind: "prime-rpc", external: managedRecord.backend?.external ?? true, activeSessionId: s.activeSessionId || managedRecord.backend?.activeSessionId || null, }; } if (!locallyActive) { managedRecord.thread.status = s.status || { type: "idle" }; if (s.activityPending && s.externalActiveUntil > Date.now()) { managedRecord._externalActiveUntil = s.externalActiveUntil; } else { delete managedRecord._externalActiveUntil; } } continue; } if (knownSessionIds.has(s.id)) continue; if (s.file && knownSessionPaths.has(path.resolve(s.file))) continue; const nativeCwd = path.resolve(s.cwd || ""); if (pendingBridgeThreads.some((pending) => { const nativeCreatedAt = Number(s.createdAt || 0); return pending.cwd === nativeCwd && nativeCreatedAt >= pending.createdAt - 1 && nativeCreatedAt <= pending.createdAt + 60; })) continue;
threadsById.set(s.id, { id: s.id, sessionId: s.id, archived: false, backend: s.backendKind === "prime-rpc" ? { kind: "prime-rpc", external: true, activeSessionId: s.activeSessionId || null, sessionId: s.id, sessionPath: s.file, resume: { sessionId: s.id, sessionPath: s.file }, } : { resume: { sessionId: s.id, sessionPath: s.file } }, approvalCache: {}, _externalActiveUntil: s.activityPending ? s.externalActiveUntil : null, thread: makeThread({ id: s.id, sessionId: s.id, cwd: s.cwd, source: "appServer", name: s.title || null, preview: s.preview || s.title || "", model: s.model, modelProvider: s.modelProvider, reasoningEffort: s.reasoningEffort, createdAt: s.createdAt, updatedAt: s.updatedAt, recencyAt: s.recencyAt, status: s.status || { type: "idle" }, turns: [], }), }); }
const allRecords = Array.from(threadsById.values());
let rows = allRecords .filter((record) => !record.thread.ephemeral) .filter((record) => Boolean(record.archived || record.thread.archived) === Boolean(archived)) .filter((record) => { const title = (record.thread.name || record.thread.preview || "").trim(); if (title === "OMP Session" || title === "cli" || title === "test") return false; if (title.startsWith("# Overview") || title.startsWith("You are a helpful assistant") || title.includes("short title for a task")) return false; const isFromDisk = !state.threads[record.id] && !state.threads[record.thread?.id]; if (isFromDisk && !title) return false; return true; }) .filter((record) => { if (!cwdFilterList) return true; const threadCwd = path.resolve(record.thread.cwd || ""); const sourceCwd = record.thread.workspace?.sourceCwd ? path.resolve(record.thread.workspace.sourceCwd) : null; const rootCwd = record.thread.workspace?.root ? path.resolve(record.thread.workspace.root) : null; return cwdFilterList.some((c) => { if (c === threadCwd || (sourceCwd && c === sourceCwd) || (rootCwd && c === rootCwd)) return true; if ((threadCwd.includes("/.codex/worktrees/") || threadCwd.includes("/.omp/codex-bridge/workspaces/")) && path.basename(threadCwd) === path.basename(c)) { return true; } return false; }); }); if (sortKey) { const dir = sortDirection === "asc" ? 1 : -1; rows.sort((a, b) => { const tA = a.thread; const tB = b.thread; if (sortKey === "created_at") return ((tA.createdAt || 0) - (tB.createdAt || 0)) * dir; if (sortKey === "updated_at") return ((tA.updatedAt || 0) - (tB.updatedAt || 0)) * dir; return ((tA.recencyAt || tA.updatedAt || 0) - (tB.recencyAt || tB.updatedAt || 0)) * dir; }); }
const { items: pageRows, nextCursor } = paginateArray(rows, { limit: cap, cursor }, true); const data = pageRows.map((record) => ({ ...normalizeStoredThread(record.thread), turns: [] })); return { data, nextCursor, backwardsCursor: null }; }
async renameThread(threadId, name) { return this.updateThread(threadId, (thread) => { thread.name = name == null ? null : String(name); }); }
async archiveThread(threadId, archived = true) { return this.updateThread(threadId, (thread, record) => { record.archived = archived; thread.archived = archived; }); }
async deleteThread(threadId) { const state = await this.load(); if (!state.threads[threadId]) return false; delete state.threads[threadId]; state.order = state.order.filter((id) => id !== threadId); // A merge would otherwise read the other process's copy back in. this.#tombstones.add(threadId); this.#dirtyThreadIds.delete(threadId); await this.flush(); return true; } async rollbackThread(threadId, numTurns = 1) { return this.updateThread(threadId, (thread) => { const turns = Array.isArray(thread.turns) ? thread.turns : []; const count = Math.max(1, Number(numTurns || 1)); thread.turns = turns.slice(0, Math.max(0, turns.length - count)); }); }
async addTurn(threadId, turn) { return this.updateThread(threadId, (thread) => { thread.turns ||= []; thread.turns.push(structuredClone(turn)); }); }
async updateTurn(threadId, turnId, updater, { flush = true, clone = true } = {}) { let result = null; await this.updateThread(threadId, (thread) => { const turn = thread.turns?.find((candidate) => candidate.id === turnId); if (!turn) return; updater(turn); turn.updatedAt = unixSeconds(); turn.recencyAt = turn.updatedAt; result = clone ? structuredClone(turn) : turn; }, { flush, clone: false }); return result; }
async addItem(threadId, turnId, item, options = {}) { return this.updateTurn(threadId, turnId, (turn) => { turn.items ||= []; turn.items.push(structuredClone(item)); }, options); }
async updateItem(threadId, turnId, itemId, updater, { flush = true, clone = true } = {}) { let updatedItem = null; await this.updateTurn(threadId, turnId, (turn) => { const item = turn.items?.find((candidate) => candidate.id === itemId); if (item) { updater(item); updatedItem = clone ? structuredClone(item) : item; } }, { flush, clone: false }); return updatedItem; }
async setPreview(threadId, text) { return this.updateThread(threadId, (thread) => { if (!thread.preview) thread.preview = safePreview(text); }); }
async getApproval(threadId, key) { const record = await this.getRecord(threadId); return record?.approvalCache?.[key] || null; }
async rememberApproval(threadId, key, decision) { return this.updateThread(threadId, (_thread, record) => { record.approvalCache ||= {}; record.approvalCache[key] = decision; }); }}