feat(cursor): drive cursor-agent acp as a second app-server dialect

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<unknown>
+}
+
+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<string, string>
+  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<TurnResult, AppServerDriverError>
+  /** The current outstanding approvals; subscribe via `.changes` for the live signal. */
+  readonly pendingApprovals: SubscriptionRef.SubscriptionRef<ReadonlyArray<PendingApproval>>
+  /** Answer an approval with a server-offered decision (the driver shapes the dialect envelope). */
+  readonly answerApproval: (id: number | string, decision: unknown) => Effect.Effect<void, AppServerDriverError>
+}

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<unknown>
-}
-
-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<TurnResult, CodexDriverError>
-  /** The current outstanding approvals; subscribe via `.changes` for the live signal. */
-  readonly pendingApprovals: SubscriptionRef.SubscriptionRef<ReadonlyArray<PendingApproval>>
-  /** Answer an approval with a server-offered decision (pass it through verbatim). */
-  readonly answerApproval: (id: number | string, decision: unknown) => Effect.Effect<void, CodexDriverError>
-}
-
 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) =>
-  <A, R>(effect: Effect.Effect<A, AppServerTransportError, R>): Effect.Effect<A, CodexDriverError, R> =>
-    effect.pipe(Effect.mapError((cause) => new CodexDriverError({ message, cause })))
+  <A, R>(effect: Effect.Effect<A, AppServerTransportError, R>): Effect.Effect<A, AppServerDriverError, R> =>
+    effect.pipe(Effect.mapError((cause) => new AppServerDriverError({ message, cause })))
 
 export const makeCodexAppServerDriver = (
   options: CodexDriverOptions,
-): Effect.Effect<CodexAppServerDriver, CodexDriverError, Scope.Scope> =>
+): Effect.Effect<AppServerDriver, AppServerDriverError, Scope.Scope> =>
   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<unknown> = []
     const usage: Array<unknown> = []
-    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,15 +179,15 @@ export const makeCodexAppServerDriver = (
       Effect.forkScoped,
     )
 
-    const runTurn = (text: string): Effect.Effect<TurnResult, CodexDriverError> =>
+    const runTurn = (text: string): Effect.Effect<TurnResult, AppServerDriverError> =>
       Effect.gen(function* () {
         yield* transport
           .request("turn/start", { threadId, input: [{ type: "text", text }] })
           .pipe(wrap("turn/start failed"))
         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<void, CodexDriverError> =>
+    const answerApproval = (id: number | string, decision: unknown): Effect.Effect<void, AppServerDriverError> =>
       transport
         .respond(id, { decision })
         .pipe(

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<string, string>
-  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<CodexAppServerDriver, CodexDriverError, Scope.Scope | IngestStore> =>
+  params: AppServerLaunchParams,
+): Effect.Effect<AppServerDriver, AppServerDriverError, Scope.Scope | IngestStore> =>
   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 }))),
         ),
       )
 

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,17 +45,35 @@ 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<unknown, AppServerTransportError>
   /** Fire-and-forget notification (no response expected). */
   readonly notify: (method: string, params?: unknown) => Effect.Effect<void, AppServerTransportError>
   /** Answer a server→client request (e.g. an approval decision). */
   readonly respond: (id: IdKey, result: unknown) => Effect.Effect<void, AppServerTransportError>
-  /** Server notifications, in arrival order. Ends when the process exits. */
-  readonly notifications: Stream.Stream<JsonRpcNotification>
-  /** Server→client requests, in arrival order. Ends when the process exits. */
-  readonly serverRequests: Stream.Stream<JsonRpcServerRequest>
+  /**
+   * 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<void, AppServerTransportError>
+  /**
+   * 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<JsonRpcNotification>
+  /** Server→client requests, in arrival order. Shut down on process exit. */
+  readonly serverRequests: Queue.Dequeue<JsonRpcServerRequest>
 }
 
 export interface AppServerTransportOptions {
@@ -191,7 +211,9 @@ export const makeAppServerTransport = (
     const writeLine = (payload: Rec): Effect.Effect<void, AppServerTransportError> =>
       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<void, AppServerTransportError> =>
       closed ? Effect.fail(closed) : writeLine({ id, result })
 
+    const respondError = (
+      id: IdKey,
+      error: JsonRpcErrorObject,
+    ): Effect.Effect<void, AppServerTransportError> =>
+      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
   })