From 6a3c0aca042fb0f01dcca3e3a003998b4589710d Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Mon, 6 Jul 2026 06:44:48 +1000 Subject: [PATCH] feat(cursor): drive cursor-agent acp as a second app-server dialect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the Cursor CLI ACP (Agent Client Protocol, JSON-RPC 2.0 over stdio) dialect alongside codex app-server, sharing the RPC session path. - New src/main/ingest/providers/cursor-acp/ (protocol, driver, normalize, launch) mirroring codex-appserver/. - Hoist the dialect-neutral driver contract to app-server-driver.ts (AppServerDriver, AppServerDriverError, PendingApproval, TurnResult); both dialects implement one interface. - AppServerCapability.protocol dialect field ("codex-app-server" | "acp"); RpcSessionManager dispatches on it. cursor declares launchCmd cursor-agent, args ["acp"], protocol "acp". - Transport always emits the "jsonrpc":"2.0" header (codex tolerates it) and gains respondError; notification/server-request channels are Queue.Dequeue. - codex-approval-view recognizes ACP {optionId, name, kind} options — labels by name, answers with optionId; codex decisions stay opaque passthrough. ACP specifics the driver handles: turn completion is the session/prompt response ({stopReason}), not a notification; the user turn is never echoed (synthesized from the prompt text); session/load replays the prior transcript as notifications before its response (drained post-handshake). The turn folds session/update after the response resolves — the transport routes stdout through one sequential fiber, so every update is queued before the response, making a single Queue.clear race-free. pnpm typecheck clean; pnpm test 531 passed / 2 skipped (live-gated). --- .../ingest/providers/app-server-driver.ts | 68 +++++ .../providers/codex-appserver/driver.ts | 69 ++--- .../providers/codex-appserver/launch.ts | 27 +- .../providers/codex-appserver/transport.ts | 51 +++- .../ingest/providers/cursor-acp/driver.ts | 283 ++++++++++++++++++ .../ingest/providers/cursor-acp/launch.ts | 39 +++ .../ingest/providers/cursor-acp/normalize.ts | 98 ++++++ .../ingest/providers/cursor-acp/protocol.ts | 100 +++++++ src/main/services/ChatMessageService.ts | 2 +- src/main/services/CodexDriverRegistry.ts | 14 +- src/main/services/ProviderRegistry.ts | 4 + src/main/services/RpcSessionManager.ts | 28 +- src/main/services/SessionRuntimeRouter.ts | 10 +- src/main/services/codex-approval-view.ts | 22 +- src/shared/provider.ts | 7 + tests/codex-approval-view.test.ts | 27 ++ tests/codex-appserver-driver.test.ts | 8 +- tests/codex-appserver-transport.test.ts | 4 +- tests/cursor-acp-driver.test.ts | 171 +++++++++++ tests/cursor-acp-live.test.ts | 92 ++++++ tests/cursor-acp-normalize.test.ts | 80 +++++ 21 files changed, 1088 insertions(+), 116 deletions(-) create mode 100644 src/main/ingest/providers/app-server-driver.ts create mode 100644 src/main/ingest/providers/cursor-acp/driver.ts create mode 100644 src/main/ingest/providers/cursor-acp/launch.ts create mode 100644 src/main/ingest/providers/cursor-acp/normalize.ts create mode 100644 src/main/ingest/providers/cursor-acp/protocol.ts create mode 100644 tests/cursor-acp-driver.test.ts create mode 100644 tests/cursor-acp-live.test.ts create mode 100644 tests/cursor-acp-normalize.test.ts diff --git a/src/main/ingest/providers/app-server-driver.ts b/src/main/ingest/providers/app-server-driver.ts new file mode 100644 index 0000000..27701c2 --- /dev/null +++ b/src/main/ingest/providers/app-server-driver.ts @@ -0,0 +1,68 @@ +import { Data, type Effect, type SubscriptionRef } from "effect" +import type { ExtractedRows } from "../db/schema.js" + +/** + * The dialect-neutral contract every resident RPC agent driver implements. Two + * dialects speak it today: `codex app-server` (thread/turn/item — see + * `codex-appserver/driver.ts`) and Cursor's ACP (session/prompt/update — see + * `cursor-acp/driver.ts`). Both fold a live JSON-RPC process into the same + * {@link ExtractedRows} the rollout-file scrapers produce, surface an approval + * signal, and answer it — so the {@link RpcSessionManager}/{@link CodexDriverRegistry} + * seam holds one interface and picks the dialect by `AppServerCapability.protocol`. + */ +export class AppServerDriverError extends Data.TaggedError("AppServerDriverError")<{ + readonly message: string + readonly cause?: unknown +}> {} + +/** An outstanding approval request — the pending-input signal for the app-server path. */ +export interface PendingApproval { + /** The JSON-RPC request id — the reliable routing key for `answerApproval`. */ + readonly id: number | string + /** Dialect approval handle when present (codex commandExecution only) — display/correlation detail. */ + readonly approvalId: string | null + /** Links to the exact tool-call item already in the projection. */ + readonly itemId: string | null + readonly command: string | null + /** + * Server-supplied allowable answers; the UI offers these verbatim. Codex + * decisions are opaque values (string or rule-carrying object) echoed back; + * ACP decisions are `{ optionId, name, kind }` options rendered by `name` and + * answered by `optionId` (see `codex-approval-view.ts`). + */ + readonly availableDecisions: ReadonlyArray +} + +export interface TurnResult { + /** `completed` | `interrupted` | `failed` | `unknown`. */ + readonly status: string + /** The session's cumulative rows through this turn — ready for `IngestStore.replaceSession`. */ + readonly rows: ExtractedRows +} + +/** + * The launch params the {@link RpcSessionManager} hands whichever dialect factory + * it picks — one shape so the factory reference is a clean union. `sandbox` / + * `approvalPolicy` are codex thread config (ACP ignores them); `resumeThreadId` + * rejoins an existing session by native id. + */ +export interface AppServerLaunchParams { + readonly cwd: string + readonly model?: string + readonly sandbox?: "read-only" | "workspace-write" | "danger-full-access" + readonly approvalPolicy?: "untrusted" | "on-failure" | "on-request" | "never" + readonly env?: Record + readonly clientName?: string + readonly resumeThreadId?: string +} + +export interface AppServerDriver { + /** The native session id — codex thread id / ACP session id. */ + readonly threadId: string + /** Send a user turn and resolve once it completes, with the session's cumulative rows + status. */ + readonly runTurn: (text: string) => Effect.Effect + /** The current outstanding approvals; subscribe via `.changes` for the live signal. */ + readonly pendingApprovals: SubscriptionRef.SubscriptionRef> + /** Answer an approval with a server-offered decision (the driver shapes the dialect envelope). */ + readonly answerApproval: (id: number | string, decision: unknown) => Effect.Effect +} diff --git a/src/main/ingest/providers/codex-appserver/driver.ts b/src/main/ingest/providers/codex-appserver/driver.ts index 6f07756..a4e53af 100644 --- a/src/main/ingest/providers/codex-appserver/driver.ts +++ b/src/main/ingest/providers/codex-appserver/driver.ts @@ -1,13 +1,18 @@ -import { Data, Effect, Option, Queue, type Scope, Stream, SubscriptionRef } from "effect" -import type { ExtractedRows } from "../../db/schema.js" +import { Effect, Option, Queue, type Scope, Stream, SubscriptionRef } from "effect" +import { + type AppServerDriver, + AppServerDriverError, + type PendingApproval, + type TurnResult, +} from "../app-server-driver.js" import { obj, str } from "../../extract/json.js" import { normalizeAppServerThread } from "./normalize.js" import { type ApprovalRequestParams, decodeApprovalParams, decodeThreadStart } from "./protocol.js" import { type AppServerTransport, type AppServerTransportError, makeAppServerTransport } from "./transport.js" /** - * The codex app-server driver: the one place that knows the protocol. It owns a - * live `codex app-server` process (via the generic transport), runs the + * The codex app-server driver: the one place that knows the codex dialect. It + * owns a live `codex app-server` process (via the generic transport), runs the * `initialize → thread/start → turn/start` handshake, folds each turn's * `item/completed` + `thread/tokenUsage/updated` stream into the shared * `ExtractedRows` (via {@link normalizeAppServerThread}), and runs the approval @@ -16,43 +21,9 @@ import { type AppServerTransport, type AppServerTransportError, makeAppServerTra * * It is a *service* that owns process + thread state, not an `AgentProvider` * (whose `collect` is a stateless parse pass). Wire types never escape this file: - * callers see `ExtractedRows`, a `PendingApproval` list, and a turn status. + * callers see the dialect-neutral {@link AppServerDriver} — `ExtractedRows`, a + * `PendingApproval` list, and a turn status. */ -export class CodexDriverError extends Data.TaggedError("CodexDriverError")<{ - readonly message: string - readonly cause?: unknown -}> {} - -/** An outstanding approval request — the pending-input signal for the app-server path. */ -export interface PendingApproval { - /** The JSON-RPC request id — the reliable routing key for `answerApproval`. */ - readonly id: number | string - /** Codex's approval handle when present (commandExecution only) — display/correlation detail. */ - readonly approvalId: string | null - /** Links to the exact tool-call item already in the projection. */ - readonly itemId: string | null - readonly command: string | null - /** Server-supplied allowable answers; the UI offers these verbatim. */ - readonly availableDecisions: ReadonlyArray -} - -export interface TurnResult { - /** `completed` | `interrupted` | `failed` | `unknown`. */ - readonly status: string - /** The session's cumulative rows through this turn — ready for `IngestStore.replaceSession`. */ - readonly rows: ExtractedRows -} - -export interface CodexAppServerDriver { - readonly threadId: string - /** Send a user turn and resolve once it completes, with the session's cumulative rows + status. */ - readonly runTurn: (text: string) => Effect.Effect - /** The current outstanding approvals; subscribe via `.changes` for the live signal. */ - readonly pendingApprovals: SubscriptionRef.SubscriptionRef> - /** Answer an approval with a server-offered decision (pass it through verbatim). */ - readonly answerApproval: (id: number | string, decision: unknown) => Effect.Effect -} - export interface CodexDriverOptions { readonly cwd: string readonly model?: string @@ -84,12 +55,12 @@ type TurnSignal = { readonly kind: "completed"; readonly outcome: TurnOutcome } const wrap = (message: string) => - (effect: Effect.Effect): Effect.Effect => - effect.pipe(Effect.mapError((cause) => new CodexDriverError({ message, cause }))) + (effect: Effect.Effect): Effect.Effect => + effect.pipe(Effect.mapError((cause) => new AppServerDriverError({ message, cause }))) export const makeCodexAppServerDriver = ( options: CodexDriverOptions, -): Effect.Effect => +): Effect.Effect => Effect.gen(function* () { const transport: AppServerTransport = yield* makeAppServerTransport({ command: options.command ?? "codex", @@ -120,7 +91,7 @@ export const makeCodexAppServerDriver = ( const thread = decodeThreadStart(started) if (Option.isNone(thread)) { return yield* Effect.fail( - new CodexDriverError({ message: `${method} returned no thread id`, cause: started }), + new AppServerDriverError({ message: `${method} returned no thread id`, cause: started }), ) } const threadId = thread.value.thread.id @@ -136,7 +107,7 @@ export const makeCodexAppServerDriver = ( // Single-consumer, so the mutable accumulators need no locking. const items: Array = [] const usage: Array = [] - yield* transport.notifications.pipe( + yield* Stream.fromQueue(transport.notifications).pipe( Stream.runForEach((notification) => Effect.gen(function* () { switch (notification.method) { @@ -185,7 +156,7 @@ export const makeCodexAppServerDriver = ( // Approvals: record each `requestApproval` as a pending-input signal. We do // not auto-answer — the UI (or a headless policy) calls `answerApproval`. // Any other server request is answered emptily so the server never blocks. - yield* transport.serverRequests.pipe( + yield* Stream.fromQueue(transport.serverRequests).pipe( Stream.runForEach((request) => Effect.gen(function* () { if (!request.method.endsWith("requestApproval")) { @@ -208,7 +179,7 @@ export const makeCodexAppServerDriver = ( Effect.forkScoped, ) - const runTurn = (text: string): Effect.Effect => + const runTurn = (text: string): Effect.Effect => Effect.gen(function* () { yield* transport .request("turn/start", { threadId, input: [{ type: "text", text }] }) @@ -216,7 +187,7 @@ export const makeCodexAppServerDriver = ( const signal = yield* Queue.take(turnOutcomes) if (signal.kind === "aborted") { return yield* Effect.fail( - new CodexDriverError({ message: "codex app-server exited before the turn completed" }), + new AppServerDriverError({ message: "codex app-server exited before the turn completed" }), ) } const outcome = signal.outcome @@ -229,7 +200,7 @@ export const makeCodexAppServerDriver = ( return { status: outcome.status, rows } }) - const answerApproval = (id: number | string, decision: unknown): Effect.Effect => + const answerApproval = (id: number | string, decision: unknown): Effect.Effect => transport .respond(id, { decision }) .pipe( diff --git a/src/main/ingest/providers/codex-appserver/launch.ts b/src/main/ingest/providers/codex-appserver/launch.ts index bf043a9..9ebb38d 100644 --- a/src/main/ingest/providers/codex-appserver/launch.ts +++ b/src/main/ingest/providers/codex-appserver/launch.ts @@ -2,22 +2,11 @@ import { Effect, type Scope } from "effect" import type { AppServerCapability } from "../../../../shared/provider.js" import { IngestStore } from "../../db/store.js" import { - type CodexAppServerDriver, - CodexDriverError, - type CodexDriverOptions, - makeCodexAppServerDriver, -} from "./driver.js" - -export interface CodexLaunchParams { - readonly cwd: string - readonly model?: string - readonly sandbox?: CodexDriverOptions["sandbox"] - readonly approvalPolicy?: CodexDriverOptions["approvalPolicy"] - readonly env?: Record - readonly clientName?: string - /** Rejoin an existing thread by id (`thread/resume`) instead of starting fresh. */ - readonly resumeThreadId?: string -} + type AppServerDriver, + AppServerDriverError, + type AppServerLaunchParams, +} from "../app-server-driver.js" +import { makeCodexAppServerDriver } from "./driver.js" /** * Launch a codex app-server driver from a provider's {@link AppServerCapability} @@ -32,8 +21,8 @@ export interface CodexLaunchParams { */ export const launchCodexAppServerSession = ( capability: AppServerCapability, - params: CodexLaunchParams, -): Effect.Effect => + params: AppServerLaunchParams, +): Effect.Effect => Effect.gen(function* () { const store = yield* IngestStore const driver = yield* makeCodexAppServerDriver({ @@ -53,7 +42,7 @@ export const launchCodexAppServerSession = ( Effect.tap((result) => store .replaceSession(result.rows) - .pipe(Effect.mapError((cause) => new CodexDriverError({ message: "persist turn failed", cause }))), + .pipe(Effect.mapError((cause) => new AppServerDriverError({ message: "persist turn failed", cause }))), ), ) diff --git a/src/main/ingest/providers/codex-appserver/transport.ts b/src/main/ingest/providers/codex-appserver/transport.ts index bc28fe0..617ab0b 100644 --- a/src/main/ingest/providers/codex-appserver/transport.ts +++ b/src/main/ingest/providers/codex-appserver/transport.ts @@ -4,10 +4,12 @@ import { Data, Deferred, Effect, Queue, type Scope, Stream } from "effect" /** * A generic newline-delimited JSON-RPC 2.0 client over a child process's stdio — - * the transport `codex app-server` speaks (`--listen stdio://`, the default). - * Deliberately Codex-agnostic: it knows only the three JSON-RPC message shapes, - * so the codex-specific method names and payload schemas live one layer up in - * the adapter. Framing omits the `"jsonrpc":"2.0"` header, matching the server. + * the transport both `codex app-server` and Cursor's ACP (`cursor-agent acp`) + * speak. Deliberately dialect-agnostic: it knows only the three JSON-RPC message + * shapes, so the per-dialect method names and payload schemas live one layer up + * in the adapter (codex-appserver / cursor-acp). Every outbound frame carries the + * `"jsonrpc":"2.0"` header — ACP requires it and codex app-server tolerates it + * (verified against codex-cli 0.142.x), so there is no framing asymmetry. * * Three inbound shapes are routed: * - **response** — has `id` + (`result`|`error`), no `method` → resolves the @@ -43,6 +45,12 @@ export interface JsonRpcServerRequest { readonly params: unknown } +export interface JsonRpcErrorObject { + readonly code: number + readonly message: string + readonly data?: unknown +} + export interface AppServerTransport { /** Send a request and await its response `result` (decode with Schema). Fails on a JSON-RPC error or process death. */ readonly request: (method: string, params?: unknown) => Effect.Effect @@ -50,10 +58,22 @@ export interface AppServerTransport { readonly notify: (method: string, params?: unknown) => Effect.Effect /** Answer a server→client request (e.g. an approval decision). */ readonly respond: (id: IdKey, result: unknown) => Effect.Effect - /** Server notifications, in arrival order. Ends when the process exits. */ - readonly notifications: Stream.Stream - /** Server→client requests, in arrival order. Ends when the process exits. */ - readonly serverRequests: Stream.Stream + /** + * Reject a server→client request with a JSON-RPC error — for a blocking request + * the client cannot fulfil (e.g. ACP's `cursor/ask_question`), where answering + * with an empty `{}` result would corrupt it. + */ + readonly respondError: (id: IdKey, error: JsonRpcErrorObject) => Effect.Effect + /** + * Server notifications, in arrival order. Exposed as the raw queue (not a + * Stream) so a caller can `Queue.takeAll` the currently-buffered set without + * blocking — the ACP driver drains `session/load` replay this way. Shut down + * on process exit, so a `Stream.fromQueue` fold ends (and a blocked `take` is + * interrupted) when the server dies. + */ + readonly notifications: Queue.Dequeue + /** Server→client requests, in arrival order. Shut down on process exit. */ + readonly serverRequests: Queue.Dequeue } export interface AppServerTransportOptions { @@ -191,7 +211,9 @@ export const makeAppServerTransport = ( const writeLine = (payload: Rec): Effect.Effect => Effect.try({ try: () => { - stdin.write(`${JSON.stringify(payload)}\n`) + // The `"jsonrpc":"2.0"` header on every outbound frame — required by ACP, + // tolerated by codex app-server (see the module doc), so always emitted. + stdin.write(`${JSON.stringify({ jsonrpc: "2.0", ...payload })}\n`) }, catch: (cause) => new AppServerTransportError({ message: "write to app-server failed", cause }), }) @@ -220,11 +242,18 @@ export const makeAppServerTransport = ( const respond = (id: IdKey, result: unknown): Effect.Effect => closed ? Effect.fail(closed) : writeLine({ id, result }) + const respondError = ( + id: IdKey, + error: JsonRpcErrorObject, + ): Effect.Effect => + closed ? Effect.fail(closed) : writeLine({ id, error: { ...error } }) + return { request, notify, respond, - notifications: Stream.fromQueue(notifications), - serverRequests: Stream.fromQueue(serverRequests), + respondError, + notifications, + serverRequests, } satisfies AppServerTransport }) diff --git a/src/main/ingest/providers/cursor-acp/driver.ts b/src/main/ingest/providers/cursor-acp/driver.ts new file mode 100644 index 0000000..ec85a3b --- /dev/null +++ b/src/main/ingest/providers/cursor-acp/driver.ts @@ -0,0 +1,283 @@ +import { Effect, Option, Queue, type Scope, Stream, SubscriptionRef } from "effect" +import { + type AppServerDriver, + AppServerDriverError, + type PendingApproval, + type TurnResult, +} from "../app-server-driver.js" +import { obj, str } from "../../extract/json.js" +import { type AcpItem, normalizeAcpSession } from "./normalize.js" +import { + decodePermissionParams, + decodePromptResult, + decodeSessionNew, + decodeSessionUpdate, + type RequestPermissionParams, +} from "./protocol.js" +import { + type AppServerTransport, + type AppServerTransportError, + type JsonRpcNotification, + makeAppServerTransport, +} from "../codex-appserver/transport.js" + +/** + * The Cursor ACP driver: the one place that knows the ACP dialect. It owns a live + * `cursor-agent acp` process (via the generic transport), runs the + * `initialize → session/new` (or `session/load` for resume) handshake, folds each + * turn's `session/update` notification stream into the shared `ExtractedRows` + * (via {@link normalizeAcpSession}), and turns `session/request_permission` + * server requests into the same {@link PendingApproval} signal the codex driver + * uses — so both dialects present one {@link AppServerDriver} to the seam. + * + * ACP differs from codex in two ways the fold handles here: the turn completion + * signal is the `session/prompt` *response* (`{ stopReason }`), not a notification, + * and the user turn is never echoed back as an update — so `runTurn` synthesizes + * the user message from the prompt text and drives the notification fold itself + * (owning the item list for the turn's duration), rather than a standing fold + * fiber offering completed turns. + */ +export interface CursorAcpDriverOptions { + readonly cwd: string + readonly model?: string + /** Defaults to `cursor-agent`. */ + readonly command?: string + /** Defaults to `["acp"]`. */ + readonly args?: ReadonlyArray + readonly env?: Record + /** Rejoin an existing session by id (`session/load`) instead of starting fresh (`session/new`). */ + readonly resumeThreadId?: string +} + +const wrap = + (message: string) => + (effect: Effect.Effect): Effect.Effect => + effect.pipe(Effect.mapError((cause) => new AppServerDriverError({ message, cause }))) + +/** Map an ACP `stopReason` to a turn status. `end_turn` is the clean completion. */ +const statusFromStopReason = (stopReason: string | undefined): string => { + switch (stopReason) { + case "end_turn": + return "completed" + case "cancelled": + return "interrupted" + default: + return stopReason ?? "unknown" + } +} + +export const makeCursorAcpDriver = ( + options: CursorAcpDriverOptions, +): Effect.Effect => + Effect.gen(function* () { + const transport: AppServerTransport = yield* makeAppServerTransport({ + command: options.command ?? "cursor-agent", + args: options.args ?? ["acp"], + cwd: options.cwd, + env: options.env, + }).pipe(wrap("failed to spawn cursor-agent acp")) + + // Handshake: initialize → session/new (fresh) or session/load (rejoin by id). + // `session/new` returns the sessionId; `session/load` takes it as input and + // returns only modes/models, so the resumed id is the one we passed in. + yield* transport + .request("initialize", { + protocolVersion: 1, + clientCapabilities: { fs: { readTextFile: false, writeTextFile: false } }, + }) + .pipe(wrap("initialize failed")) + + let sessionId: string + if (options.resumeThreadId) { + yield* transport + .request("session/load", { sessionId: options.resumeThreadId, cwd: options.cwd, mcpServers: [] }) + .pipe(wrap("session/load failed")) + sessionId = options.resumeThreadId + } else { + const created = yield* transport + .request("session/new", { cwd: options.cwd, mcpServers: [] }) + .pipe(wrap("session/new failed")) + const decoded = decodeSessionNew(created) + if (Option.isNone(decoded)) { + return yield* Effect.fail( + new AppServerDriverError({ message: "session/new returned no sessionId", cause: created }), + ) + } + sessionId = decoded.value.sessionId + } + + // Discard any `session/load` replay: cursor replays the prior transcript as + // `session/update` notifications *before* the load response resolves, so by + // now they are all buffered on the transport's notifications queue. Draining + // it here (race-free — the transport routes stdout in order, so every replay + // frame was offered before the response we already awaited) keeps `runTurn` + // accumulating only new turns, matching the codex resume semantics (the prior + // rows are already persisted). `session/new` buffers nothing, so this no-ops. + yield* Queue.clear(transport.notifications) + + const pendingApprovals = yield* SubscriptionRef.make>([]) + + // Session-cumulative item list + the folder that mutates it. `pendingText` + // buffers `agent_message_chunk` deltas; it is flushed as one assistant message + // when a `tool_call` arrives or the turn completes. `toolIndex` upserts a tool + // by `toolCallId` so a later `tool_call_update` merges onto the same item, + // preserving order. Only ever touched by the single active turn's fold (below) + // + the between-turns drain, so no locking is needed. + const items: Array = [] + const toolIndex = new Map() + let pendingText: Array = [] + + const flushText = (): void => { + if (pendingText.length === 0) return + items.push({ kind: "message", role: "assistant", text: pendingText.join("") }) + pendingText = [] + } + + const readExecOutput = (rawOutput: unknown): { exitCode: number | null; output: string | null } => { + const o = obj(rawOutput) + if (!o) return { exitCode: null, output: rawOutput != null ? JSON.stringify(rawOutput) : null } + const exitCode = typeof o["exitCode"] === "number" ? o["exitCode"] : null + const stdout = str(o["stdout"]) ?? "" + const stderr = str(o["stderr"]) ?? "" + const hasStd = "stdout" in o || "stderr" in o + const output = hasStd ? stdout + stderr : JSON.stringify(rawOutput) + return { exitCode, output } + } + + const foldNotification = (notification: JsonRpcNotification): void => { + const decoded = decodeSessionUpdate(notification.params) + if (Option.isNone(decoded)) return // unknown variant (info/commands/replay-user) → skip + const update = decoded.value.update + switch (update.sessionUpdate) { + case "agent_message_chunk": { + const text = update.content?.text + if (text) pendingText.push(text) + break + } + case "tool_call": { + // A tool boundary flushes the assistant text accumulated before it, so + // the message lands on its own ordinal ahead of the tool. + flushText() + const input = obj(update.rawInput) ?? null + items.push({ + kind: "tool", + toolCallId: update.toolCallId, + title: update.title ?? null, + toolKind: update.kind ?? null, + command: str(input?.["command"]) ?? null, + exitCode: null, + output: null, + input, + }) + toolIndex.set(update.toolCallId, items.length - 1) + break + } + case "tool_call_update": { + const idx = toolIndex.get(update.toolCallId) + const existing = idx != null ? items[idx] : undefined + const merged = update.rawOutput !== undefined ? readExecOutput(update.rawOutput) : null + if (existing && existing.kind === "tool") { + items[idx!] = { + ...existing, + exitCode: merged?.exitCode ?? existing.exitCode, + output: merged?.output ?? existing.output, + command: existing.command ?? str(obj(update.rawInput)?.["command"]) ?? null, + } + } else { + // An update with no preceding tool_call — record it so the output is + // not lost, keyed for any further updates. + const input = obj(update.rawInput) ?? null + items.push({ + kind: "tool", + toolCallId: update.toolCallId, + title: update.title ?? null, + toolKind: update.kind ?? null, + command: str(input?.["command"]) ?? null, + exitCode: merged?.exitCode ?? null, + output: merged?.output ?? null, + input, + }) + toolIndex.set(update.toolCallId, items.length - 1) + } + break + } + } + } + + // Permissions: a `session/request_permission` server request becomes the + // pending-input signal (answered via `answerApproval`). Any other blocking + // server request (Cursor's `cursor/ask_question` / `cursor/create_plan` + // extensions, etc.) is rejected with a JSON-RPC error rather than an empty + // `{}` result — a bare `{}` would corrupt those request shapes. (They could + // later ride this same pending-input surface instead of being rejected.) + yield* Stream.fromQueue(transport.serverRequests).pipe( + Stream.runForEach((request) => + Effect.gen(function* () { + if (request.method !== "session/request_permission") { + yield* transport + .respondError(request.id, { code: -32601, message: `Unsupported request: ${request.method}` }) + .pipe(Effect.ignore) + return + } + const params: RequestPermissionParams = decodePermissionParams(request.params).pipe( + Option.getOrElse(() => ({}) as RequestPermissionParams), + ) + const command = str(obj(params.toolCall?.rawInput)?.["command"]) ?? params.toolCall?.title ?? null + const approval: PendingApproval = { + id: request.id, + approvalId: null, + itemId: params.toolCall?.toolCallId ?? null, + command, + availableDecisions: params.options ?? [], + } + yield* SubscriptionRef.update(pendingApprovals, (list) => [...list, approval]) + }), + ), + Effect.forkScoped, + ) + + const runTurn = (text: string): Effect.Effect => + Effect.gen(function* () { + // ACP never echoes the user turn, so synthesize its message row from the + // prompt text (ordered before any of this turn's updates). + items.push({ kind: "message", role: "user", text }) + + // The `session/prompt` response is the turn-completion signal. The transport + // routes stdout through one sequential fiber, so every `session/update` for + // this turn was offered onto the notifications queue before the response + // resolved — fold after the response: take the whole buffered batch at once + // and flush the trailing assistant text. (No concurrent fold, so no window + // between a `Queue.take` and the fold in which an update could be dropped.) + const result = yield* transport + .request("session/prompt", { sessionId, prompt: [{ type: "text", text }] }) + .pipe(wrap("session/prompt failed")) + + const updates = yield* Queue.clear(transport.notifications) + for (const n of updates) foldNotification(n) + flushText() + + const stopReason = decodePromptResult(result).pipe( + Option.map((r) => r.stopReason), + Option.getOrUndefined, + ) + const rows = normalizeAcpSession(items.slice(), { + nativeSessionId: sessionId, + workspaceRoot: options.cwd, + sourcePath: `acp:${sessionId}`, + model: options.model ?? null, + }) + return { status: statusFromStopReason(stopReason), rows } + }) + + const answerApproval = (id: number | string, decision: unknown): Effect.Effect => + transport + .respond(id, { outcome: { outcome: "selected", optionId: decision } }) + .pipe( + Effect.andThen( + SubscriptionRef.update(pendingApprovals, (list) => list.filter((approval) => approval.id !== id)), + ), + wrap("answerApproval failed"), + ) + + return { threadId: sessionId, runTurn, pendingApprovals, answerApproval } + }) diff --git a/src/main/ingest/providers/cursor-acp/launch.ts b/src/main/ingest/providers/cursor-acp/launch.ts new file mode 100644 index 0000000..9c249aa --- /dev/null +++ b/src/main/ingest/providers/cursor-acp/launch.ts @@ -0,0 +1,39 @@ +import { Effect, type Scope } from "effect" +import type { AppServerCapability } from "../../../../shared/provider.js" +import { IngestStore } from "../../db/store.js" +import { type AppServerDriver, AppServerDriverError, type AppServerLaunchParams } from "../app-server-driver.js" +import { makeCursorAcpDriver } from "./driver.js" + +/** + * Launch a Cursor ACP driver from a provider's {@link AppServerCapability} and + * persist each completed turn's cumulative rows into the shared {@link IngestStore} + * — the cursor sibling of `launchCodexAppServerSession`. The rows and the store + * are the same ones the rollout-file provider writes, so a live-driven ACP session + * and a scraped cursor session are indistinguishable downstream. + */ +export const launchCursorAcpSession = ( + capability: AppServerCapability, + params: AppServerLaunchParams, +): Effect.Effect => + Effect.gen(function* () { + const store = yield* IngestStore + const driver = yield* makeCursorAcpDriver({ + command: capability.launchCmd, + args: [...capability.args], + cwd: params.cwd, + model: params.model, + env: params.env, + resumeThreadId: params.resumeThreadId, + }) + + const runTurn = (text: string) => + driver.runTurn(text).pipe( + Effect.tap((result) => + store + .replaceSession(result.rows) + .pipe(Effect.mapError((cause) => new AppServerDriverError({ message: "persist turn failed", cause }))), + ), + ) + + return { ...driver, runTurn } + }) diff --git a/src/main/ingest/providers/cursor-acp/normalize.ts b/src/main/ingest/providers/cursor-acp/normalize.ts new file mode 100644 index 0000000..7467796 --- /dev/null +++ b/src/main/ingest/providers/cursor-acp/normalize.ts @@ -0,0 +1,98 @@ +import type { DiagnosticRow, ExtractedRows } from "../../db/schema.js" +import type { Rec } from "../../extract/json.js" +import { SessionRowBuilder } from "../../extract/session-row-builder.js" +import { classifyTool } from "../../extract/tool-kind.js" + +/** + * The driver's accumulated, session-cumulative item list — the contract between + * the ACP driver's `session/update` fold and this normalizer. The driver has + * already done the ACP-specific work (buffering `agent_message_chunk` deltas + * into one message, upserting `tool_call` / `tool_call_update` by `toolCallId`, + * synthesizing the user message from the prompt text since ACP does not echo + * it), so each item here maps to exactly one row — the same shape + * `codex-appserver/normalize.ts` produces, just pre-folded. + */ +export type AcpItem = + | { readonly kind: "message"; readonly role: "user" | "assistant"; readonly text: string } + | { + readonly kind: "tool" + readonly toolCallId: string + /** ACP tool title, e.g. "`echo hello-acp`". */ + readonly title: string | null + /** ACP `kind`, e.g. "execute" for a shell command. */ + readonly toolKind: string | null + /** `rawInput.command` for an execute tool. */ + readonly command: string | null + /** `rawOutput.exitCode` for an execute tool. */ + readonly exitCode: number | null + /** Combined stdout+stderr for an execute tool; stringified `rawOutput` otherwise. */ + readonly output: string | null + /** Decoded `rawInput`, for file-hint derivation + the tool row `inputJson`. */ + readonly input: Rec | null + } + +export interface AcpNormalizeOptions { + /** The ACP session id — the native session id for the `cursor` provider. */ + readonly nativeSessionId: string + readonly workspaceRoot: string + /** A stable handle for this live session (`acp:`). */ + readonly sourcePath: string + readonly model?: string | null + readonly title?: string + readonly diagnostics?: ReadonlyArray> +} + +/** A completed exec's output, prefixed `[exit N]` on failure (mirrors codex-appserver). */ +const execOutput = (item: Extract): string => { + const body = item.output ?? "" + return item.exitCode != null && item.exitCode !== 0 ? `[exit ${item.exitCode}]\n${body}` : body +} + +/** + * Fold the ACP driver's accumulated {@link AcpItem}s into database-shaped rows — + * the same {@link ExtractedRows} the rollout-file provider produces, via the same + * {@link SessionRowBuilder}. The live-transport counterpart to + * `normalizeAppServerThread`: an execute tool becomes a first-class `shell` row + * carrying its command + exit code + output (no wrapper regex to strip), and any + * other tool becomes a generic tool row keyed by its ACP title/kind. No token + * usage is observed on the ACP wire, so `usageEvents` stays empty. + */ +export const normalizeAcpSession = ( + items: ReadonlyArray, + options: AcpNormalizeOptions, +): ExtractedRows => { + const b = new SessionRowBuilder("cursor", options.nativeSessionId) + const model = options.model ?? null + + for (const item of items) { + if (item.kind === "message") { + if (item.role === "user") { + b.message({ role: "user", text: item.text }) + } else { + b.message({ role: "assistant", text: item.text, model }) + } + continue + } + + // An execute tool is the shell case (structured command/exit/output); every + // other tool is a generic row named by its ACP title/kind. + const isExec = item.toolKind === "execute" || item.command != null + const name = isExec ? "Shell" : (item.title ?? item.toolKind ?? "tool") + const row = b.tool({ + name, + kind: classifyTool("cursor", name), + nativeToolId: item.toolCallId, + inputJson: item.input ? JSON.stringify(item.input) : JSON.stringify({ command: item.command }), + }) + row.outputText = execOutput(item) + b.hint(name, item.input ?? { command: item.command ?? undefined }, row.messageId, row.id) + } + + return b.finish({ + nativeSessionId: options.nativeSessionId, + workspaceRoot: options.workspaceRoot, + sourcePath: options.sourcePath, + title: options.title, + diagnostics: options.diagnostics, + }) +} diff --git a/src/main/ingest/providers/cursor-acp/protocol.ts b/src/main/ingest/providers/cursor-acp/protocol.ts new file mode 100644 index 0000000..f4806f4 --- /dev/null +++ b/src/main/ingest/providers/cursor-acp/protocol.ts @@ -0,0 +1,100 @@ +import { Schema } from "effect" + +// --- Cursor ACP (Agent Client Protocol) wire shapes ------------------------ +// `cursor-agent acp` speaks newline-delimited JSON-RPC 2.0 on stdio. A turn is a +// `session/prompt` request whose response (`{ stopReason }`) is the completion +// signal; while it runs, `session/update` notifications stream the transcript. +// Every variant is decoded the same way the rollout-file providers decode their +// records: `decode*Option`, so an unknown/malformed shape decodes to None and is +// skipped rather than throwing. `NonEmptyString` treats "" as absent. +// +// Only the subset a normal coding turn produces is mirrored here; `session_info_update` +// / `available_commands_update` (and the `user_message_chunk` replayed by +// `session/load`) intentionally have no variant, so they decode to None and are +// ignored. Shapes verified live against cursor-agent 2026.07.01. +const NeStr = Schema.NonEmptyString + +/** The `session/new` result: `sessionId` is the native session id the turn keys off. */ +export const SessionNewResult = Schema.Struct({ sessionId: NeStr }) +export const decodeSessionNew = Schema.decodeUnknownOption(SessionNewResult) + +/** The `session/prompt` result — its `stopReason` maps to the turn status. */ +export const PromptResult = Schema.Struct({ stopReason: Schema.optional(Schema.String) }) +export const decodePromptResult = Schema.decodeUnknownOption(PromptResult) + +// --- session/update variants ----------------------------------------------- + +/** An assistant text delta; `content.text` is concatenated into the pending buffer. */ +export const AgentMessageChunk = Schema.Struct({ + sessionUpdate: Schema.Literal("agent_message_chunk"), + content: Schema.optional( + Schema.Struct({ type: Schema.optional(Schema.String), text: Schema.optional(Schema.String) }), + ), +}) + +/** + * A tool call starts. `kind` is "execute" for a shell command; `rawInput` is the + * tool's args (`{ command }` for execute, other shapes otherwise) — kept + * `Unknown` and narrowed at the fold, mirroring how the codex driver keeps wire + * `item`s opaque until the normalizer. + */ +export const ToolCall = Schema.Struct({ + sessionUpdate: Schema.Literal("tool_call"), + toolCallId: NeStr, + title: Schema.optional(Schema.String), + kind: Schema.optional(Schema.String), + status: Schema.optional(Schema.String), + rawInput: Schema.optional(Schema.Unknown), +}) + +/** + * A later status/result for a tool call (merged onto the existing item by + * `toolCallId`). On completion `rawOutput` is `{ exitCode, stdout, stderr }` for + * an execute tool. + */ +export const ToolCallUpdate = Schema.Struct({ + sessionUpdate: Schema.Literal("tool_call_update"), + toolCallId: NeStr, + status: Schema.optional(Schema.String), + title: Schema.optional(Schema.String), + kind: Schema.optional(Schema.String), + rawInput: Schema.optional(Schema.Unknown), + rawOutput: Schema.optional(Schema.Unknown), +}) + +export const AcpSessionUpdate = Schema.Union([AgentMessageChunk, ToolCall, ToolCallUpdate]) +export type AcpSessionUpdate = typeof AcpSessionUpdate.Type + +/** The `session/update` notification params: `{ sessionId, update }`. */ +export const SessionUpdateParams = Schema.Struct({ update: AcpSessionUpdate }) +export const decodeSessionUpdate = Schema.decodeUnknownOption(SessionUpdateParams) + +// --- session/request_permission -------------------------------------------- + +/** One offered decision — the UI renders `name` and answers with `optionId`. */ +export const PermissionOption = Schema.Struct({ + optionId: NeStr, + name: Schema.optional(Schema.String), + kind: Schema.optional(Schema.String), +}) +export type PermissionOption = typeof PermissionOption.Type + +const PermissionToolCall = Schema.Struct({ + toolCallId: Schema.optional(Schema.String), + title: Schema.optional(Schema.String), + kind: Schema.optional(Schema.String), + status: Schema.optional(Schema.String), + rawInput: Schema.optional(Schema.Unknown), +}) + +/** + * The params of a `session/request_permission` server→client request. `toolCall` + * links to the streamed tool item; `options` is the server-supplied answer set + * the UI must offer verbatim (rendered by `name`, answered by `optionId`). + */ +export const RequestPermissionParams = Schema.Struct({ + toolCall: Schema.optional(PermissionToolCall), + options: Schema.optional(Schema.Array(PermissionOption)), +}) +export type RequestPermissionParams = typeof RequestPermissionParams.Type +export const decodePermissionParams = Schema.decodeUnknownOption(RequestPermissionParams) diff --git a/src/main/services/ChatMessageService.ts b/src/main/services/ChatMessageService.ts index 07486e2..d69c702 100644 --- a/src/main/services/ChatMessageService.ts +++ b/src/main/services/ChatMessageService.ts @@ -457,7 +457,7 @@ export const ChatMessageServiceLive = Layer.effect( const delivery = yield* router .submit({ instanceId: req.targetSessionId, text }) .pipe( - Effect.catchTag("CodexDriverError", (e) => + Effect.catchTag("AppServerDriverError", (e) => Effect.logWarning(`rpc turn failed (${req.targetSessionId}): ${e.message}`).pipe( Effect.as({ accepted: false as const }), ), diff --git a/src/main/services/CodexDriverRegistry.ts b/src/main/services/CodexDriverRegistry.ts index fea1392..c7a7d37 100644 --- a/src/main/services/CodexDriverRegistry.ts +++ b/src/main/services/CodexDriverRegistry.ts @@ -1,9 +1,9 @@ import { Context, Effect, Layer, type Scope, Stream, SubscriptionRef } from "effect" import type { - CodexAppServerDriver, - CodexDriverError, + AppServerDriver, + AppServerDriverError, PendingApproval, -} from "../ingest/providers/codex-appserver/driver.js" +} from "../ingest/providers/app-server-driver.js" /** The live approvals awaiting an answer for one session. */ export interface SessionApprovals { @@ -30,14 +30,14 @@ export class CodexDriverRegistry extends Context.Service< readonly register: (params: { readonly chatId: string readonly targetSessionId: string - readonly driver: CodexAppServerDriver + readonly driver: AppServerDriver }) => Effect.Effect /** Route an approval answer (by JSON-RPC request id) to the session's driver. */ readonly answerApproval: ( targetSessionId: string, requestId: number | string, decision: unknown, - ) => Effect.Effect + ) => Effect.Effect /** Sessions with outstanding approvals (empty sessions omitted). */ readonly pending: Effect.Effect> /** Reactive view of `pending`: the current aggregate, then every change. */ @@ -48,7 +48,7 @@ export class CodexDriverRegistry extends Context.Service< export const CodexDriverRegistryLive = Layer.effect( CodexDriverRegistry, Effect.gen(function* () { - const drivers = new Map() + const drivers = new Map() const current = new Map() const state = yield* SubscriptionRef.make>([]) @@ -63,7 +63,7 @@ export const CodexDriverRegistryLive = Layer.effect( const register = (params: { readonly chatId: string readonly targetSessionId: string - readonly driver: CodexAppServerDriver + readonly driver: AppServerDriver }) => Effect.gen(function* () { const { chatId, targetSessionId, driver } = params diff --git a/src/main/services/ProviderRegistry.ts b/src/main/services/ProviderRegistry.ts index b220091..548ca5d 100644 --- a/src/main/services/ProviderRegistry.ts +++ b/src/main/services/ProviderRegistry.ts @@ -66,6 +66,10 @@ const providers: ReadonlyArray = [ { match: "Press any key to log in", key: "\r" }, ], }, + // Drive cursor-agent directly over its ACP (Agent Client Protocol) JSON-RPC + // dialect (see ingest/providers/cursor-acp/driver.ts) — additive to the + // scraper + TUI paths above, the cursor sibling of codex's appServer. + appServer: { launchCmd: "cursor-agent", args: ["acp"], protocol: "acp" }, }, { kind: "pi", diff --git a/src/main/services/RpcSessionManager.ts b/src/main/services/RpcSessionManager.ts index f83209e..904e1a0 100644 --- a/src/main/services/RpcSessionManager.ts +++ b/src/main/services/RpcSessionManager.ts @@ -3,12 +3,11 @@ import * as Semaphore from "effect/Semaphore" import type { ChatId, TargetId } from "../../shared/ids.js" import type { TargetOrigin, TargetSession } from "../../shared/instance.js" import type { ExtractedRows } from "../ingest/db/schema.js" -import type { - CodexAppServerDriver, - CodexDriverError, - CodexDriverOptions, -} from "../ingest/providers/codex-appserver/driver.js" +import type { AppServerCapability } from "../../shared/provider.js" +import type { AppServerDriver, AppServerDriverError } from "../ingest/providers/app-server-driver.js" +import type { CodexDriverOptions } from "../ingest/providers/codex-appserver/driver.js" import { launchCodexAppServerSession } from "../ingest/providers/codex-appserver/launch.js" +import { launchCursorAcpSession } from "../ingest/providers/cursor-acp/launch.js" import { IngestStore } from "../ingest/db/store.js" import { CodexDriverRegistry } from "./CodexDriverRegistry.js" @@ -28,6 +27,8 @@ export interface RpcLaunchRequest { readonly cwd: string readonly command: string readonly args: ReadonlyArray + /** Which dialect the launch command speaks — picks the driver factory. Defaults to codex. */ + readonly protocol?: AppServerCapability["protocol"] readonly model?: string readonly sandbox?: CodexDriverOptions["sandbox"] readonly approvalPolicy?: CodexDriverOptions["approvalPolicy"] @@ -56,7 +57,7 @@ export class RpcSessionManager extends Context.Service< * turns, register its approvals. Returns the `TargetSession` with its * `nativeSessionId` bound to the driver's thread id — the caller persists that * so the timeline projection can resolve the target by (provider, native id). */ - readonly launch: (req: RpcLaunchRequest) => Effect.Effect + readonly launch: (req: RpcLaunchRequest) => Effect.Effect /** Run a user turn against a launched session. `accepted:false` if unknown; * `rows` is the session's cumulative rows for the caller to project. */ readonly submit: (req: { @@ -64,7 +65,7 @@ export class RpcSessionManager extends Context.Service< readonly text: string }) => Effect.Effect< { readonly accepted: boolean; readonly status?: string; readonly rows?: ExtractedRows }, - CodexDriverError + AppServerDriverError > /** Tear a session down: closes its scope (kills driver, deregisters approvals). */ readonly stop: (targetSessionId: string) => Effect.Effect<{ readonly stopped: boolean }> @@ -84,7 +85,7 @@ export class RpcSessionManager extends Context.Service< >()("arcwork/RpcSessionManager") {} interface LiveRpcSession { - readonly driver: CodexAppServerDriver + readonly driver: AppServerDriver readonly scope: Scope.Closeable /** Serializes turns for this session: the driver folds one turn at a time * (unkeyed turn outcomes), so concurrent submits would misattribute completions. */ @@ -120,7 +121,7 @@ export const RpcSessionManagerLive = Layer.effect( return next }) - const launch = (req: RpcLaunchRequest): Effect.Effect => + const launch = (req: RpcLaunchRequest): Effect.Effect => launchLock.withPermits(1)( Effect.gen(function* () { const current = yield* SubscriptionRef.get(store) @@ -129,9 +130,14 @@ export const RpcSessionManagerLive = Layer.effect( // Child of the layer scope: closes on `stop` or on app quit. const scope = yield* Scope.fork(parentScope) + // Pick the driver factory by dialect. Both speak the newline-delimited + // JSON-RPC transport and return the same {@link AppServerDriver}, so only + // the handshake/fold differs — the manager below is dialect-agnostic. + const launchSession = + req.protocol === "acp" ? launchCursorAcpSession : launchCodexAppServerSession const build = Effect.gen(function* () { - const driver = yield* launchCodexAppServerSession( - { launchCmd: req.command, args: req.args }, + const driver = yield* launchSession( + { launchCmd: req.command, args: req.args, protocol: req.protocol }, { cwd: req.cwd, model: req.model, diff --git a/src/main/services/SessionRuntimeRouter.ts b/src/main/services/SessionRuntimeRouter.ts index ea3338f..48c9be9 100644 --- a/src/main/services/SessionRuntimeRouter.ts +++ b/src/main/services/SessionRuntimeRouter.ts @@ -4,7 +4,7 @@ import { newArcId } from "../../shared/ids.js" import type { TargetSession } from "../../shared/instance.js" import { ArcStore } from "../db/store.js" import type { ExtractedRows } from "../ingest/db/schema.js" -import type { CodexDriverError } from "../ingest/providers/codex-appserver/driver.js" +import type { AppServerDriverError } from "../ingest/providers/app-server-driver.js" import { type ArcRequestError, arcRequestError } from "../errors.js" import { ChatService } from "./ChatService.js" import { ProviderRegistry } from "./ProviderRegistry.js" @@ -38,16 +38,16 @@ export class SessionRuntimeRouter extends Context.Service< { readonly launch: ( req: LaunchRequest, - ) => Effect.Effect + ) => Effect.Effect /** Resume a session, into `pty` (default) or `rpc` (rejoin the app-server * thread by its persisted native id) per `req.runtime`. */ readonly resume: ( req: ResumeRequest, - ) => Effect.Effect + ) => Effect.Effect /** Route a submit; `rows` is present for an rpc turn (caller projects it). */ readonly submit: ( req: SubmitRequest, - ) => Effect.Effect<{ readonly accepted: boolean; readonly rows?: ExtractedRows }, CodexDriverError> + ) => Effect.Effect<{ readonly accepted: boolean; readonly rows?: ExtractedRows }, AppServerDriverError> readonly stop: (req: StopRequest) => Effect.Effect<{ readonly stopped: boolean }> /** Whether an rpc runtime owns this session id (used to skip PTY-only checks). */ readonly ownsRpc: (targetSessionId: string) => Effect.Effect @@ -102,6 +102,7 @@ export const SessionRuntimeRouterLive = Layer.effect( cwd, command: spec.appServer.launchCmd, args: spec.appServer.args, + protocol: spec.appServer.protocol, sandbox: "workspace-write", approvalPolicy: "on-request", }) @@ -162,6 +163,7 @@ export const SessionRuntimeRouterLive = Layer.effect( cwd: row.cwd, command: spec.appServer.launchCmd, args: spec.appServer.args, + protocol: spec.appServer.protocol, sandbox: "workspace-write", approvalPolicy: "on-request", resumeThreadId: row.nativeSessionId, diff --git a/src/main/services/codex-approval-view.ts b/src/main/services/codex-approval-view.ts index 23e453e..a892d9e 100644 --- a/src/main/services/codex-approval-view.ts +++ b/src/main/services/codex-approval-view.ts @@ -1,16 +1,24 @@ import type { AppServerApproval, AppServerApprovalDecision } from "../../shared/codex-approval.js" -import { obj } from "../ingest/extract/json.js" +import { obj, str } from "../ingest/extract/json.js" import type { SessionApprovals } from "./CodexDriverRegistry.js" /** - * Normalize one raw server decision into a renderer button. A string decision - * (`"accept"`) labels itself; an object decision - * (`{ acceptWithExecpolicyAmendment: {…} }`) labels by its key. `payload` is the - * decision re-encoded as JSON so it round-trips to the driver unchanged. + * Normalize one raw server decision into a renderer button, spanning both + * dialects' answer models: + * - **ACP** — a `{ optionId, name, kind }` option: label by `name`, and encode + * `payload` as just the `optionId` string, so answering sends that id back + * (the ACP driver wraps it as `{ outcome: { outcome: "selected", optionId } }`). + * - **codex** — an opaque decision: a string (`"accept"`) labels itself; a + * rule-carrying object (`{ acceptWithExecpolicyAmendment: {…} }`) labels by + * its key. `payload` is the whole decision JSON, echoed back verbatim. */ const decisionView = (decision: unknown): AppServerApprovalDecision => { - const label = - typeof decision === "string" ? decision : (Object.keys(obj(decision) ?? {})[0] ?? "decision") + const record = obj(decision) + const optionId = str(record?.["optionId"]) + if (optionId) { + return { label: str(record?.["name"]) ?? optionId, payload: JSON.stringify(optionId) } + } + const label = typeof decision === "string" ? decision : (Object.keys(record ?? {})[0] ?? "decision") return { label, payload: JSON.stringify(decision) } } diff --git a/src/shared/provider.ts b/src/shared/provider.ts index e32eddd..624f498 100644 --- a/src/shared/provider.ts +++ b/src/shared/provider.ts @@ -89,6 +89,13 @@ export type InteractiveCapability = typeof InteractiveCapability.Type export const AppServerCapability = Schema.Struct({ launchCmd: Schema.String, args: Schema.Array(Schema.String), + /** + * Which JSON-RPC dialect `launchCmd` speaks — the driver factory the + * `RpcSessionManager` picks. `codex-app-server` (default when absent) is + * codex's thread/turn/item protocol; `acp` is Cursor's Agent Client Protocol + * (session/prompt/update). Both ride the same newline-delimited transport. + */ + protocol: Schema.optional(Schema.Literals(["codex-app-server", "acp"])), }) export type AppServerCapability = typeof AppServerCapability.Type diff --git a/tests/codex-approval-view.test.ts b/tests/codex-approval-view.test.ts index e8cf726..6029480 100644 --- a/tests/codex-approval-view.test.ts +++ b/tests/codex-approval-view.test.ts @@ -45,6 +45,33 @@ describe("codex approval projection", () => { expect(second.command).toBeNull() }) + it("labels ACP option decisions by name and answers with the optionId", () => { + const acp: ReadonlyArray = [ + { + chatId: "chat_a", + targetSessionId: "target_a", + approvals: [ + { + id: 0, + approvalId: null, + itemId: "tool_1", + command: "echo hi", + availableDecisions: [ + { optionId: "allow-once", name: "Allow once", kind: "allow_once" }, + { optionId: "reject-once", name: "Reject", kind: "reject_once" }, + ], + }, + ], + }, + ] + const view = projectApprovals(acp) + const decisions = view[0]!.decisions + expect(decisions.map((d) => d.label)).toEqual(["Allow once", "Reject"]) + // The payload is just the optionId, so answering sends the id (not the whole + // option object) — what the ACP driver's `{ outcome: { optionId } }` needs. + expect(parseDecisionPayload(decisions[0]!.payload)).toBe("allow-once") + }) + it("round-trips a decision payload back to its raw value", () => { const view = projectApprovals(sessions) const amend = view[0]!.decisions[1]! diff --git a/tests/codex-appserver-driver.test.ts b/tests/codex-appserver-driver.test.ts index e5e25b8..8aaf095 100644 --- a/tests/codex-appserver-driver.test.ts +++ b/tests/codex-appserver-driver.test.ts @@ -1,9 +1,7 @@ import { Effect, type Scope, Stream, SubscriptionRef } from "effect" import { describe, expect, it } from "vitest" -import { - CodexDriverError, - makeCodexAppServerDriver, -} from "../src/main/ingest/providers/codex-appserver/driver.js" +import { AppServerDriverError } from "../src/main/ingest/providers/app-server-driver.js" +import { makeCodexAppServerDriver } from "../src/main/ingest/providers/codex-appserver/driver.js" // A scripted thread/turn peer (stands in for `codex app-server`). It completes // the handshake, then on `turn/start` streams a user item and asks for approval @@ -152,7 +150,7 @@ describe("codex app-server driver", () => { // The peer acks turn/start then exits before turn/completed. runTurn must // resolve to a failure (via the aborted signal), not block on the queue. const failure = yield* Effect.flip(driver.runTurn("hi")) - expect(failure).toBeInstanceOf(CodexDriverError) + expect(failure).toBeInstanceOf(AppServerDriverError) }), ), 15000, diff --git a/tests/codex-appserver-transport.test.ts b/tests/codex-appserver-transport.test.ts index 5d93c98..e66e7e5 100644 --- a/tests/codex-appserver-transport.test.ts +++ b/tests/codex-appserver-transport.test.ts @@ -76,7 +76,7 @@ describe("codex app-server transport", () => { expect(failure).toBeInstanceOf(AppServerTransportError) // the peer's server→client request (buffered during initialize) → answer it - yield* t.serverRequests.pipe( + yield* Stream.fromQueue(t.serverRequests).pipe( Stream.take(1), Stream.runForEach((req) => { expect(req.id).toBe(777) @@ -87,7 +87,7 @@ describe("codex app-server transport", () => { ) // turn/started (from initialize) then approval/echo (proof respond landed) - const notifs = yield* t.notifications.pipe(Stream.take(2), Stream.runCollect) + const notifs = yield* Stream.fromQueue(t.notifications).pipe(Stream.take(2), Stream.runCollect) expect(notifs.map((n) => n.method)).toEqual(["turn/started", "approval/echo"]) expect(notifs[1]?.params).toMatchObject({ decision: "accept" }) }), diff --git a/tests/cursor-acp-driver.test.ts b/tests/cursor-acp-driver.test.ts new file mode 100644 index 0000000..8be4bd5 --- /dev/null +++ b/tests/cursor-acp-driver.test.ts @@ -0,0 +1,171 @@ +import { Effect, type Scope, Stream, SubscriptionRef } from "effect" +import { describe, expect, it } from "vitest" +import { AppServerDriverError } from "../src/main/ingest/providers/app-server-driver.js" +import { makeCursorAcpDriver } from "../src/main/ingest/providers/cursor-acp/driver.js" + +// A scripted ACP peer (stands in for `cursor-agent acp`, no auth / network). It +// handshakes, then on `session/prompt` streams an ignored info/commands update, a +// pending execute `tool_call`, and a `session/request_permission` (id 0 — numeric +// zero, the real cursor id, to prove routing). Once the client answers request 0 +// it completes the tool (rawOutput), emits the assistant reply, and resolves the +// prompt with `{ stopReason: "end_turn" }` — that response is the turn signal. +const PEER = ` +const readline = require('node:readline') +const rl = readline.createInterface({ input: process.stdin }) +const send = (o) => process.stdout.write(JSON.stringify(o) + '\\n') +const upd = (update) => send({ method: 'session/update', params: { sessionId: 'sess_test', update } }) +let promptId = null +rl.on('line', (line) => { + if (!line.trim()) return + const m = JSON.parse(line) + if (m.method === 'initialize') send({ id: m.id, result: { protocolVersion: 1, agentCapabilities: { loadSession: true } } }) + else if (m.method === 'session/new') send({ id: m.id, result: { sessionId: 'sess_test' } }) + else if (m.method === 'session/prompt') { + promptId = m.id + upd({ sessionUpdate: 'available_commands_update', availableCommands: [] }) + upd({ sessionUpdate: 'session_info_update', title: 'Echo' }) + upd({ sessionUpdate: 'tool_call', toolCallId: 'tool_1', title: '\\\`echo hi\\\`', kind: 'execute', status: 'pending', rawInput: { command: 'echo hi' } }) + upd({ sessionUpdate: 'tool_call_update', toolCallId: 'tool_1', status: 'in_progress' }) + send({ id: 0, method: 'session/request_permission', params: { sessionId: 'sess_test', toolCall: { toolCallId: 'tool_1', title: '\\\`echo hi\\\`', kind: 'execute', status: 'pending' }, options: [{ optionId: 'allow-once', name: 'Allow once', kind: 'allow_once' }, { optionId: 'reject-once', name: 'Reject', kind: 'reject_once' }] } }) + } else if (m.method == null && m.id === 0) { + upd({ sessionUpdate: 'tool_call_update', toolCallId: 'tool_1', status: 'completed', rawOutput: { exitCode: 0, stdout: 'hi\\n', stderr: '' } }) + upd({ sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'done' } }) + send({ id: promptId, result: { stopReason: 'end_turn' } }) + } +}) +` + +// Resume peer: `session/load` replays the prior transcript as session/update +// notifications BEFORE its reply (the observed cursor behavior), then a new turn +// streams only 'new'. Proves the driver drains replay and accumulates only the +// new turn. +const RESUME_PEER = ` +const readline = require('node:readline') +const rl = readline.createInterface({ input: process.stdin }) +const send = (o) => process.stdout.write(JSON.stringify(o) + '\\n') +const upd = (sid, update) => send({ method: 'session/update', params: { sessionId: sid, update } }) +let promptId = null +rl.on('line', (line) => { + if (!line.trim()) return + const m = JSON.parse(line) + if (m.method === 'initialize') send({ id: m.id, result: { agentCapabilities: { loadSession: true } } }) + else if (m.method === 'session/load') { + upd(m.params.sessionId, { sessionUpdate: 'user_message_chunk', content: { type: 'text', text: 'OLD PROMPT' } }) + upd(m.params.sessionId, { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'OLD REPLY' } }) + send({ id: m.id, result: { modes: {}, models: {} } }) + } else if (m.method === 'session/prompt') { + promptId = m.id + upd(m.params.sessionId, { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'new' } }) + send({ id: promptId, result: { stopReason: 'end_turn' } }) + } +}) +` + +// Handshakes, then exits the process on `session/prompt` before responding — the +// mid-turn crash that must fail runTurn (via the failed prompt request), not hang. +const MIDTURN_EXIT_PEER = ` +const readline = require('node:readline') +const rl = readline.createInterface({ input: process.stdin }) +const send = (o) => process.stdout.write(JSON.stringify(o) + '\\n') +rl.on('line', (line) => { + if (!line.trim()) return + const m = JSON.parse(line) + if (m.method === 'initialize') send({ id: m.id, result: {} }) + else if (m.method === 'session/new') send({ id: m.id, result: { sessionId: 'sess_x' } }) + else if (m.method === 'session/prompt') process.exit(0) +}) +` + +const run = (program: Effect.Effect): Promise => + Effect.runPromise(Effect.scoped(program)) + +describe("cursor ACP driver", () => { + it( + "handshakes, records + answers a permission (id 0), and folds the turn into rows", + () => + run( + Effect.gen(function* () { + const driver = yield* makeCursorAcpDriver({ + cwd: process.cwd(), + model: "cursor-default", + command: process.execPath, + args: ["-e", PEER], + }) + expect(driver.threadId).toBe("sess_test") + + // Answer the permission with the first option's id (as the renderer does). + yield* SubscriptionRef.changes(driver.pendingApprovals).pipe( + Stream.filter((list) => list.length > 0), + Stream.take(1), + Stream.runForEach((list) => { + const approval = list[0]! + expect(approval.id).toBe(0) // numeric zero routed correctly + expect(approval.itemId).toBe("tool_1") + expect(approval.command).toBe("`echo hi`") + // Options are surfaced verbatim as { optionId, name, kind } objects. + expect((approval.availableDecisions[0] as { optionId: string }).optionId).toBe("allow-once") + return driver.answerApproval(approval.id, "allow-once") + }), + Effect.forkScoped, + ) + + const result = yield* driver.runTurn("Run echo hi then say done.") + expect(result.status).toBe("completed") + // The user turn is synthesized from the prompt text (ACP never echoes it). + expect(result.rows.messages.map((m) => m.role)).toEqual(["user", "assistant"]) + expect(result.rows.messages[0]?.text).toContain("echo hi") + expect(result.rows.messages.find((m) => m.role === "assistant")?.text).toBe("done") + + const tool = result.rows.toolCalls.find((t) => t.nativeToolId === "tool_1") + expect(tool?.name).toBe("Shell") + expect(tool?.outputText).toBe("hi\n") + expect(result.rows.session.nativeSessionId).toBe("sess_test") + + // Answering cleared the pending signal. + expect(yield* SubscriptionRef.get(driver.pendingApprovals)).toHaveLength(0) + }), + ), + 15000, + ) + + it( + "rejoins by id via session/load and drains replay (new turn only)", + () => + run( + Effect.gen(function* () { + const driver = yield* makeCursorAcpDriver({ + cwd: process.cwd(), + command: process.execPath, + args: ["-e", RESUME_PEER], + resumeThreadId: "sess_old", + }) + expect(driver.threadId).toBe("sess_old") + + const result = yield* driver.runTurn("hi") + // The replayed prior transcript ('OLD PROMPT' / 'OLD REPLY') is discarded; + // only the synthesized user turn + the new assistant reply remain. + expect(result.rows.messages.map((m) => m.role)).toEqual(["user", "assistant"]) + expect(result.rows.messages.find((m) => m.role === "assistant")?.text).toBe("new") + expect(result.rows.messages.some((m) => (m.text ?? "").includes("OLD"))).toBe(false) + }), + ), + 15000, + ) + + it( + "fails the turn (does not hang) when the process exits mid-turn", + () => + run( + Effect.gen(function* () { + const driver = yield* makeCursorAcpDriver({ + cwd: process.cwd(), + command: process.execPath, + args: ["-e", MIDTURN_EXIT_PEER], + }) + const failure = yield* Effect.flip(driver.runTurn("hi")) + expect(failure).toBeInstanceOf(AppServerDriverError) + }), + ), + 15000, + ) +}) diff --git a/tests/cursor-acp-live.test.ts b/tests/cursor-acp-live.test.ts new file mode 100644 index 0000000..8a057c3 --- /dev/null +++ b/tests/cursor-acp-live.test.ts @@ -0,0 +1,92 @@ +import { mkdtempSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { Effect, Layer, ManagedRuntime, type Scope, Stream, SubscriptionRef } from "effect" +import { describe, expect, it } from "vitest" +import { IngestStore, IngestStoreLive } from "../src/main/ingest/db/store.js" +import { sqliteLayer } from "../src/main/ingest/db/sqlite.js" +import { launchCursorAcpSession } from "../src/main/ingest/providers/cursor-acp/launch.js" + +// Live end-to-end smoke against the real `cursor-agent acp` binary, through the +// full pipeline: launch → driver → normalizer → shared store. Runs one turn that +// executes a shell command, auto-answering the permission with `allow-once` via +// the driver's `answerApproval`. Opt-in (needs an authenticated cursor-agent), +// skipped unless CURSOR_LIVE=1. +// Run: `CURSOR_LIVE=1 pnpm exec vitest run tests/cursor-acp-live.test.ts`. +const run = (program: Effect.Effect): Promise => { + const runtime = ManagedRuntime.make(IngestStoreLive.pipe(Layer.provide(sqliteLayer(":memory:")))) + return runtime.runPromise(Effect.scoped(program)).finally(() => runtime.dispose()) +} + +describe("cursor ACP (live, full pipeline)", () => { + it.runIf(process.env.CURSOR_LIVE === "1")( + "drives a real turn that runs a shell command and persists rows", + () => + run( + Effect.gen(function* () { + const cwd = mkdtempSync(join(tmpdir(), "cursor-acp-live-")) + const driver = yield* launchCursorAcpSession( + { launchCmd: "cursor-agent", args: ["acp"], protocol: "acp" }, + { cwd }, + ) + + // Auto-answer any permission with its `allow_once` option (else the first). + yield* SubscriptionRef.changes(driver.pendingApprovals).pipe( + Stream.filter((list) => list.length > 0), + Stream.runForEach((list) => + Effect.forEach(list, (approval) => { + const options = approval.availableDecisions as ReadonlyArray<{ + readonly optionId: string + readonly kind?: string + }> + const chosen = options.find((o) => o.kind === "allow_once") ?? options[0] + return chosen ? driver.answerApproval(approval.id, chosen.optionId) : Effect.void + }), + ), + Effect.forkScoped, + ) + + const turn = yield* driver.runTurn( + "Run exactly `echo hello-acp` in the shell using a command, then reply with the single word DONE.", + ) + + const commandTool = turn.rows.toolCalls.find((t) => + (t.inputJson ?? "").includes("echo hello-acp"), + ) + const assistant = turn.rows.messages.filter((m) => m.role === "assistant") + + // Print the rows for the report. + writeFileSync( + join(tmpdir(), "cursor-acp-live-rows.json"), + JSON.stringify( + { + status: turn.status, + nativeSessionId: turn.rows.session.nativeSessionId, + provider: turn.rows.session.provider, + messages: turn.rows.messages.map((m) => ({ role: m.role, text: m.text })), + toolCalls: turn.rows.toolCalls.map((t) => ({ + name: t.name, + input: t.inputJson, + output: t.outputText, + })), + }, + null, + 2, + ), + ) + + expect(turn.status).toBe("completed") + expect(commandTool).toBeDefined() + expect(commandTool?.name).toBe("Shell") + expect(commandTool?.outputText ?? "").toContain("hello-acp") + expect(assistant.length).toBeGreaterThanOrEqual(1) + + // The turn landed in the shared store, indistinguishable from a scrape. + const store = yield* IngestStore + const stored = yield* store.getSession(turn.rows.session.id) + expect(stored?.session.provider).toBe("cursor") + }), + ), + 90000, + ) +}) diff --git a/tests/cursor-acp-normalize.test.ts b/tests/cursor-acp-normalize.test.ts new file mode 100644 index 0000000..7f75b47 --- /dev/null +++ b/tests/cursor-acp-normalize.test.ts @@ -0,0 +1,80 @@ +import { describe, expect, it } from "vitest" +import { type AcpItem, normalizeAcpSession } from "../src/main/ingest/providers/cursor-acp/normalize.js" + +// The driver's pre-folded item list for one ACP turn: a synthesized user +// message, a completed execute tool (command + exit + output), and the flushed +// assistant reply. Mirrors what the `session/update` fold produces. +const items: ReadonlyArray = [ + { kind: "message", role: "user", text: "Run echo hello-acp then say DONE." }, + { + kind: "tool", + toolCallId: "tool_5f739001", + title: "`echo hello-acp`", + toolKind: "execute", + command: "echo hello-acp", + exitCode: 0, + output: "hello-acp\n", + input: { command: "echo hello-acp" }, + }, + { kind: "message", role: "assistant", text: "DONE" }, +] + +const rows = normalizeAcpSession(items, { + nativeSessionId: "441efc01-session", + workspaceRoot: "/work", + sourcePath: "acp:441efc01-session", + model: "cursor-default", +}) + +describe("cursor ACP → ExtractedRows", () => { + it("maps the synthesized user + assistant items to messages", () => { + expect(rows.messages.map((m) => m.role)).toEqual(["user", "assistant"]) + expect(rows.messages[0]?.text).toContain("hello-acp") + const assistant = rows.messages.find((m) => m.role === "assistant") + expect(assistant?.text).toBe("DONE") + expect(assistant?.model).toBe("cursor-default") + }) + + it("maps an execute tool to a shell tool call with structured output", () => { + const tool = rows.toolCalls.find((t) => t.nativeToolId === "tool_5f739001") + expect(tool?.name).toBe("Shell") + expect(tool?.kind).toBe("shell") + expect(tool?.outputText).toBe("hello-acp\n") + expect(JSON.parse(tool!.inputJson!).command).toBe("echo hello-acp") + }) + + it("prefixes a nonzero exit code, mirroring the codex path", () => { + const failed = normalizeAcpSession( + [ + { + kind: "tool", + toolCallId: "t2", + title: "`false`", + toolKind: "execute", + command: "false", + exitCode: 2, + output: "boom\n", + input: { command: "false" }, + }, + ], + { nativeSessionId: "s", workspaceRoot: "/w", sourcePath: "acp:s" }, + ) + expect(failed.toolCalls[0]?.outputText).toBe("[exit 2]\nboom\n") + }) + + it("orders the tool call between the surrounding messages", () => { + const ordinals = new Map([...rows.messages, ...rows.toolCalls].map((r) => [r.id, r.ordinal] as const)) + const user = rows.messages.find((m) => m.role === "user")! + const tool = rows.toolCalls[0]! + const assistant = rows.messages.find((m) => m.role === "assistant")! + expect(ordinals.get(user.id)).toBeLessThan(ordinals.get(tool.id)!) + expect(ordinals.get(tool.id)).toBeLessThan(ordinals.get(assistant.id)!) + }) + + it("carries cursor provider + session identity through finish", () => { + expect(rows.session.provider).toBe("cursor") + expect(rows.session.nativeSessionId).toBe("441efc01-session") + expect(rows.session.sourcePath).toBe("acp:441efc01-session") + expect(rows.usageEvents).toHaveLength(0) + }) +})