Something went wrong. Try again.
presents omp as codex for use with chatgpt remote
Something went wrong. Try again.
40 kB · 1071 lines
JavaScript
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072import test from "node:test";import assert from "node:assert/strict";import fs from "node:fs/promises";import net from "node:net";import os from "node:os";import path from "node:path";import { makeHarness } from "./helpers.mjs";import { controlConfigure, controlInterrupt, controlSend, controlStatus, controlRespondAsk, controlCancelAsk, controlSubscribe, controlSocketPath,} from "../src/control-client.mjs";import controlSocketExtension from "../extension/control-socket.mjs";
/** Minimal stand-in for a live OMP session's control socket. */async function fakeSession(sessionId, { reply = null } = {}) { const received = []; const socketPath = controlSocketPath(sessionId); await fs.mkdir(path.dirname(socketPath), { recursive: true, mode: 0o700 }); const server = net.createServer((socket) => { socket.setEncoding("utf8"); socket.on("data", (chunk) => { for (const line of chunk.split("\n")) { if (!line.trim()) continue; const request = JSON.parse(line); received.push(request); const response = reply ? reply(request) : { ok: true, op: request.op }; if (response) socket.write(`${JSON.stringify(response)}\n`); } }); }); await new Promise((resolve) => server.listen(socketPath, resolve)); return { received, close: () => new Promise((resolve) => server.close(resolve)), };}
async function withControlDir(fn) { const previous = process.env.OMP_CONTROL_SOCKET_DIR; const dir = await fs.mkdtemp(path.join(os.tmpdir(), "omp-control-test-")); process.env.OMP_CONTROL_SOCKET_DIR = dir; try { await fn(dir); } finally { if (previous !== undefined) process.env.OMP_CONTROL_SOCKET_DIR = previous; else delete process.env.OMP_CONTROL_SOCKET_DIR; await fs.rm(dir, { recursive: true, force: true }).catch(() => {}); }}
function createMockOmp(sessionId = "ext-session", overrides = {}) { const sent = []; const configured = []; let aborted = 0; let idle = true; let pendingMessages = false; let currentModelId = "test-model"; const handlers = new Map(); const tools = new Map(); const models = [ { provider: "anthropic", id: "claude-opus-5" }, { provider: "google-antigravity", id: "gemini-3.7-flash" }, { provider: "nanbeige-local", id: "MercuriusDream/Nanbeige4.2-3B-mlx-8bit" }, ]; const api = { on: (event, handler) => { if (!handlers.has(event)) handlers.set(event, []); handlers.get(event).push(handler); }, registerTool: (tool) => tools.set(tool.name, tool), sendUserMessage: (content, options) => sent.push({ content, options }), setModel: async (model) => { configured.push({ model }); currentModelId = model.id; ctx.model = model; return true; }, setThinkingLevel: async (reasoningEffort) => { configured.push({ reasoningEffort }); }, }; const ctx = { isIdle: () => idle, hasPendingMessages: () => pendingMessages, abort: () => { aborted += 1; }, cwd: "/tmp/project", model: { id: currentModelId }, modelRegistry: { getAvailable: () => models, find: (provider, modelId) => models.find((model) => model.provider === provider && model.id === modelId), }, sessionManager: { getSessionId: () => sessionId, getSessionFile: () => `/tmp/sessions/${sessionId}.jsonl`, }, ...overrides, };
const emit = (event, payload) => { const list = handlers.get(event) || []; for (const fn of list) fn(payload, ctx); };
const start = async () => { controlSocketExtension(api); const startHandlers = handlers.get("session_start") || []; for (const fn of startHandlers) await fn({ type: "session_start" }, ctx); };
const stop = async () => { const stopHandlers = handlers.get("session_shutdown") || []; for (const fn of stopHandlers) await fn({ type: "session_shutdown" }, ctx); };
return { api, ctx, sent, configured, models, tools, emit, start, stop, get aborted() { return aborted; }, set idle(val) { idle = val; }, set pendingMessages(val) { pendingMessages = val; }, };}
test("control client reports no control for a session without a socket", () => withControlDir(async () => { assert.equal(await controlStatus("missing-session"), null); assert.equal(await controlSend("missing-session", "hello"), false); assert.equal(await controlInterrupt("missing-session"), false); assert.equal(await controlSubscribe("missing-session"), null);}));
test("control client round-trips status, configure, steer and interrupt", () => withControlDir(async () => { const session = await fakeSession("live-session", { reply: (request) => (request.op === "status" ? { ok: true, op: "status", idle: false, sessionId: "live-session", pid: 4242 } : { ok: true, op: request.op }), }); try { const status = await controlStatus("live-session"); assert.equal(status.pid, 4242); assert.equal(status.idle, false);
assert.equal(await controlConfigure("live-session", { model: "anthropic/claude-opus-5", reasoningEffort: "high", }), true); assert.equal(await controlSend("live-session", "go left", { deliverAs: "steer" }), true); assert.equal(await controlInterrupt("live-session"), true);
assert.deepEqual(session.received.map((r) => r.op), ["status", "configure", "steer", "interrupt"]); assert.equal(session.received[1].model, "anthropic/claude-opus-5"); assert.equal(session.received[2].text, "go left"); } finally { await session.close(); }}));
test("the extension answers status, configuration, steering and interrupt over its socket", () => withControlDir(async () => { const mock = createMockOmp("ext-session"); await mock.start();
try { const status = await controlStatus("ext-session"); assert.equal(status.sessionId, "ext-session"); assert.equal(status.sessionFile, "/tmp/sessions/ext-session.jsonl"); assert.equal(status.idle, true);
assert.equal(await controlConfigure("ext-session", { model: "anthropic/claude-opus-5", reasoningEffort: "high", }), true); assert.deepEqual(mock.configured, [ { model: mock.models[0] }, { reasoningEffort: "high" }, ]); assert.equal(await controlConfigure("ext-session", { model: "MercuriusDream/Nanbeige4.2-3B-mlx-8bit", }), true); assert.deepEqual(mock.configured.at(-1), { model: mock.models[2] });
assert.equal(await controlSend("ext-session", "change of plan", { deliverAs: "steer" }), true); assert.deepEqual(mock.sent, [{ content: "change of plan", options: { deliverAs: "steer" } }]);
assert.equal(await controlSend("ext-session", "and afterwards", { deliverAs: "followUp" }), true); assert.equal(mock.sent[1].options.deliverAs, "followUp");
assert.equal(await controlInterrupt("ext-session"), true); assert.equal(mock.aborted, 1);
const empty = await controlSend("ext-session", " "); assert.equal(empty, false, "blank steering text must be rejected"); } finally { await mock.stop(); }
assert.equal(await controlStatus("ext-session"), null, "shutdown must unbind the socket");}));
test("controlSubscribe delivers live ordered events with monotonic sequence and live status", () => withControlDir(async () => { const mock = createMockOmp("stream-session"); await mock.start();
try { const received = []; const sub = await controlSubscribe("stream-session", { onEvent: (frame) => received.push(frame), });
assert.ok(sub, "subscribe handle must be returned"); assert.ok(typeof sub.streamId === "string" && sub.streamId.length > 0); assert.equal(sub.sequence, 0); assert.equal(sub.reset, true, "first subscription requires a durable snapshot sync"); assert.equal(sub.status.sessionId, "stream-session"); assert.equal(sub.status.idle, true);
mock.idle = false; mock.emit("turn_start", { type: "turn_start", turnId: "turn-1" }); mock.emit("agent_start", { type: "agent_start" }); mock.emit("message_start", { type: "message_start", messageId: "msg-1" }); mock.emit("message_update", { type: "message_update", delta: "hello" }); mock.emit("message_end", { type: "message_end" }); mock.emit("tool_execution_start", { type: "tool_execution_start", toolCallId: "call-1", toolName: "read" }); mock.emit("tool_execution_end", { type: "tool_execution_end", toolCallId: "call-1", result: { ok: true } }); mock.emit("goal_updated", { type: "goal_updated", goal: "write tests" }); mock.idle = true; mock.emit("turn_end", { type: "turn_end", turnId: "turn-1" });
// Wait a brief moment for socket drain await new Promise((r) => setTimeout(r, 50));
assert.equal(received.length, 9); for (let i = 0; i < received.length; i++) { const frame = received[i]; assert.equal(frame.type, "event"); assert.equal(frame.streamId, sub.streamId); assert.equal(frame.sequence, i + 1, "sequence numbers must be strictly monotonic"); assert.ok(frame.status, "each frame must include live status"); assert.equal(frame.status.sessionId, "stream-session"); }
assert.equal(received[0].event.type, "turn_start"); assert.equal(received[0].status.idle, false); assert.equal(received[8].event.type, "turn_end"); assert.equal(received[8].status.idle, true); assert.equal(sub.sequence, 9);
sub.close(); } finally { await mock.stop(); }}));
test("external Ask tool requests survive reconnect and resolve through the control socket", () => withControlDir(async () => { const mock = createMockOmp("ask-session"); await mock.start(); const events = []; let firstSub = null; let secondSub = null; try { firstSub = await controlSubscribe("ask-session", { onEvent: (frame) => events.push(frame), }); const askTool = mock.tools.get("ask"); assert.ok(askTool, "control extension must override the built-in Ask tool"); const params = { questions: [{ id: "choice", header: "Choose", question: "Which path?", multi: false, options: [ { label: "Alpha", description: "Use alpha." }, { label: "Beta", description: "Use beta." }, ], }], }; const nativeCalls = []; const execution = askTool.execute("ask-call", params, undefined, undefined, { sessionManager: mock.ctx.sessionManager, invokeTool: async (...args) => { nativeCalls.push(args); return { content: [{ type: "text", text: "native" }] }; }, }); await new Promise((resolve) => setTimeout(resolve, 30)); const request = events.find((frame) => frame.event?.type === "ask_request"); assert.ok(request); assert.equal(request.event.toolCallId, "ask-call");
firstSub.close(); secondSub = await controlSubscribe("ask-session", { streamId: firstSub.streamId, afterSequence: firstSub.sequence, }); assert.ok(secondSub.pendingAsks.some((pending) => pending.requestId === request.event.requestId)); assert.equal(await controlRespondAsk("ask-session", request.event.requestId, { choice: { answers: ["Beta"] }, }), true); const result = await execution; assert.equal(nativeCalls.length, 0); assert.deepEqual(result.content, [{ type: "text", text: "User selected: Beta" }]); assert.deepEqual(result.details, { question: "Which path?", options: ["Alpha", "Beta"], multi: false, selectedOptions: ["Beta"], }); assert.equal(result.isError, false); assert.equal(await controlRespondAsk("ask-session", request.event.requestId, { choice: { answers: ["Alpha"] }, }), false, "only the first answer may win"); } finally { firstSub?.close(); secondSub?.close(); await mock.stop(); }}));
test("external Ask override delegates to native UI when no bridge subscribes", () => withControlDir(async () => { const mock = createMockOmp("native-ask-session"); await mock.start(); try { const askTool = mock.tools.get("ask"); const nativeResult = { content: [{ type: "text", text: "native ask" }] }; const result = await askTool.execute("ask-native", { questions: [{ id: "q", question: "Question?", options: [{ label: "Yes" }] }], }, undefined, undefined, { sessionManager: mock.ctx.sessionManager, invokeTool: async () => nativeResult, }); assert.equal(result, nativeResult); } finally { await mock.stop(); }}));
test("controlSubscribe replays missing frames after reconnect without interleaving", () => withControlDir(async () => { const mock = createMockOmp("replay-session"); await mock.start();
try { const firstEvents = []; const sub1 = await controlSubscribe("replay-session", { onEvent: (frame) => firstEvents.push(frame), });
mock.emit("turn_start", { type: "turn_start", id: "t1" }); mock.emit("message_start", { type: "message_start", id: "m1" }); mock.emit("message_end", { type: "message_end", id: "m1" });
await new Promise((r) => setTimeout(r, 50)); assert.equal(firstEvents.length, 3); assert.equal(sub1.sequence, 3);
// Close the first subscription sub1.close();
// Publish events while disconnected mock.emit("tool_execution_start", { type: "tool_execution_start", id: "call-1" }); mock.emit("tool_execution_end", { type: "tool_execution_end", id: "call-1" }); mock.emit("turn_end", { type: "turn_end", id: "t1" });
// Reconnect with streamId and afterSequence: 3 const replayedEvents = []; const sub2 = await controlSubscribe("replay-session", { streamId: sub1.streamId, afterSequence: 3, onEvent: (frame) => replayedEvents.push(frame), });
assert.ok(sub2, "reconnect subscribe must succeed"); assert.equal(sub2.reset, false, "reset must be false when within replay buffer"); assert.equal(sub2.streamId, sub1.streamId);
// Publish new live event immediately after reconnect mock.emit("turn_start", { type: "turn_start", id: "t2" });
await new Promise((r) => setTimeout(r, 50));
// Must have replayed sequences 4, 5, 6 and live sequence 7 in strict order assert.equal(replayedEvents.length, 4); assert.deepEqual(replayedEvents.map((f) => f.sequence), [4, 5, 6, 7]); assert.equal(replayedEvents[0].event.type, "tool_execution_start"); assert.equal(replayedEvents[1].event.type, "tool_execution_end"); assert.equal(replayedEvents[2].event.type, "turn_end"); assert.equal(replayedEvents[3].event.type, "turn_start"); assert.equal(sub2.sequence, 7);
sub2.close(); } finally { await mock.stop(); }}));
test("controlSubscribe keeps the reconnect cursor behind an ACK until split replay frames arrive", () => withControlDir(async () => { const socketPath = controlSocketPath("split-replay-session"); const server = net.createServer((socket) => { socket.setEncoding("utf8"); socket.once("data", () => { socket.write(`${JSON.stringify({ ok: true, op: "subscribe", streamId: "split-stream", sequence: 5, earliestSequence: 1, reset: false, status: { idle: false, hasPendingMessages: false }, })}\n`); setTimeout(() => socket.write(`${JSON.stringify({ type: "event", streamId: "split-stream", sequence: 4, event: { type: "message_end" }, status: { idle: false, hasPendingMessages: false }, })}\n`), 20); setTimeout(() => socket.write(`${JSON.stringify({ type: "event", streamId: "split-stream", sequence: 5, event: { type: "turn_end" }, status: { idle: true, hasPendingMessages: false }, })}\n`), 40); }); }); await new Promise((resolve) => server.listen(socketPath, resolve)); const received = []; const sub = await controlSubscribe("split-replay-session", { streamId: "split-stream", afterSequence: 3, onEvent: (frame) => received.push(frame), }); try { assert.ok(sub); assert.equal(sub.sequence, 3, "ACK head must not skip replay still in flight"); assert.equal(sub.headSequence, 5); await new Promise((resolve) => setTimeout(resolve, 70)); assert.deepEqual(received.map((frame) => frame.sequence), [4, 5]); assert.equal(sub.sequence, 5); } finally { sub?.close(); await new Promise((resolve) => server.close(resolve)); }}));
test("controlSubscribe aborts a pending handshake without leaking its socket", () => withControlDir(async () => { const session = await fakeSession("aborted-subscription", { reply: () => null }); try { const controller = new AbortController(); const pending = controlSubscribe("aborted-subscription", { signal: controller.signal, timeoutMs: 2_000 }); setTimeout(() => controller.abort(), 20); assert.equal(await pending, null); } finally { await session.close(); }}));
test("controlSubscribe sets reset: true on stream mismatch and evicted sequence", () => withControlDir(async () => { const mock = createMockOmp("reset-session"); await mock.start();
try { mock.emit("turn_start", { type: "turn_start" }); mock.emit("turn_end", { type: "turn_end" });
// 1. Stream mismatch const mismatchedSub = await controlSubscribe("reset-session", { streamId: "wrong-stream-uuid", afterSequence: 1, }); assert.ok(mismatchedSub); assert.equal(mismatchedSub.reset, true, "mismatched streamId must trigger reset: true"); mismatchedSub.close();
// 2. Replay buffer eviction (capacity is 256) for (let i = 0; i < 270; i++) { mock.emit("message_update", { type: "message_update", delta: `chunk-${i}` }); }
const evictedSub = await controlSubscribe("reset-session", { streamId: mismatchedSub.streamId, // correct server streamId afterSequence: 5, // evicted sequence }); assert.ok(evictedSub); assert.equal(evictedSub.reset, true, "evicted sequence must trigger reset: true"); evictedSub.close(); } finally { await mock.stop(); }}));
test("slash model configuration updates status and validates availability", () => withControlDir(async () => { const mock = createMockOmp("model-session"); await mock.start();
try { // 1. Slash model selector const ok = await controlConfigure("model-session", { model: "google-antigravity/gemini-3.7-flash", reasoningEffort: "medium", }); assert.equal(ok, true); assert.deepEqual(mock.configured.at(-2), { model: mock.models[1] }); assert.deepEqual(mock.configured.at(-1), { reasoningEffort: "medium" });
const status = await controlStatus("model-session"); assert.equal(status.model, "gemini-3.7-flash");
// 2. Unknown model selector returns false const failed = await controlConfigure("model-session", { model: "nonexistent-provider/fake-model", }); assert.equal(failed, false); } finally { await mock.stop(); }}));
test("session teardown cleans up socket and notifies subscribers via onDisconnect", () => withControlDir(async () => { const mock = createMockOmp("teardown-session"); await mock.start();
let disconnected = false; const sub = await controlSubscribe("teardown-session", { onDisconnect: () => { disconnected = true; }, }); assert.ok(sub);
// Stop session await mock.stop();
await new Promise((r) => setTimeout(r, 50)); assert.equal(disconnected, true, "active subscriber must be notified on session teardown"); assert.equal(await controlStatus("teardown-session"), null, "socket must be cleaned up");}));
test("oversized payloads degrade to invalidation events without crashing", () => withControlDir(async () => { const mock = createMockOmp("large-session"); await mock.start();
try { const events = []; const sub = await controlSubscribe("large-session", { onEvent: (frame) => events.push(frame), });
const hugeData = "x".repeat(128 * 1024); // 128 KB payload mock.emit("tool_execution_end", { type: "tool_execution_end", hugeData });
await new Promise((r) => setTimeout(r, 50));
assert.equal(events.length, 1); assert.equal(events[0].event.type, "invalidation"); assert.equal(events[0].event.originalType, "tool_execution_end"); assert.equal(events[0].event.reason, "oversized_or_unserializable");
sub.close(); } finally { await mock.stop(); }}));
test("app-server consumes live control events without waiting for the JSONL poll interval", () => withControlDir(async () => { const harness = await makeHarness({}); let mock = null; try { await harness.initialize(); const sessionId = "event-driven-owner"; const threadId = "event-driven-thread"; const sessionFile = path.join(harness.temp, "event-driven.jsonl"); await fs.writeFile(sessionFile, [ JSON.stringify({ type: "session", version: 3, id: sessionId, timestamp: new Date().toISOString(), cwd: harness.temp }), JSON.stringify({ type: "title", title: "Event driven session" }), JSON.stringify({ type: "message", id: "user-1", message: { role: "user", content: [{ type: "text", text: "baseline" }] } }), JSON.stringify({ type: "message", id: "assistant-1", message: { role: "assistant", stopReason: "stop", content: [{ type: "text", text: "baseline done" }] } }), ].join("\n") + "\n");
const thread = await harness.server.store.createThread({ threadId, cwd: harness.temp, preview: "Event driven session", }); await harness.server.store.setBackendState(thread.id, { sessionId, sessionPath: sessionFile, resume: { sessionId, sessionPath: sessionFile }, }); mock = createMockOmp(sessionId, { cwd: harness.temp, sessionManager: { getSessionId: () => sessionId, getSessionFile: () => sessionFile, }, }); await mock.start(); await harness.request("thread/read", { threadId });
const startedAt = Date.now(); const startedPromise = harness.waitFor((message) => message.method === "turn/started" && message.params?.threadId === threadId , 750); const userMessage = { role: "user", content: [{ type: "text", text: "live prompt" }] }; await fs.appendFile(sessionFile, `${JSON.stringify({ type: "message", id: "user-live", timestamp: new Date().toISOString(), message: userMessage, })}\n`); mock.idle = false; mock.emit("message_start", { type: "message_start", message: userMessage }); const started = await startedPromise; assert.ok(Date.now() - startedAt < 750, "live event should beat the one-second fallback poll");
const retryPromise = harness.waitFor((message) => message.method === "error" && message.params?.turnId === started.params.turn.id , 750); mock.emit("auto_retry_start", { type: "auto_retry_start", attempt: 1, maxAttempts: 3, delayMs: 250, errorMessage: "provider quota exhausted", }); const retry = await retryPromise; assert.equal(retry.params.willRetry, true); assert.equal(retry.params.error.codexErrorInfo, "usageLimitExceeded");
const warningPromise = harness.waitFor((message) => message.method === "warning" && /continuing with anthropic\/claude-opus-5/.test(message.params?.message || "") , 750); const settingsPromise = harness.waitFor((message) => message.method === "thread/settings/updated" && message.params?.threadId === threadId , 750); mock.emit("retry_fallback_applied", { type: "retry_fallback_applied", from: "openai-codex/gpt-5.6-sol", to: "anthropic/claude-opus-5", role: "default", }); const [, settings] = await Promise.all([warningPromise, settingsPromise]); assert.equal(settings.params.threadSettings.model, "claude-opus-5"); assert.equal(settings.params.threadSettings.modelProvider, "anthropic");
const completedPromise = harness.waitFor((message) => message.method === "turn/completed" && message.params?.threadId === threadId && message.params?.turn?.id === started.params.turn.id , 750); const assistantMessage = { role: "assistant", stopReason: "stop", content: [{ type: "text", text: "live answer" }], }; await fs.appendFile(sessionFile, `${JSON.stringify({ type: "message", id: "assistant-live", timestamp: new Date().toISOString(), message: assistantMessage, })}\n`); mock.idle = true; mock.emit("message_end", { type: "message_end", message: assistantMessage }); const completed = await completedPromise; assert.equal(completed.params.turn.status, "completed"); } finally { await mock?.stop(); await harness.close(); }}));
test("app-server relays an external OMP Ask to Desktop and returns the winning answer", () => withControlDir(async () => { let desktopRequest = null; const harness = await makeHarness({ clientResponse(message) { if (message.method === "item/tool/requestUserInput") { desktopRequest = message; return { answers: { deploy: { answers: ["Continue"] } } }; } return { decision: "accept" }; }, }); let mock = null; try { await harness.initialize(); const sessionId = "external-ask-owner"; const threadId = "external-ask-thread"; const sessionFile = path.join(harness.temp, "external-ask.jsonl"); await fs.writeFile(sessionFile, [ JSON.stringify({ type: "session", id: sessionId, timestamp: new Date().toISOString(), cwd: harness.temp }), JSON.stringify({ type: "title", title: "External Ask" }), JSON.stringify({ type: "message", id: "ask-user", message: { role: "user", content: [{ type: "text", text: "test" }] } }), ].join("\n") + "\n"); const thread = await harness.store.createThread({ threadId, cwd: harness.temp, preview: "External Ask", }); await harness.store.setBackendState(thread.id, { sessionId, sessionPath: sessionFile, resume: { sessionId, sessionPath: sessionFile }, }); mock = createMockOmp(sessionId, { cwd: harness.temp, sessionManager: { getSessionId: () => sessionId, getSessionFile: () => sessionFile, }, }); await mock.start(); await harness.request("thread/read", { threadId });
const result = await mock.tools.get("ask").execute("external-ask-call", { questions: [{ id: "deploy", header: "Deploy", question: "Continue deployment?", options: [{ label: "Continue", description: "Proceed." }], }], }, undefined, undefined, { sessionManager: mock.ctx.sessionManager, invokeTool: async () => { throw new Error("native Ask must not run while bridge is subscribed"); }, }); assert.deepEqual(result.content, [{ type: "text", text: "User selected: Continue" }]); assert.equal(result.isError, false); assert.ok(desktopRequest); assert.equal(desktopRequest.params.threadId, threadId); assert.equal(desktopRequest.params.itemId, "external-ask-call"); assert.deepEqual(desktopRequest.params.questions, [{ id: "deploy", header: "Deploy", question: "Continue deployment?", isOther: true, isSecret: false, options: [{ label: "Continue", description: "Proceed." }], }]); assert.ok(harness.messages.some((message) => message.method === "serverRequest/resolved" && message.params?.requestId === desktopRequest.id )); } finally { await mock?.stop(); await harness.close(); }}));
test("bridge-owned RPC turns subscribe before Ask even while active polling is suppressed", () => withControlDir(async () => { let mock = null; let nativeAskCalls = 0; const backendFactory = { async listModels() { return [{ id: "mock/omp", model: "mock/omp", displayName: "Mock", isDefault: true }]; }, async openSession() { return { async runTurn(_prompt, context) { const socketStart = new Promise((resolve) => setTimeout(resolve, 500)).then(() => mock.start()); await context.persistSessionState({ kind: "rpc", sessionId: "owned-ask-session", sessionPath: path.join(context.options.cwd, "owned-ask.jsonl"), resume: { sessionId: "owned-ask-session", sessionPath: path.join(context.options.cwd, "owned-ask.jsonl"), }, }); await socketStart; await new Promise((resolve) => setTimeout(resolve, 1_250)); const result = await mock.tools.get("ask").execute("owned-ask-call", { questions: [{ id: "owned", question: "Does owned Ask work?", options: [{ label: "Yes" }, { label: "No" }], }], }, context.signal, undefined, { sessionManager: mock.ctx.sessionManager, invokeTool: async () => { nativeAskCalls += 1; return { content: [{ type: "text", text: "native" }] }; }, }); await context.emit({ type: "textDelta", delta: result.content[0].text }); return { status: "completed" }; }, async getPersistentState() { return {}; }, async interrupt() {}, async dispose() {}, }; }, }; const harness = await makeHarness({ backendFactory, clientResponse(message) { if (message.method === "item/tool/requestUserInput") { return { answers: { owned: { answers: ["Yes"] } } }; } return { decision: "accept" }; }, }); try { await harness.initialize(); const sessionFile = path.join(harness.temp, "owned-ask.jsonl"); await fs.writeFile(sessionFile, [ JSON.stringify({ type: "session", id: "owned-ask-session", timestamp: new Date().toISOString(), cwd: harness.temp }), JSON.stringify({ type: "title", title: "Owned Ask" }), ].join("\n") + "\n"); mock = createMockOmp("owned-ask-session", { cwd: harness.temp, sessionManager: { getSessionId: () => "owned-ask-session", getSessionFile: () => sessionFile, }, }); const thread = await harness.request("thread/start", { cwd: harness.temp, model: "mock/omp" }); const started = await harness.request("turn/start", { threadId: thread.thread.id, input: [{ type: "text", text: "ask" }], }); const completed = await harness.waitFor((message) => message.method === "turn/completed" && message.params?.turn?.id === started.turn.id ); assert.equal(nativeAskCalls, 0); assert.match(completed.params.turn.items.find((item) => item.type === "agentMessage").text, /User selected: Yes/); } finally { await mock?.stop(); await harness.close(); }}));
test("turn/interrupt reaches the OMP process that owns an external thread", () => withControlDir(async () => { const harness = await makeHarness({}); try { await harness.initialize(); const session = await fakeSession("external-owner"); try { harness.server.store.getRecord = async () => ({ thread: { id: "external-thread", sessionId: "external-owner" }, backend: { resume: { sessionId: "external-owner", sessionPath: "/tmp/external.jsonl" } }, }); await harness.request("turn/interrupt", { threadId: "external-thread" }); assert.deepEqual(session.received.map((r) => r.op), ["interrupt"]); } finally { await session.close(); } } finally { await harness.close(); }}));
test("turn/start steers a thread another OMP client is driving", () => withControlDir(async () => { const harness = await makeHarness({}); try { await harness.initialize(); const session = await fakeSession("busy-owner", { reply: (request) => request.op === "status" ? { ok: true, op: "status", idle: true, hasPendingMessages: false, sessionId: "busy-owner" } : { ok: true, op: request.op }, }); try { const inputText = "actually, stop and summarize"; const userTurn = (id, text) => ({ id, status: "inProgress", items: [{ id: `${id}-user`, type: "userMessage", status: "completed", content: [{ type: "text", text }], }], }); const unrelatedTurn = userTurn("turn-unrelated", "someone else's prompt"); const externalTurn = userTurn("turn-external-1", inputText); let turns = []; harness.server.store.getRecord = async () => ({ thread: { id: "busy-thread", sessionId: "busy-owner" }, backend: { resume: { sessionId: "busy-owner", sessionPath: "/tmp/busy.jsonl" } }, }); harness.server.store.getThread = async () => ({ id: "busy-thread", cwd: harness.temp, status: { type: "active", activeFlags: [] }, turns, }); // A concurrent unrelated turn appears first; correlation must wait for // the turn containing this request's prompt. setTimeout(() => { turns = [unrelatedTurn]; }, 100); setTimeout(() => { turns = [unrelatedTurn, externalTurn]; }, 300);
const result = await harness.request("turn/start", { threadId: "busy-thread", input: [{ type: "text", text: inputText }], }); assert.equal(result.turn.id, "turn-external-1", "client must receive the real external turn"); assert.deepEqual(session.received.map((r) => r.op), ["status", "steer"]); assert.equal(session.received[1].text, "actually, stop and summarize"); } finally { await session.close(); } } finally { await harness.close(); }}));
test("turn/start steering an active external session returns its in-progress turn immediately", () => withControlDir(async () => { const harness = await makeHarness({}); try { await harness.initialize(); const session = await fakeSession("active-owner", { reply: (request) => request.op === "status" ? { ok: true, op: "status", idle: false, hasPendingMessages: false, sessionId: "active-owner" } : { ok: true, op: request.op }, }); try { const activeTurn = { id: "turn-active", status: "inProgress", items: [] }; harness.server.store.getRecord = async () => ({ thread: { id: "active-thread", sessionId: "active-owner" }, backend: { resume: { sessionId: "active-owner", sessionPath: "/tmp/active.jsonl" } }, }); harness.server.store.getThread = async () => ({ id: "active-thread", cwd: harness.temp, model: "claude-opus-5", modelSelector: "anthropic/claude-opus-5", reasoningEffort: "high", status: { type: "active", activeFlags: [] }, turns: [activeTurn], });
const result = await harness.request("turn/start", { threadId: "active-thread", input: [{ type: "text", text: "change direction" }], }); assert.equal(result.turn.id, activeTurn.id); assert.equal(result.turn.status, "inProgress"); assert.deepEqual(session.received.map((request) => request.op), ["status", "configure", "steer"]); } finally { await session.close(); } } finally { await harness.close(); }}));
test("rejected live model settings do not mutate persisted thread state", () => withControlDir(async () => { const harness = await makeHarness({}); try { await harness.initialize(); const thread = await harness.server.store.createThread({ threadId: "configured-thread", cwd: harness.temp, preview: "configured", model: { id: "claude-opus-5", provider: "anthropic", selector: "anthropic/claude-opus-5" }, reasoningEffort: "medium", }); await harness.server.store.setBackendState(thread.id, { sessionId: "configured-owner", sessionPath: "/tmp/configured-owner.jsonl", resume: { sessionId: "configured-owner", sessionPath: "/tmp/configured-owner.jsonl" }, }); const session = await fakeSession("configured-owner", { reply: (request) => { if (request.op === "status") { return { ok: true, op: "status", idle: true, hasPendingMessages: false, sessionId: "configured-owner" }; } if (request.op === "configure") return { ok: false, op: "configure", error: "unknown model" }; return { ok: true, op: request.op }; }, }); try { await assert.rejects( () => harness.request("thread/resume", { threadId: thread.id, model: "unknown-provider/unknown-model", config: { model_reasoning_effort: "high" }, }), /rejected the requested model settings/, ); let persisted = await harness.server.store.getThread(thread.id); assert.equal(persisted.model, "claude-opus-5"); assert.equal(persisted.reasoningEffort, "medium");
await assert.rejects( () => harness.request("thread/settings/update", { threadId: thread.id, model: "unknown-provider/unknown-model", reasoningEffort: "high", }), /rejected the requested model settings/, ); persisted = await harness.server.store.getThread(thread.id); assert.equal(persisted.model, "claude-opus-5"); assert.equal(persisted.reasoningEffort, "medium"); } finally { await session.close(); } } finally { await harness.close(); }}));
test("turn/start still reports BUSY when the owning session has no control socket", () => withControlDir(async () => { const harness = await makeHarness({}); try { await harness.initialize(); harness.server.store.getRecord = async () => ({ thread: { id: "opaque-thread", sessionId: "opaque-owner" }, backend: { resume: { sessionId: "opaque-owner", sessionPath: "/tmp/opaque.jsonl" } }, }); harness.server.store.getThread = async () => ({ id: "opaque-thread", cwd: harness.temp, status: { type: "active", activeFlags: [] }, turns: [], }); await assert.rejects( () => harness.request("turn/start", { threadId: "opaque-thread", input: [{ type: "text", text: "hi" }] }), /active in another OMP client/, ); } finally { await harness.close(); }}));
test("compacting a thread another OMP process drives goes over its control socket", () => withControlDir(async () => { const harness = await makeHarness({}); try { await harness.initialize(); const session = await fakeSession("compact-owner", { reply: (request) => request.op === "status" ? { ok: true, op: "status", idle: true, hasPendingMessages: false, sessionId: "compact-owner" } : { ok: true, op: request.op, started: true }, }); try { const { thread } = await harness.request("thread/start", { cwd: harness.temp }); await harness.server.store.setBackendState(thread.id, { kind: "rpc", resume: { sessionId: "compact-owner", sessionPath: "/tmp/compact-owner.jsonl" }, }); const sessionsBefore = harness.server.sessions.size;
await harness.request("thread/compact/start", { threadId: thread.id }); const compacted = await harness.waitFor((message) => message.method === "thread/compacted");
assert.equal(compacted.params.compacted, true); assert.ok(session.received.some((request) => request.op === "compact"), `control ops seen: ${JSON.stringify(session.received.map((r) => r.op))}`); assert.equal(harness.server.sessions.size, sessionsBefore, "a live external thread must not get a second session"); const stored = await harness.server.store.getThread(thread.id); const turn = stored.turns.find((candidate) => (candidate.items || []).some((item) => item.type === "contextCompaction")); assert.equal(turn.status, "completed"); } finally { await session.close(); } } finally { await harness.close(); }}));