From 4c48ce1d89cc62bf3fdcc5269a42bbdcd9a21eb1 Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sat, 4 Jul 2026 13:34:27 +1000 Subject: [PATCH 01/26] feat(ingest): drive codex via the app-server JSON-RPC transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a live-driver path for Codex built on `codex app-server` (JSON-RPC 2.0 over stdio) — the interface the Codex VS Code extension uses — as a third acquisition transport beside the rollout-file scraper and the PTY TUI launch. It lands in the same `ExtractedRows` + pending-input signal as the existing provider, never a parallel pipeline; codex protocol types stay inside these files. Three composable layers, all verified: - transport.ts — generic newline-delimited JSON-RPC client over a child process's stdio. Scoped spawn (killed on scope close); raw readline/exit/error callbacks only offer onto an inbound queue; one scoped fiber resolves id-correlated Deferreds and fans notifications / server-requests to Streams. On process death every pending request fails and both streams end. - protocol.ts — Effect Schema for the item variants a turn produces (userMessage, agentMessage, reasoning, commandExecution), token usage, and the driver-facing thread/start + approval payloads. Mirrors codex.ts's Schema.Union + decode*Option idiom (unknown shape -> skipped, never throws). - normalize.ts — folds a turn's item/completed payloads into the shared SessionRowBuilder / ExtractedRows. Structured commandExecution fields (command, exitCode, aggregatedOutput) drop the EXEC_WRAPPER telemetry-regex the rollout provider needs. - driver.ts — handshake (initialize -> thread/start), notification fold, and the approval state machine: each item/*/requestApproval becomes an addressed PendingApproval in a SubscriptionRef (with server-supplied availableDecisions), cleared on serverRequest/resolved — the deterministic replacement for the racy outputText-flash inference. Tests: hermetic node-peer coverage for the transport and the full driver flow (handshake, approval surface+answer, turn->rows), plus an env-gated live smoke (CODEX_LIVE=1) that drives the real codex binary end to end. --- .../providers/codex-appserver/driver.ts | 210 ++++++++++++++++++ .../providers/codex-appserver/normalize.ts | 127 +++++++++++ .../providers/codex-appserver/protocol.ts | 112 ++++++++++ .../providers/codex-appserver/transport.ts | 204 +++++++++++++++++ tests/codex-appserver-driver.test.ts | 84 +++++++ tests/codex-appserver-live.test.ts | 44 ++++ tests/codex-appserver-normalize.test.ts | 94 ++++++++ tests/codex-appserver-transport.test.ts | 70 ++++++ 8 files changed, 945 insertions(+) create mode 100644 src/main/ingest/providers/codex-appserver/driver.ts create mode 100644 src/main/ingest/providers/codex-appserver/normalize.ts create mode 100644 src/main/ingest/providers/codex-appserver/protocol.ts create mode 100644 src/main/ingest/providers/codex-appserver/transport.ts create mode 100644 tests/codex-appserver-driver.test.ts create mode 100644 tests/codex-appserver-live.test.ts create mode 100644 tests/codex-appserver-normalize.test.ts create mode 100644 tests/codex-appserver-transport.test.ts diff --git a/src/main/ingest/providers/codex-appserver/driver.ts b/src/main/ingest/providers/codex-appserver/driver.ts new file mode 100644 index 0000000..f8f8ed0 --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/driver.ts @@ -0,0 +1,210 @@ +import { Data, Effect, Option, Queue, type Scope, Stream, SubscriptionRef } from "effect" +import type { ExtractedRows } from "../../db/schema.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 + * `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 + * state machine so `item//requestApproval` server requests become a deterministic + * pending-input signal instead of the racy `outputText`-flash inference. + * + * 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. + */ +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 { + readonly id: number | string + /** 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 + readonly rows: ExtractedRows +} + +export interface CodexAppServerDriver { + readonly threadId: string + /** Send a user turn and resolve once the turn completes, with its 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 + readonly sandbox?: "read-only" | "workspace-write" | "danger-full-access" + readonly approvalPolicy?: "untrusted" | "on-failure" | "on-request" | "never" + /** Defaults to `codex`. */ + readonly command?: string + /** Defaults to `["app-server"]`. */ + readonly args?: ReadonlyArray + readonly env?: Record + readonly clientName?: string +} + +interface TurnOutcome { + readonly items: ReadonlyArray + readonly usage: ReadonlyArray + readonly status: string +} + +const wrap = + (message: string) => + (effect: Effect.Effect): Effect.Effect => + effect.pipe(Effect.mapError((cause) => new CodexDriverError({ message, cause }))) + +export const makeCodexAppServerDriver = ( + options: CodexDriverOptions, +): Effect.Effect => + Effect.gen(function* () { + const transport: AppServerTransport = yield* makeAppServerTransport({ + command: options.command ?? "codex", + args: options.args ?? ["app-server"], + cwd: options.cwd, + env: options.env, + }).pipe(wrap("failed to spawn codex app-server")) + + // Handshake: initialize → initialized → thread/start. + yield* transport + .request("initialize", { + clientInfo: { name: options.clientName ?? "arc", title: "Arc", version: "0.0.1" }, + }) + .pipe(wrap("initialize failed")) + yield* transport.notify("initialized", {}).pipe(wrap("initialized notify failed")) + const started = yield* transport + .request("thread/start", { + cwd: options.cwd, + model: options.model, + sandbox: options.sandbox, + approvalPolicy: options.approvalPolicy, + }) + .pipe(wrap("thread/start failed")) + const thread = decodeThreadStart(started) + if (Option.isNone(thread)) { + return yield* Effect.fail( + new CodexDriverError({ message: "thread/start returned no thread id", cause: started }), + ) + } + const threadId = thread.value.thread.id + + const pendingApprovals = yield* SubscriptionRef.make>([]) + const turnOutcomes = yield* Queue.make() + + // One sequential fiber folds the notification stream. `turn/started` opens a + // fresh accumulator; each `item/completed` and token-usage snapshot appends; + // `turn/completed` hands the accumulated turn off to `runTurn` via the queue. + // Single-consumer, so the mutable accumulators need no locking. + let items: Array = [] + let usage: Array = [] + yield* transport.notifications.pipe( + Stream.runForEach((notification) => + Effect.gen(function* () { + switch (notification.method) { + case "turn/started": { + items = [] + usage = [] + break + } + case "item/completed": { + const item = obj(notification.params)?.["item"] + if (item !== undefined) items.push(item) + break + } + case "thread/tokenUsage/updated": { + usage.push(notification.params) + break + } + case "turn/completed": { + const turn = obj(obj(notification.params)?.["turn"]) + const status = str(turn?.["status"]) ?? "unknown" + const outcome: TurnOutcome = { items, usage, status } + items = [] + usage = [] + yield* Queue.offer(turnOutcomes, outcome) + break + } + case "serverRequest/resolved": { + const requestId = obj(notification.params)?.["requestId"] + yield* SubscriptionRef.update(pendingApprovals, (list) => + list.filter((approval) => approval.id !== requestId), + ) + break + } + } + }), + ), + Effect.forkScoped, + ) + + // 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( + Stream.runForEach((request) => + Effect.gen(function* () { + if (!request.method.endsWith("requestApproval")) { + yield* transport.respond(request.id, {}).pipe(Effect.ignore) + return + } + const params: ApprovalRequestParams = decodeApprovalParams(request.params).pipe( + Option.getOrElse(() => ({}) as ApprovalRequestParams), + ) + const approval: PendingApproval = { + id: request.id, + itemId: params.itemId ?? null, + command: params.command ?? null, + availableDecisions: params.availableDecisions ?? [], + } + yield* SubscriptionRef.update(pendingApprovals, (list) => [...list, approval]) + }), + ), + Effect.forkScoped, + ) + + const runTurn = (text: string): Effect.Effect => + Effect.gen(function* () { + yield* transport + .request("turn/start", { threadId, input: [{ type: "text", text }] }) + .pipe(wrap("turn/start failed")) + const outcome = yield* Queue.take(turnOutcomes) + const rows = normalizeAppServerThread(outcome.items, outcome.usage, { + nativeSessionId: threadId, + workspaceRoot: options.cwd, + sourcePath: `appserver:${threadId}`, + model: options.model ?? null, + }) + return { status: outcome.status, rows } + }) + + const answerApproval = (id: number | string, decision: unknown): Effect.Effect => + transport + .respond(id, { decision }) + .pipe( + Effect.andThen( + SubscriptionRef.update(pendingApprovals, (list) => list.filter((approval) => approval.id !== id)), + ), + wrap("answerApproval failed"), + ) + + return { threadId, runTurn, pendingApprovals, answerApproval } + }) diff --git a/src/main/ingest/providers/codex-appserver/normalize.ts b/src/main/ingest/providers/codex-appserver/normalize.ts new file mode 100644 index 0000000..c3fee5f --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/normalize.ts @@ -0,0 +1,127 @@ +import { Option } from "effect" +import type { DiagnosticRow, ExtractedRows } from "../../db/schema.js" +import { obj, str } from "../../extract/json.js" +import { SessionRowBuilder } from "../../extract/session-row-builder.js" +import { classifyTool } from "../../extract/tool-kind.js" +import { type AppServerItem, decodeItem, decodeUsage } from "./protocol.js" + +export interface AppServerNormalizeOptions { + /** The app-server thread id — the native session id for the `codex` provider. */ + readonly nativeSessionId: string + readonly workspaceRoot: string + /** A stable handle for this live thread (e.g. `appserver:`). */ + readonly sourcePath: string + readonly model?: string | null + readonly title?: string + readonly createdAt?: string | null + readonly updatedAt?: string | null + readonly diagnostics?: ReadonlyArray> +} + +const finite = (value: number | undefined): number | null => + value === undefined || !Number.isFinite(value) ? null : value + +/** A completed exec's output, prefixed `[exit N]` on failure (mirrors codex.ts). */ +const execOutput = (item: Extract): string => { + const body = item.aggregatedOutput ?? "" + return item.exitCode != null && item.exitCode !== 0 ? `[exit ${item.exitCode}]\n${body}` : body +} + +/** Join reasoning summary/content parts (string parts or `{ text }` parts). */ +const reasoningText = (item: Extract): string | null => { + const parts = [...(item.summary ?? []), ...(item.content ?? [])] + const texts: Array = [] + for (const part of parts) { + const text = typeof part === "string" ? part : str(obj(part)?.["text"]) + if (text) texts.push(text) + } + const joined = texts.join("\n\n").trim() + return joined.length > 0 ? joined : null +} + +/** + * Fold a codex app-server thread's `item/completed` payloads (plus its + * `thread/tokenUsage/updated` snapshots) into database-shaped rows — the same + * `ExtractedRows` the rollout-file provider produces, via the same + * {@link SessionRowBuilder}. This is the live-transport counterpart to + * `normalizeCodexRecords`: the switch mirrors that provider's `switch (p.type)`, + * but tool results arrive structured (no `EXEC_WRAPPER` regex) and each exec is + * one completed item carrying its own command + exit code + output. + * + * `rawItems` are the items in `item/completed` order; `rawUsage` the usage + * notification params in arrival order. Both are decoded here (untrusted wire + * JSON), and an undecodable entry is skipped, never thrown. + */ +export const normalizeAppServerThread = ( + rawItems: ReadonlyArray, + rawUsage: ReadonlyArray, + options: AppServerNormalizeOptions, +): ExtractedRows => { + const b = new SessionRowBuilder("codex", options.nativeSessionId) + const model = options.model ?? null + + for (const raw of rawItems) { + const decoded = decodeItem(raw) + if (Option.isNone(decoded)) continue + const item = decoded.value + + switch (item.type) { + case "userMessage": { + const text = (item.content ?? []) + .filter((c) => c.type === "text") + .map((c) => c.text ?? "") + .join("") + .trim() + if (text) b.message({ role: "user", text, nativeMessageId: item.id ?? null }) + break + } + case "agentMessage": { + const text = item.text?.trim() + if (text) b.message({ role: "assistant", text: item.text, model, nativeMessageId: item.id ?? null }) + break + } + case "reasoning": { + const thinking = reasoningText(item) + if (thinking) b.message({ role: "assistant", thinking, model, nativeMessageId: item.id ?? null }) + break + } + case "commandExecution": { + const row = b.tool({ + name: "shell", + kind: classifyTool("codex", "shell"), + nativeToolId: item.id, + inputJson: JSON.stringify({ command: item.command ?? null, cwd: item.cwd ?? null }), + }) + row.outputText = execOutput(item) + b.hint("shell", { command: item.command }, row.messageId, row.id) + break + } + } + } + + for (const raw of rawUsage) { + const decoded = decodeUsage(raw) + if (Option.isNone(decoded)) continue + const usage = decoded.value.tokenUsage + const last = usage.last + const inputTokens = finite(last?.inputTokens) + b.usage({ + model, + contextUsedTokens: inputTokens, + contextWindowTokens: finite(usage.modelContextWindow), + inputTokens, + outputTokens: finite(last?.outputTokens), + rawJson: JSON.stringify(raw), + }) + } + + return b.finish({ + nativeSessionId: options.nativeSessionId, + workspaceRoot: options.workspaceRoot, + sourcePath: options.sourcePath, + title: options.title, + createdAt: options.createdAt, + updatedAt: options.updatedAt, + diagnostics: options.diagnostics, + }) +} diff --git a/src/main/ingest/providers/codex-appserver/protocol.ts b/src/main/ingest/providers/codex-appserver/protocol.ts new file mode 100644 index 0000000..c0dccdc --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/protocol.ts @@ -0,0 +1,112 @@ +import { Schema } from "effect" + +// --- codex app-server wire items ------------------------------------------- +// `codex app-server` (JSON-RPC 2.0 over stdio) streams a turn as `item/*` +// notifications; the authoritative final state of each arrives on +// `item/completed` as an `item` whose own `type` discriminates the variant. +// These schemas cover the item variants a normal coding turn produces, decoded +// the same way as the rollout-file provider (`codex.ts`): a `Schema.Union` per +// discriminant, `decode*Option` so an unknown/malformed shape decodes to None +// and is skipped rather than throwing. `NonEmptyString` mirrors the flat +// provider's treatment of "" as absent. +// +// The full protocol for the installed binary is emitted by +// `codex app-server generate-json-schema --out DIR` / `generate-ts --out DIR`; +// this module hand-mirrors only the subset the normalizer consumes today. +// mcpToolCall / fileChange follow the same shape and are added when fixtured. +const NeStr = Schema.NonEmptyString + +/** A user turn: `content` is a part array; only text parts are renderable here. */ +export const UserMessageItem = Schema.Struct({ + type: Schema.Literal("userMessage"), + id: Schema.optional(Schema.String), + content: Schema.optional( + Schema.Array(Schema.Struct({ type: Schema.String, text: Schema.optional(Schema.String) })), + ), +}) + +/** An assistant message chunk. `phase` is "final_answer" | "commentary"; both are text. */ +export const AgentMessageItem = Schema.Struct({ + type: Schema.Literal("agentMessage"), + id: Schema.optional(Schema.String), + text: Schema.optional(Schema.String), + phase: Schema.optional(Schema.String), +}) + +/** A reasoning block. `summary`/`content` are part arrays (often empty). */ +export const ReasoningItem = Schema.Struct({ + type: Schema.Literal("reasoning"), + id: Schema.optional(Schema.String), + summary: Schema.optional(Schema.Array(Schema.Unknown)), + content: Schema.optional(Schema.Array(Schema.Unknown)), +}) + +/** + * A shell command run. Unlike the rollout-file provider — which recovers the + * body + exit code by regex-stripping a `Chunk ID / Wall time / Process exited + * with code N / Output:` telemetry preamble out of the result *text* — the + * app-server delivers `command`, `exitCode`, and `aggregatedOutput` as first- + * class fields. `aggregatedOutput` is null when the command produced no output. + */ +export const CommandExecutionItem = Schema.Struct({ + type: Schema.Literal("commandExecution"), + id: NeStr, + command: Schema.optional(Schema.String), + cwd: Schema.optional(Schema.String), + status: Schema.optional(Schema.String), + exitCode: Schema.optional(Schema.NullOr(Schema.Number)), + aggregatedOutput: Schema.optional(Schema.NullOr(Schema.String)), +}) + +export const AppServerItem = Schema.Union([ + UserMessageItem, + AgentMessageItem, + ReasoningItem, + CommandExecutionItem, +]) +export type AppServerItem = typeof AppServerItem.Type +export const decodeItem = Schema.decodeUnknownOption(AppServerItem) + +// --- token usage ----------------------------------------------------------- +// `thread/tokenUsage/updated` params. `last` is the delta for the turn just +// completed; `total` the running thread total. Mirrors the rollout provider's +// `token_count` handling (which keys off `last`). +const TokenAmounts = Schema.Struct({ + totalTokens: Schema.optional(Schema.Number), + inputTokens: Schema.optional(Schema.Number), + cachedInputTokens: Schema.optional(Schema.Number), + outputTokens: Schema.optional(Schema.Number), + reasoningOutputTokens: Schema.optional(Schema.Number), +}) + +export const TokenUsageParams = Schema.Struct({ + tokenUsage: Schema.Struct({ + total: Schema.optional(TokenAmounts), + last: Schema.optional(TokenAmounts), + modelContextWindow: Schema.optional(Schema.Number), + }), +}) +export type TokenUsageParams = typeof TokenUsageParams.Type +export const decodeUsage = Schema.decodeUnknownOption(TokenUsageParams) + +// --- driver-facing method payloads ----------------------------------------- +// The `thread/start` response carries the thread id the rest of the session +// keys off — a missing one is fatal, so the driver decodes to Option and fails. +export const ThreadStartResult = Schema.Struct({ thread: Schema.Struct({ id: NeStr }) }) +export const decodeThreadStart = Schema.decodeUnknownOption(ThreadStartResult) + +/** + * The params of an `item//requestApproval` server→client request. `itemId` + * links to the exact tool-call item already streamed; `availableDecisions` is + * the server-supplied set of allowable answers (accept / acceptForSession / + * acceptWithExecpolicyAmendment / cancel …) the UI must offer verbatim. + */ +export const ApprovalRequestParams = Schema.Struct({ + itemId: Schema.optional(Schema.String), + command: Schema.optional(Schema.String), + cwd: Schema.optional(Schema.String), + reason: Schema.optional(Schema.String), + availableDecisions: Schema.optional(Schema.Array(Schema.Unknown)), +}) +export type ApprovalRequestParams = typeof ApprovalRequestParams.Type +export const decodeApprovalParams = Schema.decodeUnknownOption(ApprovalRequestParams) diff --git a/src/main/ingest/providers/codex-appserver/transport.ts b/src/main/ingest/providers/codex-appserver/transport.ts new file mode 100644 index 0000000..faabeda --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/transport.ts @@ -0,0 +1,204 @@ +import { spawn } from "node:child_process" +import { createInterface } from "node:readline" +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. + * + * Three inbound shapes are routed: + * - **response** — has `id` + (`result`|`error`), no `method` → resolves the + * matching in-flight {@link AppServerTransport.request}. + * - **server request** — has `id` + `method` → the server asking the client + * (e.g. an approval); surfaced on {@link AppServerTransport.serverRequests} + * for the client to answer with {@link AppServerTransport.respond}. + * - **notification** — `method`, no `id` → surfaced on + * {@link AppServerTransport.notifications}. + * + * The raw `readline`/`exit` callbacks only `Queue.offerUnsafe` onto an inbound + * queue (a no-op once shut down at scope close); a single scoped fiber drains it + * and applies all Effectful routing — the same raw-callback→queue→scoped-fiber + * seam the PTY manager uses. On process exit every pending request is failed and + * the two output streams end, so no caller hangs on a dead server. + */ +export class AppServerTransportError extends Data.TaggedError("AppServerTransportError")<{ + readonly message: string + readonly cause?: unknown +}> {} + +type IdKey = number | string +type Rec = Record + +export interface JsonRpcNotification { + readonly method: string + readonly params: unknown +} + +export interface JsonRpcServerRequest { + readonly id: IdKey + readonly method: string + readonly params: 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 + /** Fire-and-forget notification (no response expected). */ + 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 +} + +export interface AppServerTransportOptions { + /** Command to spawn, e.g. `"codex"`. */ + readonly command: string + /** Args, e.g. `["app-server"]`. */ + readonly args?: ReadonlyArray + readonly cwd?: string + readonly env?: Record +} + +type Inbound = { readonly kind: "message"; readonly msg: Rec } | { readonly kind: "exit" } + +/** Spawn an app-server child and return a scoped transport (killed on scope close). */ +export const makeAppServerTransport = ( + options: AppServerTransportOptions, +): Effect.Effect => + Effect.gen(function* () { + const child = yield* Effect.acquireRelease( + Effect.sync(() => + spawn(options.command, [...(options.args ?? [])], { + stdio: ["pipe", "pipe", "inherit"], + cwd: options.cwd, + env: options.env ? { ...process.env, ...options.env } : process.env, + }), + ), + (proc) => + Effect.sync(() => { + try { + proc.kill() + } catch { + /* already gone */ + } + }), + ) + + const stdin = child.stdin + const stdout = child.stdout + if (!stdin || !stdout) { + return yield* Effect.fail( + new AppServerTransportError({ message: `${options.command} spawned without stdio pipes` }), + ) + } + + const inbound = yield* Queue.make() + const notifications = yield* Queue.make() + const serverRequests = yield* Queue.make() + const pending = new Map>() + let nextId = 1 + + // Raw callbacks only offer; `offerUnsafe` is a no-op after scope-close shutdown. + const lines = createInterface({ input: stdout }) + lines.on("line", (line) => { + const trimmed = line.trim() + if (trimmed.length === 0) return + let msg: unknown + try { + msg = JSON.parse(trimmed) + } catch { + return // app-server emits only JSON on stdout; stderr is inherited + } + if (msg !== null && typeof msg === "object") { + Queue.offerUnsafe(inbound, { kind: "message", msg: msg as Rec }) + } + }) + child.on("exit", () => Queue.offerUnsafe(inbound, { kind: "exit" })) + // A spawn failure (bad command, missing cwd) fires `error`, not `exit`; treat + // it as an exit so pending requests fail instead of an unhandled throw. + child.on("error", () => Queue.offerUnsafe(inbound, { kind: "exit" })) + + const route = (event: Inbound): Effect.Effect => + Effect.gen(function* () { + if (event.kind === "exit") { + // Fail every in-flight request and end both output streams so nobody hangs. + for (const [, deferred] of pending) { + yield* Deferred.fail( + deferred, + new AppServerTransportError({ message: `${options.command} exited` }), + ) + } + pending.clear() + yield* Queue.shutdown(notifications) + yield* Queue.shutdown(serverRequests) + return + } + const msg = event.msg + const id = msg["id"] as IdKey | undefined + const method = msg["method"] + if (id != null && typeof method !== "string") { + const deferred = pending.get(id) + if (!deferred) return + pending.delete(id) + if ("error" in msg) { + const error = msg["error"] + const detail = typeof error === "object" && error !== null ? error : { message: String(error) } + yield* Deferred.fail( + deferred, + new AppServerTransportError({ message: "app-server request failed", cause: detail }), + ) + } else { + yield* Deferred.succeed(deferred, msg["result"]) + } + return + } + if (id != null && typeof method === "string") { + yield* Queue.offer(serverRequests, { id, method, params: msg["params"] }) + return + } + if (typeof method === "string") { + yield* Queue.offer(notifications, { method, params: msg["params"] }) + } + }) + + yield* Stream.fromQueue(inbound).pipe(Stream.runForEach(route), Effect.forkScoped) + + const writeLine = (payload: Rec): Effect.Effect => + Effect.try({ + try: () => { + stdin.write(`${JSON.stringify(payload)}\n`) + }, + catch: (cause) => new AppServerTransportError({ message: "write to app-server failed", cause }), + }) + + const request = (method: string, params?: unknown): Effect.Effect => + Effect.gen(function* () { + const id = nextId++ + const deferred = yield* Deferred.make() + pending.set(id, deferred) + yield* writeLine({ method, id, params: params ?? {} }).pipe( + Effect.tapError(() => Effect.sync(() => pending.delete(id))), + ) + return yield* Deferred.await(deferred) + }) + + const notify = (method: string, params?: unknown): Effect.Effect => + writeLine({ method, params: params ?? {} }) + + const respond = (id: IdKey, result: unknown): Effect.Effect => + writeLine({ id, result }) + + return { + request, + notify, + respond, + notifications: Stream.fromQueue(notifications), + serverRequests: Stream.fromQueue(serverRequests), + } satisfies AppServerTransport + }) diff --git a/tests/codex-appserver-driver.test.ts b/tests/codex-appserver-driver.test.ts new file mode 100644 index 0000000..a83037a --- /dev/null +++ b/tests/codex-appserver-driver.test.ts @@ -0,0 +1,84 @@ +import { Effect, type Scope, Stream, SubscriptionRef } from "effect" +import { describe, expect, it } from "vitest" +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 +// before "running" the command; once the client answers request 501 it streams +// the commandExecution + agentMessage + token usage, completes the turn, and +// confirms the approval resolved. +const 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: { userAgent: 'peer' } }) + else if (m.method === 'thread/start') send({ id: m.id, result: { thread: { id: 'thr_test' } } }) + else if (m.method === 'turn/start') { + send({ id: m.id, result: { turn: { id: 'turn_1', status: 'inProgress' } } }) + send({ method: 'turn/started', params: { turn: { id: 'turn_1' } } }) + send({ method: 'item/completed', params: { item: { type: 'userMessage', id: 'u1', content: [{ type: 'text', text: 'hi' }] } } }) + send({ id: 501, method: 'item/commandExecution/requestApproval', params: { itemId: 'call_1', command: 'echo hi', availableDecisions: ['accept', 'cancel'] } }) + } else if (m.method == null && m.id === 501) { + send({ method: 'item/completed', params: { item: { type: 'commandExecution', id: 'call_1', command: 'echo hi', status: 'completed', exitCode: 0, aggregatedOutput: 'hi\\n' } } }) + send({ method: 'item/completed', params: { item: { type: 'agentMessage', id: 'a1', text: 'done', phase: 'final_answer' } } }) + send({ method: 'thread/tokenUsage/updated', params: { tokenUsage: { last: { inputTokens: 100, outputTokens: 5 }, modelContextWindow: 200000 } } }) + send({ method: 'turn/completed', params: { turn: { id: 'turn_1', status: 'completed' } } }) + send({ method: 'serverRequest/resolved', params: { requestId: 501 } }) + } +}) +` + +const run = (program: Effect.Effect): Promise => + Effect.runPromise(Effect.scoped(program)) + +describe("codex app-server driver", () => { + it( + "handshakes, records + answers an approval, and folds the turn into rows", + () => + run( + Effect.gen(function* () { + const driver = yield* makeCodexAppServerDriver({ + cwd: process.cwd(), + model: "gpt-5.4", + sandbox: "read-only", + approvalPolicy: "untrusted", + command: process.execPath, + args: ["-e", PEER], + }) + expect(driver.threadId).toBe("thr_test") + + // Answer approvals as they surface (mirrors the UI answering the signal). + yield* SubscriptionRef.changes(driver.pendingApprovals).pipe( + Stream.filter((list) => list.length > 0), + Stream.take(1), + Stream.runForEach((list) => { + expect(list[0]?.itemId).toBe("call_1") + expect(list[0]?.availableDecisions).toContain("accept") + return driver.answerApproval(list[0]!.id, "accept") + }), + Effect.forkScoped, + ) + + const result = yield* driver.runTurn("hi") + expect(result.status).toBe("completed") + expect(result.rows.messages.map((m) => m.role)).toEqual(["user", "assistant"]) + expect(result.rows.messages.find((m) => m.role === "assistant")?.text).toBe("done") + + const tool = result.rows.toolCalls.find((t) => t.nativeToolId === "call_1") + expect(tool?.name).toBe("shell") + expect(tool?.outputText).toBe("hi\n") + + expect(result.rows.usageEvents[0]?.inputTokens).toBe(100) + expect(result.rows.session.nativeSessionId).toBe("thr_test") + + // serverRequest/resolved cleared the pending signal. + const remaining = yield* SubscriptionRef.get(driver.pendingApprovals) + expect(remaining).toHaveLength(0) + }), + ), + 15000, + ) +}) diff --git a/tests/codex-appserver-live.test.ts b/tests/codex-appserver-live.test.ts new file mode 100644 index 0000000..e87d910 --- /dev/null +++ b/tests/codex-appserver-live.test.ts @@ -0,0 +1,44 @@ +import { mkdtempSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { Effect, type Scope, Stream, SubscriptionRef } from "effect" +import { describe, expect, it } from "vitest" +import { makeCodexAppServerDriver } from "../src/main/ingest/providers/codex-appserver/driver.js" + +// Live end-to-end smoke against the real `codex app-server` binary. Opt-in +// (needs an authenticated codex + network), so it is skipped unless CODEX_LIVE=1. +// Run: `CODEX_LIVE=1 pnpm exec vitest run tests/codex-appserver-live.test.ts`. +const run = (program: Effect.Effect): Promise => + Effect.runPromise(Effect.scoped(program)) + +describe("codex app-server driver (live)", () => { + it.runIf(process.env.CODEX_LIVE === "1")( + "drives a real codex turn and folds it into rows", + () => + run( + Effect.gen(function* () { + const cwd = mkdtempSync(join(tmpdir(), "codex-live-")) + const driver = yield* makeCodexAppServerDriver({ + cwd, + sandbox: "read-only", + approvalPolicy: "on-request", + }) + // Auto-accept any approval so the turn can complete unattended. + yield* SubscriptionRef.changes(driver.pendingApprovals).pipe( + Stream.filter((list) => list.length > 0), + Stream.runForEach((list) => driver.answerApproval(list[0]!.id, "accept")), + Effect.forkScoped, + ) + + const result = yield* driver.runTurn( + "Reply with exactly the word DONE and nothing else. Do not run any commands.", + ) + expect(result.status).toBe("completed") + const assistant = result.rows.messages.filter((m) => m.role === "assistant") + expect(assistant.length).toBeGreaterThan(0) + expect(result.rows.session.provider).toBe("codex") + }), + ), + 60000, + ) +}) diff --git a/tests/codex-appserver-normalize.test.ts b/tests/codex-appserver-normalize.test.ts new file mode 100644 index 0000000..5444345 --- /dev/null +++ b/tests/codex-appserver-normalize.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from "vitest" +import { normalizeAppServerThread } from "../src/main/ingest/providers/codex-appserver/normalize.js" + +// Real `item/completed` payloads + `thread/tokenUsage/updated` params captured +// from `codex app-server` (codex-cli 0.142.2) driving one turn: a prompt asking +// to write a file under a read-only sandbox, one shell exec, and a "DONE" reply. +// The reasoning item arrived with empty summary/content (no visible thinking). +const items: ReadonlyArray = [ + { + type: "userMessage", + id: "5015c0d2-8c42-4e28-b950-d5d7817814a6", + content: [ + { type: "text", text: "Using a single shell command, create a file named spike.txt containing the text hi in the current directory, then reply with exactly DONE." }, + ], + }, + { type: "reasoning", id: "rs_0ef5", summary: [], content: [] }, + { + type: "commandExecution", + id: "call_h0jxrRzNJjyd8GdGvVgJbrZF", + command: "/bin/zsh -lc \"printf 'hi\\n' > spike.txt\"", + cwd: "/tmp/appserver-spike-LEvqUJ", + status: "completed", + aggregatedOutput: null, + exitCode: 0, + }, + { type: "agentMessage", id: "msg_0ef5", text: "DONE", phase: "final_answer" }, +] + +const usage: ReadonlyArray = [ + { tokenUsage: { last: { inputTokens: 12534, outputTokens: 51 }, modelContextWindow: 258400 } }, + { tokenUsage: { last: { inputTokens: 12619, outputTokens: 71 }, modelContextWindow: 258400 } }, +] + +const rows = normalizeAppServerThread(items, usage, { + nativeSessionId: "019f2af5-thread", + workspaceRoot: "/work", + sourcePath: "appserver:019f2af5-thread", + model: "gpt-5.4", +}) + +describe("codex app-server → ExtractedRows", () => { + it("maps user + assistant items to messages and skips empty reasoning", () => { + expect(rows.messages.map((m) => m.role)).toEqual(["user", "assistant"]) + expect(rows.messages[0]?.text).toContain("create a file named spike.txt") + const assistant = rows.messages.find((m) => m.role === "assistant") + expect(assistant?.text).toBe("DONE") + expect(assistant?.model).toBe("gpt-5.4") + expect(rows.messages.some((m) => m.thinking)).toBe(false) + }) + + it("maps a commandExecution to a shell tool call with structured output", () => { + const tool = rows.toolCalls.find((t) => t.nativeToolId === "call_h0jxrRzNJjyd8GdGvVgJbrZF") + expect(tool).toBeDefined() + expect(tool?.name).toBe("shell") + expect(tool?.kind).toBe("shell") + // exit 0 with no output → empty string (no EXEC_WRAPPER preamble to strip). + expect(tool?.outputText).toBe("") + expect(JSON.parse(tool!.inputJson!).command).toContain("printf 'hi") + }) + + it("prefixes a nonzero exit code, mirroring the rollout provider", () => { + const failed = normalizeAppServerThread( + [{ type: "commandExecution", id: "call_x", command: "false", status: "completed", exitCode: 2, aggregatedOutput: "boom\n" }], + [], + { nativeSessionId: "s", workspaceRoot: "/w", sourcePath: "appserver:s" }, + ) + expect(failed.toolCalls[0]?.outputText).toBe("[exit 2]\nboom\n") + }) + + it("orders 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("projects each token-usage snapshot from its `last` delta", () => { + expect(rows.usageEvents).toHaveLength(usage.length) + expect(rows.usageEvents.map((u) => u.inputTokens)).toEqual([12534, 12619]) + expect(rows.usageEvents[0]?.contextWindowTokens).toBe(258400) + expect(rows.usageEvents[0]?.contextUsedTokens).toBe(12534) + }) + + it("carries provider + session identity through finish", () => { + expect(rows.session.provider).toBe("codex") + expect(rows.session.nativeSessionId).toBe("019f2af5-thread") + expect(rows.session.workspaceRoot).toBe("/work") + expect(rows.session.title).toContain("create a file named spike.txt") + }) +}) diff --git a/tests/codex-appserver-transport.test.ts b/tests/codex-appserver-transport.test.ts new file mode 100644 index 0000000..0710caf --- /dev/null +++ b/tests/codex-appserver-transport.test.ts @@ -0,0 +1,70 @@ +import { Effect, type Scope, Stream } from "effect" +import { describe, expect, it } from "vitest" +import { + AppServerTransportError, + makeAppServerTransport, +} from "../src/main/ingest/providers/codex-appserver/transport.js" + +// A scripted NDJSON JSON-RPC peer (stands in for `codex app-server`, no auth / +// network). On `initialize` it answers, then emits one notification and one +// server→client request; `boom` returns a JSON-RPC error; a client response to +// request id 777 (the approval) is echoed back as an `approval/echo` +// notification so the test can prove `respond` reached the peer. +const 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: { ok: true } }) + send({ method: 'turn/started', params: { n: 1 } }) + send({ id: 777, method: 'item/commandExecution/requestApproval', params: { itemId: 'call_x' } }) + } else if (m.method === 'boom') { + send({ id: m.id, error: { code: 1, message: 'nope' } }) + } else if (m.method == null && m.id === 777) { + send({ method: 'approval/echo', params: m.result }) + } +}) +` + +const run = (program: Effect.Effect): Promise => + Effect.runPromise(Effect.scoped(program)) + +describe("codex app-server transport", () => { + it( + "round-trips a request, surfaces a JSON-RPC error, and delivers notifications + server requests", + () => + run( + Effect.gen(function* () { + const t = yield* makeAppServerTransport({ command: process.execPath, args: ["-e", PEER] }) + + // request → response + const res = yield* t.request("initialize", { clientInfo: { name: "test" } }) + expect(res).toEqual({ ok: true }) + + // a JSON-RPC error becomes a typed failure (flip: error → success value) + const failure = yield* Effect.flip(t.request("boom")) + expect(failure).toBeInstanceOf(AppServerTransportError) + + // the peer's server→client request (buffered during initialize) → answer it + yield* t.serverRequests.pipe( + Stream.take(1), + Stream.runForEach((req) => { + expect(req.id).toBe(777) + expect(req.method).toContain("requestApproval") + expect(req.params).toMatchObject({ itemId: "call_x" }) + return t.respond(req.id, { decision: "accept" }) + }), + ) + + // turn/started (from initialize) then approval/echo (proof respond landed) + const notifs = yield* 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" }) + }), + ), + 15000, + ) +}) From df57d0d12422c52a6a03bf996a5c4bee005fbffd Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sat, 4 Jul 2026 13:43:58 +1000 Subject: [PATCH 02/26] feat(provider): declare an app-server launch capability on ProviderSpec MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add an optional `appServer` capability to ProviderSpec — a third launch mode beside `batch` (non-interactive runs) and `interactive` (PTY TUI): a long-lived JSON-RPC process on stdio the codex-appserver driver spawns and drives. Populate it for codex (`codex app-server`). Optional field, so the ListProviders RPC Schema stays backward-compatible. The capability is additive — a provider that declares it can still be scraped post-hoc and launched as a TUI. Its reader is the driver-launch wiring (next step); this just makes the capability discoverable via ProviderRegistry. --- src/main/services/ProviderRegistry.ts | 4 ++++ src/shared/provider.ts | 16 ++++++++++++++++ 2 files changed, 20 insertions(+) diff --git a/src/main/services/ProviderRegistry.ts b/src/main/services/ProviderRegistry.ts index 72b2f51..b220091 100644 --- a/src/main/services/ProviderRegistry.ts +++ b/src/main/services/ProviderRegistry.ts @@ -35,6 +35,10 @@ const providers: ReadonlyArray = [ promptInjectionMode: "stdin-after-start", readyPromptGlyph: "›", }, + // Drive codex directly over the app-server JSON-RPC protocol (see + // ingest/providers/codex-appserver/driver.ts) — additive to the scraper + + // TUI paths above. + appServer: { launchCmd: "codex", args: ["app-server"] }, }, { kind: "cursor", diff --git a/src/shared/provider.ts b/src/shared/provider.ts index 8d29faa..e32eddd 100644 --- a/src/shared/provider.ts +++ b/src/shared/provider.ts @@ -77,6 +77,21 @@ export const InteractiveCapability = Schema.Struct({ }) export type InteractiveCapability = typeof InteractiveCapability.Type +/** + * App-server launch capability: a long-lived JSON-RPC 2.0 process on stdio + * (`codex app-server`) that arc drives directly — a third acquisition transport + * beside `batch` (non-interactive runs) and `interactive` (PTY TUI). The driver + * (`ingest/providers/codex-appserver/driver.ts`) spawns `launchCmd` with `args`, + * speaks the thread/turn/item protocol, and folds each turn into the same + * `ExtractedRows` the rollout-file provider produces. Additive: a provider that + * declares this can still be scraped post-hoc and launched as a TUI. + */ +export const AppServerCapability = Schema.Struct({ + launchCmd: Schema.String, + args: Schema.Array(Schema.String), +}) +export type AppServerCapability = typeof AppServerCapability.Type + export const ProviderSpec = Schema.Struct({ kind: Provider, displayName: Schema.String, @@ -84,5 +99,6 @@ export const ProviderSpec = Schema.Struct({ concurrency: Concurrency, batch: Schema.optional(BatchCapability), interactive: Schema.optional(InteractiveCapability), + appServer: Schema.optional(AppServerCapability), }) export type ProviderSpec = typeof ProviderSpec.Type From 6e52a418a551e85de3beb6f8573a4c1dd4a51c7a Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sat, 4 Jul 2026 13:49:32 +1000 Subject: [PATCH 03/26] feat(ingest): land live codex app-server turns in the shared store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the driver into IngestStore so a live-driven codex session is indistinguishable from a scraped one downstream (one store, one projection). - driver: accumulate the thread cumulatively rather than per-turn. The store's replaceSession replaces a session with its whole row set (the file scraper re-parses the entire rollout each time), so each completed turn now yields the session's rows *so far*, not just that turn's delta. - launch.ts: launchCodexAppServerSession reads a ProviderSpec app-server capability, spawns the driver, and persists each completed turn's cumulative rows via IngestStore.replaceSession — the reader of the appServer capability. pendingApprovals / answerApproval pass through unchanged. Tests: hermetic node-peer coverage that a completed turn lands in an in-memory IngestStore; the env-gated live smoke now drives two real turns through launch -> driver -> normalizer -> store and asserts cumulative rows persist. --- .../providers/codex-appserver/driver.ts | 25 ++++---- .../providers/codex-appserver/launch.ts | 58 +++++++++++++++++++ tests/codex-appserver-launch.test.ts | 56 ++++++++++++++++++ tests/codex-appserver-live.test.ts | 56 +++++++++++------- 4 files changed, 158 insertions(+), 37 deletions(-) create mode 100644 src/main/ingest/providers/codex-appserver/launch.ts create mode 100644 tests/codex-appserver-launch.test.ts diff --git a/src/main/ingest/providers/codex-appserver/driver.ts b/src/main/ingest/providers/codex-appserver/driver.ts index f8f8ed0..4423f01 100644 --- a/src/main/ingest/providers/codex-appserver/driver.ts +++ b/src/main/ingest/providers/codex-appserver/driver.ts @@ -36,12 +36,13 @@ export interface PendingApproval { 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 the turn completes, with its rows + status. */ + /** 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> @@ -110,21 +111,18 @@ export const makeCodexAppServerDriver = ( const pendingApprovals = yield* SubscriptionRef.make>([]) const turnOutcomes = yield* Queue.make() - // One sequential fiber folds the notification stream. `turn/started` opens a - // fresh accumulator; each `item/completed` and token-usage snapshot appends; - // `turn/completed` hands the accumulated turn off to `runTurn` via the queue. + // One sequential fiber folds the notification stream. Accumulation is + // thread-cumulative, not per-turn: the store's `replaceSession` replaces a + // session with the whole row set (the file scraper re-parses the entire + // rollout each time), so each `turn/completed` hands `runTurn` the session's + // rows *so far* — a snapshot copy, since the accumulators keep growing. // Single-consumer, so the mutable accumulators need no locking. - let items: Array = [] - let usage: Array = [] + const items: Array = [] + const usage: Array = [] yield* transport.notifications.pipe( Stream.runForEach((notification) => Effect.gen(function* () { switch (notification.method) { - case "turn/started": { - items = [] - usage = [] - break - } case "item/completed": { const item = obj(notification.params)?.["item"] if (item !== undefined) items.push(item) @@ -137,10 +135,7 @@ export const makeCodexAppServerDriver = ( case "turn/completed": { const turn = obj(obj(notification.params)?.["turn"]) const status = str(turn?.["status"]) ?? "unknown" - const outcome: TurnOutcome = { items, usage, status } - items = [] - usage = [] - yield* Queue.offer(turnOutcomes, outcome) + yield* Queue.offer(turnOutcomes, { items: items.slice(), usage: usage.slice(), status }) break } case "serverRequest/resolved": { diff --git a/src/main/ingest/providers/codex-appserver/launch.ts b/src/main/ingest/providers/codex-appserver/launch.ts new file mode 100644 index 0000000..3afe968 --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/launch.ts @@ -0,0 +1,58 @@ +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 +} + +/** + * Launch a codex app-server driver from a provider's {@link AppServerCapability} + * and persist each completed turn's cumulative rows into the shared + * {@link IngestStore}. The rows and the store are the same ones the rollout-file + * provider writes, so a live-driven session and a scraped session are + * indistinguishable downstream — one store, one projection. This is the reader + * of the `ProviderSpec.appServer` capability. + * + * Returns the driver with a `runTurn` that also persists; `pendingApprovals` / + * `answerApproval` are unchanged (the UI answers the pending-input signal). + */ +export const launchCodexAppServerSession = ( + capability: AppServerCapability, + params: CodexLaunchParams, +): Effect.Effect => + Effect.gen(function* () { + const store = yield* IngestStore + const driver = yield* makeCodexAppServerDriver({ + command: capability.launchCmd, + args: [...capability.args], + cwd: params.cwd, + model: params.model, + sandbox: params.sandbox, + approvalPolicy: params.approvalPolicy, + env: params.env, + clientName: params.clientName, + }) + + const runTurn = (text: string) => + driver.runTurn(text).pipe( + Effect.tap((result) => + store + .replaceSession(result.rows) + .pipe(Effect.mapError((cause) => new CodexDriverError({ message: "persist turn failed", cause }))), + ), + ) + + return { ...driver, runTurn } + }) diff --git a/tests/codex-appserver-launch.test.ts b/tests/codex-appserver-launch.test.ts new file mode 100644 index 0000000..f507f2b --- /dev/null +++ b/tests/codex-appserver-launch.test.ts @@ -0,0 +1,56 @@ +import { Effect, Layer, ManagedRuntime, type Scope } 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 { launchCodexAppServerSession } from "../src/main/ingest/providers/codex-appserver/launch.js" + +// A minimal thread/turn peer (no approval) that completes one turn with a +// user + assistant message, so we can assert the turn lands in the store. +const 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 === 'thread/start') send({ id: m.id, result: { thread: { id: 'thr_persist' } } }) + else if (m.method === 'turn/start') { + send({ id: m.id, result: { turn: { id: 't1', status: 'inProgress' } } }) + send({ method: 'item/completed', params: { item: { type: 'userMessage', id: 'u1', content: [{ type: 'text', text: 'hello' }] } } }) + send({ method: 'item/completed', params: { item: { type: 'agentMessage', id: 'a1', text: 'hi there', phase: 'final_answer' } } }) + send({ method: 'turn/completed', params: { turn: { status: 'completed' } } }) + } +}) +` + +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("codex app-server launch → store", () => { + it( + "persists a completed turn's rows into the shared IngestStore", + () => + run( + Effect.gen(function* () { + const driver = yield* launchCodexAppServerSession( + { launchCmd: process.execPath, args: ["-e", PEER] }, + { cwd: process.cwd() }, + ) + const turn = yield* driver.runTurn("hello") + expect(turn.status).toBe("completed") + + const store = yield* IngestStore + const stored = yield* store.getSession(turn.rows.session.id) + expect(stored).toBeDefined() + expect(stored?.session.provider).toBe("codex") + expect(stored?.session.nativeSessionId).toBe("thr_persist") + expect(stored?.messages.map((m) => m.role)).toEqual(["user", "assistant"]) + expect(stored?.messages.find((m) => m.role === "assistant")?.text).toBe("hi there") + }), + ), + 15000, + ) +}) diff --git a/tests/codex-appserver-live.test.ts b/tests/codex-appserver-live.test.ts index e87d910..bd60dea 100644 --- a/tests/codex-appserver-live.test.ts +++ b/tests/codex-appserver-live.test.ts @@ -1,44 +1,56 @@ import { mkdtempSync } from "node:fs" import { tmpdir } from "node:os" import { join } from "node:path" -import { Effect, type Scope, Stream, SubscriptionRef } from "effect" +import { Effect, Layer, ManagedRuntime, type Scope, Stream, SubscriptionRef } from "effect" import { describe, expect, it } from "vitest" -import { makeCodexAppServerDriver } from "../src/main/ingest/providers/codex-appserver/driver.js" +import { IngestStore, IngestStoreLive } from "../src/main/ingest/db/store.js" +import { sqliteLayer } from "../src/main/ingest/db/sqlite.js" +import { launchCodexAppServerSession } from "../src/main/ingest/providers/codex-appserver/launch.js" -// Live end-to-end smoke against the real `codex app-server` binary. Opt-in -// (needs an authenticated codex + network), so it is skipped unless CODEX_LIVE=1. +// Live end-to-end smoke against the real `codex app-server` binary, through the +// full pipeline: launch → driver → normalizer → shared store, across two turns +// (proving cumulative accumulation + store persistence). Opt-in (needs an +// authenticated codex + network), skipped unless CODEX_LIVE=1. // Run: `CODEX_LIVE=1 pnpm exec vitest run tests/codex-appserver-live.test.ts`. -const run = (program: Effect.Effect): Promise => - Effect.runPromise(Effect.scoped(program)) +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("codex app-server driver (live)", () => { +describe("codex app-server (live, full pipeline)", () => { it.runIf(process.env.CODEX_LIVE === "1")( - "drives a real codex turn and folds it into rows", + "drives two real turns and persists cumulative rows to the store", () => run( Effect.gen(function* () { const cwd = mkdtempSync(join(tmpdir(), "codex-live-")) - const driver = yield* makeCodexAppServerDriver({ - cwd, - sandbox: "read-only", - approvalPolicy: "on-request", - }) - // Auto-accept any approval so the turn can complete unattended. + const driver = yield* launchCodexAppServerSession( + { launchCmd: "codex", args: ["app-server"] }, + { cwd, sandbox: "read-only", approvalPolicy: "on-request" }, + ) + // Auto-accept any approval so turns complete unattended. yield* SubscriptionRef.changes(driver.pendingApprovals).pipe( Stream.filter((list) => list.length > 0), Stream.runForEach((list) => driver.answerApproval(list[0]!.id, "accept")), Effect.forkScoped, ) - const result = yield* driver.runTurn( - "Reply with exactly the word DONE and nothing else. Do not run any commands.", - ) - expect(result.status).toBe("completed") - const assistant = result.rows.messages.filter((m) => m.role === "assistant") - expect(assistant.length).toBeGreaterThan(0) - expect(result.rows.session.provider).toBe("codex") + const t1 = yield* driver.runTurn("Reply with exactly the word ONE and nothing else. No commands.") + expect(t1.status).toBe("completed") + const usersAfter1 = t1.rows.messages.filter((m) => m.role === "user").length + + const t2 = yield* driver.runTurn("Reply with exactly the word TWO and nothing else. No commands.") + expect(t2.status).toBe("completed") + const usersAfter2 = t2.rows.messages.filter((m) => m.role === "user").length + // Cumulative: the second turn's rows include the first turn's user message too. + expect(usersAfter2).toBeGreaterThan(usersAfter1) + + const store = yield* IngestStore + const stored = yield* store.getSession(t2.rows.session.id) + expect(stored?.session.provider).toBe("codex") + expect(stored?.messages.filter((m) => m.role === "assistant").length).toBeGreaterThanOrEqual(2) }), ), - 60000, + 90000, ) }) From 6544b544413b30932219c930f5e8b3fe11153f59 Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sat, 4 Jul 2026 14:43:55 +1000 Subject: [PATCH 04/26] feat(codex): driver registry + approvalId, the app-server answer surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit App-server approvals can't be answered by focusing a PTY (there is none), so Arc must own the answer surface. First main-side pieces: - PendingApproval / ApprovalRequestParams gain approvalId — codex's own approval handle (present on commandExecution approvals, absent on fileChange). It is display/correlation detail; the reliable routing key stays the always-present JSON-RPC request id the request arrived with. - CodexDriverRegistry: owns the live drivers keyed by targetSessionId, mirrors each driver's pendingApprovals into one reactive aggregate (SessionApprovals), and routes AnswerApproval back to the right driver. register is scoped to the caller (the session launch): the mirror fiber + deregistration finalizer live in that scope, so a session that ends drops out of the aggregate. Hermetic test registers a peer-backed driver, answers the approval *through the registry* (proving per-session routing), and asserts the aggregate clears on serverRequest/resolved. --- .../providers/codex-appserver/driver.ts | 4 + .../providers/codex-appserver/protocol.ts | 7 ++ src/main/services/CodexDriverRegistry.ts | 113 ++++++++++++++++++ tests/codex-driver-registry.test.ts | 75 ++++++++++++ 4 files changed, 199 insertions(+) create mode 100644 src/main/services/CodexDriverRegistry.ts create mode 100644 tests/codex-driver-registry.test.ts diff --git a/src/main/ingest/providers/codex-appserver/driver.ts b/src/main/ingest/providers/codex-appserver/driver.ts index 4423f01..96a33a2 100644 --- a/src/main/ingest/providers/codex-appserver/driver.ts +++ b/src/main/ingest/providers/codex-appserver/driver.ts @@ -25,7 +25,10 @@ export class CodexDriverError extends Data.TaggedError("CodexDriverError")<{ /** 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 @@ -166,6 +169,7 @@ export const makeCodexAppServerDriver = ( ) const approval: PendingApproval = { id: request.id, + approvalId: params.approvalId ?? null, itemId: params.itemId ?? null, command: params.command ?? null, availableDecisions: params.availableDecisions ?? [], diff --git a/src/main/ingest/providers/codex-appserver/protocol.ts b/src/main/ingest/providers/codex-appserver/protocol.ts index c0dccdc..3e003a9 100644 --- a/src/main/ingest/providers/codex-appserver/protocol.ts +++ b/src/main/ingest/providers/codex-appserver/protocol.ts @@ -102,6 +102,13 @@ export const decodeThreadStart = Schema.decodeUnknownOption(ThreadStartResult) * acceptWithExecpolicyAmendment / cancel …) the UI must offer verbatim. */ export const ApprovalRequestParams = Schema.Struct({ + /** + * Codex's own approval handle. Present on commandExecution approvals but not + * all types (fileChange approvals omit it), so it is display/correlation + * detail — the reliable routing key stays the always-present JSON-RPC request + * id the request arrived with. + */ + approvalId: Schema.optional(Schema.String), itemId: Schema.optional(Schema.String), command: Schema.optional(Schema.String), cwd: Schema.optional(Schema.String), diff --git a/src/main/services/CodexDriverRegistry.ts b/src/main/services/CodexDriverRegistry.ts new file mode 100644 index 0000000..fea1392 --- /dev/null +++ b/src/main/services/CodexDriverRegistry.ts @@ -0,0 +1,113 @@ +import { Context, Effect, Layer, type Scope, Stream, SubscriptionRef } from "effect" +import type { + CodexAppServerDriver, + CodexDriverError, + PendingApproval, +} from "../ingest/providers/codex-appserver/driver.js" + +/** The live approvals awaiting an answer for one session. */ +export interface SessionApprovals { + readonly chatId: string + readonly targetSessionId: string + readonly approvals: ReadonlyArray +} + +/** + * Owns the live `codex app-server` drivers keyed by `targetSessionId`, so the + * app-server path has an answer surface a PTY session never needed: it + * aggregates every driver's `pendingApprovals` into one reactive list the + * renderer signal + inline approval card read, and routes `AnswerApproval` back + * to the right driver's `answerApproval`. + * + * `register` is scoped to the caller (the session launch): the mirror fiber and + * the deregistration finalizer live in that scope, so a driver whose session + * ends drops out of the aggregate automatically — no stale approvals. + */ +export class CodexDriverRegistry extends Context.Service< + CodexDriverRegistry, + { + /** Register a driver for a session; deregisters when the caller's scope closes. */ + readonly register: (params: { + readonly chatId: string + readonly targetSessionId: string + readonly driver: CodexAppServerDriver + }) => 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 + /** Sessions with outstanding approvals (empty sessions omitted). */ + readonly pending: Effect.Effect> + /** Reactive view of `pending`: the current aggregate, then every change. */ + readonly changes: Stream.Stream> + } +>()("arcwork/CodexDriverRegistry") {} + +export const CodexDriverRegistryLive = Layer.effect( + CodexDriverRegistry, + Effect.gen(function* () { + const drivers = new Map() + const current = new Map() + const state = yield* SubscriptionRef.make>([]) + + // Publish only sessions that actually have approvals — the aggregate is the + // attention signal, so an emptied session should disappear from it. + const republish = () => + SubscriptionRef.set( + state, + [...current.values()].filter((s) => s.approvals.length > 0), + ) + + const register = (params: { + readonly chatId: string + readonly targetSessionId: string + readonly driver: CodexAppServerDriver + }) => + Effect.gen(function* () { + const { chatId, targetSessionId, driver } = params + drivers.set(targetSessionId, { chatId, driver }) + + // Mirror the driver's live approvals into the aggregate. `changes` on a + // SubscriptionRef replays its current value, so the initial (usually + // empty) state is reflected immediately. + yield* SubscriptionRef.changes(driver.pendingApprovals).pipe( + Stream.runForEach((approvals) => + Effect.gen(function* () { + current.set(targetSessionId, { chatId, targetSessionId, approvals }) + yield* republish() + }), + ), + Effect.forkScoped, + ) + + // Drop the session from the aggregate when its launch scope closes. + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + drivers.delete(targetSessionId) + current.delete(targetSessionId) + yield* republish() + }), + ) + }) + + const answerApproval = (targetSessionId: string, requestId: number | string, decision: unknown) => + Effect.gen(function* () { + const entry = drivers.get(targetSessionId) + if (!entry) { + // No live driver: the session ended or was never app-server-driven. + // Nothing to answer — a stale click is a no-op, not an error. + return + } + yield* entry.driver.answerApproval(requestId, decision) + }) + + return { + register, + answerApproval, + pending: SubscriptionRef.get(state), + changes: SubscriptionRef.changes(state), + } + }), +) diff --git a/tests/codex-driver-registry.test.ts b/tests/codex-driver-registry.test.ts new file mode 100644 index 0000000..e6d2d89 --- /dev/null +++ b/tests/codex-driver-registry.test.ts @@ -0,0 +1,75 @@ +import { Effect, ManagedRuntime, type Scope, Stream } from "effect" +import { describe, expect, it } from "vitest" +import { makeCodexAppServerDriver } from "../src/main/ingest/providers/codex-appserver/driver.js" +import { CodexDriverRegistry, CodexDriverRegistryLive } from "../src/main/services/CodexDriverRegistry.js" + +// Scripted thread/turn peer that asks for approval mid-turn (request id 501) and +// completes once answered. Mirrors the driver test's peer. +const 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 === 'thread/start') send({ id: m.id, result: { thread: { id: 'thr_reg' } } }) + else if (m.method === 'turn/start') { + send({ id: m.id, result: { turn: { id: 't1', status: 'inProgress' } } }) + send({ method: 'turn/started', params: { turn: { id: 't1' } } }) + send({ id: 501, method: 'item/commandExecution/requestApproval', params: { approvalId: 'appr_1', itemId: 'call_1', command: 'echo hi', availableDecisions: ['accept', 'cancel'] } }) + } else if (m.method == null && m.id === 501) { + send({ method: 'item/completed', params: { item: { type: 'agentMessage', id: 'a1', text: 'done', phase: 'final_answer' } } }) + send({ method: 'turn/completed', params: { turn: { status: 'completed' } } }) + send({ method: 'serverRequest/resolved', params: { requestId: 501 } }) + } +}) +` + +const run = (program: Effect.Effect): Promise => { + const runtime = ManagedRuntime.make(CodexDriverRegistryLive) + return runtime.runPromise(Effect.scoped(program)).finally(() => runtime.dispose()) +} + +describe("codex driver registry", () => { + it( + "aggregates a session's approvals and routes the answer to its driver", + () => + run( + Effect.gen(function* () { + const registry = yield* CodexDriverRegistry + const driver = yield* makeCodexAppServerDriver({ + cwd: process.cwd(), + command: process.execPath, + args: ["-e", PEER], + }) + yield* registry.register({ chatId: "chat_1", targetSessionId: "target_1", driver }) + + // Answer through the REGISTRY (not the driver) — proves routing by session. + yield* registry.changes.pipe( + Stream.filter((list) => list.length > 0), + Stream.take(1), + Stream.runForEach((list) => { + const session = list[0]! + expect(session.targetSessionId).toBe("target_1") + expect(session.chatId).toBe("chat_1") + const approval = session.approvals[0]! + expect(approval.approvalId).toBe("appr_1") + expect(approval.itemId).toBe("call_1") + expect(approval.availableDecisions).toContain("accept") + return registry.answerApproval(session.targetSessionId, approval.id, "accept") + }), + Effect.forkScoped, + ) + + const result = yield* driver.runTurn("hi") + expect(result.status).toBe("completed") + + // serverRequest/resolved → driver ref → mirror → aggregate empties. + const pending = yield* registry.pending + expect(pending).toHaveLength(0) + }), + ), + 15000, + ) +}) From f6d3fe0c9d3866ccd056f2e15af26978d5bcecc6 Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sat, 4 Jul 2026 14:47:53 +1000 Subject: [PATCH 05/26] feat(rpc): expose codex app-server approvals to the renderer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The answer surface app-server needs, reaching the renderer: - shared AppServerApproval schema — the inline card's view of an approval (requestId, approvalId, itemId, command, decisions). Ephemeral, never persisted. Each decision is {label, payload} where payload is the raw server decision re-encoded as JSON, so acceptWithExecpolicyAmendment and friends round-trip verbatim — the decision model is never collapsed to approve/deny. - codex-approval-view: projectApprovals flattens the registry's per-session aggregate into that view; parseDecisionPayload decodes a chosen decision back to the raw value to answer the driver. - three RPCs: ListAppServerApprovals (one-shot), WatchAppServerApprovals (stream), AnswerAppServerApproval (echo a decision payload back). Handlers read/route through CodexDriverRegistry, now provided in the runtime layer graph. Pure test covers the projection + verbatim decision round-trip (string and rule-carrying object decisions). --- src/main/rpc.ts | 10 ++++ src/main/runtime.ts | 10 +++- src/main/services/codex-approval-view.ts | 44 ++++++++++++++++++ src/shared/codex-approval.ts | 37 +++++++++++++++ src/shared/rpc.ts | 19 ++++++++ tests/codex-approval-view.test.ts | 59 ++++++++++++++++++++++++ 6 files changed, 178 insertions(+), 1 deletion(-) create mode 100644 src/main/services/codex-approval-view.ts create mode 100644 src/shared/codex-approval.ts create mode 100644 tests/codex-approval-view.test.ts diff --git a/src/main/rpc.ts b/src/main/rpc.ts index 692ee07..7e92ab2 100644 --- a/src/main/rpc.ts +++ b/src/main/rpc.ts @@ -14,6 +14,8 @@ import { ChatMessageService } from "./services/ChatMessageService.js" import { LocalModelService } from "./services/LocalModelService.js" import { TargetSessionManager } from "./services/TargetSessionManager.js" import { LiveTargetStateService } from "./services/LiveTargetStateService.js" +import { CodexDriverRegistry } from "./services/CodexDriverRegistry.js" +import { projectApprovals, parseDecisionPayload } from "./services/codex-approval-view.js" import { ArtifactIngestService } from "./services/ArtifactIngestService.js" import { WorkService } from "./work/service.js" import { ReadService } from "./read/service.js" @@ -132,6 +134,8 @@ export const ArcRpcHandlersLive = ArcRpcs.toLayer( WatchWorkspaces: () => Stream.unwrap(Effect.map(WorkspaceService, (_) => _.changes)), WatchLiveTargetStates: () => Stream.unwrap(Effect.map(LiveTargetStateService, (_) => _.changes)), + WatchAppServerApprovals: () => + Stream.unwrap(Effect.map(CodexDriverRegistry, (_) => Stream.map(_.changes, projectApprovals))), // Invalidation-signal streams: forward each service's change PubSub as a // stream of tiny descriptors (the renderer re-pulls the affected list). Unlike // the SubscriptionRef lists above these PubSubs don't replay, so a fresh @@ -145,6 +149,12 @@ export const ArcRpcHandlersLive = ArcRpcs.toLayer( TestLocalModel: svc("TestLocalModel", LocalModelService, (_) => _.status), ListPendingRequests: svc("ListPendingRequests", ChatMessageService, (_) => _.listPending), ListLiveTargetStates: svc("ListLiveTargetStates", LiveTargetStateService, (_) => _.list), + ListAppServerApprovals: svc("ListAppServerApprovals", CodexDriverRegistry, (_) => + Effect.map(_.pending, projectApprovals), + ), + AnswerAppServerApproval: svc("AnswerAppServerApproval", CodexDriverRegistry, (_, req) => + _.answerApproval(req.targetSessionId, req.requestId, parseDecisionPayload(req.decisionPayload)), + ), SearchArc: svc("SearchArc", ReadService, (_, req) => _.search(req.params)), GetArc: svc("GetArc", ReadService, (_, req) => _.get(req.params)), CreateChat: svc("CreateChat", ChatService, (_, req) => _.create(req.workspaceId, req.title)), diff --git a/src/main/runtime.ts b/src/main/runtime.ts index 471d1c2..7d5ea82 100644 --- a/src/main/runtime.ts +++ b/src/main/runtime.ts @@ -2,6 +2,7 @@ import { Layer, ManagedRuntime, References } from "effect" import * as NodeFileSystem from "@effect/platform-node/NodeFileSystem" import * as NodePath from "@effect/platform-node/NodePath" import { ProviderRegistryLive } from "./services/ProviderRegistry.js" +import { CodexDriverRegistryLive } from "./services/CodexDriverRegistry.js" import { PresetRegistryLive } from "./services/PresetRegistry.js" import { WorkspaceServiceLive } from "./services/WorkspaceService.js" import { WorkspaceFilesServiceLive } from "./services/WorkspaceFilesService.js" @@ -43,7 +44,14 @@ const WorkLive = WorkServiceLive.pipe(Layer.provide(Layer.mergeAll(WorkStoreLive const PersistenceLayers = [StoreLive, IngestStoreLiveLayer, WorkStoreLiveLayer] as const // Static registries and small local facades that do not depend on durable state. -const RegistryLayers = [ProviderRegistryLive, PresetRegistryLive, LocalModelServiceLive] as const +// CodexDriverRegistry holds live app-server drivers keyed by target session and +// aggregates their approvals for the renderer answer surface. +const RegistryLayers = [ + ProviderRegistryLive, + CodexDriverRegistryLive, + PresetRegistryLive, + LocalModelServiceLive, +] as const // Core domain services. These are intentionally built from the persistence and // registry constants above so shared in-memory projections and the session diff --git a/src/main/services/codex-approval-view.ts b/src/main/services/codex-approval-view.ts new file mode 100644 index 0000000..23e453e --- /dev/null +++ b/src/main/services/codex-approval-view.ts @@ -0,0 +1,44 @@ +import type { AppServerApproval, AppServerApprovalDecision } from "../../shared/codex-approval.js" +import { obj } 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. + */ +const decisionView = (decision: unknown): AppServerApprovalDecision => { + const label = + typeof decision === "string" ? decision : (Object.keys(obj(decision) ?? {})[0] ?? "decision") + return { label, payload: JSON.stringify(decision) } +} + +/** Flatten the registry's per-session aggregate into the renderer-facing list. */ +export const projectApprovals = ( + sessions: ReadonlyArray, +): ReadonlyArray => + sessions.flatMap((session) => + session.approvals.map((approval) => ({ + chatId: session.chatId, + targetSessionId: session.targetSessionId, + requestId: approval.id, + approvalId: approval.approvalId, + itemId: approval.itemId, + command: approval.command, + decisions: approval.availableDecisions.map(decisionView), + })), + ) + +/** + * Decode a decision `payload` back to the raw value to answer the driver. + * Payloads originate from {@link decisionView}'s `JSON.stringify`, so parsing + * succeeds; a malformed one falls back to the literal string rather than throwing. + */ +export const parseDecisionPayload = (payload: string): unknown => { + try { + return JSON.parse(payload) + } catch { + return payload + } +} diff --git a/src/shared/codex-approval.ts b/src/shared/codex-approval.ts new file mode 100644 index 0000000..c361e27 --- /dev/null +++ b/src/shared/codex-approval.ts @@ -0,0 +1,37 @@ +import { Schema } from "effect" + +/** + * The renderer-facing view of a codex app-server approval — the answer surface a + * PTY session never needed. Unlike hook permissions (a coarse sidebar flag), + * app-server approvals must be *answered inside Arc*, so this carries the detail + * the inline card needs. Ephemeral: never persisted, mirrored live from the + * driver's in-memory state (see CodexDriverRegistry). + */ + +/** + * One allowable answer, server-owned. `payload` is the raw decision re-encoded as + * JSON so it survives the round-trip verbatim — the decision model is not + * collapsed to approve/deny (`acceptWithExecpolicyAmendment` carries a rule + * payload, etc.); the renderer echoes `payload` back unchanged. + */ +export const AppServerApprovalDecision = Schema.Struct({ + /** Display label for the button (a string decision's text, or an object decision's key). */ + label: Schema.String, + /** JSON of the raw server decision, sent back to `AnswerAppServerApproval` unchanged. */ + payload: Schema.String, +}) +export type AppServerApprovalDecision = typeof AppServerApprovalDecision.Type + +export const AppServerApproval = Schema.Struct({ + chatId: Schema.String, + targetSessionId: Schema.String, + /** JSON-RPC request id — the routing key echoed back to answer this approval. */ + requestId: Schema.Union([Schema.String, Schema.Number]), + /** Codex's approval handle when present (commandExecution only); display detail. */ + approvalId: Schema.NullOr(Schema.String), + /** The tool-call item this approval gates. */ + itemId: Schema.NullOr(Schema.String), + command: Schema.NullOr(Schema.String), + decisions: Schema.Array(AppServerApprovalDecision), +}) +export type AppServerApproval = typeof AppServerApproval.Type diff --git a/src/shared/rpc.ts b/src/shared/rpc.ts index 12272a3..fccd333 100644 --- a/src/shared/rpc.ts +++ b/src/shared/rpc.ts @@ -8,6 +8,7 @@ import { ActivityEvent } from "./activity-event.js" import { ChatMessage } from "./chat-message.js" import { GitCommit, GitFileDiff, GitStatus, PullRequest, Worktree, WorkspaceGitContext } from "./git.js" import { PendingRequest } from "./chat-request.js" +import { AppServerApproval } from "./codex-approval.js" import { Instance, TargetSession } from "./instance.js" import { LiveTargetState } from "./live-target-state.js" import { ArcGetParams, ArcGetResult, ArcSearchParams, ArcSearchResult } from "./read.js" @@ -278,6 +279,24 @@ export const ArcRpcs = RpcGroup.make( error: RpcError, stream: true, }), + /** Outstanding codex app-server approvals — the one-shot floor under `WatchAppServerApprovals`. */ + Rpc.make("ListAppServerApprovals", { success: Schema.Array(AppServerApproval), error: RpcError }), + /** Live codex app-server approvals as a server stream (the inline-card answer surface). */ + Rpc.make("WatchAppServerApprovals", { + success: Schema.Array(AppServerApproval), + error: RpcError, + stream: true, + }), + /** Answer a codex app-server approval by echoing a decision's `payload` back verbatim. */ + Rpc.make("AnswerAppServerApproval", { + payload: { + targetSessionId: Schema.String, + requestId: Schema.Union([Schema.String, Schema.Number]), + decisionPayload: Schema.String, + }, + success: Schema.Void, + error: RpcError, + }), /** Renderer door onto the same unified read surface as MCP `arc.search`. */ Rpc.make("SearchArc", { payload: { params: ArcSearchParams }, diff --git a/tests/codex-approval-view.test.ts b/tests/codex-approval-view.test.ts new file mode 100644 index 0000000..e8cf726 --- /dev/null +++ b/tests/codex-approval-view.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it } from "vitest" +import { parseDecisionPayload, projectApprovals } from "../src/main/services/codex-approval-view.js" +import type { SessionApprovals } from "../src/main/services/CodexDriverRegistry.js" + +const sessions: ReadonlyArray = [ + { + chatId: "chat_1", + targetSessionId: "target_1", + approvals: [ + { + id: 501, + approvalId: "appr_1", + itemId: "call_1", + command: "printf hi > f", + // A string decision and an object (rule-carrying) decision. + availableDecisions: ["accept", { acceptWithExecpolicyAmendment: { execpolicy_amendment: ["x"] } }, "cancel"], + }, + ], + }, + { + chatId: "chat_2", + targetSessionId: "target_2", + approvals: [{ id: "req-9", approvalId: null, itemId: "call_2", command: null, availableDecisions: ["accept"] }], + }, +] + +describe("codex approval projection", () => { + it("flattens sessions and normalizes decisions to label + verbatim payload", () => { + const view = projectApprovals(sessions) + expect(view).toHaveLength(2) + + const first = view.find((a) => a.requestId === 501)! + expect(first.chatId).toBe("chat_1") + expect(first.targetSessionId).toBe("target_1") + expect(first.approvalId).toBe("appr_1") + expect(first.itemId).toBe("call_1") + expect(first.decisions.map((d) => d.label)).toEqual(["accept", "acceptWithExecpolicyAmendment", "cancel"]) + // Object decision's payload preserves the rule verbatim. + expect(JSON.parse(first.decisions[1]!.payload)).toEqual({ + acceptWithExecpolicyAmendment: { execpolicy_amendment: ["x"] }, + }) + + const second = view.find((a) => a.requestId === "req-9")! + expect(second.approvalId).toBeNull() + expect(second.command).toBeNull() + }) + + it("round-trips a decision payload back to its raw value", () => { + const view = projectApprovals(sessions) + const amend = view[0]!.decisions[1]! + // The renderer echoes payload back; the driver receives the exact object. + expect(parseDecisionPayload(amend.payload)).toEqual({ + acceptWithExecpolicyAmendment: { execpolicy_amendment: ["x"] }, + }) + expect(parseDecisionPayload('"accept"')).toBe("accept") + // A malformed payload degrades to the literal string, never throws. + expect(parseDecisionPayload("not json")).toBe("not json") + }) +}) From a131e1e6eac20f047cc01df7b747225eabcebe23 Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sat, 4 Jul 2026 14:51:23 +1000 Subject: [PATCH 06/26] feat(chat): inline approval card for the app-server answer surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The renderer half of the pty-less answer surface. Unlike the Question card (deliberately read-only — a PTY provider's picker owns the answer), AppServerApproval *is* the answer surface: it renders the server-supplied decisions verbatim as buttons and calls onAnswer(payload) with the raw decision JSON, so acceptWithExecpolicyAmendment and friends round-trip unchanged. Presentational only (approval + onAnswer + answering props), styled with the shared Button/Badge primitives. Verified in Storybook across command-exec, rule-carrying (execpolicy amendment), fileChange (no command), and answering (disabled) states. --- .../src/chat/AppServerApproval.stories.tsx | 75 +++++++++++++++++++ src/renderer/src/chat/AppServerApproval.tsx | 72 ++++++++++++++++++ 2 files changed, 147 insertions(+) create mode 100644 src/renderer/src/chat/AppServerApproval.stories.tsx create mode 100644 src/renderer/src/chat/AppServerApproval.tsx diff --git a/src/renderer/src/chat/AppServerApproval.stories.tsx b/src/renderer/src/chat/AppServerApproval.stories.tsx new file mode 100644 index 0000000..ced7469 --- /dev/null +++ b/src/renderer/src/chat/AppServerApproval.stories.tsx @@ -0,0 +1,75 @@ +import type { ReactNode } from "react" +import type { AppServerApproval as AppServerApprovalData } from "../../../shared/codex-approval.js" +import { AppServerApproval } from "./AppServerApproval.js" + +export default { + title: "Chat / App-Server Approval", +} + +// Approvals sit flat on the pane background, like the Question card. +const Frame = ({ children }: { readonly children: ReactNode }) => ( +
{children}
+) + +const noop = () => {} + +const decision = (label: string, raw: unknown = label) => ({ label, payload: JSON.stringify(raw) }) + +// A read-only sandbox write: the common shell-exec approval. +const commandExec: AppServerApprovalData = { + chatId: "chat_1", + targetSessionId: "target_1", + requestId: 501, + approvalId: "appr_1", + itemId: "call_lmCyiThzGf03OzBhckpeeMHB", + command: "/bin/zsh -lc \"printf 'hi\\n' > spike.txt\"", + decisions: [decision("accept"), decision("acceptForSession"), decision("cancel")], +} + +// The rule-carrying decision the card must not collapse away. +const withExecpolicyAmendment: AppServerApprovalData = { + ...commandExec, + requestId: 502, + decisions: [ + decision("accept"), + decision("acceptWithExecpolicyAmendment", { + acceptWithExecpolicyAmendment: { execpolicy_amendment: ["/bin/zsh", "-lc", "printf 'hi\\n' > spike.txt"] }, + }), + decision("cancel"), + ], +} + +// A fileChange approval: no command, only an item + reason. +const fileChange: AppServerApprovalData = { + chatId: "chat_1", + targetSessionId: "target_1", + requestId: "req-9", + approvalId: null, + itemId: "call_fileChange_7", + command: null, + decisions: [decision("accept"), decision("acceptForSession"), decision("decline")], +} + +export const CommandExecution = () => ( + + + +) + +export const WithExecpolicyAmendment = () => ( + + + +) + +export const FileChange = () => ( + + + +) + +export const Answering = () => ( + + + +) diff --git a/src/renderer/src/chat/AppServerApproval.tsx b/src/renderer/src/chat/AppServerApproval.tsx new file mode 100644 index 0000000..1fe00cb --- /dev/null +++ b/src/renderer/src/chat/AppServerApproval.tsx @@ -0,0 +1,72 @@ +import type { JSX } from "react" +import type { AppServerApproval as AppServerApprovalData } from "../../../shared/codex-approval.js" +import { Badge } from "../ui/Badge.js" +import { Button } from "../ui/Button.js" + +const CARD = "grid gap-2.5 min-w-0" +const HEAD = "flex items-center justify-between gap-2" +const TITLE = "font-mono text-xs font-semibold text-foreground" +const COMMAND = + "px-[7px] py-1 border border-border bg-input text-foreground font-mono text-[11px] leading-[1.35] whitespace-pre-wrap [overflow-wrap:anywhere]" +const PROMPT = "text-[13px] leading-[1.45] text-foreground [overflow-wrap:anywhere]" +const HINT = "font-mono text-[11px] text-fg-faint [overflow-wrap:anywhere]" + +/** + * Tone a decision button by its label. Unlike a PTY provider's own picker, this + * card *is* the answer surface, so the affordances must read at a glance: + * accept-family solid, cancel/decline danger, everything else (e.g. + * acceptWithExecpolicyAmendment) a quieter ghost. + */ +const decisionVariant = (label: string): "solid" | "danger" | "ghost" => { + const l = label.toLowerCase() + if (l.startsWith("accept") || l === "approve" || l === "allow") return "solid" + if (l === "cancel" || l === "decline" || l === "deny" || l === "reject") return "danger" + return "ghost" +} + +export interface AppServerApprovalProps { + readonly approval: AppServerApprovalData + /** + * Answer the approval with a decision's `payload` (the raw server decision, + * JSON-encoded). The container echoes it back verbatim via + * `AnswerAppServerApproval` — the decision model is never collapsed. + */ + readonly onAnswer?: (payload: string) => void + /** Disable the buttons while an answer is in flight (optimistic). */ + readonly answering?: boolean +} + +/** + * A codex app-server approval awaiting an answer — the inline card that replaces + * "focus the PTY and use the provider's picker" for the pty-less app-server path. + * Renders the server-supplied decisions verbatim as buttons; there is no PTY to + * defer to, so Arc owns the interaction. + */ +export function AppServerApproval({ approval, onAnswer, answering }: AppServerApprovalProps): JSX.Element { + return ( +
+
+ approval + awaiting answer +
+ {approval.command ? ( +
{approval.command}
+ ) : ( +
This action needs your approval.
+ )} +
+ {approval.decisions.map((decision) => ( + + ))} +
+ {approval.itemId && {approval.itemId}} +
+ ) +} From 16b14b17626e27a667a35cdceff47803500cfad2 Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sat, 4 Jul 2026 15:38:58 +1000 Subject: [PATCH 07/26] =?UTF-8?q?feat(session):=20RpcSessionManager=20?= =?UTF-8?q?=E2=80=94=20runtime=20owner=20for=20app-server=20sessions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The structured-process sibling of the PTY TargetSessionManager, and the piece that makes app-server sessions live. TargetSession identity is unchanged; only the runtime owner differs (per the pty/rpc runtime split). - launch(req): spawns the driver (persisting turns to the shared store via launchCodexAppServerSession) and registers its approvals with CodexDriverRegistry. Idempotent per targetSessionId. Each session gets a scope forked off the layer scope, so stop() closes just that session (kills the driver, deregisters its approvals) and app quit closes them all. - submit(req): runs a user turn (driver.runTurn); accepted:false for an unknown session. - stop / list round out the lifecycle. Provider-agnostic (command/args come from the app-server capability, mapped in by the caller); codex is the only RPC driver today, pi's rpc mode is the same family. Wired into the runtime layer graph. Hermetic test drives a node peer through the whole cycle: launch, answer an approval *through the registry* (proving launch registered the driver), submit a turn, assert it persisted to the shared IngestStore, then stop and assert both the manager and the registry aggregate clear. --- src/main/runtime.ts | 9 ++ src/main/services/RpcSessionManager.ts | 122 +++++++++++++++++++++++++ tests/rpc-session-manager.test.ts | 87 ++++++++++++++++++ 3 files changed, 218 insertions(+) create mode 100644 src/main/services/RpcSessionManager.ts create mode 100644 tests/rpc-session-manager.test.ts diff --git a/src/main/runtime.ts b/src/main/runtime.ts index 7d5ea82..b6831bf 100644 --- a/src/main/runtime.ts +++ b/src/main/runtime.ts @@ -3,6 +3,7 @@ import * as NodeFileSystem from "@effect/platform-node/NodeFileSystem" import * as NodePath from "@effect/platform-node/NodePath" import { ProviderRegistryLive } from "./services/ProviderRegistry.js" import { CodexDriverRegistryLive } from "./services/CodexDriverRegistry.js" +import { RpcSessionManagerLive } from "./services/RpcSessionManager.js" import { PresetRegistryLive } from "./services/PresetRegistry.js" import { WorkspaceServiceLive } from "./services/WorkspaceService.js" import { WorkspaceFilesServiceLive } from "./services/WorkspaceFilesService.js" @@ -92,6 +93,13 @@ const TargetInboxLive = TargetInboxServiceLive.pipe( Layer.provide(SessionsLive), Layer.provide(LiveTargetStatesLive), ) +// Runtime owner for RPC-backed (app-server) sessions — the structured-process +// sibling of the PTY TargetSessionManager. Launches drivers, persists their +// turns, and registers their approvals for the renderer answer surface. +const RpcSessionsLive = RpcSessionManagerLive.pipe( + Layer.provide(CodexDriverRegistryLive), + Layer.provide(IngestStoreLiveLayer), +) const DomainServiceLayers = [ WorkLive, @@ -105,6 +113,7 @@ const DomainServiceLayers = [ ChatMessagesLive, LiveTargetStatesLive, TargetInboxLive, + RpcSessionsLive, SessionsLive, ] as const diff --git a/src/main/services/RpcSessionManager.ts b/src/main/services/RpcSessionManager.ts new file mode 100644 index 0000000..62c0867 --- /dev/null +++ b/src/main/services/RpcSessionManager.ts @@ -0,0 +1,122 @@ +import { Context, Effect, Exit, Layer, Scope } from "effect" +import type { + CodexAppServerDriver, + CodexDriverError, + CodexDriverOptions, +} from "../ingest/providers/codex-appserver/driver.js" +import { launchCodexAppServerSession } from "../ingest/providers/codex-appserver/launch.js" +import { IngestStore } from "../ingest/db/store.js" +import { CodexDriverRegistry } from "./CodexDriverRegistry.js" + +/** + * Launch a resident structured (RPC-backed) session. `command`/`args` come from + * the provider's app-server capability (e.g. `codex` + `["app-server"]`), mapped + * in by the caller/router, so this manager stays provider-agnostic. + */ +export interface RpcLaunchRequest { + readonly chatId: string + readonly targetSessionId: string + readonly cwd: string + readonly command: string + readonly args: ReadonlyArray + readonly model?: string + readonly sandbox?: CodexDriverOptions["sandbox"] + readonly approvalPolicy?: CodexDriverOptions["approvalPolicy"] +} + +/** + * The runtime owner for **RPC-backed** target sessions — the structured-process + * sibling of the PTY-backed `TargetSessionManager`. Where that one spawns a + * pseudo-terminal and drives it with keystrokes/paste/resize, this drives a + * resident JSON-RPC agent: `submit` runs a turn, approvals flow through the + * Arc-owned `CodexDriverRegistry` → answer RPC, and there is no byte stream or + * terminal surface. `TargetSession` identity is unchanged; only the live runtime + * differs (the renderer branches on session kind for terminal vs transcript). + * + * Each session gets a scope forked off the layer scope, so `stop` closes just + * that session (killing its driver + deregistering its approvals) and app quit + * closes them all. Today the only RPC driver is codex app-server; pi's rpc mode + * is the same family and would join here rather than through the PTY machinery. + */ +export class RpcSessionManager extends Context.Service< + RpcSessionManager, + { + /** Launch (idempotent per targetSessionId): spawn the driver, persist its + * turns, and register its approvals. */ + readonly launch: (req: RpcLaunchRequest) => Effect.Effect + /** Run a user turn against a launched session. `accepted:false` if unknown. */ + readonly submit: (req: { + readonly targetSessionId: string + readonly text: string + }) => Effect.Effect<{ readonly accepted: boolean; readonly status?: string }, CodexDriverError> + /** Tear a session down: closes its scope (kills driver, deregisters approvals). */ + readonly stop: (targetSessionId: string) => Effect.Effect<{ readonly stopped: boolean }> + /** The target session ids currently live under this manager. */ + readonly list: Effect.Effect> + } +>()("arcwork/RpcSessionManager") {} + +interface LiveRpcSession { + readonly driver: CodexAppServerDriver + readonly scope: Scope.Closeable +} + +export const RpcSessionManagerLive = Layer.effect( + RpcSessionManager, + Effect.gen(function* () { + const registry = yield* CodexDriverRegistry + const ingest = yield* IngestStore + const parentScope = yield* Effect.scope + const sessions = new Map() + + const launch = (req: RpcLaunchRequest): Effect.Effect => + Effect.gen(function* () { + if (sessions.has(req.targetSessionId)) return // idempotent + + // Child of the layer scope: closes on `stop` or on app quit. + const scope = yield* Scope.fork(parentScope) + const build = Effect.gen(function* () { + const driver = yield* launchCodexAppServerSession( + { launchCmd: req.command, args: req.args }, + { + cwd: req.cwd, + model: req.model, + sandbox: req.sandbox, + approvalPolicy: req.approvalPolicy, + }, + ) + yield* registry.register({ + chatId: req.chatId, + targetSessionId: req.targetSessionId, + driver, + }) + return driver + }).pipe(Scope.provide(scope), Effect.provideService(IngestStore, ingest)) + + const driver = yield* build.pipe( + // A failed launch must not leak the forked scope (or a half-spawned child). + Effect.tapError(() => Scope.close(scope, Exit.void)), + ) + sessions.set(req.targetSessionId, { driver, scope }) + }) + + const submit = (req: { readonly targetSessionId: string; readonly text: string }) => + Effect.gen(function* () { + const session = sessions.get(req.targetSessionId) + if (!session) return { accepted: false as const } + const result = yield* session.driver.runTurn(req.text) + return { accepted: true as const, status: result.status } + }) + + const stop = (targetSessionId: string) => + Effect.gen(function* () { + const session = sessions.get(targetSessionId) + if (!session) return { stopped: false } + sessions.delete(targetSessionId) + yield* Scope.close(session.scope, Exit.void) + return { stopped: true } + }) + + return { launch, submit, stop, list: Effect.sync(() => [...sessions.keys()]) } + }), +) diff --git a/tests/rpc-session-manager.test.ts b/tests/rpc-session-manager.test.ts new file mode 100644 index 0000000..bd01aee --- /dev/null +++ b/tests/rpc-session-manager.test.ts @@ -0,0 +1,87 @@ +import { Effect, Layer, ManagedRuntime, type Scope, Stream } 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 { CodexDriverRegistry, CodexDriverRegistryLive } from "../src/main/services/CodexDriverRegistry.js" +import { RpcSessionManager, RpcSessionManagerLive } from "../src/main/services/RpcSessionManager.js" + +// A thread/turn peer that asks for approval (request id 501) then completes. +const 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 === 'thread/start') send({ id: m.id, result: { thread: { id: 'thr_mgr' } } }) + else if (m.method === 'turn/start') { + send({ id: m.id, result: { turn: { id: 't1', status: 'inProgress' } } }) + send({ method: 'item/completed', params: { item: { type: 'userMessage', id: 'u1', content: [{ type: 'text', text: 'hi' }] } } }) + send({ id: 501, method: 'item/commandExecution/requestApproval', params: { itemId: 'call_1', command: 'echo hi', availableDecisions: ['accept', 'cancel'] } }) + } else if (m.method == null && m.id === 501) { + send({ method: 'item/completed', params: { item: { type: 'agentMessage', id: 'a1', text: 'done', phase: 'final_answer' } } }) + send({ method: 'turn/completed', params: { turn: { status: 'completed' } } }) + send({ method: 'serverRequest/resolved', params: { requestId: 501 } }) + } +}) +` + +const run = ( + program: Effect.Effect, +): Promise => { + const stores = IngestStoreLive.pipe(Layer.provide(sqliteLayer(":memory:"))) + const layer = Layer.mergeAll( + RpcSessionManagerLive.pipe(Layer.provide(Layer.mergeAll(CodexDriverRegistryLive, stores))), + CodexDriverRegistryLive, + stores, + ) + const runtime = ManagedRuntime.make(layer) + return runtime.runPromise(Effect.scoped(program)).finally(() => runtime.dispose()) +} + +describe("RpcSessionManager", () => { + it( + "launches, registers approvals, submits a turn, persists, and stops", + () => + run( + Effect.gen(function* () { + const manager = yield* RpcSessionManager + const registry = yield* CodexDriverRegistry + const store = yield* IngestStore + + yield* manager.launch({ + chatId: "chat_1", + targetSessionId: "target_1", + cwd: process.cwd(), + command: process.execPath, + args: ["-e", PEER], + }) + expect(yield* manager.list).toContain("target_1") + + // Answer the approval through the registry — proves launch registered the driver. + yield* registry.changes.pipe( + Stream.filter((list) => list.length > 0), + Stream.take(1), + Stream.runForEach((list) => + registry.answerApproval(list[0]!.targetSessionId, list[0]!.approvals[0]!.id, "accept"), + ), + Effect.forkScoped, + ) + + const res = yield* manager.submit({ targetSessionId: "target_1", text: "hi" }) + expect(res).toEqual({ accepted: true, status: "completed" }) + + // The turn landed in the shared store (indistinguishable from a scraped session). + const stored = yield* store.listSessions() + expect(stored.some((s) => s.provider === "codex" && s.nativeSessionId === "thr_mgr")).toBe(true) + + // Stop tears the session down: manager forgets it, registry deregisters. + expect(yield* manager.stop("target_1")).toEqual({ stopped: true }) + expect(yield* manager.list).toHaveLength(0) + expect(yield* registry.pending).toHaveLength(0) + }), + ), + 15000, + ) +}) From 24f7cbf6fb9b6c0f787d2104a49baf04a867db4d Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sat, 4 Jul 2026 17:35:16 +1000 Subject: [PATCH 08/26] =?UTF-8?q?feat(session):=20SessionRuntimeRouter=20?= =?UTF-8?q?=E2=80=94=20route=20pty=20vs=20rpc,=20project=20rpc=20turns?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The one door onto both session runtimes, and the wiring that makes an app-server session usable from the composer. - SessionRuntimeRouter: launch dispatches by *intent* (LaunchRequest.runtime: "pty" | "rpc") — never by provider identity, since codex declares both interactive and appServer. submit/stop dispatch by *ownership* (which manager holds the id), so no persisted kind is needed. rpc-launch resolves cwd, creates + persists a TargetSession row, spawns the driver, and bindNatives the thread id (so the timeline projection can find the target). No migration. - The router does NOT project rpc turns itself — that would depend on ChatMessageService, which depends back on the router. Instead submit returns the turn's rows and sendPrompt projects them via ingestArtifactSession, keeping the graph acyclic. RpcSessionManager.launch now returns the thread id; submit returns the cumulative rows. - sendPrompt routes through the router: it skips the PTY attached-check for rpc sessions (no byte stream), submits via the router, and projects the returned rows into the chat timeline (the driver wrote IngestStore; the renderer reads the ArcStore projection). - LaunchTarget / StopTarget RPCs route through the router; LaunchTarget gains a runtime field. SubmitPrompt stays PTY-raw. Tests: router dispatch (submit/stop/ownsRpc by ownership, rpc turn carries rows, unknown id falls through to pty). Full suite 506 passed / 1 skipped. --- src/main/rpc.ts | 5 +- src/main/runtime.ts | 29 +++-- src/main/services/ChatMessageService.ts | 39 +++++-- src/main/services/RpcSessionManager.ts | 24 ++-- src/main/services/SessionRuntimeRouter.ts | 136 ++++++++++++++++++++++ src/main/services/TargetSessionManager.ts | 8 ++ src/shared/rpc.ts | 3 + tests/live-pending-permission.test.ts | 14 +++ tests/rpc-session-manager.test.ts | 4 +- tests/session-runtime-router.test.ts | 108 +++++++++++++++++ 10 files changed, 345 insertions(+), 25 deletions(-) create mode 100644 src/main/services/SessionRuntimeRouter.ts create mode 100644 tests/session-runtime-router.test.ts diff --git a/src/main/rpc.ts b/src/main/rpc.ts index 7e92ab2..44311d2 100644 --- a/src/main/rpc.ts +++ b/src/main/rpc.ts @@ -15,6 +15,7 @@ import { LocalModelService } from "./services/LocalModelService.js" import { TargetSessionManager } from "./services/TargetSessionManager.js" import { LiveTargetStateService } from "./services/LiveTargetStateService.js" import { CodexDriverRegistry } from "./services/CodexDriverRegistry.js" +import { SessionRuntimeRouter } from "./services/SessionRuntimeRouter.js" import { projectApprovals, parseDecisionPayload } from "./services/codex-approval-view.js" import { ArtifactIngestService } from "./services/ArtifactIngestService.js" import { WorkService } from "./work/service.js" @@ -168,9 +169,9 @@ export const ArcRpcHandlersLive = ArcRpcs.toLayer( ReingestAndReprojectChatMessages: svc("ReingestAndReprojectChatMessages", ArtifactIngestService, (_, req) => _.reingestAndReprojectChat(req.chatId, req.provider), ), - LaunchTarget: svc("LaunchTarget", TargetSessionManager, (_, req) => _.launch(req)), + LaunchTarget: svc("LaunchTarget", SessionRuntimeRouter, (_, req) => _.launch(req)), ResumeTarget: svc("ResumeTarget", TargetSessionManager, (_, req) => _.resume(req)), - StopTarget: svc("StopTarget", TargetSessionManager, (_, req) => _.stop(req)), + StopTarget: svc("StopTarget", SessionRuntimeRouter, (_, req) => _.stop(req)), SubmitPrompt: svc("SubmitPrompt", TargetSessionManager, (_, req) => _.submit(req)), SendChatPrompt: svc("SendChatPrompt", ChatMessageService, (_, req) => _.sendPrompt(req)), ListWork: svc("ListWork", WorkService, (_) => _.listOpen), diff --git a/src/main/runtime.ts b/src/main/runtime.ts index b6831bf..4df2849 100644 --- a/src/main/runtime.ts +++ b/src/main/runtime.ts @@ -4,6 +4,7 @@ import * as NodePath from "@effect/platform-node/NodePath" import { ProviderRegistryLive } from "./services/ProviderRegistry.js" import { CodexDriverRegistryLive } from "./services/CodexDriverRegistry.js" import { RpcSessionManagerLive } from "./services/RpcSessionManager.js" +import { SessionRuntimeRouterLive } from "./services/SessionRuntimeRouter.js" import { PresetRegistryLive } from "./services/PresetRegistry.js" import { WorkspaceServiceLive } from "./services/WorkspaceService.js" import { WorkspaceFilesServiceLive } from "./services/WorkspaceFilesService.js" @@ -70,11 +71,31 @@ const SessionsLive = TargetSessionManagerLive.pipe( Layer.provide(HookSignalServerLive), Layer.provide(StoreLive), ) +// Runtime owner for RPC-backed (app-server) sessions — the structured-process +// sibling of the PTY TargetSessionManager. Launches drivers, persists their +// turns, and registers their approvals for the renderer answer surface. +const RpcSessionsLive = RpcSessionManagerLive.pipe( + Layer.provide(CodexDriverRegistryLive), + Layer.provide(IngestStoreLiveLayer), +) +// The one door onto both session runtimes: dispatches launch/submit/stop to the +// PTY or RPC manager (launch by intent, submit/stop by ownership). `sendPrompt` +// routes through it; it returns rpc turn rows for the caller to project (staying +// acyclic — the router does not depend on ChatMessageService). +const SessionRouterLive = SessionRuntimeRouterLive.pipe( + Layer.provide(ProviderRegistryLive), + Layer.provide(WorkspacesLive), + Layer.provide(ChatsLive), + Layer.provide(StoreLive), + Layer.provide(SessionsLive), + Layer.provide(RpcSessionsLive), +) const ChatMessagesLive = ChatMessageServiceLive.pipe( Layer.provide(StoreLive), Layer.provide(IngestStoreLiveLayer), Layer.provide(SessionsLive), + Layer.provide(SessionRouterLive), Layer.provide(ChatsLive), Layer.provide(LocalModelServiceLive), Layer.provide(ActivityEventsLive), @@ -93,13 +114,6 @@ const TargetInboxLive = TargetInboxServiceLive.pipe( Layer.provide(SessionsLive), Layer.provide(LiveTargetStatesLive), ) -// Runtime owner for RPC-backed (app-server) sessions — the structured-process -// sibling of the PTY TargetSessionManager. Launches drivers, persists their -// turns, and registers their approvals for the renderer answer surface. -const RpcSessionsLive = RpcSessionManagerLive.pipe( - Layer.provide(CodexDriverRegistryLive), - Layer.provide(IngestStoreLiveLayer), -) const DomainServiceLayers = [ WorkLive, @@ -114,6 +128,7 @@ const DomainServiceLayers = [ LiveTargetStatesLive, TargetInboxLive, RpcSessionsLive, + SessionRouterLive, SessionsLive, ] as const diff --git a/src/main/services/ChatMessageService.ts b/src/main/services/ChatMessageService.ts index 5766b96..e898bd2 100644 --- a/src/main/services/ChatMessageService.ts +++ b/src/main/services/ChatMessageService.ts @@ -15,6 +15,7 @@ import type { ChatMessage } from "../../shared/chat-message.js" import type { PendingRequest } from "../../shared/chat-request.js" import { type ChatId, newArcId, type TargetId } from "../../shared/ids.js" import { TargetSessionManager } from "./TargetSessionManager.js" +import { SessionRuntimeRouter } from "./SessionRuntimeRouter.js" import { ChatService } from "./ChatService.js" import { LocalModelService } from "./LocalModelService.js" import { ActivityEventService } from "./ActivityEventService.js" @@ -78,6 +79,7 @@ export const ChatMessageServiceLive = Layer.effect( const db = yield* ArcStore const ingest = yield* IngestStore const sessions = yield* TargetSessionManager + const router = yield* SessionRuntimeRouter const chats = yield* ChatService const localModel = yield* LocalModelService const activity = yield* ActivityEventService @@ -398,13 +400,19 @@ export const ChatMessageServiceLive = Layer.effect( ) } - const live = (yield* sessions.list).find((session) => session.id === req.targetSessionId) - if (!live?.attached) { - return yield* Effect.fail( - arcRequestError( - `Target session "${stored.provider}" is not running — attach or resume it before sending`, - ), - ) + // rpc (app-server) sessions live under RpcSessionManager, not the PTY + // manager, and have no "attached" byte-stream — skip the PTY liveness + // check for them (router.submit reports its own not-running). + const isRpc = yield* router.ownsRpc(req.targetSessionId) + if (!isRpc) { + const live = (yield* sessions.list).find((session) => session.id === req.targetSessionId) + if (!live?.attached) { + return yield* Effect.fail( + arcRequestError( + `Target session "${stored.provider}" is not running — attach or resume it before sending`, + ), + ) + } } const occurredAt = yield* nowIso @@ -442,7 +450,15 @@ export const ChatMessageServiceLive = Layer.effect( } } - const delivery = yield* sessions.submit({ instanceId: req.targetSessionId, text }) + const delivery = yield* router + .submit({ instanceId: req.targetSessionId, text }) + .pipe( + Effect.catchTag("CodexDriverError", (e) => + Effect.logWarning(`rpc turn failed (${req.targetSessionId}): ${e.message}`).pipe( + Effect.as({ accepted: false as const }), + ), + ), + ) if (!delivery.accepted) { yield* db.deleteChatMessageByDedupKey(row.dedupKey).pipe( Effect.tapError((e: SqlError) => @@ -457,6 +473,13 @@ export const ChatMessageServiceLive = Layer.effect( ) } + // An rpc turn returns its cumulative rows; project them into the chat + // timeline (the driver wrote IngestStore, the renderer reads the ArcStore + // projection). A pty turn's transcript arrives via the file watcher instead. + if ("rows" in delivery && delivery.rows) { + yield* ingestArtifactSession(delivery.rows) + } + const titleSeed = titleSeedFromMessages( yield* db.loadChatMessagesForChat(req.chatId).pipe( Effect.tapError((e: SqlError) => diff --git a/src/main/services/RpcSessionManager.ts b/src/main/services/RpcSessionManager.ts index 62c0867..ccb3734 100644 --- a/src/main/services/RpcSessionManager.ts +++ b/src/main/services/RpcSessionManager.ts @@ -1,4 +1,5 @@ import { Context, Effect, Exit, Layer, Scope } from "effect" +import type { ExtractedRows } from "../ingest/db/schema.js" import type { CodexAppServerDriver, CodexDriverError, @@ -42,13 +43,20 @@ export class RpcSessionManager extends Context.Service< RpcSessionManager, { /** Launch (idempotent per targetSessionId): spawn the driver, persist its - * turns, and register its approvals. */ - readonly launch: (req: RpcLaunchRequest) => Effect.Effect - /** Run a user turn against a launched session. `accepted:false` if unknown. */ + * turns, register its approvals; returns the driver's thread id so the caller + * can `bindNative` it (needed for the timeline projection). */ + readonly launch: ( + req: RpcLaunchRequest, + ) => Effect.Effect<{ readonly threadId: string }, CodexDriverError> + /** 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: { readonly targetSessionId: string readonly text: string - }) => Effect.Effect<{ readonly accepted: boolean; readonly status?: string }, CodexDriverError> + }) => Effect.Effect< + { readonly accepted: boolean; readonly status?: string; readonly rows?: ExtractedRows }, + CodexDriverError + > /** Tear a session down: closes its scope (kills driver, deregisters approvals). */ readonly stop: (targetSessionId: string) => Effect.Effect<{ readonly stopped: boolean }> /** The target session ids currently live under this manager. */ @@ -69,9 +77,10 @@ export const RpcSessionManagerLive = Layer.effect( const parentScope = yield* Effect.scope const sessions = new Map() - const launch = (req: RpcLaunchRequest): Effect.Effect => + const launch = (req: RpcLaunchRequest): Effect.Effect<{ readonly threadId: string }, CodexDriverError> => Effect.gen(function* () { - if (sessions.has(req.targetSessionId)) return // idempotent + const existing = sessions.get(req.targetSessionId) + if (existing) return { threadId: existing.driver.threadId } // idempotent // Child of the layer scope: closes on `stop` or on app quit. const scope = yield* Scope.fork(parentScope) @@ -98,6 +107,7 @@ export const RpcSessionManagerLive = Layer.effect( Effect.tapError(() => Scope.close(scope, Exit.void)), ) sessions.set(req.targetSessionId, { driver, scope }) + return { threadId: driver.threadId } }) const submit = (req: { readonly targetSessionId: string; readonly text: string }) => @@ -105,7 +115,7 @@ export const RpcSessionManagerLive = Layer.effect( const session = sessions.get(req.targetSessionId) if (!session) return { accepted: false as const } const result = yield* session.driver.runTurn(req.text) - return { accepted: true as const, status: result.status } + return { accepted: true as const, status: result.status, rows: result.rows } }) const stop = (targetSessionId: string) => diff --git a/src/main/services/SessionRuntimeRouter.ts b/src/main/services/SessionRuntimeRouter.ts new file mode 100644 index 0000000..c9220a7 --- /dev/null +++ b/src/main/services/SessionRuntimeRouter.ts @@ -0,0 +1,136 @@ +import { Clock, Context, Effect, Layer } from "effect" +import type { SqlError } from "effect/unstable/sql/SqlError" +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 ArcRequestError, arcRequestError } from "../errors.js" +import { ChatService } from "./ChatService.js" +import { ProviderRegistry } from "./ProviderRegistry.js" +import { RpcSessionManager } from "./RpcSessionManager.js" +import { + type LaunchRequest, + type StopRequest, + type SubmitRequest, + TargetSessionManager, +} from "./TargetSessionManager.js" +import { WorkspaceService } from "./WorkspaceService.js" + +/** + * Dispatches launch/submit/stop across the two session runtimes — the PTY + * `TargetSessionManager` and the RPC-backed `RpcSessionManager` — so the rest of + * the app has one door. Launch picks the runtime by **intent** (`req.runtime`), + * never by provider identity: codex declares both `interactive` and `appServer`, + * so it can be launched either way. Submit/stop route by *ownership* (which + * manager holds the id), which needs no persisted kind. + * + * The router deliberately does NOT project rpc turns into the chat timeline + * itself — that would require depending on `ChatMessageService`, which depends + * back on the router (`sendPrompt`). Instead `submit` returns the turn's `rows` + * and the caller (`sendPrompt`) projects them, keeping the dependency acyclic. + */ +export class SessionRuntimeRouter extends Context.Service< + SessionRuntimeRouter, + { + readonly launch: ( + req: LaunchRequest, + ) => 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> + 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 + } +>()("arcwork/SessionRuntimeRouter") {} + +export const SessionRuntimeRouterLive = Layer.effect( + SessionRuntimeRouter, + Effect.gen(function* () { + const providers = yield* ProviderRegistry + const workspaces = yield* WorkspaceService + const chats = yield* ChatService + const db = yield* ArcStore + const pty = yield* TargetSessionManager + const rpc = yield* RpcSessionManager + + const nowIso = Effect.map(Clock.currentTimeMillis, (ms) => new Date(ms).toISOString()) + const ownsRpc = (id: string) => Effect.map(rpc.list, (ids) => ids.includes(id)) + + // Create + persist a TargetSession row for an rpc launch, spawn the driver, + // and bind the thread id so `ingestArtifactSession` can find the target when + // the caller projects a turn's rows. + const launchRpc = (req: LaunchRequest) => + Effect.gen(function* () { + const spec = yield* providers.get(req.provider) + if (!spec?.appServer) { + return yield* Effect.fail( + arcRequestError(`Provider "${req.provider}" has no app-server capability`), + ) + } + const chat = yield* chats.get(req.chatId) + const workspace = yield* workspaces.get(req.workspaceId ?? chat.workspaceId) + const cwd = workspace.path + const id = newArcId("target") + const startedAt = yield* nowIso + + yield* db.upsertTargetSession({ + id, + chatId: req.chatId, + provider: req.provider, + origin: req.origin ?? "manual", + spawnedBy: req.spawnedBy ?? null, + preset: req.preset ?? null, + cwd, + nativeSessionId: null, + nativeTranscriptPath: null, + state: "running", + startedAt, + }) + + const { threadId } = yield* rpc.launch({ + chatId: req.chatId, + targetSessionId: id, + cwd, + command: spec.appServer.launchCmd, + args: spec.appServer.args, + sandbox: "workspace-write", + approvalPolicy: "on-request", + }) + yield* pty.bindNative(id, threadId) + + return { + _tag: "TargetSession", + id, + provider: req.provider, + origin: req.origin ?? "manual", + chatId: req.chatId, + cwd, + nativeSessionId: threadId, + attached: true, + state: "running", + startedAt, + } satisfies TargetSession + }) + + const launch = (req: LaunchRequest) => (req.runtime === "rpc" ? launchRpc(req) : pty.launch(req)) + + const submit = (req: SubmitRequest) => + Effect.gen(function* () { + if (yield* ownsRpc(req.instanceId)) { + return yield* rpc.submit({ targetSessionId: req.instanceId, text: req.text }) + } + return yield* pty.submit(req) + }) + + const stop = (req: StopRequest) => + Effect.gen(function* () { + if (yield* ownsRpc(req.sessionId)) return yield* rpc.stop(req.sessionId) + return yield* pty.stop(req) + }) + + return { launch, submit, stop, ownsRpc } + }), +) diff --git a/src/main/services/TargetSessionManager.ts b/src/main/services/TargetSessionManager.ts index 8d345ff..658ca7b 100644 --- a/src/main/services/TargetSessionManager.ts +++ b/src/main/services/TargetSessionManager.ts @@ -39,6 +39,14 @@ import { export interface LaunchRequest { readonly provider: string readonly chatId: ChatId + /** + * Which live runtime backs this session. A launch-time *intent*, not a + * provider property — codex declares both `interactive` (PTY TUI) and + * `appServer` (rpc), so the same provider can be launched either way. Defaults + * to `pty` (unchanged behaviour); `rpc` routes to `RpcSessionManager` via + * `SessionRuntimeRouter`. + */ + readonly runtime?: "pty" | "rpc" readonly origin?: "manual" | "orchestrated" /** the orchestrator spawning this session (its target id), for an orchestrated * launch — persisted as the durable parent→child back-channel link. */ diff --git a/src/shared/rpc.ts b/src/shared/rpc.ts index fccd333..1aa9458 100644 --- a/src/shared/rpc.ts +++ b/src/shared/rpc.ts @@ -361,6 +361,9 @@ export const ArcRpcs = RpcGroup.make( payload: { provider: Schema.String, chatId: ChatId, + /** Which live runtime backs the session — `pty` (terminal TUI, default) or + * `rpc` (app-server). A launch-time intent, not a provider property. */ + runtime: Schema.optional(Schema.Literals(["pty", "rpc"])), /** Diff endpoint to run in — the worker writes here and hooks/file refs * resolve against it. Omit to use the chat's own workspace. */ workspaceId: Schema.optional(WorkspaceId), diff --git a/tests/live-pending-permission.test.ts b/tests/live-pending-permission.test.ts index e33de45..84f1eaf 100644 --- a/tests/live-pending-permission.test.ts +++ b/tests/live-pending-permission.test.ts @@ -8,6 +8,7 @@ import { ActivityEventServiceLive } from "../src/main/services/ActivityEventServ import { ChatServiceLive } from "../src/main/services/ChatService.js" import { LocalModelServiceLive } from "../src/main/services/LocalModelService.js" import { TargetSessionManager } from "../src/main/services/TargetSessionManager.js" +import { SessionRuntimeRouter } from "../src/main/services/SessionRuntimeRouter.js" import { ChatMessageService, ChatMessageServiceLive } from "../src/main/services/ChatMessageService.js" import type { HookSignal } from "../src/main/hooks/signals.js" import { toSignal } from "../src/main/hooks/signals.js" @@ -20,6 +21,18 @@ const TARGET = "target_1" // The live-pending-permission flag never drives a PTY, so TargetSessionManager is // only *acquired* by the layer — a stub keeps node-pty/socket setup out of the // test while satisfying the dependency. Any method actually called is a bug. +// sendPrompt is not exercised here, so the router is only acquired — a stub +// that dies on use keeps its full dependency graph out of the test. +const stubRouter = Layer.succeed( + SessionRuntimeRouter, + SessionRuntimeRouter.of({ + launch: () => Effect.die("SessionRuntimeRouter.launch is unused in this test"), + submit: () => Effect.die("SessionRuntimeRouter.submit is unused in this test"), + stop: () => Effect.succeed({ stopped: false }), + ownsRpc: () => Effect.succeed(false), + }), +) + const stubSessions = Layer.succeed( TargetSessionManager, TargetSessionManager.of({ @@ -50,6 +63,7 @@ const run = async ( ChatServiceLive.pipe(Layer.provide(arc)), LocalModelServiceLive, stubSessions, + stubRouter, ) const runtime = ManagedRuntime.make(ChatMessageServiceLive.pipe(Layer.provideMerge(deps))) try { diff --git a/tests/rpc-session-manager.test.ts b/tests/rpc-session-manager.test.ts index bd01aee..0b7a2d0 100644 --- a/tests/rpc-session-manager.test.ts +++ b/tests/rpc-session-manager.test.ts @@ -70,7 +70,9 @@ describe("RpcSessionManager", () => { ) const res = yield* manager.submit({ targetSessionId: "target_1", text: "hi" }) - expect(res).toEqual({ accepted: true, status: "completed" }) + expect(res.accepted).toBe(true) + expect(res.status).toBe("completed") + expect(res.rows?.session.nativeSessionId).toBe("thr_mgr") // The turn landed in the shared store (indistinguishable from a scraped session). const stored = yield* store.listSessions() diff --git a/tests/session-runtime-router.test.ts b/tests/session-runtime-router.test.ts new file mode 100644 index 0000000..9c46f65 --- /dev/null +++ b/tests/session-runtime-router.test.ts @@ -0,0 +1,108 @@ +import { EventEmitter } from "node:events" +import { Effect, Layer, ManagedRuntime, Stream } from "effect" +import { describe, expect, it } from "vitest" +import { ArcStoreLive } from "../src/main/db/store.js" +import { sqliteLayer } from "../src/main/db/sqlite.js" +import { IngestStoreLive } from "../src/main/ingest/db/store.js" +import { CodexDriverRegistryLive } from "../src/main/services/CodexDriverRegistry.js" +import { ChatServiceLive } from "../src/main/services/ChatService.js" +import { ProviderRegistryLive } from "../src/main/services/ProviderRegistry.js" +import { RpcSessionManager, RpcSessionManagerLive } from "../src/main/services/RpcSessionManager.js" +import { + SessionRuntimeRouter, + SessionRuntimeRouterLive, +} from "../src/main/services/SessionRuntimeRouter.js" +import { TargetSessionManager } from "../src/main/services/TargetSessionManager.js" +import { WorkspaceServiceLive } from "../src/main/services/WorkspaceService.js" + +const 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 === 'thread/start') send({ id: m.id, result: { thread: { id: 'thr_router' } } }) + else if (m.method === 'turn/start') { + send({ id: m.id, result: { turn: { id: 't1', status: 'inProgress' } } }) + send({ method: 'item/completed', params: { item: { type: 'agentMessage', id: 'a1', text: 'ok', phase: 'final_answer' } } }) + send({ method: 'turn/completed', params: { turn: { status: 'completed' } } }) + } +}) +` + +// PTY manager isn't launched here — only its submit is exercised (the non-rpc +// dispatch branch), which reports the session isn't attached. +const stubPty = Layer.succeed( + TargetSessionManager, + TargetSessionManager.of({ + list: Effect.succeed([]), + changes: Stream.empty, + launch: () => Effect.die("pty launch unused"), + resume: () => Effect.die("pty resume unused"), + stop: () => Effect.succeed({ stopped: false }), + bindNative: () => Effect.void, + submit: () => Effect.succeed({ accepted: false }), + write: () => Effect.void, + resize: () => Effect.void, + events: new EventEmitter(), + }), +) + +const run = (program: Effect.Effect): Promise => { + const sql = sqliteLayer(":memory:") + const arc = ArcStoreLive.pipe(Layer.provide(sql)) + const ingest = IngestStoreLive.pipe(Layer.provide(sql)) + const rpc = RpcSessionManagerLive.pipe(Layer.provide(CodexDriverRegistryLive), Layer.provide(ingest)) + const base = Layer.mergeAll( + arc, + rpc, + ProviderRegistryLive, + WorkspaceServiceLive.pipe(Layer.provide(arc)), + ChatServiceLive.pipe(Layer.provide(arc)), + stubPty, + ) + const runtime = ManagedRuntime.make(SessionRuntimeRouterLive.pipe(Layer.provideMerge(base))) + return runtime.runPromise(program).finally(() => runtime.dispose()) +} + +describe("SessionRuntimeRouter dispatch", () => { + it( + "routes submit/stop/ownsRpc by which manager owns the session", + () => + run( + Effect.gen(function* () { + const router = yield* SessionRuntimeRouter + const rpc = yield* RpcSessionManager + + // Launch an rpc session directly, then drive it through the router. + yield* rpc.launch({ + chatId: "chat_1", + targetSessionId: "t1", + cwd: process.cwd(), + command: process.execPath, + args: ["-e", PEER], + }) + + expect(yield* router.ownsRpc("t1")).toBe(true) + expect(yield* router.ownsRpc("unknown")).toBe(false) + + // Owned by rpc → routes there, and carries the turn rows for projection. + const rpcTurn = yield* router.submit({ instanceId: "t1", text: "hi" }) + expect(rpcTurn.accepted).toBe(true) + expect(rpcTurn.rows?.session.nativeSessionId).toBe("thr_router") + + // Not owned by rpc → falls through to the PTY manager (stub: not attached). + const ptyTurn = yield* router.submit({ instanceId: "unknown", text: "hi" }) + expect(ptyTurn.accepted).toBe(false) + expect("rows" in ptyTurn).toBe(false) + + // Stop routes to rpc and the session leaves the aggregate. + expect(yield* router.stop({ sessionId: "t1" })).toEqual({ stopped: true }) + expect(yield* router.ownsRpc("t1")).toBe(false) + }), + ), + 15000, + ) +}) From b2ad8e3977acf5164b876603390d76454b8d1c2d Mon Sep 17 00:00:00 2001 From: Tim Hanlon Date: Sun, 5 Jul 2026 06:14:12 +1000 Subject: [PATCH 09/26] feat(chat): app-server approval cards + rpc launch entry in the renderer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pty-less answer surface: an app-server session has no terminal to defer approvals to, so Arc owns the interaction. - appServerApprovalsAtom (live WatchAppServerApprovals stream) + answerAppServerApprovalAtom mutation. - ChatApprovals: chat-scoped container filtering the live approvals to the current chat, rendering AppServerApproval cards, echoing a decision's payload back verbatim, with optimistic per-request in-flight disabling. Mounted above the composer in UnifiedChatPane. - Launch entry: LaunchableProvider carries runtime + label; a provider declaring both interactive and appServer surfaces two options (codex / codex . app-server). onLaunch branches — rpc fires LaunchTarget directly (no pane, no xterm measure), pty keeps the pane/measure path. No terminal pane for rpc falls out for free. - ChatApprovals story (Pending filters an other-chat approval out; Empty renders null) — verified in Storybook. --- src/renderer/src/App.tsx | 26 +++++- src/renderer/src/atoms.ts | 7 ++ .../src/chat/ChatApprovals.stories.tsx | 84 +++++++++++++++++++ src/renderer/src/chat/ChatApprovals.tsx | 58 +++++++++++++ src/renderer/src/chat/UnifiedChatPane.tsx | 29 +++++-- 5 files changed, 192 insertions(+), 12 deletions(-) create mode 100644 src/renderer/src/chat/ChatApprovals.stories.tsx create mode 100644 src/renderer/src/chat/ChatApprovals.tsx diff --git a/src/renderer/src/App.tsx b/src/renderer/src/App.tsx index 34f32f1..aff89e9 100644 --- a/src/renderer/src/App.tsx +++ b/src/renderer/src/App.tsx @@ -17,7 +17,7 @@ import type { ChatId, PaneId, TargetId, WorkspaceId } from "../../shared/ids.js" import { ArcSidebarTree } from "./sidebar/ArcSidebarTree.js" import { TargetSessionPane } from "./chat/TargetSessionPane.js" import { sync as syncTerminals } from "./terminal/terminalRegistry.js" -import { UnifiedChatPane, type ChatPaneHandle } from "./chat/UnifiedChatPane.js" +import { UnifiedChatPane, type ChatPaneHandle, type LaunchableProvider } from "./chat/UnifiedChatPane.js" import { WorkPane, type WorkPaneHandle } from "./work/WorkPane.js" import { GitPane } from "./git/GitPane.js" import { GitPrefetch } from "./git/GitPrefetch.js" @@ -89,7 +89,16 @@ export function App(): JSX.Element { () => deriveShellViewModel(shell.state, { workspaces, chats, sessions }), [shell.state, workspaces, chats, sessions], ) - const interactiveProviders = providers.filter((p) => p.interactive) + // One launch option per (provider, runtime): a provider that declares both an + // `interactive` (PTY TUI) and an `appServer` (codex app-server) capability + // surfaces both, individually labelled, so the user picks the runtime at launch. + const launchableProviders: ReadonlyArray = providers.flatMap((p) => { + const options: Array = [] + if (p.interactive) options.push({ kind: p.kind, displayName: p.displayName, runtime: "pty", label: p.kind }) + if (p.appServer) + options.push({ kind: p.kind, displayName: p.displayName, runtime: "rpc", label: `${p.kind} · app-server` }) + return options + }) const centerView = center.surface.kind === "work" ? "work" : "chat" const rightView = right.surface.kind === "git" ? "git" : "terminal" @@ -224,7 +233,16 @@ export function App(): JSX.Element { shell.actions.selectChat(workspaceId, chatId) } - const onLaunch = (provider: string, chatId: ChatId): void => { + const onLaunch = (provider: string, chatId: ChatId, runtime: "pty" | "rpc"): void => { + // An rpc (app-server) session has no terminal, so it skips the pane/measure + // machinery entirely: fire `LaunchTarget` directly and let the session appear + // through `WatchSessions`. Its transcript, composer, and approval cards render + // from that projection — no xterm to bind. PTY launches still go through the + // shell machine, which creates a pane and defers the RPC until xterm measures. + if (runtime === "rpc") { + void runLaunchTarget({ payload: { provider, chatId, runtime: "rpc" } }) + return + } shell.actions.launchTarget(provider, chatId) } @@ -355,7 +373,7 @@ export function App(): JSX.Element { liveStateById={liveStateById} activeSessionId={vm.activeSessionId} sessionCount={vm.sessionCount} - providers={interactiveProviders} + providers={launchableProviders} onLaunch={onLaunch} onFocusSession={focusSession} onRenameChat={renameChat} diff --git a/src/renderer/src/atoms.ts b/src/renderer/src/atoms.ts index 07bdd43..b88cb4d 100644 --- a/src/renderer/src/atoms.ts +++ b/src/renderer/src/atoms.ts @@ -50,6 +50,8 @@ export const launchTargetAtom = ArcRpcAtomClient.mutation("LaunchTarget") export const resumeTargetAtom = ArcRpcAtomClient.mutation("ResumeTarget") export const stopTargetAtom = ArcRpcAtomClient.mutation("StopTarget") export const listWorkspaceFilesAtom = ArcRpcAtomClient.mutation("ListWorkspaceFiles") +/** Answer a codex app-server approval by echoing a decision's `payload` back. */ +export const answerAppServerApprovalAtom = ArcRpcAtomClient.mutation("AnswerAppServerApproval") /** * One subscription per chat change feed (`WatchChatMessageChanges` / @@ -119,6 +121,11 @@ export const chatsAtom = ArcRpcAtomClient.runtime.atom( export const liveTargetStatesAtom = ArcRpcAtomClient.runtime.atom( Stream.unwrap(ArcRpcAtomClient.use((client) => Effect.succeed(client("WatchLiveTargetStates", undefined)))), ) +/** Outstanding codex app-server approvals, live — the inline-card answer surface. + * Ephemeral (never persisted); the driver mirrors its in-memory state here. */ +export const appServerApprovalsAtom = ArcRpcAtomClient.runtime.atom( + Stream.unwrap(ArcRpcAtomClient.use((client) => Effect.succeed(client("WatchAppServerApprovals", undefined)))), +) /** Per-chat message-change signal, derived from the shared {@link chatMessageCountsAtom} * as the bare per-chat count — so a change for another chat is deduped away by the diff --git a/src/renderer/src/chat/ChatApprovals.stories.tsx b/src/renderer/src/chat/ChatApprovals.stories.tsx new file mode 100644 index 0000000..ea38af9 --- /dev/null +++ b/src/renderer/src/chat/ChatApprovals.stories.tsx @@ -0,0 +1,84 @@ +import type { ReactNode } from "react" +import { RegistryProvider } from "@effect/atom-react" +import * as AsyncResult from "effect/unstable/reactivity/AsyncResult" +import type { AppServerApproval } from "../../../shared/codex-approval.js" +import type { ChatId } from "../../../shared/ids.js" +import { appServerApprovalsAtom } from "../atoms.js" +import { ChatApprovals } from "./ChatApprovals.js" + +export default { + title: "Chat / ChatApprovals", +} + +const CHAT = "chat_demo" as ChatId +const OTHER = "chat_other" as ChatId + +const shellApproval: AppServerApproval = { + chatId: CHAT, + targetSessionId: "target_demo", + requestId: 7, + approvalId: "appr_1", + itemId: "item_cmd_1", + command: "rm -rf node_modules && pnpm install", + decisions: [ + { label: "accept", payload: JSON.stringify({ decision: "accept" }) }, + { label: "acceptForSession", payload: JSON.stringify({ decision: "acceptForSession" }) }, + { label: "cancel", payload: JSON.stringify({ decision: "cancel" }) }, + ], +} + +const patchApproval: AppServerApproval = { + chatId: CHAT, + targetSessionId: "target_demo", + requestId: "req-abc", + approvalId: null, + itemId: "item_patch_1", + command: null, + decisions: [ + { label: "accept", payload: JSON.stringify({ decision: "accept" }) }, + { label: "cancel", payload: JSON.stringify({ decision: "cancel" }) }, + ], +} + +// An approval for a different chat — proves the container filters by chatId. +const otherChatApproval: AppServerApproval = { + ...shellApproval, + chatId: OTHER, + requestId: 99, + command: "should not appear", +} + +function Frame({ + approvals, + children, +}: { + readonly approvals: ReadonlyArray + readonly children: ReactNode +}) { + return ( + +
{children}
+ + ) +} + +/** A shell-command approval and a pty-less file-patch approval, both for this chat. */ +export function Pending() { + return ( + + + + ) +} + +/** No approvals for this chat → the container renders nothing (the other chat's is filtered out). */ +export function Empty() { + return ( + +
+ (nothing renders below — no approvals for this chat) +
+ + + ) +} diff --git a/src/renderer/src/chat/ChatApprovals.tsx b/src/renderer/src/chat/ChatApprovals.tsx new file mode 100644 index 0000000..0b04a34 --- /dev/null +++ b/src/renderer/src/chat/ChatApprovals.tsx @@ -0,0 +1,58 @@ +import { type JSX, useState } from "react" +import { useAtomSet, useAtomValue } from "@effect/atom-react" +import type { ChatId } from "../../../shared/ids.js" +import { answerAppServerApprovalAtom, appServerApprovalsAtom, successList } from "../atoms.js" +import { AppServerApproval } from "./AppServerApproval.js" + +export interface ChatApprovalsProps { + readonly chatId: ChatId +} + +/** + * The chat-scoped stack of codex app-server approvals awaiting an answer. An + * app-server session has no PTY to defer the prompt to, so Arc owns the + * interaction: this subscribes to the live `WatchAppServerApprovals` stream, + * filters to this chat, and answers each by echoing a decision's `payload` back + * verbatim. The list is ephemeral — an answered approval drops off the stream, so + * there's nothing to clear here. + */ +export function ChatApprovals({ chatId }: ChatApprovalsProps): JSX.Element | null { + const approvals = successList(useAtomValue(appServerApprovalsAtom)).filter((a) => a.chatId === chatId) + const answer = useAtomSet(answerAppServerApprovalAtom, { mode: "promiseExit" }) + // requestIds currently in flight, keyed by their stringified id so the buttons + // disable optimistically before the approval falls off the stream. + const [answering, setAnswering] = useState>(new Set()) + + if (approvals.length === 0) return null + + return ( +
+ {approvals.map((approval) => { + const key = String(approval.requestId) + return ( + { + setAnswering((prev) => new Set(prev).add(key)) + void answer({ + payload: { + targetSessionId: approval.targetSessionId, + requestId: approval.requestId, + decisionPayload, + }, + }).finally(() => { + setAnswering((prev) => { + const next = new Set(prev) + next.delete(key) + return next + }) + }) + }} + /> + ) + })} +
+ ) +} diff --git a/src/renderer/src/chat/UnifiedChatPane.tsx b/src/renderer/src/chat/UnifiedChatPane.tsx index 5f0ae4e..d12d4bd 100644 --- a/src/renderer/src/chat/UnifiedChatPane.tsx +++ b/src/renderer/src/chat/UnifiedChatPane.tsx @@ -9,6 +9,7 @@ import { useChatMessages } from "./useChatMessages.js" import { useStreamingMessages } from "./useStreamingMessages.js" import { useChatWork } from "./useChatWork.js" import { ChatWork } from "./ChatWork.js" +import { ChatApprovals } from "./ChatApprovals.js" import { ChatComposer, type ComposerHandle } from "./composer/ChatComposer.js" import { ComposerTargetIndicators, formatAddressee } from "./composer/ComposerTargetIndicators.js" import { useReferenceTargets } from "./composer/useReferenceTargets.js" @@ -21,10 +22,17 @@ import { type TranscriptFilter, TranscriptFilterMenu, showsMessage } from "./Tra import { rpc } from "../rpc-client.js" import type { LiveStateById } from "../sidebar/grouping.js" -/** A provider the composer can launch a new target session against. */ +/** A provider+runtime the composer can launch a new target session against. A + * provider that declares both `interactive` and `appServer` appears twice — once + * per runtime — so the two are distinct, individually-labelled launch options. */ export interface LaunchableProvider { readonly kind: string readonly displayName: string + /** Which live runtime this option launches — `pty` (terminal TUI) or `rpc` + * (codex app-server, no terminal; answered via the inline approval cards). */ + readonly runtime: "pty" | "rpc" + /** Button text; distinguishes the app-server option from the pty one. */ + readonly label: string } export interface UnifiedChatPaneProps { @@ -36,7 +44,7 @@ export interface UnifiedChatPaneProps { readonly activeSessionId?: TargetId readonly sessionCount: number readonly providers: ReadonlyArray - readonly onLaunch: (provider: string, chatId: ChatId) => void + readonly onLaunch: (provider: string, chatId: ChatId, runtime: "pty" | "rpc") => void /** focus the live target session waiting on a pending question */ readonly onFocusSession: (sessionId: TargetId) => void readonly onRenameChat: (chatId: ChatId, title: string) => Promise @@ -315,13 +323,13 @@ export const UnifiedChatPane = forwardRef(
{launchableProviders.map((provider) => ( ))}
@@ -340,8 +348,11 @@ export const UnifiedChatPane = forwardRef( {providers.length > 0 ? (
{providers.map((provider) => ( - ))}
@@ -403,6 +414,8 @@ export const UnifiedChatPane = forwardRef( )} + +