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..6f07756 --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/driver.ts @@ -0,0 +1,243 @@ +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 { + /** The JSON-RPC request id — the reliable routing key for `answerApproval`. */ + readonly id: number | string + /** Codex's approval handle when present (commandExecution only) — display/correlation detail. */ + readonly approvalId: string | null + /** Links to the exact tool-call item already in the projection. */ + readonly itemId: string | null + readonly command: string | null + /** Server-supplied allowable answers; the UI offers these verbatim. */ + readonly availableDecisions: ReadonlyArray +} + +export interface TurnResult { + /** `completed` | `interrupted` | `failed` | `unknown`. */ + readonly status: string + /** The session's cumulative rows through this turn — ready for `IngestStore.replaceSession`. */ + readonly rows: ExtractedRows +} + +export interface CodexAppServerDriver { + readonly threadId: string + /** Send a user turn and resolve once it completes, with the session's cumulative rows + status. */ + readonly runTurn: (text: string) => Effect.Effect + /** The current outstanding approvals; subscribe via `.changes` for the live signal. */ + readonly pendingApprovals: SubscriptionRef.SubscriptionRef> + /** Answer an approval with a server-offered decision (pass it through verbatim). */ + readonly answerApproval: (id: number | string, decision: unknown) => Effect.Effect +} + +export interface CodexDriverOptions { + readonly cwd: string + readonly model?: string + 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 + /** Resume an existing thread by id (`thread/resume`) instead of starting a + * fresh one (`thread/start`). The id is a codex session id — the same one the + * PTY path resumes with — so a session started here rejoins under its old id. */ + readonly resumeThreadId?: string +} + +interface TurnOutcome { + readonly items: ReadonlyArray + readonly usage: ReadonlyArray + readonly status: string +} + +// What `runTurn` awaits: a completed turn, or an `aborted` signal offered when +// the notification stream ends (the process died mid-turn) so the awaiting +// `Queue.take` fails fast instead of blocking forever on a completion that will +// never arrive. +type TurnSignal = { readonly kind: "completed"; readonly outcome: TurnOutcome } | { readonly kind: "aborted" } + +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 (fresh) or thread/resume + // (rejoin an existing thread by id). Both answer with `{ thread: { id } }`, so + // the same decoder extracts the thread id either way. + 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 threadConfig = { + cwd: options.cwd, + model: options.model, + sandbox: options.sandbox, + approvalPolicy: options.approvalPolicy, + } + const [method, params] = options.resumeThreadId + ? (["thread/resume", { threadId: options.resumeThreadId, ...threadConfig }] as const) + : (["thread/start", threadConfig] as const) + const started = yield* transport.request(method, params).pipe(wrap(`${method} failed`)) + const thread = decodeThreadStart(started) + if (Option.isNone(thread)) { + return yield* Effect.fail( + new CodexDriverError({ message: `${method} 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. 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. + const items: Array = [] + const usage: Array = [] + yield* transport.notifications.pipe( + Stream.runForEach((notification) => + Effect.gen(function* () { + switch (notification.method) { + 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" + yield* Queue.offer(turnOutcomes, { + kind: "completed", + outcome: { items: items.slice(), usage: usage.slice(), status }, + }) + break + } + case "serverRequest/resolved": { + // Guard the id: an absent/malformed field would leave `requestId` + // undefined, and `approval.id !== undefined` matches every pending + // approval — so the filter would clear nothing and the card would + // stick until scope close. Only act on a real id. + const requestId = obj(notification.params)?.["requestId"] + if (typeof requestId !== "number" && typeof requestId !== "string") break + yield* SubscriptionRef.update(pendingApprovals, (list) => + list.filter((approval) => approval.id !== requestId), + ) + break + } + } + }), + ), + // The transport shuts its queues on process exit, which *interrupts* this + // fold (a blocked `Queue.take` on a shut-down queue is interrupted, not a + // graceful end) — so use `ensuring`, not `andThen`, to signal an in-flight + // turn. Its `turn/completed` will never arrive; the waiting `runTurn` must + // fail instead of blocking on the outcome queue forever. + Effect.ensuring(Queue.offer(turnOutcomes, { kind: "aborted" as const })), + 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, + approvalId: params.approvalId ?? null, + 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 signal = yield* Queue.take(turnOutcomes) + if (signal.kind === "aborted") { + return yield* Effect.fail( + new CodexDriverError({ message: "codex app-server exited before the turn completed" }), + ) + } + const outcome = signal.outcome + 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/launch.ts b/src/main/ingest/providers/codex-appserver/launch.ts new file mode 100644 index 0000000..bf043a9 --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/launch.ts @@ -0,0 +1,61 @@ +import { Effect, type Scope } from "effect" +import type { AppServerCapability } from "../../../../shared/provider.js" +import { IngestStore } from "../../db/store.js" +import { + type CodexAppServerDriver, + CodexDriverError, + type CodexDriverOptions, + makeCodexAppServerDriver, +} from "./driver.js" + +export interface CodexLaunchParams { + readonly cwd: string + readonly model?: string + readonly sandbox?: CodexDriverOptions["sandbox"] + readonly approvalPolicy?: CodexDriverOptions["approvalPolicy"] + readonly env?: Record + readonly clientName?: string + /** Rejoin an existing thread by id (`thread/resume`) instead of starting fresh. */ + readonly resumeThreadId?: string +} + +/** + * 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, + resumeThreadId: params.resumeThreadId, + }) + + 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/src/main/ingest/providers/codex-appserver/normalize.ts b/src/main/ingest/providers/codex-appserver/normalize.ts new file mode 100644 index 0000000..5d60447 --- /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, 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..3e003a9 --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/protocol.ts @@ -0,0 +1,119 @@ +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({ + /** + * 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), + 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..bc28fe0 --- /dev/null +++ b/src/main/ingest/providers/codex-appserver/transport.ts @@ -0,0 +1,230 @@ +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 + // Set once the child exits (see the exit branch of `route`). A request made + // *after* the exit event would otherwise register a Deferred that no future + // exit can fail — so every request/notify/respond checks this first and fails + // fast instead of hanging on a dead server. + let closed: AppServerTransportError | null = null + + // 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" })) + // Mark the transport closed and fail every in-flight request. Called both + // when the child's exit is *routed* (process crash while alive) and as a scope + // finalizer (intentional `stop`): on scope close the drain fiber is + // interrupted before the child is killed, so the exit event is never routed — + // without the finalizer a request blocked on `Deferred.await` would hang. + // Idempotent: whichever runs first clears `pending`, the other finds it empty. + const failAllPending = () => + Effect.gen(function* () { + closed ??= new AppServerTransportError({ message: `${options.command} exited` }) + for (const [, deferred] of pending) { + yield* Deferred.fail(deferred, closed) + } + pending.clear() + }) + + // Release the readline interface + fail pending requests at scope close, + // alongside the child kill the acquireRelease above registers. + yield* Effect.addFinalizer(() => Effect.sync(() => lines.close())) + yield* Effect.addFinalizer(() => failAllPending()) + + const route = (event: Inbound): Effect.Effect => + Effect.gen(function* () { + if (event.kind === "exit") { + // Fail every in-flight request, mark closed so later calls fail fast, and + // end both output streams so nobody hangs. + yield* failAllPending() + 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* () { + if (closed) return yield* Effect.fail(closed) + const id = nextId++ + const deferred = yield* Deferred.make() + pending.set(id, deferred) + // `Deferred.make` yields, so an exit could have drained `pending` between + // the guard above and here — re-check rather than await a dead deferred. + if (closed) { + pending.delete(id) + return yield* Effect.fail(closed) + } + 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 => + closed ? Effect.fail(closed) : writeLine({ method, params: params ?? {} }) + + const respond = (id: IdKey, result: unknown): Effect.Effect => + closed ? Effect.fail(closed) : writeLine({ id, result }) + + return { + request, + notify, + respond, + notifications: Stream.fromQueue(notifications), + serverRequests: Stream.fromQueue(serverRequests), + } satisfies AppServerTransport + }) diff --git a/src/main/rpc.ts b/src/main/rpc.ts index 692ee07..72a3cce 100644 --- a/src/main/rpc.ts +++ b/src/main/rpc.ts @@ -14,6 +14,9 @@ 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 { 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" import { ReadService } from "./read/service.js" @@ -121,17 +124,19 @@ export const ArcRpcHandlersLive = ArcRpcs.toLayer( ), ListPresets: svc("ListPresets", PresetRegistry, (_) => _.list), ListInstances: svc("ListInstances", TargetSessionManager, (_) => _.list), - ListSessions: svc("ListSessions", TargetSessionManager, (_) => _.list), + ListSessions: svc("ListSessions", SessionRuntimeRouter, (_) => _.sessions), // Streaming handlers: return each service's reactive `changes` stream // directly. The streams replay (or derive) their current value on subscribe, // so a fresh client gets the snapshot then live updates — no separate boot // push. Their error channel is `never`, so the `rpcEffect` request/error // wrapper (Effect-shaped) doesn't apply here. - WatchSessions: () => Stream.unwrap(Effect.map(TargetSessionManager, (_) => _.changes)), + WatchSessions: () => Stream.unwrap(Effect.map(SessionRuntimeRouter, (_) => _.changes)), WatchChats: () => Stream.unwrap(Effect.map(ChatService, (_) => _.changes)), 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 +150,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)), @@ -158,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)), - ResumeTarget: svc("ResumeTarget", TargetSessionManager, (_, req) => _.resume(req)), - StopTarget: svc("StopTarget", TargetSessionManager, (_, req) => _.stop(req)), + LaunchTarget: svc("LaunchTarget", SessionRuntimeRouter, (_, req) => _.launch(req)), + ResumeTarget: svc("ResumeTarget", SessionRuntimeRouter, (_, req) => _.resume(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 471d1c2..d394427 100644 --- a/src/main/runtime.ts +++ b/src/main/runtime.ts @@ -2,6 +2,9 @@ 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 { 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" @@ -43,7 +46,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 @@ -61,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), @@ -74,7 +104,8 @@ const ChatMessagesLive = ChatMessageServiceLive.pipe( // requests + hook turn lifecycle. Pure read model — it shares the one // TargetSessionManager / ChatMessageService instance (memoized by reference). const LiveTargetStatesLive = LiveTargetStateServiceLive.pipe( - Layer.provide(SessionsLive), + Layer.provide(SessionRouterLive), + Layer.provide(RpcSessionsLive), Layer.provide(ChatMessagesLive), ) // Deliver messages into a running target session: queues on the inbox, pastes @@ -97,6 +128,8 @@ const DomainServiceLayers = [ ChatMessagesLive, LiveTargetStatesLive, TargetInboxLive, + RpcSessionsLive, + SessionRouterLive, SessionsLive, ] as const diff --git a/src/main/services/ChatMessageService.ts b/src/main/services/ChatMessageService.ts index 5766b96..07486e2 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 @@ -440,9 +448,21 @@ export const ChatMessageServiceLive = Layer.effect( if (!ok) { return yield* Effect.fail(arcRequestError("Failed to record prompt in chat transcript")) } + // Publish now so the bubble shows immediately — an rpc turn holds this + // handler open until it completes, so the end-of-send publish would + // otherwise delay the user's own message until codex replies. + yield* PubSub.publish(updates, { chatId: req.chatId }) } - 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) => @@ -450,6 +470,9 @@ export const ChatMessageServiceLive = Layer.effect( ), Effect.ignore, ) + // The optimistic echo already published to show the bubble; publish the + // rollback too, or the renderer keeps showing a message whose row is gone. + yield* PubSub.publish(updates, { chatId: req.chatId }) return yield* Effect.fail( arcRequestError( `Target session "${stored.provider}" is not attached — prompt was not delivered`, @@ -457,6 +480,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/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/src/main/services/LiveTargetStateService.ts b/src/main/services/LiveTargetStateService.ts index 67a7f8d..13bcc61 100644 --- a/src/main/services/LiveTargetStateService.ts +++ b/src/main/services/LiveTargetStateService.ts @@ -2,7 +2,8 @@ import { Context, Effect, Layer, Stream, SubscriptionRef } from "effect" import type { TargetSession } from "../../shared/instance.js" import type { LiveTargetActivity, LiveTargetState } from "../../shared/live-target-state.js" import type { PendingRequest } from "../../shared/chat-request.js" -import { TargetSessionManager } from "./TargetSessionManager.js" +import { SessionRuntimeRouter } from "./SessionRuntimeRouter.js" +import { RpcSessionManager } from "./RpcSessionManager.js" import { ChatMessageService } from "./ChatMessageService.js" /** @@ -13,14 +14,15 @@ import { ChatMessageService } from "./ChatMessageService.js" * {@link TargetSession.state} (which is lifecycle, not activity — see * shared/live-target-state.ts): * - * 1. PTY ownership + lifecycle — `TargetSessionManager.changes` carries - * `attached` (this process holds the live PTY) and the persisted `state` - * (only `exited` is read here). + * 1. Ownership + lifecycle — the unified `SessionRuntimeRouter.changes` (PTY + + * rpc) carries `attached` (a live PTY, or a running app-server session) and + * the persisted `state` (only `exited` is read here). * 2. Pending questions/permissions — `ChatMessageService.listPending`, the * attention signal (a persisted question row or an in-memory permission). - * 3. The hook turn lifecycle — `noteTurn`, fed by the controller from the hook - * signal stream (UserPromptSubmit opens a turn, Stop closes it). This is the - * only input that distinguishes "actively generating" from "attached idle". + * 3. Turn lifecycle — for PTY, `noteTurn` fed by the controller from the hook + * stream (UserPromptSubmit opens, Stop closes); for rpc, the + * `RpcSessionManager.generating` marker set around each turn. Either + * distinguishes "actively generating" from "attached idle". * * The open-turn set is ephemeral (a SubscriptionRef, never persisted): it is * rebuilt from the live hook stream each process, the same way the PTY map is. @@ -63,18 +65,26 @@ export const deriveActivity = ( export const LiveTargetStateServiceLive = Layer.effect( LiveTargetStateService, Effect.gen(function* () { - const sessions = yield* TargetSessionManager + // Read the *unified* session list (PTY + rpc) via the router, not just the PTY + // manager — an rpc app-server session lives under RpcSessionManager and would + // otherwise be invisible here (no live state at all). + const router = yield* SessionRuntimeRouter + const rpc = yield* RpcSessionManager const chatMessages = yield* ChatMessageService const openTurns = yield* SubscriptionRef.make(new Set()) const derive = Effect.gen(function* () { - const sessionList = yield* sessions.list + const sessionList = yield* router.sessions // A pending-list read failure must not blank every session's status, so it // degrades to "no pending" rather than failing the projection. const pending = yield* chatMessages.listPending.pipe( Effect.orElseSucceed(() => [] as ReadonlyArray), ) - const turns = yield* SubscriptionRef.get(openTurns) + // Both turn signals mean "generating": hook-driven (PTY) and the rpc turn + // marker (app-server sessions have no hook stream). + const hookTurns = yield* SubscriptionRef.get(openTurns) + const rpcTurns = yield* rpc.generating + const turns = new Set([...hookTurns, ...rpcTurns]) const pendingByTarget = new Map(pending.map((p) => [p.targetSessionId, p.kind])) return sessionList.map( (session): LiveTargetState => ({ @@ -92,7 +102,12 @@ export const LiveTargetStateServiceLive = Layer.effect( const tick = (stream: Stream.Stream): Stream.Stream => Stream.map(stream, () => undefined) const changes = Stream.mergeAll( - [tick(sessions.changes), tick(chatMessages.changes), tick(SubscriptionRef.changes(openTurns))], + [ + tick(router.changes), + tick(chatMessages.changes), + tick(SubscriptionRef.changes(openTurns)), + tick(rpc.generatingChanges), + ], { concurrency: "unbounded" }, ).pipe(Stream.mapEffect(() => derive)) 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/main/services/RpcSessionManager.ts b/src/main/services/RpcSessionManager.ts new file mode 100644 index 0000000..f83209e --- /dev/null +++ b/src/main/services/RpcSessionManager.ts @@ -0,0 +1,218 @@ +import { Context, Effect, Exit, Layer, Scope, Stream, SubscriptionRef } from "effect" +import * as Semaphore from "effect/Semaphore" +import type { ChatId, TargetId } from "../../shared/ids.js" +import type { TargetOrigin, TargetSession } from "../../shared/instance.js" +import type { ExtractedRows } from "../ingest/db/schema.js" +import type { + CodexAppServerDriver, + CodexDriverError, + CodexDriverOptions, +} from "../ingest/providers/codex-appserver/driver.js" +import { 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: ChatId + readonly targetSessionId: TargetId + /** Provider kind and launch identity — this manager owns the `TargetSession` + * state (so it surfaces in the unified `WatchSessions`), built from these. */ + readonly provider: string + readonly origin?: TargetOrigin + readonly startedAt: string + readonly cwd: string + readonly command: string + readonly args: ReadonlyArray + readonly model?: string + readonly sandbox?: CodexDriverOptions["sandbox"] + readonly approvalPolicy?: CodexDriverOptions["approvalPolicy"] + /** Rejoin an existing thread by id (`thread/resume`) instead of starting fresh. */ + readonly resumeThreadId?: string +} + +/** + * 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, register its approvals. Returns the `TargetSession` with its + * `nativeSessionId` bound to the driver's thread id — the caller persists that + * so the timeline projection can resolve the target by (provider, native id). */ + readonly launch: (req: RpcLaunchRequest) => Effect.Effect + /** 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; 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. */ + readonly list: Effect.Effect> + /** The live `TargetSession` states (full objects) — the router merges these + * into the unified sessions view so rpc sessions surface in the renderer. */ + readonly sessions: Effect.Effect> + /** Reactive view of {@link sessions}: current value, then every change. */ + readonly changes: Stream.Stream> + /** Target ids with a turn in flight — feeds the live "generating" activity + * (an rpc session has no hook stream, so this is its turn-lifecycle signal). */ + readonly generating: Effect.Effect> + /** Reactive view of {@link generating}: current value, then every change. */ + readonly generatingChanges: Stream.Stream> + } +>()("arcwork/RpcSessionManager") {} + +interface LiveRpcSession { + readonly driver: CodexAppServerDriver + readonly scope: Scope.Closeable + /** Serializes turns for this session: the driver folds one turn at a time + * (unkeyed turn outcomes), so concurrent submits would misattribute completions. */ + readonly sem: Semaphore.Semaphore +} + +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() + // Serialize launches so the idempotency check + spawn + `sessions.set` are + // atomic: two concurrent resumes of the same id (a double-clicked "resume") + // would otherwise both pass the guard, both spawn a driver, and the second + // set would orphan the first scope (a leaked process). Mirrors the PTY + // manager's `launchLock`. + const launchLock = yield* Semaphore.make(1) + // Observable `TargetSession` state, so the router can fold these into the + // unified `WatchSessions` stream — an rpc session has no PTY registry to + // appear in, so it lives here. + const store = yield* SubscriptionRef.make>(new Map()) + // Target ids with a turn in flight — the rpc equivalent of the hook-driven + // open-turn set, read by LiveTargetStateService to paint "generating". + const generating = yield* SubscriptionRef.make>(new Set()) + const markGenerating = (id: string, on: boolean) => + SubscriptionRef.update(generating, (s) => { + if (on === s.has(id)) return s + const next = new Set(s) + if (on) next.add(id) + else next.delete(id) + return next + }) + + const launch = (req: RpcLaunchRequest): Effect.Effect => + launchLock.withPermits(1)( + Effect.gen(function* () { + const current = yield* SubscriptionRef.get(store) + const already = current.get(req.targetSessionId) + if (already && sessions.has(req.targetSessionId)) return already // 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, + resumeThreadId: req.resumeThreadId, + }, + ) + 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)), + ) + const sem = yield* Semaphore.make(1) + sessions.set(req.targetSessionId, { driver, scope, sem }) + const session: TargetSession = { + _tag: "TargetSession", + id: req.targetSessionId, + provider: req.provider, + origin: req.origin ?? "manual", + chatId: req.chatId, + cwd: req.cwd, + nativeSessionId: driver.threadId, + attached: true, + runtime: "rpc", + state: "running", + startedAt: req.startedAt, + } + yield* SubscriptionRef.update(store, (m) => new Map(m).set(session.id, session)) + return session + }), + ) + + 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 } + // One turn at a time per session (the driver's turn outcomes are unkeyed), + // and mark the session "generating" for the duration so the composer/sidebar + // paint live activity the way a PTY turn does off its hooks. + const result = yield* session.sem.withPermits(1)( + Effect.acquireUseRelease( + markGenerating(req.targetSessionId, true), + () => session.driver.runTurn(req.text), + () => markGenerating(req.targetSessionId, false), + ), + ) + return { accepted: true as const, status: result.status, rows: result.rows } + }) + + const stop = (targetSessionId: string) => + Effect.gen(function* () { + const session = sessions.get(targetSessionId) + if (!session) return { stopped: false } + sessions.delete(targetSessionId) + yield* SubscriptionRef.update(store, (m) => { + const next = new Map(m) + next.delete(targetSessionId) + return next + }) + yield* markGenerating(targetSessionId, false) + yield* Scope.close(session.scope, Exit.void) + return { stopped: true } + }) + + return { + launch, + submit, + stop, + list: Effect.sync(() => [...sessions.keys()]), + sessions: SubscriptionRef.get(store).pipe(Effect.map((m) => [...m.values()])), + changes: Stream.map(SubscriptionRef.changes(store), (m) => [...m.values()]), + generating: SubscriptionRef.get(generating).pipe(Effect.map((s) => [...s])), + generatingChanges: Stream.map(SubscriptionRef.changes(generating), (s) => [...s]), + } + }), +) diff --git a/src/main/services/SessionRuntimeRouter.ts b/src/main/services/SessionRuntimeRouter.ts new file mode 100644 index 0000000..ea3338f --- /dev/null +++ b/src/main/services/SessionRuntimeRouter.ts @@ -0,0 +1,247 @@ +import { Clock, Context, Effect, Layer, Stream } 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 { restoredSessionFromRow } from "./target-session/boot-restore.js" +import { presentTargetSession } from "./target-session/provider-args.js" +import { + type LaunchRequest, + type ResumeRequest, + 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 + /** Resume a session, into `pty` (default) or `rpc` (rejoin the app-server + * thread by its persisted native id) per `req.runtime`. */ + readonly resume: ( + req: ResumeRequest, + ) => Effect.Effect + /** 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 + /** The unified session list — PTY + rpc — backing `ListSessions`. */ + readonly sessions: Effect.Effect> + /** Reactive union of both managers' change streams, backing `WatchSessions`. */ + readonly changes: Stream.Stream> + } +>()("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)) + + // Spawn the driver for an rpc launch, then persist its TargetSession row (with + // the thread id as `nativeSessionId`) so `ingestArtifactSession` can resolve + // the target by (provider, native id) 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 + + // The manager owns the live TargetSession (with the thread id bound), so it + // surfaces in the unified `sessions`/`changes` below. We persist that row — + // `nativeSessionId` = the thread id — so `ingestArtifactSession` can resolve + // the target by (provider, native id) when a turn projects. (No `bindNative`: + // that's the PTY store's op and no-ops for an rpc target.) + const session = yield* rpc.launch({ + chatId: req.chatId, + targetSessionId: id, + provider: req.provider, + origin: req.origin ?? "manual", + startedAt, + cwd, + command: spec.appServer.launchCmd, + args: spec.appServer.args, + sandbox: "workspace-write", + approvalPolicy: "on-request", + }) + + // If persistence fails the launch fails — but the driver is already live + // and published in `rpc.sessions`. Tear it back down so we don't leak a + // running session the DB (and thus a restart) knows nothing about. + yield* db + .upsertTargetSession({ + id, + chatId: req.chatId, + provider: req.provider, + origin: req.origin ?? "manual", + spawnedBy: req.spawnedBy ?? null, + preset: req.preset ?? null, + cwd, + nativeSessionId: session.nativeSessionId ?? null, + nativeTranscriptPath: null, + state: "running", + startedAt, + }) + .pipe(Effect.tapError(() => rpc.stop(id))) + + return session + }) + + const launch = (req: LaunchRequest) => (req.runtime === "rpc" ? launchRpc(req) : pty.launch(req)) + + // Rejoin an app-server thread by the row's persisted native id (a codex + // session id). The session keeps its identity (same target id), so it resumes + // where it left off — mirroring the PTY `codex resume ` path, just into the + // rpc runtime instead of a terminal. + const resumeRpc = (req: ResumeRequest) => + Effect.gen(function* () { + const rows = yield* db.loadTargetSessions + const row = rows.find((r) => r.id === req.sessionId) + if (!row) { + return yield* Effect.fail(arcRequestError(`No target session "${req.sessionId}" to resume`)) + } + if (!row.nativeSessionId) { + return yield* Effect.fail( + arcRequestError(`Target session "${req.sessionId}" has no native session id to resume`), + ) + } + const spec = yield* providers.get(row.provider) + if (!spec?.appServer) { + return yield* Effect.fail( + arcRequestError(`Provider "${row.provider}" has no app-server capability`), + ) + } + + const session = yield* rpc.launch({ + chatId: row.chatId, + targetSessionId: row.id, + provider: row.provider, + origin: row.origin === "orchestrated" ? "orchestrated" : "manual", + startedAt: row.startedAt, + cwd: row.cwd, + command: spec.appServer.launchCmd, + args: spec.appServer.args, + sandbox: "workspace-write", + approvalPolicy: "on-request", + resumeThreadId: row.nativeSessionId, + }) + // No cross-manager handoff: a detached session isn't held by the PTY + // manager (the store is live-only) — it was surfaced from the DB. Launching + // it into rpc makes it live here; the unified list's detached set (DB rows + // minus live ids) drops it automatically because it's now a live id. + // Flip the durable row back to running (stop/quit may have left it exited). + yield* db + .setTargetSessionState(row.id, "running") + .pipe( + Effect.tapError((e) => Effect.logWarning(`rpc resume persist failed (${row.id}): ${e}`)), + Effect.ignore, + ) + return session + }) + + const resume = (req: ResumeRequest) => (req.runtime === "rpc" ? resumeRpc(req) : pty.resume(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)) { + // Mark the row exited *before* `rpc.stop` removes the session from the + // rpc store: that removal ticks the unified `changes`, which re-reads the + // DB for the detached set — if the row still read "running" it would + // resurrect the just-stopped session as a detached row. Best-effort + + // logged: a DB hiccup shouldn't fail an otherwise-good stop. + yield* db + .setTargetSessionState(req.sessionId, "exited") + .pipe( + Effect.tapError((e) => Effect.logWarning(`rpc stop persist failed (${req.sessionId}): ${e}`)), + Effect.ignore, + ) + return yield* rpc.stop(req.sessionId) + } + return yield* pty.stop(req) + }) + + // The unified view: the live sessions this process owns (PTY + rpc, disjoint + // by id) plus every persisted row not currently live, read from the DB and + // runtime-neutral. The non-live set carries its persisted state — `exited` + // rows show as exited (a stopped rpc session stays visible, the way a PTY exit + // stays in its store), everything else as `unknown`/detached. An id is either + // live in one manager or non-live here; never both, so there's no duplicate + // and no ownership handoff. Resuming a non-live session makes it live → it + // drops out on the next tick because its id is now in `live`. + const unify = (live: ReadonlyArray) => + Effect.map( + db.loadTargetSessions.pipe(Effect.orElseSucceed(() => [])), + (rows) => { + const liveIds = new Set(live.map((s) => s.id)) + // Not live in this process → detached; stamp the same derived fields + // (`attached: false`, `resumable`) the PTY manager's list uses, via the + // shared presenter so the two paths can't diverge on which fields they set. + const persisted = rows + .filter((r) => !liveIds.has(r.id)) + .map((r) => presentTargetSession(restoredSessionFromRow(r), false)) + return [...live, ...persisted] + }, + ) + // `rechunk(1)` per side so `zipLatestWith` tracks each list emission + // individually; both sides replay their current value on subscribe, so the + // union is live from the first pull. Each tick re-reads the DB for the + // detached set (cheap; only on a session change). + const sessions = Effect.zipWith(pty.list, rpc.sessions, (a, b) => [...a, ...b]).pipe(Effect.flatMap(unify)) + const changes = pty.changes.pipe( + Stream.rechunk(1), + Stream.zipLatestWith(Stream.rechunk(rpc.changes, 1), (a, b) => [...a, ...b]), + Stream.mapEffect(unify), + ) + + return { launch, resume, submit, stop, ownsRpc, sessions, changes } + }), +) diff --git a/src/main/services/TargetSessionManager.ts b/src/main/services/TargetSessionManager.ts index 8d345ff..0a61abe 100644 --- a/src/main/services/TargetSessionManager.ts +++ b/src/main/services/TargetSessionManager.ts @@ -15,8 +15,8 @@ import { withSqlOperation } from "../db/sql-operation.js" import { resolveArcDb } from "../db/paths.js" import { type ArcRequestError, arcRequestError } from "../errors.js" import { arcIdOrNull, type ChatId, newArcId } from "../../shared/ids.js" -import { restorePersistedSessions } from "./target-session/boot-restore.js" -import { buildProviderArgs, canResume, resumeArgs } from "./target-session/provider-args.js" +import { rearmPersistedSessionHooks, restoredSessionFromRow } from "./target-session/boot-restore.js" +import { buildProviderArgs, canResume, presentTargetSession, resumeArgs } from "./target-session/provider-args.js" import { drivePtySpawn, type FirstOutput, @@ -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. */ @@ -72,6 +80,10 @@ export interface ResumeRequest { readonly sessionId: string readonly cols?: number readonly rows?: number + /** Which runtime to resume into — `pty` (terminal, default) or `rpc` + * (app-server). A resume-time intent, not a session property: the same codex + * session resumes in either transport. Only the router reads it. */ + readonly runtime?: "pty" | "rpc" } export interface StopRequest { readonly sessionId: string @@ -128,9 +140,12 @@ export const TargetSessionManagerLive = Layer.effect( const db = yield* ArcStore const scope = yield* Effect.scope - // Restore persisted sessions on boot (and re-arm their hook sockets); see - // target-session/boot-restore.ts for the unconfirmed-state reconciliation. - const initialMap = yield* restorePersistedSessions + // Re-arm the hook sockets of persisted sessions on boot. The store itself + // starts empty — a boot-restored session isn't live in any runtime, so it's + // surfaced from the DB as *detached* by the router's unified list, not held + // here. This store holds only sessions this process has a live PTY for. + yield* rearmPersistedSessionHooks + const initialMap = new Map() // SubscriptionRef (not Ref) so the session list is observable: `changes` // pushes the current value, then every update. This is the Effect-idiomatic @@ -270,11 +285,7 @@ export const TargetSessionManagerLive = Layer.effect( ) const asList = (m: ReadonlyMap): ReadonlyArray => - Array.from(m.values()).map((s) => ({ - ...s, - attached: ptys.has(s.id), - resumable: canResume(s), - })) + Array.from(m.values()).map((s) => presentTargetSession(s, ptys.has(s.id))) const list = SubscriptionRef.get(store).pipe(Effect.map(asList)) const changes = Stream.map(SubscriptionRef.changes(store), asList) @@ -449,11 +460,20 @@ export const TargetSessionManagerLive = Layer.effect( launchLock.withPermits(1)( Effect.gen(function* () { const current = yield* SubscriptionRef.get(store) - const existing = current.get(req.sessionId) + const held = current.get(req.sessionId) + if (held && ptys.has(held.id)) return { ...held, attached: true } // already live + // A detached session is no longer held here (the store is live-only), so + // read the persisted row from the DB — the same runtime-neutral projection + // the router's detached set uses. + const row = held + ? undefined + : (yield* db.loadTargetSessions.pipe(Effect.orElseSucceed(() => []))).find( + (r) => r.id === req.sessionId, + ) + const existing = held ?? (row ? restoredSessionFromRow(row) : undefined) if (!existing) { return yield* Effect.fail(arcRequestError(`Unknown target session "${req.sessionId}"`)) } - if (ptys.has(existing.id)) return { ...existing, attached: true } const spec = yield* registry.get(existing.provider) if (!spec?.interactive) { @@ -545,6 +565,7 @@ export const TargetSessionManagerLive = Layer.effect( }), ) + const bindNative = ( targetSessionId: string, nativeSessionId: string, 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/main/services/target-session/boot-restore.ts b/src/main/services/target-session/boot-restore.ts index 70a4fdb..6c6807d 100644 Binary files a/src/main/services/target-session/boot-restore.ts and b/src/main/services/target-session/boot-restore.ts differ diff --git a/src/main/services/target-session/provider-args.ts b/src/main/services/target-session/provider-args.ts index e0b8631..fc4b39d 100644 --- a/src/main/services/target-session/provider-args.ts +++ b/src/main/services/target-session/provider-args.ts @@ -99,3 +99,16 @@ export const canResume = (s: TargetSession): boolean => { if (s.provider === "claude") return Boolean(transcriptPath && fs.existsSync(transcriptPath)) return true } + +/** + * Stamp the renderer-facing *derived* fields onto a session — `attached` (is a + * live handle held for it this process) and `resumable` (can it be re-launched). + * The single place these are computed, so the PTY manager's live list and the + * router's DB-derived detached set can't drift on which fields they set (a + * divergence that shipped as three separate bugs before this existed). + */ +export const presentTargetSession = (s: TargetSession, attached: boolean): TargetSession => ({ + ...s, + attached, + resumable: canResume(s), +}) diff --git a/src/renderer/src/App.tsx b/src/renderer/src/App.tsx index 34f32f1..44eaf1a 100644 --- a/src/renderer/src/App.tsx +++ b/src/renderer/src/App.tsx @@ -14,10 +14,11 @@ import { workspacesAtom, } from "./atoms.js" import type { ChatId, PaneId, TargetId, WorkspaceId } from "../../shared/ids.js" +import type { TargetSession } from "../../shared/instance.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 +90,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" @@ -114,11 +124,15 @@ export function App(): JSX.Element { // launch mid-bind), so this stays idempotent across every `arc:sessions` push. useEffect(() => { for (const session of unadoptedSessions(sessions, panes)) { + // rpc (app-server) sessions have no terminal — adopting one would mount an + // empty xterm (a stray cursor when focused). They live in the chat pane only. + if (session.runtime === "rpc") continue shell.actions.adoptSession({ id: session.id, provider: session.provider, chatId: session.chatId, attached: session.attached ?? false, + runtime: session.runtime, }) } }, [sessions, panes, shell.actions]) @@ -224,7 +238,32 @@ export function App(): JSX.Element { shell.actions.selectChat(workspaceId, chatId) } - const onLaunch = (provider: string, chatId: ChatId): void => { + // rpc launch/resume bypass the shell's pane/measure machinery (no `TARGET_BOUND` + // to make the session current), so once the RPC returns the session we focus it + // by ref — making it the composer target without waiting for it to land in the + // live list (a `focusSession(id)` lookup would race and miss). + const focusRpcTarget = (session: TargetSession): void => { + shell.actions.focusTarget({ + id: session.id, + provider: session.provider, + chatId: session.chatId, + attached: session.attached ?? false, + runtime: session.runtime, + }) + } + + 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" } }).then((exit) => { + if (Exit.isSuccess(exit)) focusRpcTarget(exit.value) + }) + return + } shell.actions.launchTarget(provider, chatId) } @@ -278,6 +317,23 @@ export function App(): JSX.Element { shell.actions.resumeDetached() } + // Whether the detached session can resume into the rpc (app-server) runtime — + // its provider must declare an appServer capability. Resuming that way fires + // `ResumeTarget` directly (no pane): the session comes back attached, which + // clears the detached overlay (it keys on `!attached`) and surfaces it in the + // chat pane — the resume mirror of the rpc launch entry. + const detachedProvider = providers.find((p) => p.kind === vm.detachedSession?.provider) + const canResumeDetachedRpc = Boolean(vm.detachedSession && detachedProvider?.appServer) + const onResumeDetachedRpc = (): void => { + if (vm.detachedSession) { + void runResumeTarget({ payload: { sessionId: vm.detachedSession.id, runtime: "rpc" } }).then( + (exit) => { + if (Exit.isSuccess(exit)) focusRpcTarget(exit.value) + }, + ) + } + } + const selectGitPath = (filePath: string): void => { shell.actions.open({ kind: "git", path: filePath }, "right") } @@ -353,9 +409,9 @@ export function App(): JSX.Element { workspace={vm.chatWorkspace} sessions={sessions} liveStateById={liveStateById} - activeSessionId={vm.activeSessionId} + activeTargetId={vm.activeTargetId} sessionCount={vm.sessionCount} - providers={interactiveProviders} + providers={launchableProviders} onLaunch={onLaunch} onFocusSession={focusSession} onRenameChat={renameChat} @@ -397,6 +453,8 @@ export function App(): JSX.Element { panes={panes} activePaneId={shell.state.selection.terminalPaneId} detachedSession={vm.detachedSession} + canResumeRpc={canResumeDetachedRpc} + onResumeDetachedRpc={onResumeDetachedRpc} hasWorkspaces={workspaces.length > 0} onResumeDetached={onResumeDetached} /> 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/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}} +
+ ) +} 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/TargetSessionPane.tsx b/src/renderer/src/chat/TargetSessionPane.tsx index f84a642..edc1de1 100644 --- a/src/renderer/src/chat/TargetSessionPane.tsx +++ b/src/renderer/src/chat/TargetSessionPane.tsx @@ -20,6 +20,10 @@ export interface TargetSessionPaneProps { readonly detachedSession?: TargetSession readonly hasWorkspaces: boolean readonly onResumeDetached: () => void + /** Whether the detached session's provider can resume into the app-server (rpc) + * runtime — offers a second, terminal-less resume that lands in the chat pane. */ + readonly canResumeRpc?: boolean + readonly onResumeDetachedRpc?: () => void } export function TargetSessionPane(props: TargetSessionPaneProps): JSX.Element { @@ -38,6 +42,17 @@ export function TargetSessionPane(props: TargetSessionPaneProps): JSX.Element { resume {detachedSession.provider} {RESUME_BINDING && } + {props.canResumeRpc && ( + + )} ) : (
{detachedSession.provider} session is not resumable
diff --git a/src/renderer/src/chat/UnifiedChatPane.tsx b/src/renderer/src/chat/UnifiedChatPane.tsx index 5f0ae4e..ba89c80 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 { @@ -33,10 +41,11 @@ export interface UnifiedChatPaneProps { readonly sessions: ReadonlyArray /** session id → live activity, from the `arc:live-target-states` projection */ readonly liveStateById?: LiveStateById - readonly activeSessionId?: TargetId + /** the current composer target (any runtime) — the addressee, when attached */ + readonly activeTargetId?: 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 @@ -45,12 +54,15 @@ export interface UnifiedChatPaneProps { const addressableTarget = ( chatId: ChatId, sessions: ReadonlyArray, - activeSessionId?: TargetId, + activeTargetId?: TargetId, ): TargetSession | undefined => { const inChat = sessions.filter((session) => session.chatId === chatId) + // The focused target is the addressee when it's attached and in this chat; the + // first-attached fallback is only a cold-start default (no/invalid active + // target) — never the primary path, which would be arbitrary with 2+ targets. const active = - activeSessionId !== undefined - ? inChat.find((session) => session.id === activeSessionId && session.attached) + activeTargetId !== undefined + ? inChat.find((session) => session.id === activeTargetId && session.attached) : undefined if (active) return active return inChat.find((session) => session.attached) @@ -73,7 +85,7 @@ export interface ChatPaneHandle { export const UnifiedChatPane = forwardRef( function UnifiedChatPane(props, ref): JSX.Element { - const { chat, workspace, sessions, activeSessionId, sessionCount, providers, onLaunch, onFocusSession } = + const { chat, workspace, sessions, activeTargetId, sessionCount, providers, onLaunch, onFocusSession } = props const liveStateById = props.liveStateById ?? EMPTY_LIVE_STATES const messages = useChatMessages(chat?.id) @@ -89,7 +101,6 @@ export const UnifiedChatPane = forwardRef( // the auto-picked addressee. Reset when the chat changes (it's chat-scoped). const [targetOverride, setTargetOverride] = useState(undefined) const [composerError, setComposerError] = useState(undefined) - const [sending, setSending] = useState(false) const [titleEditing, setTitleEditing] = useState(false) const [titleDraft, setTitleDraft] = useState(chat?.title ?? "") const [titleError, setTitleError] = useState(undefined) @@ -150,7 +161,7 @@ export const UnifiedChatPane = forwardRef( ? sessionsInChat.find((session) => session.id === targetOverride) : undefined const addressee = - overrideSession ?? (chat ? addressableTarget(chat.id, sessions, activeSessionId) : undefined) + overrideSession ?? (chat ? addressableTarget(chat.id, sessions, activeTargetId) : undefined) // Targets the composer's `@` picker can reference: this chat's work + sessions // (in memory) and the workspace's files (lazily fetched on first mention). @@ -180,7 +191,12 @@ export const UnifiedChatPane = forwardRef( return } - setSending(true) + // Optimistic: clear the composer immediately and let the turn run in the + // background. An rpc (app-server) turn runs to completion server-side, so + // awaiting it would hold the composer disabled for the whole turn; the reply + // lands via the chat-changes stream and "generating" shows live activity + // meanwhile — matching how a PTY submit returns instantly. + setDraft("") setComposerError(undefined) try { await rpc("SendChatPrompt", { @@ -188,12 +204,11 @@ export const UnifiedChatPane = forwardRef( targetSessionId: addressee.id, text, }) - setDraft("") } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error) setComposerError(message) - } finally { - setSending(false) + // Restore the prompt so it isn't lost — unless the user already typed a new one. + setDraft((current) => (current === "" ? text : current)) } }, [addressee, attachedInChat.length, chat, draft]) @@ -315,13 +330,13 @@ export const UnifiedChatPane = forwardRef(
{launchableProviders.map((provider) => ( ))}
@@ -340,8 +355,11 @@ export const UnifiedChatPane = forwardRef( {providers.length > 0 ? (
{providers.map((provider) => ( - ))}
@@ -403,6 +421,8 @@ export const UnifiedChatPane = forwardRef( )} + +