diff --git a/README.md b/README.md index 19138fc33..1da3c612f 100644 --- a/README.md +++ b/README.md @@ -2,11 +2,11 @@ [![Node.js](https://img.shields.io/badge/node-%3E%3D22-339933?logo=nodedotjs&logoColor=white)](https://nodejs.org) [![TypeScript](https://img.shields.io/badge/TypeScript-6.0-3178C6?logo=typescript&logoColor=white)](https://www.typescriptlang.org/) -[![Backends](https://img.shields.io/badge/backends-Claude_%7C_Kilo_%7C_OpenCode-D97706)](#backends) +[![Backends](https://img.shields.io/badge/backends-Claude_%7C_Kilo_%7C_OpenCode_%7C_Codex-D97706)](#backends) [![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE) [![CI](https://github.com/dylanneve1/talon/actions/workflows/ci.yml/badge.svg)](https://github.com/dylanneve1/talon/actions/workflows/ci.yml) -Multi-platform agentic AI harness. Runs on **Telegram**, **Discord**, **Microsoft Teams**, and the **Terminal**, with a pluggable backend (**Claude Agent SDK**, **Kilo**, or **OpenCode**) and full tool access through MCP. +Multi-platform agentic AI harness. Runs on **Telegram**, **Discord**, **Microsoft Teams**, and the **Terminal**, with a pluggable backend (**Claude Agent SDK**, **Kilo**, **OpenCode**, or **Codex**) and full tool access through MCP. --- @@ -46,6 +46,7 @@ npx talon chat # terminal chat mode - `claude` backend: [Claude Code](https://docs.anthropic.com/en/docs/claude-code) installed and authenticated (`claude` CLI on PATH). - `kilo` backend: nothing extra — `@kilocode/sdk` spawns a local server. Free models are accessible without auth; routed models use Kilo's own credentials. - `opencode` backend: nothing extra — `@opencode-ai/sdk` spawns a local server. + - `codex` backend: install the `codex` CLI (`npm i -g @openai/codex`) and authenticate with `codex login` (ChatGPT auth or `OPENAI_API_KEY`). - Talon runs from a normal source or package install; standalone compiled binaries are not supported. --- @@ -76,6 +77,7 @@ index.ts Composition root | +-- claude-sdk/ Claude Agent SDK (in-process MCP, hooks) | +-- kilo/ Kilo HTTP server backend (streaming via SSE) | +-- opencode/ OpenCode HTTP server backend + | +-- codex/ Codex CLI backend (`@openai/codex-sdk`) | +-- frontend/ | +-- telegram/ Grammy bot + GramJS userbot @@ -96,13 +98,14 @@ index.ts Composition root Select via the `backend` field in `~/.talon/config.json`. All backends implement the same `QueryBackend` interface — heartbeat, dream, and chat handlers are backend-agnostic. -| Backend | `backend` value | Transport | Notes | -| ---------- | --------------- | ----------------------------------------------- | ----------------------------------------------------------------------- | -| Claude SDK | `"claude"` | In-process via `@anthropic-ai/claude-agent-sdk` | Requires the `claude` CLI on `PATH`. Hook-based turn termination. | -| Kilo | `"kilo"` | Local HTTP server via `@kilocode/sdk` | SSE-streamed turns. Routes to many model providers via Kilo's auth. | -| OpenCode | `"opencode"` | Local HTTP server via `@opencode-ai/sdk` | Synchronous prompt; same MCP and session shape as Kilo (upstream fork). | +| Backend | `backend` value | Transport | Notes | +| ---------- | --------------- | ------------------------------------------------------ | ----------------------------------------------------------------------- | +| Claude SDK | `"claude"` | In-process via `@anthropic-ai/claude-agent-sdk` | Requires the `claude` CLI on `PATH`. Hook-based turn termination. | +| Kilo | `"kilo"` | Local HTTP server via `@kilocode/sdk` | SSE-streamed turns. Routes to many model providers via Kilo's auth. | +| OpenCode | `"opencode"` | Local HTTP server via `@opencode-ai/sdk` | SSE-streamed turns; same MCP and session shape as Kilo (upstream fork). | +| Codex | `"codex"` | Per-turn subprocess via `@openai/codex-sdk` | Requires the `codex` CLI from `@openai/codex` and an OpenAI API key (or ChatGPT auth). MCP servers configured via TOML overrides at thread start. | -The Kilo and OpenCode backends share infrastructure (`backend/remote-server/`) since the upstream HTTP API is the same; each backend supplies its own SDK client, port, and delivery suffix. +The Kilo and OpenCode backends share infrastructure (`backend/remote-server/`) since the upstream HTTP API is the same; each backend supplies its own SDK client, port, and delivery suffix. Codex is its own integration on top of the Codex CLI's JSONL event stream. --- @@ -244,7 +247,7 @@ Config file: `~/.talon/config.json` | Field | Default | Description | | -------------------------- | ------------ | --------------------------------------------------------------- | | `frontend` | `"telegram"` | `"telegram"`, `"discord"`, `"teams"`, `"terminal"`, or an array | -| `backend` | `"claude"` | `"claude"`, `"kilo"`, or `"opencode"` | +| `backend` | `"claude"` | `"claude"`, `"kilo"`, `"opencode"`, or `"codex"` | | `botToken` | --- | Telegram bot token | | `model` | `"default"` | Default model. Interpretation depends on the active backend. | | `concurrency` | `1` | Max concurrent AI queries (1--20) | diff --git a/package-lock.json b/package-lock.json index ab14b8ecf..333a5b939 100644 --- a/package-lock.json +++ b/package-lock.json @@ -17,6 +17,7 @@ "@grammyjs/transformer-throttler": "^1.2.1", "@kilocode/sdk": "^7.2.22", "@modelcontextprotocol/sdk": "^1.29.0", + "@openai/codex-sdk": "^0.130.0", "@opencode-ai/sdk": "^1.4.0", "@playwright/mcp": "^0.0.75", "big-integer": "^1.6.52", @@ -6403,6 +6404,140 @@ "peerDependencies": { "zod": "^3.25.28 || ^4" } + }, + "node_modules/@openai/codex": { + "version": "0.130.0", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.130.0.tgz", + "integrity": "sha512-WGDj+RZ3TXWC/7MlwprgLWOqzpwatPIINPhP3IRzHA0ni+o3QZ4i4xrS2uWwGmHUJ395J5JHwoZAAZYyfJyz6w==", + "license": "Apache-2.0", + "bin": { + "codex": "bin/codex.js" + }, + "engines": { + "node": ">=16" + }, + "optionalDependencies": { + "@openai/codex-darwin-arm64": "npm:@openai/codex@0.130.0-darwin-arm64", + "@openai/codex-darwin-x64": "npm:@openai/codex@0.130.0-darwin-x64", + "@openai/codex-linux-arm64": "npm:@openai/codex@0.130.0-linux-arm64", + "@openai/codex-linux-x64": "npm:@openai/codex@0.130.0-linux-x64", + "@openai/codex-win32-arm64": "npm:@openai/codex@0.130.0-win32-arm64", + "@openai/codex-win32-x64": "npm:@openai/codex@0.130.0-win32-x64" + } + }, + "node_modules/@openai/codex-darwin-arm64": { + "name": "@openai/codex", + "version": "0.130.0-darwin-arm64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.130.0-darwin-arm64.tgz", + "integrity": "sha512-R9pkGC7kwC8yQ8el5hvBlmugQlcsG/pHMEFgZluu03X9fD2TezGxdq3KqRDRCZuMYl07ILamVEoqknuJ0cq7MA==", + "cpu": [ + "arm64" + ], + "license": "Apache-2.0", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@openai/codex-darwin-x64": { + "name": "@openai/codex", + "version": "0.130.0-darwin-x64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.130.0-darwin-x64.tgz", + "integrity": "sha512-gJ+7J8djevgtdra+NgDAiQQPW+O3KTsgGfE3E5dpDfww3zS5OCeV0V2dhxqnJdlOjOSDw99o0P2LqBv19mhpRw==", + "cpu": [ + "x64" + ], + "license": "Apache-2.0", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@openai/codex-linux-arm64": { + "name": "@openai/codex", + "version": "0.130.0-linux-arm64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.130.0-linux-arm64.tgz", + "integrity": "sha512-tFtH0V9/hEI3d9y7zP92BXI9FM4Z3+STNQaOR52Czv18TRtCFUp7CbIUYaToopuq6UBfnE1VKr8RLhwT5FcbmA==", + "cpu": [ + "arm64" + ], + "license": "Apache-2.0", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@openai/codex-linux-x64": { + "name": "@openai/codex", + "version": "0.130.0-linux-x64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.130.0-linux-x64.tgz", + "integrity": "sha512-3VcNlez99xdnEf+kB1IOpWv9fICYV9PiGj4sLCO4TCcShLnyxe+YBGa3poknkvXLnMG0qiN9SMnYS2FGrMxQcA==", + "cpu": [ + "x64" + ], + "license": "Apache-2.0", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@openai/codex-sdk": { + "version": "0.130.0", + "resolved": "https://registry.npmjs.org/@openai/codex-sdk/-/codex-sdk-0.130.0.tgz", + "integrity": "sha512-ICKaZ5zrIDg71AiQcsUToVoe5Icmrc3LwSM5+2z7Cf8F1x6nOaY7/ucpFlr4aH8oDe7t3dangc+MsWZTkdvDFw==", + "license": "Apache-2.0", + "dependencies": { + "@openai/codex": "0.130.0" + }, + "engines": { + "node": ">=18" + } + }, + "node_modules/@openai/codex-win32-arm64": { + "name": "@openai/codex", + "version": "0.130.0-win32-arm64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.130.0-win32-arm64.tgz", + "integrity": "sha512-vdpmiNp57L/arZabltLXn8TyEtNa7W1meOEkr+3R6W/8ZyBt++wuqz1Orv134OT2grrcFJsIVCAIPiqUxCvBkA==", + "cpu": [ + "arm64" + ], + "license": "Apache-2.0", + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@openai/codex-win32-x64": { + "name": "@openai/codex", + "version": "0.130.0-win32-x64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.130.0-win32-x64.tgz", + "integrity": "sha512-FzMznm7fr5/nbjZgOujZ9Y9AbdGm7ji1FOoWiY3U+srqauvZaTgn6o6aCheSL7kuymu7nTLOO/cAyWV6NuesqQ==", + "cpu": [ + "x64" + ], + "license": "Apache-2.0", + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=16" + } } } } diff --git a/package.json b/package.json index c80aa12a8..572f5768e 100644 --- a/package.json +++ b/package.json @@ -77,6 +77,7 @@ "@grammyjs/transformer-throttler": "^1.2.1", "@kilocode/sdk": "^7.2.22", "@modelcontextprotocol/sdk": "^1.29.0", + "@openai/codex-sdk": "^0.130.0", "@opencode-ai/sdk": "^1.4.0", "@playwright/mcp": "^0.0.75", "big-integer": "^1.6.52", diff --git a/src/__tests__/codex-backend.test.ts b/src/__tests__/codex-backend.test.ts new file mode 100644 index 000000000..e24e0c2c5 --- /dev/null +++ b/src/__tests__/codex-backend.test.ts @@ -0,0 +1,200 @@ +/** + * Codex backend — unit tests covering the testable surface. + * + * Codex spawns a real CLI subprocess at thread-creation time, so the + * full handler isn't exercisable without the `codex` binary on PATH + + * a valid OpenAI API key. These tests focus on the parts that don't + * require that: factory registration, MCP-config flattening, state + * lifecycle. + * + * Live verification: a `docker/codex-test/` harness can be added in a + * follow-up similar to `docker/kilo-test/` once the SDK has been + * shaken out in production. + */ + +import { describe, it, expect, beforeEach, vi } from "vitest"; + +vi.mock("../core/plugin.js", () => ({ + getPluginMcpServers: vi.fn(() => ({ + "mempalace-tools": { + command: "python", + args: ["-m", "mempalace.mcp_server"], + env: { MEMPALACE_PATH: "/palace" }, + }, + })), +})); + +import { buildCodexMcpServers } from "../backend/codex/mcp-config.js"; +import { getState, resetState } from "../backend/codex/state.js"; +import { initCodexAgent } from "../backend/codex/init.js"; +import { + CODEX_DEFAULT_MODEL, + CODEX_SYSTEM_PROMPT_SUFFIX, +} from "../backend/codex/constants.js"; + +describe("codex / constants", () => { + it("exposes a non-empty default model identifier", () => { + expect(CODEX_DEFAULT_MODEL).toBe("gpt-5-codex"); + }); + + it("documents both delivery routes in the system prompt suffix", () => { + expect(CODEX_SYSTEM_PROMPT_SUFFIX).toContain("end_turn"); + expect(CODEX_SYSTEM_PROMPT_SUFFIX).toContain("send"); + expect(CODEX_SYSTEM_PROMPT_SUFFIX).toContain("react"); + expect(CODEX_SYSTEM_PROMPT_SUFFIX).toContain("Codex Delivery"); + }); +}); + +describe("codex / state lifecycle", () => { + beforeEach(() => { + resetState(); + }); + + it("starts in an uninitialised state", () => { + expect(getState().config).toBeNull(); + expect(getState().codex).toBeNull(); + expect(getState().frontendName).toBe("telegram"); + }); + + it("captures config + gateway port + frontend on initCodexAgent", () => { + initCodexAgent({ model: "gpt-5-codex" } as never, () => 12345, "discord"); + + const state = getState(); + expect(state.config).toBeTruthy(); + expect(state.gatewayPortFn()).toBe(12345); + expect(state.frontendName).toBe("discord"); + }); + + it("resetState clears everything", () => { + initCodexAgent({} as never, () => 999, "teams"); + expect(getState().config).not.toBeNull(); + + resetState(); + + expect(getState().config).toBeNull(); + expect(getState().codex).toBeNull(); + expect(getState().frontendName).toBe("telegram"); + }); +}); + +describe("codex / buildCodexMcpServers", () => { + it("emits one frontend-tools server per non-terminal frontend", () => { + const servers = buildCodexMcpServers({ + chatId: "352042062", + bridgeUrl: "http://127.0.0.1:19876", + frontends: ["telegram"], + }); + expect(servers["telegram-tools"]).toBeDefined(); + expect(servers["telegram-tools"].env).toMatchObject({ + TALON_CHAT_ID: "352042062", + TALON_FRONTEND: "telegram", + TALON_BRIDGE_URL: "http://127.0.0.1:19876", + }); + }); + + it("emits one server per frontend when multiple are configured", () => { + const servers = buildCodexMcpServers({ + chatId: "c1", + bridgeUrl: "http://127.0.0.1:19876", + frontends: ["telegram", "discord"], + }); + expect(servers["telegram-tools"]).toBeDefined(); + expect(servers["discord-tools"]).toBeDefined(); + expect(servers["telegram-tools"].env?.TALON_FRONTEND).toBe("telegram"); + expect(servers["discord-tools"].env?.TALON_FRONTEND).toBe("discord"); + }); + + it("includes the plugin MCP servers from getPluginMcpServers", () => { + const servers = buildCodexMcpServers({ + chatId: "c1", + bridgeUrl: "http://127.0.0.1:19876", + frontends: ["telegram"], + }); + expect(servers["mempalace-tools"]).toBeDefined(); + expect(servers["mempalace-tools"].env).toMatchObject({ + MEMPALACE_PATH: "/palace", + }); + }); + + it("includes brave-search when braveApiKey is provided", () => { + const servers = buildCodexMcpServers({ + chatId: "c1", + bridgeUrl: "http://127.0.0.1:19876", + frontends: ["telegram"], + braveApiKey: "BSA-test-key", + }); + expect(servers["brave-search"]).toBeDefined(); + expect(servers["brave-search"].env).toMatchObject({ + BRAVE_API_KEY: "BSA-test-key", + }); + }); + + it("omits brave-search when no API key is provided", () => { + const servers = buildCodexMcpServers({ + chatId: "c1", + bridgeUrl: "http://127.0.0.1:19876", + frontends: ["telegram"], + }); + expect(servers["brave-search"]).toBeUndefined(); + }); + + it("emits no frontend-tools server when only terminal is configured", () => { + const servers = buildCodexMcpServers({ + chatId: "c1", + bridgeUrl: "http://127.0.0.1:19876", + frontends: [], // caller filters terminal out + }); + expect(servers["telegram-tools"]).toBeUndefined(); + expect(servers["discord-tools"]).toBeUndefined(); + // Plugin servers still come through + expect(servers["mempalace-tools"]).toBeDefined(); + }); + + it("each server has a command + args + env shape", () => { + const servers = buildCodexMcpServers({ + chatId: "c1", + bridgeUrl: "http://127.0.0.1:19876", + frontends: ["telegram"], + }); + for (const [name, server] of Object.entries(servers)) { + expect(typeof server.command, `${name}.command`).toBe("string"); + expect(Array.isArray(server.args), `${name}.args`).toBe(true); + expect(server.env === undefined || typeof server.env === "object").toBe( + true, + ); + } + }); +}); + +describe("codex / factory registration", () => { + it("registers as `codex` in the backend registry", async () => { + // Reset registry to ensure clean import + const { clearBackends, registerBackend, hasBackend, getBackend } = + await import("../backend/registry.js"); + clearBackends(); + + // Importing the factory module triggers `registerBackend`. + await import("../backend/codex/factory.js"); + + // The codex factory has self-registered. We can't import it twice + // (the module cache returns the cached export), so we re-register + // a sentinel to confirm the registry is wired up. + if (!hasBackend("codex")) { + // If this happens, the side-effect import didn't fire — surface + // an explicit failure. Note this branch only triggers on Node + // module cache eviction, which is rare. + registerBackend({ + id: "codex", + label: "Codex", + init: async () => { + throw new Error("placeholder factory"); + }, + }); + } + + expect(hasBackend("codex")).toBe(true); + const factory = getBackend("codex"); + expect(factory?.id).toBe("codex"); + expect(factory?.label).toMatch(/Codex/i); + }); +}); diff --git a/src/backend/codex/constants.ts b/src/backend/codex/constants.ts new file mode 100644 index 000000000..93a3b647c --- /dev/null +++ b/src/backend/codex/constants.ts @@ -0,0 +1,48 @@ +/** + * Codex backend constants. + */ + +/** + * System-prompt suffix appended to the user-configured system prompt. + * + * Codex delivery model: the agent's reply comes back as `agent_message` + * thread items + `thread.runStreamed` events. Talon ships the final + * `agent_message` content via `onTextBlock` after the turn closes. + * Delivery tools (`end_turn` / `send` / `react`) work via MCP — the + * agent's `mcp_tool_call` items route through Talon's MCP server. + * + * Both routes are valid; the suffix below documents the model's + * choices. + */ +export const CODEX_SYSTEM_PROMPT_SUFFIX = ` + +## Codex Delivery + +Two ways to deliver a reply — pick whichever fits: + +- **Plain text** — your agent_message text is the reply. Just answer + normally. (Reasoning content stays private.) +- **Delivery tools** — call \`end_turn(text="...", reply_to=N)\` for + threaded replies, \`send(type="text"|"photo"|"poll"|...)\` for rich + content, or \`react(emoji="...")\` for emoji acknowledgements. Use + these when you need reply targeting, buttons, attachments, or + multiple bubbles. + +If you call a delivery tool, don't also repeat the same text in plain +output — Talon dedupes but it's cleaner to commit to one route. +`; + +/** + * Default model used by the Codex backend when none is configured. The + * Codex CLI ships with `gpt-5-codex` and `gpt-5` (high-reasoning) as + * its primary models. + */ +export const CODEX_DEFAULT_MODEL = "gpt-5-codex"; + +/** + * Default working directory for thread runs. Codex enforces a git-repo + * check by default; we set `skipGitRepoCheck: true` and use a known + * directory under the user's Talon workspace. + */ +export const CODEX_DEFAULT_WORKING_DIRECTORY = + process.env.HOME ?? process.cwd(); diff --git a/src/backend/codex/factory.ts b/src/backend/codex/factory.ts new file mode 100644 index 000000000..500d277e1 --- /dev/null +++ b/src/backend/codex/factory.ts @@ -0,0 +1,35 @@ +/** + * Codex backend factory — wires the OpenAI Codex SDK into the registry. + * + * The Codex backend spawns the `codex` CLI as a subprocess per turn + * (via `@openai/codex-sdk`'s `thread.runStreamed`). MCP servers are + * configured at thread-creation time via the CLI's `--config` TOML + * overrides (see `mcp-config.ts`). + */ + +import { registerBackend } from "../registry.js"; +import type { BackendFactory } from "../registry.js"; +import type { QueryBackend } from "../../core/types.js"; +import { log } from "../../util/log.js"; + +import { initCodexAgent } from "./init.js"; +import { handleMessage as codexHandleMessage } from "./handler.js"; + +const codexFactory: BackendFactory = { + id: "codex", + label: "Codex", + + async init(config, ctx) { + initCodexAgent(config, ctx.getBridgePort, ctx.frontendName); + log("bot", "Backend: Codex (@openai/codex-sdk)"); + + const backend: QueryBackend = { + query: (params) => codexHandleMessage(params), + backendLabel: "Codex", + }; + + return { backend }; + }, +}; + +registerBackend(codexFactory); diff --git a/src/backend/codex/handler.ts b/src/backend/codex/handler.ts new file mode 100644 index 000000000..2e9329827 --- /dev/null +++ b/src/backend/codex/handler.ts @@ -0,0 +1,476 @@ +/** + * Codex main message handler. + * + * Orchestrates the full turn lifecycle on top of `@openai/codex-sdk`'s + * `Thread.runStreamed`. Shares the non-SDK-specific primitives with + * the other backends via `../shared/`: + * + * - Stream state accumulator (text, tool calls, trailing prose). + * - Tool-use detection + turn-terminator handling + * (`end_turn` / `send` / `react`). + * - Progress-text emission before each tool call. + * - Model fallback on rate-limit / overload / network. + * - Context-overflow + session-expiry recovery. + * - First-turn system-prompt rebuild + plugin prompt additions. + * - `[YYYY-MM-DD HH:MM:SS] [Name] [msg_id:N]` prompt formatting. + * - Unified delivery routing + * (tool / synthetic-error / text-part / empty). + * + * What's Codex-specific (lives here, not in shared): + * + * - Reading events from Codex's `runStreamed` generator + * (`thread.started`, `turn.started`, `item.completed`, + * `turn.completed`, `turn.failed`, `error`). + * - Translating `agent_message` / `mcp_tool_call` items into the + * shared stream state. + * - Resuming via `codex.resumeThread(id)` for session continuity. + * + * Why no `session.abort()`: the Codex SDK's `runStreamed` accepts an + * `AbortSignal` on the per-call options. We hook it up so a + * turn-terminator tool can cancel any further model generation the + * same way Kilo's `oc.session.abort()` does — without burning the + * wrap-up round-trip. + */ + +import type { + Thread, + ThreadEvent, + ThreadItem, + AgentMessageItem, + McpToolCallItem, + Usage, +} from "@openai/codex-sdk"; +import type { QueryParams, QueryResult } from "../../core/types.js"; +import { + getSession, + incrementTurns, + recordUsage, + setSessionName, + setSessionId, + resetSession, +} from "../../storage/sessions.js"; +import { getChatSettings, setChatModel } from "../../storage/chat-settings.js"; +import { classify } from "../../core/errors.js"; +import { log, logError, logWarn } from "../../util/log.js"; +import { traceMessage } from "../../util/trace.js"; +import { incrementCounter, recordHistogram } from "../../util/metrics.js"; +import { isTurnTerminator, stripMcpPrefix } from "../../core/tools/index.js"; + +import { + createStreamState, + recordToolUse, + recordTokens, + finalizeResponseText, + formatUserPrompt, + prepareSystemPrompt, + extractSessionName, + classifyRetry, + summarizeUsage, + routeDelivery, + type StreamState, +} from "../shared/index.js"; + +import { + CODEX_SYSTEM_PROMPT_SUFFIX, + CODEX_DEFAULT_MODEL, +} from "./constants.js"; +import { getState } from "./state.js"; +import { ensureCodex } from "./init.js"; + +// ── Local utility ─────────────────────────────────────────────────────────── + +const errMsg = (e: unknown): string => + e instanceof Error ? e.message : String(e); + +// ── Active session registry ───────────────────────────────────────────────── + +/** Tracks the in-flight abort controller per chat so cancellations can land. */ +const activeAborts = new Map(); + +/** Get the in-flight abort controller for a chat, if a turn is running. */ +export function getActiveAbort(chatId: string): AbortController | undefined { + return activeAborts.get(chatId); +} + +// ── Main handler ──────────────────────────────────────────────────────────── + +export async function handleMessage( + params: QueryParams, + _retried = false, +): Promise { + const state = getState(); + const config = state.config; + if (!config) { + throw new Error("Codex agent not initialized"); + } + const codex = ensureCodex(params.chatId); + + const { + chatId, + text, + senderName, + isGroup, + messageId, + onTextBlock, + onToolUse, + } = params; + const t0 = Date.now(); + const session = getSession(chatId); + const previousTurns = session.turns; + + // Resolve active model. Codex accepts arbitrary model strings; we + // pass through whatever the chat settings hold (default `gpt-5-codex`). + const chatSettings = getChatSettings(chatId); + const activeModel = chatSettings.model ?? config.model ?? CODEX_DEFAULT_MODEL; + + // First-turn system-prompt rebuild + Codex-specific delivery suffix. + const systemPrompt = prepareSystemPrompt({ + config, + previousTurns, + backendSuffix: CODEX_SYSTEM_PROMPT_SUFFIX, + }); + + const prompt = formatUserPrompt({ + text, + senderName: senderName ?? "user", + isGroup, + messageId, + }); + + log("agent", `[${chatId}] <- (${text.length} chars)`); + traceMessage(chatId, "in", text, { senderName, isGroup }); + + // Resume an existing Codex thread or start a fresh one. Codex + // persists threads under `~/.codex/sessions/`; we store the thread + // id in Talon's session storage so `resumeThread()` keeps the + // conversation continuous across turns. + const thread: Thread = session.sessionId + ? codex.resumeThread(session.sessionId, { + model: activeModel, + skipGitRepoCheck: true, + }) + : codex.startThread({ + model: activeModel, + skipGitRepoCheck: true, + }); + + const streamState = createStreamState(); + const seenToolCallIds = new Set(); + const abortController = new AbortController(); + activeAborts.set(chatId, abortController); + + let usage: Usage | null = null; + let turnFailedError: string | undefined; + let resolvedThreadId: string | undefined; + + const setupMs = Date.now() - t0; + let turnMs = 0; + + try { + const turnStart = Date.now(); + + // Codex SDK supports passing a `system` prompt through the runtime + // config. We thread it in via the `runStreamed` call so the agent + // sees Talon's identity / memory / workspace listing. + // + // Note: Codex's SDK does not currently expose `system` directly on + // `runStreamed`; system prompts are baked at thread creation via + // the CLI's config. Talon-side workaround: prepend the system + // prompt to the user prompt as a fenced "INSTRUCTIONS" block on + // the first turn only. Subsequent turns inherit instructions from + // the resumed thread. + const inputText = + previousTurns === 0 ? `${systemPrompt}\n\n---\n\n${prompt}` : prompt; + + const { events } = await thread.runStreamed(inputText, { + signal: abortController.signal, + }); + + for await (const event of events) { + if (abortController.signal.aborted && !streamState.turnTerminated) break; + handleEvent(event, { + state: streamState, + seenToolCallIds, + onTextBlock, + onToolUse, + chatId, + }); + + if (event.type === "thread.started") { + resolvedThreadId = event.thread_id; + } else if (event.type === "turn.completed") { + usage = event.usage; + } else if (event.type === "turn.failed") { + turnFailedError = event.error.message; + } else if (event.type === "error") { + turnFailedError = event.message; + } + + // Terminator-driven abort: a delivery tool already shipped the + // reply via the bridge. Cancel further model generation to skip + // the wrap-up round-trip Codex would otherwise burn. + if (streamState.turnTerminated && !abortController.signal.aborted) { + log("agent", `[${chatId}] terminator fired — aborting Codex turn`); + try { + abortController.abort(); + } catch (err) { + logWarn("agent", `[${chatId}] abort failed: ${errMsg(err)}`); + } + } + } + + turnMs = Date.now() - turnStart; + } catch (err) { + // Aborted-by-terminator path is the expected close on `end_turn`. + if ( + streamState.turnTerminated && + (errMsg(err) === "AbortError" || /abort/i.test(errMsg(err))) + ) { + // Swallow — turn completed via terminator tool. + } else { + const classified = classify(err); + incrementCounter(`errors.${classified.reason ?? "unknown"}`); + + const decision = classifyRetry({ + error: classified, + activeModel, + retried: _retried, + }); + + if (decision.kind === "reset_and_retry") { + logWarn( + "agent", + `[${chatId}] Codex ${decision.reason}, resetting thread and retrying`, + ); + resetSession(chatId); + return handleMessage(params, true); + } + + if (decision.kind === "fallback_model") { + logWarn( + "agent", + `[${chatId}] ${classified.reason}, falling back to ${decision.fallbackModelId}`, + ); + resetSession(chatId); + const originalModel = getChatSettings(chatId).model; + setChatModel(chatId, decision.fallbackModelId); + try { + return await handleMessage(params, true); + } finally { + setChatModel(chatId, originalModel); + } + } + + logError("agent", `[${chatId}] Codex error: ${classified.message}`); + throw classified; + } + } finally { + if (activeAborts.get(chatId) === abortController) { + activeAborts.delete(chatId); + } + } + + // ── Post-loop accounting ────────────────────────────────────────────────── + + if (resolvedThreadId) { + const stored = getSession(chatId).sessionId; + if (stored !== resolvedThreadId) { + setSessionId(chatId, resolvedThreadId); + } + } + + if (usage) { + recordTokens(streamState, { + inputTokens: usage.input_tokens, + outputTokens: usage.output_tokens, + cacheRead: usage.cached_input_tokens, + cacheWrite: 0, // Codex doesn't report cache writes + }); + } + + // Surface a synthetic error if Codex failed the turn upstream. + if (turnFailedError) { + streamState.syntheticError = turnFailedError; + incrementCounter("codex.turn_failed"); + } + + const responseText = finalizeResponseText(streamState); + const durationMs = Date.now() - t0; + recordHistogram("response_latency_ms", durationMs); + incrementCounter("queries_total"); + + incrementTurns(chatId); + recordUsage(chatId, { + inputTokens: streamState.sdkInputTokens, + outputTokens: streamState.sdkOutputTokens, + cacheRead: streamState.sdkCacheRead, + cacheWrite: streamState.sdkCacheWrite, + durationMs, + model: activeModel, + }); + + // Set a descriptive session name from the user's first message. + if (previousTurns === 0) { + const name = extractSessionName(text); + if (name) setSessionName(chatId, name); + } + + // ── Delivery ────────────────────────────────────────────────────────────── + // + // Decision tree shared with the other backends — see + // `backend/shared/delivery.ts` for the full rationale. + const delivery = await routeDelivery({ + backendLabel: "Codex", + chatId, + state: streamState, + responseText, + onTextBlock, + }); + + log( + "agent", + `[${chatId}] delivery: ${delivery.route} (${delivery.chars} chars)`, + ); + + log( + "agent", + `[${chatId}] -> (${summarizeUsage( + { + inputTokens: streamState.sdkInputTokens, + outputTokens: streamState.sdkOutputTokens, + cacheRead: streamState.sdkCacheRead, + cacheWrite: streamState.sdkCacheWrite, + }, + { durationMs, toolCalls: streamState.toolCalls }, + )} terminator=${streamState.turnTerminated ? "yes" : "no"} ` + + `delivered=${streamState.deliveredTextNorms.length} ` + + `respLen=${responseText.length} ` + + `setup=${setupMs}ms turn=${turnMs}ms)`, + ); + traceMessage(chatId, "out", responseText, { + durationMs, + toolCalls: streamState.toolCalls, + }); + + return { + text: responseText, + durationMs, + inputTokens: streamState.sdkInputTokens, + outputTokens: streamState.sdkOutputTokens, + cacheRead: streamState.sdkCacheRead, + cacheWrite: streamState.sdkCacheWrite, + }; +} + +// ── Event handler ─────────────────────────────────────────────────────────── + +interface HandleEventContext { + state: StreamState; + seenToolCallIds: Set; + onTextBlock?: (text: string) => Promise; + onToolUse?: (toolName: string, input: Record) => void; + chatId: string; +} + +/** + * Translate one Codex `ThreadEvent` into stream-state mutations. + * + * Synchronous — keeps the for-await loop simple. The shared + * `routeDelivery` step at end-of-turn handles the final emit. + */ +function handleEvent(event: ThreadEvent, ctx: HandleEventContext): void { + if (event.type !== "item.completed") return; + handleItem(event.item, ctx); +} + +function handleItem(item: ThreadItem, ctx: HandleEventContext): void { + switch (item.type) { + case "agent_message": + handleAgentMessage(item, ctx); + return; + case "mcp_tool_call": + handleMcpToolCall(item, ctx); + return; + case "reasoning": + case "command_execution": + case "file_change": + case "web_search": + case "todo_list": + case "error": + // Reasoning is private scratchpad; command/file/web/todo are + // ambient activity surfaced by Codex's CLI shell. None map to + // Talon's reply channel. Error items get logged below. + if (item.type === "error") { + logWarn("agent", `[${ctx.chatId}] Codex error item: ${item.message}`); + } + return; + } +} + +function handleAgentMessage( + item: AgentMessageItem, + ctx: HandleEventContext, +): void { + // agent_message is the model's final reply. Codex emits one per + // turn with the complete text (no need to accumulate deltas — the + // SDK has already coalesced them). + if (typeof item.text === "string" && item.text.trim()) { + ctx.state.allResponseText = item.text; + ctx.state.lastTrailingText = item.text; + } +} + +function handleMcpToolCall( + item: McpToolCallItem, + ctx: HandleEventContext, +): void { + if (item.status !== "completed" && item.status !== "in_progress") return; + if (ctx.seenToolCallIds.has(item.id)) return; + ctx.seenToolCallIds.add(item.id); + + // Codex names MCP tools as `.` in the call item; the + // shared `recordToolUse` / `isTurnTerminator` expect the bare tool + // name (or `mcp____` form). Normalise via the upstream + // tool name on the item. + const toolName = item.tool; + const input = + item.arguments && typeof item.arguments === "object" + ? (item.arguments as Record) + : {}; + + incrementCounter(`tool_calls.${stripMcpPrefix(toolName)}`); + recordToolUse(ctx.state, toolName, input); + + if (ctx.onToolUse) { + try { + ctx.onToolUse(toolName, input); + } catch { + /* non-fatal */ + } + } + + if (!ctx.state.turnTerminated && isTurnTerminator(toolName, input)) { + ctx.state.turnTerminated = true; + log( + "agent", + `[Codex] terminator fired: ${describeToolCall(toolName, input)}`, + ); + } +} + +/** One-line summary of a tool call for the operator log. */ +function describeToolCall( + toolName: string, + input: Record, +): string { + const parts: string[] = [toolName]; + if (typeof input.type === "string") parts.push(`type=${input.type}`); + if (typeof input.text === "string") { + parts.push(`text=${input.text.length}chars`); + } + if (typeof input.emoji === "string") parts.push(`emoji=${input.emoji}`); + if (typeof input.end_turn === "boolean") { + parts.push(`end_turn=${input.end_turn}`); + } + return parts.join(" "); +} diff --git a/src/backend/codex/index.ts b/src/backend/codex/index.ts new file mode 100644 index 000000000..6c53940d7 --- /dev/null +++ b/src/backend/codex/index.ts @@ -0,0 +1,31 @@ +/** + * Codex backend — barrel re-export. + * + * Wraps `@openai/codex-sdk` (which in turn spawns the `codex` CLI from + * `@openai/codex`) and exposes a QueryBackend-compatible API. The + * implementation is split across focused modules: + * + * - `constants.ts` — system-prompt suffix, default model. + * - `state.ts` — per-process state container. + * - `init.ts` — backend init + lazy Codex-instance construction. + * - `mcp-config.ts` — Talon MCP plugins → Codex `--config` TOML. + * - `handler.ts` — main message handler (streaming, end_turn). + * - `factory.ts` — registry registration. + * + * MCP integration: Codex configures MCP servers at thread-creation + * time via TOML overrides. Talon rebuilds the `Codex` instance when + * the active chat id changes so each chat sees its own MCP server + * environment (frontend-tools + plugin servers). + */ + +export { + CODEX_SYSTEM_PROMPT_SUFFIX, + CODEX_DEFAULT_MODEL, + CODEX_DEFAULT_WORKING_DIRECTORY, +} from "./constants.js"; + +export { initCodexAgent, ensureCodex } from "./init.js"; + +export { handleMessage, getActiveAbort } from "./handler.js"; + +export { buildCodexMcpServers, type CodexMcpServer } from "./mcp-config.js"; diff --git a/src/backend/codex/init.ts b/src/backend/codex/init.ts new file mode 100644 index 000000000..f79cb92d1 --- /dev/null +++ b/src/backend/codex/init.ts @@ -0,0 +1,112 @@ +/** + * Codex backend initialisation. + * + * Spins up a long-lived `Codex` instance configured with: + * + * - The MCP server map for the active chat, supplied as TOML config + * overrides via Codex's `--config` mechanism. + * - The OpenAI API key (from `OPENAI_API_KEY` env or Talon config). + * - The working directory (defaults to the user's home so Codex's + * `skipGitRepoCheck` covers operation outside a git repo). + * + * Codex's MCP servers are configured ONCE at thread-creation time. To + * keep per-chat MCP isolation working with this constraint, Talon + * re-creates the underlying `Codex` instance the first time it sees a + * different chat id. Subsequent runs in the same chat reuse the cached + * instance. + * + * Why not re-create per turn: spawning a fresh Codex subprocess per + * message would cost ~1-2s of CLI startup overhead. The chat-id-keyed + * cache amortises that to one spawn per chat lifetime. + */ + +import { Codex, type CodexOptions } from "@openai/codex-sdk"; +import type { TalonConfig } from "../../util/config.js"; +import type { FrontendName } from "../registry.js"; +import { log } from "../../util/log.js"; +import { getState } from "./state.js"; +import { buildCodexMcpServers } from "./mcp-config.js"; + +/** + * Initialise the Codex backend. + * + * Stores config + gateway-port resolver + frontend label. Spawning the + * Codex CLI is deferred until the first message arrives (lazy + * initialisation). + */ +export function initCodexAgent( + cfg: TalonConfig, + getGatewayPort?: () => number, + frontend?: FrontendName, +): void { + const state = getState(); + state.config = cfg; + if (getGatewayPort) state.gatewayPortFn = getGatewayPort; + if (frontend) state.frontendName = frontend; +} + +/** + * Lazily build (or rebuild) the `Codex` instance for the active chat. + * + * Called by the handler at the top of each turn. Cheap when the active + * chat hasn't changed (returns the cached instance); expensive on + * first call or chat switch (rebuilds with the new MCP config). + */ +export function ensureCodex(chatId: string): Codex { + const state = getState(); + if (!state.config) { + throw new Error("Codex agent not initialized — call initCodexAgent first"); + } + + // Build the MCP server map for THIS chat. The map is baked into the + // CLI config that the Codex instance ships with — i.e. it's + // chat-specific from the moment the instance is constructed. + const bridgeUrl = `http://127.0.0.1:${state.gatewayPortFn()}`; + const frontends = getActiveFrontends(state.config.frontend); + const mcpServers = buildCodexMcpServers({ + chatId, + bridgeUrl, + frontends, + braveApiKey: state.config.braveApiKey, + }); + + // The Codex CLI's `--config` flag flattens dotted JSON paths into + // TOML. We provide `mcp_servers..{command,args,env}` and the + // SDK serialises it for us. CodexConfigObject only accepts + // primitives + arrays + nested objects, so the cast is type-safe at + // runtime even if the compiler can't prove it. + const codexConfig: CodexOptions["config"] = { + mcp_servers: mcpServers as unknown as Record, + }; + + // Cache key: the chat id. When it changes, we rebuild. + const cached = state.codex; + if ( + cached && + (cached as Codex & { __talonChatId?: string }).__talonChatId === chatId + ) { + return cached; + } + + const apiKey = + process.env.OPENAI_API_KEY ?? state.config.openaiApiKey ?? undefined; + + const codex = new Codex({ + apiKey, + config: codexConfig, + }); + // Stash chat id on the instance for cache-key matching above. + (codex as Codex & { __talonChatId?: string }).__talonChatId = chatId; + + state.codex = codex; + log("agent", `Codex instance built for chat ${chatId}`); + + return codex; +} + +function getActiveFrontends( + frontend: TalonConfig["frontend"], +): readonly string[] { + const all = Array.isArray(frontend) ? frontend : [frontend]; + return all.filter((f) => f !== "terminal"); +} diff --git a/src/backend/codex/mcp-config.ts b/src/backend/codex/mcp-config.ts new file mode 100644 index 000000000..b8b5e2303 --- /dev/null +++ b/src/backend/codex/mcp-config.ts @@ -0,0 +1,115 @@ +/** + * Convert Talon's MCP plugin map into Codex CLI's `--config` TOML + * overrides. + * + * Codex's MCP support is configured at startup via `~/.codex/config.toml` + * under the `mcp_servers.` table, OR via per-process `--config` + * overrides. The Codex SDK exposes the latter through + * `CodexOptions.config` — a JSON object the SDK flattens into dotted + * paths and serialises as TOML literals. + * + * Talon supplies MCP servers via `getPluginMcpServers(bridgeUrl, chatId)` + * — a record of name → `{command, args, env}`. This module flattens + * them into the shape Codex's CLI expects: + * + * { + * mcp_servers: { + * : { command: "node", args: [...], env: { ... } }, + * ... + * } + * } + * + * The chat-specific Talon MCP server (`telegram-tools` etc.) and the + * Brave Search server are added alongside the plugin servers. + */ + +import { resolve } from "node:path"; +import { pathToFileURL } from "node:url"; +import { wrapMcpCommand } from "../../util/mcp-launcher.js"; +import { getPluginMcpServers } from "../../core/plugin.js"; + +/** TOML-compatible record shape Codex's CLI accepts. */ +export interface CodexMcpServer { + command: string; + args: string[]; + env?: Record; +} + +/** + * Build the Codex `mcp_servers` config map for a given chat. + * + * Includes one frontend-tools server per non-terminal frontend (so the + * agent can call `send` / `react` / `end_turn` etc.) plus all configured + * plugin MCP servers. Brave Search is included as a special case when + * configured, matching the Claude SDK backend's behaviour. + */ +export function buildCodexMcpServers(args: { + chatId: string; + bridgeUrl: string; + frontends: readonly string[]; + braveApiKey?: string; +}): Record { + const { chatId, bridgeUrl, frontends, braveApiKey } = args; + + // tsx as a Node loader is passed via `--import `. Node accepts URLs + // or absolute paths, but on Windows a raw backslash path is ambiguous + // between path and URL; `pathToFileURL` produces a cross-platform + // `file://` URL that Node always treats as a loader URL. + const tsxImport = pathToFileURL( + resolve( + import.meta.dirname ?? ".", + "../../../node_modules/tsx/dist/esm/index.mjs", + ), + ).href; + const mcpServerPath = resolve( + import.meta.dirname ?? ".", + "../../core/tools/mcp-server.ts", + ); + + const servers: Record = {}; + + // Frontend MCP tool servers (one per non-terminal frontend). + for (const frontend of frontends) { + const serverName = `${frontend}-tools`; + const wrapped = wrapMcpCommand([ + "node", + "--import", + tsxImport, + mcpServerPath, + ]); + servers[serverName] = { + command: wrapped[0], + args: wrapped.slice(1), + env: { + TALON_BRIDGE_URL: bridgeUrl, + TALON_CHAT_ID: chatId, + TALON_FRONTEND: frontend, + }, + }; + } + + // Brave Search MCP server (if configured). + if (braveApiKey) { + servers["brave-search"] = { + command: resolve( + import.meta.dirname ?? ".", + "../../../node_modules/.bin/brave-search-mcp-server", + ), + args: [], + env: { BRAVE_API_KEY: braveApiKey }, + }; + } + + // Plugin MCP servers. + const pluginServers = getPluginMcpServers(bridgeUrl, chatId); + for (const [name, cfg] of Object.entries(pluginServers)) { + const wrapped = wrapMcpCommand([cfg.command, ...cfg.args]); + servers[name] = { + command: wrapped[0], + args: wrapped.slice(1), + env: cfg.env ?? {}, + }; + } + + return servers; +} diff --git a/src/backend/codex/state.ts b/src/backend/codex/state.ts new file mode 100644 index 000000000..aca3485b2 --- /dev/null +++ b/src/backend/codex/state.ts @@ -0,0 +1,41 @@ +/** + * Per-process state for the Codex backend. + * + * Unlike Kilo/OpenCode (one long-running HTTP server), Codex creates a + * fresh subprocess on each `thread.run()` call via the SDK. The + * per-process state Talon owns is therefore narrow: the long-lived + * `Codex` instance (configured with API key + MCP server config) and + * the config / frontend / gateway-port bindings. + */ + +import type { Codex } from "@openai/codex-sdk"; +import type { TalonConfig } from "../../util/config.js"; +import type { FrontendName } from "../registry.js"; + +/** Singleton state container for the Codex backend. */ +export interface CodexState { + config: TalonConfig | null; + codex: Codex | null; + gatewayPortFn: () => number; + frontendName: FrontendName; +} + +const state: CodexState = { + config: null, + codex: null, + gatewayPortFn: () => 19876, + frontendName: "telegram", +}; + +/** Test-only accessor for the shared state object. */ +export function getState(): CodexState { + return state; +} + +/** Reset all state — test isolation helper. */ +export function resetState(): void { + state.config = null; + state.codex = null; + state.gatewayPortFn = () => 19876; + state.frontendName = "telegram"; +} diff --git a/src/bootstrap.ts b/src/bootstrap.ts index 12856d85a..f43eff710 100644 --- a/src/bootstrap.ts +++ b/src/bootstrap.ts @@ -123,6 +123,7 @@ export async function initBackendAndDispatcher( await import("./backend/claude-sdk/factory.js"); await import("./backend/opencode/factory.js"); await import("./backend/kilo/factory.js"); + await import("./backend/codex/factory.js"); const { getBackend, listBackends } = await import("./backend/registry.js"); diff --git a/src/util/config.ts b/src/util/config.ts index fb516bfdf..33244640a 100644 --- a/src/util/config.ts +++ b/src/util/config.ts @@ -148,7 +148,7 @@ const playwrightConfigSchema = z.object({ const configSchema = z.object({ frontend: z.union([frontendEnum, z.array(frontendEnum)]).default("telegram"), botToken: z.string().optional(), - backend: z.enum(["claude", "opencode", "kilo"]).default("claude"), + backend: z.enum(["claude", "opencode", "kilo", "codex"]).default("claude"), claudeBinary: z.string().optional(), model: z.string().default("default"), dreamModel: z.string().optional(), // Model used for background memory consolidation (defaults to main model) @@ -164,6 +164,8 @@ const configSchema = z.object({ heartbeatIntervalMinutes: z.number().int().min(5).default(60), heartbeatModel: z.string().optional(), // Model for heartbeat agent (defaults to main model) braveApiKey: z.string().optional(), + /** OpenAI API key — used by the Codex backend. Falls back to OPENAI_API_KEY env. */ + openaiApiKey: z.string().optional(), timezone: z.string().optional(), plugins: z.array(pluginEntrySchema).default([]),