Something went wrong. Try again.
presents omp as codex for use with chatgpt remote
Something went wrong. Try again.
50 kB · 1091 lines
JavaScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092import test from "node:test";import assert from "node:assert/strict";import fs from "node:fs/promises";import os from "node:os";import path from "node:path";import { pathToFileURL } from "node:url";import { CodexAppServer } from "../src/app-server.mjs";import { OmpRpcBackendFactory, OmpSdkBackendFactory, extractRpcResume, loadOmpSdk, mapOmpSessionEvent, parseOmpModelRows } from "../src/backend/omp.mjs";import { StateStore } from "../src/store.mjs";import { cwdToSessionDirName, parseOmpSessionFile } from "../src/omp-sessions.mjs";
async function makeServerHarness(backendFactory, temp) { const store = new StateStore(path.join(temp, "state.json")); const server = new CodexAppServer({ backendFactory, store, cwd: temp }); const messages = []; const waiters = []; const peer = server.createPeer((message) => { messages.push(message); for (const waiter of [...waiters]) waiter(message); if (message.id !== undefined && message.method) queueMicrotask(() => server.handle(peer, { id: message.id, result: { decision: "accept" } })); }); let id = 0; const request = async (method, params = {}) => { const requestId = ++id; await server.handle(peer, { id: requestId, method, params }); const response = messages.find((message) => message.id === requestId && !message.method); if (response?.error) throw new Error(response.error.message); return response?.result; }; const waitFor = (predicate, timeoutMs = 3000) => { const found = messages.find(predicate); if (found) return Promise.resolve(found); return new Promise((resolve, reject) => { const timer = setTimeout(() => reject(new Error(`timeout: ${JSON.stringify(messages)}`)), timeoutMs); const waiter = (message) => { if (!predicate(message)) return; clearTimeout(timer); waiters.splice(waiters.indexOf(waiter), 1); resolve(message); }; waiters.push(waiter); }); }; await request("initialize", { clientInfo: { name: "adapter-test" } }); await server.handle(peer, { method: "initialized", params: {} }); return { server, store, peer, messages, request, waitFor };}
test("direct SDK adapter maps OMP events and ACP-style permission calls", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-sdk-adapter-")); try { const fake = path.join(temp, "fake-omp.mjs"); await fs.writeFile(fake, ` export async function discoverAuthStorage() { return { kind: "fake-auth" }; } export async function discoverModels() { return [{ id: "fake/model", provider: "fake", displayName: "Fake model" }]; } export class SessionManager { static create(cwd) { return { cwd, sessionPath: cwd + "/fake-session.jsonl" }; } } export async function createAgentSession(options) { let listener = null; return { session: { sessionId: "fake-session", sessionPath: options.sessionManager.sessionPath, subscribe(fn) { listener = fn; return () => { listener = null; }; }, async prompt(text) { await listener({ type: "message_update", assistantMessageEvent: { type: "thinking_delta", delta: "considering" } }); await listener({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "sdk:" + text } }); await listener({ type: "tool_execution_start", toolCallId: "sdk-tool", toolName: "bash", args: { command: "echo sdk", cwd: options.cwd } }); const permission = await options.clientBridge.requestPermission({ toolCallId: "sdk-tool", toolName: "bash", input: { command: "echo sdk", cwd: options.cwd } }); if (!permission.approved) throw new Error("permission declined"); await listener({ type: "tool_execution_update", toolCallId: "sdk-tool", delta: "sdk output\\n" }); listener({ type: "tool_execution_end", toolCallId: "sdk-tool", result: { exitCode: 0 } }); listener({ type: "message_end", message: { usage: { inputTokens: 20, outputTokens: 5, totalTokens: 25 } } }); listener({ type: "agent_end", isTerminal: true }); }, async abort() {}, async steer() {}, }}; } `); const loaded = await loadOmpSdk({ module: pathToFileURL(fake).href }); assert.ok(loaded); const factory = new OmpSdkBackendFactory(loaded); const models = await factory.listModels(); assert.equal(models[0].id, "fake/model"); const h = await makeServerHarness(factory, temp); try { const { thread } = await h.request("thread/start", { cwd: temp, model: "fake/model", approvalPolicy: "on-request" }); const { turn } = await h.request("turn/start", { threadId: thread.id, input: [{ type: "text", text: "hello" }], approvalPolicy: "on-request" }); const completed = await h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === turn.id); assert.equal(completed.params.turn.status, "completed"); assert.match(completed.params.turn.items.find((item) => item.type === "agentMessage").text, /sdk:hello/); assert.equal(completed.params.turn.items.find((item) => item.type === "commandExecution").exitCode, 0); assert.equal(completed.params.turn.tokenUsage.totalTokens, 25); assert.ok(h.messages.some((message) => message.method === "item/commandExecution/requestApproval")); assert.ok(h.messages.some((message) => message.method?.startsWith("item/reasoning/"))); } finally { await h.server.close(); } } finally { await fs.rm(temp, { recursive: true, force: true }); }});
test("SDK session applies turn model and thinking overrides to the underlying session before prompt", async () => { const calls = []; const fakeSession = { sessionId: "fake-sdk-sess", sessionPath: "/tmp/fake-sdk-sess.jsonl", subscribe(fn) { return () => {}; }, async setModel(model) { calls.push({ method: "setModel", args: [model] }); }, async setThinkingLevel(level) { calls.push({ method: "setThinkingLevel", args: [level] }); }, async prompt(text) { calls.push({ method: "prompt", args: [text] }); }, };
const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-sdk-overrides-")); try { const fake = path.join(temp, "fake-omp.mjs"); await fs.writeFile(fake, ` export async function discoverAuthStorage() { return { kind: "fake-auth" }; } export async function discoverModels() { return [{ id: "fake/model", provider: "fake", displayName: "Fake model" }]; } export class SessionManager { static create(cwd) { return { cwd, sessionPath: cwd + "/fake-session.jsonl" }; } } export async function createAgentSession(options) { return { session: globalThis.__testFakeSession }; } `); globalThis.__testFakeSession = fakeSession; const loaded = await loadOmpSdk({ module: pathToFileURL(fake).href }); const factory = new OmpSdkBackendFactory(loaded); const session = await factory.openSession({ cwd: temp });
// Turn 1: model + reasoningEffort overrides await session.runTurn({ text: "turn 1" }, { options: { model: "claude-sonnet-5", reasoningEffort: "high" }, emit: async () => {}, });
assert.equal(calls.length, 3); assert.deepEqual(calls[0], { method: "setModel", args: ["claude-sonnet-5"] }); assert.deepEqual(calls[1], { method: "setThinkingLevel", args: ["high"] }); assert.deepEqual(calls[2], { method: "prompt", args: ["turn 1"] });
// Turn 2: reasoningEffort none -> maps to minimal, no model change calls.length = 0; await session.runTurn({ text: "turn 2" }, { options: { reasoningEffort: "none" }, emit: async () => {}, }); assert.equal(calls.length, 2); assert.deepEqual(calls[0], { method: "setThinkingLevel", args: ["minimal"] }); assert.deepEqual(calls[1], { method: "prompt", args: ["turn 2"] });
// Turn 3: fallback to setReasoningEffort if setThinkingLevel is absent const fallbackCalls = []; const fallbackSession = { sessionId: "fallback-sdk-sess", sessionPath: "/tmp/fallback.jsonl", subscribe() { return () => {}; }, async setReasoningEffort(effort) { fallbackCalls.push({ method: "setReasoningEffort", args: [effort] }); }, async prompt(text) { fallbackCalls.push({ method: "prompt", args: [text] }); }, }; globalThis.__testFakeSession = fallbackSession; const session2 = await factory.openSession({ cwd: temp }); await session2.runTurn({ text: "turn 3" }, { options: { reasoningEffort: "low" }, emit: async () => {}, }); assert.deepEqual(fallbackCalls, [ { method: "setReasoningEffort", args: ["low"] }, { method: "prompt", args: ["turn 3"] }, ]); } finally { delete globalThis.__testFakeSession; await fs.rm(temp, { recursive: true, force: true }); }});
test("RPC model discovery exposes one canonical bare slug row per slug and chooses first non-openrouter candidate", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-models-")); try { const fake = path.join(temp, "fake-models.cjs"); await fs.writeFile(fake, `#!/usr/bin/env node if (process.argv[2] !== "models" || process.argv[3] !== "--json") process.exit(2); console.log(JSON.stringify({ models: [ { id: "claude-sonnet-5", provider: "anthropic", name: "Claude Sonnet 5", selector: "anthropic/claude-sonnet-5", thinking: ["low", "high"] }, { id: "gpt-4.1", provider: "openai", name: "GPT-4.1", selector: "openai/gpt-4.1", reasoning: true }, { id: "gemini-3.7-flash", provider: "google-antigravity", name: "Gemini 3.7 Flash", selector: "google-antigravity/gemini-3.7-flash", thinking: ["minimal", "low", "medium", "high"] }, { id: "shared-model", provider: "anthropic", name: "Shared Anthropic", selector: "anthropic/shared-model", thinking: ["low"] }, { id: "shared-model", provider: "google-antigravity", name: "Shared Google", selector: "google-antigravity/shared-model", thinking: ["high"] }, { id: "shared-model", provider: "openrouter", name: "Shared OpenRouter", selector: "openrouter/shared-model" }, { id: "openrouter-only", provider: "openrouter", name: "OR Only", selector: "openrouter/openrouter-only" }, ] })); `); await fs.chmod(fake, 0o755); const modelCache = path.join(temp, "model-catalog-cache.json"); await fs.writeFile(modelCache, "null"); const models = await new OmpRpcBackendFactory({ ompExecutable: fake, modelCatalogCacheFile: modelCache, }).listModels();
// 1. Exactly one canonical bare slug per model, never provider-prefixed IDs assert.deepEqual( models.map((model) => model.id), ["claude-sonnet-5", "gpt-4.1", "gemini-3.7-flash", "shared-model"] ); assert.equal(models.some((model) => model.id.includes("/")), false); assert.equal(models.some((model) => model.model.includes("/")), false);
// 2. Deterministically choose the first non-openrouter candidate when duplicates exist const shared = models.find((model) => model.id === "shared-model"); assert.equal(shared.provider, "anthropic"); assert.equal(shared.selector, "anthropic/shared-model"); assert.equal(shared.displayName, "Shared Anthropic"); assert.deepEqual(shared.supportedReasoningEfforts, ["low"]);
// 3. Openrouter models are excluded assert.equal(models.some((model) => model.id === "openrouter-only"), false);
// 4. Chosen provider-qualified selector is kept internal on row assert.equal(models.find((model) => model.id === "claude-sonnet-5").selector, "anthropic/claude-sonnet-5"); assert.equal(models.find((model) => model.id === "gemini-3.7-flash").selector, "google-antigravity/gemini-3.7-flash");
const persisted = JSON.parse(await fs.readFile(modelCache, "utf8")); assert.ok(Array.isArray(persisted.catalogs[fake].models)); assert.deepEqual(persisted.catalogs[fake].models.map((m) => m.id), ["claude-sonnet-5", "gpt-4.1", "gemini-3.7-flash", "shared-model"]); } finally { await fs.rm(temp, { recursive: true, force: true }); }});
test("RPC forks copy OMP history and persist injected side-chat boundaries", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-fork-")); const previousAgentDir = process.env.PI_CODING_AGENT_DIR; process.env.PI_CODING_AGENT_DIR = path.join(temp, "agent"); try { const cwd = path.join(temp, "project"); const sourcePath = path.join(temp, "source.jsonl"); const sourceArtifacts = sourcePath.slice(0, -".jsonl".length); await fs.mkdir(cwd, { recursive: true }); await fs.mkdir(sourceArtifacts, { recursive: true }); await fs.writeFile(path.join(sourceArtifacts, "0.read.log"), "artifact"); await fs.writeFile(sourcePath, [ JSON.stringify({ type: "title", v: 1, title: "parent", source: "auto", updatedAt: "2026-08-25T00:00:00.000Z" }), JSON.stringify({ type: "session", version: 3, id: "parent-session", timestamp: "2026-08-25T00:00:00.000Z", cwd }), JSON.stringify({ type: "message", id: "parent-message", parentId: null, timestamp: "2026-08-25T00:00:01.000Z", message: { role: "user", content: [{ type: "text", text: "parent context" }] }, }), ].join("\n") + "\n");
const session = await new OmpRpcBackendFactory().forkSession({ sessionPath: sourcePath }, { cwd }); await session.injectItems([{ type: "message", role: "user", content: [{ type: "input_text", text: "Side conversation boundary." }], }]);
const state = await session.getPersistentState(); assert.notEqual(state.sessionPath, sourcePath); const records = (await fs.readFile(state.sessionPath, "utf8")) .trim() .split("\n") .map((line) => JSON.parse(line)); const header = records.find((record) => record.type === "session"); const injected = records.at(-1); assert.equal(header.parentSession, "parent-session"); assert.equal(header.cwd, cwd); assert.notEqual(header.id, "parent-session"); assert.equal(records.some((record) => record.id === "parent-message"), true); assert.equal(injected.type, "message"); assert.equal(injected.parentId, "parent-message"); assert.equal(injected.message.role, "user"); assert.equal(injected.message.content[0].text, "Side conversation boundary."); assert.equal(injected.message.codexBridgeInjected, true); assert.equal((await parseOmpSessionFile(state.sessionPath)).thread.turns.length, 0); assert.equal( await fs.readFile(path.join(state.sessionPath.slice(0, -".jsonl".length), "0.read.log"), "utf8"), "artifact" ); } finally { if (previousAgentDir === undefined) delete process.env.PI_CODING_AGENT_DIR; else process.env.PI_CODING_AGENT_DIR = previousAgentDir; await fs.rm(temp, { recursive: true, force: true }); }});
test("extractRpcResume recursively unwraps nested metadata, strings, and direct objects preferring sessionPath", () => { // String file path assert.deepEqual(extractRpcResume("/path/to/session.jsonl"), { sessionPath: "/path/to/session.jsonl", sessionId: null, resumeTarget: "/path/to/session.jsonl", raw: "/path/to/session.jsonl", });
// String session ID assert.deepEqual(extractRpcResume("sess-1234-uuid"), { sessionPath: null, sessionId: "sess-1234-uuid", resumeTarget: "sess-1234-uuid", raw: "sess-1234-uuid", });
// Direct object with both sessionPath and sessionId (prefers sessionPath as target) assert.deepEqual(extractRpcResume({ sessionPath: "/path/to/session.jsonl", sessionId: "sess-123" }), { sessionPath: "/path/to/session.jsonl", sessionId: "sess-123", resumeTarget: "/path/to/session.jsonl", raw: { sessionPath: "/path/to/session.jsonl", sessionId: "sess-123" }, });
// Direct object with only sessionId assert.deepEqual(extractRpcResume({ sessionId: "sess-123" }), { sessionPath: null, sessionId: "sess-123", resumeTarget: "sess-123", raw: { sessionId: "sess-123" }, });
// Deeply nested .resume object const deeplyNested = { resume: { resume: { resume: { sessionPath: "/deep/nested.jsonl", sessionId: "deep-uuid", }, }, }, }; assert.deepEqual(extractRpcResume(deeplyNested), { sessionPath: "/deep/nested.jsonl", sessionId: "deep-uuid", resumeTarget: "/deep/nested.jsonl", raw: { sessionPath: "/deep/nested.jsonl", sessionId: "deep-uuid" }, });
// Nested string assert.deepEqual(extractRpcResume({ resume: { resume: "/nested/path.jsonl" } }), { sessionPath: "/nested/path.jsonl", sessionId: null, resumeTarget: "/nested/path.jsonl", raw: "/nested/path.jsonl", });
// Null, undefined, empty object assert.equal(extractRpcResume(null), null); assert.equal(extractRpcResume(undefined), null); assert.equal(extractRpcResume({}), null); assert.equal(extractRpcResume({ resume: null }), null); assert.equal(extractRpcResume({ resume: {} }), null);});
test("RPC session with nested resume passes --resume target preferring sessionPath to spawned process", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-resume-args-")); try { const argsLogFile = path.join(temp, "spawned-args.json"); const fake = path.join(temp, "fake-omp-resume.cjs"); await fs.writeFile(fake, `#!/usr/bin/env node const fs = require("node:fs"); const readline = require("node:readline"); fs.writeFileSync(process.env.TEST_ARGS_LOG, JSON.stringify(process.argv.slice(2))); const rl = readline.createInterface({ input: process.stdin }); rl.on("line", (line) => { const msg = JSON.parse(line); if (msg.type === "get_state") { console.log(JSON.stringify({ id: msg.id, type: "response", command: "get_state", success: true, data: { sessionId: "resumed-session", sessionFile: process.cwd() + "/resumed.jsonl" } })); } else if (msg.type === "prompt") { console.log(JSON.stringify({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "ok" } })); console.log(JSON.stringify({ type: "agent_end", isTerminal: true })); } }); `); await fs.chmod(fake, 0o755);
process.env.TEST_ARGS_LOG = argsLogFile;
// Test 1: nested object with both sessionPath and sessionId -> --resume sessionPath const nestedResume = { resume: { resume: { sessionPath: path.join(temp, "nested-target.jsonl"), sessionId: "nested-uuid-123", }, }, }; const factory = new OmpRpcBackendFactory({ ompExecutable: fake, sessionStateRequestTimeoutMs: 500 }); const resumedLifecycle = []; const session1 = await factory.openSession({ cwd: temp, resume: nestedResume }); await session1.runTurn({ text: "hi" }, { emit: async (event) => resumedLifecycle.push(event.type), persistSessionState: async () => resumedLifecycle.push("persist"), signal: new AbortController().signal, }); await session1.dispose();
const spawnedArgs1 = JSON.parse(await fs.readFile(argsLogFile, "utf8")); const resumeIndex1 = spawnedArgs1.indexOf("--resume"); assert.notEqual(resumeIndex1, -1); assert.equal(spawnedArgs1[resumeIndex1 + 1], path.join(temp, "nested-target.jsonl")); assert.equal(resumedLifecycle[0], "persist");
// Test 2: nested object with only sessionId -> --resume sessionId const idOnlyResume = { resume: { resume: { sessionId: "sess-only-id-999", }, }, }; const session2 = await factory.openSession({ cwd: temp, resume: idOnlyResume }); await session2.runTurn({ text: "hi" }, { emit: async () => {}, signal: new AbortController().signal }); await session2.dispose();
const spawnedArgs2 = JSON.parse(await fs.readFile(argsLogFile, "utf8")); const resumeIndex2 = spawnedArgs2.indexOf("--resume"); assert.notEqual(resumeIndex2, -1); assert.equal(spawnedArgs2[resumeIndex2 + 1], "sess-only-id-999"); } finally { delete process.env.TEST_ARGS_LOG; await fs.rm(temp, { recursive: true, force: true }); }});
test("RPC model discovery failure returns a selectable featured catalog", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-model-fallback-")); try { const fake = path.join(temp, "failing-models.cjs"); await fs.writeFile(fake, "#!/usr/bin/env node\nprocess.exit(1);\n"); await fs.chmod(fake, 0o755); const models = await new OmpRpcBackendFactory({ ompExecutable: fake, modelCatalogCacheFile: path.join(temp, "model-catalog-cache.json"), }).listModels(); assert.ok(models.length > 1); assert.ok(models.some((model) => model.id === "gpt-5.6-sol" && model.selector === "openai-codex/gpt-5.6-sol")); assert.equal( models.find((model) => model.id === "gpt-oss-120b")?.selector, "google-antigravity/gpt-oss-120b" ); assert.equal( models.find((model) => model.id === "MercuriusDream/Nanbeige4.2-3B-mlx-8bit")?.selector, "nanbeige-local/MercuriusDream/Nanbeige4.2-3B-mlx-8bit" ); assert.equal( models.find((model) => model.id === "mlx-community/Laguna-XS-2.1-3bit")?.selector, "laguna-local/mlx-community/Laguna-XS-2.1-3bit" ); assert.ok(models.every((model) => model.id !== "omp/default")); } finally { await fs.rm(temp, { recursive: true, force: true }); }});
test("fresh RPC sessions do not claim an older session before the process starts", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-state-")); const originalAgentDir = process.env.PI_CODING_AGENT_DIR; try { const agentDir = path.join(temp, "agent"); const cwd = path.join(temp, "worktree"); const sessionDir = path.join(agentDir, "sessions", cwdToSessionDirName(cwd)); await fs.mkdir(sessionDir, { recursive: true }); await fs.writeFile(path.join(sessionDir, "old.jsonl"), [ JSON.stringify({ type: "session", version: 3, id: "old-session", timestamp: "2026-08-20T00:00:00.000Z", cwd }), "", ].join("\n")); process.env.PI_CODING_AGENT_DIR = agentDir;
const factory = new OmpRpcBackendFactory({ ompExecutable: "/does/not/start" }); const session = await factory.openSession({ cwd }); assert.deepEqual(await session.getPersistentState(), { kind: "rpc" }); await session.dispose(); } finally { if (originalAgentDir !== undefined) process.env.PI_CODING_AGENT_DIR = originalAgentDir; else delete process.env.PI_CODING_AGENT_DIR; await fs.rm(temp, { recursive: true, force: true }); }});
test("RPC session identity is persisted before the first turn can appear as a duplicate", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-identity-")); const originalAgentDir = process.env.PI_CODING_AGENT_DIR; const originalFakeSessionPath = process.env.FAKE_OMP_SESSION_PATH; try { const agentDir = path.join(temp, "agent"); const cwd = path.join(temp, "worktree"); const sessionDir = path.join(agentDir, "sessions", cwdToSessionDirName(cwd)); const sessionId = "fresh-session"; const sessionPath = path.join(sessionDir, `fresh_${sessionId}.jsonl`); const fake = path.join(temp, "fake-omp.cjs"); await fs.mkdir(sessionDir, { recursive: true }); await fs.mkdir(cwd, { recursive: true }); await fs.writeFile(path.join(sessionDir, "old.jsonl"), [ JSON.stringify({ type: "session", version: 3, id: "old-session", timestamp: new Date().toISOString(), cwd }), JSON.stringify({ type: "message", message: { role: "user", content: [{ type: "text", text: "older same-cwd session" }] } }), "", ].join("\n")); await fs.writeFile(fake, `#!/usr/bin/env node const fs = require("node:fs"); const readline = require("node:readline"); const session = { type: "session", version: 3, id: "fresh-session", timestamp: new Date().toISOString(), cwd: process.cwd() }; const user = { type: "message", message: { role: "user", content: [{ type: "text", text: "visible session" }] } }; fs.writeFileSync(process.env.FAKE_OMP_SESSION_PATH, [session, user].map((entry) => JSON.stringify(entry)).join("\\n") + "\\n"); const rl = readline.createInterface({ input: process.stdin }); rl.on("line", (line) => { const message = JSON.parse(line); if (message.type === "get_state") { console.log(JSON.stringify({ id: message.id, type: "response", command: "get_state", success: true, data: { sessionId: "fresh-session", sessionFile: process.env.FAKE_OMP_SESSION_PATH } })); return; } if (message.type === "prompt") { console.log(JSON.stringify({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "working" } })); setTimeout(() => console.log(JSON.stringify({ type: "agent_end", isTerminal: true })), 200); } }); `); await fs.chmod(fake, 0o755); process.env.PI_CODING_AGENT_DIR = agentDir; process.env.FAKE_OMP_SESSION_PATH = sessionPath;
const h = await makeServerHarness(new OmpRpcBackendFactory({ ompExecutable: fake }), temp); try { const { thread } = await h.request("thread/start", { cwd }); const { turn } = await h.request("turn/start", { threadId: thread.id, input: [{ type: "text", text: "hello" }] }); await h.waitFor((message) => message.method === "item/agentMessage/delta" && message.params.turnId === turn.id, 6_000);
const record = await h.store.getRecord(thread.id); assert.equal(record.backend.sessionId, sessionId); assert.equal(record.backend.sessionPath, sessionPath); const listed = await h.request("thread/list", { cwd, limit: 100 }); assert.equal(listed.data.some((item) => item.id === sessionId), false); await h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === turn.id); } finally { await h.server.close(); } } finally { if (originalAgentDir !== undefined) process.env.PI_CODING_AGENT_DIR = originalAgentDir; else delete process.env.PI_CODING_AGENT_DIR; if (originalFakeSessionPath !== undefined) process.env.FAKE_OMP_SESSION_PATH = originalFakeSessionPath; else delete process.env.FAKE_OMP_SESSION_PATH; await new Promise((resolve) => setTimeout(resolve, 100)); await fs.rm(temp, { recursive: true, force: true }); }});
test("concurrent fresh RPC sessions in one cwd bind distinct native files", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-concurrent-")); const originalAgentDir = process.env.PI_CODING_AGENT_DIR; const originalFakeSessionDir = process.env.FAKE_OMP_SESSION_DIR; try { const agentDir = path.join(temp, "agent"); const cwd = path.join(temp, "worktree"); const sessionDir = path.join(agentDir, "sessions", cwdToSessionDirName(cwd)); const fake = path.join(temp, "fake-omp.cjs"); await fs.mkdir(sessionDir, { recursive: true }); await fs.mkdir(cwd, { recursive: true }); await fs.writeFile(fake, `#!/usr/bin/env node const fs = require("node:fs"); const path = require("node:path"); const readline = require("node:readline"); const sessionId = "session-" + process.pid; const sessionPath = path.join(process.env.FAKE_OMP_SESSION_DIR, sessionId + ".jsonl"); const session = { type: "session", version: 3, id: sessionId, timestamp: new Date().toISOString(), cwd: process.cwd() }; const user = { type: "message", message: { role: "user", content: [{ type: "text", text: sessionId }] } }; fs.writeFileSync(sessionPath, [session, user].map((entry) => JSON.stringify(entry)).join("\\n") + "\\n"); const rl = readline.createInterface({ input: process.stdin }); rl.on("line", (line) => { const message = JSON.parse(line); if (message.type === "get_state") { console.log(JSON.stringify({ id: message.id, type: "response", command: "get_state", success: true, data: { sessionId, sessionFile: sessionPath } })); return; } if (message.type === "prompt") { console.log(JSON.stringify({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: sessionId } })); setTimeout(() => console.log(JSON.stringify({ type: "agent_end", isTerminal: true })), 100); } }); `); await fs.chmod(fake, 0o755); process.env.PI_CODING_AGENT_DIR = agentDir; process.env.FAKE_OMP_SESSION_DIR = sessionDir;
const h = await makeServerHarness(new OmpRpcBackendFactory({ ompExecutable: fake }), temp); try { const [{ thread: first }, { thread: second }] = await Promise.all([ h.request("thread/start", { cwd }), h.request("thread/start", { cwd }), ]); const [{ turn: firstTurn }, { turn: secondTurn }] = await Promise.all([ h.request("turn/start", { threadId: first.id, input: [{ type: "text", text: "first" }] }), h.request("turn/start", { threadId: second.id, input: [{ type: "text", text: "second" }] }), ]); await Promise.all([ h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === firstTurn.id, 8_000), h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === secondTurn.id, 8_000), ]);
const firstRecord = await h.store.getRecord(first.id); const secondRecord = await h.store.getRecord(second.id); assert.match(firstRecord.backend.sessionId, /^session-/); assert.match(secondRecord.backend.sessionId, /^session-/); assert.notEqual(firstRecord.backend.sessionId, secondRecord.backend.sessionId); assert.notEqual(firstRecord.backend.sessionPath, secondRecord.backend.sessionPath); } finally { await h.server.close(); } } finally { if (originalAgentDir !== undefined) process.env.PI_CODING_AGENT_DIR = originalAgentDir; else delete process.env.PI_CODING_AGENT_DIR; if (originalFakeSessionDir !== undefined) process.env.FAKE_OMP_SESSION_DIR = originalFakeSessionDir; else delete process.env.FAKE_OMP_SESSION_DIR; await new Promise((resolve) => setTimeout(resolve, 100)); await fs.rm(temp, { recursive: true, force: true }); }});
test("RPC adapter translates the documented JSONL event stream", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-adapter-")); try { const fake = path.join(temp, "fake-omp.cjs"); await fs.writeFile(fake, `#!/usr/bin/env node const readline = require("node:readline"); const rl = readline.createInterface({ input: process.stdin }); rl.on("line", (line) => { const message = JSON.parse(line); if (message.type === "get_state") { console.log(JSON.stringify({ id: message.id, type: "response", command: "get_state", success: true, data: { sessionId: "rpc-session", sessionFile: process.cwd() + "/rpc-session.jsonl" } })); return; } if (message.type === "prompt") { console.log(JSON.stringify({ type: "auto_retry_start", attempt: 1, maxAttempts: 3, delayMs: 250, errorMessage: "provider quota exhausted", })); console.log(JSON.stringify({ type: "retry_fallback_applied", from: "openai-codex/gpt-5.6-sol", to: "anthropic/claude-opus-5", role: "default", })); console.log(JSON.stringify({ type: "agent_end", willContinue: true })); const answer = "rpc:" + message.message; const assistant = { role: "assistant", content: [{ type: "text", text: answer }], usage: { input: 10, output: 5, totalTokens: 15 }, stopReason: "stop", }; console.log(JSON.stringify({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: answer } })); console.log(JSON.stringify({ type: "message_end", message: assistant })); console.log(JSON.stringify({ type: "turn_end", message: assistant })); console.log(JSON.stringify({ type: "retry_fallback_succeeded", model: "anthropic/claude-opus-5", role: "default", })); console.log(JSON.stringify({ type: "agent_end", isTerminal: true })); } }); `); await fs.chmod(fake, 0o755); const factory = new OmpRpcBackendFactory({ ompExecutable: fake, cwd: temp, sessionDiscoveryTimeoutMs: 0 }); const h = await makeServerHarness(factory, temp); try { const { thread } = await h.request("thread/start", { cwd: temp }); const { turn } = await h.request("turn/start", { threadId: thread.id, input: [{ type: "text", text: "hello" }] }); const completed = await h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === turn.id); assert.equal(completed.params.turn.items.find((item) => item.type === "agentMessage").text, "rpc:hello"); const retryError = h.messages.find((message) => message.method === "error" && message.params?.turnId === turn.id); assert.equal(retryError.params.willRetry, true); assert.equal(retryError.params.error.codexErrorInfo, "usageLimitExceeded"); assert.match(retryError.params.error.additionalDetails, /attempt 1 of 3/); const fallbackWarning = h.messages.find((message) => message.method === "warning"); assert.match(fallbackWarning.params.message, /continuing with anthropic\/claude-opus-5/); const settings = h.messages.findLast((message) => message.method === "thread/settings/updated"); assert.equal(settings.params.threadSettings.model, "claude-opus-5"); assert.equal(settings.params.threadSettings.modelProvider, "anthropic"); } finally { await h.server.close(); } } finally { await fs.rm(temp, { recursive: true, force: true }); }});
test("exhausted OMP retries map to a terminal turn error", async () => { const emitted = []; await mapOmpSessionEvent({ type: "auto_retry_end", success: false, attempt: 3, finalError: "all provider credentials are exhausted", }, { emit: async (event) => emitted.push(event) });
assert.deepEqual(emitted, [{ type: "turnError", message: "all provider credentials are exhausted", additionalDetails: "failed after 3 attempts", code: "omp_retry_exhausted", willRetry: false, terminal: true, }]);});
test("RPC process failures surface their stderr in the native turn error", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-stderr-")); try { const fake = path.join(temp, "failing-omp.cjs"); await fs.writeFile(fake, `#!/usr/bin/env node const readline = require("node:readline"); const rl = readline.createInterface({ input: process.stdin }); rl.on("line", (line) => { const message = JSON.parse(line); if (message.type === "get_state") { console.log(JSON.stringify({ id: message.id, type: "response", command: "get_state", success: true, data: { sessionId: "rpc-failure", sessionFile: process.cwd() + "/rpc-failure.jsonl" } })); return; } if (message.type === "prompt") { console.error("provider returned 503: capacity exhausted"); process.exit(7); } }); `); await fs.chmod(fake, 0o755); const factory = new OmpRpcBackendFactory({ ompExecutable: fake, cwd: temp, sessionDiscoveryTimeoutMs: 0 }); const h = await makeServerHarness(factory, temp); try { const { thread } = await h.request("thread/start", { cwd: temp }); const { turn } = await h.request("turn/start", { threadId: thread.id, input: [{ type: "text", text: "hello" }] }); const error = await h.waitFor((message) => message.method === "error" && message.params?.turnId === turn.id); assert.match(error.params.error.message, /provider returned 503: capacity exhausted/); assert.equal(error.params.error.codexErrorInfo, "serverOverloaded"); assert.match(error.params.error.additionalDetails, /OMP_RPC_EXIT_7/); const completed = await h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === turn.id); assert.equal(completed.params.turn.status, "failed"); } finally { await h.server.close(); } } finally { await fs.rm(temp, { recursive: true, force: true }); }});
test("OMP event mapper preserves rich tool kinds, plans, and goals", async () => { const emitted = []; const contextCwd = path.resolve("/tmp/omp-context"); const context = { options: { cwd: contextCwd }, emit: async (event) => emitted.push(event) }; const states = new Map();
await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "web-1", toolName: "web_search", args: { query: "codex protocol" }, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_end", toolCallId: "web-1", toolName: "web_search", result: { content: [{ type: "text", text: "result" }] }, isError: false, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "read-1", toolName: "read", args: { path: "src/app-server.mjs:120-:raw", cwd: "subdir" }, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "edit-1", toolName: "edit", args: { path: "src/app-server.mjs" }, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "todo-1", toolName: "todo", args: { phases: [{ step: "inspect", status: "active" }, { step: "verify", status: "pending" }] }, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_end", toolCallId: "todo-1", toolName: "todo", result: { details: { phases: [{ name: "implementation", tasks: [{ content: "inspect", status: "in_progress" }, { content: "verify", status: "pending" }], }], }, }, isError: false, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "mcp-1", toolName: "mcp__hospital_notify_parent", args: { message: "done" }, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_end", toolCallId: "mcp-1", toolName: "mcp__hospital_notify_parent", result: { content: [{ type: "text", text: "sent" }] }, }, context, states); await mapOmpSessionEvent({ type: "goal_updated", goal: { objective: "finish the bridge", status: "budget-limited", tokenBudget: 100, tokensUsed: 100, timeUsedSeconds: 5, createdAt: 1_800_000_000_000, updatedAt: 1_800_000_001_000, }, }, context, states);
assert.equal(emitted.find((event) => event.toolCallId === "web-1" && event.type === "toolStart").itemType, "webSearch"); assert.equal(emitted.find((event) => event.toolCallId === "web-1" && event.type === "toolEnd").itemType, "webSearch"); assert.equal(emitted.find((event) => event.toolCallId === "read-1").itemType, "commandExecution"); assert.equal(emitted.find((event) => event.toolCallId === "read-1").cwd, path.join(contextCwd, "subdir")); assert.equal(emitted.find((event) => event.toolCallId === "read-1").commandActions[0].path, path.join(contextCwd, "subdir", "src/app-server.mjs")); assert.equal(emitted.find((event) => event.toolCallId === "read-1").commandActions[0].type, "read"); assert.equal(emitted.find((event) => event.toolCallId === "edit-1").type, "fileStart"); const mcpStart = emitted.find((event) => event.toolCallId === "mcp-1" && event.type === "toolStart"); assert.equal(mcpStart.itemType, "mcpToolCall"); assert.equal(mcpStart.server, "hospital"); assert.equal(mcpStart.toolName, "notify_parent"); assert.deepEqual(emitted.find((event) => event.type === "planUpdate").plan, [ { step: "inspect", status: "inProgress" }, { step: "verify", status: "pending" }, ]); assert.equal(emitted.find((event) => event.type === "goalUpdate").goal.status, "budgetLimited"); assert.equal(emitted.find((event) => event.type === "goalUpdate").goal.createdAt, 1_800_000_000);});
test("OMP edit events produce native unified diffs from workspace snapshots", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-edit-diff-")); try { const file = path.join(temp, "sample.txt"); const beforeLines = Array.from({ length: 100 }, (_value, index) => `line ${index + 1}`); await fs.writeFile(file, `${beforeLines.join("\n")}\n`); const emitted = []; const context = { options: { cwd: temp }, emit: async (event) => emitted.push(event), }; const states = new Map(); await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "edit-diff", toolName: "edit", args: { path: "sample.txt" }, }, context, states); const afterLines = beforeLines.slice(); afterLines[9] = "changed ten"; afterLines[89] = "changed ninety"; await fs.writeFile(file, `${afterLines.join("\n")}\n`); await mapOmpSessionEvent({ type: "tool_execution_end", toolCallId: "edit-diff", toolName: "edit", result: { content: [{ type: "text", text: "updated" }] }, isError: false, }, context, states); const completed = emitted.find((event) => event.type === "fileEnd"); assert.equal(completed.changes[0].path, "sample.txt"); assert.match(completed.changes[0].diff, /-line 10/); assert.match(completed.changes[0].diff, /\+changed ten/); assert.match(completed.changes[0].diff, /-line 90/); assert.match(completed.changes[0].diff, /\+changed ninety/); assert.equal((completed.changes[0].diff.match(/^@@ /gm) || []).length, 2); assert.equal((completed.changes[0].diff.match(/^-(?!--)/gm) || []).length, 2); assert.equal((completed.changes[0].diff.match(/^\+(?!\+\+)/gm) || []).length, 2); assert.doesNotMatch(completed.changes[0].diff, /^-line 1$/m); } finally { await fs.rm(temp, { recursive: true, force: true }); }});
test("OMP edit result details recover per-file diffs when the workspace snapshot misses them", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-edit-detail-diff-")); try { const emitted = []; const context = { options: { cwd: temp }, emit: async (event) => emitted.push(event), }; const states = new Map(); await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "edit-details", toolName: "edit", args: { input: "*** Begin Patch\n[src/one.mjs#A1B2]\n[test/two.mjs#C3D4]\n*** End Patch" }, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_end", toolCallId: "edit-details", toolName: "edit", result: { content: [{ type: "text", text: "applied" }], details: { perFileResults: [ { path: "src/one.mjs", op: "update", diff: " 10|keep ten\n-11|old eleven\n+11|new eleven\n 12|keep twelve", }, { path: "test/two.mjs", op: "add", diff: "+1|brand new\n+2|second line", }, ], }, }, isError: false, }, context, states);
const completed = emitted.find((event) => event.type === "fileEnd"); assert.deepEqual(completed.changes.map((change) => change.path), ["src/one.mjs", "test/two.mjs"]); assert.match(completed.changes[0].diff, /^@@ -10,3 \+10,3 @@$/m); assert.match(completed.changes[0].diff, /^-old eleven$/m); assert.match(completed.changes[0].diff, /^\+new eleven$/m); assert.deepEqual(completed.changes[1].kind, { type: "add" }); assert.match(completed.changes[1].diff, /^new file mode 100644$/m); assert.match(completed.changes[1].diff, /^@@ -0,0 \+1,2 @@$/m); assert.match(completed.changes[1].diff, /^\+brand new$/m); } finally { await fs.rm(temp, { recursive: true, force: true }); }});
test("live mapper keeps device writes visible and still sees route-bracketed patch paths", async () => { const emitted = []; const contextCwd = path.resolve("/tmp/omp-device-context"); const context = { options: { cwd: contextCwd }, emit: async (event) => emitted.push(event) }; const states = new Map(); const routeFile = "web/src/routes/[handle]/[repo]/+page.svelte";
await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "device-1", toolName: "write", args: { path: "xd://browser", content: JSON.stringify({ action: "run", name: "live-status", code: "await tab.observe()" }), }, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_end", toolCallId: "device-1", toolName: "write", result: { content: [{ type: "text", text: "ran" }] }, isError: false, }, context, states); await mapOmpSessionEvent({ type: "tool_execution_start", toolCallId: "route-edit", toolName: "edit", args: { input: `*** Begin Patch\n[${routeFile}#2B51]\nPUT 2.=2:\n+\tconst live = true;\n*** End Patch\n` }, }, context, states);
const deviceStart = emitted.find((event) => event.toolCallId === "device-1" && event.type === "toolStart"); assert.equal(deviceStart.itemType, "commandExecution"); assert.equal(deviceStart.command, "write xd://browser run live-status"); assert.equal(emitted.find((event) => event.toolCallId === "device-1" && event.type === "toolEnd").itemType, "commandExecution"); const routeStart = emitted.find((event) => event.toolCallId === "route-edit"); assert.equal(routeStart.type, "fileStart"); assert.deepEqual(routeStart.changes.map((change) => change.path), [routeFile]);});
test("a final answer with no text deltas still produces the answer", async () => { const emitted = []; const context = { options: { cwd: "/tmp" }, emit: async (event) => emitted.push(event) }; const states = new Map(); const signedText = { type: "text", text: "yep. tantivy does the heavy lifting", textSignature: "{\"v\":1,\"phase\":\"final_answer\"}", }; const thinkingPart = { type: "thinking", thinking: "**Weighing FST cost**" };
await mapOmpSessionEvent({ type: "message_update", assistantMessageEvent: { type: "thinking_delta", contentIndex: 0, delta: "**Weighing" }, }, context, states); // real rpc-ui closes the block whole and may send no deltas await mapOmpSessionEvent({ type: "message_update", assistantMessageEvent: { type: "text_end", contentIndex: 1, content: signedText.text, partial: { role: "assistant", content: [thinkingPart, signedText] }, }, }, context, states); await mapOmpSessionEvent({ type: "message_end", message: { role: "assistant", content: [thinkingPart, signedText], stopReason: "stop", usage: { input: 10, output: 5, totalTokens: 15 }, }, }, context, states); await mapOmpSessionEvent({ type: "turn_end", message: { role: "assistant", content: [thinkingPart, signedText], stopReason: "stop", usage: { input: 10, output: 5, totalTokens: 15 }, }, }, context, states);
assert.deepEqual(emitted.filter((event) => event.type === "assistantText").map((event) => event.text), [ "yep. tantivy does the heavy lifting", "yep. tantivy does the heavy lifting", ]); assert.equal(emitted.filter((event) => event.type === "assistantThinking").at(-1).thinking, "**Weighing FST cost**"); assert.equal(emitted.filter((event) => event.type === "tokenUsage").length, 2);
// don't mistake a user message closing for an answer const before = emitted.length; await mapOmpSessionEvent({ type: "message_end", message: { role: "user", content: [{ type: "text", text: "ask" }] }, }, context, states); assert.equal(emitted.length, before);});
test("rpc-ui cumulative answer partials wait for the complete block", async () => { const emitted = []; const context = { emit: async (event) => emitted.push(event) }; const partial = (text) => ({ role: "assistant", content: [{ type: "text", text }] });
await mapOmpSessionEvent({ type: "message_update", assistantMessageEvent: { type: "text_delta", contentIndex: 0, delta: "nothing,", partial: partial("nothing,"), }, }, context); await mapOmpSessionEvent({ type: "message_update", assistantMessageEvent: { type: "text_delta", contentIndex: 0, delta: "nothing, just a typo.", partial: partial("nothing, just a typo."), }, }, context); assert.deepEqual(emitted, []); await mapOmpSessionEvent({ type: "message_update", assistantMessageEvent: { type: "text_end", contentIndex: 0, content: "nothing, just a typo.", partial: partial("nothing, just a typo."), }, }, context);
assert.deepEqual(emitted, [ { type: "assistantText", text: "nothing, just a typo." }, ]);});