Something went wrong. Try again.
presents omp as codex for use with chatgpt remote
Something went wrong. Try again.
9.3 kB · 293 lines
JavaScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294/** * Client for the OMP session control socket. * * The bridge can observe an externally driven thread by tailing its session * JSONL, but it cannot write to one. When the OMP side is running the * `extension/control-socket.mjs` extension, each live session exposes a Unix * socket keyed by session id, which turns "someone else is driving this * thread" from read-only into steerable. * * Every call is best-effort: no socket means the session is not running the * extension (or is not running at all), which callers treat as "no control * available" rather than as an error. */import net from "node:net";import os from "node:os";import path from "node:path";
const DEFAULT_TIMEOUT_MS = 2_000;
export function controlSocketDir() { return process.env.OMP_CONTROL_SOCKET_DIR || path.join(os.homedir(), ".omp", "codex-bridge", "control");}
export function controlSocketPath(sessionId) { return path.join(controlSocketDir(), `${sessionId}.sock`);}
/** * Send one request and resolve its reply, or null when the session offers no * control socket. */export function controlRequest(sessionId, request, { timeoutMs = DEFAULT_TIMEOUT_MS } = {}) { if (!sessionId) return Promise.resolve(null); const socketPath = controlSocketPath(sessionId); return new Promise((resolve) => { let settled = false; let buffer = ""; const finish = (value) => { if (settled) return; settled = true; socket.destroy(); clearTimeout(timer); resolve(value); }; const socket = net.connect(socketPath); const timer = setTimeout(() => finish(null), timeoutMs); timer.unref?.(); socket.setEncoding("utf8"); socket.once("connect", () => socket.write(`${JSON.stringify(request)}\n`)); socket.on("data", (chunk) => { buffer += chunk; const newline = buffer.indexOf("\n"); if (newline < 0) return; try { finish(JSON.parse(buffer.slice(0, newline))); } catch { finish(null); } }); socket.on("error", () => finish(null)); socket.on("close", () => finish(null)); });}
/** Live session snapshot, or null when the session is not controllable. */export async function controlStatus(sessionId, options = {}) { const reply = await controlRequest(sessionId, { op: "status" }, options); return reply?.ok ? reply : null;}
/** True once the running session accepted the message. */export async function controlSend(sessionId, text, { deliverAs = "steer", ...options } = {}) { const reply = await controlRequest(sessionId, { op: deliverAs, text }, options); return Boolean(reply?.ok);}
/** True once the running session accepted model/thinking overrides. */export async function controlConfigure(sessionId, { model = null, reasoningEffort = null } = {}, options = {}) { if (!model && !reasoningEffort) return true; const reply = await controlRequest(sessionId, { op: "configure", model, reasoningEffort }, options); return Boolean(reply?.ok);}
export async function controlRespondAsk(sessionId, requestId, answers, options = {}) { const reply = await controlRequest(sessionId, { op: "ask_response", requestId, answers }, options); return Boolean(reply?.ok);}
export async function controlCancelAsk(sessionId, requestId, options = {}) { const reply = await controlRequest(sessionId, { op: "ask_cancel", requestId }, options); return Boolean(reply?.ok);}
/** True once the running session accepted the interrupt. */export async function controlInterrupt(sessionId, options = {}) { const reply = await controlRequest(sessionId, { op: "interrupt" }, options); return Boolean(reply?.ok);}
/** * Ask a live session to compact its own context. * * Compaction runs far longer than a control round trip, so the session only * acknowledges that it started; the compaction itself shows up as a lifecycle * event and in the session log. */export async function controlCompact(sessionId, { instructions = null, ...options } = {}) { const reply = await controlRequest(sessionId, { op: "compact", instructions }, options); return Boolean(reply?.ok);}
/** * Subscribe to live session lifecycle events. * * Performs handshake, consumes ordered frames, deduplicates sequence numbers, * and calls onEvent for each validated frame and onDisconnect when the connection closes. * * Resolves to a subscription handle { streamId, sequence, status, reset, close } * or null if the session offers no control socket or the handshake fails. */export function controlSubscribe( sessionId, { onEvent = null, onDisconnect = null, streamId = null, afterSequence = null, timeoutMs = DEFAULT_TIMEOUT_MS, signal = null, } = {}) { if (!sessionId || signal?.aborted) return Promise.resolve(null); const socketPath = controlSocketPath(sessionId);
return new Promise((resolve) => { let settled = false; let closed = false; let disconnectedCalled = false; let buffer = ""; let handle = null; let lastSeenSequence = -1;
const socket = net.connect(socketPath);
const timer = setTimeout(() => { if (!settled) { finish(null); } }, timeoutMs); timer.unref?.();
const finish = (val) => { if (settled) return; settled = true; clearTimeout(timer); signal?.removeEventListener("abort", onAbort); if (!val) { socket.destroy(); } resolve(val); }; const onAbort = () => finish(null); signal?.addEventListener("abort", onAbort, { once: true });
const triggerDisconnect = () => { if (disconnectedCalled) return; disconnectedCalled = true; clearTimeout(timer); signal?.removeEventListener("abort", onAbort); if (!settled) { settled = true; resolve(null); } onDisconnect?.(); };
socket.setEncoding("utf8");
socket.once("connect", () => { const subscribeReq = { op: "subscribe", streamId: streamId ?? undefined, afterSequence: afterSequence ?? undefined, }; socket.write(`${JSON.stringify(subscribeReq)}\n`); });
socket.on("data", (chunk) => { buffer += chunk; let newline = buffer.indexOf("\n"); while (newline >= 0) { const line = buffer.slice(0, newline).trim(); buffer = buffer.slice(newline + 1); if (line) { let msg; try { msg = JSON.parse(line); } catch { newline = buffer.indexOf("\n"); continue; }
if (!handle) { // Expecting subscription ACK handshake if (msg && msg.ok && msg.op === "subscribe") { const ackStreamId = typeof msg.streamId === "string" ? msg.streamId : ""; const ackSequence = typeof msg.sequence === "number" ? msg.sequence : 0; const isReset = Boolean(msg.reset);
if (!isReset && typeof afterSequence === "number" && Number.isInteger(afterSequence) && afterSequence >= 0) { lastSeenSequence = afterSequence; } else { lastSeenSequence = ackSequence; }
let currentSeq = lastSeenSequence; let currentStatus = msg.status ?? null; let currentStreamId = ackStreamId;
handle = { get streamId() { return currentStreamId; }, set streamId(val) { currentStreamId = val; }, get sequence() { return currentSeq; }, set sequence(val) { currentSeq = val; }, get status() { return currentStatus; }, set status(val) { currentStatus = val; }, pendingAsks: Array.isArray(msg.pendingAsks) ? msg.pendingAsks : [], headSequence: ackSequence, reset: isReset, close() { if (closed) return; closed = true; socket.destroy(); }, };
finish(handle); } else { finish(null); return; } } else { // Consuming subsequent event frames if (msg && msg.type === "event") { if ( msg.streamId === handle.streamId && typeof msg.sequence === "number" && msg.sequence > lastSeenSequence ) { lastSeenSequence = msg.sequence; handle.sequence = msg.sequence; if (msg.status !== undefined) { handle.status = msg.status; } try { onEvent?.(msg); } catch { // User callback error should not crash client loop } } } } } newline = buffer.indexOf("\n"); } });
socket.on("error", () => { triggerDisconnect(); });
socket.on("close", () => { triggerDisconnect(); });
socket.on("end", () => { triggerDisconnect(); }); });}