mirror of
https://github.com/Gitlawb/openclaude.git
synced 2026-08-24 02:34:15 -05:00
fix(autocompact): retry circuit breaker after cooldown (#1375)
* fix(autocompact): retry circuit breaker after cooldown * fix: honor auto-compact cooldown fallback Derive the circuit breaker retry time from lastFailureAtMs plus the configured cooldown when nextRetryAtMs is missing or invalid. Also clear SDK auto-compact tracking after manual compact boundaries and cover both paths with regressions. * test: make auto-compact cooldown fixture portable * test: extend auto-compact cooldown fixture timeout
This commit is contained in:
chioarub
@@ -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 })
|
||||
+14
-1
@@ -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 (
|
||||
|
||||
+79
-19
@@ -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,
|
||||
|
||||
@@ -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<T, TReturn>(
|
||||
generator: AsyncGenerator<T, TReturn>,
|
||||
): 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<AutoCompactTrackingState | undefined> = []
|
||||
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)
|
||||
})
|
||||
+52
-8
@@ -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<ReturnType<typeof getSessionId>, AutoCompactTrackingState>());
|
||||
const getAutoCompactTrackingForSession = useCallback((sessionId: ReturnType<typeof getSessionId>) => autoCompactTrackingBySessionRef.current.get(sessionId), []);
|
||||
const setAutoCompactTrackingForSession = useCallback((sessionId: ReturnType<typeof getSessionId>, tracking: AutoCompactTrackingState | undefined) => {
|
||||
if (tracking) {
|
||||
autoCompactTrackingBySessionRef.current.set(sessionId, tracking);
|
||||
} else {
|
||||
autoCompactTrackingBySessionRef.current.delete(sessionId);
|
||||
}
|
||||
}, []);
|
||||
const setAutoCompactTrackingForSessionIfUnchanged = useCallback((sessionId: ReturnType<typeof getSessionId>, 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<boolean>, input?: string, effort?: EffortValue): Promise<void> => {
|
||||
// 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<typeof pending>) {
|
||||
// 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);
|
||||
}
|
||||
|
||||
@@ -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<typeof mock>
|
||||
trySessionMemoryCompaction?: ReturnType<typeof mock>
|
||||
}
|
||||
|
||||
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()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<unknown>): Promise<void> {
|
||||
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<string>(),
|
||||
builtInCommandNames: new Set<string>(),
|
||||
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 })
|
||||
}
|
||||
Reference in New Issue
Block a user