diff --git a/src/QueryEngine.autoCompactCooldown.test.ts b/src/QueryEngine.autoCompactCooldown.test.ts new file mode 100644 index 000000000..86c3094a9 --- /dev/null +++ b/src/QueryEngine.autoCompactCooldown.test.ts @@ -0,0 +1,41 @@ +import { test } from 'bun:test' +import { spawnSync } from 'node:child_process' +import { dirname, resolve } from 'node:path' +import { fileURLToPath } from 'node:url' + +const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), '..') +// Cold Windows checkouts can exceed Bun's default 5s while loading QueryEngine. +const FIXTURE_TIMEOUT_MS = 60_000 +const TEST_TIMEOUT_MS = FIXTURE_TIMEOUT_MS + 5_000 + +test('SDK manual compact clears stale auto-compact cooldown tracking', () => { + const fixture = resolve( + repoRoot, + 'src/test/fixtures/queryEngineManualCompactCooldown.fixture.ts', + ) + // Keep QueryEngine module mocks out of Bun's shared test-process cache. + const result = spawnSync(process.execPath, [fixture], { + cwd: repoRoot, + encoding: 'utf8', + timeout: FIXTURE_TIMEOUT_MS, + env: { + ...process.env, + FORCE_COLOR: '0', + }, + }) + + if (result.error) { + throw result.error + } + if (result.status !== 0) { + throw new Error( + [ + `Fixture exited with status ${result.status ?? 'unknown'}.`, + result.stdout.trim(), + result.stderr.trim(), + ] + .filter(Boolean) + .join('\n\n'), + ) + } +}, { timeout: TEST_TIMEOUT_MS }) diff --git a/src/QueryEngine.ts b/src/QueryEngine.ts index 2d2187936..407de56fc 100644 --- a/src/QueryEngine.ts +++ b/src/QueryEngine.ts @@ -34,6 +34,7 @@ import { loadMemoryPrompt } from './memdir/memdir.js' import { hasAutoMemPathOverride } from './memdir/paths.js' import { query } from './query.js' import { categorizeRetryableAPIError } from './services/api/errors.js' +import type { AutoCompactTrackingState } from './services/compact/autoCompact.js' import type { MCPServerConnection } from './services/mcp/types.js' import type { AppState } from './state/AppState.js' import { type Tools, type ToolUseContext, toolMatchesName } from './Tool.js' @@ -62,7 +63,11 @@ import { import { headlessProfilerCheckpoint } from './utils/headlessProfiler.js' import { registerStructuredOutputEnforcement } from './utils/hooks/hookHelpers.js' import { getInMemoryErrors } from './utils/log.js' -import { countToolCalls, SYNTHETIC_MESSAGES } from './utils/messages.js' +import { + countToolCalls, + isCompactBoundaryMessage, + SYNTHETIC_MESSAGES, +} from './utils/messages.js' import { getMainLoopModel, parseUserSpecifiedModel, @@ -187,6 +192,7 @@ export class QueryEngine { private totalUsage: NonNullableUsage private hasHandledOrphanedPermission = false private readFileState: FileStateCache + private autoCompactTracking: AutoCompactTrackingState | undefined // Turn-scoped skill discovery tracking (feeds was_discovered on // tengu_skill_tool_invocation). Must persist across the two // processUserInputContext rebuilds inside submitMessage, but is cleared @@ -427,6 +433,9 @@ export class QueryEngine { // Push new messages, including user input and any attachments this.mutableMessages.push(...messagesFromUserInput) + if (messagesFromUserInput.some(isCompactBoundaryMessage)) { + this.autoCompactTracking = undefined + } // Update params to reflect updates from processing /slash commands const messages = [...this.mutableMessages] @@ -681,6 +690,10 @@ export class QueryEngine { querySource: 'sdk', maxTurns, taskBudget, + autoCompactTracking: this.autoCompactTracking, + onAutoCompactTrackingChange: tracking => { + this.autoCompactTracking = tracking + }, })) { // Record assistant, user, and compact boundary messages if ( diff --git a/src/query.ts b/src/query.ts index a6ce40521..1789f7955 100644 --- a/src/query.ts +++ b/src/query.ts @@ -8,6 +8,7 @@ import { FallbackTriggeredError } from './services/api/withRetry.js' import { calculateTokenWarningState, isAutoCompactEnabled, + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, type AutoCompactTrackingState, } from './services/compact/autoCompact.js' import { buildPostCompactMessages } from './services/compact/compact.js' @@ -169,6 +170,15 @@ function* yieldMissingToolResultBlocks( const MAX_OUTPUT_TOKENS_RECOVERY_LIMIT = 3 const MAX_CONTINUATION_NUDGES = 3 +function formatAutoCompactRetryDelay(delayMs: number): string { + const totalSeconds = Math.max(1, Math.ceil(delayMs / 1000)) + if (totalSeconds < 60) { + return `${totalSeconds} second${totalSeconds === 1 ? '' : 's'}` + } + const totalMinutes = Math.ceil(totalSeconds / 60) + return `${totalMinutes} minute${totalMinutes === 1 ? '' : 's'}` +} + /** * Is this a max_output_tokens error message? If so, the streaming loop should * withhold it from SDK callers until we know whether the recovery loop can @@ -196,6 +206,10 @@ export type QueryParams = { maxOutputTokensOverride?: number maxTurns?: number skipCacheWrite?: boolean + autoCompactTracking?: AutoCompactTrackingState + onAutoCompactTrackingChange?: ( + tracking: AutoCompactTrackingState | undefined, + ) => void // API task_budget (output_config.task_budget, beta task-budgets-2026-03-13). // Distinct from the tokenBudget +500k auto-continue feature. `total` is the // budget for the whole agentic turn; `remaining` is computed per iteration @@ -288,7 +302,7 @@ async function* queryLoop( messages: params.messages, toolUseContext: params.toolUseContext, maxOutputTokensOverride: params.maxOutputTokensOverride, - autoCompactTracking: undefined, + autoCompactTracking: params.autoCompactTracking, stopHookActive: undefined, maxOutputTokensRecoveryCount: 0, hasAttemptedReactiveCompact: false, @@ -299,6 +313,12 @@ async function* queryLoop( } const budgetTracker = feature('TOKEN_BUDGET') ? createBudgetTracker() : null + const updateAutoCompactTracking = ( + tracking: AutoCompactTrackingState | undefined, + ) => { + params.onAutoCompactTrackingChange?.(tracking) + } + // task_budget.remaining tracking across compaction boundaries. Undefined // until first compact fires — while context is uncompacted the server can // see the full history and handles the countdown from {total} itself (see @@ -507,7 +527,14 @@ async function* queryLoop( ) queryCheckpoint('query_autocompact_start') - const { compactionResult, consecutiveFailures } = await deps.autocompact( + const { + compactionResult, + consecutiveFailures, + nextRetryAtMs, + lastFailureAtMs, + circuitBreakerActive, + circuitBreakerTripped, + } = await deps.autocompact( messagesForQuery, toolUseContext, { @@ -580,6 +607,7 @@ async function* queryLoop( turnCounter: 0, consecutiveFailures: 0, } + updateAutoCompactTracking(tracking) const postCompactMessages = buildPostCompactMessages(compactionResult) @@ -589,13 +617,31 @@ async function* queryLoop( // Continue on with the current query call using the post compact messages messagesForQuery = postCompactMessages - } else if (consecutiveFailures !== undefined) { - // Autocompact failed — propagate failure count so the circuit breaker - // can stop retrying on the next iteration. - tracking = { + } else if ( + consecutiveFailures !== undefined || + nextRetryAtMs !== undefined || + lastFailureAtMs !== undefined || + circuitBreakerActive !== undefined || + circuitBreakerTripped !== undefined + ) { + // Autocompact returned breaker metadata. Thread it through the loop so + // cooldown can skip retry storms, expire, and then half-open retry. + const nextTracking: AutoCompactTrackingState = { ...(tracking ?? { compacted: false, turnId: '', turnCounter: 0 }), - consecutiveFailures, } + if (consecutiveFailures !== undefined) { + nextTracking.consecutiveFailures = consecutiveFailures + } + if (nextRetryAtMs !== undefined) { + nextTracking.nextRetryAtMs = nextRetryAtMs + } else { + delete nextTracking.nextRetryAtMs + } + if (lastFailureAtMs !== undefined) { + nextTracking.lastFailureAtMs = lastFailureAtMs + } + tracking = nextTracking + updateAutoCompactTracking(tracking) } //TODO: no need to set toolUseContext.messages during set-up since it is updated here @@ -701,16 +747,15 @@ async function* queryLoop( } } - // Safety net: when auto-compact's circuit breaker has tripped (3+ - // consecutive failures), the normal blocking check above is gated on - // reactiveCompact. If reactiveCompact is also enabled but ALSO fails - // (or is disabled), the oversized context goes straight to the API and - // gets a 500. This check catches that gap — if compaction is exhausted - // and context is still over the autocompact threshold, block immediately - // with a clear message instead of burning an API call that will 500. + // Safety net: when auto-compact's circuit breaker has tripped, the normal + // blocking check above may be gated on reactiveCompact. If compaction is + // cooling down or otherwise exhausted and context is still over the + // autocompact threshold, block immediately with a clear message instead + // of burning an oversized API call. if ( tracking?.consecutiveFailures !== undefined && - tracking.consecutiveFailures >= 3 && + tracking.consecutiveFailures >= + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES && isAutoCompactEnabled() ) { const model = toolUseContext.options.mainLoopModel @@ -720,10 +765,20 @@ async function* queryLoop( model, ) if (isAboveAutoCompactThreshold) { + const nowMs = Date.now() + const retryDelayMs = + tracking.nextRetryAtMs !== undefined + ? tracking.nextRetryAtMs - nowMs + : undefined + const content = + retryDelayMs !== undefined && retryDelayMs > 0 + ? 'The conversation is over the auto-compact threshold, but automatic compaction is cooling down after repeated failures. ' + + 'OpenClaude stopped before sending another oversized request. ' + + `Retry after ${formatAutoCompactRetryDelay(retryDelayMs)}, run /compact, or start a new session with /new.` + : 'The conversation is over the auto-compact threshold and automatic compaction has failed repeatedly. ' + + 'OpenClaude stopped before sending another oversized request. Run /compact, undo recent large tool output, or start a new session with /new.' yield createAssistantAPIErrorMessage({ - content: - 'The conversation has exceeded the context limit and automatic compaction has failed. ' + - 'Press esc twice to go up a few messages and try again, or start a new session with /new.', + content, error: 'invalid_request', }) return { reason: 'blocking_limit' } @@ -1231,6 +1286,7 @@ async function* queryLoop( for (const msg of postCompactMessages) { yield msg } + updateAutoCompactTracking(undefined) const next: State = { messages: postCompactMessages, toolUseContext, @@ -1720,7 +1776,11 @@ async function* queryLoop( } if (tracking?.compacted) { - tracking.turnCounter++ + tracking = { + ...tracking, + turnCounter: tracking.turnCounter + 1, + } + updateAutoCompactTracking(tracking) logEvent('tengu_post_autocompact_turn', { turnId: tracking.turnId as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, diff --git a/src/query/autoCompactCooldown.test.ts b/src/query/autoCompactCooldown.test.ts new file mode 100644 index 000000000..214ff2a32 --- /dev/null +++ b/src/query/autoCompactCooldown.test.ts @@ -0,0 +1,339 @@ +import { afterEach, beforeEach, expect, mock, test } from 'bun:test' + +import { + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + type AutoCompactTrackingState, +} from '../services/compact/autoCompact.js' +import { + acquireSharedMutationLock, + releaseSharedMutationLock, +} from '../test/sharedMutationLock.js' +import type { Message } from '../types/message.js' +import { query } from '../query.js' +import { asSystemPrompt } from '../utils/systemPromptType.js' + +const SAVED_ENV = { + CLAUDE_AUTOCOMPACT_PCT_OVERRIDE: + process.env.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE, +} + +beforeEach(async () => { + await acquireSharedMutationLock('query/autoCompactCooldown.test.ts') + process.env.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE = '1' +}) + +afterEach(() => { + if (SAVED_ENV.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE === undefined) { + delete process.env.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE + } else { + process.env.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE = + SAVED_ENV.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE + } + releaseSharedMutationLock() +}) + +function userMessage(content: string): Message { + return { + type: 'user', + message: { role: 'user', content }, + uuid: `test-${Math.random()}`, + timestamp: new Date().toISOString(), + } +} + +function toolUseContext() { + const abortController = new AbortController() + return { + abortController, + agentId: undefined, + contentReplacementState: undefined, + options: { + agentDefinitions: { activeAgents: [] }, + allowedAgentTypes: undefined, + appendSystemPrompt: undefined, + isNonInteractiveSession: false, + mainLoopModel: 'claude-sonnet-4', + mcpClients: [], + providerOverride: undefined, + thinkingConfig: undefined, + tools: [], + }, + readFileState: {}, + getAppState: () => ({ + fastMode: false, + effortValue: undefined, + advisorModel: undefined, + mainLoopModel: 'claude-sonnet-4', + mainLoopModelForSession: undefined, + mcp: { tools: [], clients: [] }, + toolPermissionContext: { mode: 'default' }, + }), + setInProgressToolUseIDs: () => {}, + } as never +} + +function assistantToolUseMessage(): Message { + return { + type: 'assistant', + message: { + id: 'msg-test-tool-use', + role: 'assistant', + content: [ + { + type: 'tool_use', + id: 'tool-use-test', + name: 'MissingTool', + input: {}, + }, + ], + }, + uuid: 'assistant-tool-use', + timestamp: new Date().toISOString(), + } +} + +async function canUseTool() { + return { behavior: 'allow' as const } +} + +async function drain( + generator: AsyncGenerator, +): Promise<{ yielded: T[]; terminal: TReturn }> { + const yielded: T[] = [] + while (true) { + const next = await generator.next() + if (next.done) { + return { yielded, terminal: next.value } + } + yielded.push(next.value) + } +} + +test('active auto-compact cooldown blocks before model call with cooldown guidance', async () => { + const messages = [userMessage('x'.repeat(100_000))] + const nextRetryAtMs = Date.now() + 60_000 + const callModel = mock(() => { + throw new Error('model should not be called while autocompact cools down') + }) + const deps = { + callModel, + microcompact: mock(async (input: Message[]) => ({ + messages: input, + })), + autocompact: mock( + async (): Promise<{ + wasCompacted: boolean + consecutiveFailures: number + nextRetryAtMs: number + circuitBreakerActive: boolean + circuitBreakerTripped: boolean + }> => ({ + wasCompacted: false, + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs, + circuitBreakerActive: true, + circuitBreakerTripped: false, + }), + ), + uuid: () => 'test-uuid', + } as never + + const { yielded, terminal } = await drain( + query({ + messages, + systemPrompt: asSystemPrompt([]), + userContext: {}, + systemContext: {}, + canUseTool, + toolUseContext: toolUseContext(), + querySource: 'repl_main_thread', + deps, + }), + ) + + expect(callModel).not.toHaveBeenCalled() + expect(terminal.reason).toBe('blocking_limit') + + const apiError = yielded.find( + (message): message is Message => + (message as { isApiErrorMessage?: boolean }).isApiErrorMessage === true, + ) + expect(apiError).toBeDefined() + const text = apiError!.message.content[0].text + expect(text).toContain('automatic compaction is cooling down') + expect(text).toContain('Retry after') +}) + +test('auto-compact cooldown tracking is carried into the next query call', async () => { + const messages = [userMessage('x'.repeat(100_000))] + const nextRetryAtMs = Date.now() + 60_000 + const seenTracking: Array = [] + const callModel = mock(() => { + throw new Error('model should not be called while autocompact cools down') + }) + const deps = { + callModel, + microcompact: mock(async (input: Message[]) => ({ + messages: input, + })), + autocompact: mock( + async ( + _messages: never, + _toolUseContext: never, + _params: never, + _querySource: never, + tracking: AutoCompactTrackingState | undefined, + ) => { + seenTracking.push(tracking) + return { + wasCompacted: false, + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs, + circuitBreakerActive: true, + circuitBreakerTripped: false, + } + }, + ), + uuid: () => 'test-uuid', + } as never + + let persistedTracking: AutoCompactTrackingState | undefined + const queryParams = () => ({ + messages, + systemPrompt: asSystemPrompt([]), + userContext: {}, + systemContext: {}, + canUseTool, + toolUseContext: toolUseContext(), + querySource: 'repl_main_thread' as const, + deps, + autoCompactTracking: persistedTracking, + onAutoCompactTrackingChange: ( + tracking: AutoCompactTrackingState | undefined, + ) => { + persistedTracking = tracking + }, + }) + + const first = await drain(query(queryParams())) + expect(first.terminal.reason).toBe('blocking_limit') + expect(persistedTracking?.nextRetryAtMs).toBe(nextRetryAtMs) + + const second = await drain(query(queryParams())) + expect(second.terminal.reason).toBe('blocking_limit') + expect(callModel).not.toHaveBeenCalled() + expect(seenTracking).toHaveLength(2) + expect(seenTracking[0]).toBeUndefined() + expect(seenTracking[1]?.nextRetryAtMs).toBe(nextRetryAtMs) + expect(seenTracking[1]?.consecutiveFailures).toBe( + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + ) +}) + +test('post-compact turn tracking callback publishes a fresh object', async () => { + const initialTracking: AutoCompactTrackingState = { + compacted: true, + turnId: 'compact-turn', + turnCounter: 0, + consecutiveFailures: 0, + } + const trackingUpdates: AutoCompactTrackingState[] = [] + const deps = { + callModel: mock(async function* () { + yield assistantToolUseMessage() + }), + microcompact: mock(async (input: Message[]) => ({ + messages: input, + })), + autocompact: mock(async () => ({ + wasCompacted: false, + })), + uuid: () => 'test-uuid', + } as never + + const { terminal } = await drain( + query({ + messages: [userMessage('hello')], + systemPrompt: asSystemPrompt([]), + userContext: {}, + systemContext: {}, + canUseTool, + toolUseContext: toolUseContext(), + querySource: 'repl_main_thread', + maxTurns: 1, + deps, + autoCompactTracking: initialTracking, + onAutoCompactTrackingChange: tracking => { + if (tracking) { + trackingUpdates.push(tracking) + } + }, + }), + ) + + expect(terminal.reason).toBe('max_turns') + expect(trackingUpdates).toHaveLength(1) + expect(trackingUpdates[0]).not.toBe(initialTracking) + expect(trackingUpdates[0]?.turnCounter).toBe(1) + expect(initialTracking.turnCounter).toBe(0) +}) + +test('breaker metadata tracking callback publishes a fresh object', async () => { + const initialTracking: AutoCompactTrackingState = { + compacted: false, + turnId: 'turn', + turnCounter: 0, + consecutiveFailures: 2, + nextRetryAtMs: 10_000, + lastFailureAtMs: 5_000, + } + const trackingUpdates: AutoCompactTrackingState[] = [] + const deps = { + callModel: mock(() => { + throw new Error('model should not be called while autocompact cools down') + }), + microcompact: mock(async (input: Message[]) => ({ + messages: input, + })), + autocompact: mock(async () => ({ + wasCompacted: false, + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: 20_000, + lastFailureAtMs: 15_000, + circuitBreakerActive: true, + circuitBreakerTripped: true, + })), + uuid: () => 'test-uuid', + } as never + + const { terminal } = await drain( + query({ + messages: [userMessage('x'.repeat(100_000))], + systemPrompt: asSystemPrompt([]), + userContext: {}, + systemContext: {}, + canUseTool, + toolUseContext: toolUseContext(), + querySource: 'repl_main_thread', + deps, + autoCompactTracking: initialTracking, + onAutoCompactTrackingChange: tracking => { + if (tracking) { + trackingUpdates.push(tracking) + } + }, + }), + ) + + expect(terminal.reason).toBe('blocking_limit') + expect(trackingUpdates).toHaveLength(1) + expect(trackingUpdates[0]).not.toBe(initialTracking) + expect(trackingUpdates[0]?.consecutiveFailures).toBe( + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + ) + expect(trackingUpdates[0]?.nextRetryAtMs).toBe(20_000) + expect(trackingUpdates[0]?.lastFailureAtMs).toBe(15_000) + expect(initialTracking.consecutiveFailures).toBe(2) + expect(initialTracking.nextRetryAtMs).toBe(10_000) + expect(initialTracking.lastFailureAtMs).toBe(5_000) +}) diff --git a/src/screens/REPL.tsx b/src/screens/REPL.tsx index 8d7109fb4..b8c3f3501 100644 --- a/src/screens/REPL.tsx +++ b/src/screens/REPL.tsx @@ -147,6 +147,7 @@ import { useMailboxBridge } from '../hooks/useMailboxBridge.js'; import { queryCheckpoint, logQueryProfileReport } from '../utils/queryProfiler.js'; import type { Message as MessageType, UserMessage, ProgressMessage, HookResultMessage, PartialCompactDirection } from '../types/message.js'; import { query } from '../query.js'; +import type { AutoCompactTrackingState } from '../services/compact/autoCompact.js'; import { mergeClients, useMergedClients } from '../hooks/useMergedClients.js'; import { getQuerySourceForREPL } from '../utils/promptCategory.js'; import { useMergedTools } from '../hooks/useMergedTools.js'; @@ -648,6 +649,29 @@ export function REPL({ const ultraplanLaunchPending = useAppState(s => s.ultraplanLaunchPending); const viewingAgentTaskId = useAppState(s => s.viewingAgentTaskId); const setAppState = useSetAppState(); + const autoCompactTrackingBySessionRef = useRef(new Map, AutoCompactTrackingState>()); + const getAutoCompactTrackingForSession = useCallback((sessionId: ReturnType) => autoCompactTrackingBySessionRef.current.get(sessionId), []); + const setAutoCompactTrackingForSession = useCallback((sessionId: ReturnType, tracking: AutoCompactTrackingState | undefined) => { + if (tracking) { + autoCompactTrackingBySessionRef.current.set(sessionId, tracking); + } else { + autoCompactTrackingBySessionRef.current.delete(sessionId); + } + }, []); + const setAutoCompactTrackingForSessionIfUnchanged = useCallback((sessionId: ReturnType, expected: AutoCompactTrackingState | undefined, tracking: AutoCompactTrackingState | undefined) => { + if (autoCompactTrackingBySessionRef.current.get(sessionId) !== expected) { + return false; + } + if (tracking) { + autoCompactTrackingBySessionRef.current.set(sessionId, tracking); + } else { + autoCompactTrackingBySessionRef.current.delete(sessionId); + } + return true; + }, []); + const resetAutoCompactTracking = useCallback(() => { + autoCompactTrackingBySessionRef.current.clear(); + }, []); // Bootstrap: retained local_agent that hasn't loaded disk yet → read // sidechain JSONL and UUID-merge with whatever stream has appended so far. @@ -1806,6 +1830,7 @@ export function REPL({ const resume = useCallback(async (sessionId: UUID, log: LogOption, entrypoint: ResumeEntrypoint) => { const resumeStart = performance.now(); try { + resetAutoCompactTracking(); // Deserialize messages to properly clean up the conversation // This filters unresolved tool uses and adds a synthetic assistant message if needed const messages = deserializeMessages(log.messages); @@ -2024,7 +2049,7 @@ export function REPL({ }); throw error; } - }, [resetLoadingState, setAppState]); + }, [resetLoadingState, resetAutoCompactTracking, setAppState]); // Lazy init: useRef(createX()) would call createX on every render and // discard the result. LRUCache construction inside FileStateCache is @@ -2630,6 +2655,7 @@ export function REPL({ // Session backgrounding (Ctrl+B to background/foreground) const handleBackgroundQuery = useCallback(() => { + const backgroundSessionId = getSessionId(); // Stop the foreground query so the background one takes over abortController?.abort('background'); // Aborting subagents may produce task-completed notifications. @@ -2670,14 +2696,18 @@ export function REPL({ systemContext, canUseTool, toolUseContext, - querySource: getQuerySourceForREPL() + querySource: getQuerySourceForREPL(), + autoCompactTracking: getAutoCompactTrackingForSession(backgroundSessionId), + onAutoCompactTrackingChange: tracking => { + setAutoCompactTrackingForSession(backgroundSessionId, tracking); + } }, description: terminalTitle, setAppState, agentDefinition: mainThreadAgentDefinition }); })(); - }, [abortController, mainLoopModel, toolPermissionContext, mainThreadAgentDefinition, getToolUseContext, customSystemPrompt, appendSystemPrompt, canUseTool, setAppState]); + }, [abortController, mainLoopModel, toolPermissionContext, mainThreadAgentDefinition, getToolUseContext, customSystemPrompt, appendSystemPrompt, canUseTool, setAppState, getAutoCompactTrackingForSession, setAutoCompactTrackingForSession]); const { handleBackgroundSession } = useSessionBackgrounding({ @@ -2838,6 +2868,7 @@ export function REPL({ // handleMessageFromStream. Clear context-blocked if a compact boundary // is present so proactive ticks resume after compaction. if (newMessages.some(isCompactBoundaryMessage)) { + setAutoCompactTrackingForSession(getSessionId(), undefined); // Bump conversationId so Messages.tsx row keys change and // stale memoized rows remount with post-compact content. setConversationId(randomUUID()); @@ -2850,6 +2881,8 @@ export function REPL({ return; } const toolUseContext = getToolUseContext(messagesIncludingNewMessages, newMessages, abortController, mainLoopModelParam); + const querySessionId = getSessionId(); + const queryAutoCompactTracking = getAutoCompactTrackingForSession(querySessionId); // getToolUseContext reads tools/mcpClients fresh from store.getState() // (via computeTools/mergeClients). Use those rather than the closure- // captured `tools`/`mcpClients` — useManageMCPConnections may have @@ -2896,6 +2929,7 @@ export function REPL({ resetTurnHookDuration(); resetTurnToolDuration(); resetTurnClassifierDuration(); + let expectedAutoCompactTracking = queryAutoCompactTracking; for await (const event of query({ messages: messagesIncludingNewMessages, systemPrompt, @@ -2903,7 +2937,13 @@ export function REPL({ systemContext, canUseTool, toolUseContext, - querySource: getQuerySourceForREPL() + querySource: getQuerySourceForREPL(), + autoCompactTracking: queryAutoCompactTracking, + onAutoCompactTrackingChange: tracking => { + if (setAutoCompactTrackingForSessionIfUnchanged(querySessionId, expectedAutoCompactTracking, tracking)) { + expectedAutoCompactTracking = tracking; + } + } })) { onQueryEvent(event); } @@ -2957,7 +2997,7 @@ export function REPL({ // Signal that a query turn has completed successfully await onTurnComplete?.(messagesRef.current); - }, [initialMcpClients, resetLoadingState, getToolUseContext, toolPermissionContext, setAppState, customSystemPrompt, onTurnComplete, appendSystemPrompt, canUseTool, mainThreadAgentDefinition, onQueryEvent, sessionTitle, titleDisabled]); + }, [initialMcpClients, resetLoadingState, getToolUseContext, toolPermissionContext, setAppState, customSystemPrompt, onTurnComplete, appendSystemPrompt, canUseTool, mainThreadAgentDefinition, onQueryEvent, sessionTitle, titleDisabled, getAutoCompactTrackingForSession, setAutoCompactTrackingForSession, setAutoCompactTrackingForSessionIfUnchanged]); const onQuery = useCallback(async (newMessages: MessageType[], abortController: AbortController, shouldQuery: boolean, additionalAllowedTools: string[], mainLoopModelParam: string, onBeforeQueryCallback?: (input: string, newMessages: MessageType[]) => Promise, input?: string, effort?: EffortValue): Promise => { // If this is a teammate, mark them as active when starting a turn if (isAgentSwarmsEnabled()) { @@ -3180,6 +3220,7 @@ export function REPL({ async function processInitialMessage(initialMsg: NonNullable) { // Clear context if requested (plan mode exit) if (initialMsg.clearContext) { + resetAutoCompactTracking(); // Preserve the plan slug before clearing context, so the new session // can access the same plan file after regenerateSessionId() const oldPlanSlug = initialMsg.message.planContent ? getPlanSlug() : undefined; @@ -3283,7 +3324,7 @@ export function REPL({ }, 100, initialMessageRef); } void processInitialMessage(pending); - }, [initialMessage, isLoading, setMessages, setAppState, onQuery, mainLoopModel, tools]); + }, [initialMessage, isLoading, setMessages, setAppState, onQuery, mainLoopModel, tools, resetAutoCompactTracking]); const onSubmit = useCallback(async (input: string, helpers: PromptInputHelpers, speculationAccept?: { state: ActiveSpeculationState; speculationSessionTimeSavedMs: number; @@ -3802,6 +3843,7 @@ export function REPL({ rewindToMessageIndex: messageIndex }); setMessages(prev.slice(0, messageIndex)); + resetAutoCompactTracking(); // Careful, this has to happen after setMessages setConversationId(randomUUID()); // Reset cached microcompact state so stale pinned cache edits @@ -3837,7 +3879,7 @@ export function REPL({ generationRequestId: null } })); - }, [setMessages, setAppState]); + }, [setMessages, resetAutoCompactTracking, setAppState]); // Synchronous rewind + input population. Used directly by auto-restore on // interrupt (so React batches with the abort's setMessages → single render, @@ -4904,6 +4946,7 @@ export function REPL({ }); } if (action === 'clear') { + resetAutoCompactTracking(); const { clearConversation } = await import('../commands/clear/conversation.js'); @@ -5089,7 +5132,8 @@ export function REPL({ setMessages(postCompact); } // Partial compact bypasses handleMessageFromStream — clear - // the context-blocked flag so proactive ticks resume. + // the auto-compact breaker and context-blocked flag. + setAutoCompactTrackingForSession(getSessionId(), undefined); if (feature('PROACTIVE') || feature('KAIROS')) { proactiveModule?.setContextBlocked(false); } diff --git a/src/services/compact/autoCompact.test.ts b/src/services/compact/autoCompact.test.ts index 4f6a2ab5e..5691caf1f 100644 --- a/src/services/compact/autoCompact.test.ts +++ b/src/services/compact/autoCompact.test.ts @@ -1,12 +1,44 @@ -import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' +import { + afterEach, + beforeEach, + describe, + expect, + mock, + test, +} from 'bun:test' import { acquireSharedMutationLock, releaseSharedMutationLock, } from '../../test/sharedMutationLock.js' +import type { Message } from '../../types/message.js' +import * as realConfig from '../../utils/config.js' -async function importAutoCompact() { +const USER_ABORT_MESSAGE = 'API Error: Request was aborted.' + +type ImportAutoCompactOptions = { + compactConversation?: ReturnType + trySessionMemoryCompaction?: ReturnType +} + +async function importAutoCompact(options: ImportAutoCompactOptions = {}) { mock.restore() + mock.module('../../utils/config.js', () => ({ + ...realConfig, + getGlobalConfig: () => ({ autoCompactEnabled: true }), + })) + if (options.compactConversation) { + mock.module('./compact.js', () => ({ + ERROR_MESSAGE_USER_ABORT: USER_ABORT_MESSAGE, + buildPostCompactMessages: mock(() => []), + compactConversation: options.compactConversation, + })) + } + if (options.trySessionMemoryCompaction) { + mock.module('./sessionMemoryCompact.js', () => ({ + trySessionMemoryCompaction: options.trySessionMemoryCompaction, + })) + } const nonce = `${Date.now()}-${Math.random()}` return import(`./autoCompact.ts?test=${nonce}`) } @@ -39,6 +71,12 @@ const SAVED_ENV = { process.env.CLAUDE_CODE_AUTO_COMPACT_WINDOW, CLAUDE_CODE_MAX_OUTPUT_TOKENS: process.env.CLAUDE_CODE_MAX_OUTPUT_TOKENS, + CLAUDE_AUTOCOMPACT_PCT_OVERRIDE: + process.env.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE, + OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS: + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS, + DISABLE_COMPACT: process.env.DISABLE_COMPACT, + DISABLE_AUTO_COMPACT: process.env.DISABLE_AUTO_COMPACT, } function restoreEnv(): void { @@ -53,16 +91,70 @@ function restoreEnv(): void { beforeEach(async () => { await acquireSharedMutationLock('services/compact/autoCompact.test.ts') + delete process.env.DISABLE_COMPACT + delete process.env.DISABLE_AUTO_COMPACT + delete process.env.CLAUDE_CODE_MAX_CONTEXT_TOKENS + delete process.env.CLAUDE_CODE_AUTO_COMPACT_WINDOW + delete process.env.CLAUDE_CODE_MAX_OUTPUT_TOKENS }) afterEach(() => { try { + mock.restore() restoreEnv() } finally { releaseSharedMutationLock() } }) +function userMessage(content: string): Message { + return { + type: 'user', + message: { role: 'user', content }, + uuid: `test-${Math.random()}`, + timestamp: new Date().toISOString(), + } +} + +function overThresholdMessages(): Message[] { + return [userMessage('x'.repeat(100_000))] +} + +function underThresholdMessages(): Message[] { + return [userMessage('small conversation')] +} + +function toolUseContext() { + return { + agentId: undefined, + options: { + mainLoopModel: 'claude-sonnet-4', + }, + } as never +} + +function cacheSafeParams(messages: Message[]) { + const context = toolUseContext() + return { + systemPrompt: [], + userContext: {}, + systemContext: {}, + toolUseContext: context, + forkContextMessages: messages, + } as never +} + +function compactResult() { + return { + summaryMessages: [userMessage('summary')], + attachments: [], + hookResults: [], + preCompactTokenCount: 10_000, + postCompactTokenCount: 100, + truePostCompactTokenCount: 100, + } as never +} + describe('getEffectiveContextWindowSize', () => { test('returns positive value for known models with large context windows', async () => { const { getEffectiveContextWindowSize } = await importAutoCompact() @@ -153,3 +245,512 @@ describe('getAutoCompactThreshold', () => { } }) }) + +describe('getAutoCompactFailureCooldownMs', () => { + test('uses valid positive integer override', async () => { + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = ' 5000 ' + const { getAutoCompactFailureCooldownMs } = await importAutoCompact() + + expect(getAutoCompactFailureCooldownMs()).toBe(5000) + }) + + test('ignores partial or invalid override values', async () => { + const { + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + getAutoCompactFailureCooldownMs, + } = await importAutoCompact() + + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '5000ms' + expect(getAutoCompactFailureCooldownMs()).toBe( + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + ) + + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '-1' + expect(getAutoCompactFailureCooldownMs()).toBe( + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + ) + + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '1.5' + expect(getAutoCompactFailureCooldownMs()).toBe( + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + ) + + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '1e3' + expect(getAutoCompactFailureCooldownMs()).toBe( + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + ) + + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '0x10' + expect(getAutoCompactFailureCooldownMs()).toBe( + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + ) + + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '0b10' + expect(getAutoCompactFailureCooldownMs()).toBe( + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + ) + + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '+5' + expect(getAutoCompactFailureCooldownMs()).toBe( + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + ) + + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '5.0' + expect(getAutoCompactFailureCooldownMs()).toBe( + AUTOCOMPACT_FAILURE_COOLDOWN_MS, + ) + }) +}) + +describe('resolveAutoCompactCircuitBreakerState', () => { + test('skips compaction while cooldown is active', async () => { + const { + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + resolveAutoCompactCircuitBreakerState, + } = await importAutoCompact() + + expect( + resolveAutoCompactCircuitBreakerState({ + tracking: { + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: 10_000, + }, + nowMs: 9_000, + cooldownMs: 5_000, + }), + ).toEqual({ + action: 'skip', + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: 10_000, + circuitBreakerActive: true, + }) + }) + + test('allows exactly one half-open retry after cooldown expires', async () => { + const { + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + resolveAutoCompactCircuitBreakerState, + } = await importAutoCompact() + + expect( + resolveAutoCompactCircuitBreakerState({ + tracking: { + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: 10_000, + }, + nowMs: 10_001, + cooldownMs: 5_000, + }), + ).toEqual({ + action: 'allow', + effectiveConsecutiveFailures: + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES - 1, + wasHalfOpen: true, + }) + }) + + test('derives active cooldown from failure time when retry time is absent', async () => { + const { + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + resolveAutoCompactCircuitBreakerState, + } = await importAutoCompact() + + expect( + resolveAutoCompactCircuitBreakerState({ + tracking: { + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + lastFailureAtMs: 5_000, + }, + nowMs: 11_000, + cooldownMs: 7_000, + }), + ).toEqual({ + action: 'skip', + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: 12_000, + circuitBreakerActive: true, + }) + }) + + test.each([ + ['NaN', Number.NaN], + ['Infinity', Number.POSITIVE_INFINITY], + ])( + 'derives active cooldown from failure time when retry time is %s', + async (_label, nextRetryAtMs) => { + const { + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + resolveAutoCompactCircuitBreakerState, + } = await importAutoCompact() + + expect( + resolveAutoCompactCircuitBreakerState({ + tracking: { + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs, + lastFailureAtMs: 5_000, + }, + nowMs: 11_000, + cooldownMs: 7_000, + }), + ).toEqual({ + action: 'skip', + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: 12_000, + circuitBreakerActive: true, + }) + }, + ) + + test('uses explicit retry time before deriving cooldown from failure time', async () => { + const { + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + resolveAutoCompactCircuitBreakerState, + } = await importAutoCompact() + + expect( + resolveAutoCompactCircuitBreakerState({ + tracking: { + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: 10_000, + lastFailureAtMs: 50_000, + }, + nowMs: 10_001, + cooldownMs: 7_000, + }), + ).toEqual({ + action: 'allow', + effectiveConsecutiveFailures: + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES - 1, + wasHalfOpen: true, + }) + }) + + test('allows half-open retry after derived cooldown expires', async () => { + const { + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + resolveAutoCompactCircuitBreakerState, + } = await importAutoCompact() + + expect( + resolveAutoCompactCircuitBreakerState({ + tracking: { + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + lastFailureAtMs: 5_000, + }, + nowMs: 10_001, + cooldownMs: 5_000, + }), + ).toEqual({ + action: 'allow', + effectiveConsecutiveFailures: + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES - 1, + wasHalfOpen: true, + }) + }) +}) + +describe('autoCompactIfNeeded circuit breaker', () => { + beforeEach(() => { + process.env.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE = '1' + process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS = '5000' + }) + + test('trips after three non-user failures and records a retry time', async () => { + const compactConversation = mock(async () => { + throw new Error('provider down') + }) + const trySessionMemoryCompaction = mock(async () => null) + const { + autoCompactIfNeeded, + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + } = await importAutoCompact({ + compactConversation, + trySessionMemoryCompaction, + }) + + const messages = overThresholdMessages() + let tracking: { + compacted: boolean + turnCounter: number + turnId: string + consecutiveFailures?: number + } = { + compacted: false, + turnCounter: 0, + turnId: 'turn', + } + let result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + tracking, + ) + expect(result.consecutiveFailures).toBe(1) + expect(result.nextRetryAtMs).toBeUndefined() + + tracking = { ...tracking, consecutiveFailures: result.consecutiveFailures } + result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + tracking, + ) + expect(result.consecutiveFailures).toBe(2) + expect(result.nextRetryAtMs).toBeUndefined() + + tracking = { ...tracking, consecutiveFailures: result.consecutiveFailures } + result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + tracking, + ) + + expect(compactConversation).toHaveBeenCalledTimes(3) + expect(result.consecutiveFailures).toBe( + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + ) + expect(result.nextRetryAtMs).toBeGreaterThan(Date.now()) + expect(result.circuitBreakerTripped).toBe(true) + }) + + test('active cooldown skips compaction attempts', async () => { + const compactConversation = mock(async () => compactResult()) + const trySessionMemoryCompaction = mock(async () => null) + const { + autoCompactIfNeeded, + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + } = await importAutoCompact({ + compactConversation, + trySessionMemoryCompaction, + }) + + const messages = overThresholdMessages() + const result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + { + compacted: false, + turnCounter: 0, + turnId: 'turn', + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: Date.now() + 60_000, + }, + ) + + expect(compactConversation).not.toHaveBeenCalled() + expect(result.wasCompacted).toBe(false) + expect(result.circuitBreakerActive).toBe(true) + expect(result.nextRetryAtMs).toBeGreaterThan(Date.now()) + }) + + test('expired cooldown allows a half-open compaction attempt', async () => { + const compactConversation = mock(async () => compactResult()) + const trySessionMemoryCompaction = mock(async () => null) + const { + autoCompactIfNeeded, + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + } = await importAutoCompact({ + compactConversation, + trySessionMemoryCompaction, + }) + + const messages = overThresholdMessages() + const result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + { + compacted: false, + turnCounter: 0, + turnId: 'turn', + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: Date.now() - 1, + }, + ) + + expect(compactConversation).toHaveBeenCalledTimes(1) + expect(result.wasCompacted).toBe(true) + expect(result.consecutiveFailures).toBe(0) + expect(result.nextRetryAtMs).toBeUndefined() + }) + + test('half-open failure immediately re-trips instead of growing unbounded', async () => { + const compactConversation = mock(async () => { + throw new Error('still broken') + }) + const trySessionMemoryCompaction = mock(async () => null) + const { + autoCompactIfNeeded, + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + } = await importAutoCompact({ + compactConversation, + trySessionMemoryCompaction, + }) + + const messages = overThresholdMessages() + const result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + { + compacted: false, + turnCounter: 0, + turnId: 'turn', + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: Date.now() - 1, + }, + ) + + expect(compactConversation).toHaveBeenCalledTimes(1) + expect(result.consecutiveFailures).toBe( + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + ) + expect(result.nextRetryAtMs).toBeGreaterThan(Date.now()) + expect(result.circuitBreakerTripped).toBe(true) + }) + + test('failed compaction cooldown starts at failure time, not attempt start', async () => { + let nowMs = 100_000 + const originalDateNow = Date.now + Date.now = mock(() => nowMs) as never + try { + const compactConversation = mock(async () => { + nowMs = 106_000 + throw new Error('slow provider failure') + }) + const trySessionMemoryCompaction = mock(async () => null) + const { autoCompactIfNeeded } = await importAutoCompact({ + compactConversation, + trySessionMemoryCompaction, + }) + + const messages = overThresholdMessages() + const result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + { + compacted: false, + turnCounter: 0, + turnId: 'turn', + consecutiveFailures: 2, + }, + ) + + expect(result.lastFailureAtMs).toBe(106_000) + expect(result.nextRetryAtMs).toBe(111_000) + } finally { + Date.now = originalDateNow + } + }) + + test('user abort does not increment failures or trip cooldown', async () => { + const compactConversation = mock(async () => { + throw new Error(USER_ABORT_MESSAGE) + }) + const trySessionMemoryCompaction = mock(async () => null) + const { autoCompactIfNeeded } = await importAutoCompact({ + compactConversation, + trySessionMemoryCompaction, + }) + + const messages = overThresholdMessages() + const result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + { + compacted: false, + turnCounter: 0, + turnId: 'turn', + consecutiveFailures: 2, + }, + ) + + expect(compactConversation).toHaveBeenCalledTimes(1) + expect(result.consecutiveFailures).toBe(2) + expect(result.nextRetryAtMs).toBeUndefined() + expect(result.circuitBreakerTripped).toBe(false) + }) + + test('user abort during half-open retry clears expired cooldown without retripping', async () => { + const compactConversation = mock(async () => { + throw new Error(USER_ABORT_MESSAGE) + }) + const trySessionMemoryCompaction = mock(async () => null) + const { + autoCompactIfNeeded, + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + } = await importAutoCompact({ + compactConversation, + trySessionMemoryCompaction, + }) + + const messages = overThresholdMessages() + const result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + { + compacted: false, + turnCounter: 0, + turnId: 'turn', + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: Date.now() - 1, + }, + ) + + expect(compactConversation).toHaveBeenCalledTimes(1) + expect(result.consecutiveFailures).toBe( + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES - 1, + ) + expect(result.nextRetryAtMs).toBeUndefined() + expect(result.circuitBreakerActive).toBe(false) + expect(result.circuitBreakerTripped).toBe(false) + }) + + test('below-threshold conversations clear stale breaker state', async () => { + const compactConversation = mock(async () => compactResult()) + const trySessionMemoryCompaction = mock(async () => null) + const { + autoCompactIfNeeded, + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + } = await importAutoCompact({ + compactConversation, + trySessionMemoryCompaction, + }) + + const messages = underThresholdMessages() + const result = await autoCompactIfNeeded( + messages, + toolUseContext(), + cacheSafeParams(messages), + 'repl_main_thread', + { + compacted: false, + turnCounter: 0, + turnId: 'turn', + consecutiveFailures: MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + nextRetryAtMs: Date.now() + 60_000, + }, + ) + + expect(compactConversation).not.toHaveBeenCalled() + expect(result.wasCompacted).toBe(false) + expect(result.circuitBreakerActive).toBe(false) + expect(result.consecutiveFailures).toBe(0) + expect(result.nextRetryAtMs).toBeUndefined() + }) +}) diff --git a/src/services/compact/autoCompact.ts b/src/services/compact/autoCompact.ts index d7e9b28eb..7767aae11 100644 --- a/src/services/compact/autoCompact.ts +++ b/src/services/compact/autoCompact.ts @@ -61,9 +61,13 @@ export type AutoCompactTrackingState = { // Unique ID per turn turnId: string // Consecutive autocompact failures. Reset on success. - // Used as a circuit breaker to stop retrying when the context is - // irrecoverably over the limit (e.g., prompt_too_long). + // Used by the cooldown circuit breaker to avoid retry storms when the + // context is irrecoverably over the limit (e.g., prompt_too_long). consecutiveFailures?: number + // Process-local retry timestamp for the cooldown breaker. This state is + // threaded through query() callers rather than serialized into transcripts. + nextRetryAtMs?: number + lastFailureAtMs?: number } export const AUTOCOMPACT_BUFFER_TOKENS = 13_000 @@ -71,10 +75,84 @@ export const WARNING_THRESHOLD_BUFFER_TOKENS = 20_000 export const ERROR_THRESHOLD_BUFFER_TOKENS = 20_000 export const MANUAL_COMPACT_BUFFER_TOKENS = 3_000 -// Stop trying autocompact after this many consecutive failures. +export const AUTOCOMPACT_FAILURE_COOLDOWN_MS = 5 * 60 * 1000 + +// Pause autocompact after this many consecutive failures. // BQ 2026-03-10: 1,279 sessions had 50+ consecutive failures (up to 3,272) // in a single session, wasting ~250K API calls/day globally. -const MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES = 3 +export const MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES = 3 + +export function getAutoCompactFailureCooldownMs(): number { + const override = process.env.OPENCLAUDE_AUTOCOMPACT_FAILURE_COOLDOWN_MS + if (override) { + const trimmed = override.trim() + const parsed = Number(trimmed) + if (/^[1-9]\d*$/.test(trimmed) && Number.isSafeInteger(parsed)) { + return parsed + } + } + return AUTOCOMPACT_FAILURE_COOLDOWN_MS +} + +export function resolveAutoCompactCircuitBreakerState(args: { + tracking?: Pick< + AutoCompactTrackingState, + 'consecutiveFailures' | 'nextRetryAtMs' | 'lastFailureAtMs' + > + nowMs: number + cooldownMs: number +}): + | { + action: 'allow' + effectiveConsecutiveFailures: number + wasHalfOpen: boolean + } + | { + action: 'skip' + consecutiveFailures: number + nextRetryAtMs: number + circuitBreakerActive: true + } { + const { tracking, nowMs, cooldownMs } = args + const consecutiveFailures = Math.max(0, tracking?.consecutiveFailures ?? 0) + if (consecutiveFailures < MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES) { + return { + action: 'allow', + effectiveConsecutiveFailures: consecutiveFailures, + wasHalfOpen: false, + } + } + + let nextRetryAtMs = tracking?.nextRetryAtMs + if ( + (typeof nextRetryAtMs !== 'number' || + !Number.isFinite(nextRetryAtMs)) && + typeof tracking?.lastFailureAtMs === 'number' && + Number.isFinite(tracking.lastFailureAtMs) && + Number.isFinite(cooldownMs) + ) { + nextRetryAtMs = tracking.lastFailureAtMs + cooldownMs + } + if ( + typeof nextRetryAtMs === 'number' && + Number.isFinite(nextRetryAtMs) && + nowMs < nextRetryAtMs + ) { + return { + action: 'skip', + consecutiveFailures, + nextRetryAtMs, + circuitBreakerActive: true, + } + } + + return { + action: 'allow', + effectiveConsecutiveFailures: + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES - 1, + wasHalfOpen: true, + } +} export function getAutoCompactThreshold(model: string): number { const effectiveContextWindow = getEffectiveContextWindowSize(model) @@ -261,21 +339,15 @@ export async function autoCompactIfNeeded( wasCompacted: boolean compactionResult?: CompactionResult consecutiveFailures?: number + nextRetryAtMs?: number + lastFailureAtMs?: number + circuitBreakerActive?: boolean + circuitBreakerTripped?: boolean }> { if (isEnvTruthy(process.env.DISABLE_COMPACT)) { return { wasCompacted: false } } - // Circuit breaker: stop retrying after N consecutive failures. - // Without this, sessions where context is irrecoverably over the limit - // hammer the API with doomed compaction attempts on every turn. - if ( - tracking?.consecutiveFailures !== undefined && - tracking.consecutiveFailures >= MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES - ) { - return { wasCompacted: false } - } - const model = toolUseContext.options.mainLoopModel const shouldCompact = await shouldAutoCompact( messages, @@ -285,9 +357,44 @@ export async function autoCompactIfNeeded( ) if (!shouldCompact) { + if ((tracking?.consecutiveFailures ?? 0) > 0 || tracking?.nextRetryAtMs) { + return { + wasCompacted: false, + consecutiveFailures: 0, + circuitBreakerActive: false, + circuitBreakerTripped: false, + } + } return { wasCompacted: false } } + const now = Date.now() + const cooldownMs = getAutoCompactFailureCooldownMs() + const breakerState = resolveAutoCompactCircuitBreakerState({ + tracking, + nowMs: now, + cooldownMs, + }) + + if (breakerState.action === 'skip') { + return { + wasCompacted: false, + consecutiveFailures: breakerState.consecutiveFailures, + nextRetryAtMs: breakerState.nextRetryAtMs, + circuitBreakerActive: true, + circuitBreakerTripped: false, + } + } + + const effectiveTracking: AutoCompactTrackingState | undefined = + tracking && breakerState.wasHalfOpen + ? { + ...tracking, + consecutiveFailures: breakerState.effectiveConsecutiveFailures, + nextRetryAtMs: undefined, + } + : tracking + const contextWindow = getContextWindowForModel(model, getSdkBetas()) const partitioned = partitionContext(messages, { @@ -322,9 +429,9 @@ export async function autoCompactIfNeeded( } const recompactionInfo: RecompactionInfo = { - isRecompactionInChain: tracking?.compacted === true, - turnsSincePreviousCompact: tracking?.turnCounter ?? -1, - previousCompactTurnId: tracking?.turnId, + isRecompactionInChain: effectiveTracking?.compacted === true, + turnsSincePreviousCompact: effectiveTracking?.turnCounter ?? -1, + previousCompactTurnId: effectiveTracking?.turnId, autoCompactThreshold: getAutoCompactThreshold(model), querySource, } @@ -351,6 +458,7 @@ export async function autoCompactIfNeeded( return { wasCompacted: true, compactionResult: sessionMemoryResult, + consecutiveFailures: 0, } } @@ -377,20 +485,46 @@ export async function autoCompactIfNeeded( consecutiveFailures: 0, } } catch (error) { - if (!hasExactErrorMessage(error, ERROR_MESSAGE_USER_ABORT)) { - logError(error) + const wasUserAbort = hasExactErrorMessage(error, ERROR_MESSAGE_USER_ABORT) + if (wasUserAbort) { + return { + wasCompacted: false, + consecutiveFailures: breakerState.effectiveConsecutiveFailures, + nextRetryAtMs: breakerState.wasHalfOpen + ? undefined + : tracking?.nextRetryAtMs, + circuitBreakerActive: false, + circuitBreakerTripped: false, + } } + + logError(error) // Increment consecutive failure count for circuit breaker. // The caller threads this through autoCompactTracking so the - // next query loop iteration can skip futile retry attempts. - const prevFailures = tracking?.consecutiveFailures ?? 0 - const nextFailures = prevFailures + 1 - if (nextFailures >= MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES) { + // next query loop iteration can skip futile retry attempts until cooldown. + const nextFailures = Math.min( + breakerState.effectiveConsecutiveFailures + 1, + MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES, + ) + const circuitBreakerTripped = + nextFailures >= MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES + const failureAtMs = Date.now() + const nextRetryAtMs = circuitBreakerTripped + ? failureAtMs + cooldownMs + : undefined + if (circuitBreakerTripped) { logForDebugging( - `autocompact: circuit breaker tripped after ${nextFailures} consecutive failures — skipping future attempts this session`, + `autocompact: circuit breaker tripped after ${nextFailures} consecutive failures — retrying after cooldown`, { level: 'warn' }, ) } - return { wasCompacted: false, consecutiveFailures: nextFailures } + return { + wasCompacted: false, + consecutiveFailures: nextFailures, + nextRetryAtMs, + lastFailureAtMs: failureAtMs, + circuitBreakerActive: circuitBreakerTripped, + circuitBreakerTripped, + } } } diff --git a/src/test/fixtures/queryEngineManualCompactCooldown.fixture.ts b/src/test/fixtures/queryEngineManualCompactCooldown.fixture.ts new file mode 100644 index 000000000..554fe364f --- /dev/null +++ b/src/test/fixtures/queryEngineManualCompactCooldown.fixture.ts @@ -0,0 +1,156 @@ +import { mock } from 'bun:test' +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { + isSessionPersistenceDisabled, + setSessionPersistenceDisabled, +} from '../../bootstrap/state.js' +import type { AutoCompactTrackingState } from '../../services/compact/autoCompact.js' +import type { AppState } from '../../state/AppState.js' +import type { Message } from '../../types/message.js' + +function compactBoundaryMessage(): Message { + return { + type: 'system', + subtype: 'compact_boundary', + content: 'Conversation compacted', + isMeta: false, + timestamp: new Date().toISOString(), + uuid: `test-${Math.random()}`, + level: 'info', + compactMetadata: { + trigger: 'manual', + preTokens: 100_000, + }, + } +} + +async function drain(generator: AsyncGenerator): Promise { + for await (const _event of generator) { + // Drain the SDK generator so submitMessage completes. + } +} + +const savedSessionPersistenceDisabled = isSessionPersistenceDisabled() +const fixtureCwd = mkdtempSync( + join(tmpdir(), 'openclaude-query-engine-cooldown-'), +) + +try { + const trippedTracking: AutoCompactTrackingState = { + compacted: false, + turnCounter: 0, + turnId: 'turn', + consecutiveFailures: 3, + nextRetryAtMs: Date.now() + 60_000, + lastFailureAtMs: Date.now(), + } + const processUserInputResults = [ + { + messages: [compactBoundaryMessage()], + shouldQuery: false, + allowedTools: [], + model: undefined, + resultText: 'compacted', + }, + ] + + mock.module('../../utils/processUserInput/processUserInput.js', () => ({ + processUserInput: mock(async () => { + const result = processUserInputResults.shift() + if (!result) { + throw new Error('unexpected processUserInput call') + } + return result + }), + })) + mock.module('../../utils/queryContext.js', () => ({ + fetchSystemPromptParts: mock(async () => ({ + defaultSystemPrompt: [], + userContext: {}, + systemContext: {}, + })), + })) + mock.module('../../utils/messages/systemInit.js', () => ({ + buildSystemInitMessage: mock(() => ({ + type: 'system', + subtype: 'init', + session_id: 'test-session', + tools: [], + mcp_servers: [], + model: 'claude-sonnet-4', + permissionMode: 'default', + apiKeySource: 'none', + cwd: fixtureCwd, + })), + sdkCompatToolName: (name: string) => name, + })) + mock.module('../../commands.js', () => ({ + REMOTE_SAFE_COMMANDS: new Set(), + builtInCommandNames: new Set(), + clearCommandsCache: () => {}, + findCommand: () => undefined, + getCommand: () => undefined, + getCommandName: (command: { name?: string }) => command.name ?? '', + getCommands: () => [], + getMcpSkillCommands: () => [], + getSkillToolCommands: () => [], + getSlashCommandToolSkills: mock(async () => []), + hasCommand: () => false, + isCommandEnabled: () => true, + })) + mock.module('src/entrypoints/agentSdkTypes.js', () => ({ + HOOK_EVENTS: [], + })) + setSessionPersistenceDisabled(true) + + const { QueryEngine } = await import('../../QueryEngine.js') + let appState = { + fastMode: false, + toolPermissionContext: { + mode: 'default', + additionalWorkingDirectories: new Map(), + alwaysAllowRules: {}, + }, + fileHistory: {}, + attribution: {}, + } as unknown as AppState + const engine = new QueryEngine({ + cwd: fixtureCwd, + tools: [], + commands: [], + mcpClients: [], + agents: [], + canUseTool: async () => ({ behavior: 'allow' }), + getAppState: () => appState, + setAppState: updater => { + appState = updater(appState) + }, + readFileCache: {} as never, + userSpecifiedModel: 'claude-sonnet-4', + thinkingConfig: { type: 'disabled' }, + }) + ;( + engine as unknown as { + autoCompactTracking?: AutoCompactTrackingState + } + ).autoCompactTracking = trippedTracking + + await drain(engine.submitMessage('/compact')) + + assert.equal( + ( + engine as unknown as { + autoCompactTracking?: AutoCompactTrackingState + } + ).autoCompactTracking, + undefined, + ) +} finally { + mock.restore() + setSessionPersistenceDisabled(savedSessionPersistenceDisabled) + rmSync(fixtureCwd, { recursive: true, force: true }) +}