diff --git a/CLAUDE.md b/CLAUDE.md index 1e17df235..21bbe4d0c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -121,7 +121,6 @@ Run `npx expo-doctor` to diagnose version mismatches and native module issues. - **Launch agents in the most permissive mode**: Use full access or bypass permissions mode. - **Set cwd to the repository root** - The agent's working directory should usually be the repo root - **CRITICAL: ALWAYS RUN TYPECHECK AFTER EVERY CHANGE.** ## Agent Authentication @@ -132,7 +131,6 @@ All agent providers (Claude, Codex, OpenCode) handle their own authentication ou ## NEVER DO THESE THINGS -- **NEVER restart the main Paseo daemon on port 6767** - This is the production daemon that launches and manages agents. If you are reading this, you are probably running as an agent under it. Restarting it will kill your own process and all other running agents. The daemon is managed by the user in Tmux. -- **NEVER kill or restart processes in Tmux** without explicit user permission +- **NEVER restart the main Paseo daemon on port 6767 without permission** - This is the production daemon that launches and manages agents. If you are reading this, you are probably running as an agent under it. Restarting it will kill your own process and all other running agents. The daemon is managed by the user in Tmux. - **NEVER assume a timeout means the service needs restarting** - Timeouts can be transient network issues, not service failures - **NEVER add authentication checks to tests** - Agent providers handle their own auth. If tests fail due to auth issues, report it rather than adding conditional skips or env var checks diff --git a/packages/app/e2e/helpers/app.ts b/packages/app/e2e/helpers/app.ts index 025819f0f..767444b6f 100644 --- a/packages/app/e2e/helpers/app.ts +++ b/packages/app/e2e/helpers/app.ts @@ -137,7 +137,7 @@ async function assertE2EUsesSeededTestDaemon(page: Page): Promise { export const gotoHome = async (page: Page) => { await page.goto('/'); await ensureE2EStorageSeeded(page); - await expect(page.getByText('New Agent', { exact: true }).first()).toBeVisible(); + await expect(page.getByText('New agent', { exact: true }).first()).toBeVisible(); await expect(page.getByRole('textbox', { name: 'Message agent...' })).toBeVisible(); }; diff --git a/packages/app/src/components/agent-form/agent-form-dropdowns.tsx b/packages/app/src/components/agent-form/agent-form-dropdowns.tsx index d9bbc7346..2cbdcf2e1 100644 --- a/packages/app/src/components/agent-form/agent-form-dropdowns.tsx +++ b/packages/app/src/components/agent-form/agent-form-dropdowns.tsx @@ -851,7 +851,6 @@ export interface GitOptionsSectionProps { currentBranch: string | null; baseBranch: string; onBaseBranchChange: (value: string) => void; - branches: Array<{ name: string; isCurrent: boolean }>; status: "idle" | "loading" | "ready" | "error"; repoError: string | null; gitValidationError: string | null; @@ -871,7 +870,6 @@ export function GitOptionsSection({ currentBranch, baseBranch, onBaseBranchChange, - branches, status, repoError, gitValidationError, diff --git a/packages/app/src/screens/agent/agent-ready-screen.tsx b/packages/app/src/screens/agent/agent-ready-screen.tsx index ff9872855..221350fd2 100644 --- a/packages/app/src/screens/agent/agent-ready-screen.tsx +++ b/packages/app/src/screens/agent/agent-ready-screen.tsx @@ -373,16 +373,14 @@ function AgentScreenContent({ const modelDisplayValue = agentModel ?? "Unknown"; const repoInfoQuery = useQuery({ - queryKey: ["gitRepoInfo", serverId, agent?.cwd ?? ""], + queryKey: ["checkoutStatus", serverId, agent?.cwd ?? ""], queryFn: async () => { if (!client) { throw new Error("Daemon client unavailable"); } - const payload = await client.getGitRepoInfo({ - cwd: agent?.cwd ?? ".", - }); + const payload = await client.getCheckoutStatus(agent?.cwd ?? "."); if (payload.error) { - throw new Error(payload.error); + throw new Error(payload.error.message); } return { cwd: payload.cwd, diff --git a/packages/app/src/screens/agent/draft-agent-screen.tsx b/packages/app/src/screens/agent/draft-agent-screen.tsx index 58cc392c6..8d58ebc55 100644 --- a/packages/app/src/screens/agent/draft-agent-screen.tsx +++ b/packages/app/src/screens/agent/draft-agent-screen.tsx @@ -312,29 +312,29 @@ export function DraftAgentScreen({ type RepoInfoState = { cwd: string; repoRoot: string; - branches: Array<{ name: string; isCurrent: boolean }>; currentBranch: string | null; isDirty: boolean; }; const repoInfoQuery = useQuery({ - queryKey: ["gitRepoInfo", selectedServerId, trimmedWorkingDir], + queryKey: ["checkoutStatus", selectedServerId, trimmedWorkingDir], queryFn: async () => { const client = sessionClient; if (!client) { throw new Error("Daemon client unavailable"); } - const payload = await client.getGitRepoInfo({ - cwd: trimmedWorkingDir || ".", - }); + const payload = await client.getCheckoutStatus(trimmedWorkingDir || "."); if (payload.error) { - throw new Error(payload.error); + throw new Error(payload.error.message); } + if (!payload.isGit) { + throw new Error("Not a git repository"); + } + // After the isGit check, TypeScript knows we have a git repo return { cwd: payload.cwd, repoRoot: payload.repoRoot, - branches: payload.branches ?? [], - currentBranch: payload.currentBranch ?? null, - isDirty: Boolean(payload.isDirty), + currentBranch: payload.currentBranch, + isDirty: payload.isDirty, }; }, enabled: @@ -501,6 +501,30 @@ export function DraftAgentScreen({ validateWorktreeName, ]); + // Validate branch exists (checks local first, then remote) + const branchValidationQuery = useQuery({ + queryKey: ["validateBranch", selectedServerId, trimmedWorkingDir, baseBranch], + queryFn: async () => { + const client = sessionClient; + if (!client) { + throw new Error("Daemon client unavailable"); + } + return client.validateBranch({ + cwd: trimmedWorkingDir || ".", + branchName: baseBranch, + }); + }, + enabled: + isCreateWorktree && + !isNonGitDirectory && + Boolean(baseBranch) && + Boolean(trimmedWorkingDir) && + Boolean(sessionClient) && + isConnected, + retry: false, + staleTime: 30_000, + }); + const baseBranchError = useMemo(() => { if (!isCreateWorktree || isNonGitDirectory) { return null; @@ -508,16 +532,21 @@ export function DraftAgentScreen({ if (!baseBranch) { return "Base branch is required"; } - const branches = repoInfo?.branches ?? []; - if (branches.length === 0) { + // While validating, don't show error + if (branchValidationQuery.isPending || branchValidationQuery.isFetching) { return null; } - const branchExists = branches.some((b) => b.name === baseBranch); - if (!branchExists) { + // If validation query errored, show generic error + if (branchValidationQuery.isError) { + return "Failed to validate branch"; + } + // If validation completed and branch doesn't exist + const validationResult = branchValidationQuery.data; + if (validationResult && !validationResult.exists) { return `Branch "${baseBranch}" not found in repository`; } return null; - }, [isCreateWorktree, isNonGitDirectory, baseBranch, repoInfo?.branches]); + }, [isCreateWorktree, isNonGitDirectory, baseBranch, branchValidationQuery]); const handleBaseBranchChange = useCallback((value: string) => { setBaseBranch(value); @@ -865,7 +894,6 @@ export function DraftAgentScreen({ currentBranch={repoInfo?.currentBranch ?? null} baseBranch={baseBranch} onBaseBranchChange={handleBaseBranchChange} - branches={repoInfo?.branches ?? []} status={repoInfoStatus} repoError={repoInfoError} gitValidationError={gitBlockingError} diff --git a/packages/server/src/client/daemon-client-v2.ts b/packages/server/src/client/daemon-client-v2.ts index e561d90b7..1cc400a4e 100644 --- a/packages/server/src/client/daemon-client-v2.ts +++ b/packages/server/src/client/daemon-client-v2.ts @@ -19,17 +19,17 @@ import type { FileExplorerResponse, GitDiffResponse, GitSetupOptions, - GitRepoInfoResponse, HighlightedDiffResponse, CheckoutStatusResponse, - CheckoutDiffResponse, - CheckoutCommitResponse, - CheckoutMergeResponse, - CheckoutMergeFromBaseResponse, - CheckoutPushResponse, - CheckoutPrCreateResponse, - CheckoutPrStatusResponse, - PaseoWorktreeListResponse, + CheckoutDiffResponse, + CheckoutCommitResponse, + CheckoutMergeResponse, + CheckoutMergeFromBaseResponse, + CheckoutPushResponse, + CheckoutPrCreateResponse, + CheckoutPrStatusResponse, + ValidateBranchResponse, + PaseoWorktreeListResponse, PaseoWorktreeArchiveResponse, ProjectIconResponse, ListCommandsResponse, @@ -177,7 +177,6 @@ type ListVoiceConversationsPayload = ListVoiceConversationsResponseMessage["payl type DeleteVoiceConversationPayload = DeleteVoiceConversationResponseMessage["payload"]; type GitDiffPayload = GitDiffResponse["payload"]; type HighlightedDiffPayload = HighlightedDiffResponse["payload"]; -type GitRepoInfoPayload = GitRepoInfoResponse["payload"]; type CheckoutStatusPayload = CheckoutStatusResponse["payload"]; type CheckoutDiffPayload = CheckoutDiffResponse["payload"]; type CheckoutCommitPayload = CheckoutCommitResponse["payload"]; @@ -186,6 +185,7 @@ type CheckoutMergeFromBasePayload = CheckoutMergeFromBaseResponse["payload"]; type CheckoutPushPayload = CheckoutPushResponse["payload"]; type CheckoutPrCreatePayload = CheckoutPrCreateResponse["payload"]; type CheckoutPrStatusPayload = CheckoutPrStatusResponse["payload"]; +type ValidateBranchPayload = ValidateBranchResponse["payload"]; type PaseoWorktreeListPayload = PaseoWorktreeListResponse["payload"]; type PaseoWorktreeArchivePayload = PaseoWorktreeArchiveResponse["payload"]; type FileExplorerPayload = FileExplorerResponse["payload"]; @@ -200,9 +200,6 @@ type SubscribeTerminalPayload = SubscribeTerminalResponse["payload"]; type TerminalOutputPayload = TerminalOutput["payload"]; type KillTerminalPayload = KillTerminalResponse["payload"]; -type AgentCreateFailedStatusPayload = z.infer< - typeof AgentCreateFailedStatusPayloadSchema ->; type AgentRefreshedStatusPayload = z.infer< typeof AgentRefreshedStatusPayloadSchema >; @@ -889,43 +886,7 @@ export class DaemonClientV2 { throw new Error(status.error); } - return this.waitForAgentUpsert( - status.agentId, - (snapshot) => snapshot.status === "idle", - 60000 - ); - } - - async createAgentExpectFail( - options: CreateAgentRequestOptions - ): Promise { - const requestId = this.createRequestId(options.requestId); - const config = resolveAgentConfig(options); - - const message = SessionInboundMessageSchema.parse({ - type: "create_agent_request", - requestId, - config, - ...(options.initialPrompt ? { initialPrompt: options.initialPrompt } : {}), - }); - - const response = this.waitFor( - (msg) => { - if (msg.type !== "status") { - return null; - } - const failed = AgentCreateFailedStatusPayloadSchema.safeParse(msg.payload); - if (failed.success && failed.data.requestId === requestId) { - return failed.data; - } - return null; - }, - 10000, - { skipQueue: true } - ); - - await this.sendSessionMessageOrThrow(message); - return response; + return status.agent; } async deleteAgent(agentId: string): Promise { @@ -1007,11 +968,7 @@ export class DaemonClientV2 { await this.sendSessionMessageOrThrow(message); const status = await statusPromise; - return this.waitForAgentUpsert( - status.agentId, - (snapshot) => snapshot.status === "idle", - 60000 - ); + return status.agent; } async refreshAgent( @@ -1639,37 +1596,20 @@ export class DaemonClientV2 { return response; } - async getGitRepoInfo( - input: string | { cwd: string } | { agentId: string }, + async validateBranch( + options: { cwd: string; branchName: string }, requestId?: string - ): Promise { - const normalizedInput = - typeof input === "string" ? { agentId: input } : input; + ): Promise { const resolvedRequestId = this.createRequestId(requestId); - const cwd = - "cwd" in normalizedInput - ? normalizedInput.cwd - : (await this.fetchAgent(normalizedInput.agentId).catch(() => null))?.cwd; - - if (!cwd) { - return { - cwd: "cwd" in normalizedInput ? normalizedInput.cwd : "", - repoRoot: "", - requestId: resolvedRequestId, - error: `Agent not found: ${ - "agentId" in normalizedInput ? normalizedInput.agentId : "" - }`, - }; - } - const message = SessionInboundMessageSchema.parse({ - type: "git_repo_info_request", - cwd, + type: "validate_branch_request", + cwd: options.cwd, + branchName: options.branchName, requestId: resolvedRequestId, }); const response = this.waitFor( (msg) => { - if (msg.type !== "git_repo_info_response") { + if (msg.type !== "validate_branch_response") { return null; } if (msg.payload.requestId !== resolvedRequestId) { diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index 733e51ba2..37327f2a3 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/packages/server/src/server/agent/provider-registry.ts @@ -7,7 +7,7 @@ import type { import type { Logger } from "pino"; import { ClaudeAgentClient } from "./providers/claude-agent.js"; -import { CodexMcpAgentClient } from "./providers/codex-mcp-agent.js"; +import { CodexAppServerAgentClient } from "./providers/codex-app-server-agent.js"; import { OpenCodeAgentClient, OpenCodeServerManager } from "./providers/opencode-agent.js"; import { @@ -32,7 +32,7 @@ export interface ProviderDefinition extends AgentProviderDefinition { export function buildProviderRegistry(logger: Logger): Record { const claudeClient = new ClaudeAgentClient({ logger }); - const codexClient = new CodexMcpAgentClient(logger); + const codexClient = new CodexAppServerAgentClient(logger); const opencodeClient = new OpenCodeAgentClient(logger); return { @@ -43,7 +43,7 @@ export function buildProviderRegistry(logger: Logger): Record d.id === "codex")!, - createClient: (logger: Logger) => new CodexMcpAgentClient(logger), + createClient: (logger: Logger) => new CodexAppServerAgentClient(logger), fetchModels: (options) => codexClient.listModels(options), }, opencode: { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.e2e.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.e2e.test.ts new file mode 100644 index 000000000..b5330af11 --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.e2e.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, test } from "vitest"; +import { execFileSync } from "node:child_process"; +import os from "node:os"; +import path from "node:path"; +import { mkdtempSync } from "node:fs"; + +import { CodexAppServerAgentClient } from "./codex-app-server-agent.js"; +import { createTestLogger } from "../../../test-utils/test-logger.js"; + +const CODEX_TEST_MODEL = "gpt-5.1-codex-mini"; +const CODEX_TEST_REASONING_EFFORT = "low"; + +function isCodexInstalled(): boolean { + try { + const out = execFileSync("which", ["codex"], { encoding: "utf8" }).trim(); + return out.length > 0; + } catch { + return false; + } +} + +describe("Codex app-server provider (e2e)", () => { + test.runIf(isCodexInstalled())("lists models and runs a simple prompt", async () => { + const client = new CodexAppServerAgentClient(createTestLogger()); + const models = await client.listModels(); + expect(models.some((m) => m.id.includes("gpt-5.1-codex"))).toBe(true); + + const session = await client.createSession({ + provider: "codex", + cwd: mkdtempSync(path.join(os.tmpdir(), "codex-app-server-e2e-")), + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const result = await session.run("Say hello in one sentence."); + expect(result.finalText.length).toBeGreaterThan(0); + await session.close(); + }, 30000); +}); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts new file mode 100644 index 000000000..a2250e42f --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -0,0 +1,710 @@ +import { describe, expect, test } from "vitest"; +import { execFileSync } from "node:child_process"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync, mkdirSync, existsSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { CodexAppServerAgentClient } from "./codex-app-server-agent.js"; +import { createTestLogger } from "../../../test-utils/test-logger.js"; +import type { + AgentPermissionRequest, + AgentPromptContentBlock, + AgentTimelineItem, +} from "../agent-sdk-types.js"; + +const CODEX_TEST_MODEL = "gpt-5.1-codex-mini"; +const CODEX_TEST_REASONING_EFFORT = "low"; + +function isCodexInstalled(): boolean { + try { + const out = execFileSync("which", ["codex"], { encoding: "utf8" }).trim(); + return out.length > 0; + } catch { + return false; + } +} + +function tmpCwd(prefix = "codex-app-server-e2e-"): string { + return mkdtempSync(path.join(os.tmpdir(), prefix)); +} + +function useTempCodexSessionDir(): () => void { + const codexSessionDir = tmpCwd("codex-sessions-"); + const prevSessionDir = process.env.CODEX_SESSION_DIR; + process.env.CODEX_SESSION_DIR = codexSessionDir; + return () => { + if (prevSessionDir === undefined) { + delete process.env.CODEX_SESSION_DIR; + } else { + process.env.CODEX_SESSION_DIR = prevSessionDir; + } + rmSync(codexSessionDir, { recursive: true, force: true }); + }; +} + +function hasShellCommand(item: AgentTimelineItem, commandFragment: string): boolean { + if (item.type !== "tool_call" || item.name !== "shell") return false; + const input = item.input as { command?: string } | undefined; + const command = input?.command ?? ""; + return command.includes(commandFragment); +} + +function hasApplyPatchFile(item: AgentTimelineItem, fileName: string): boolean { + if (item.type !== "tool_call" || item.name !== "apply_patch") return false; + const input = item.input as { files?: Array<{ path?: string }> } | undefined; + const output = item.output as + | { files?: Array<{ path?: string; patch?: string }>; diff?: string } + | undefined; + const inInput = (input?.files ?? []).some((file) => file?.path === fileName); + const inOutput = (output?.files ?? []).some((file) => file?.path === fileName); + const inDiff = typeof output?.diff === "string" && output.diff.includes(fileName); + return inInput || inOutput || inDiff; +} + +describe("Codex app-server provider (integration)", () => { + const logger = createTestLogger(); + + test.runIf(isCodexInstalled())("listModels returns live Codex models", async () => { + const client = new CodexAppServerAgentClient(logger); + const models = await client.listModels(); + expect(models.some((model) => model.id.includes("gpt-5.1-codex"))).toBe(true); + }, 30000); + + test.runIf(isCodexInstalled())("getRuntimeInfo reflects model + mode", async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-runtime-"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + await session.run("Reply with OK."); + const info = await session.getRuntimeInfo(); + await session.close(); + + expect(info.model).toBe(CODEX_TEST_MODEL); + expect(info.modeId).toBe("auto"); + expect(info.sessionId?.length).toBeGreaterThan(0); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 30000); + + test.runIf(isCodexInstalled())("listCommands includes custom prompts and executeCommand runs them", async () => { + const cleanup = useTempCodexSessionDir(); + const codexHome = process.env.CODEX_HOME ?? path.join(os.homedir(), ".codex"); + const promptsDir = path.join(codexHome, "prompts"); + const promptPath = path.join(promptsDir, "test.md"); + const cwd = tmpCwd("codex-cmd-"); + + mkdirSync(promptsDir, { recursive: true }); + writeFileSync( + promptPath, + "---\ndescription: Test Prompt\n---\nReply with exactly: OK-$ARGUMENTS.", + "utf8" + ); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const commands = await session.listCommands?.(); + expect(commands?.some((cmd) => cmd.name === "prompts:test")).toBe(true); + + const result = await session.executeCommand?.("prompts:test", "world"); + await session.close(); + + expect(result?.text.trim()).toContain("OK-world"); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + rmSync(promptPath, { force: true }); + } + }, 60000); + + test.runIf(isCodexInstalled())("command approval flow requests permission and runs command", async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-cmd-approval-"); + const filePath = path.join(cwd, "permission.txt"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + approvalPolicy: "on-request", + paseoPromptInstructions: + "You must use the shell tool for command execution tasks. Do not answer without running the command.", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + let sawPermission = false; + let captured: AgentPermissionRequest | null = null; + let sawPermissionResolved = false; + const timelineItems: AgentTimelineItem[] = []; + + const events = session.stream( + [ + "You must use your shell tool to run the exact command", + "`printf \"ok\" > permission.txt`.", + "If you need approval before running it, request approval first.", + "After approval, run it and reply DONE.", + ].join(" ") + ); + + let failure: string | null = null; + for await (const event of events) { + if (event.type === "permission_requested" && event.request.name === "CodexBash") { + sawPermission = true; + captured = event.request; + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if ( + event.type === "permission_resolved" && + captured && + event.requestId === captured.id && + event.resolution.behavior === "allow" + ) { + sawPermissionResolved = true; + } + if (event.type === "timeline" && event.item.type === "tool_call") { + timelineItems.push(event.item); + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + + if (failure) { + throw new Error(failure); + } + if (captured) { + expect(sawPermissionResolved).toBe(true); + } + expect(sawPermission || timelineItems.length > 0).toBe(true); + expect( + timelineItems.some((item) => item.type === "tool_call" && item.name === "shell") + ).toBe(true); + expect(existsSync(filePath)).toBe(true); + expect(readFileSync(filePath, "utf8")).toContain("ok"); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 60000); + + test.runIf(isCodexInstalled())( + "streams responses and maps shell + file change tool calls into timeline items", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-stream-"); + const shellFile = path.join(cwd, "shell.txt"); + const patchFile = path.join(cwd, "patch.txt"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "full-access", + approvalPolicy: "on-request", + paseoPromptInstructions: + "You must use shell for commands and apply_patch for file edits. Do not skip tool usage.", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + let sawAssistantMessage = false; + let sawShellTool = false; + let sawPatchTool = false; + let sawShellCompleted = false; + let sawPatchCompleted = false; + const timelineItems: AgentTimelineItem[] = []; + + const shellEvents = session.stream( + "Run the exact shell command `printf \"ok\" > shell.txt`. After it completes, reply SHELL_DONE." + ); + let failure: string | null = null; + for await (const event of shellEvents) { + if (event.type === "permission_requested") { + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if (event.type === "timeline") { + timelineItems.push(event.item); + if (event.item.type === "assistant_message") { + sawAssistantMessage = true; + } + if (hasShellCommand(event.item, "printf")) { + sawShellTool = true; + if (event.item.status === "completed") { + sawShellCompleted = true; + } + } + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + if (failure) { + throw new Error(failure); + } + + const patch = [ + "*** Begin Patch", + "*** Add File: patch.txt", + "+patched", + "*** End Patch", + ].join("\\n"); + const patchEvents = session.stream( + [ + "Use the apply_patch tool and nothing else.", + "Do not use the shell tool for file changes.", + "Do not respond with any message until the apply_patch tool completes.", + "Apply the following patch exactly:", + patch, + "After it completes, reply PATCH_DONE.", + ].join("\\n") + ); + + for await (const event of patchEvents) { + if (event.type === "permission_requested") { + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if (event.type === "timeline") { + timelineItems.push(event.item); + if (event.item.type === "assistant_message") { + sawAssistantMessage = true; + } + if (hasApplyPatchFile(event.item, "patch.txt")) { + sawPatchTool = true; + if (event.item.status === "completed") { + sawPatchCompleted = true; + } + } + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + + if (failure) { + throw new Error(failure); + } + + expect(sawAssistantMessage).toBe(true); + expect(sawShellTool).toBe(true); + expect(sawPatchTool).toBe(true); + expect(sawShellCompleted || existsSync(shellFile)).toBe(true); + expect(sawPatchCompleted || existsSync(patchFile)).toBe(true); + expect(readFileSync(shellFile, "utf8")).toContain("ok"); + expect(readFileSync(patchFile, "utf8").trim()).toBe("patched"); + + const shellItem = timelineItems.find((item) => hasShellCommand(item, "printf")); + const patchItem = timelineItems.find((item) => hasApplyPatchFile(item, "patch.txt")); + expect(shellItem?.type).toBe("tool_call"); + expect(patchItem?.type).toBe("tool_call"); + if (shellItem?.type === "tool_call") { + expect(shellItem.name).toBe("shell"); + expect(shellItem.input).toBeTruthy(); + } + if (patchItem?.type === "tool_call") { + expect(patchItem.name).toBe("apply_patch"); + expect(patchItem.input || patchItem.output).toBeTruthy(); + } + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 120000 + ); + + test.runIf(isCodexInstalled())( + "interrupts long-running commands and emits a canceled turn", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-interrupt-"); + + let session: Awaited> | null = null; + let followupSession: Awaited> | null = + null; + let interruptAt: number | null = null; + let stoppedAt: number | null = null; + + try { + const client = new CodexAppServerAgentClient(logger); + session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + approvalPolicy: "on-request", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const stream = session.stream( + "Run the exact shell command `sleep 60` using your shell tool and do not respond until it finishes." + ); + + let sawSleepCommand = false; + for await (const event of stream) { + if (event.type === "permission_requested") { + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if ( + event.type === "timeline" && + event.item.type === "tool_call" && + event.item.name === "shell" && + hasShellCommand(event.item, "sleep 60") + ) { + sawSleepCommand = true; + if (!interruptAt) { + interruptAt = Date.now(); + await session.interrupt(); + } + } + if (event.type === "turn_canceled") { + stoppedAt = Date.now(); + break; + } + if (event.type === "turn_completed" || event.type === "turn_failed") { + stoppedAt = Date.now(); + break; + } + } + + if (!interruptAt) { + throw new Error("Did not issue interrupt for long-running command"); + } + if (!stoppedAt) { + stoppedAt = Date.now(); + } + const latencyMs = stoppedAt - interruptAt; + expect(sawSleepCommand).toBe(true); + expect(latencyMs).toBeGreaterThanOrEqual(0); + expect(latencyMs).toBeLessThan(10_000); + + await session.close(); + session = null; + + followupSession = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + const followup = await followupSession.run("Reply OK and stop."); + expect(followup.finalText.toLowerCase()).toContain("ok"); + } finally { + await session?.close(); + await followupSession?.close(); + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 120000 + ); + + test.runIf(isCodexInstalled())( + "persists session metadata and resumes with history", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-resume-"); + const token = `ALPHA-${Date.now()}`; + + let session: Awaited> | null = null; + let resumed: Awaited> | null = null; + + try { + const client = new CodexAppServerAgentClient(logger); + session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const first = await session.run(`Remember the word ${token} and reply ACK.`); + expect(first.finalText.toLowerCase()).toContain("ack"); + + const handle = session.describePersistence(); + expect(handle?.sessionId).toBeTruthy(); + expect(handle?.metadata?.threadId).toBe(handle?.sessionId); + + await session.close(); + session = null; + + resumed = await client.resumeSession(handle!); + const history: AgentTimelineItem[] = []; + for await (const event of resumed.streamHistory()) { + if (event.type === "timeline") { + history.push(event.item); + } + } + + expect( + history.some( + (item) => item.type === "assistant_message" || item.type === "user_message" + ) + ).toBe(true); + + const response = await resumed.run( + `Respond with the exact token ${token} and stop.` + ); + expect(response.finalText).toContain(token); + + const resumedHandle = resumed.describePersistence(); + expect(resumedHandle?.sessionId).toBe(handle?.sessionId); + expect(resumedHandle?.metadata?.threadId).toBe(handle?.sessionId); + } finally { + await session?.close(); + await resumed?.close(); + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 180000 + ); + + test.runIf(isCodexInstalled())( + "emits plan items and resolves collaboration mode mapping", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-plan-"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "read-only", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + const result = await Promise.race([ + session.run("Provide a concise 2-step plan and do not execute. Reply PLAN_DONE."), + new Promise((_, reject) => + setTimeout(() => reject(new Error("Plan run timed out")), 60000) + ), + ]); + + const info = await session.getRuntimeInfo(); + const sawCollaborationMode = Boolean(info.extra?.collaborationMode); + await session.close(); + + const sawTodo = result.timeline.some( + (item) => item.type === "todo" && item.items.length > 0 + ); + expect(sawTodo).toBe(true); + expect(sawCollaborationMode).toBe(true); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 120000 + ); + + test.runIf(isCodexInstalled())("file change approval flow requests permission and applies change", async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-file-approval-"); + const targetPath = path.join(cwd, "approval-test.txt"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "full-access", + approvalPolicy: "on-request", + paseoPromptInstructions: + "You must use the apply_patch tool for file edits. Do not use the shell tool for file changes.", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + let sawPermission = false; + let captured: AgentPermissionRequest | null = null; + let sawPermissionResolved = false; + const timelineItems: AgentTimelineItem[] = []; + + const patch = [ + "*** Begin Patch", + "*** Add File: approval-test.txt", + "+ok", + "*** End Patch", + ].join("\\n"); + const events = session.stream( + [ + "Use the apply_patch tool and nothing else.", + "If you need approval before writing files, request approval first.", + "Do not respond with any message until the apply_patch tool completes.", + "Apply the following patch exactly:", + patch, + "After approval, reply FILE_DONE.", + ].join("\\n") + ); + + let failure: string | null = null; + for await (const event of events) { + if (event.type === "permission_requested" && event.request.name === "CodexFileChange") { + sawPermission = true; + captured = event.request; + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if ( + event.type === "permission_resolved" && + captured && + event.requestId === captured.id && + event.resolution.behavior === "allow" + ) { + sawPermissionResolved = true; + } + if (event.type === "timeline" && event.item.type === "tool_call") { + timelineItems.push(event.item); + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + + if (failure) { + throw new Error(failure); + } + if (captured) { + expect(sawPermissionResolved).toBe(true); + } + expect(sawPermission || timelineItems.length > 0).toBe(true); + expect( + timelineItems.some((item) => item.type === "tool_call" && item.name === "apply_patch") + ).toBe(true); + expect(readFileSync(targetPath, "utf8").trim()).toBe("ok"); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 60000); + + test.runIf(isCodexInstalled())("tool approval flow requests user input for app tools", async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-tool-approval-"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "auto", + model: CODEX_TEST_MODEL, + reasoningEffort: CODEX_TEST_REASONING_EFFORT, + }); + + await session.connect(); + const rawClient = (session as any).client as { request: (method: string, params?: any) => Promise } | null; + if (!rawClient) { + throw new Error("Codex app-server client unavailable for app/list"); + } + const appsResult = await rawClient.request("app/list", { cursor: null, limit: 10 }); + const apps = Array.isArray(appsResult?.data) ? appsResult.data : []; + const app = apps.find((entry: any) => entry?.isAccessible) ?? apps[0]; + if (!app) { + await session.close(); + return; + } + + const input = [ + { type: "text", text: `$${app.id} Perform a minimal action and wait for approval if required. Reply with TOOL_DONE.` }, + { type: "mention", name: app.name ?? app.id, path: `app://${app.id}` }, + ] as unknown as AgentPromptContentBlock[]; + + let sawPermission = false; + let captured: AgentPermissionRequest | null = null; + let sawPermissionResolved = false; + let failure: string | null = null; + const timelineItems: AgentTimelineItem[] = []; + + for await (const event of session.stream(input)) { + if (event.type === "permission_requested" && event.request.name === "CodexTool") { + sawPermission = true; + captured = event.request; + await session.respondToPermission(event.request.id, { behavior: "allow" }); + } + if ( + event.type === "permission_resolved" && + captured && + event.requestId === captured.id && + event.resolution.behavior === "allow" + ) { + sawPermissionResolved = true; + } + if (event.type === "timeline" && event.item.type === "tool_call") { + timelineItems.push(event.item); + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + + if (failure) { + throw new Error(failure); + } + if (captured) { + expect(sawPermissionResolved).toBe(true); + } + expect(sawPermission || timelineItems.length > 0).toBe(true); + if (captured) { + expect(Array.isArray(captured?.metadata?.questions)).toBe(true); + } + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 90000); +}); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts new file mode 100644 index 000000000..8e1fed0f7 --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -0,0 +1,1723 @@ +import type { + AgentCapabilityFlags, + AgentClient, + AgentCommandResult, + AgentMode, + AgentModelDefinition, + McpServerConfig, + AgentPermissionRequest, + AgentPermissionResponse, + AgentPromptContentBlock, + AgentPromptInput, + AgentRunOptions, + AgentRunResult, + AgentRuntimeInfo, + AgentSession, + AgentSessionConfig, + AgentSlashCommand, + AgentStreamEvent, + AgentTimelineItem, + AgentUsage, + ListModelsOptions, + ListPersistedAgentsOptions, + PersistedAgentDescriptor, + ToolCallTimelineItem, +} from "../agent-sdk-types.js"; +import type { Logger } from "pino"; + +import { execSync, spawn } from "node:child_process"; +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import fs from "node:fs/promises"; +import { Dirent } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import readline from "node:readline"; + +import { injectLeadingPaseoInstructionTag } from "../paseo-instructions-tag.js"; + +const DEFAULT_TIMEOUT_MS = 14 * 24 * 60 * 60 * 1000; +const CODEX_PROVIDER = "codex" as const; + +const CODEX_APP_SERVER_CAPABILITIES: AgentCapabilityFlags = { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: false, + supportsMcpServers: true, + supportsReasoningStream: true, + supportsToolInvocations: true, +}; + +const CODEX_MODES: AgentMode[] = [ + { + id: "read-only", + label: "Read Only", + description: + "Read files and answer questions. Manual approval required for edits, commands, or network ops.", + }, + { + id: "auto", + label: "Auto", + description: + "Edit files and run commands but still request approval before escalating scope.", + }, + { + id: "full-access", + label: "Full Access", + description: "Edit files, run commands, and access the network without additional prompts.", + }, +]; + +const DEFAULT_CODEX_MODE_ID = "auto"; + +const MODE_PRESETS: Record< + string, + { approvalPolicy: string; sandbox: string; networkAccess?: boolean } +> = { + "read-only": { + approvalPolicy: "on-request", + sandbox: "read-only", + }, + auto: { + approvalPolicy: "on-request", + sandbox: "workspace-write", + }, + "full-access": { + approvalPolicy: "never", + sandbox: "danger-full-access", + networkAccess: true, + }, +}; + +function validateCodexMode(modeId: string): void { + if (!(modeId in MODE_PRESETS)) { + const validModes = Object.keys(MODE_PRESETS).join(", "); + throw new Error( + `Invalid Codex mode "${modeId}". Valid modes are: ${validModes}` + ); + } +} + +function resolveCodexBinary(): string { + try { + const codexPath = execSync("which codex", { encoding: "utf8" }).trim(); + if (codexPath) { + return codexPath; + } + } catch { + // Fall through to error + } + throw new Error( + "Codex CLI not found. Please install codex globally: npm install -g @openai/codex" + ); +} + +function resolveCodexHomeDir(): string { + return process.env.CODEX_HOME ?? path.join(os.homedir(), ".codex"); +} + +function tokenizeCommandArgs(args: string): string[] { + const tokens: string[] = []; + let current = ""; + let quote: "'" | '"' | null = null; + for (let i = 0; i < args.length; i += 1) { + const ch = args[i]!; + if (quote) { + if (ch === quote) { + quote = null; + continue; + } + if (ch === "\\" && i + 1 < args.length) { + const next = args[i + 1]!; + if (next === quote || next === "\\" || next === "n" || next === "t") { + i += 1; + current += next === "n" ? "\n" : next === "t" ? "\t" : next; + continue; + } + } + current += ch; + continue; + } + + if (ch === "'" || ch === '"') { + quote = ch; + continue; + } + + if (/\s/.test(ch)) { + if (current) { + tokens.push(current); + current = ""; + } + continue; + } + current += ch; + } + if (current) { + tokens.push(current); + } + return tokens; +} + +function parseFrontMatter(markdown: string): { + frontMatter: Record; + body: string; +} { + const lines = markdown.split("\n"); + if (lines[0]?.trim() !== "---") { + return { frontMatter: {}, body: markdown }; + } + let end = -1; + for (let i = 1; i < lines.length; i += 1) { + if (lines[i]?.trim() === "---") { + end = i; + break; + } + } + if (end === -1) { + return { frontMatter: {}, body: markdown }; + } + const metaLines = lines.slice(1, end); + const body = lines.slice(end + 1).join("\n"); + const frontMatter: Record = {}; + for (const line of metaLines) { + const trimmed = line.trim(); + if (!trimmed || trimmed.startsWith("#")) { + continue; + } + const idx = trimmed.indexOf(":"); + if (idx <= 0) { + continue; + } + const key = trimmed.slice(0, idx).trim(); + let value = trimmed.slice(idx + 1).trim(); + value = value.replace(/^['"]/, "").replace(/['"]$/, ""); + if (key && value) { + frontMatter[key] = value; + } + } + return { frontMatter, body }; +} + +async function listCodexCustomPrompts(): Promise { + const codexHome = resolveCodexHomeDir(); + const promptsDir = path.join(codexHome, "prompts"); + let entries: Dirent[]; + try { + entries = await fs.readdir(promptsDir, { withFileTypes: true }); + } catch { + return []; + } + + const commands: AgentSlashCommand[] = []; + for (const entry of entries) { + if (!entry.isFile()) { + continue; + } + if (!entry.name.endsWith(".md")) { + continue; + } + const name = entry.name.slice(0, -".md".length); + if (!name) { + continue; + } + const fullPath = path.join(promptsDir, entry.name); + let content: string; + try { + content = await fs.readFile(fullPath, "utf8"); + } catch { + continue; + } + const parsed = parseFrontMatter(content); + const description = parsed.frontMatter["description"] ?? "Custom prompt"; + const argumentHint = + parsed.frontMatter["argument-hint"] ?? + parsed.frontMatter["argument_hint"] ?? + ""; + commands.push({ + name: `prompts:${name}`, + description, + argumentHint, + }); + } + return commands.sort((a, b) => a.name.localeCompare(b.name)); +} + +async function listCodexSkills(cwd: string): Promise { + const candidates: string[] = []; + candidates.push(path.join(cwd, ".codex", "skills")); + + const repoRoot = (() => { + try { + const output = execSync("git rev-parse --show-toplevel", { + cwd, + encoding: "utf8", + stdio: ["ignore", "pipe", "ignore"], + }); + const trimmed = output.trim(); + return trimmed ? trimmed : null; + } catch { + return null; + } + })(); + if (repoRoot) { + candidates.push(path.join(path.dirname(cwd), ".codex", "skills")); + candidates.push(path.join(repoRoot, ".codex", "skills")); + } + + candidates.push(path.join(resolveCodexHomeDir(), "skills")); + + const commandsByName = new Map(); + + for (const dir of candidates) { + let entries: Dirent[]; + try { + entries = await fs.readdir(dir, { withFileTypes: true }); + } catch { + continue; + } + + for (const entry of entries) { + if (!entry.isDirectory() && !entry.isSymbolicLink()) { + continue; + } + const skillDir = path.join(dir, entry.name); + const skillPath = path.join(skillDir, "SKILL.md"); + let content: string; + try { + content = await fs.readFile(skillPath, "utf8"); + } catch { + continue; + } + const { frontMatter } = parseFrontMatter(content); + const name = frontMatter["name"]; + const description = frontMatter["description"]; + if (!name || !description) { + continue; + } + if (!commandsByName.has(name)) { + commandsByName.set(name, { + name, + description, + argumentHint: "", + }); + } + } + } + + return Array.from(commandsByName.values()).sort((a, b) => + a.name.localeCompare(b.name) + ); +} + +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); +} + +function expandCodexCustomPrompt(template: string, args: string | undefined): string { + const trimmedArgs = args ? args.trim() : ""; + const tokens = trimmedArgs ? tokenizeCommandArgs(trimmedArgs) : []; + const named: Record = {}; + const positional: string[] = []; + + for (const token of tokens) { + const idx = token.indexOf("="); + if (idx > 0) { + const key = token.slice(0, idx); + const value = token.slice(idx + 1); + if (key) { + named[key] = value; + continue; + } + } + positional.push(token); + } + + const dollarPlaceholder = "__CODEX_DOLLAR_PLACEHOLDER__"; + let out = template.split("$$").join(dollarPlaceholder); + + out = out.split("$ARGUMENTS").join(trimmedArgs); + + for (let i = 1; i <= 9; i += 1) { + const value = positional[i - 1] ?? ""; + out = out.split(`$${i}`).join(value); + } + + const namedKeys = Object.keys(named).sort((a, b) => b.length - a.length); + for (const key of namedKeys) { + const value = named[key] ?? ""; + const re = new RegExp(`\\$${escapeRegExp(key)}\\b`, "g"); + out = out.replace(re, value); + } + + out = out.split(dollarPlaceholder).join("$"); + return out; +} + +interface CodexMcpServerConfig { + url?: string; + http_headers?: Record; + command?: string; + args?: string[]; + env?: Record; + tool_timeout_sec?: number; +} + +function toCodexMcpConfig(config: McpServerConfig): CodexMcpServerConfig { + switch (config.type) { + case "stdio": + return { + command: config.command, + args: config.args, + env: config.env, + }; + case "http": + return { + url: config.url, + http_headers: config.headers, + }; + case "sse": + return { + url: config.url, + http_headers: config.headers, + }; + } +} + +class Pushable implements AsyncIterable { + private queue: T[] = []; + private resolvers: ((value: IteratorResult) => void)[] = []; + private closed = false; + + push(item: T) { + if (this.closed) { + return; + } + if (this.resolvers.length > 0) { + const resolve = this.resolvers.shift()!; + resolve({ value: item, done: false }); + } else { + this.queue.push(item); + } + } + + end() { + this.closed = true; + while (this.resolvers.length > 0) { + const resolve = this.resolvers.shift()!; + resolve({ value: undefined, done: true }); + } + } + + [Symbol.asyncIterator](): AsyncIterator { + return { + next: (): Promise> => { + if (this.queue.length > 0) { + const value = this.queue.shift()!; + return Promise.resolve({ value, done: false }); + } + if (this.closed) { + return Promise.resolve({ value: undefined, done: true }); + } + return new Promise>((resolve) => { + this.resolvers.push(resolve); + }); + }, + }; + } +} + +type JsonRpcRequest = { + id: number; + method: string; + params?: unknown; +}; + +type JsonRpcResponse = { + id: number; + result?: unknown; + error?: { code?: number; message: string }; +}; + +type JsonRpcNotification = { + method: string; + params?: unknown; +}; + +type JsonRpcMessage = JsonRpcRequest | JsonRpcResponse | JsonRpcNotification; + +type PendingRequest = { + resolve: (value: unknown) => void; + reject: (error: Error) => void; + timer: ReturnType; +}; + +type RequestHandler = (params: unknown) => Promise | unknown; + +type NotificationHandler = (method: string, params: unknown) => void; + +class CodexAppServerClient { + private readonly rl: readline.Interface; + private readonly pending = new Map(); + private readonly requestHandlers = new Map(); + private notificationHandler: NotificationHandler | null = null; + private nextId = 1; + private disposed = false; + private stderrBuffer = ""; + + constructor( + private readonly child: ChildProcessWithoutNullStreams, + private readonly logger: Logger + ) { + this.rl = readline.createInterface({ input: child.stdout }); + this.rl.on("line", (line) => this.handleLine(line)); + + child.stderr.on("data", (chunk) => { + this.stderrBuffer += chunk.toString(); + if (this.stderrBuffer.length > 8192) { + this.stderrBuffer = this.stderrBuffer.slice(-8192); + } + }); + + child.on("exit", (code, signal) => { + const message = + code === 0 && !signal + ? "Codex app-server exited" + : `Codex app-server exited with code ${code ?? "null"} and signal ${signal ?? "null"}`; + const error = new Error(`${message}\n${this.stderrBuffer}`.trim()); + for (const pending of this.pending.values()) { + clearTimeout(pending.timer); + pending.reject(error); + } + this.pending.clear(); + this.disposed = true; + }); + } + + setNotificationHandler(handler: NotificationHandler): void { + this.notificationHandler = handler; + } + + setRequestHandler(method: string, handler: RequestHandler): void { + this.requestHandlers.set(method, handler); + } + + request(method: string, params?: unknown, timeoutMs = DEFAULT_TIMEOUT_MS): Promise { + if (this.disposed) { + return Promise.reject(new Error("Codex app-server client is closed")); + } + const id = this.nextId++; + const payload: JsonRpcRequest = { id, method, params }; + const serialized = JSON.stringify(payload); + this.child.stdin.write(`${serialized}\n`); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.pending.delete(id); + reject(new Error(`Codex app-server request timed out for ${method}`)); + }, timeoutMs); + this.pending.set(id, { resolve, reject, timer }); + }); + } + + notify(method: string, params?: unknown): void { + if (this.disposed) { + return; + } + const payload: JsonRpcNotification = { method, params }; + this.child.stdin.write(`${JSON.stringify(payload)}\n`); + } + + async dispose(): Promise { + if (this.disposed) return; + this.disposed = true; + this.rl.close(); + try { + this.child.kill(); + } catch { + // ignore + } + } + + private async handleLine(line: string): Promise { + if (!line.trim()) return; + let msg: JsonRpcMessage; + try { + msg = JSON.parse(line); + } catch (error) { + this.logger.warn({ error, line }, "Failed to parse Codex app-server JSON"); + return; + } + + if (typeof (msg as JsonRpcResponse).id === "number") { + const id = (msg as JsonRpcResponse).id; + if ((msg as JsonRpcResponse).result !== undefined || (msg as JsonRpcResponse).error) { + const pending = this.pending.get(id); + if (!pending) return; + clearTimeout(pending.timer); + this.pending.delete(id); + if ((msg as JsonRpcResponse).error) { + pending.reject(new Error((msg as JsonRpcResponse).error?.message ?? "Unknown error")); + } else { + pending.resolve((msg as JsonRpcResponse).result); + } + return; + } + + // Server-initiated request + if (typeof (msg as JsonRpcRequest).method === "string") { + const request = msg as JsonRpcRequest; + const handler = this.requestHandlers.get(request.method); + try { + const result = handler ? await handler(request.params) : {}; + const response: JsonRpcResponse = { id: request.id, result }; + this.child.stdin.write(`${JSON.stringify(response)}\n`); + } catch (error) { + const response: JsonRpcResponse = { + id: request.id, + error: { message: error instanceof Error ? error.message : String(error) }, + }; + this.child.stdin.write(`${JSON.stringify(response)}\n`); + } + return; + } + } + + if (typeof (msg as JsonRpcNotification).method === "string") { + const notification = msg as JsonRpcNotification; + this.notificationHandler?.(notification.method, notification.params); + } + } +} + +function toAgentUsage(tokenUsage: unknown): AgentUsage | undefined { + if (!tokenUsage || typeof tokenUsage !== "object") return undefined; + const usage = tokenUsage as { last?: { inputTokens?: number; cachedInputTokens?: number; outputTokens?: number } }; + return { + inputTokens: usage.last?.inputTokens, + cachedInputTokens: usage.last?.cachedInputTokens, + outputTokens: usage.last?.outputTokens, + }; +} + +function createToolCallTimelineItem( + data: Omit +): AgentTimelineItem { + return { type: "tool_call", ...data }; +} + +function extractUserText(content: unknown): string | null { + if (!Array.isArray(content)) return null; + const parts: string[] = []; + for (const item of content) { + if (item && typeof item === "object") { + const obj = item as { type?: string; text?: string }; + if (obj.type === "text" && typeof obj.text === "string") { + parts.push(obj.text); + } + } + } + return parts.length > 0 ? parts.join("\n") : null; +} + +function extractContentText(content: unknown): string | null { + if (!Array.isArray(content)) return null; + const parts: string[] = []; + for (const item of content) { + if (item && typeof item === "object") { + const obj = item as { text?: string }; + if (typeof obj.text === "string") { + parts.push(obj.text); + } + } + } + return parts.length > 0 ? parts.join("\n") : null; +} + +function parsePlanTextToTodoItems(text: string): { text: string; completed: boolean }[] { + const lines = text + .split("\n") + .map((line) => line.trim()) + .filter((line) => line.length > 0); + if (lines.length === 0) { + return [{ text, completed: false }]; + } + return lines.map((line) => ({ + text: line.replace(/^[-*]\s+/, ""), + completed: false, + })); +} + +function planStepsToTodoItems(steps: Array<{ step: string; status: string }>): { + text: string; + completed: boolean; +}[] { + return steps.map((entry) => ({ + text: entry.step, + completed: entry.status === "completed", + })); +} + +function threadItemToTimeline( + item: any, + options?: { includeUserMessage?: boolean } +): AgentTimelineItem | null { + if (!item || typeof item !== "object") return null; + const includeUserMessage = options?.includeUserMessage ?? true; + switch (item.type) { + case "userMessage": { + if (!includeUserMessage) { + return null; + } + const text = extractUserText(item.content) ?? ""; + return { type: "user_message", text }; + } + case "agentMessage": { + return { type: "assistant_message", text: item.text ?? "" }; + } + case "plan": { + const text = item.text ?? ""; + const items = parsePlanTextToTodoItems(text); + return { type: "todo", items }; + } + case "reasoning": { + const summary = Array.isArray(item.summary) ? item.summary.join("\n") : ""; + const content = Array.isArray(item.content) ? item.content.join("\n") : ""; + const text = summary || content; + return text ? { type: "reasoning", text } : null; + } + case "commandExecution": { + const output = { + type: "command", + command: item.command, + output: item.aggregatedOutput ?? "", + exitCode: item.exitCode ?? undefined, + }; + return createToolCallTimelineItem({ + name: "shell", + status: item.status, + callId: item.id, + input: { command: item.command, cwd: item.cwd }, + output, + }); + } + case "fileChange": { + const files = Array.isArray(item.changes) + ? item.changes.map((change: any) => ({ + path: change.path, + kind: change.kind, + })) + : []; + const outputFiles = Array.isArray(item.changes) + ? item.changes.map((change: any) => ({ + path: change.path, + patch: change.diff, + kind: change.kind, + })) + : []; + return createToolCallTimelineItem({ + name: "apply_patch", + status: item.status, + callId: item.id, + input: { files }, + output: { files: outputFiles }, + }); + } + case "mcpToolCall": { + if (item.tool === "read_file") { + const pathValue = item.arguments?.path ?? item.arguments?.file_path ?? null; + const content = extractContentText(item.result?.content) ?? ""; + return createToolCallTimelineItem({ + name: "read_file", + status: item.status, + callId: item.id, + input: pathValue ? { path: pathValue } : item.arguments, + output: pathValue + ? { type: "read_file", path: pathValue, content } + : item.result ?? undefined, + error: item.error ?? undefined, + }); + } + return createToolCallTimelineItem({ + name: `${item.server}.${item.tool}`, + status: item.status, + callId: item.id, + input: item.arguments, + output: item.result ?? undefined, + error: item.error ?? undefined, + }); + } + case "webSearch": { + return createToolCallTimelineItem({ + name: "web_search", + status: "completed", + callId: item.id, + input: { query: item.query }, + output: item.action ?? undefined, + }); + } + default: + return null; + } +} + +function toSandboxPolicy(type: string, networkAccess?: boolean): Record { + switch (type) { + case "read-only": + return { type: "readOnly" }; + case "workspace-write": + return { type: "workspaceWrite", networkAccess: networkAccess ?? false }; + case "danger-full-access": + return { type: "dangerFullAccess" }; + default: + return { type: "workspaceWrite", networkAccess: networkAccess ?? false }; + } +} + +class CodexAppServerAgentSession implements AgentSession { + readonly provider = CODEX_PROVIDER; + readonly capabilities = CODEX_APP_SERVER_CAPABILITIES; + + private readonly logger: Logger; + private readonly config: AgentSessionConfig; + private currentMode: string; + private currentThreadId: string | null = null; + private currentTurnId: string | null = null; + private client: CodexAppServerClient | null = null; + private eventQueue: Pushable | null = null; + private cachedRuntimeInfo: AgentRuntimeInfo | null = null; + private historyPending = false; + private persistedHistory: AgentTimelineItem[] = []; + private pendingPermissions = new Map(); + private pendingPermissionHandlers = new Map< + string, + { + resolve: (value: unknown) => void; + kind: "command" | "file" | "tool"; + questions?: Array<{ id: string; options?: Array<{ label?: string; value?: string }> }>; + } + >(); + private resolvedPermissionRequests = new Set(); + private pendingAgentMessages = new Map(); + private pendingReasoning = new Map(); + private latestUsage: AgentUsage | undefined; + private connected = false; + private paseoInstructionsInjected = false; + private collaborationModes: Array<{ + name: string; + mode?: string | null; + model?: string | null; + reasoning_effort?: string | null; + developer_instructions?: string | null; + }> = []; + private resolvedCollaborationMode: { mode: string; settings: Record; name: string } | null = null; + private cachedSkills: Array<{ name: string; description: string; path: string }> = []; + + constructor( + config: AgentSessionConfig, + private readonly resumeHandle: { sessionId: string; metadata?: Record } | null, + logger: Logger + ) { + this.logger = logger.child({ module: "agent", provider: CODEX_PROVIDER }); + if (config.modeId === undefined) { + throw new Error("Codex agent requires modeId to be specified"); + } + validateCodexMode(config.modeId); + this.currentMode = config.modeId; + this.config = config; + + if (this.resumeHandle?.sessionId) { + this.currentThreadId = this.resumeHandle.sessionId; + this.historyPending = true; + this.paseoInstructionsInjected = true; + } + } + + get id(): string | null { + return this.currentThreadId; + } + + async connect(): Promise { + if (this.connected) return; + const binaryPath = resolveCodexBinary(); + const child = spawn(binaryPath, ["app-server"], { + stdio: ["pipe", "pipe", "pipe"], + env: process.env, + }); + this.client = new CodexAppServerClient(child, this.logger); + this.client.setNotificationHandler((method, params) => this.handleNotification(method, params)); + this.registerRequestHandlers(); + + await this.client.request("initialize", { + clientInfo: { + name: "paseo", + title: "Paseo", + version: "0.0.0", + }, + }); + this.client.notify("initialized", {}); + + await this.loadCollaborationModes(); + await this.loadSkills(); + + if (this.currentThreadId) { + await this.loadPersistedHistory(); + await this.ensureThreadLoaded(); + } + + this.connected = true; + } + + private async loadCollaborationModes(): Promise { + if (!this.client) return; + try { + const response = (await this.client.request("collaborationMode/list", {})) as { + data?: Array; + }; + const data = Array.isArray(response?.data) ? response.data : []; + this.collaborationModes = data.map((entry) => ({ + name: String(entry.name ?? ""), + mode: entry.mode ?? null, + model: entry.model ?? null, + reasoning_effort: entry.reasoning_effort ?? null, + developer_instructions: entry.developer_instructions ?? null, + })); + } catch (error) { + this.logger.debug({ error }, "Failed to load collaboration modes"); + this.collaborationModes = []; + } + this.resolvedCollaborationMode = this.resolveCollaborationMode(this.currentMode); + } + + private async loadSkills(): Promise { + if (!this.client) return; + try { + const response = (await this.client.request("skills/list", { + cwd: [this.config.cwd], + })) as { data?: Array }; + const entries = Array.isArray(response?.data) ? response.data : []; + const skills: Array<{ name: string; description: string; path: string }> = []; + for (const entry of entries) { + const list = Array.isArray(entry.skills) ? entry.skills : []; + for (const skill of list) { + if (!skill?.name || !skill?.path) continue; + skills.push({ + name: skill.name, + description: skill.description ?? skill.shortDescription ?? "Skill", + path: skill.path, + }); + } + } + this.cachedSkills = skills; + } catch (error) { + this.logger.debug({ error }, "Failed to load skills list"); + this.cachedSkills = []; + } + } + + private resolveCollaborationMode(modeId: string): { mode: string; settings: Record; name: string } | null { + if (this.collaborationModes.length === 0) return null; + const normalized = modeId.toLowerCase(); + const findByName = (predicate: (name: string) => boolean) => + this.collaborationModes.find((entry) => predicate(entry.name.toLowerCase())); + let match = + normalized === "read-only" + ? findByName((name) => name.includes("read") || name.includes("plan")) + : normalized === "full-access" + ? findByName((name) => name.includes("full") || name.includes("exec")) + : findByName((name) => name.includes("auto") || name.includes("code")); + if (!match) { + match = this.collaborationModes[0] ?? null; + } + if (!match) return null; + const settings: Record = {}; + if (match.model) settings.model = match.model; + if (match.reasoning_effort) settings.reasoning_effort = match.reasoning_effort; + if (match.developer_instructions) settings.developer_instructions = match.developer_instructions; + if (this.config.model) settings.model = this.config.model; + if (this.config.reasoningEffort) settings.reasoning_effort = this.config.reasoningEffort; + return { mode: match.mode ?? "code", settings, name: match.name }; + } + + private registerRequestHandlers(): void { + if (!this.client) return; + + this.client.setRequestHandler("item/commandExecution/requestApproval", (params) => + this.handleCommandApprovalRequest(params) + ); + this.client.setRequestHandler("item/fileChange/requestApproval", (params) => + this.handleFileChangeApprovalRequest(params) + ); + this.client.setRequestHandler("tool/requestUserInput", (params) => + this.handleToolApprovalRequest(params) + ); + } + + private async loadPersistedHistory(): Promise { + if (!this.client || !this.currentThreadId) return; + try { + const response = (await this.client.request("thread/read", { + threadId: this.currentThreadId, + includeTurns: true, + })) as { thread?: { turns?: Array<{ items?: any[] }> } }; + const thread = response?.thread; + if (!thread || !Array.isArray(thread.turns)) return; + const timeline: AgentTimelineItem[] = []; + for (const turn of thread.turns) { + const items = Array.isArray(turn.items) ? turn.items : []; + for (const item of items) { + const timelineItem = threadItemToTimeline(item); + if (timelineItem) { + timeline.push(timelineItem); + } + } + } + if (timeline.length > 0) { + this.persistedHistory = timeline; + this.historyPending = true; + } + } catch (error) { + this.logger.warn({ error }, "Failed to load Codex thread history"); + } + } + + private async ensureThreadLoaded(): Promise { + if (!this.client || !this.currentThreadId) return; + try { + const loaded = (await this.client.request("thread/loaded/list", {})) as { data?: string[] }; + const ids = Array.isArray(loaded?.data) ? loaded.data : []; + if (ids.includes(this.currentThreadId)) { + return; + } + await this.client.request("thread/resume", { threadId: this.currentThreadId }); + } catch (error) { + this.logger.warn({ error }, "Failed to resume Codex thread, starting new thread"); + this.currentThreadId = null; + await this.ensureThread(); + } + } + + async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise { + const events = this.stream(prompt, options); + const timeline: AgentTimelineItem[] = []; + let finalText = ""; + let usage: AgentUsage | undefined; + + for await (const event of events) { + if (event.type === "timeline") { + timeline.push(event.item); + if (event.item.type === "assistant_message") { + finalText = event.item.text; + } + } else if (event.type === "turn_completed") { + usage = event.usage; + } else if (event.type === "turn_failed") { + throw new Error(event.error); + } + } + + const info = await this.getRuntimeInfo(); + return { + sessionId: info.sessionId ?? "", + finalText, + usage, + timeline, + }; + } + + async *stream( + prompt: AgentPromptInput, + options?: AgentRunOptions + ): AsyncGenerator { + await this.connect(); + if (!this.client) return; + + const queue = new Pushable(); + this.eventQueue = queue; + + try { + if (this.currentThreadId) { + await this.ensureThreadLoaded(); + } else { + await this.ensureThread(); + } + const input = this.buildUserInput(prompt); + const preset = MODE_PRESETS[this.currentMode] ?? MODE_PRESETS[DEFAULT_CODEX_MODE_ID]; + const approvalPolicy = this.config.approvalPolicy ?? preset.approvalPolicy; + const sandboxPolicyType = this.config.sandboxMode ?? preset.sandbox; + + const params: Record = { + threadId: this.currentThreadId, + input, + approvalPolicy, + sandboxPolicy: toSandboxPolicy( + sandboxPolicyType, + typeof this.config.networkAccess === "boolean" + ? this.config.networkAccess + : preset.networkAccess + ), + }; + + if (this.config.model) { + params.model = this.config.model; + } + if (this.config.reasoningEffort) { + params.effort = this.config.reasoningEffort; + } + if (this.resolvedCollaborationMode) { + params.collaborationMode = { + mode: this.resolvedCollaborationMode.mode, + settings: this.resolvedCollaborationMode.settings, + }; + } + if (this.config.cwd) { + params.cwd = this.config.cwd; + } + if (options?.outputSchema) { + params.outputSchema = options.outputSchema; + } + + await this.client.request("turn/start", params); + + for await (const event of queue) { + yield event; + if ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ) { + break; + } + } + } finally { + if (this.eventQueue === queue) { + this.eventQueue = null; + } + } + } + + async *streamHistory(): AsyncGenerator { + if (!this.historyPending || this.persistedHistory.length === 0) { + return; + } + const history = this.persistedHistory; + this.persistedHistory = []; + this.historyPending = false; + for (const item of history) { + yield { type: "timeline", provider: CODEX_PROVIDER, item }; + } + } + + async getRuntimeInfo(): Promise { + if (this.cachedRuntimeInfo) return { ...this.cachedRuntimeInfo }; + const info: AgentRuntimeInfo = { + provider: CODEX_PROVIDER, + sessionId: this.currentThreadId, + model: this.config.model ?? null, + modeId: this.currentMode ?? null, + extra: this.resolvedCollaborationMode + ? { collaborationMode: this.resolvedCollaborationMode.name } + : undefined, + }; + this.cachedRuntimeInfo = info; + return { ...info }; + } + + async getAvailableModes(): Promise { + return CODEX_MODES; + } + + async getCurrentMode(): Promise { + return this.currentMode ?? null; + } + + async setMode(modeId: string): Promise { + validateCodexMode(modeId); + this.currentMode = modeId; + this.resolvedCollaborationMode = this.resolveCollaborationMode(modeId); + this.cachedRuntimeInfo = null; + } + + getPendingPermissions(): AgentPermissionRequest[] { + return Array.from(this.pendingPermissions.values()); + } + + async respondToPermission( + requestId: string, + response: AgentPermissionResponse + ): Promise { + const pending = this.pendingPermissionHandlers.get(requestId); + if (!pending) { + throw new Error(`No pending Codex app-server permission request with id '${requestId}'`); + } + this.pendingPermissionHandlers.delete(requestId); + this.pendingPermissions.delete(requestId); + this.resolvedPermissionRequests.add(requestId); + + this.emitEvent({ + type: "permission_resolved", + provider: CODEX_PROVIDER, + requestId, + resolution: response, + }); + + if (pending.kind === "command") { + const decision = response.behavior === "allow" + ? "accept" + : response.interrupt + ? "cancel" + : "decline"; + pending.resolve({ decision }); + return; + } + + if (pending.kind === "file") { + const decision = response.behavior === "allow" + ? "accept" + : response.interrupt + ? "cancel" + : "decline"; + pending.resolve({ decision }); + return; + } + + // tool/requestUserInput + const answers: Record = {}; + const questions = pending.questions ?? []; + const decision = response.behavior === "allow" ? "accept" : response.interrupt ? "cancel" : "decline"; + for (const question of questions) { + let picked = decision; + const options = question.options ?? []; + if (options.length > 0) { + const byLabel = options.find((opt) => (opt.label ?? "").toLowerCase().includes(decision)); + const byValue = options.find((opt) => (opt.value ?? "").toLowerCase().includes(decision)); + const option = byLabel ?? byValue ?? options[0]!; + picked = option.value ?? option.label ?? decision; + } + answers[question.id] = { answers: [picked] }; + } + if (questions.length === 0) { + answers["default"] = { answers: [decision] }; + } + pending.resolve({ answers }); + } + + describePersistence(): { provider: typeof CODEX_PROVIDER; sessionId: string; nativeHandle: string; metadata: Record } | null { + if (!this.currentThreadId) return null; + return { + provider: CODEX_PROVIDER, + sessionId: this.currentThreadId, + nativeHandle: this.currentThreadId, + metadata: { + provider: CODEX_PROVIDER, + cwd: this.config.cwd, + title: this.config.title ?? null, + threadId: this.currentThreadId, + modeId: this.currentMode, + model: this.config.model ?? null, + reasoningEffort: this.config.reasoningEffort ?? null, + }, + }; + } + + async interrupt(): Promise { + if (!this.client || !this.currentThreadId || !this.currentTurnId) return; + try { + await this.client.request("turn/interrupt", { + threadId: this.currentThreadId, + turnId: this.currentTurnId, + }); + } catch (error) { + this.logger.warn({ error }, "Failed to interrupt Codex turn"); + } + } + + async close(): Promise { + for (const pending of this.pendingPermissionHandlers.values()) { + pending.resolve({ decision: "cancel" }); + } + this.pendingPermissionHandlers.clear(); + this.pendingPermissions.clear(); + this.resolvedPermissionRequests.clear(); + this.eventQueue?.end(); + this.eventQueue = null; + if (this.client) { + await this.client.dispose(); + } + this.client = null; + this.connected = false; + this.currentThreadId = null; + this.currentTurnId = null; + } + + async listCommands(): Promise { + const prompts = await listCodexCustomPrompts(); + if (!this.connected) { + await this.connect(); + } else { + await this.loadSkills(); + } + const appServerSkills = this.cachedSkills.map((skill) => ({ + name: skill.name, + description: skill.description, + argumentHint: "", + })); + const fallbackSkills = appServerSkills.length === 0 ? await listCodexSkills(this.config.cwd) : []; + return [...appServerSkills, ...fallbackSkills, ...prompts].sort((a, b) => + a.name.localeCompare(b.name) + ); + } + + async executeCommand(commandName: string, args?: string): Promise { + if (commandName.startsWith("prompts:")) { + const promptName = commandName.slice("prompts:".length); + const codexHome = resolveCodexHomeDir(); + const promptPath = path.join(codexHome, "prompts", `${promptName}.md`); + const raw = await fs.readFile(promptPath, "utf8"); + const parsed = parseFrontMatter(raw); + const expanded = expandCodexCustomPrompt(parsed.body, args); + const result = await this.run(expanded); + return { text: result.finalText, timeline: result.timeline, usage: result.usage }; + } + + if (!this.connected) { + await this.connect(); + } else { + await this.loadSkills(); + } + const skill = this.cachedSkills.find((entry) => entry.name === commandName); + if (skill) { + const input = [ + { type: "skill", name: skill.name, path: skill.path }, + ] as unknown as AgentPromptContentBlock[]; + if (args && args.trim().length > 0) { + input.push({ type: "text", text: args.trim() }); + } else { + input.push({ type: "text", text: `$${skill.name}` }); + } + const result = await this.run(input); + return { text: result.finalText, timeline: result.timeline, usage: result.usage }; + } + + const skillPrompt = args ? `$${commandName} ${args}` : `$${commandName}`; + const result = await this.run(skillPrompt); + return { text: result.finalText, timeline: result.timeline, usage: result.usage }; + } + + private async ensureThread(): Promise { + if (!this.client) return; + if (this.currentThreadId) return; + const preset = MODE_PRESETS[this.currentMode] ?? MODE_PRESETS[DEFAULT_CODEX_MODE_ID]; + const approvalPolicy = this.config.approvalPolicy ?? preset.approvalPolicy; + const sandbox = this.config.sandboxMode ?? preset.sandbox; + const innerConfig: Record = {}; + if (this.config.mcpServers) { + const mcpServers: Record = {}; + for (const [name, serverConfig] of Object.entries(this.config.mcpServers)) { + mcpServers[name] = toCodexMcpConfig(serverConfig); + } + innerConfig.mcp_servers = mcpServers; + } + if (this.config.extra?.codex) { + Object.assign(innerConfig, this.config.extra.codex); + } + const response = (await this.client.request("thread/start", { + model: this.config.model ?? null, + cwd: this.config.cwd ?? null, + approvalPolicy, + sandbox, + ...(Object.keys(innerConfig).length > 0 ? { config: innerConfig } : {}), + })) as { thread?: { id?: string } }; + const threadId = response?.thread?.id; + if (!threadId) { + throw new Error("Codex app-server did not return thread id"); + } + this.currentThreadId = threadId; + } + + private buildUserInput(prompt: AgentPromptInput): unknown[] { + if (typeof prompt === "string") { + const text = this.paseoInstructionsInjected + ? prompt + : injectLeadingPaseoInstructionTag(prompt, this.config.paseoPromptInstructions); + this.paseoInstructionsInjected = true; + return [{ type: "text", text }]; + } + const blocks = prompt as AgentPromptContentBlock[]; + if (this.paseoInstructionsInjected) { + return blocks; + } + this.paseoInstructionsInjected = true; + if (blocks.length === 0) { + return blocks; + } + const first = blocks[0]; + if (first && typeof first === "object" && (first as { type?: string }).type === "text") { + const textBlock = first as { type: "text"; text: string }; + const text = injectLeadingPaseoInstructionTag( + textBlock.text ?? "", + this.config.paseoPromptInstructions + ); + return [{ ...textBlock, text }, ...blocks.slice(1)]; + } + const injected = injectLeadingPaseoInstructionTag( + "", + this.config.paseoPromptInstructions + ); + if (injected.trim().length === 0) { + return blocks; + } + return [{ type: "text", text: injected }, ...blocks]; + } + + private emitEvent(event: AgentStreamEvent): void { + if (event.type === "timeline") { + if (event.item.type === "assistant_message") { + this.pendingAgentMessages.clear(); + } + } + this.eventQueue?.push(event); + } + + private handleNotification(method: string, params: unknown): void { + this.emitEvent({ type: "provider_event", provider: CODEX_PROVIDER, raw: { method, params } }); + + switch (method) { + case "thread/started": { + const threadId = (params as any)?.thread?.id; + if (threadId) { + this.currentThreadId = threadId; + this.emitEvent({ type: "thread_started", provider: CODEX_PROVIDER, sessionId: threadId }); + } + break; + } + case "turn/started": { + const turnId = (params as any)?.turn?.id; + if (turnId) { + this.currentTurnId = turnId; + } + this.emitEvent({ type: "turn_started", provider: CODEX_PROVIDER }); + break; + } + case "turn/completed": { + const turn = (params as any)?.turn; + const status = turn?.status; + if (status === "failed") { + const message = turn?.error?.message ?? "Codex turn failed"; + this.emitEvent({ type: "turn_failed", provider: CODEX_PROVIDER, error: message }); + } else if (status === "interrupted") { + this.emitEvent({ type: "turn_canceled", provider: CODEX_PROVIDER, reason: "interrupted" }); + } else { + this.emitEvent({ type: "turn_completed", provider: CODEX_PROVIDER, usage: this.latestUsage }); + } + this.eventQueue?.end(); + break; + } + case "turn/plan/updated": { + const steps = (params as any)?.plan; + if (Array.isArray(steps)) { + const items = planStepsToTodoItems( + steps.map((entry: any) => ({ + step: entry.step ?? "", + status: entry.status ?? "pending", + })) + ); + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: { type: "todo", items }, + }); + } + break; + } + case "turn/diff/updated": { + const diff = (params as any)?.diff; + if (typeof diff === "string" && diff.trim().length > 0) { + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: createToolCallTimelineItem({ + name: "apply_patch", + status: "running", + output: { diff }, + }), + }); + } + break; + } + case "thread/tokenUsage/updated": { + const tokenUsage = (params as any)?.tokenUsage; + this.latestUsage = toAgentUsage(tokenUsage); + break; + } + case "item/agentMessage/delta": { + const itemId = (params as any)?.itemId; + const delta = (params as any)?.delta; + if (itemId && typeof delta === "string") { + const prev = this.pendingAgentMessages.get(itemId) ?? ""; + this.pendingAgentMessages.set(itemId, prev + delta); + } + break; + } + case "item/reasoning/summaryTextDelta": { + const itemId = (params as any)?.itemId; + const delta = (params as any)?.delta; + if (itemId && typeof delta === "string") { + const prev = this.pendingReasoning.get(itemId) ?? []; + prev.push(delta); + this.pendingReasoning.set(itemId, prev); + } + break; + } + case "item/completed": { + const item = (params as any)?.item; + const timelineItem = threadItemToTimeline(item, { includeUserMessage: false }); + if (timelineItem) { + if (timelineItem.type === "assistant_message" && item?.id) { + const buffered = this.pendingAgentMessages.get(item.id); + if (buffered && buffered.length > 0) { + timelineItem.text = buffered; + } + } + if (timelineItem.type === "reasoning" && item?.id) { + const buffered = this.pendingReasoning.get(item.id); + if (buffered && buffered.length > 0) { + timelineItem.text = buffered.join(""); + } + } + this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + } + break; + } + case "item/started": { + const item = (params as any)?.item; + const timelineItem = threadItemToTimeline(item, { includeUserMessage: false }); + if (timelineItem && timelineItem.type === "tool_call") { + this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + } + break; + } + default: + break; + } + } + + private handleCommandApprovalRequest(params: unknown): Promise { + const parsed = params as { + itemId: string; + threadId: string; + turnId: string; + command?: string | null; + cwd?: string | null; + reason?: string | null; + }; + const requestId = `permission-${parsed.itemId}`; + const title = parsed.command ? `Run command: ${parsed.command}` : "Run command"; + const request: AgentPermissionRequest = { + id: requestId, + provider: CODEX_PROVIDER, + name: "CodexBash", + kind: "tool", + title, + description: parsed.reason ?? undefined, + input: { + command: parsed.command ?? undefined, + cwd: parsed.cwd ?? undefined, + }, + metadata: { + itemId: parsed.itemId, + threadId: parsed.threadId, + turnId: parsed.turnId, + }, + }; + this.pendingPermissions.set(requestId, request); + this.emitEvent({ type: "permission_requested", provider: CODEX_PROVIDER, request }); + return new Promise((resolve) => { + this.pendingPermissionHandlers.set(requestId, { resolve, kind: "command" }); + }); + } + + private handleFileChangeApprovalRequest(params: unknown): Promise { + const parsed = params as { itemId: string; threadId: string; turnId: string; reason?: string | null }; + const requestId = `permission-${parsed.itemId}`; + const request: AgentPermissionRequest = { + id: requestId, + provider: CODEX_PROVIDER, + name: "CodexFileChange", + kind: "tool", + title: "Apply file changes", + description: parsed.reason ?? undefined, + metadata: { + itemId: parsed.itemId, + threadId: parsed.threadId, + turnId: parsed.turnId, + }, + }; + this.pendingPermissions.set(requestId, request); + this.emitEvent({ type: "permission_requested", provider: CODEX_PROVIDER, request }); + return new Promise((resolve) => { + this.pendingPermissionHandlers.set(requestId, { resolve, kind: "file" }); + }); + } + + private handleToolApprovalRequest(params: unknown): Promise { + const parsed = params as { itemId: string; threadId: string; turnId: string; questions: any[] }; + const requestId = `permission-${parsed.itemId}`; + const request: AgentPermissionRequest = { + id: requestId, + provider: CODEX_PROVIDER, + name: "CodexTool", + kind: "tool", + title: "Tool action requires approval", + description: undefined, + metadata: { + itemId: parsed.itemId, + threadId: parsed.threadId, + turnId: parsed.turnId, + questions: parsed.questions, + }, + }; + this.pendingPermissions.set(requestId, request); + this.emitEvent({ type: "permission_requested", provider: CODEX_PROVIDER, request }); + return new Promise((resolve) => { + this.pendingPermissionHandlers.set(requestId, { + resolve, + kind: "tool", + questions: Array.isArray(parsed.questions) ? parsed.questions : [], + }); + }); + } +} + +export class CodexAppServerAgentClient implements AgentClient { + readonly provider = CODEX_PROVIDER; + readonly capabilities = CODEX_APP_SERVER_CAPABILITIES; + + constructor(private readonly logger: Logger) {} + + async createSession(config: AgentSessionConfig): Promise { + const sessionConfig: AgentSessionConfig = { ...config, provider: CODEX_PROVIDER }; + const session = new CodexAppServerAgentSession(sessionConfig, null, this.logger); + await session.connect(); + return session; + } + + async resumeSession(handle: { sessionId: string; metadata?: Record }, overrides?: Partial): Promise { + const storedConfig = (handle.metadata ?? {}) as AgentSessionConfig; + const merged: AgentSessionConfig = { + ...storedConfig, + ...overrides, + provider: CODEX_PROVIDER, + cwd: overrides?.cwd ?? storedConfig.cwd ?? process.cwd(), + }; + const session = new CodexAppServerAgentSession(merged, handle, this.logger); + await session.connect(); + return session; + } + + async listPersistedAgents( + options?: ListPersistedAgentsOptions + ): Promise { + const binaryPath = resolveCodexBinary(); + const child = spawn(binaryPath, ["app-server"], { + stdio: ["pipe", "pipe", "pipe"], + }); + const client = new CodexAppServerClient(child, this.logger); + + try { + await client.request("initialize", { + clientInfo: { name: "paseo", title: "Paseo", version: "0.0.0" }, + }); + client.notify("initialized", {}); + + const limit = options?.limit ?? 20; + const response = (await client.request("thread/list", { limit })) as { + data?: Array; + }; + const threads = Array.isArray(response?.data) ? response.data : []; + const descriptors: PersistedAgentDescriptor[] = []; + + for (const thread of threads.slice(0, limit)) { + const threadId = thread.id; + const cwd = thread.cwd ?? process.cwd(); + const title = thread.preview ?? null; + let timeline: AgentTimelineItem[] = []; + try { + const read = (await client.request("thread/read", { + threadId, + includeTurns: true, + })) as { thread?: { turns?: Array<{ items?: any[] }> } }; + const turns = read.thread?.turns ?? []; + const items: AgentTimelineItem[] = []; + for (const turn of turns) { + for (const item of turn.items ?? []) { + const timelineItem = threadItemToTimeline(item); + if (timelineItem) items.push(timelineItem); + } + } + timeline = items; + } catch { + timeline = []; + } + + descriptors.push({ + provider: CODEX_PROVIDER, + sessionId: threadId, + cwd, + title, + lastActivityAt: new Date((thread.updatedAt ?? thread.createdAt ?? 0) * 1000), + persistence: { + provider: CODEX_PROVIDER, + sessionId: threadId, + nativeHandle: threadId, + metadata: { + provider: CODEX_PROVIDER, + cwd, + title, + threadId, + }, + }, + timeline, + }); + } + + return descriptors; + } finally { + await client.dispose(); + } + } + + async listModels(_options?: ListModelsOptions): Promise { + const binaryPath = resolveCodexBinary(); + const child = spawn(binaryPath, ["app-server"], { + stdio: ["pipe", "pipe", "pipe"], + }); + const client = new CodexAppServerClient(child, this.logger); + + try { + await client.request("initialize", { + clientInfo: { + name: "paseo", + title: "Paseo", + version: "0.0.0", + }, + }); + client.notify("initialized", {}); + + const response = (await client.request("model/list", {})) as { data?: Array }; + const models = Array.isArray(response?.data) ? response.data : []; + return models.map((model) => ({ + provider: CODEX_PROVIDER, + id: model.id, + label: model.displayName, + description: model.description, + isDefault: model.isDefault, + metadata: { + model: model.model, + defaultReasoningEffort: model.defaultReasoningEffort, + supportedReasoningEfforts: model.supportedReasoningEfforts, + }, + })); + } finally { + await client.dispose(); + } + } +} diff --git a/packages/server/src/server/daemon-client-v2.e2e.test.ts b/packages/server/src/server/daemon-client-v2.e2e.test.ts index 0828a784e..d959154ff 100644 --- a/packages/server/src/server/daemon-client-v2.e2e.test.ts +++ b/packages/server/src/server/daemon-client-v2.e2e.test.ts @@ -185,13 +185,14 @@ describe("daemon client v2 E2E", () => { return unsubscribe; }); - const failResult = await ctx.client.createAgentExpectFail({ - ...getFullAccessConfig("codex"), - cwd: "/this/path/does/not/exist/12345", - title: "Should Fail", - requestId: failRequestId, - }); - expect(failResult.error).toContain("Working directory does not exist"); + await expect( + ctx.client.createAgent({ + ...getFullAccessConfig("codex"), + cwd: "/this/path/does/not/exist/12345", + title: "Should Fail", + requestId: failRequestId, + }) + ).rejects.toThrow("Working directory does not exist"); await failedStatusPromise; let sawRefresh = false; @@ -783,36 +784,11 @@ describe("daemon client v2 E2E", () => { title: "Git/File Test", }); - const repoInfoRequestId = `repo-info-${Date.now()}`; - const repoInfoMessagePromise = waitForSignal(15000, (resolve) => { - const unsubscribeRepo = ctx.client.on( - "git_repo_info_response", - (message) => { - if (message.type !== "git_repo_info_response") { - return; - } - if (message.payload.cwd !== cwd) { - return; - } - if (message.payload.requestId !== repoInfoRequestId) { - return; - } - resolve(message); - } - ); - return unsubscribeRepo; - }); - - const repoInfo = await ctx.client.getGitRepoInfo( - { cwd }, - repoInfoRequestId - ); - const repoInfoMessage = await repoInfoMessagePromise; - expect(repoInfo.error ?? null).toBeNull(); - expect(repoInfo.repoRoot).toContain(cwd); - expect(repoInfo.requestId).toBe(repoInfoRequestId); - expect(repoInfoMessage.payload.cwd).toBe(cwd); - expect(repoInfoMessage.payload.requestId).toBe(repoInfoRequestId); + // Test checkout status RPC + const checkoutStatus = await ctx.client.getCheckoutStatus(cwd); + expect(checkoutStatus.error).toBeNull(); + expect(checkoutStatus.isGit).toBe(true); + expect(checkoutStatus.repoRoot).toContain(cwd); const diffRequestId = `diff-${Date.now()}`; const diffMessagePromise = waitForSignal(15000, (resolve) => { diff --git a/packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts index e99c1af20..e675b5769 100644 --- a/packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts @@ -93,14 +93,13 @@ describe("daemon E2E", () => { test("fails to create agent with non-existent cwd", async () => { const nonExistentCwd = "/this/path/does/not/exist/12345"; - const result = await ctx.client.createAgentExpectFail({ - provider: "codex", model: CODEX_TEST_MODEL, reasoningEffort: CODEX_TEST_REASONING_EFFORT, - cwd: nonExistentCwd, - title: "Should Fail Agent", - }); - - expect(result.error).toContain("Working directory does not exist"); - expect(result.error).toContain(nonExistentCwd); + await expect( + ctx.client.createAgent({ + provider: "codex", model: CODEX_TEST_MODEL, reasoningEffort: CODEX_TEST_REASONING_EFFORT, + cwd: nonExistentCwd, + title: "Should Fail Agent", + }) + ).rejects.toThrow(nonExistentCwd); }); diff --git a/packages/server/src/server/daemon-e2e/git-operations.e2e.test.ts b/packages/server/src/server/daemon-e2e/git-operations.e2e.test.ts index 2cba0af9e..67be69383 100644 --- a/packages/server/src/server/daemon-e2e/git-operations.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/git-operations.e2e.test.ts @@ -234,7 +234,7 @@ describe("daemon E2E", () => { }); - describe("getGitRepoInfo", () => { + describe("getCheckoutStatus", () => { test( "returns repo info for git repo with branch and dirty state", async () => { @@ -268,16 +268,15 @@ describe("daemon E2E", () => { expect(agent.id).toBeTruthy(); expect(agent.status).toBe("idle"); - // Get git repo info - const result = await ctx.client.getGitRepoInfo(agent.id); + // Get checkout status + const result = await ctx.client.getCheckoutStatus(cwd); // Verify repo info returned without error expect(result.error).toBeNull(); + expect(result.isGit).toBe(true); // macOS symlinks /var to /private/var, so we check containment expect(result.repoRoot).toContain("daemon-e2e-"); expect(result.currentBranch).toBeTruthy(); - expect(result.branches.length).toBeGreaterThan(0); - expect(result.branches.some((b) => b.isCurrent)).toBe(true); expect(result.isDirty).toBe(true); // Cleanup @@ -316,10 +315,11 @@ describe("daemon E2E", () => { expect(agent.id).toBeTruthy(); - // Get git repo info - const result = await ctx.client.getGitRepoInfo(agent.id); + // Get checkout status + const result = await ctx.client.getCheckoutStatus(cwd); expect(result.error).toBeNull(); + expect(result.isGit).toBe(true); expect(result.isDirty).toBe(false); expect(result.currentBranch).toBeTruthy(); @@ -331,7 +331,7 @@ describe("daemon E2E", () => { ); test( - "returns error for non-git directory", + "returns isGit false for non-git directory", async () => { const cwd = tmpCwd(); // Don't initialize git - just a regular directory @@ -345,11 +345,10 @@ describe("daemon E2E", () => { expect(agent.id).toBeTruthy(); - // Get git repo info - should return error - const result = await ctx.client.getGitRepoInfo(agent.id); + // Get checkout status - should return isGit: false + const result = await ctx.client.getCheckoutStatus(cwd); - // Server returns cwd as repoRoot even on error, so we just check for error - expect(result.error).toBeTruthy(); + expect(result.isGit).toBe(false); // Cleanup await ctx.client.deleteAgent(agent.id); diff --git a/packages/server/src/server/dictation/dictation-stream-manager.test.ts b/packages/server/src/server/dictation/dictation-stream-manager.test.ts new file mode 100644 index 000000000..c0581942c --- /dev/null +++ b/packages/server/src/server/dictation/dictation-stream-manager.test.ts @@ -0,0 +1,153 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from "vitest"; +import { EventEmitter } from "node:events"; +import pino from "pino"; + +import { + DictationStreamManager, + type RealtimeTranscriptionSession, + type RealtimeTranscriptionSessionFactory, +} from "./dictation-stream-manager.js"; + +class FakeRealtimeSession extends EventEmitter implements RealtimeTranscriptionSession { + connected = false; + appended: string[] = []; + commitCalls = 0; + clearCalls = 0; + closed = false; + + async connect(): Promise { + this.connected = true; + } + + appendPcm16Base64(base64Audio: string): void { + this.appended.push(base64Audio); + } + + commit(): void { + this.commitCalls += 1; + } + + clear(): void { + this.clearCalls += 1; + } + + close(): void { + this.closed = true; + } + + emitCommitted(itemId: string): void { + this.emit("committed", { itemId, previousItemId: null }); + } + + emitTranscript(itemId: string, transcript: string, isFinal: boolean): void { + this.emit("transcript", { itemId, transcript, isFinal }); + } + + emitError(message: string): void { + this.emit("error", new Error(message)); + } +} + +const buildPcmBase64 = (sampleValue: number, sampleCount: number): string => { + const samples = new Int16Array(sampleCount); + samples.fill(sampleValue); + return Buffer.from(samples.buffer).toString("base64"); +}; + +const tick = async (): Promise => { + await Promise.resolve(); + await Promise.resolve(); +}; + +describe("DictationStreamManager (semantic VAD grace fallback)", () => { + const env = { + turnDetection: process.env.OPENAI_REALTIME_DICTATION_TURN_DETECTION, + dictationDebug: process.env.PASEO_DICTATION_DEBUG, + }; + + beforeEach(() => { + vi.useFakeTimers(); + process.env.OPENAI_REALTIME_DICTATION_TURN_DETECTION = "semantic_vad"; + process.env.PASEO_DICTATION_DEBUG = "false"; + }); + + afterEach(() => { + vi.useRealTimers(); + process.env.OPENAI_REALTIME_DICTATION_TURN_DETECTION = env.turnDetection; + process.env.PASEO_DICTATION_DEBUG = env.dictationDebug; + }); + + it("treats buffer-too-small as benign and finalizes with existing transcripts", async () => { + const session = new FakeRealtimeSession(); + const factory: RealtimeTranscriptionSessionFactory = () => session; + const emitted: Array<{ type: string; payload: any }> = []; + const manager = new DictationStreamManager({ + logger: pino({ level: "silent" }), + emit: (msg) => emitted.push(msg), + sessionId: "s1", + openaiApiKey: "k", + sessionFactory: factory, + finalTimeoutMs: 5000, + }); + + await manager.handleStart("d1", "audio/pcm;rate=24000;bits=16"); + await manager.handleChunk({ + dictationId: "d1", + seq: 0, + audioBase64: buildPcmBase64(2000, 2400), + format: "audio/pcm;rate=24000;bits=16", + }); + + session.emitTranscript("i1", "hello world", true); + + await manager.handleFinish("d1", 0); + await tick(); + + vi.advanceTimersByTime(2000); + await tick(); + + session.emitError( + "Error committing input audio buffer: buffer too small. Expected at least 100ms of audio, but buffer only has 0.00ms of audio." + ); + await tick(); + + const final = emitted.find((msg) => msg.type === "dictation_stream_final"); + const error = emitted.find((msg) => msg.type === "dictation_stream_error"); + expect(error).toBeUndefined(); + expect(final?.payload.text).toBe("hello world"); + expect(session.closed).toBe(true); + }); + + it("does not fallback-commit if committed event arrives during grace window", async () => { + const session = new FakeRealtimeSession(); + const factory: RealtimeTranscriptionSessionFactory = () => session; + const emitted: Array<{ type: string; payload: any }> = []; + const manager = new DictationStreamManager({ + logger: pino({ level: "silent" }), + emit: (msg) => emitted.push(msg), + sessionId: "s1", + openaiApiKey: "k", + sessionFactory: factory, + finalTimeoutMs: 5000, + }); + + await manager.handleStart("d1", "audio/pcm;rate=24000;bits=16"); + await manager.handleChunk({ + dictationId: "d1", + seq: 0, + audioBase64: buildPcmBase64(2000, 2400), + format: "audio/pcm;rate=24000;bits=16", + }); + + await manager.handleFinish("d1", 0); + session.emitCommitted("i1"); + session.emitTranscript("i1", "hi there", true); + + vi.advanceTimersByTime(2000); + await tick(); + + expect(session.commitCalls).toBe(0); + const final = emitted.find((msg) => msg.type === "dictation_stream_final"); + expect(final?.payload.text).toBe("hi there"); + }); +}); diff --git a/packages/server/src/server/dictation/dictation-stream-manager.ts b/packages/server/src/server/dictation/dictation-stream-manager.ts index 4ae9ec317..20a9581cf 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.ts +++ b/packages/server/src/server/dictation/dictation-stream-manager.ts @@ -12,7 +12,11 @@ import { OpenAIRealtimeTranscriptionSession } from "../agent/openai-realtime-tra const PCM_CHANNELS = 1; const PCM_BITS_PER_SAMPLE = 16; const DICTATION_PCM_OUTPUT_RATE = 24000; -const DEFAULT_DICTATION_FINAL_TIMEOUT_MS = 30000; +const DEFAULT_DICTATION_FINAL_TIMEOUT_MS = 10000; +const DICTATION_VAD_GRACE_TIMEOUT_MS = Number.parseInt( + process.env.OPENAI_REALTIME_DICTATION_VAD_GRACE_TIMEOUT_MS ?? "2000", + 10 +); const DICTATION_SILENCE_PEAK_THRESHOLD = Number.parseInt( process.env.OPENAI_REALTIME_DICTATION_SILENCE_PEAK_THRESHOLD ?? "300", 10 @@ -101,6 +105,15 @@ export type RealtimeTranscriptionSession = { on(event: "error", handler: (err: unknown) => void): unknown; }; +export type RealtimeTranscriptionSessionFactory = (params: { + apiKey: string; + logger: pino.Logger; + transcriptionModel: string; + language?: string; + prompt?: string; + turnDetection: OpenAITurnDetection; +}) => RealtimeTranscriptionSession; + function convertPCMToWavBuffer( pcmBuffer: Buffer, sampleRate: number, @@ -149,10 +162,13 @@ type DictationStreamState = { transcriptsByItemId: Map; finalTranscriptItemIds: Set; awaitingFinalCommit: boolean; + vadGraceTimeout: ReturnType | null; + fallbackCommitAttempted: boolean; finishRequested: boolean; finishSealed: boolean; finalSeq: number | null; finalTimeout: ReturnType | null; + isSemanticVad: boolean; }; export type DictationStreamOutboundMessage = @@ -177,6 +193,7 @@ export class DictationStreamManager { private readonly sessionId: string; private readonly openaiApiKey: string | null; private readonly finalTimeoutMs: number; + private readonly createSession: RealtimeTranscriptionSessionFactory; private readonly streams = new Map(); constructor(params: { @@ -185,12 +202,17 @@ export class DictationStreamManager { sessionId: string; openaiApiKey?: string | null; finalTimeoutMs?: number; + sessionFactory?: RealtimeTranscriptionSessionFactory; }) { this.logger = params.logger.child({ component: "dictation-stream-manager" }); this.emit = params.emit; this.sessionId = params.sessionId; this.openaiApiKey = params.openaiApiKey ?? null; this.finalTimeoutMs = params.finalTimeoutMs ?? DEFAULT_DICTATION_FINAL_TIMEOUT_MS; + this.createSession = + params.sessionFactory ?? + ((factoryParams) => + new OpenAIRealtimeTranscriptionSession(factoryParams)); } public cleanupAll(): void { @@ -214,13 +236,14 @@ export class DictationStreamManager { process.env.OPENAI_REALTIME_DICTATION_TRANSCRIPTION_PROMPT ?? "Transcribe only what the speaker says. Do not add words. Preserve punctuation and casing. If the audio is silence or non-speech noise, return an empty transcript."; - const openai = new OpenAIRealtimeTranscriptionSession({ + const turnDetection = parseDictationTurnDetection(); + const openai = this.createSession({ apiKey, logger: this.logger.child({ dictationId }), transcriptionModel, language: "en", prompt: transcriptionPrompt, - turnDetection: parseDictationTurnDetection(), + turnDetection, }); openai.on("committed", ({ itemId }: { itemId: string }) => { @@ -228,6 +251,7 @@ export class DictationStreamManager { if (!state) { return; } + this.clearVadGraceTimeout(state); state.committedItemIds.push(itemId); state.bytesSinceCommit = 0; state.peakSinceCommit = 0; @@ -263,6 +287,7 @@ export class DictationStreamManager { // If we triggered a finish commit but OpenAI doesn't emit committed events (or they arrive late), // allow final transcripts to unblock finalization. if (state.finishRequested && state.awaitingFinalCommit && isFinal) { + this.clearVadGraceTimeout(state); state.awaitingFinalCommit = false; } @@ -282,6 +307,15 @@ export class DictationStreamManager { openai.on("error", (err) => { const message = err instanceof Error ? err.message : String(err); + const state = this.streams.get(dictationId); + if (state && state.finishRequested && isBufferTooSmallError(message)) { + this.clearVadGraceTimeout(state); + if (state.awaitingFinalCommit) { + state.awaitingFinalCommit = false; + } + this.maybeFinalizeDictationStream(dictationId); + return; + } void this.failAndCleanupDictationStream(dictationId, message, true); }); @@ -329,10 +363,13 @@ export class DictationStreamManager { transcriptsByItemId: new Map(), finalTranscriptItemIds: new Set(), awaitingFinalCommit: false, + vadGraceTimeout: null, + fallbackCommitAttempted: false, finishRequested: false, finishSealed: false, finalSeq: null, finalTimeout: null, + isSemanticVad: turnDetection?.type === "semantic_vad", }); this.emitDictationAck(dictationId, -1); @@ -529,6 +566,7 @@ export class DictationStreamManager { if (!state) { return; } + this.clearVadGraceTimeout(state); if (state.finalTimeout) { clearTimeout(state.finalTimeout); } @@ -585,13 +623,17 @@ export class DictationStreamManager { state.bytesSinceCommit += silenceBytes; } - try { - state.awaitingFinalCommit = true; - state.openai.commit(); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - void this.failAndCleanupDictationStream(dictationId, message, true); - return; + state.awaitingFinalCommit = true; + if (state.isSemanticVad) { + this.startVadGraceTimeout(state); + } else { + try { + state.openai.commit(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + void this.failAndCleanupDictationStream(dictationId, message, true); + return; + } } } } else { @@ -690,4 +732,43 @@ export class DictationStreamManager { this.cleanupDictationStream(dictationId); })(); } + + private startVadGraceTimeout(state: DictationStreamState): void { + if (state.vadGraceTimeout || DICTATION_VAD_GRACE_TIMEOUT_MS <= 0) { + return; + } + state.vadGraceTimeout = setTimeout(() => { + state.vadGraceTimeout = null; + if (!state.finishRequested || !state.awaitingFinalCommit) { + return; + } + if (state.bytesSinceCommit <= 0 || state.fallbackCommitAttempted) { + return; + } + state.fallbackCommitAttempted = true; + try { + state.openai.commit(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (isBufferTooSmallError(message)) { + state.awaitingFinalCommit = false; + this.maybeFinalizeDictationStream(state.dictationId); + return; + } + void this.failAndCleanupDictationStream(state.dictationId, message, true); + } + }, DICTATION_VAD_GRACE_TIMEOUT_MS); + } + + private clearVadGraceTimeout(state: DictationStreamState): void { + if (!state.vadGraceTimeout) { + return; + } + clearTimeout(state.vadGraceTimeout); + state.vadGraceTimeout = null; + } +} + +function isBufferTooSmallError(message: string): boolean { + return /buffer too small/i.test(message); } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index d4857c7ae..201439867 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -922,10 +922,6 @@ export class Session { await this.handleFileDownloadTokenRequest(msg); break; - case "git_repo_info_request": - await this.handleGitRepoInfoRequest(msg); - break; - case "list_provider_models_request": await this.handleListProviderModelsRequest(msg); break; @@ -1441,12 +1437,17 @@ export class Session { } if (requestId) { + const agentPayload = await this.getAgentPayloadById(snapshot.id); + if (!agentPayload) { + throw new Error(`Agent ${snapshot.id} not found after creation`); + } this.emit({ type: "status", payload: { status: "agent_created", agentId: snapshot.id, requestId, + agent: agentPayload, }, }); } @@ -1516,6 +1517,10 @@ export class Session { await this.forwardAgentUpdate(snapshot); const timelineSize = this.emitAgentTimelineSnapshot(snapshot); if (requestId) { + const agentPayload = await this.getAgentPayloadById(snapshot.id); + if (!agentPayload) { + throw new Error(`Agent ${snapshot.id} not found after resume`); + } this.emit({ type: "status", payload: { @@ -1523,6 +1528,7 @@ export class Session { agentId: snapshot.id, requestId, timelineSize, + agent: agentPayload, }, }); } @@ -1787,59 +1793,6 @@ export class Session { } } - private async handleGitRepoInfoRequest( - msg: Extract - ): Promise { - const { cwd, requestId } = msg; - const resolvedCwd = expandTilde(cwd); - - try { - const status = await getCheckoutStatus(resolvedCwd, { paseoHome: this.paseoHome }); - if (!status.isGit) { - throw new NotGitRepoError(resolvedCwd); - } - const repoRoot = status.repoRoot ?? resolvedCwd; - const { stdout: branchesRaw } = await execAsync( - "git branch --format='%(refname:short)'", - { cwd: repoRoot, env: READ_ONLY_GIT_ENV } - ); - const currentBranch = status.currentBranch ?? ""; - const branches = branchesRaw - .split("\n") - .map((line) => line.trim()) - .filter((line) => line.length > 0) - .map((name) => ({ - name, - isCurrent: name === currentBranch, - })); - - const isDirty = status.isDirty ?? false; - - this.emit({ - type: "git_repo_info_response", - payload: { - cwd: resolvedCwd, - repoRoot, - requestId, - branches, - currentBranch: currentBranch || null, - isDirty, - error: null, - }, - }); - } catch (error) { - this.emit({ - type: "git_repo_info_response", - payload: { - cwd, - repoRoot: cwd, - requestId, - error: (error as Error)?.message ?? String(error), - }, - }); - } - } - private async handleListProviderModelsRequest( msg: Extract ): Promise { diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index 33d61b925..728c67eda 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -623,6 +623,13 @@ export const CheckoutPrStatusRequestSchema = z.object({ requestId: z.string(), }); +export const ValidateBranchRequestSchema = z.object({ + type: z.literal("validate_branch_request"), + cwd: z.string(), + branchName: z.string(), + requestId: z.string(), +}); + export const PaseoWorktreeListRequestSchema = z.object({ type: z.literal("paseo_worktree_list_request"), cwd: z.string().optional(), @@ -842,13 +849,13 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ CheckoutPushRequestSchema, CheckoutPrCreateRequestSchema, CheckoutPrStatusRequestSchema, + ValidateBranchRequestSchema, PaseoWorktreeListRequestSchema, PaseoWorktreeArchiveRequestSchema, HighlightedDiffRequestSchema, FileExplorerRequestSchema, ProjectIconRequestSchema, FileDownloadTokenRequestSchema, - GitRepoInfoRequestMessageSchema, ClearAgentAttentionMessageSchema, ClientHeartbeatMessageSchema, ListCommandsRequestSchema, @@ -979,6 +986,7 @@ const AgentStatusWithTimelineSchema = AgentStatusWithRequestSchema.extend({ export const AgentCreatedStatusPayloadSchema = z .object({ status: z.literal("agent_created"), + agent: AgentSnapshotPayloadSchema, }) .extend(AgentStatusWithRequestSchema.shape); @@ -991,6 +999,7 @@ export const AgentCreateFailedStatusPayloadSchema = z.object({ export const AgentResumedStatusPayloadSchema = z .object({ status: z.literal("agent_resumed"), + agent: AgentSnapshotPayloadSchema, }) .extend(AgentStatusWithTimelineSchema.shape); @@ -1333,6 +1342,17 @@ export const CheckoutPrStatusResponseSchema = z.object({ }), }); +export const ValidateBranchResponseSchema = z.object({ + type: z.literal("validate_branch_response"), + payload: z.object({ + exists: z.boolean(), + resolvedRef: z.string().nullable(), + isRemote: z.boolean(), + error: z.string().nullable(), + requestId: z.string(), + }), +}); + const PaseoWorktreeSchema = z.object({ worktreePath: z.string(), branchName: z.string().nullable().optional(), @@ -1410,24 +1430,6 @@ export const FileDownloadTokenResponseSchema = z.object({ }), }); -const GitBranchInfoSchema = z.object({ - name: z.string(), - isCurrent: z.boolean(), -}); - -export const GitRepoInfoResponseSchema = z.object({ - type: z.literal("git_repo_info_response"), - payload: z.object({ - cwd: z.string(), - repoRoot: z.string(), - requestId: z.string(), - branches: z.array(GitBranchInfoSchema).optional(), - currentBranch: z.string().nullable().optional(), - isDirty: z.boolean().optional(), - error: z.string().nullable().optional(), - }), -}); - export const ListProviderModelsResponseMessageSchema = z.object({ type: z.literal("list_provider_models_response"), payload: z.object({ @@ -1581,13 +1583,13 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ CheckoutPushResponseSchema, CheckoutPrCreateResponseSchema, CheckoutPrStatusResponseSchema, + ValidateBranchResponseSchema, PaseoWorktreeListResponseSchema, PaseoWorktreeArchiveResponseSchema, HighlightedDiffResponseSchema, FileExplorerResponseSchema, ProjectIconResponseSchema, FileDownloadTokenResponseSchema, - GitRepoInfoResponseSchema, ListProviderModelsResponseMessageSchema, ListCommandsResponseSchema, ExecuteCommandResponseSchema, @@ -1685,6 +1687,8 @@ export type CheckoutPrCreateRequest = z.infer; export type CheckoutPrStatusRequest = z.infer; export type CheckoutPrStatusResponse = z.infer; +export type ValidateBranchRequest = z.infer; +export type ValidateBranchResponse = z.infer; export type PaseoWorktreeListRequest = z.infer; export type PaseoWorktreeListResponse = z.infer; export type PaseoWorktreeArchiveRequest = z.infer; @@ -1698,7 +1702,6 @@ export type ProjectIconResponse = z.infer; export type ProjectIcon = z.infer; export type FileDownloadTokenRequest = z.infer; export type FileDownloadTokenResponse = z.infer; -export type GitRepoInfoResponse = z.infer; export type RestartServerRequestMessage = z.infer; export type ClearAgentAttentionMessage = z.infer; export type ClientHeartbeatMessage = z.infer; diff --git a/packages/server/src/utils/worktree.ts b/packages/server/src/utils/worktree.ts index ba30eaf5a..9de91341b 100644 --- a/packages/server/src/utils/worktree.ts +++ b/packages/server/src/utils/worktree.ts @@ -542,10 +542,19 @@ export async function createWorktree({ if (normalizedBaseBranch === "HEAD") { throw new Error("Base branch cannot be HEAD when creating a Paseo worktree"); } + + // Resolve the base branch - try local first, then remote + let resolvedBaseBranch = normalizedBaseBranch; try { await execAsync(`git rev-parse --verify ${normalizedBaseBranch}`, { cwd }); } catch { - throw new Error(`Base branch not found: ${normalizedBaseBranch}`); + // Local branch doesn't exist, try remote (origin/{branch}) + try { + await execAsync(`git rev-parse --verify origin/${normalizedBaseBranch}`, { cwd }); + resolvedBaseBranch = `origin/${normalizedBaseBranch}`; + } catch { + throw new Error(`Base branch not found: ${normalizedBaseBranch}`); + } } let worktreePath: string; @@ -568,8 +577,8 @@ export async function createWorktree({ // Always create a new branch for the worktree // If branchName already exists, use it as base and create worktree-slug as branch name - // If branchName doesn't exist, create it from baseBranch - const base = branchExists ? branchName : normalizedBaseBranch; + // If branchName doesn't exist, create it from baseBranch (resolved to remote if needed) + const base = branchExists ? branchName : resolvedBaseBranch; const candidateBranch = branchExists ? desiredSlug : branchName; // Find unique branch name if collision