diff --git a/src/shared/dynamic-truncator.test.ts b/src/shared/dynamic-truncator.test.ts index d090dd260..aae20aed9 100644 --- a/src/shared/dynamic-truncator.test.ts +++ b/src/shared/dynamic-truncator.test.ts @@ -2,7 +2,11 @@ import { describe, expect, it, afterEach } from "bun:test" -import { getContextWindowUsage, invalidateContextWindowUsageCache } from "./dynamic-truncator" +import { + _setContextWindowUsageFetchTimeoutMsForTesting, + getContextWindowUsage, + invalidateContextWindowUsageCache, +} from "./dynamic-truncator" const ANTHROPIC_CONTEXT_ENV_KEY = "ANTHROPIC_1M_CONTEXT" const VERTEX_CONTEXT_ENV_KEY = "VERTEX_ANTHROPIC_1M_CONTEXT" @@ -89,6 +93,114 @@ function createCountingContextUsageMockContext(inputTokens: number) { describe("getContextWindowUsage", () => { afterEach(() => { resetContextLimitEnv() + _setContextWindowUsageFetchTimeoutMsForTesting(undefined) + }) + + describe("#given client.session.messages never settles", () => { + describe("#when getContextWindowUsage is called with a fast fetch timeout", () => { + it("#then returns null instead of hanging forever", async () => { + // given + _setContextWindowUsageFetchTimeoutMsForTesting(50) + const ctx = { + client: { + session: { + messages: () => new Promise(() => {}), + }, + }, + } + + // when + const start = Date.now() + const usage = await getContextWindowUsage(ctx as never, "ses_hang_messages", { + anthropicContext1MEnabled: false, + }) + const elapsed = Date.now() - start + + // then + expect(usage).toBeNull() + expect(elapsed).toBeLessThan(2000) + }) + + it("#then a parallel concurrent caller also resolves to null instead of hanging on the cached promise", async () => { + // given + _setContextWindowUsageFetchTimeoutMsForTesting(50) + const ctx = { + client: { + session: { + messages: () => new Promise(() => {}), + }, + }, + } + + // when + const start = Date.now() + const [first, second] = await Promise.all([ + getContextWindowUsage(ctx as never, "ses_hang_messages_parallel", { + anthropicContext1MEnabled: false, + }), + getContextWindowUsage(ctx as never, "ses_hang_messages_parallel", { + anthropicContext1MEnabled: false, + }), + ]) + const elapsed = Date.now() - start + + // then + expect(first).toBeNull() + expect(second).toBeNull() + expect(elapsed).toBeLessThan(2000) + }) + + it("#then a follow-up call after invalidation retries fresh instead of being poisoned by the timeout", async () => { + // given + _setContextWindowUsageFetchTimeoutMsForTesting(50) + let messagesCalls = 0 + let shouldHang = true + const ctx = { + client: { + session: { + messages: () => { + messagesCalls += 1 + if (shouldHang) { + return new Promise(() => {}) + } + return Promise.resolve({ + data: [ + { + info: { + role: "assistant", + providerID: "anthropic", + modelID: "claude-sonnet-4-5", + tokens: { + input: 100000, + output: 0, + reasoning: 0, + cache: { read: 0, write: 0 }, + }, + }, + }, + ], + }) + }, + }, + }, + } + + // when + const firstUsage = await getContextWindowUsage(ctx as never, "ses_hang_then_recover", { + anthropicContext1MEnabled: false, + }) + invalidateContextWindowUsageCache(ctx as never, "ses_hang_then_recover") + shouldHang = false + const secondUsage = await getContextWindowUsage(ctx as never, "ses_hang_then_recover", { + anthropicContext1MEnabled: false, + }) + + // then + expect(firstUsage).toBeNull() + expect(secondUsage?.remainingTokens).toBe(100000) + expect(messagesCalls).toBe(2) + }) + }) }) it("uses 1M limit when model cache flag is enabled", async () => { diff --git a/src/shared/dynamic-truncator.ts b/src/shared/dynamic-truncator.ts index 504f1a73b..59cbc81ca 100644 --- a/src/shared/dynamic-truncator.ts +++ b/src/shared/dynamic-truncator.ts @@ -3,10 +3,20 @@ import { resolveActualContextLimit, type ContextLimitModelCacheState, } from "./context-limit-resolver" +import { log } from "./logger" import { normalizeSDKResponse } from "./normalize-sdk-response" const CHARS_PER_TOKEN_ESTIMATE = 4; const DEFAULT_TARGET_MAX_TOKENS = 50_000; +// Hard ceiling on how long `session.messages()` is allowed to block inside +// `fetchContextWindowUsage`. Without it, a stuck OpenCode RPC (observed when +// `session.processor` enters an "Aborted process" loop) would leave the cached +// promise pending forever and every hook that calls `truncator.truncate(...)` +// would hang on it (issue #4086). +export const DEFAULT_CONTEXT_WINDOW_USAGE_FETCH_TIMEOUT_MS = 5_000; + +declare function setTimeout(callback: () => void, delay?: number): ReturnType +declare function clearTimeout(timeout: ReturnType): void interface AssistantMessageInfo { role: "assistant"; @@ -34,6 +44,16 @@ type ContextWindowUsageClient = Pick const usageCacheByClient = new WeakMap>>>() +// Test-only override for the fetch timeout used by `fetchContextWindowUsage`. +// `undefined` means "use the production default". +let contextWindowUsageFetchTimeoutMsForTesting: number | undefined = undefined + +export function _setContextWindowUsageFetchTimeoutMsForTesting( + ms: number | undefined, +): void { + contextWindowUsageFetchTimeoutMsForTesting = ms +} + function createModelCacheKey(modelCacheState?: ContextLimitModelCacheState): string { if (!modelCacheState) { return "default" @@ -184,15 +204,41 @@ export async function getContextWindowUsage( return usagePromise } +function withFetchTimeout(operation: Promise, timeoutMs: number): Promise { + if (timeoutMs <= 0) { + return operation + } + let timeoutID: ReturnType | undefined + const timeoutPromise = new Promise((_, reject) => { + timeoutID = setTimeout( + () => + reject( + new Error( + `[dynamic-truncator] session.messages timed out after ${timeoutMs}ms`, + ), + ), + timeoutMs, + ) + }) + return Promise.race([operation, timeoutPromise]).finally(() => { + if (timeoutID !== undefined) clearTimeout(timeoutID) + }) +} + async function fetchContextWindowUsage( ctx: PluginInput, sessionID: string, modelCacheState?: ContextLimitModelCacheState, ): Promise { + const fetchTimeoutMs = + contextWindowUsageFetchTimeoutMsForTesting ?? DEFAULT_CONTEXT_WINDOW_USAGE_FETCH_TIMEOUT_MS try { - const response = await ctx.client.session.messages({ - path: { id: sessionID }, - }); + const response = await withFetchTimeout( + ctx.client.session.messages({ + path: { id: sessionID }, + }), + fetchTimeoutMs, + ); const messages = normalizeSDKResponse(response, [] as MessageWrapper[], { preferResponseOnMissingData: true }) @@ -228,7 +274,11 @@ async function fetchContextWindowUsage( remainingTokens, usagePercentage: usedTokens / actualLimit, }; - } catch { + } catch (error) { + log("[dynamic-truncator] fetchContextWindowUsage failed; falling back to null", { + sessionID, + error: error instanceof Error ? error.message : String(error), + }) return null; } }