diff --git a/src/__tests__/codex-auth.test.ts b/src/__tests__/codex-auth.test.ts new file mode 100644 index 000000000..bbecff840 --- /dev/null +++ b/src/__tests__/codex-auth.test.ts @@ -0,0 +1,223 @@ +/** + * Codex auth-detection unit tests. + * + * Covers `detectCodexAuth` (priority order, file-parse edge cases), + * `isChatGptModelMismatchError` (substring matching tolerance), and + * the catalog helpers `isCodexApiKeyOnlyModel` / `chatGptFallbackFor`. + */ + +import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { describe, it, expect } from "vitest"; +import { + detectCodexAuth, + isChatGptModelMismatchError, +} from "../backend/codex/auth.js"; +import { + isCodexApiKeyOnlyModel, + chatGptFallbackFor, +} from "../backend/codex/models.js"; + +/** + * Build a frozen env map that overrides HOME so the file-check path + * resolves to a test-controlled directory. + */ +function envWith( + overrides: Record, +): NodeJS.ProcessEnv { + // Don't inherit the host's OPENAI_API_KEY etc — start from a clean + // slate so test outcomes don't depend on the caller's shell. + const base: NodeJS.ProcessEnv = {}; + for (const [k, v] of Object.entries(overrides)) { + if (v !== undefined) base[k] = v; + } + return base; +} + +function makeFakeHome(authJson: string | null): string { + const dir = mkdtempSync(join(tmpdir(), "talon-codex-auth-")); + if (authJson !== null) { + mkdirSync(join(dir, ".codex"), { recursive: true }); + writeFileSync(join(dir, ".codex", "auth.json"), authJson); + } + return dir; +} + +describe("codex / detectCodexAuth — priority order", () => { + it("env OPENAI_API_KEY wins over everything else", () => { + const fakeHome = makeFakeHome(`{"auth_mode":"chatgpt"}`); + try { + const info = detectCodexAuth( + "from-config", + envWith({ OPENAI_API_KEY: "env-key", HOME: fakeHome }), + ); + expect(info.mode).toBe("api-key"); + expect(info.source).toBe("env:OPENAI_API_KEY"); + } finally { + rmSync(fakeHome, { recursive: true, force: true }); + } + }); + + it("config openaiApiKey wins over the auth file", () => { + const fakeHome = makeFakeHome(`{"auth_mode":"chatgpt"}`); + try { + const info = detectCodexAuth("from-config", envWith({ HOME: fakeHome })); + expect(info.mode).toBe("api-key"); + expect(info.source).toBe("config:openaiApiKey"); + } finally { + rmSync(fakeHome, { recursive: true, force: true }); + } + }); + + it("falls back to ~/.codex/auth.json chatgpt mode", () => { + const fakeHome = makeFakeHome( + `{"auth_mode":"chatgpt","OPENAI_API_KEY":null}`, + ); + try { + const info = detectCodexAuth(undefined, envWith({ HOME: fakeHome })); + expect(info.mode).toBe("chatgpt"); + expect(info.source).toBe("file:~/.codex/auth.json"); + expect(info.authFilePath).toBe(join(fakeHome, ".codex", "auth.json")); + expect(info.authFileParsed).toBe(true); + } finally { + rmSync(fakeHome, { recursive: true, force: true }); + } + }); + + it("auth.json with non-null OPENAI_API_KEY → api-key mode", () => { + // The codex CLI ships auth.json with `OPENAI_API_KEY` as a + // top-level field; when non-null it takes precedence over + // `auth_mode`. This mirrors what the CLI does internally. + const fakeHome = makeFakeHome( + `{"auth_mode":"chatgpt","OPENAI_API_KEY":"sk-from-file"}`, + ); + try { + const info = detectCodexAuth(undefined, envWith({ HOME: fakeHome })); + expect(info.mode).toBe("api-key"); + expect(info.source).toBe("file:~/.codex/auth.json"); + } finally { + rmSync(fakeHome, { recursive: true, force: true }); + } + }); + + it("returns `none` when no auth source is available", () => { + const fakeHome = makeFakeHome(null); + try { + const info = detectCodexAuth(undefined, envWith({ HOME: fakeHome })); + expect(info.mode).toBe("none"); + expect(info.source).toBe("missing"); + } finally { + rmSync(fakeHome, { recursive: true, force: true }); + } + }); + + it("returns `none` when auth.json exists but doesn't carry a recognised mode", () => { + const fakeHome = makeFakeHome(`{"some_other_field":"value"}`); + try { + const info = detectCodexAuth(undefined, envWith({ HOME: fakeHome })); + expect(info.mode).toBe("none"); + expect(info.authFileParsed).toBe(true); + } finally { + rmSync(fakeHome, { recursive: true, force: true }); + } + }); + + it("surfaces a parse error when auth.json is malformed JSON", () => { + const fakeHome = makeFakeHome(`{not valid json`); + try { + const info = detectCodexAuth(undefined, envWith({ HOME: fakeHome })); + expect(info.mode).toBe("none"); + expect(info.authFileParsed).toBe(false); + expect(info.parseError).toBeTruthy(); + } finally { + rmSync(fakeHome, { recursive: true, force: true }); + } + }); + + it("returns `none` when HOME is unset and no other auth is configured", () => { + const info = detectCodexAuth(undefined, envWith({})); + expect(info.mode).toBe("none"); + expect(info.authFilePath).toBeUndefined(); + }); +}); + +describe("codex / isChatGptModelMismatchError", () => { + it("matches the canonical OpenAI 400 message", () => { + expect( + isChatGptModelMismatchError( + "The 'gpt-5-codex' model is not supported when using Codex with a ChatGPT account.", + ), + ).toBe(true); + }); + + it("matches a JSON-wrapped payload", () => { + const payload = JSON.stringify({ + type: "error", + status: 400, + error: { + type: "invalid_request_error", + message: + "The 'gpt-5-codex' model is not supported when using Codex with a ChatGPT account.", + }, + }); + expect(isChatGptModelMismatchError(payload)).toBe(true); + }); + + it("is case-insensitive", () => { + expect( + isChatGptModelMismatchError( + "NOT SUPPORTED WHEN USING CODEX WITH A CHATGPT ACCOUNT", + ), + ).toBe(true); + }); + + it("tolerates extra whitespace", () => { + expect( + isChatGptModelMismatchError( + "not\tsupported when using\ncodex with a chatgpt account", + ), + ).toBe(true); + }); + + it("does not match an unrelated 400 error", () => { + expect( + isChatGptModelMismatchError( + "context_length_exceeded: too many tokens in the prompt", + ), + ).toBe(false); + }); + + it("does not match a network error", () => { + expect(isChatGptModelMismatchError("fetch failed — ECONNRESET")).toBe( + false, + ); + }); +}); + +describe("codex / api-key-only catalog helpers", () => { + it("isCodexApiKeyOnlyModel returns true for gpt-5-codex", () => { + expect(isCodexApiKeyOnlyModel("gpt-5-codex")).toBe(true); + }); + + it("returns false for chatgpt-compatible models", () => { + expect(isCodexApiKeyOnlyModel("gpt-5.5")).toBe(false); + expect(isCodexApiKeyOnlyModel("gpt-5")).toBe(false); + expect(isCodexApiKeyOnlyModel("gpt-5-mini")).toBe(false); + }); + + it("returns false for unknown models (no over-correction)", () => { + expect(isCodexApiKeyOnlyModel("totally-made-up-model")).toBe(false); + expect(isCodexApiKeyOnlyModel("")).toBe(false); + }); + + it("chatGptFallbackFor returns gpt-5.5 for gpt-5-codex", () => { + expect(chatGptFallbackFor("gpt-5-codex")).toBe("gpt-5.5"); + }); + + it("chatGptFallbackFor returns undefined for non-api-key-only models", () => { + expect(chatGptFallbackFor("gpt-5.5")).toBeUndefined(); + expect(chatGptFallbackFor("gpt-5")).toBeUndefined(); + expect(chatGptFallbackFor("unknown-model")).toBeUndefined(); + }); +}); diff --git a/src/__tests__/codex-handler.test.ts b/src/__tests__/codex-handler.test.ts index 61d37269d..16eb31d50 100644 --- a/src/__tests__/codex-handler.test.ts +++ b/src/__tests__/codex-handler.test.ts @@ -13,6 +13,9 @@ * These tests verify each path with hand-built event sequences. */ +import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; import { describe, it, expect, vi, beforeEach } from "vitest"; // ── Shared mock state ─────────────────────────────────────────────────────── @@ -23,6 +26,11 @@ interface MockEvent { } let MOCK_EVENTS: MockEvent[] = []; +// Per-call event queue. When non-empty, each `runStreamed` call shifts +// one entry off the front and yields those events INSTEAD of +// `MOCK_EVENTS`. Lets retry-path tests configure +// "fail-with-events-on-call-1, succeed-with-different-events-on-call-2". +let MOCK_EVENTS_QUEUE: MockEvent[][] = []; let MOCK_RUN_STREAMED_CALLS: Array<{ input: string; signal?: AbortSignal }> = []; let MOCK_THREAD_OPTIONS_SEEN: Array> = []; @@ -57,8 +65,12 @@ vi.mock("@openai/codex-sdk", () => { if (MOCK_RUN_STREAMED_THROWS) { throw MOCK_RUN_STREAMED_THROWS; } + const eventsForThisCall = + MOCK_EVENTS_QUEUE.length > 0 + ? (MOCK_EVENTS_QUEUE.shift() ?? MOCK_EVENTS) + : MOCK_EVENTS; const events = (async function* (): AsyncGenerator { - for (const event of MOCK_EVENTS) { + for (const event of eventsForThisCall) { yield event; } })(); @@ -101,13 +113,19 @@ const coreModels = await import("../core/models.js"); // ── Helpers ───────────────────────────────────────────────────────────────── -function setupHandler(): void { +function setupHandler(overrides: Record = {}): void { + // Pass an explicit `openaiApiKey` so `detectCodexAuth` lands in + // api-key mode (and doesn't read the host's `~/.codex/auth.json`, + // which on developer machines may be in chatgpt OAuth mode and + // would otherwise change the auth-aware default model under us). initCodexAgent( { model: "gpt-5-codex", + openaiApiKey: "test-api-key", workspace: "/tmp", systemPrompt: "Test system prompt.", frontend: "telegram", + ...overrides, } as never, () => 19876, "telegram", @@ -116,6 +134,7 @@ function setupHandler(): void { function resetMocks(): void { MOCK_EVENTS = []; + MOCK_EVENTS_QUEUE = []; MOCK_RUN_STREAMED_CALLS = []; MOCK_THREAD_OPTIONS_SEEN = []; MOCK_RESUME_CALLS = []; @@ -811,9 +830,12 @@ describe("codex / handleMessage — model resolution", () => { it("falls back to CODEX_DEFAULT_MODEL when config.model is absent", async () => { // Init without an explicit model — handler should fall back to - // CODEX_DEFAULT_MODEL ("gpt-5-codex"). + // CODEX_DEFAULT_MODEL ("gpt-5-codex") on the api-key auth path. + // `openaiApiKey` forces api-key mode so the auth-aware default + // is the api-key default and not the chatgpt fallback. initCodexAgent( { + openaiApiKey: "test-api-key", workspace: "/tmp", systemPrompt: "Test system prompt.", frontend: "telegram", @@ -1344,3 +1366,223 @@ describe("codex / handleMessage — usage propagation", () => { expect(result.cacheWrite).toBe(0); }); }); + +describe("codex / handleMessage — ChatGPT-auth model fallback", () => { + // These tests assert the two recovery paths around ChatGPT-OAuth + // accounts: (1) pre-emptive swap at request build time when the + // configured model is known-incompatible, (2) post-hoc retry when + // the model passed validation but Codex returned a 400 anyway. + + it("pre-emptively swaps gpt-5-codex → gpt-5.5 under ChatGPT auth", async () => { + // Set up a fake HOME with a chatgpt auth.json so this test is + // self-contained and works on CI (not just on Dylan's machine where + // the real ~/.codex/auth.json happens to be in chatgpt mode). + const fakeHome = mkdtempSync(join(tmpdir(), "talon-codex-handler-")); + mkdirSync(join(fakeHome, ".codex"), { recursive: true }); + writeFileSync( + join(fakeHome, ".codex", "auth.json"), + '{"auth_mode":"chatgpt"}', + ); + const origHome = process.env.HOME; + const origUserProfile = process.env.USERPROFILE; + const origApiKey = process.env.OPENAI_API_KEY; + process.env.HOME = fakeHome; + delete process.env.USERPROFILE; + // Suppress OPENAI_API_KEY so detectCodexAuth doesn't short-circuit + // to api-key mode before reaching the auth-file check. + delete process.env.OPENAI_API_KEY; + try { + // No openaiApiKey + no OPENAI_API_KEY env → init reads the fake + // ~/.codex/auth.json we just created (chatgpt mode). With + // `gpt-5-codex` set in config, the handler should detect the + // mismatch before calling `runStreamed` and pass `gpt-5.5` to + // ThreadOptions. + initCodexAgent( + { + model: "gpt-5-codex", + workspace: "/tmp", + systemPrompt: "Test system prompt.", + frontend: "telegram", + } as never, + () => 19876, + "telegram", + ); + } finally { + // Restore environment regardless of init outcome. + if (origHome === undefined) { + delete process.env.HOME; + } else { + process.env.HOME = origHome; + } + if (origUserProfile !== undefined) { + process.env.USERPROFILE = origUserProfile; + } + if (origApiKey !== undefined) { + process.env.OPENAI_API_KEY = origApiKey; + } + rmSync(fakeHome, { recursive: true, force: true }); + } + + MOCK_EVENTS = [ + { type: "thread.started", thread_id: "thr_preempt" }, + { type: "turn.started" }, + { + type: "item.completed", + item: { id: "i1", type: "agent_message", text: "ok" }, + }, + { + type: "turn.completed", + usage: { + input_tokens: 1, + output_tokens: 1, + cached_input_tokens: 0, + reasoning_output_tokens: 0, + }, + }, + ]; + + await handleMessage({ + chatId: "test-chat", + text: "hi", + senderName: "Dylan", + isGroup: false, + }); + + // ThreadOptions saw `gpt-5.5` even though config said `gpt-5-codex`. + expect(MOCK_THREAD_OPTIONS_SEEN[0].model).toBe("gpt-5.5"); + // No retry needed — the pre-emptive swap means runStreamed is + // called exactly once. + expect(MOCK_RUN_STREAMED_CALLS).toHaveLength(1); + }); + + it("post-hoc retries with gpt-5.5 when Codex returns the ChatGPT 400", async () => { + // Api-key auth so the pre-emptive swap doesn't fire — the handler + // believes gpt-5-codex is valid, hands it to Codex, and Codex + // returns the chatgpt-mismatch 400. The handler should classify + // the error, reset the session, and retry on gpt-5.5. + setupHandler(); + + const chatgptMismatch = new Error( + `Codex Exec exited with code 1: Reading prompt from stdin...\n` + + `{"type":"error","status":400,"error":{"type":"invalid_request_error",` + + `"message":"The 'gpt-5-codex' model is not supported when using Codex ` + + `with a ChatGPT account."}}`, + ); + MOCK_RUN_STREAMED_THROW_QUEUE = [chatgptMismatch, null]; + MOCK_EVENTS = [ + { type: "thread.started", thread_id: "thr_after_fallback" }, + { type: "turn.started" }, + { + type: "item.completed", + item: { id: "i1", type: "agent_message", text: "on fallback" }, + }, + { + type: "turn.completed", + usage: { + input_tokens: 1, + output_tokens: 1, + cached_input_tokens: 0, + reasoning_output_tokens: 0, + }, + }, + ]; + + const result = await handleMessage({ + chatId: "test-chat", + text: "force the 400", + senderName: "Dylan", + isGroup: false, + }); + + // Two runStreamed calls — the original + the retry on the + // chatgpt-compatible model. + expect(MOCK_RUN_STREAMED_CALLS).toHaveLength(2); + expect(MOCK_THREAD_OPTIONS_SEEN[0].model).toBe("gpt-5-codex"); + expect(MOCK_THREAD_OPTIONS_SEEN[1].model).toBe("gpt-5.5"); + expect(result.text).toBe("on fallback"); + + // Chat-settings restored to the original after the retry — the + // fallback is one-shot, not a permanent switch. + expect(chatSettings.getChatSettings("test-chat").model).toBeUndefined(); + }); + + it("retries via the turn.failed event channel (no thrown error)", async () => { + // Some failure modes surface only as `turn.failed` events without + // the SDK rethrowing — the handler still has to detect the + // mismatch text and retry. (The current SDK does both; this test + // covers the event-only path defensively via the post-loop check.) + setupHandler(); + + // First call: events include the mismatch turn.failed, then the + // generator ends cleanly (no throw). The handler sets + // `turnFailedError` during iteration, the loop exits normally, + // the post-loop ChatGPT-mismatch check fires the retry. + // Second call: clean recovery on gpt-5.5. + MOCK_EVENTS_QUEUE = [ + // Call 1 — events only, no throw + [ + { type: "thread.started", thread_id: "thr_event_path_1" }, + { type: "turn.started" }, + { + type: "turn.failed", + error: { + message: + 'The "gpt-5-codex" model is not supported when using Codex with a ChatGPT account.', + }, + }, + ], + // Call 2 — clean recovery + [ + { type: "thread.started", thread_id: "thr_event_path_2" }, + { type: "turn.started" }, + { + type: "item.completed", + item: { id: "i1", type: "agent_message", text: "recovered" }, + }, + { + type: "turn.completed", + usage: { + input_tokens: 1, + output_tokens: 1, + cached_input_tokens: 0, + reasoning_output_tokens: 0, + }, + }, + ], + ]; + + const result = await handleMessage({ + chatId: "test-chat", + text: "force the event-path failure", + senderName: "Dylan", + isGroup: false, + }); + + expect(MOCK_RUN_STREAMED_CALLS).toHaveLength(2); + expect(MOCK_THREAD_OPTIONS_SEEN[1].model).toBe("gpt-5.5"); + expect(result.text).toBe("recovered"); + }); + + it("does not retry when the chatgpt-mismatch fires a second time", async () => { + // Defensive: if the post-hoc retry ALSO returns the 400 (shouldn't + // happen since gpt-5.5 is supported, but guards against a future + // breakage), don't loop. `_retried` is true on the recursive call + // so the chatgpt-mismatch branch short-circuits. + setupHandler(); + + const mismatch = new Error( + "not supported when using Codex with a ChatGPT account", + ); + MOCK_RUN_STREAMED_THROW_QUEUE = [mismatch, mismatch]; + + await expect( + handleMessage({ + chatId: "test-chat", + text: "double-failure", + senderName: "Dylan", + isGroup: false, + }), + ).rejects.toThrow(); + expect(MOCK_RUN_STREAMED_CALLS).toHaveLength(2); + }); +}); diff --git a/src/backend/claude-sdk/handler.ts b/src/backend/claude-sdk/handler.ts index 7e6343bd2..0f7190bee 100644 --- a/src/backend/claude-sdk/handler.ts +++ b/src/backend/claude-sdk/handler.ts @@ -171,24 +171,22 @@ export async function handleMessage( } } - // Note: we previously called `qi.interrupt()` here when a turn- - // terminator tool fired, intending to short-circuit the SDK's - // wasted "wrap up after end_turn tool_result" follow-up API call - // (~3s of phantom typing while the model says nothing useful). - // That interrupt races with in-flight MCP tool dispatches in the - // same assistant message — `end_turn` itself is an MCP tool, and - // the model frequently emits sibling tool_use blocks in the same - // message. interrupt cancels their AbortController mid-flight, - // which surfaces as `MCP error -32001: AbortError` in the SDK - // result and bubbles up to the user as "Something went wrong". + // Turn-terminator detection happens here (sets `state.turnTerminated` + // for the flow-violation check below) but the actual SDK loop exit + // is owned by the `PostToolBatch` hook in `options.ts`. The hook + // fires after every tool in the batch has resolved, returns + // `{ continue: false }`, and the SDK exits with TerminalReason + // `'hook_stopped'` — no extra "wrap up after end_turn" round-trip, + // no phantom typing, no token spend on a stop_turn. // - // The natural-close path is fine: the SDK does one more API call - // after end_turn returns (the model has nothing to say so it - // returns a stop turn quickly, ~2-3s typing lag), then yields a - // result message and exits the iterator cleanly. We accept the - // typing lag in exchange for not breaking turns. `state.turnTerminated` - // is still tracked so the flow-violation re-prompt path below can - // skip its retry when the model explicitly ended its turn. + // Historical note: an earlier implementation called `qi.interrupt()` + // here directly. That raced with in-flight MCP tool dispatches in + // the same assistant message — `end_turn` itself is an MCP tool, + // and the model frequently emits sibling tool_use blocks alongside + // it. `interrupt()` cancelled their AbortController mid-flight, + // surfacing as `MCP error -32001: AbortError` and bubbling up as + // "Something went wrong". The `PostToolBatch` hook avoids the race + // by definition (it fires once the entire batch has resolved). continue; } @@ -245,7 +243,11 @@ export async function handleMessage( recordHistogram("response_latency_ms", durationMs); incrementCounter("queries_total"); if (state.newSessionId) setSessionId(chatId, state.newSessionId); - incrementTurns(chatId); + // Token usage is recorded for THIS attempt unconditionally — the running + // session totals are additive, so a flow-violation retry that recurses + // through this same path will record its own tokens on top. The turn + // counter, in contrast, must only increment ONCE per user message (see + // the post-violation block below). recordUsage(chatId, { inputTokens: state.sdkInputTokens, outputTokens: state.sdkOutputTokens, @@ -270,6 +272,11 @@ export async function handleMessage( // shared `detectFlowViolation` decides whether trailing prose constitutes // a missed delivery (and whether to re-prompt the model once with the // synthetic reminder). + // + // `incrementTurns` is deferred until AFTER this check so the retry path + // (which recurses through `handleMessage` and hits its own + // `incrementTurns` at the end of that call) doesn't double-count a + // single user message as two turns. const violation = detectFlowViolation({ trailingText: state.lastTrailingText, turnTerminated: state.turnTerminated, @@ -290,10 +297,15 @@ export async function handleMessage( if (violation.shouldRetry) { incrementCounter("scratchpad.flow_violation_retried"); + // The recursive call owns the `incrementTurns` for this user message. + // We deliberately don't increment here. return handleMessage({ ...params, text: violation.reminder }, true); } } + // Reached the non-retry path — this turn counts as one user-visible turn. + incrementTurns(chatId); + // ── Build result ────────────────────────────────────────────────────────── state.allResponseText += state.currentBlockText; diff --git a/src/backend/claude-sdk/options.ts b/src/backend/claude-sdk/options.ts index 4f98321f1..0300eba7a 100644 --- a/src/backend/claude-sdk/options.ts +++ b/src/backend/claude-sdk/options.ts @@ -251,6 +251,15 @@ export function buildSdkOptions(chatId: string): BuildSdkOptionsResult { model: resolvedActiveModel, systemPrompt: config.systemPrompt, cwd: config.workspace, + // The SDK's permission system is designed for an interactive Claude + // Code IDE session where a human approves each tool call. Talon runs + // the SDK as a server-side bot — there's no human at the keyboard + // to confirm Bash/Edit/etc., and our security boundary is the bot + // account itself (its OS user, its workspace dir, its mempalace), + // not the SDK's per-tool prompts. `bypassPermissions` skips the + // interactive prompts; `allowDangerouslySkipPermissions` is the + // explicit acknowledgement the SDK requires alongside it. Treat + // these as a unit — flipping either alone is a configuration bug. permissionMode: "bypassPermissions", allowDangerouslySkipPermissions: true, ...(config.claudeBinary diff --git a/src/backend/codex/auth.ts b/src/backend/codex/auth.ts new file mode 100644 index 000000000..1efcf7300 --- /dev/null +++ b/src/backend/codex/auth.ts @@ -0,0 +1,165 @@ +/** + * Codex authentication detection. + * + * The Codex CLI supports two distinct authentication modes, with + * materially different capabilities: + * + * - **API key** (`OPENAI_API_KEY` env var, or `apiKey` passed to the + * SDK's `Codex` constructor, or `OPENAI_API_KEY` stored in + * `~/.codex/auth.json`). Full model catalog access, billed via the + * standard OpenAI API. `gpt-5-codex` is the flagship. + * + * - **ChatGPT OAuth** (`~/.codex/auth.json` `auth_mode: "chatgpt"` + * after running `codex login`). Restricted to a subset of models — + * `gpt-5-codex` is explicitly rejected with a 400 + * "invalid_request_error" telling the user the model isn't + * supported on a ChatGPT account. `gpt-5.5` is the flagship under + * this auth mode. + * + * Talon needs to know which mode is active at startup so it can pick a + * sensible default model and surface useful errors. This module owns + * the detection logic. + * + * Detection order (each step short-circuits if it succeeds): + * + * 1. `OPENAI_API_KEY` env var → `"api-key"`. + * 2. `openaiApiKey` in Talon config → `"api-key"`. + * 3. `~/.codex/auth.json` exists + `auth_mode` field is read → returns + * that value (`"chatgpt"` for OAuth, `"api-key"` if the JSON also + * has a non-null `OPENAI_API_KEY`). + * 4. Nothing → `"none"`. First turn will fail with an auth error; + * caller emits a startup warning pointing to `codex login`. + */ + +import { existsSync, readFileSync } from "node:fs"; +import { join } from "node:path"; + +/** Detected auth mode. */ +export type CodexAuthMode = "api-key" | "chatgpt" | "none"; + +/** Resolved auth state plus diagnostics for the startup log. */ +export interface CodexAuthInfo { + mode: CodexAuthMode; + /** Where the credential was found, for logging. */ + source: + | "env:OPENAI_API_KEY" + | "config:openaiApiKey" + | "file:~/.codex/auth.json" + | "missing"; + /** Path to the auth file when present (for diagnostics). */ + authFilePath?: string; + /** Whether the auth file (if present) parsed correctly. */ + authFileParsed: boolean; + /** Raw parse error when the file existed but couldn't be parsed. */ + parseError?: string; +} + +/** + * Detect the active Codex auth mode. + * + * `configKey` is the value of `config.openaiApiKey` from Talon's main + * config — passed in by the caller so this module doesn't depend on + * the config module. `envOverride` defaults to `process.env` so tests + * can substitute a frozen env map. + */ +export function detectCodexAuth( + configKey?: string, + envOverride: NodeJS.ProcessEnv = process.env, +): CodexAuthInfo { + // 1. Explicit API key in the environment. + if (envOverride.OPENAI_API_KEY) { + return { + mode: "api-key", + source: "env:OPENAI_API_KEY", + authFileParsed: false, + }; + } + + // 2. Explicit API key in Talon's config. + if (configKey) { + return { + mode: "api-key", + source: "config:openaiApiKey", + authFileParsed: false, + }; + } + + // 3. Codex CLI auth file (created by `codex login`). + const home = envOverride.HOME ?? envOverride.USERPROFILE; + if (home) { + const authFilePath = join(home, ".codex", "auth.json"); + if (existsSync(authFilePath)) { + try { + const raw = readFileSync(authFilePath, "utf8"); + const parsed = JSON.parse(raw) as { + auth_mode?: string; + OPENAI_API_KEY?: string | null; + }; + // The auth.json schema includes `OPENAI_API_KEY` as a top-level + // field even on ChatGPT-mode installs; when it's non-null it + // takes precedence (mirrors what the CLI itself does). + if ( + parsed.OPENAI_API_KEY && + typeof parsed.OPENAI_API_KEY === "string" + ) { + return { + mode: "api-key", + source: "file:~/.codex/auth.json", + authFilePath, + authFileParsed: true, + }; + } + if (parsed.auth_mode === "chatgpt") { + return { + mode: "chatgpt", + source: "file:~/.codex/auth.json", + authFilePath, + authFileParsed: true, + }; + } + // File parsed but doesn't carry a recognised mode — treat as + // missing so the caller surfaces a `codex login` reminder. + return { + mode: "none", + source: "missing", + authFilePath, + authFileParsed: true, + }; + } catch (err) { + return { + mode: "none", + source: "missing", + authFilePath, + authFileParsed: false, + parseError: err instanceof Error ? err.message : String(err), + }; + } + } + } + + // 4. Nothing. + return { + mode: "none", + source: "missing", + authFileParsed: false, + }; +} + +/** + * Detect whether an error from `runStreamed` is the + * "model-not-supported-on-ChatGPT-account" 400. Used by the handler's + * recovery ladder to auto-fall-back to a ChatGPT-compatible model when + * the configured one is API-key-only. + * + * Codex surfaces this in two places — the JSON error payload nested + * inside an `error` event, AND the textual `turn.failed.error.message` + * the SDK wraps around it. Both contain the substring + * `"not supported when using Codex with a ChatGPT account"`. Match on + * that substring (case-insensitive, generous on whitespace) so a + * future wording shift still trips a soft-match. + */ +export function isChatGptModelMismatchError(message: string): boolean { + return /not\s+supported\s+when\s+using\s+codex\s+with\s+a\s+chatgpt\s+account/i.test( + message, + ); +} diff --git a/src/backend/codex/constants.ts b/src/backend/codex/constants.ts index 93a3b647c..a96850c61 100644 --- a/src/backend/codex/constants.ts +++ b/src/backend/codex/constants.ts @@ -33,12 +33,37 @@ 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. + * Default model used by the Codex backend when none is configured AND + * an API key is present (`OPENAI_API_KEY` env or `openaiApiKey` in + * config). The `gpt-5-codex` model is the highest-quality coding model + * available through the Codex CLI but requires API-key billing — it is + * not granted to ChatGPT subscription accounts. */ export const CODEX_DEFAULT_MODEL = "gpt-5-codex"; +/** + * Default model used by the Codex backend when the user is signed in + * via ChatGPT OAuth (`~/.codex/auth.json` `auth_mode: "chatgpt"`). + * The `gpt-5-codex` model is rejected with a 400 + * `invalid_request_error` ("not supported when using Codex with a + * ChatGPT account") on this auth path; `gpt-5.5` is the supported + * flagship for ChatGPT users. + */ +export const CODEX_CHATGPT_DEFAULT_MODEL = "gpt-5.5"; + +/** + * Set of Codex models that require an API key (i.e. won't work under + * ChatGPT OAuth). Used by the handler's recovery ladder to detect a + * model-not-supported error and automatically fall back to a + * chatgpt-compatible model on retry. + * + * Keep this list in sync with `CODEX_MODELS` in `models.ts` — every + * entry there with `apiKeyOnly: true` should appear here. + */ +export const CODEX_API_KEY_ONLY_MODELS: ReadonlySet = new Set([ + "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 diff --git a/src/backend/codex/handler.ts b/src/backend/codex/handler.ts index 123bae3b4..f6e8af27d 100644 --- a/src/backend/codex/handler.ts +++ b/src/backend/codex/handler.ts @@ -73,15 +73,62 @@ import { import { CODEX_SYSTEM_PROMPT_SUFFIX, CODEX_DEFAULT_MODEL, + CODEX_CHATGPT_DEFAULT_MODEL, } from "./constants.js"; import { getState } from "./state.js"; -import { ensureCodex } from "./init.js"; +import { ensureCodex, getCodexAuthInfo } from "./init.js"; +import { isChatGptModelMismatchError } from "./auth.js"; +import { chatGptFallbackFor, isCodexApiKeyOnlyModel } from "./models.js"; // ── Local utility ─────────────────────────────────────────────────────────── const errMsg = (e: unknown): string => e instanceof Error ? e.message : String(e); +/** + * One-shot ChatGPT-OAuth model-mismatch recovery. + * + * Returns a retry promise when the error text indicates Codex rejected + * an api-key-only model on a ChatGPT account AND we haven't already + * retried. Returns `undefined` otherwise (caller falls through to its + * normal error handling). + * + * Shared between the catch block (when the SDK rethrows the 400) and + * the post-event-loop check (when the SDK only emits `turn.failed` + * without rethrowing). The retry side-effects are confined here: + * session reset, transient `setChatModel` flip, `_retried = true` + * sentinel on the recursive call. + */ +async function maybeFallbackForChatGptMismatch( + probeText: string, + activeModel: string, + params: QueryParams, + retried: boolean, + chatId: string, +): Promise { + if (retried) return undefined; + if (!isChatGptModelMismatchError(probeText)) return undefined; + + const fallbackModel = + chatGptFallbackFor(activeModel) ?? CODEX_CHATGPT_DEFAULT_MODEL; + if (fallbackModel === activeModel) return undefined; + + logWarn( + "agent", + `[${chatId}] Codex returned ChatGPT model-mismatch for ${activeModel}; ` + + `resetting thread and retrying on ${fallbackModel}. ` + + `Set OPENAI_API_KEY for billing-based access to api-key-only models.`, + ); + resetSession(chatId); + const originalModel = getChatSettings(chatId).model; + setChatModel(chatId, fallbackModel); + try { + return await handleMessage(params, true); + } finally { + setChatModel(chatId, originalModel); + } +} + // ── Active session registry ───────────────────────────────────────────────── /** Tracks the in-flight abort controller per chat so cancellations can land. */ @@ -119,9 +166,37 @@ export async function handleMessage( const previousTurns = session.turns; // Resolve active model. Codex accepts arbitrary model strings; we - // pass through whatever the chat settings hold (default `gpt-5-codex`). + // pass through whatever the chat settings hold. The fallback chain + // is: chat-settings → config → auth-aware default. The auth-aware + // default is `gpt-5-codex` when an API key is present, `gpt-5.5` + // when only ChatGPT OAuth is configured (because `gpt-5-codex` is + // rejected with a 400 on ChatGPT-mode accounts). const chatSettings = getChatSettings(chatId); - const activeModel = chatSettings.model ?? config.model ?? CODEX_DEFAULT_MODEL; + const authInfo = getCodexAuthInfo(); + const authAwareDefault = + authInfo?.mode === "chatgpt" + ? CODEX_CHATGPT_DEFAULT_MODEL + : CODEX_DEFAULT_MODEL; + const requestedModel = chatSettings.model ?? config.model ?? authAwareDefault; + // If the resolved model is api-key-only AND we're on ChatGPT OAuth, + // pre-emptively swap to the chatgpt-compatible fallback rather than + // letting the first turn fail. This is the "user configured + // gpt-5-codex in talon.json and then ran codex login" case — the + // handler's retry ladder will catch it post-hoc but doing it here + // saves the 400 round-trip + makes the resolved-model log line + // match what Codex actually sees. + let activeModel = requestedModel; + if (authInfo?.mode === "chatgpt" && isCodexApiKeyOnlyModel(requestedModel)) { + const fallback = + chatGptFallbackFor(requestedModel) ?? CODEX_CHATGPT_DEFAULT_MODEL; + logWarn( + "agent", + `[${chatId}] Codex model ${requestedModel} is api-key-only and ` + + `current auth is ChatGPT OAuth — pre-emptively falling back to ${fallback}. ` + + `Set OPENAI_API_KEY or change the configured model to silence this.`, + ); + activeModel = fallback; + } log("agent", `[${chatId}] Codex model resolved: ${activeModel}`); // First-turn system-prompt rebuild + Codex-specific delivery suffix. @@ -243,6 +318,19 @@ export async function handleMessage( ) { // Swallow — turn completed via terminator tool. } else { + // ChatGPT-OAuth model-mismatch path: a 400 invalid_request_error + // saying "not supported when using Codex with a ChatGPT account". + // Check both the captured event-stream message and the thrown + // error — Codex SDK surfaces it via both channels. + const fallback = await maybeFallbackForChatGptMismatch( + `${turnFailedError ?? ""} ${errMsg(err)}`, + activeModel, + params, + _retried, + chatId, + ); + if (fallback) return fallback; + const classified = classify(err); incrementCounter(`errors.${classified.reason ?? "unknown"}`); @@ -287,6 +375,22 @@ export async function handleMessage( // ── Post-loop accounting ────────────────────────────────────────────────── + // Event-only ChatGPT-mismatch recovery: if the SDK emitted a + // `turn.failed` carrying the mismatch text but DIDN'T rethrow, the + // catch block above never fired. Catch it here too. (Current SDK + // rethrows in this case, but this path defends against a future + // change where it only surfaces via the event stream.) + if (turnFailedError && !_retried) { + const fallback = await maybeFallbackForChatGptMismatch( + turnFailedError, + activeModel, + params, + _retried, + chatId, + ); + if (fallback) return fallback; + } + if (resolvedThreadId) { const stored = getSession(chatId).sessionId; if (stored !== resolvedThreadId) { diff --git a/src/backend/codex/init.ts b/src/backend/codex/init.ts index 6daba967c..b68fca1e6 100644 --- a/src/backend/codex/init.ts +++ b/src/backend/codex/init.ts @@ -20,19 +20,33 @@ * cache amortises that to one spawn per chat lifetime. */ -import { Codex, type CodexOptions } from "@openai/codex-sdk"; +import { Codex } from "@openai/codex-sdk"; import type { TalonConfig } from "../../util/config.js"; import type { FrontendName } from "../registry.js"; import { log, logWarn } from "../../util/log.js"; import { getState } from "./state.js"; -import { buildCodexMcpServers } from "./mcp-config.js"; +import { asCodexConfig, buildCodexMcpServers } from "./mcp-config.js"; +import { detectCodexAuth, type CodexAuthInfo } from "./auth.js"; + +/** Cached auth-mode detection result — updated on every `initCodexAgent`. */ +let cachedAuthInfo: CodexAuthInfo | null = null; + +/** + * Return the auth-mode detection result captured at the last + * `initCodexAgent` call. Used by the handler to pick the right default + * model and by tests to assert init-time behaviour. + */ +export function getCodexAuthInfo(): CodexAuthInfo | null { + return cachedAuthInfo; +} /** * Initialise the Codex backend. * * Stores config + gateway-port resolver + frontend label. Spawning the * Codex CLI is deferred until the first message arrives (lazy - * initialisation). + * initialisation). Auth-mode detection runs synchronously here so the + * startup log surfaces the result before any turn fires. */ export function initCodexAgent( cfg: TalonConfig, @@ -48,32 +62,42 @@ export function initCodexAgent( // `ensureCodex(chatId)` call will rebuild from scratch. state.codex = null; - // Friendly startup check: warn early if neither the env var nor a - // configured key nor the host's `~/.codex/auth.json` is available. - // Codex will fail per-turn with a less helpful error otherwise. - const hasEnvKey = Boolean(process.env.OPENAI_API_KEY); - const hasCfgKey = Boolean(cfg.openaiApiKey); - // Lazy: only check the auth.json file if neither key is present - // (avoids spurious filesystem reads on the happy path). - if (!hasEnvKey && !hasCfgKey) { - void checkCodexAuthFile(); - } + // Detect auth mode synchronously so the startup line carries it. + // Knowing whether we're on api-key vs chatgpt vs nothing drives both + // the default-model pick and the handler's recovery decisions. + const authInfo = detectCodexAuth(cfg.openaiApiKey); + cachedAuthInfo = authInfo; + logAuthInfo(authInfo); } -async function checkCodexAuthFile(): Promise { - if (!process.env.HOME) return; - try { - const { existsSync } = await import("node:fs"); - if (!existsSync(`${process.env.HOME}/.codex/auth.json`)) { +function logAuthInfo(info: CodexAuthInfo): void { + switch (info.mode) { + case "api-key": + log("agent", `Codex auth: api-key (source: ${info.source})`); + return; + case "chatgpt": + log( + "agent", + `Codex auth: chatgpt OAuth (source: ${info.source}). ` + + `Note: gpt-5-codex is unavailable on this auth mode — ` + + `gpt-5.5 will be used as the default.`, + ); + return; + case "none": logWarn( "agent", "Codex: no OPENAI_API_KEY env, no openaiApiKey in talon.json, " + - "and no ~/.codex/auth.json — first turn will fail. " + - "Run `codex login` or set OPENAI_API_KEY.", + "and no usable ~/.codex/auth.json — first turn will fail. " + + "Run `codex login` (for ChatGPT OAuth) or set OPENAI_API_KEY " + + "(for API-key billing).", ); - } - } catch { - /* non-fatal — auth check is best-effort */ + if (info.authFilePath && !info.authFileParsed && info.parseError) { + logWarn( + "agent", + `Codex: ~/.codex/auth.json exists but failed to parse: ${info.parseError}`, + ); + } + return; } } @@ -104,12 +128,10 @@ export function ensureCodex(chatId: string): Codex { // 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, - }; + // SDK serialises it for us. The type narrowing (SDK's + // `CodexConfigObject` vs our `CodexMcpServer` shape) is centralised + // in `asCodexConfig`. + const codexConfig = asCodexConfig(mcpServers); // Cache key: the chat id. When it changes, we rebuild. const cached = state.codex; diff --git a/src/backend/codex/mcp-config.ts b/src/backend/codex/mcp-config.ts index b8b5e2303..a0e94d370 100644 --- a/src/backend/codex/mcp-config.ts +++ b/src/backend/codex/mcp-config.ts @@ -25,6 +25,7 @@ import { resolve } from "node:path"; import { pathToFileURL } from "node:url"; +import type { CodexOptions } from "@openai/codex-sdk"; import { wrapMcpCommand } from "../../util/mcp-launcher.js"; import { getPluginMcpServers } from "../../core/plugin.js"; @@ -35,6 +36,27 @@ export interface CodexMcpServer { env?: Record; } +/** + * Wrap a `Record` in the `{ mcp_servers: ... }` + * envelope the Codex SDK expects on `CodexOptions.config`. + * + * The SDK's `CodexConfigObject` type recursively narrows values to + * `string | number | boolean | array | object`. Our `CodexMcpServer` + * shape satisfies that at runtime (every leaf is a string or string + * array), but TypeScript can't prove the structural compatibility + * because `Record` isn't a direct subtype of + * the recursive `CodexConfigObject` type. The cast through `unknown` + * is centralised here so the lie lives in exactly one documented spot + * — both `init.ts` and any future caller use this helper. + */ +export function asCodexConfig( + mcpServers: Record, +): CodexOptions["config"] { + return { + mcp_servers: mcpServers, + } as unknown as CodexOptions["config"]; +} + /** * Build the Codex `mcp_servers` config map for a given chat. * diff --git a/src/backend/codex/models.ts b/src/backend/codex/models.ts index fe80b6619..55f550f72 100644 --- a/src/backend/codex/models.ts +++ b/src/backend/codex/models.ts @@ -19,16 +19,38 @@ import type { ModelButton, } from "../../core/types.js"; -/** Models available through the Codex CLI. */ -export const CODEX_MODELS: UnifiedModelInfo[] = [ +/** + * Codex-specific model metadata extension. + * + * The Codex CLI accepts a fixed set of model strings, but which ones + * actually resolve depends on the auth mode in use. `apiKeyOnly: true` + * marks models that the OpenAI API rejects with a 400 when called from + * a ChatGPT-OAuth account (`auth_mode: "chatgpt"` in `~/.codex/auth.json`). + * The handler's recovery ladder reads this flag. + */ +export interface CodexModelInfo extends UnifiedModelInfo { + /** True if this model requires API-key billing (not available on ChatGPT OAuth). */ + apiKeyOnly?: boolean; +} + +/** + * Models available through the Codex CLI. + * + * Order matters: `getSettingsPresentation` lists models in this order, + * and `gpt-5.5` is intentionally first because it's the broadest-access + * model (works on both auth modes) — it's the safe default for + * Talon-on-Codex deployments where the operator hasn't explicitly + * picked a model. + */ +export const CODEX_MODELS: CodexModelInfo[] = [ { - id: "gpt-5-codex", - displayName: "GPT-5 Codex", + id: "gpt-5.5", + displayName: "GPT-5.5", provider: "openai", providerName: "OpenAI", selectable: true, reasoning: true, - contextWindow: 200_000, + contextWindow: 400_000, }, { id: "gpt-5", @@ -37,7 +59,7 @@ export const CODEX_MODELS: UnifiedModelInfo[] = [ providerName: "OpenAI", selectable: true, reasoning: true, - contextWindow: 200_000, + contextWindow: 400_000, }, { id: "gpt-5-mini", @@ -46,7 +68,17 @@ export const CODEX_MODELS: UnifiedModelInfo[] = [ providerName: "OpenAI", selectable: true, reasoning: true, - contextWindow: 200_000, + contextWindow: 400_000, + }, + { + id: "gpt-5-codex", + displayName: "GPT-5 Codex", + provider: "openai", + providerName: "OpenAI", + selectable: true, + reasoning: true, + contextWindow: 400_000, + apiKeyOnly: true, }, { id: "o4-mini", @@ -59,6 +91,27 @@ export const CODEX_MODELS: UnifiedModelInfo[] = [ }, ]; +/** + * True when the given model id is in the catalog AND flagged as + * api-key-only. Returns `false` for unknown models — the caller should + * not over-correct on unrecognised inputs. + */ +export function isCodexApiKeyOnlyModel(id: string): boolean { + return CODEX_MODELS.some((m) => m.id === id && m.apiKeyOnly === true); +} + +/** + * Return a chatgpt-OAuth-compatible fallback for an api-key-only model. + * Returns `undefined` when the model isn't recognised as api-key-only + * (caller can skip the fallback path) or when no compatible fallback + * exists. Currently the only api-key-only entry is `gpt-5-codex`; this + * function points it at `gpt-5.5` as the broadest-access flagship. + */ +export function chatGptFallbackFor(id: string): string | undefined { + if (!isCodexApiKeyOnlyModel(id)) return undefined; + return "gpt-5.5"; +} + /** * Resolve a user query string against the Codex model catalog. * diff --git a/src/backend/kilo/handler.ts b/src/backend/kilo/handler.ts index b9f87e9c1..74fd455f6 100644 --- a/src/backend/kilo/handler.ts +++ b/src/backend/kilo/handler.ts @@ -64,8 +64,11 @@ import { classifyRetry, summarizeUsage, routeDelivery, + sleep, } from "../shared/index.js"; import { processStreamEvent, finalizePartsIntoState } from "./events.js"; +import { subscribeSseStream } from "../remote-server/sse-stream.js"; +import { findLastAssistantMessage as findLastAssistantMessageShared } from "../remote-server/messages.js"; // ── Active session registry ───────────────────────────────────────────────── // @@ -537,20 +540,15 @@ async function runKiloTurn(inputs: RunKiloTurnInputs): Promise { * last assistant message (the one that just landed via the prompt * we issued). */ -function findLastAssistantMessage( +// Shared walker — see `remote-server/messages.ts`. The runtime guard + +// `info` typing live there once; the generic `Info` parameter labels +// what we consume. +const findLastAssistantMessage = ( messages: Array>, -): { parts: Array>; info?: KiloAssistantInfo } | null { - for (let i = messages.length - 1; i >= 0; i--) { - const m = messages[i]; - const info = m?.info as { role?: string } | undefined; - if (info?.role !== "assistant") continue; - const parts = Array.isArray(m.parts) - ? (m.parts as Array>) - : []; - return { parts, info: info as unknown as KiloAssistantInfo }; - } - return null; -} +): { + parts: Array>; + info?: KiloAssistantInfo; +} | null => findLastAssistantMessageShared(messages); // ── SSE subscription ─────────────────────────────────────────────────────── @@ -590,19 +588,11 @@ async function subscribeToTurnEvents(inputs: SubscribeInputs): Promise { abortSignal, } = inputs; - let stream: AsyncIterable | undefined; - try { - // The SDK's `global.event()` returns a `ServerSentEventsResult` whose - // `stream` field is an async iterable of typed events. - const sse = (await oc.global.event()) as unknown as { - stream?: AsyncIterable; - }; - stream = sse?.stream; - } catch (err) { - logWarn("agent", `[${chatId}] SSE subscribe failed: ${errMsg(err)}`); - return; - } - + // `subscribeSseStream` handles the `as unknown as { stream }` narrowing + // with a runtime guard + the subscribe-failed warning. Returns + // `undefined` for both "subscribe rejected" and "no iterable in the + // response shape" — either way we exit without iterating. + const stream = await subscribeSseStream(oc, chatId); if (!stream) return; try { @@ -714,22 +704,4 @@ async function subscribeToTurnEvents(inputs: SubscribeInputs): Promise { } } -// ── Sleep with abort ──────────────────────────────────────────────────────── - -function sleep(ms: number, signal?: AbortSignal): Promise { - return new Promise((resolve) => { - if (signal?.aborted) { - resolve(); - return; - } - const timer = setTimeout(() => resolve(), ms); - if (signal) { - const onAbort = (): void => { - clearTimeout(timer); - signal.removeEventListener("abort", onAbort); - resolve(); - }; - signal.addEventListener("abort", onAbort, { once: true }); - } - }); -} +// (`sleep` lives in `../shared/sleep.ts` — see import at the top.) diff --git a/src/backend/opencode/handler.ts b/src/backend/opencode/handler.ts index b14436bfe..9f6ed403f 100644 --- a/src/backend/opencode/handler.ts +++ b/src/backend/opencode/handler.ts @@ -66,11 +66,14 @@ import { classifyRetry, summarizeUsage, routeDelivery, + sleep, } from "../shared/index.js"; import { processStreamEvent, finalizePartsIntoState, } from "../remote-server/events.js"; +import { subscribeSseStream } from "../remote-server/sse-stream.js"; +import { findLastAssistantMessage as findLastAssistantMessageShared } from "../remote-server/messages.js"; // ── Local utility ─────────────────────────────────────────────────────────── @@ -521,21 +524,15 @@ async function runOpenCodeTurn(inputs: RunOpenCodeTurnInputs): Promise { * Find the most recent assistant message in a session-messages list * and surface its parts + assistant info in a uniform shape. */ -function findLastAssistantMessage(messages: Array>): { +// Shared walker — see `remote-server/messages.ts`. The runtime guard + +// `info` typing live there once; the generic `Info` parameter labels +// what we consume. +const findLastAssistantMessage = ( + messages: Array>, +): { parts: Array>; info?: OpenCodeAssistantInfo; -} | null { - for (let i = messages.length - 1; i >= 0; i--) { - const m = messages[i]; - const info = m?.info as { role?: string } | undefined; - if (info?.role !== "assistant") continue; - const parts = Array.isArray(m.parts) - ? (m.parts as Array>) - : []; - return { parts, info: info as unknown as OpenCodeAssistantInfo }; - } - return null; -} +} | null => findLastAssistantMessageShared(messages); // ── SSE subscription ─────────────────────────────────────────────────────── @@ -574,17 +571,9 @@ async function subscribeToTurnEvents(inputs: SubscribeInputs): Promise { abortSignal, } = inputs; - let stream: AsyncIterable | undefined; - try { - const sse = (await oc.global.event()) as unknown as { - stream?: AsyncIterable; - }; - stream = sse?.stream; - } catch (err) { - logWarn("agent", `[${chatId}] SSE subscribe failed: ${errMsg(err)}`); - return; - } - + // `subscribeSseStream` handles the `as unknown as { stream }` narrowing + // with a runtime guard + the subscribe-failed warning. + const stream = await subscribeSseStream(oc, chatId); if (!stream) return; try { @@ -671,22 +660,4 @@ async function subscribeToTurnEvents(inputs: SubscribeInputs): Promise { } } -// ── Sleep with abort ──────────────────────────────────────────────────────── - -function sleep(ms: number, signal?: AbortSignal): Promise { - return new Promise((resolve) => { - if (signal?.aborted) { - resolve(); - return; - } - const timer = setTimeout(() => resolve(), ms); - if (signal) { - const onAbort = (): void => { - clearTimeout(timer); - signal.removeEventListener("abort", onAbort); - resolve(); - }; - signal.addEventListener("abort", onAbort, { once: true }); - } - }); -} +// (`sleep` lives in `../shared/sleep.ts` — see import at the top.) diff --git a/src/backend/opencode/one-shot.ts b/src/backend/opencode/one-shot.ts index b34e53dec..07cda2593 100644 --- a/src/backend/opencode/one-shot.ts +++ b/src/backend/opencode/one-shot.ts @@ -3,6 +3,12 @@ * * Mirrors src/backend/kilo/one-shot.ts (Kilo is a fork of OpenCode and the * SDK shape is the same). See that file's header for the design notes. + * + * Note on abort semantics: the OpenCode SDK exposes `session.abort` but the + * REST `prompt` endpoint blocks until the underlying provider returns. The + * heartbeat module's outer abort grace is what actually unblocks the lock + * — we call `session.abort` as a best-effort cleanup so the model stops + * spending tokens once the timeout fires. */ import type { OneShotAgentParams } from "../../core/types.js"; @@ -16,7 +22,9 @@ import { resolveProviderID, parseStoredOpenCodeModelSelection, OPENCODE_SYSTEM_PROMPT_SUFFIX, + errMsg, } from "./server.js"; +import { appendBackendSuffix } from "../shared/index.js"; export async function runOneShotAgent( params: OneShotAgentParams, @@ -47,16 +55,38 @@ export async function runOneShotAgent( const sessionID = typeof sessionData?.id === "string" ? sessionData.id : String(Date.now()); + // Abort handler: when the heartbeat timeout fires, ask OpenCode to stop + // generating. The await on `session.prompt` will reject; we propagate. + // Without this, runaway generation on a hung heartbeat keeps spending + // tokens until the provider returns naturally — kilo/one-shot.ts has the + // same handler and this file got out of sync with it. + const onAbort = (): void => { + oc.session + .abort({ sessionID }) + .catch((err) => + logWarn( + "agent", + `One-shot session.abort failed for ${sessionID}: ${errMsg(err)}`, + ), + ); + }; + abortController.signal.addEventListener("abort", onAbort, { once: true }); + try { if (abortController.signal.aborted) { throw new Error("Aborted before prompt was sent"); } + const finalSystemPrompt = appendBackendSuffix( + systemPrompt, + OPENCODE_SYSTEM_PROMPT_SUFFIX, + ); + const resp = await oc.session.prompt({ sessionID, parts: [{ type: "text", text: prompt }], model: { providerID, modelID }, - system: systemPrompt + OPENCODE_SYSTEM_PROMPT_SUFFIX, + system: finalSystemPrompt, ...(toolOverrides ? { tools: toolOverrides } : {}), }); @@ -69,15 +99,14 @@ export async function runOneShotAgent( await appendOpenCodePart(appendLog, part); } } finally { + abortController.signal.removeEventListener("abort", onAbort); await disconnectChatMcpServer(oc, chatMcpServerName); try { await oc.session.delete({ sessionID }); } catch (err) { logWarn( "agent", - `Failed to delete one-shot OpenCode session ${sessionID}: ${ - err instanceof Error ? err.message : String(err) - }`, + `Failed to delete one-shot OpenCode session ${sessionID}: ${errMsg(err)}`, ); } } diff --git a/src/backend/remote-server/events.ts b/src/backend/remote-server/events.ts index 772a57b51..dc6e5e4d4 100644 --- a/src/backend/remote-server/events.ts +++ b/src/backend/remote-server/events.ts @@ -31,7 +31,12 @@ import { recordToolUse, type StreamState, } from "../shared/index.js"; -import { log } from "../../util/log.js"; +import { log, logDebug } from "../../util/log.js"; + +/** Format an error for a debug log line. */ +function errMsg(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} import { incrementCounter } from "../../util/metrics.js"; import { stripMcpPrefix } from "../../core/tools/index.js"; @@ -208,8 +213,15 @@ async function processPartUpdate( if (progress && ctx.onTextBlock) { try { await ctx.onTextBlock(progress); - } catch { - /* non-fatal — never break the stream loop on a UI callback */ + } catch (err) { + // Non-fatal — never break the stream loop on a UI callback. Logged + // at debug level so a repeatedly-failing frontend is visible to + // anyone tailing with LOG_LEVEL=debug, without polluting the + // default operator log. + logDebug( + "agent", + `[${ctx.backendLabel}] onTextBlock (progress) threw: ${errMsg(err)}`, + ); } } // Count every tool the model calls — parity with claude-sdk which @@ -219,8 +231,11 @@ async function processPartUpdate( if (ctx.onToolUse) { try { ctx.onToolUse(toolName, input); - } catch { - /* non-fatal */ + } catch (err) { + logDebug( + "agent", + `[${ctx.backendLabel}] onToolUse(${toolName}) threw: ${errMsg(err)}`, + ); } } if (ctx.state.turnTerminated) { @@ -271,8 +286,8 @@ export function maybeFireStreamDelta( state.lastStreamUpdate = now; try { onStreamDelta(state.currentBlockText, phase); - } catch { - /* non-fatal */ + } catch (err) { + logDebug("agent", `onStreamDelta threw: ${errMsg(err)}`); } } @@ -376,8 +391,11 @@ export function finalizePartsIntoState(inputs: FinalizePartsInputs): { if (onToolUse) { try { onToolUse(toolName, stateObj.input); - } catch { - /* non-fatal */ + } catch (err) { + logDebug( + "agent", + `onToolUse(${toolName}) (parts backfill) threw: ${errMsg(err)}`, + ); } } } diff --git a/src/backend/remote-server/messages.ts b/src/backend/remote-server/messages.ts new file mode 100644 index 000000000..89814b403 --- /dev/null +++ b/src/backend/remote-server/messages.ts @@ -0,0 +1,52 @@ +/** + * Shared message-list walkers for the remote-server backends. + * + * Both `KiloClient.session.messages` and `OpencodeClient.session.messages` + * return the same wire format: an array of `{ info: { role, ... }, parts: [...] }` + * objects. Kilo and OpenCode previously duplicated the same `findLastAssistantMessage` + * helper — same body, drifted formatting, identical `info as unknown as + * KiloAssistantInfo` / `OpenCodeAssistantInfo` casts. This module centralises + * the walker and pushes the `info` typing out as a generic the caller supplies. + * + * The runtime guard (`info.role === "assistant"`) lives here and runs once; + * the generic just labels what the caller intends to consume. No new + * unsafe-cast surface is introduced — callers no longer cast through + * `unknown` themselves. + */ + +/** + * Shape every message in `session.messages` carries — confirmed by curling + * both servers directly. `info.role` is the only field consulted at runtime; + * everything else is forwarded to the caller via the `Info` generic. + */ +export interface BackendAssistantMatch { + parts: Array>; + info?: Info; +} + +/** + * Walk a message list backwards and return the most recent assistant + * message + its parts list, or `null` when no assistant message exists. + * + * `Info` is the backend's own "assistant info" type — both Kilo and + * OpenCode have one, named `KiloAssistantInfo` / `OpenCodeAssistantInfo`, + * with the same shape. The cast happens once here behind the runtime + * `role === "assistant"` check, so callers consume a properly-typed + * `Info` without further narrowing. + */ +export function findLastAssistantMessage( + messages: Array>, +): BackendAssistantMatch | null { + for (let i = messages.length - 1; i >= 0; i--) { + const m = messages[i]; + const info = m?.info as { role?: string } | undefined; + if (info?.role !== "assistant") continue; + const parts = Array.isArray(m.parts) + ? (m.parts as Array>) + : []; + // Runtime-confirmed assistant info — the generic just labels what the + // backend consumes. Single documented cast, not duplicated per backend. + return { parts, info: info as unknown as Info }; + } + return null; +} diff --git a/src/backend/remote-server/sse-stream.ts b/src/backend/remote-server/sse-stream.ts new file mode 100644 index 000000000..59543ba76 --- /dev/null +++ b/src/backend/remote-server/sse-stream.ts @@ -0,0 +1,80 @@ +/** + * Typed access to the remote-server SSE event iterator. + * + * Both `@kilocode/sdk/v2` and `@opencode-ai/sdk/v2` expose + * `client.global.event()` which (per the upstream wire format) returns + * a `ServerSentEventsResult` whose `stream` field is an async iterable + * of typed events. The SDK's published types under-promise this shape — + * the call signature returns a wider type than the value it produces. + * + * Previously, every backend that consumed this called + * `await oc.global.event() as unknown as { stream?: AsyncIterable }` + * inline. Three copies of the same lie. This helper centralises it + * behind one narrowing function with a runtime guard, so subsequent + * backends (or the next SDK revision) can update the typing in one + * place. + */ + +import { logWarn } from "../../util/log.js"; + +/** + * Minimal client surface — what both `KiloClient` and `OpencodeClient` + * expose for SSE subscription. The structural shape avoids depending on + * either SDK directly from this module. + */ +export interface SseSubscribableClient { + global: { + event(): Promise; + }; +} + +/** + * Subscribe to the remote agent's SSE event stream. + * + * Returns the event iterator on success, or `undefined` when the + * subscription call rejected (network blip, server not ready, etc.) or + * when the response shape doesn't expose an iterable stream. Callers + * should treat `undefined` as "no events to read; carry on with the + * sync prompt response as source of truth." + * + * The warning is logged here so callers don't all need to duplicate the + * `errMsg(err)` formatting. + */ +export async function subscribeSseStream( + client: SseSubscribableClient, + chatId: string, +): Promise | undefined> { + let result: unknown; + try { + result = await client.global.event(); + } catch (err) { + logWarn( + "agent", + `[${chatId}] SSE subscribe failed: ${err instanceof Error ? err.message : String(err)}`, + ); + return undefined; + } + return narrowSseResult(result); +} + +/** + * Pull the iterable `stream` field out of a `global.event()` response. + * Exported for tests; production code should call `subscribeSseStream`. + */ +export function narrowSseResult( + result: unknown, +): AsyncIterable | undefined { + if (!result || typeof result !== "object") return undefined; + const stream = (result as { stream?: unknown }).stream; + if (!stream || typeof stream !== "object") return undefined; + // `Symbol.asyncIterator` is the only reliable runtime predicate for + // an async iterable. Some SDK shapes also expose `[Symbol.iterator]` + // but the for-await-of loop in callers requires the async variant. + if ( + typeof (stream as AsyncIterable)[Symbol.asyncIterator] !== + "function" + ) { + return undefined; + } + return stream as AsyncIterable; +} diff --git a/src/backend/shared/index.ts b/src/backend/shared/index.ts index 3866bf4f3..f84ba8fda 100644 --- a/src/backend/shared/index.ts +++ b/src/backend/shared/index.ts @@ -71,3 +71,5 @@ export { type DeliveryDecision, type RouteDeliveryInputs, } from "./delivery.js"; + +export { sleep } from "./sleep.js"; diff --git a/src/backend/shared/sleep.ts b/src/backend/shared/sleep.ts new file mode 100644 index 000000000..55e1a67bf --- /dev/null +++ b/src/backend/shared/sleep.ts @@ -0,0 +1,26 @@ +/** + * Abort-aware sleep helper. + * + * Resolves after `ms` milliseconds, or immediately when the optional + * `AbortSignal` fires. Used by the remote-server backends (kilo, + * opencode) for throttled retries and stream pacing. Previously + * duplicated inline in `kilo/handler.ts` and `opencode/handler.ts`. + */ + +export function sleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal?.aborted) { + resolve(); + return; + } + const timer = setTimeout(() => resolve(), ms); + if (signal) { + const onAbort = (): void => { + clearTimeout(timer); + signal.removeEventListener("abort", onAbort); + resolve(); + }; + signal.addEventListener("abort", onAbort, { once: true }); + } + }); +}