From 19cc1084ded9fe9c1c5db299db57c0e79420a689 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Mon, 24 Aug 2026 18:47:19 +0900 Subject: [PATCH] multi client ask --- extension/control-socket.mjs | 179 +++++++++++++++++++ src/app-server.mjs | 328 ++++++++++++++++++++++++++++++++++- src/backend/omp.mjs | 77 ++++++-- src/client-connection.mjs | 3 +- src/control-client.mjs | 11 ++ src/omp-sessions.mjs | 21 ++- src/store.mjs | 14 +- src/util.mjs | 114 ++++++++++++ test/app-server.test.mjs | 151 ++++++++++++++++ test/control.test.mjs | 162 +++++++++++++++++ test/helpers.mjs | 5 +- test/omp-adapter.test.mjs | 18 +- test/store.test.mjs | 47 ++++- 13 files changed, 1089 insertions(+), 41 deletions(-) diff --git a/extension/control-socket.mjs b/extension/control-socket.mjs index c347125..e2cd77d 100644 --- a/extension/control-socket.mjs +++ b/extension/control-socket.mjs @@ -55,6 +55,8 @@ const MAX_SUBSCRIBER_BUFFER = 512 * 1024; /** One bridge needs one stream; cap accidental/malicious fan-out. */ const MAX_SUBSCRIBERS = 8; +const EXTERNAL_ASK_TIMEOUT_MS = 10 * 60_000; + export function controlSocketPath(sessionId) { return path.join(controlDir(), `${sessionId}.sock`); } @@ -115,6 +117,8 @@ export default function controlSocketExtension(api) { let sequence = 0; const replayBuffer = []; const subscribers = new Set(); + const pendingAsks = new Map(); + let askCounter = 0; let server = null; let boundPath = null; let heartbeatTimer = null; @@ -196,6 +200,157 @@ export default function controlSocketExtension(api) { } }; + const askParameters = { + type: "object", + additionalProperties: false, + required: ["questions"], + properties: { + questions: { + type: "array", + minItems: 1, + maxItems: 5, + items: { + type: "object", + additionalProperties: false, + required: ["id", "question", "options"], + properties: { + id: { type: "string" }, + header: { type: "string" }, + question: { type: "string" }, + multi: { type: "boolean" }, + recommended: { type: "number" }, + options: { + type: "array", + items: { + type: "object", + additionalProperties: false, + required: ["label"], + properties: { + label: { type: "string" }, + description: { type: "string" }, + preview: { type: "string" }, + }, + }, + }, + }, + }, + }, + }, + }; + const Type = api.typebox; + const askToolParameters = Type?.Object + ? Type.Object({ + questions: Type.Array(Type.Object({ + id: Type.String(), + header: Type.Optional(Type.String()), + question: Type.String(), + multi: Type.Optional(Type.Boolean()), + recommended: Type.Optional(Type.Number()), + options: Type.Array(Type.Object({ + label: Type.String(), + description: Type.Optional(Type.String()), + preview: Type.Optional(Type.String()), + })), + }), { minItems: 1, maxItems: 5 }), + }, { additionalProperties: false }) + : askParameters; + + const askResult = (questions, response) => { + if (response?.cancelled) { + return { + content: [{ type: "text", text: "Ask input was cancelled" }], + details: { results: [] }, + isError: true, + }; + } + const answers = response?.answers && typeof response.answers === "object" ? response.answers : {}; + const results = questions.map((question) => { + const selectedOptions = Array.isArray(answers[question.id]?.answers) + ? answers[question.id].answers.map(String) + : []; + const optionLabels = (question.options || []).map((option) => String(option.label || "")); + const customInput = selectedOptions.find((answer) => !optionLabels.includes(answer)); + return { + id: question.id, + question: question.question, + options: optionLabels, + multi: Boolean(question.multi), + selectedOptions: customInput ? selectedOptions.filter((answer) => answer !== customInput) : selectedOptions, + ...(customInput ? { customInput } : {}), + }; + }); + let text; + if (results.length === 1) { + const result = results[0]; + text = result.customInput + ? `User provided custom input: ${result.customInput}` + : `User selected: ${result.selectedOptions.join(", ")}`; + } else { + text = `User answers:\n${results.map((result) => { + const answer = result.customInput || result.selectedOptions.join(", "); + return `${result.id}: ${JSON.stringify(answer)}`; + }).join("\n")}`; + } + const first = results[0]; + return { + content: [{ type: "text", text }], + details: results.length === 1 + ? { + question: first.question, + options: first.options, + multi: first.multi, + selectedOptions: first.selectedOptions, + ...(first.customInput ? { customInput: first.customInput } : {}), + } + : { results }, + isError: false, + }; + }; + + if (typeof api.registerTool === "function") { + api.registerTool({ + name: "ask", + label: "Ask", + description: "Ask the user one or more clarifying questions.", + parameters: askToolParameters, + approval: "read", + strict: true, + async execute(toolCallId, params, signal, onUpdate, ctx) { + if (subscribers.size === 0) { + if (typeof ctx?.invokeTool === "function") { + return ctx.invokeTool(params, { signal, onUpdate }); + } + return askResult(params.questions, { cancelled: true }); + } + const requestId = `ask:${streamId}:${++askCounter}`; + let timer = null; + let abortHandler = null; + const response = await new Promise((resolve, reject) => { + const settle = (value, error = null) => { + clearTimeout(timer); + signal?.removeEventListener("abort", abortHandler); + pendingAsks.delete(requestId); + if (error) reject(error); + else resolve(value); + }; + abortHandler = () => settle(null, signal.reason instanceof Error ? signal.reason : new Error("Ask interrupted")); + signal?.addEventListener("abort", abortHandler, { once: true }); + timer = setTimeout(() => settle({ cancelled: true }), EXTERNAL_ASK_TIMEOUT_MS); + timer.unref?.(); + pendingAsks.set(requestId, { + requestId, + toolCallId, + questions: params.questions, + createdAt: Date.now(), + settle, + }); + publishEvent({ type: "ask_request", requestId, toolCallId, questions: params.questions }, ctx); + }); + return askResult(params.questions, response); + }, + }); + } + for (const eventName of LIFECYCLE_EVENTS) { try { api.on?.(eventName, (event, eventCtx) => { @@ -273,6 +428,12 @@ export default function controlSocketExtension(api) { earliestSequence, reset, status: getStatus(ctx), + pendingAsks: Array.from(pendingAsks.values(), (pending) => ({ + requestId: pending.requestId, + toolCallId: pending.toolCallId, + questions: pending.questions, + createdAt: pending.createdAt, + })), }; return { @@ -316,6 +477,20 @@ export default function controlSocketExtension(api) { api.sendUserMessage(text, { deliverAs: op === "steer" ? "steer" : "followUp" }); return { ok: true, op, delivered: op, idle: ctx.isIdle() }; } + case "ask_response": { + const pending = pendingAsks.get(String(request.requestId || "")); + if (!pending) return { ok: false, op, error: "Ask request is no longer pending" }; + pending.settle({ answers: request.answers || {} }); + publishEvent({ type: "ask_resolved", requestId: pending.requestId, toolCallId: pending.toolCallId }, ctx); + return { ok: true, op, requestId: pending.requestId }; + } + case "ask_cancel": { + const pending = pendingAsks.get(String(request.requestId || "")); + if (!pending) return { ok: false, op, error: "Ask request is no longer pending" }; + pending.settle({ cancelled: true }); + publishEvent({ type: "ask_cancelled", requestId: pending.requestId, toolCallId: pending.toolCallId }, ctx); + return { ok: true, op, requestId: pending.requestId }; + } case "interrupt": ctx.abort(); return { ok: true, op }; @@ -419,6 +594,8 @@ export default function controlSocketExtension(api) { replayBuffer.length = 0; currentCtx = null; clearInterval(heartbeatTimer); + for (const pending of [...pendingAsks.values()]) pending.settle({ cancelled: true }); + pendingAsks.clear(); heartbeatTimer = null; const listener = server; @@ -433,6 +610,8 @@ export default function controlSocketExtension(api) { currentCtx = ctx; streamId = crypto.randomUUID(); sequence = 0; + for (const pending of [...pendingAsks.values()]) pending.settle({ cancelled: true }); + pendingAsks.clear(); replayBuffer.length = 0; for (const socket of subscribers) { try { socket.destroy(); } catch {} diff --git a/src/app-server.mjs b/src/app-server.mjs index d3f5bab..c797d11 100644 --- a/src/app-server.mjs +++ b/src/app-server.mjs @@ -3,7 +3,7 @@ import os from "node:os"; import path from "node:path"; import crypto from "node:crypto"; import { spawn } from "node:child_process"; -import { controlConfigure, controlInterrupt, controlSend, controlStatus, controlSubscribe } from "./control-client.mjs"; +import { controlCancelAsk, controlConfigure, controlInterrupt, controlRespondAsk, controlSend, controlStatus, controlSubscribe } from "./control-client.mjs"; import { ClientConnection } from "./client-connection.mjs"; import { promptMatches, StateStore, turnPromptText } from "./store.mjs"; import { createBackendFactory, FEATURED_MODEL_IDS, syncCodexModelsCache } from "./backend/omp.mjs"; @@ -100,6 +100,126 @@ function itemSignature(item) { } } +const FUZZY_SEARCH_IGNORED_DIRECTORIES = new Set([ + ".git", + ".jj", + ".direnv", + "node_modules", + "target", + "dist", + "build", + ".next", + ".svelte-kit", +]); +const FUZZY_SEARCH_MAX_ENTRIES = 50_000; +const FUZZY_SEARCH_MAX_RESULTS = 100; + +function fuzzyMatch(candidate, query) { + const haystack = String(candidate || "").toLowerCase(); + const needle = String(query || "").trim().toLowerCase(); + if (!needle) return { score: 1, indices: [] }; + const contiguous = haystack.indexOf(needle); + if (contiguous >= 0) { + return { + score: Math.max(1, 10_000 - contiguous * 10 - haystack.length), + indices: Array.from({ length: needle.length }, (_, index) => contiguous + index), + }; + } + const indices = []; + let cursor = 0; + for (const character of needle) { + const found = haystack.indexOf(character, cursor); + if (found < 0) return null; + indices.push(found); + cursor = found + 1; + } + const gaps = indices.at(-1) - indices[0] + 1 - needle.length; + return { score: Math.max(1, 5_000 - gaps * 20 - haystack.length), indices }; +} + +async function searchWorkspaceFiles(query, roots, signal) { + const results = []; + let seenEntries = 0; + for (const requestedRoot of roots || []) { + if (signal?.aborted || seenEntries >= FUZZY_SEARCH_MAX_ENTRIES) break; + if (typeof requestedRoot !== "string" || !path.isAbsolute(requestedRoot)) continue; + const root = await fs.realpath(requestedRoot).catch(() => null); + if (!root) continue; + const stat = await fs.stat(root).catch(() => null); + if (!stat?.isDirectory()) continue; + const queue = [{ absolute: root, relative: "" }]; + while (queue.length && seenEntries < FUZZY_SEARCH_MAX_ENTRIES && !signal?.aborted) { + const current = queue.shift(); + const entries = await fs.readdir(current.absolute, { withFileTypes: true }).catch(() => []); + entries.sort((left, right) => left.name.localeCompare(right.name)); + for (const entry of entries) { + if (signal?.aborted || seenEntries >= FUZZY_SEARCH_MAX_ENTRIES) break; + seenEntries += 1; + const relative = current.relative ? `${current.relative}/${entry.name}` : entry.name; + if (entry.isDirectory() && FUZZY_SEARCH_IGNORED_DIRECTORIES.has(entry.name)) continue; + const match = fuzzyMatch(relative, query); + if (match) { + results.push({ + root, + path: relative, + match_type: entry.isDirectory() ? "directory" : "file", + file_name: entry.name, + score: match.score, + indices: match.indices.length ? match.indices : null, + }); + } + if (entry.isDirectory()) { + queue.push({ absolute: path.join(current.absolute, entry.name), relative }); + } + } + } + } + return results + .sort((left, right) => right.score - left.score || left.path.localeCompare(right.path)) + .slice(0, FUZZY_SEARCH_MAX_RESULTS); +} + +function normalizeUserInputQuestions(request = {}) { + const rawQuestions = Array.isArray(request.questions) && request.questions.length + ? request.questions.slice(0, 3) + : [{ + id: "answer", + header: request.title || "OMP", + question: request.prompt || request.message || "Input requested", + options: request.options || null, + }]; + return rawQuestions.map((question, index) => ({ + id: String(question.id || `answer-${index + 1}`), + header: String(question.header || request.title || "OMP"), + question: String(question.question || question.prompt || question.message || "Input requested"), + isOther: question.isOther !== false, + isSecret: Boolean(question.isSecret), + options: Array.isArray(question.options) && question.options.length + ? question.options.map((option, optionIndex) => { + const rawLabel = String(typeof option === "string" ? option : option?.label || option?.value || ""); + const recommended = Number(question.recommended) === optionIndex && !/\(Recommended\)$/i.test(rawLabel); + return { + label: recommended ? `${rawLabel} (Recommended)` : rawLabel, + description: String(typeof option === "string" ? "" : option?.description || ""), + }; + }) + : null, + })); +} + +function normalizeUserInputAnswers(answers) { + if (!answers || typeof answers !== "object") return {}; + return Object.fromEntries(Object.entries(answers).map(([questionId, answer]) => [ + questionId, + { + ...answer, + answers: Array.isArray(answer?.answers) + ? answer.answers.map((value) => String(value).replace(/\s+\(Recommended\)$/i, "")) + : [], + }, + ])); +} + const FULL_PROTOCOL_TRACE_METHODS = new Set([ "model/list", "collaborationMode/list", @@ -160,6 +280,9 @@ export class CodexAppServer { this.externalControlSyncPromises = new Map(); this.externalControlConnectControllers = new Map(); this.externalRefreshPromise = null; + this.fuzzySearchSessions = new Map(); + this.fuzzySearchRequests = new Map(); + this.externalAskRequests = new Map(); this.closing = false; } @@ -199,6 +322,17 @@ export class CodexAppServer { try { proc.child.kill("SIGTERM"); } catch {} } peer.processes?.clear(); + const fuzzyPrefix = `${peer.id}:`; + for (const [key, session] of this.fuzzySearchSessions) { + if (!key.startsWith(fuzzyPrefix)) continue; + session.controller?.abort(); + this.fuzzySearchSessions.delete(key); + } + for (const [key, controller] of this.fuzzySearchRequests) { + if (!key.startsWith(fuzzyPrefix)) continue; + controller.abort(); + this.fuzzySearchRequests.delete(key); + } } async close() { @@ -212,6 +346,15 @@ export class CodexAppServer { for (const subscription of this.externalControlSubscriptions.values()) subscription.close?.(); this.externalControlSubscriptions.clear(); this.externalControlConnecting.clear(); + for (const session of this.fuzzySearchSessions.values()) session.controller?.abort(); + this.fuzzySearchSessions.clear(); + for (const controller of this.fuzzySearchRequests.values()) controller.abort(); + this.fuzzySearchRequests.clear(); + for (const pending of this.externalAskRequests.values()) { + pending.controller.abort(); + void controlCancelAsk(pending.sessionId, pending.requestId); + } + this.externalAskRequests.clear(); const runningTurns = []; for (const active of this.activeTurns.values()) { active.controller.abort(abortError("Server closing")); @@ -336,6 +479,72 @@ export class CodexAppServer { this.externalControlSyncTimers.set(threadId, { timer, terminal }); } + async #handleExternalAskRequest(threadId, request = {}) { + const requestId = String(request.requestId || ""); + if (!requestId) return; + const key = `${threadId}:${requestId}`; + if (this.externalAskRequests.has(key)) return; + const peers = [...this.peers].filter((peer) => + peer.initialized + && !peer.client.closed + && peer.subscriptions?.has(threadId) + ); + if (!peers.length) return; + const { sessionId } = await this.#controlTarget(threadId); + const turnId = (await this.store.getThread(threadId))?.turns?.at(-1)?.id; + if (!turnId) return; + const controller = new AbortController(); + this.externalAskRequests.set(key, { controller, sessionId, requestId }); + const questions = normalizeUserInputQuestions({ questions: request.questions || [] }); + const itemId = String(request.toolCallId || id("ask")); + const askPeer = async (peer) => { + let serverRequestId = null; + try { + const result = await peer.client.request("item/tool/requestUserInput", { + threadId, + turnId, + itemId, + questions, + isBlocking: true, + autoResolutionMs: null, + }, { + signal: controller.signal, + onRequestId: (value) => { serverRequestId = value; }, + }); + if (!result?.answers || typeof result.answers !== "object") { + throw new Error("Ask response contained no answers"); + } + return result; + } finally { + if (serverRequestId) { + peer.client.notify("serverRequest/resolved", { threadId, requestId: serverRequestId }); + } + } + }; + try { + const result = await Promise.any(peers.map(askPeer)); + controller.abort(abortError("Ask answered by another client")); + await controlRespondAsk(sessionId, requestId, normalizeUserInputAnswers(result.answers)); + } catch (error) { + if (!controller.signal.aborted) { + debug("external ask had no client answer", threadId, requestId, error); + await controlCancelAsk(sessionId, requestId); + } + } finally { + if (this.externalAskRequests.get(key)?.controller === controller) { + this.externalAskRequests.delete(key); + } + } + } + + #settleExternalAsk(threadId, requestId) { + const key = `${threadId}:${String(requestId || "")}`; + const pending = this.externalAskRequests.get(key); + if (!pending) return; + pending.controller.abort(abortError("Ask settled by another client")); + this.externalAskRequests.delete(key); + } + async #handleExternalControlEvent(threadId, frame) { const current = this.externalControlCursors.get(threadId); if (current?.streamId === frame.streamId && Number(frame.sequence || 0) <= Number(current.sequence || 0)) return; @@ -356,6 +565,13 @@ export class CodexAppServer { sequence: frame.sequence ?? null, eventType: eventType || null, }); + if (eventType === "ask_request") { + void this.#handleExternalAskRequest(threadId, frame.event).catch((error) => { + debug("external ask relay failed", threadId, error); + }); + } else if (eventType === "ask_resolved" || eventType === "ask_cancelled") { + this.#settleExternalAsk(threadId, frame.event?.requestId); + } this.#scheduleExternalControlSync(threadId, eventType); } @@ -419,6 +635,12 @@ export class CodexAppServer { }); if (handle.status) await this.#publishExternalStatus(threadId, this.#codexStatusFromControl(handle.status)); if (handle.reset) this.#scheduleExternalControlSync(threadId, "snapshot_reset"); + for (const pendingAsk of handle.pendingAsks || []) { + void this.#handleExternalAskRequest(threadId, { + type: "ask_request", + ...pendingAsk, + }).catch((error) => debug("pending external ask relay failed", threadId, error)); + } return subscription; }).catch((error) => { debug("external control subscribe failed", threadId, error); @@ -601,6 +823,10 @@ export class CodexAppServer { if (!peer.initialized) throw new RpcError(ERROR.NOT_INITIALIZED, "initialize must be called before other methods"); switch (method) { + case "fuzzyFileSearch": return this.#fuzzyFileSearch(peer, params); + case "fuzzyFileSearch/sessionStart": return this.#fuzzyFileSearchSessionStart(peer, params); + case "fuzzyFileSearch/sessionUpdate": return this.#fuzzyFileSearchSessionUpdate(peer, params); + case "fuzzyFileSearch/sessionStop": return this.#fuzzyFileSearchSessionStop(peer, params); case "model/list": return this.#modelList(params); case "modelProvider/capabilities/read": return { namespaceTools: true, imageGeneration: false, webSearch: true }; case "permissionProfile/list": return { data: [], nextCursor: null }; @@ -732,6 +958,71 @@ export class CodexAppServer { return initializeResult(); } + async #fuzzyFileSearch(peer, params = {}) { + const token = typeof params.cancellationToken === "string" && params.cancellationToken + ? params.cancellationToken + : "legacy"; + const key = `${peer.id}:${token}`; + this.fuzzySearchRequests.get(key)?.abort(); + const controller = new AbortController(); + this.fuzzySearchRequests.set(key, controller); + try { + const files = await searchWorkspaceFiles(params.query || "", params.roots || [], controller.signal); + return { files }; + } finally { + if (this.fuzzySearchRequests.get(key) === controller) this.fuzzySearchRequests.delete(key); + } + } + + #fuzzyFileSearchSessionStart(peer, params = {}) { + const sessionId = requireString(params.sessionId, "sessionId"); + const key = `${peer.id}:${sessionId}`; + this.fuzzySearchSessions.get(key)?.controller?.abort(); + this.fuzzySearchSessions.set(key, { + roots: Array.isArray(params.roots) ? params.roots.slice() : [], + generation: 0, + controller: null, + }); + return {}; + } + + #fuzzyFileSearchSessionUpdate(peer, params = {}) { + const sessionId = requireString(params.sessionId, "sessionId"); + const key = `${peer.id}:${sessionId}`; + const session = this.fuzzySearchSessions.get(key); + if (!session) throw new RpcError(ERROR.NOT_FOUND, `Fuzzy file search session not found: ${sessionId}`); + session.controller?.abort(); + const controller = new AbortController(); + const generation = ++session.generation; + session.controller = controller; + const query = String(params.query || ""); + return { + result: {}, + after: async () => { + const files = await searchWorkspaceFiles(query, session.roots, controller.signal).catch((error) => { + debug("fuzzy file search failed", sessionId, error); + return []; + }); + const current = this.fuzzySearchSessions.get(key); + if (controller.signal.aborted || current !== session || current.generation !== generation || peer.client.closed) return; + if (!peer.disabledNotifications.has("fuzzyFileSearch/sessionUpdated")) { + peer.client.notify("fuzzyFileSearch/sessionUpdated", { sessionId, query, files }); + } + if (!peer.disabledNotifications.has("fuzzyFileSearch/sessionCompleted")) { + peer.client.notify("fuzzyFileSearch/sessionCompleted", { sessionId }); + } + }, + }; + } + + #fuzzyFileSearchSessionStop(peer, params = {}) { + const sessionId = requireString(params.sessionId, "sessionId"); + const key = `${peer.id}:${sessionId}`; + this.fuzzySearchSessions.get(key)?.controller?.abort(); + this.fuzzySearchSessions.delete(key); + return {}; + } + async #modelList(params) { const [backend, config] = await Promise.all([this.backendFactoryPromise, this.configPromise]); const codexHome = process.env.CODEX_HOME || path.join(os.homedir(), ".codex"); @@ -2323,6 +2614,7 @@ export class CodexAppServer { if (existingId) return { id: existingId }; return this.#startItem(active, create(), key); } + async #finishItem(active, itemId, status, mutate) { const updated = await this.store.updateItem(active.threadId, active.turnId, itemId, (item) => { if (mutate) mutate(item); @@ -2379,17 +2671,35 @@ export class CodexAppServer { async #requestInput(active, request = {}) { if (active.peer.client.closed) return undefined; - const result = await active.peer.client.request("item/tool/requestUserInput", { - threadId: active.threadId, - turnId: active.turnId, - itemId: request.itemId || request.toolCallId || null, - isBlocking: true, - questions: request.questions || [{ id: "answer", header: request.title || "OMP", question: request.prompt || request.message || "Input requested", options: request.options || null }], - }, { signal: active.controller.signal }); + const questions = normalizeUserInputQuestions(request); + let serverRequestId = null; + let result; + try { + result = await active.peer.client.request("item/tool/requestUserInput", { + threadId: active.threadId, + turnId: active.turnId, + itemId: String(request.itemId || request.toolCallId || id("input")), + questions, + isBlocking: true, + autoResolutionMs: null, + }, { + signal: active.controller.signal, + onRequestId: (requestId) => { serverRequestId = requestId; }, + }); + } finally { + if (serverRequestId) { + active.peer.client.notify("serverRequest/resolved", { + threadId: active.threadId, + requestId: serverRequestId, + }); + } + } if (typeof result === "string") return result; if (result?.answers && typeof result.answers === "object") { const first = Object.values(result.answers)[0]; - if (Array.isArray(first?.answers) && first.answers.length) return first.answers[0]; + if (Array.isArray(first?.answers) && first.answers.length) { + return String(first.answers[0]).replace(/\s+\(Recommended\)$/i, ""); + } if (typeof first === "string") return first; } return result?.answer ?? result?.value ?? null; diff --git a/src/backend/omp.mjs b/src/backend/omp.mjs index 1455ad3..626f3a7 100644 --- a/src/backend/omp.mjs +++ b/src/backend/omp.mjs @@ -1,6 +1,7 @@ import { EventEmitter } from "node:events"; -import { spawn } from "node:child_process"; +import { execFile, spawn } from "node:child_process"; import { createInterface } from "node:readline"; +import { promisify } from "node:util"; import fsSync from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; @@ -9,7 +10,9 @@ import { fileURLToPath, pathToFileURL } from "node:url"; import { MockBackendFactory } from "./mock.mjs"; import { OMP_PACKAGE_SPECIFIERS, OMP_SOURCE_CANDIDATES } from "../generated-omp-bindings.mjs"; import { findSessionDirForCwd, parseOmpSessionIdentity } from "../omp-sessions.mjs"; -import { abortError, atomicWriteJson, commandActionsForOmpTool, debug, extractToolOutputText, formatToolCommand, id, log, serializeError } from "../util.mjs"; +import { abortError, atomicWriteJson, commandActionsForOmpTool, debug, extractToolOutputText, formatToolCommand, id, log, serializeError, unifiedDiffFromText } from "../util.mjs"; + +const diffExecFileAsync = promisify(execFile); const SDK_SPECIFIERS = OMP_PACKAGE_SPECIFIERS; export const FEATURED_MODEL_IDS = [ @@ -778,6 +781,59 @@ async function snapshotFileChanges(changes = [], cwd = process.cwd()) { return snapshots; } +async function gitUnifiedDiff(file, before, after) { + const temporaryDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "omp-codex-diff-")); + const beforePath = path.join(temporaryDirectory, "before"); + const afterPath = path.join(temporaryDirectory, "after"); + try { + if (before != null) await fs.writeFile(beforePath, before); + if (after != null) await fs.writeFile(afterPath, after); + const left = before == null ? "/dev/null" : beforePath; + const right = after == null ? "/dev/null" : afterPath; + let output = ""; + try { + const result = await diffExecFileAsync("git", [ + "diff", + "--no-index", + "--no-color", + "--text", + "--unified=3", + "--", + left, + right, + ], { + timeout: 5_000, + maxBuffer: 32 * 1024 * 1024, + }); + output = String(result.stdout || ""); + } catch (error) { + if (error?.code !== 1 || !error?.stdout) throw error; + output = String(error.stdout); + } + if (!output) return unifiedDiffFromText(file, before, after); + let replacedOldPath = false; + let replacedNewPath = false; + const lines = output.split("\n").map((line) => { + if (line.startsWith("diff --git ")) return `diff --git a/${file} b/${file}`; + if (!replacedOldPath && line.startsWith("--- ")) { + replacedOldPath = true; + return `--- ${before == null ? "/dev/null" : `a/${file}`}`; + } + if (!replacedNewPath && line.startsWith("+++ ")) { + replacedNewPath = true; + return `+++ ${after == null ? "/dev/null" : `b/${file}`}`; + } + return line; + }); + return lines.join("\n"); + } catch (error) { + debug("git diff unavailable; using JavaScript fallback", error); + return unifiedDiffFromText(file, before, after); + } finally { + await fs.rm(temporaryDirectory, { recursive: true, force: true }).catch(() => {}); + } +} + async function finalizeFileChanges(changes = [], snapshots = null, cwd = process.cwd()) { return Promise.all(changes.map(async (change) => { if (change.diff) return change; @@ -791,26 +847,11 @@ async function finalizeFileChanges(changes = [], snapshots = null, cwd = process return { ...change, kind: before == null ? "add" : after == null ? "delete" : "update", - diff: fullFileUnifiedDiff(file, before, after), + diff: await gitUnifiedDiff(file, before, after), }; })); } -function fullFileUnifiedDiff(file, before, after) { - const oldLines = before == null || before === "" ? [] : before.replace(/\n$/, "").split("\n"); - const newLines = after == null || after === "" ? [] : after.replace(/\n$/, "").split("\n"); - const oldPath = before == null ? "/dev/null" : `a/${file}`; - const newPath = after == null ? "/dev/null" : `b/${file}`; - const header = [ - `diff --git a/${file} b/${file}`, - ...(before == null ? ["new file mode 100644"] : []), - ...(after == null ? ["deleted file mode 100644"] : []), - `--- ${oldPath}`, - `+++ ${newPath}`, - `@@ -${oldLines.length ? 1 : 0},${oldLines.length} +${newLines.length ? 1 : 0},${newLines.length} @@`, - ]; - return `${header.concat(oldLines.map((line) => `-${line}`), newLines.map((line) => `+${line}`)).join("\n")}\n`; -} function subscribe(session, listener) { if (typeof session.subscribe === "function") { diff --git a/src/client-connection.mjs b/src/client-connection.mjs index 122eb5e..7a32eac 100644 --- a/src/client-connection.mjs +++ b/src/client-connection.mjs @@ -24,9 +24,10 @@ export class ClientConnection { return this.send({ method, params }); } - async request(method, params = {}, { signal, timeoutMs = 10 * 60_000 } = {}) { + async request(method, params = {}, { signal, timeoutMs = 10 * 60_000, onRequestId = null } = {}) { if (this.#closed) throw new Error("Client connection is closed"); const requestId = `omp:${process.pid}:${++this.#counter}`; + onRequestId?.(requestId); const wait = deferred(); const timer = setTimeout(() => { this.#pending.delete(requestId); diff --git a/src/control-client.mjs b/src/control-client.mjs index 4cf7763..1d07590 100644 --- a/src/control-client.mjs +++ b/src/control-client.mjs @@ -82,6 +82,16 @@ export async function controlConfigure(sessionId, { model = null, reasoningEffor 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); @@ -215,6 +225,7 @@ export function controlSubscribe( set status(val) { currentStatus = val; }, + pendingAsks: Array.isArray(msg.pendingAsks) ? msg.pendingAsks : [], headSequence: ackSequence, reset: isReset, close() { diff --git a/src/omp-sessions.mjs b/src/omp-sessions.mjs index 78f29b6..daaaa5c 100644 --- a/src/omp-sessions.mjs +++ b/src/omp-sessions.mjs @@ -227,14 +227,25 @@ function activityFromMessage(message, mtimeMs) { }; } -function activityMessageFromContent(content) { +const TERMINAL_SESSION_CUSTOM_TYPES = new Set([ + "session_exit", + "session_stop", + "session_shutdown", +]); + +function activityTailFromContent(content) { const lines = content.split("\n"); for (let index = lines.length - 1; index >= 0; index -= 1) { const line = lines[index]; if (!line || line.charCodeAt(0) !== 123) continue; try { const entry = JSON.parse(line); - if (entry.type === "message" && entry.message) return entry.message; + if (entry.type === "custom" && TERMINAL_SESSION_CUSTOM_TYPES.has(entry.customType)) { + return { terminal: true, message: null }; + } + if (entry.type === "message" && entry.message) { + return { terminal: false, message: entry.message }; + } } catch {} } return null; @@ -255,8 +266,8 @@ async function inspectActivityHandle(handle, stat) { const { bytesRead } = await handle.read(buffer, 0, length, start); content = buffer.toString("utf8", 0, bytesRead) + content; scannedBytes += bytesRead; - const message = activityMessageFromContent(content); - if (message) return activityFromMessage(message, stat.mtimeMs); + const tail = activityTailFromContent(content); + if (tail) return activityFromMessage(tail.terminal ? null : tail.message, stat.mtimeMs); if (!bytesRead) break; } if (start > 0) { @@ -515,6 +526,8 @@ export async function parseOmpSessionFile(filePath, options = {}) { }); turns.push(currentTurn); currentTurn = null; + } else if (entry.type === "custom" && TERMINAL_SESSION_CUSTOM_TYPES.has(entry.customType)) { + lastMessage = null; } else if (entry.type === "message") { const msg = entry.message; if (!msg) continue; diff --git a/src/store.mjs b/src/store.mjs index fb54b53..d3d62e2 100644 --- a/src/store.mjs +++ b/src/store.mjs @@ -1,7 +1,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { atomicWriteJson, commandActionsForOmpTool, debug, decodeCursor, encodeCursor, formatToolCommand, paginateArray, readJson, safePreview, unixSeconds } from "./util.mjs"; +import { atomicWriteJson, commandActionsForOmpTool, compactWholeFileUnifiedDiff, debug, decodeCursor, encodeCursor, formatToolCommand, paginateArray, readJson, safePreview, unixSeconds } from "./util.mjs"; import { makeThread } from "./protocol.mjs"; import { inspectOmpSessionActivity, listOmpSessionsForCwd, listAllOmpSessions, parseOmpSessionFile, defaultSessionsDir } from "./omp-sessions.mjs"; const SCHEMA_VERSION = 1; @@ -33,7 +33,17 @@ function storedModelId(thread) { } function normalizeStoredItem(item, cwd) { - if (!item || item.type !== "commandExecution") return item; + if (!item) return item; + if (item.type === "fileChange" && Array.isArray(item.changes)) { + return { + ...item, + changes: item.changes.map((change) => ({ + ...change, + diff: change?.diff ? compactWholeFileUnifiedDiff(change.path || change.file || "file", change.diff) : change?.diff, + })), + }; + } + if (item.type !== "commandExecution") return item; let normalized = item; if ( typeof item.command === "string" diff --git a/src/util.mjs b/src/util.mjs index f2adf2d..2e8e560 100644 --- a/src/util.mjs +++ b/src/util.mjs @@ -383,6 +383,120 @@ export function commandActionsForOmpTool(toolName, args = {}, command = "", cwd return [{ type: "unknown", command }]; } +function diffTextLines(beforeLines, afterLines) { + const beforeLength = beforeLines.length; + const afterLength = afterLines.length; + const maxDistance = beforeLength + afterLength; + const frontier = new Map([[1, 0]]); + const trace = []; + for (let distance = 0; distance <= maxDistance; distance += 1) { + trace.push(new Map(frontier)); + for (let diagonal = -distance; diagonal <= distance; diagonal += 2) { + const moveDown = diagonal === -distance + || (diagonal !== distance && (frontier.get(diagonal - 1) ?? -1) < (frontier.get(diagonal + 1) ?? -1)); + let beforeIndex = moveDown + ? (frontier.get(diagonal + 1) ?? 0) + : (frontier.get(diagonal - 1) ?? 0) + 1; + let afterIndex = beforeIndex - diagonal; + while ( + beforeIndex < beforeLength + && afterIndex < afterLength + && beforeLines[beforeIndex] === afterLines[afterIndex] + ) { + beforeIndex += 1; + afterIndex += 1; + } + frontier.set(diagonal, beforeIndex); + if (beforeIndex >= beforeLength && afterIndex >= afterLength) { + const edits = []; + let x = beforeLength; + let y = afterLength; + for (let backtrackDistance = distance; backtrackDistance >= 0; backtrackDistance -= 1) { + const previous = trace[backtrackDistance]; + const currentDiagonal = x - y; + const movedDown = currentDiagonal === -backtrackDistance + || ( + currentDiagonal !== backtrackDistance + && (previous.get(currentDiagonal - 1) ?? -1) < (previous.get(currentDiagonal + 1) ?? -1) + ); + const previousDiagonal = movedDown ? currentDiagonal + 1 : currentDiagonal - 1; + const previousX = previous.get(previousDiagonal) ?? 0; + const previousY = previousX - previousDiagonal; + while (x > previousX && y > previousY) { + edits.push({ type: " ", line: beforeLines[x - 1] }); + x -= 1; + y -= 1; + } + if (backtrackDistance === 0) break; + if (movedDown) { + edits.push({ type: "+", line: afterLines[y - 1] }); + y -= 1; + } else { + edits.push({ type: "-", line: beforeLines[x - 1] }); + x -= 1; + } + } + return edits.reverse(); + } + } + } + return []; +} + +export function unifiedDiffFromText(file, before, after, contextLines = 3) { + const oldLines = before == null || before === "" ? [] : String(before).replace(/\n$/, "").split("\n"); + const newLines = after == null || after === "" ? [] : String(after).replace(/\n$/, "").split("\n"); + const oldPath = before == null ? "/dev/null" : `a/${file}`; + const newPath = after == null ? "/dev/null" : `b/${file}`; + const header = [ + `diff --git a/${file} b/${file}`, + ...(before == null ? ["new file mode 100644"] : []), + ...(after == null ? ["deleted file mode 100644"] : []), + `--- ${oldPath}`, + `+++ ${newPath}`, + ]; + const edits = diffTextLines(oldLines, newLines); + const changed = edits.flatMap((edit, index) => edit.type === " " ? [] : [index]); + if (!changed.length) return `${header.join("\n")}\n`; + const ranges = []; + for (const index of changed) { + const start = Math.max(0, index - contextLines); + const end = Math.min(edits.length, index + contextLines + 1); + const previous = ranges.at(-1); + if (previous && start <= previous.end) previous.end = Math.max(previous.end, end); + else ranges.push({ start, end }); + } + const body = []; + for (const range of ranges) { + let oldStart = 1; + let newStart = 1; + for (let index = 0; index < range.start; index += 1) { + if (edits[index].type !== "+") oldStart += 1; + if (edits[index].type !== "-") newStart += 1; + } + const slice = edits.slice(range.start, range.end); + const oldCount = slice.filter((edit) => edit.type !== "+").length; + const newCount = slice.filter((edit) => edit.type !== "-").length; + body.push(`@@ -${oldCount ? oldStart : 0},${oldCount} +${newCount ? newStart : 0},${newCount} @@`); + body.push(...slice.map((edit) => `${edit.type}${edit.line}`)); + } + return `${header.concat(body).join("\n")}\n`; +} + +export function compactWholeFileUnifiedDiff(file, diff) { + const lines = String(diff || "").split("\n"); + const hunk = lines.findIndex((line) => line.startsWith("@@ ")); + if (hunk < 0) return diff; + const body = lines.slice(hunk + 1).filter((line) => line.length > 0); + if (body.some((line) => line.startsWith(" "))) return diff; + if (body.some((line) => !line.startsWith("-") && !line.startsWith("+") && line !== "\\ No newline at end of file")) { + return diff; + } + const before = body.filter((line) => line.startsWith("-")).map((line) => line.slice(1)).join("\n"); + const after = body.filter((line) => line.startsWith("+")).map((line) => line.slice(1)).join("\n"); + return unifiedDiffFromText(file, before ? `${before}\n` : "", after ? `${after}\n` : ""); +} + export function formatToolCommand(toolName, args = {}, intent = null) { const name = String(toolName || "tool").toLowerCase(); const i = intent || args?.i; diff --git a/test/app-server.test.mjs b/test/app-server.test.mjs index 607bb64..6218a72 100644 --- a/test/app-server.test.mjs +++ b/test/app-server.test.mjs @@ -191,6 +191,52 @@ test("initialization and model discovery", () => withHarness({}, async (h) => { assert.deepEqual(await h.request("remoteControl/enable", { enabled: true }), {}); })); +test("fuzzy file search supports single-shot and streaming mention completion", () => withHarness({}, async (h) => { + await h.initialize(); + await fs.mkdir(path.join(h.temp, "src"), { recursive: true }); + await fs.mkdir(path.join(h.temp, "test"), { recursive: true }); + await fs.mkdir(path.join(h.temp, "node_modules", "ignored"), { recursive: true }); + await fs.writeFile(path.join(h.temp, "src", "app-server.mjs"), ""); + await fs.writeFile(path.join(h.temp, "test", "app-server.test.mjs"), ""); + await fs.writeFile(path.join(h.temp, "node_modules", "ignored", "app-server-copy.mjs"), ""); + + const single = await h.request("fuzzyFileSearch", { + query: "apptest", + roots: [h.temp], + cancellationToken: "mention-1", + }); + assert.ok(single.files.some((file) => file.path === "test/app-server.test.mjs")); + assert.equal(single.files.some((file) => file.path.includes("node_modules")), false); + assert.ok(single.files.every((file) => + file.root + && file.file_name + && ["file", "directory"].includes(file.match_type) + && Number.isFinite(file.score) + )); + + assert.deepEqual(await h.request("fuzzyFileSearch/sessionStart", { + sessionId: "mention-session", + roots: [h.temp], + }), {}); + assert.deepEqual(await h.request("fuzzyFileSearch/sessionUpdate", { + sessionId: "mention-session", + query: "appserver", + }), {}); + const updated = await h.waitFor((message) => + message.method === "fuzzyFileSearch/sessionUpdated" + && message.params?.sessionId === "mention-session" + ); + assert.equal(updated.params.query, "appserver"); + assert.ok(updated.params.files.some((file) => file.path === "src/app-server.mjs")); + await h.waitFor((message) => + message.method === "fuzzyFileSearch/sessionCompleted" + && message.params?.sessionId === "mention-session" + ); + assert.deepEqual(await h.request("fuzzyFileSearch/sessionStop", { + sessionId: "mention-session", + }), {}); +})); + test("model discovery keeps the dynamic catalog visible and prioritizes requested models", () => { const openedSessions = []; const turnModels = []; @@ -373,6 +419,111 @@ test("a turn streams a native agent item and completes", () => withHarness({}, a const agent = completed.params.turn.items.find((item) => item.type === "agentMessage"); assert.match(agent.text, /Mock OMP response: hello/); })); + +test("tool user-input requests use the complete Desktop schema and return answers", () => { + const base = new MockBackendFactory(); + let clientRequest = null; + const backendFactory = { + listModels: (...args) => base.listModels(...args), + async openSession() { + return { + async runTurn(_prompt, context) { + const answer = await context.requestInput({ + itemId: "ask-item", + questions: [{ + id: "vsock", + header: "vsock blocked", + question: "Load vhost_vsock?", + options: [{ label: "loaded", description: "Continue the test." }], + }], + }); + await context.emit({ type: "textDelta", delta: `answer:${answer}` }); + return { status: "completed" }; + }, + async getPersistentState() { return {}; }, + async interrupt() {}, + async dispose() {}, + }; + }, + }; + return withHarness({ + backendFactory, + clientResponse(message) { + if (message.method === "item/tool/requestUserInput") { + clientRequest = message; + return { answers: { vsock: { answers: ["loaded"] } } }; + } + return { decision: "accept" }; + }, + }, async (h) => { + await h.initialize(); + const started = await h.request("thread/start", { cwd: h.temp, model: "mock/omp" }); + const response = await h.request("turn/start", { + threadId: started.thread.id, + input: [{ type: "text", text: "ask me" }], + }); + const completed = await h.waitFor((message) => + message.method === "turn/completed" && message.params?.turn?.id === response.turn.id + ); + assert.ok(clientRequest); + assert.deepEqual(clientRequest.params, { + threadId: started.thread.id, + turnId: response.turn.id, + itemId: "ask-item", + questions: [{ + id: "vsock", + header: "vsock blocked", + question: "Load vhost_vsock?", + isOther: true, + isSecret: false, + options: [{ label: "loaded", description: "Continue the test." }], + }], + isBlocking: true, + autoResolutionMs: null, + }); + assert.match(completed.params.turn.items.find((item) => item.type === "agentMessage").text, /answer:loaded/); + const resolved = h.messages.find((message) => + message.method === "serverRequest/resolved" + && message.params?.threadId === started.thread.id + ); + assert.deepEqual(resolved?.params, { + threadId: started.thread.id, + requestId: clientRequest.id, + }); + }); +}); + +test("a terminal session marker prevents cancelled external asks from blocking the next turn", () => withHarness({}, async (h) => { + await h.initialize(); + const sessionFile = path.join(h.temp, "cancelled-ask.jsonl"); + await fs.writeFile(sessionFile, [ + JSON.stringify({ type: "session", id: "cancelled-ask-owner", timestamp: new Date().toISOString(), cwd: h.temp }), + JSON.stringify({ type: "title", title: "Cancelled ask" }), + JSON.stringify({ type: "message", message: { role: "user", content: [{ type: "text", text: "first" }] } }), + JSON.stringify({ type: "message", message: { role: "toolResult", toolName: "ask", isError: true, content: [{ type: "text", text: "Ask input was cancelled" }] } }), + JSON.stringify({ type: "custom", customType: "session_exit", data: { reason: "sigterm" } }), + ].join("\n") + "\n"); + const thread = await h.store.createThread({ + threadId: "cancelled-ask-thread", + cwd: h.temp, + preview: "Cancelled ask", + status: { type: "active", activeFlags: [] }, + }); + await h.store.setBackendState(thread.id, { + sessionId: "cancelled-ask-owner", + sessionPath: sessionFile, + resume: { sessionId: "cancelled-ask-owner", sessionPath: sessionFile }, + }); + + const started = await h.request("turn/start", { + threadId: thread.id, + input: [{ type: "text", text: "continue after cancellation" }], + }); + const completed = await h.waitFor((message) => + message.method === "turn/completed" && message.params?.turn?.id === started.turn.id + ); + assert.equal(completed.params.turn.status, "completed"); +})); test("interrupt aborts an active backend turn", () => withHarness({ chunkDelayMs: 15 }, async (h) => { await h.initialize(); const { thread } = await h.request("thread/start", { cwd: h.temp }); diff --git a/test/control.test.mjs b/test/control.test.mjs index d00a308..7a64904 100644 --- a/test/control.test.mjs +++ b/test/control.test.mjs @@ -10,6 +10,8 @@ import { controlInterrupt, controlSend, controlStatus, + controlRespondAsk, + controlCancelAsk, controlSubscribe, controlSocketPath, } from "../src/control-client.mjs"; @@ -60,6 +62,7 @@ function createMockOmp(sessionId = "ext-session", overrides = {}) { 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" }, @@ -70,6 +73,7 @@ function createMockOmp(sessionId = "ext-session", overrides = {}) { 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 }); @@ -120,6 +124,7 @@ function createMockOmp(sessionId = "ext-session", overrides = {}) { sent, configured, models, + tools, emit, start, stop, @@ -257,6 +262,88 @@ test("controlSubscribe delivers live ordered events with monotonic sequence and } })); + +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, { + 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, { + 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(); @@ -560,6 +647,81 @@ test("app-server consumes live control events without waiting for the JSONL poll } })); +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, { + 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("turn/interrupt reaches the OMP process that owns an external thread", () => withControlDir(async () => { const harness = await makeHarness({}); try { diff --git a/test/helpers.mjs b/test/helpers.mjs index 6945f4a..f4370c9 100644 --- a/test/helpers.mjs +++ b/test/helpers.mjs @@ -5,7 +5,7 @@ import { CodexAppServer } from "../src/app-server.mjs"; import { MockBackendFactory } from "../src/backend/mock.mjs"; import { StateStore } from "../src/store.mjs"; -export async function makeHarness({ approval = "accept", chunkDelayMs = 0, backendFactory = null, config = null } = {}) { +export async function makeHarness({ approval = "accept", chunkDelayMs = 0, backendFactory = null, config = null, clientResponse = null } = {}) { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-codex-test-")); const store = new StateStore(path.join(temp, "state.json"), path.join(temp, "sessions")); const configFile = path.join(temp, "config.json"); @@ -22,7 +22,8 @@ export async function makeHarness({ approval = "accept", chunkDelayMs = 0, backe 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: approval } })); + const result = clientResponse ? clientResponse(message) : { decision: approval }; + queueMicrotask(() => server.handle(peer, { id: message.id, result })); } }); let requestId = 0; diff --git a/test/omp-adapter.test.mjs b/test/omp-adapter.test.mjs index 486efae..f52e781 100644 --- a/test/omp-adapter.test.mjs +++ b/test/omp-adapter.test.mjs @@ -697,7 +697,8 @@ test("OMP edit events produce native unified diffs from workspace snapshots", as const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-edit-diff-")); try { const file = path.join(temp, "sample.txt"); - await fs.writeFile(file, "before\n"); + 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 }, @@ -710,7 +711,10 @@ test("OMP edit events produce native unified diffs from workspace snapshots", as toolName: "edit", args: { path: "sample.txt" }, }, context, states); - await fs.writeFile(file, "after\n"); + 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", @@ -720,8 +724,14 @@ test("OMP edit events produce native unified diffs from workspace snapshots", as }, context, states); const completed = emitted.find((event) => event.type === "fileEnd"); assert.equal(completed.changes[0].path, "sample.txt"); - assert.match(completed.changes[0].diff, /-before/); - assert.match(completed.changes[0].diff, /\+after/); + 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 }); } diff --git a/test/store.test.mjs b/test/store.test.mjs index 0f5a748..33cd090 100644 --- a/test/store.test.mjs +++ b/test/store.test.mjs @@ -5,7 +5,7 @@ import os from "node:os"; import path from "node:path"; import { StateStore, turnPromptText, canonicalizePromptText, extractGoalPayload, promptMatches, reconcileParsedTurns } from "../src/store.mjs"; import { cwdToSessionDirName, findSessionDirForCwd, inspectOmpSessionActivity, parseOmpSessionFile, parseOmpSessionHeader, parseOmpSessionIdentity, DEFAULT_EXTERNAL_ACTIVITY_LEASE_MS, MIN_EXTERNAL_ACTIVITY_LEASE_MS, getExternalActivityLeaseMs } from "../src/omp-sessions.mjs"; -import { commandActionsForOmpTool, filesystemPathFromReadTarget, formatToolCommand } from "../src/util.mjs"; +import { commandActionsForOmpTool, compactWholeFileUnifiedDiff, filesystemPathFromReadTarget, formatToolCommand } from "../src/util.mjs"; test("StateStore survives a new instance", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-store-test-")); @@ -142,6 +142,28 @@ test("StateStore can defer streamed item writes until a completion flush", async } }); +test("terminal session markers clear activity after a cancelled ask tool result", async () => { + const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-session-exit-test-")); + try { + const sessionFile = path.join(temp, "session.jsonl"); + await fs.writeFile(sessionFile, [ + JSON.stringify({ type: "session", id: "ask-exit", timestamp: new Date().toISOString(), cwd: temp }), + JSON.stringify({ type: "message", message: { role: "user", content: [{ type: "text", text: "test" }] } }), + JSON.stringify({ type: "message", message: { role: "toolResult", toolName: "ask", isError: true, content: [{ type: "text", text: "Ask input was cancelled" }] } }), + JSON.stringify({ type: "custom", customType: "session_exit", data: { reason: "sigterm" } }), + ].join("\n") + "\n"); + const activity = await inspectOmpSessionActivity(sessionFile); + assert.equal(activity.pending, false); + assert.deepEqual(activity.status, { type: "idle" }); + assert.equal(activity.activeUntil, 0); + const parsed = await parseOmpSessionFile(sessionFile); + assert.deepEqual(parsed.thread.status, { type: "idle" }); + assert.equal(parsed.thread.turns.at(-1).status, "completed"); + } finally { + await fs.rm(temp, { recursive: true, force: true }); + } +}); + test("StateStore discovers and parses native OMP sessions from disk", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-store-sessions-test-")); const originalAgentDir = process.env.PI_CODING_AGENT_DIR; @@ -631,6 +653,29 @@ test("stored SQLite read calls are normalized for native Desktop read pills", as } }); +test("stored whole-file diffs are compacted into contextual hunks", () => { + const oldLines = Array.from({ length: 50 }, (_value, index) => `line ${index + 1}`); + const newLines = oldLines.slice(); + newLines[4] = "changed five"; + newLines[44] = "changed forty-five"; + const full = [ + "diff --git a/sample.txt b/sample.txt", + "--- a/sample.txt", + "+++ b/sample.txt", + "@@ -1,50 +1,50 @@", + ...oldLines.map((line) => `-${line}`), + ...newLines.map((line) => `+${line}`), + "", + ].join("\n"); + const compact = compactWholeFileUnifiedDiff("sample.txt", full); + assert.equal((compact.match(/^@@ /gm) || []).length, 2); + assert.equal((compact.match(/^-(?!--)/gm) || []).length, 2); + assert.equal((compact.match(/^\+(?!\+\+)/gm) || []).length, 2); + assert.match(compact, /-line 5/); + assert.match(compact, /\+changed forty-five/); + assert.doesNotMatch(compact, /^-line 1$/m); +}); + test("StateStore clears fork metadata that points back to the thread's backend session", async () => { const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-dangling-fork-")); try { -- knotmirror 0.0.1