diff --git a/src/builtins/providers.zig b/src/builtins/providers.zig index 1eac08876..5d7b456f1 100644 --- a/src/builtins/providers.zig +++ b/src/builtins/providers.zig @@ -18,6 +18,7 @@ pub const native = provider_set.Set{ .configured_fn = configuredBundle, .gateway = gateway.provider_bundle, .codex = .{ + .capabilities = .{ .sends_output_limit = false }, .presentation = provider_catalog.find(.codex), .auth_strategy = .chatgpt, .title_model = openai_codex_models.title_model, diff --git a/src/core/agent/runtime/orchestrator.zig b/src/core/agent/runtime/orchestrator.zig index bc8a13779..e327d1a20 100644 --- a/src/core/agent/runtime/orchestrator.zig +++ b/src/core/agent/runtime/orchestrator.zig @@ -6862,8 +6862,8 @@ fn processQueuedPromptLoop( defer interrupted_persisted_ptr.* = interrupted_persisted; var silent_tool_steps: usize = 0; var continuation_injected = false; - // Set once, for the turn's single retry of an empty reply the provider - // cut short while the request named no output limit. + // The limit the turn's one retry of an empty, cut-short reply set. Later + // steps keep it until the catalog names a limit of its own. var length_retry_output_tokens: ?u32 = null; var last_step_ctx = finish_trace.ctx; var current_step_index: usize = 0; @@ -7431,7 +7431,7 @@ fn processQueuedPromptLoop( .tool_choice = tool_choice, .vision_mode = vision_mode, .provider_options = provider_opts, - .max_output_tokens = length_retry_output_tokens orelse model_capabilities.requestOutputTokens(request_capabilities), + .max_output_tokens = model_capabilities.requestOutputTokens(request_capabilities) orelse length_retry_output_tokens, .budget = .{ .cancel_flag = config.cancel_flag }, }; // The request re-sends every image the session keeps, and @@ -9410,16 +9410,21 @@ fn processQueuedPromptLoop( try runtime_telemetry.traceReturnedToolCalls(arena, step_ctx, completion.tool_calls); // A reply the provider cut short before any text or tool call, from a // request that named no output limit, ran out of the provider's own - // default. It is sent again once with an explicit limit. - if (disposition == .length_limited and + // default. A provider that takes fx's limit gets the request once more + // with one. + const empty_length_retry: ?u32 = if (disposition == .length_limited and completion.tool_calls.len == 0 and std.mem.trim(u8, partial_assistant, " \t\r\n").len == 0 and length_retry_output_tokens == null and + config.provider_capabilities.sends_output_limit and model_capabilities.requestOutputTokens(request_capabilities) == null and agent_steps.allowsStep(config.agent_step_limit, step + 1)) - { - length_retry_output_tokens = model_capabilities.empty_length_retry_output_tokens; - debug_trace.eventf("agent", "empty_length_limited_retry", step_ctx, "max_output_tokens={d}", .{model_capabilities.empty_length_retry_output_tokens}); + model_capabilities.emptyLengthRetryOutputTokens(request_capabilities) + else + null; + if (empty_length_retry) |limit| { + length_retry_output_tokens = limit; + debug_trace.eventf("agent", "empty_length_limited_retry", step_ctx, "max_output_tokens={d}", .{limit}); try deps.push_text(deps.ctx, .{ .operational = empty_length_retry_notice }); try deps.push_text(deps.ctx, .{ .operational = "\n" }); continue; diff --git a/src/core/app/model_cache_runtime.zig b/src/core/app/model_cache_runtime.zig index e0b1805c4..a07fde2fa 100644 --- a/src/core/app/model_cache_runtime.zig +++ b/src/core/app/model_cache_runtime.zig @@ -359,8 +359,9 @@ pub const Runtime = struct { thread: ?std.Thread = null, state: ModelCacheState = .idle, /// A fetch is running. Only the first load for an access shows as - /// `.loading`; a revalidation or retry leaves the state readers see alone, - /// so no reader ever waits on it. + /// `.loading`; a revalidation or retry leaves the state readers see alone. + /// A request waits up to `failed_catalog_retry_wait_ms` for a failed + /// catalog's retry, and on nothing else but a first load. fetch_in_flight: bool = false, /// Consecutive failed fetches. A success or reset clears it. failure_streak: u32 = 0, @@ -502,7 +503,8 @@ pub const Runtime = struct { self.mutex.unlock(io_mod.getIo()); return false; } - // Only a first load blocks request resolution. A revalidation or retry + // A first load blocks request resolution, and a failed catalog's retry + // holds it for at most `failed_catalog_retry_wait_ms`. A revalidation // runs behind the state readers already see. const first_load = self.state == .idle; if (first_load) self.state = .loading; diff --git a/src/core/config/model_capabilities.zig b/src/core/config/model_capabilities.zig index d2cc58533..8aa047307 100644 --- a/src/core/config/model_capabilities.zig +++ b/src/core/config/model_capabilities.zig @@ -131,8 +131,21 @@ const full_window_output_tokens: u32 = 32_768; const full_window_output_divisor: u32 = 8; /// The output limit for the one retry of a reply the provider cut short before -/// any text, when the first request set no limit and the provider's default ran out. -pub const empty_length_retry_output_tokens: u32 = full_window_output_tokens; +/// any text, when the first request set no limit and the provider's default ran +/// out. A known context window bounds it as it bounds a full-window limit; null +/// when that leaves no room. +pub fn emptyLengthRetryOutputTokens(capabilities: Capabilities) ?u32 { + const window = capabilities.context_window orelse return full_window_output_tokens; + const bounded = @min(full_window_output_tokens, window / full_window_output_divisor); + return if (bounded == 0) null else bounded; +} + +test "empty length retry limit stays inside a known context window" { + try std.testing.expectEqual(@as(?u32, 32_768), emptyLengthRetryOutputTokens(.{})); + try std.testing.expectEqual(@as(?u32, 32_768), emptyLengthRetryOutputTokens(.{ .context_window = 1_000_000 })); + try std.testing.expectEqual(@as(?u32, 8_192), emptyLengthRetryOutputTokens(.{ .context_window = 65_536 })); + try std.testing.expectEqual(@as(?u32, null), emptyLengthRetryOutputTokens(.{ .context_window = 7 })); +} /// Output tokens to request and reserve for one model call, or null to leave the limit unset. pub fn requestOutputTokens(capabilities: Capabilities) ?u32 { diff --git a/src/core/gateway/provider_set.zig b/src/core/gateway/provider_set.zig index e36a7a7ab..cde3fd796 100644 --- a/src/core/gateway/provider_set.zig +++ b/src/core/gateway/provider_set.zig @@ -20,6 +20,9 @@ pub const Bundle = struct { gateway_prompt_caching: bool = false, fx_search: bool = false, vision_fallback: bool = false, + /// Whether a request carries fx's output limit. ChatGPT's Codex + /// endpoint chooses its own and rejects the parameter. + sends_output_limit: bool = true, }; capabilities: Capabilities = .{}, diff --git a/src/core/session/command_replay_store.zig b/src/core/session/command_replay_store.zig index e04a60a23..ef4576d22 100644 --- a/src/core/session/command_replay_store.zig +++ b/src/core/session/command_replay_store.zig @@ -263,12 +263,10 @@ const ReplayBacking = union(enum) { } }; -/// A v2 session's replay handle: `fx-command-replay-`, the name of its -/// blob, and `.bin` (D44). Caller owns. /// One run's handle for the blob that holds its output. The blob's hash comes /// last, where `artifact_digest.blobHash` reads it, after a random id for the /// run, so two commands with the same output still get handles of their own -/// (D91). +/// (D44, D91). Caller owns. fn blobReplayHandle(alloc: Allocator, hash: session_child_store.Blobs.Hash) ![]u8 { var run_id: [6]u8 = undefined; try std.Io.randomSecure(io_mod.getIo(), &run_id); diff --git a/tests/e2e/acp.test.ts b/tests/e2e/acp.test.ts index 816304dd6..95d4e3df8 100644 --- a/tests/e2e/acp.test.ts +++ b/tests/e2e/acp.test.ts @@ -8853,7 +8853,9 @@ describe("acp: model-independent", () => { // limit; neither request asks for a summary. expect(gateway.requests).toHaveLength(4); expect(JSON.parse(gateway.requests[2]!.body).maxOutputTokens).toBeUndefined(); - expect(JSON.parse(gateway.requests[3]!.body).maxOutputTokens).toBe(32_768); + const retryLimit = JSON.parse(gateway.requests[3]!.body).maxOutputTokens; + expect(retryLimit).toBeGreaterThan(0); + expect(retryLimit).toBeLessThanOrEqual(32_768); expect(gateway.requests.some(({ body }) => body.includes("Summarize what you just did.") )).toBe(false); diff --git a/tests/e2e/gateway-stream-lifecycle.test.ts b/tests/e2e/gateway-stream-lifecycle.test.ts index 2ec1e161e..785978fa3 100644 --- a/tests/e2e/gateway-stream-lifecycle.test.ts +++ b/tests/e2e/gateway-stream-lifecycle.test.ts @@ -8965,10 +8965,11 @@ printf '%s' ${JSON.stringify(trailingMarker)} > ${JSON.stringify(effectPath)} } test("an empty reply cut short by the provider's default limit is asked again with a limit", async () => { + // A window this large bounds the retry at fx's full-window limit. const { result, json, bodies, trace } = await askWithReplies("empty-length-retry", [ emptyLengthLimitedReply, () => fakeGatewayFinalText("answered after the limit retry"), - ]); + ], { context_window: 1_000_000 }); expect(result.code).toBe(0); expect(json.output).toContain("answered after the limit retry"); @@ -8983,6 +8984,18 @@ printf '%s' ${JSON.stringify(trailingMarker)} > ${JSON.stringify(effectPath)} expect(trace).toContain("event=empty_length_limited_retry"); }, 40_000); + test("an empty reply's retry limit stays inside the model's known context window", async () => { + const { result, bodies } = await askWithReplies("empty-length-window", [ + emptyLengthLimitedReply, + () => fakeGatewayFinalText("answered inside the window"), + ], { context_window: 65_536 }); + + expect(result.code).toBe(0); + expect(bodies).toHaveLength(2); + expect(bodies[0]!.maxOutputTokens).toBeUndefined(); + expect(bodies[1]!.maxOutputTokens).toBe(8_192); + }, 40_000); + test("an empty reply still cut short after the retry says so instead of Done.", async () => { const { result, json, bodies } = await askWithReplies("empty-length-twice", [ emptyLengthLimitedReply, diff --git a/tests/e2e/host-managed-auth.test.ts b/tests/e2e/host-managed-auth.test.ts index 2d1ca17db..c282cc075 100644 --- a/tests/e2e/host-managed-auth.test.ts +++ b/tests/e2e/host-managed-auth.test.ts @@ -21,6 +21,9 @@ describe("host-managed authentication", () => { let server: ReturnType; let baseUrl = ""; let codexUnauthorizedResponses = 0; + // Replies the Codex and Grok fakes cut short before any text. + let emptyLengthReplies = 0; + const responseBodies: Array<{ path: string; body: string }> = []; beforeAll(() => { root = mkdtempSync(join(tmpdir(), "fx-host-managed-auth-")); @@ -67,6 +70,16 @@ describe("host-managed authentication", () => { context_window: 272000, }] }); } + if (path === "/codex/responses" || path === "/grok/responses") { + responseBodies.push({ path, body: await request.text() }); + if (emptyLengthReplies > 0) { + emptyLengthReplies -= 1; + return new Response( + 'data: {"type":"response.incomplete","response":{"id":"resp_empty_length","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[],"usage":{"input_tokens":4,"output_tokens":4096}}}\n\n', + { headers: { "content-type": "text/event-stream" } }, + ); + } + } if (path === "/codex/responses") { if (codexUnauthorizedResponses > 0) { codexUnauthorizedResponses -= 1; @@ -239,6 +252,44 @@ describe("host-managed authentication", () => { expect(existsSync(join(home, ".fx", "auth.json"))).toBe(false); }, TIMEOUT); + test("an empty reply cut short is asked again only through a provider that takes fx's output limit", async () => { + const childEnv = env(); + const turnBodies = (path: string) => responseBodies + .filter((entry) => entry.path === path && !entry.body.includes("Generate a short title")) + .map((entry) => JSON.parse(entry.body) as { max_output_tokens?: number }); + for (const provider of ["codex", "grok"] as const) { + const selected = await runFx(["provider", provider], { + cwd: workspace, + env: childEnv, + timeoutMs: TIMEOUT, + }); + expect(selected.code).toBe(0); + const path = `/${provider}/responses`; + const before = turnBodies(path).length; + emptyLengthReplies = 1; + const asked = await runFx(["ask", "--json", "--no-save", "Reply once."], { + cwd: workspace, + env: childEnv, + timeoutMs: TIMEOUT, + }); + emptyLengthReplies = 0; + const sent = turnBodies(path).slice(before); + expect(asked.code).toBe(0); + expect(sent[0]!.max_output_tokens).toBeUndefined(); + if (provider === "codex") { + // The Codex endpoint ignores fx's limit, so a retry would repeat the request. + expect(sent).toHaveLength(1); + expect(asked.stderr).not.toContain("Retrying with a higher limit."); + expect(JSON.parse(asked.stdout).final_output).toBe("The reply ran out of output tokens before any text."); + } else { + expect(sent).toHaveLength(2); + expect(sent[1]!.max_output_tokens).toBe(32_768); + expect(asked.stderr).toContain("Retrying with a higher limit."); + expect(asked.stdout).toContain("GROK_HOST_MANAGED_OK"); + } + } + }, TIMEOUT * 2); + test("interactive host-managed session streams through the same authority", async () => { const childEnv = env(); const selected = await runFx(["provider", "gateway"], { diff --git a/tests/e2e/mcp-stdio.test.ts b/tests/e2e/mcp-stdio.test.ts index f98b028dd..49ed93e2b 100644 --- a/tests/e2e/mcp-stdio.test.ts +++ b/tests/e2e/mcp-stdio.test.ts @@ -1159,7 +1159,10 @@ describe("modern MCP stdio compatibility", () => { }); expect(laterGateway.requests).toHaveLength(3); // A name no live server has still fails without reaching any server. - expect(toolResultText(laterGateway.requests[1]!.body, "missing_call")).toContain(missingTool); + const missingResult = toolResultText(laterGateway.requests[1]!.body, "missing_call"); + expect(missingResult).toMatch(/not selected for this model step|Unsupported tool/); + expect(missingResult).toContain(missingTool); + expect(missingResult).not.toContain(MODERN_RESULT); const laterResult = toolResultText(laterGateway.requests[2]!.body, "later_call"); expect(laterResult).toContain(`${MODERN_RESULT}:later`); expect(laterResult).not.toContain("not selected");