From 92776028d2539013c7c783435360f07c502a52c3 Mon Sep 17 00:00:00 2001 From: chioarub Date: Sun, 5 Jul 2026 21:40:00 +0300 Subject: [PATCH 1/3] fix(api): quiet expected side-task aborts --- src/query/toolFailureLoopGuard.test.ts | 19 ++ src/query/toolFailureLoopGuard.ts | 2 + .../AgentSummary/agentSummary.test.ts | 98 ++++++++++ src/services/AgentSummary/agentSummary.ts | 7 +- .../api/claude.abortClassification.test.ts | 62 ++++++- src/services/api/claude.ts | 39 ++++ src/services/api/withRetry.test.ts | 115 +++++++++++- src/services/api/withRetry.ts | 14 ++ .../extractMemories.abort.test.ts | 174 ++++++++++++++++++ .../extractMemories/extractMemories.ts | 29 ++- src/utils/abortReasons.test.ts | 32 ++++ src/utils/abortReasons.ts | 30 +++ src/utils/queryLifecycle.test.ts | 28 +++ src/utils/queryLifecycle.ts | 2 + 14 files changed, 645 insertions(+), 6 deletions(-) create mode 100644 src/services/AgentSummary/agentSummary.test.ts create mode 100644 src/services/extractMemories/extractMemories.abort.test.ts diff --git a/src/query/toolFailureLoopGuard.test.ts b/src/query/toolFailureLoopGuard.test.ts index 1833e04b07..aaa989f4c9 100644 --- a/src/query/toolFailureLoopGuard.test.ts +++ b/src/query/toolFailureLoopGuard.test.ts @@ -179,6 +179,8 @@ test('user aborts, user rejections, and streaming fallback discards are ignored' getMissingToolResultAbortMessage('hard-max-query-timeout'), getMissingToolResultAbortMessage('background'), getMissingToolResultAbortMessage('side-task-cancelled'), + getMissingToolResultAbortMessage('agent-summary-superseded'), + getMissingToolResultAbortMessage('memory-extraction-superseded'), getMissingToolResultAbortMessage('parent-ended'), getMissingToolResultAbortMessage('unknown-abort'), 'Streaming fallback - tool execution discarded', @@ -227,6 +229,23 @@ test('reason-aware synthetic aborts are ignored through wrappers and memory hint expect(decision.tripped).toBe(false) }) +test('expected side-task cancellation messages do not trip repeated failure guard', () => { + const state = createToolFailureLoopGuardState() + const message = getMissingToolResultAbortMessage( + 'memory-extraction-superseded', + ) + + update(state, [toolUse('a', 'Read')], [toolResult('a', message)], 2) + const decision = update( + state, + [toolUse('b', 'Read')], + [toolResult('b', message)], + 2, + ) + + expect(decision.tripped).toBe(false) +}) + test('tool timeout messages still count as real tool failures', () => { const state = createToolFailureLoopGuardState() const timeoutMessage = getMissingToolResultAbortMessage('tool-timeout') diff --git a/src/query/toolFailureLoopGuard.ts b/src/query/toolFailureLoopGuard.ts index 08e1940606..cc5c200b43 100644 --- a/src/query/toolFailureLoopGuard.ts +++ b/src/query/toolFailureLoopGuard.ts @@ -13,6 +13,8 @@ const SYNTHETIC_ABORT_TOOL_RESULT_PREFIXES = [ getMissingToolResultAbortMessage('hard-max-query-timeout'), getMissingToolResultAbortMessage('background'), getMissingToolResultAbortMessage('side-task-cancelled'), + getMissingToolResultAbortMessage('agent-summary-superseded'), + getMissingToolResultAbortMessage('memory-extraction-superseded'), getMissingToolResultAbortMessage('parent-ended'), getMissingToolResultAbortMessage('unknown-abort'), ].map(message => message.toLowerCase()) diff --git a/src/services/AgentSummary/agentSummary.test.ts b/src/services/AgentSummary/agentSummary.test.ts new file mode 100644 index 0000000000..52ace26d90 --- /dev/null +++ b/src/services/AgentSummary/agentSummary.test.ts @@ -0,0 +1,98 @@ +import { afterEach, describe, expect, mock, test, vi } from 'bun:test' +import * as runAgentNs from '../../tools/AgentTool/runAgent.js' +import * as forkedAgentNs from '../../utils/forkedAgent.js' +import * as sessionStorageNs from '../../utils/sessionStorage.js' +import * as localAgentTaskNs from '../../tasks/LocalAgentTask/LocalAgentTask.js' +import * as debugNs from '../../utils/debug.js' + +const realRunAgent = { ...runAgentNs } +const realForkedAgent = { ...forkedAgentNs } +const realSessionStorage = { ...sessionStorageNs } +const realLocalAgentTask = { ...localAgentTaskNs } +const realDebug = { ...debugNs } + +describe('agent summary cancellation', () => { + afterEach(() => { + vi.useRealTimers() + vi.restoreAllMocks() + mock.restore() + mock.module('../../tools/AgentTool/runAgent.js', () => realRunAgent) + mock.module('../../utils/forkedAgent.js', () => realForkedAgent) + mock.module('../../utils/sessionStorage.js', () => realSessionStorage) + mock.module( + '../../tasks/LocalAgentTask/LocalAgentTask.js', + () => realLocalAgentTask, + ) + mock.module('../../utils/debug.js', () => realDebug) + }) + + test('stopping an in-flight summary aborts with an expected side-task reason', async () => { + vi.useFakeTimers() + + let summaryAbortController: AbortController | undefined + const debugLog = mock( + (_message: string, _options?: { level?: string }) => {}, + ) + const runForkedAgent = mock( + ({ overrides }: { overrides?: { abortController?: AbortController } }) => { + summaryAbortController = overrides?.abortController + return new Promise(() => {}) + }, + ) + + mock.module('../../tools/AgentTool/runAgent.js', () => ({ + filterIncompleteToolCalls: (messages: T) => messages, + runAgent: async function* () {}, + })) + mock.module('../../utils/forkedAgent.js', () => ({ + ...realForkedAgent, + runForkedAgent, + })) + mock.module('../../utils/sessionStorage.js', () => ({ + ...realSessionStorage, + getAgentTranscript: mock(async () => ({ + messages: [{ type: 'user' }, { type: 'assistant' }, { type: 'user' }], + })), + })) + mock.module('../../tasks/LocalAgentTask/LocalAgentTask.js', () => ({ + updateAgentSummary: mock(() => {}), + })) + mock.module('../../utils/debug.js', () => ({ + ...realDebug, + logForDebugging: debugLog, + })) + + const { startAgentSummarization } = await import( + `./agentSummary.js?ts=${Date.now()}-${Math.random()}` + ) + const { stop } = startAgentSummarization( + 'task-1', + 'agent-1' as never, + {} as never, + mock(() => {}) as never, + ) + + vi.advanceTimersByTime(30_000) + await Promise.resolve() + await Promise.resolve() + + expect(runForkedAgent).toHaveBeenCalledTimes(1) + expect(summaryAbortController).toBeDefined() + + stop() + + expect(summaryAbortController!.signal.aborted).toBe(true) + expect(summaryAbortController!.signal.reason).toBe( + 'agent-summary-superseded', + ) + expect( + debugLog.mock.calls.some(([message, options]) => { + return ( + String(message).includes('[AgentSummary] Stopping summarization') && + String(message).includes('agent-summary-superseded') && + (options as { level?: string } | undefined)?.level !== 'error' + ) + }), + ).toBe(true) + }) +}) diff --git a/src/services/AgentSummary/agentSummary.ts b/src/services/AgentSummary/agentSummary.ts index a44ad5d181..8cdea93dbd 100644 --- a/src/services/AgentSummary/agentSummary.ts +++ b/src/services/AgentSummary/agentSummary.ts @@ -24,6 +24,7 @@ import { createUserMessage } from '../../utils/messages.js' import { getAgentTranscript } from '../../utils/sessionStorage.js' const SUMMARY_INTERVAL_MS = 30_000 +const SUMMARY_SUPERSEDED_ABORT_REASON = 'agent-summary-superseded' function buildSummaryPrompt(previousSummary: string | null): string { const prevLine = previousSummary @@ -160,14 +161,16 @@ export function startAgentSummarization( } function stop(): void { - logForDebugging(`[AgentSummary] Stopping summarization for ${taskId}`) + logForDebugging( + `[AgentSummary] Stopping summarization for ${taskId}: ${SUMMARY_SUPERSEDED_ABORT_REASON}`, + ) stopped = true if (timeoutId) { clearTimeout(timeoutId) timeoutId = null } if (summaryAbortController) { - summaryAbortController.abort() + summaryAbortController.abort(SUMMARY_SUPERSEDED_ABORT_REASON) summaryAbortController = null } } diff --git a/src/services/api/claude.abortClassification.test.ts b/src/services/api/claude.abortClassification.test.ts index 8e7429f61a..35439bbf32 100644 --- a/src/services/api/claude.abortClassification.test.ts +++ b/src/services/api/claude.abortClassification.test.ts @@ -1,6 +1,11 @@ import { describe, expect, test } from 'bun:test' +import { APIUserAbortError } from '@anthropic-ai/sdk' import { shouldCreateUserInterruptionMessage } from '../../utils/abortReasons.js' -import { getClaudeStreamingAbortLogMessage } from './claude.js' +import { + getClaudeExpectedSideTaskApiAbortLogMessage, + getClaudeStreamingAbortLogMessage, +} from './claude.js' +import { CannotRetryError } from './withRetry.js' describe('Claude stream abort classification wiring', () => { test('formats timeout abort logs as timeout aborts, not user aborts', () => { @@ -39,6 +44,28 @@ describe('Claude stream abort classification wiring', () => { expect(shouldCreateUserInterruptionMessage('streaming_fallback')).toBe( false, ) + expect( + getClaudeStreamingAbortLogMessage( + { reason: 'agent-summary-superseded' }, + new Error('Request was aborted.'), + ), + ).toBe( + 'Streaming aborted because agent summary was superseded: Request was aborted.', + ) + expect( + getClaudeStreamingAbortLogMessage( + { reason: 'memory-extraction-superseded' }, + new Error('Request was aborted.'), + ), + ).toBe( + 'Streaming aborted because memory extraction was superseded: Request was aborted.', + ) + expect(shouldCreateUserInterruptionMessage('agent-summary-superseded')).toBe( + false, + ) + expect( + shouldCreateUserInterruptionMessage('memory-extraction-superseded'), + ).toBe(false) }) test('keeps actual user aborts user-facing for stream logs and advisor gating', () => { @@ -55,4 +82,37 @@ describe('Claude stream abort classification wiring', () => { true, ) }) + + test('labels expected side-task retry aborts without reclassifying user aborts', () => { + const expectedAbort = new AbortController() + expectedAbort.abort('agent-summary-superseded') + const retryAbort = new CannotRetryError(new APIUserAbortError(), { + model: 'test-model', + thinkingConfig: { type: 'disabled' }, + }) + + expect( + getClaudeExpectedSideTaskApiAbortLogMessage( + expectedAbort.signal, + retryAbort, + ), + ).toBe( + 'Expected side-task API abort (agent-summary-superseded): Request was aborted.', + ) + + const userAbort = new AbortController() + userAbort.abort() + expect( + getClaudeExpectedSideTaskApiAbortLogMessage( + userAbort.signal, + new APIUserAbortError(), + ), + ).toBeNull() + expect( + getClaudeExpectedSideTaskApiAbortLogMessage( + expectedAbort.signal, + new Error('boom'), + ), + ).toBeNull() + }) }) diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index dbfe052902..80ab8e16ca 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -59,6 +59,8 @@ import { import { getOauthAccountInfo } from '../../utils/auth.js' import { getStreamingAbortMessage, + isExpectedSideTaskAbortReason, + normalizeAbortReason, shouldCreateUserInterruptionMessage, } from '../../utils/abortReasons.js' import { @@ -812,6 +814,23 @@ export function getClaudeStreamingAbortLogMessage( return getStreamingAbortMessage(signal.reason, errorMessage(streamingError)) } +export function getClaudeExpectedSideTaskApiAbortLogMessage( + signal: Pick, + errorFromRetry: unknown, +): string | null { + if (!signal.aborted || !isExpectedSideTaskAbortReason(signal.reason)) { + return null + } + const error = + errorFromRetry instanceof CannotRetryError + ? errorFromRetry.originalError + : errorFromRetry + if (!(error instanceof APIUserAbortError)) { + return null + } + return `Expected side-task API abort (${normalizeAbortReason(signal.reason)}): ${errorMessage(error)}` +} + /** * Determines if an LSP tool should be deferred (tool appears with defer_loading: true) * because LSP initialization is not yet complete. @@ -2815,6 +2834,16 @@ async function* queryModel( throw errorFromRetry } + const expectedSideTaskAbortLogMessage = + getClaudeExpectedSideTaskApiAbortLogMessage(signal, errorFromRetry) + if (expectedSideTaskAbortLogMessage) { + if (!(errorFromRetry instanceof CannotRetryError)) { + logForDebugging(expectedSideTaskAbortLogMessage) + } + releaseStreamResources() + return + } + if (signal.aborted) { releaseStreamResources() return @@ -2917,6 +2946,16 @@ async function* queryModel( throw fallbackError } + const expectedSideTaskAbortLogMessage = + getClaudeExpectedSideTaskApiAbortLogMessage(signal, fallbackError) + if (expectedSideTaskAbortLogMessage) { + if (!(fallbackError instanceof CannotRetryError)) { + logForDebugging(expectedSideTaskAbortLogMessage) + } + releaseStreamResources() + return + } + // Fallback also failed, handle as normal error logForDebugging( `Non-streaming fallback also failed: ${errorMessage(fallbackError)}`, diff --git a/src/services/api/withRetry.test.ts b/src/services/api/withRetry.test.ts index 63bc273d8a..9710749d42 100644 --- a/src/services/api/withRetry.test.ts +++ b/src/services/api/withRetry.test.ts @@ -1,7 +1,8 @@ import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' import type Anthropic from '@anthropic-ai/sdk' -import { APIError } from '@anthropic-ai/sdk' +import { APIError, APIUserAbortError } from '@anthropic-ai/sdk' import { acquireSharedMutationLock, releaseSharedMutationLock } from '../../test/sharedMutationLock.js' +import * as debugNs from '../../utils/debug.js' type ProvidersModule = typeof import('../../utils/model/providers.js') // Helper to build a mock APIError with specific headers @@ -17,6 +18,7 @@ function makeError(headers: Record): APIError { // Save/restore env vars between tests const originalEnv = { ...process.env } +const originalDebugModule = { ...debugNs } let originalProvidersModule: ProvidersModule | undefined const envKeys = [ @@ -52,6 +54,7 @@ afterEach(() => { if (originalProvidersModule) { mock.module('src/utils/model/providers.js', () => originalProvidersModule!) } + mock.module('src/utils/debug.js', () => originalDebugModule) } finally { releaseSharedMutationLock() } @@ -73,12 +76,21 @@ async function importFreshWithRetryModule( | 'gemini' | 'codex' | 'foundry' = 'firstParty', + options?: { + logForDebugging?: ReturnType + }, ) { mock.restore() originalProvidersModule ??= await importActualProviders() mock.module('src/utils/sleep.js', () => ({ sleep: async () => undefined, })) + if (options?.logForDebugging) { + mock.module('src/utils/debug.js', () => ({ + ...originalDebugModule, + logForDebugging: options.logForDebugging!, + })) + } mock.module('src/utils/model/providers.js', () => ({ ...originalProvidersModule!, getAPIProvider: () => provider, @@ -177,6 +189,107 @@ describe('retry configuration', () => { }) }) +describe('abort retry classification', () => { + test('does not retry or error-log expected side task aborts', async () => { + const debugLog = mock( + (_message: string, _options?: { level?: string }) => {}, + ) + const { CannotRetryError, withRetry } = await importFreshWithRetryModule( + 'firstParty', + { logForDebugging: debugLog }, + ) + const controller = new AbortController() + let attempts = 0 + + await expect( + drainAsyncGenerator( + withRetry( + async () => ({} as Anthropic), + async () => { + attempts++ + controller.abort('agent-summary-superseded') + throw new APIUserAbortError() + }, + { + maxRetries: 2, + model: 'test-model', + thinkingConfig: { type: 'disabled' }, + signal: controller.signal, + querySource: 'agent_summary', + }, + ), + ), + ).rejects.toBeInstanceOf(CannotRetryError) + + expect(attempts).toBe(1) + expect( + debugLog.mock.calls.some(([message, options]) => { + return ( + String(message).startsWith('API error (attempt') && + (options as { level?: string } | undefined)?.level === 'error' + ) + }), + ).toBe(false) + expect( + debugLog.mock.calls.some(([message, options]) => { + return ( + String(message).includes('Expected side-task API abort') && + String(message).includes('agent-summary-superseded') && + (options as { level?: string } | undefined)?.level !== 'error' + ) + }), + ).toBe(true) + }) + + test('still logs and retries real retryable API errors', async () => { + process.env.OPENCLAUDE_RETRY_DELAY_MS = '1' + const debugLog = mock( + (_message: string, _options?: { level?: string }) => {}, + ) + const { withRetry } = await importFreshWithRetryModule('firstParty', { + logForDebugging: debugLog, + }) + const retryableError = APIError.generate( + 500, + undefined, + 'internal server error', + new Headers(), + ) + let attempts = 0 + + const result = await drainAsyncGenerator( + withRetry( + async () => ({} as Anthropic), + async () => { + attempts++ + if (attempts === 1) { + throw retryableError + } + return { ok: true } + }, + { + maxRetries: 2, + model: 'test-model', + thinkingConfig: { type: 'disabled' }, + querySource: 'repl_main_thread', + }, + ), + ) + + expect(result).toEqual({ ok: true }) + expect(attempts).toBe(2) + expect( + debugLog.mock.calls.some(([message, options]) => { + return ( + String(message).startsWith('API error (attempt 1/3)') && + String(message).includes('500 internal server error') && + (options as { level?: string } | undefined)?.level === 'error' + ) + }), + ).toBe(true) + }) +}) + describe('OpenAI-compatible retry classification', () => { test('does not retry marked non-retryable auth failures', async () => { process.env.OPENCLAUDE_RETRY_DELAY_MS = '1' diff --git a/src/services/api/withRetry.ts b/src/services/api/withRetry.ts index b11b30a3c8..80e5c45a40 100644 --- a/src/services/api/withRetry.ts +++ b/src/services/api/withRetry.ts @@ -7,6 +7,10 @@ import { } from '@anthropic-ai/sdk' import type { QuerySource } from 'src/constants/querySource.js' import type { SystemAPIErrorMessage } from 'src/types/message.js' +import { + isExpectedSideTaskAbortReason, + normalizeAbortReason, +} from 'src/utils/abortReasons.js' import { isAwsCredentialsProviderError } from 'src/utils/aws.js' import { logForDebugging } from 'src/utils/debug.js' import { logError } from 'src/utils/log.js' @@ -277,6 +281,16 @@ export async function* withRetry( return await operation(client, attempt, retryContext) } catch (error) { lastError = error + if ( + error instanceof APIUserAbortError && + options.signal?.aborted && + isExpectedSideTaskAbortReason(options.signal.reason) + ) { + logForDebugging( + `Expected side-task API abort (${normalizeAbortReason(options.signal.reason)}): ${errorMessage(error)}`, + ) + throw new CannotRetryError(error, retryContext) + } logForDebugging( `API error (attempt ${attempt}/${maxRetries + 1}): ${error instanceof APIError ? `${error.status} ${error.message}` : errorMessage(error)}`, { level: 'error' }, diff --git a/src/services/extractMemories/extractMemories.abort.test.ts b/src/services/extractMemories/extractMemories.abort.test.ts new file mode 100644 index 0000000000..ca72f0bc95 --- /dev/null +++ b/src/services/extractMemories/extractMemories.abort.test.ts @@ -0,0 +1,174 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' +import { APIUserAbortError } from '@anthropic-ai/sdk' +import { + acquireSharedMutationLock, + releaseSharedMutationLock, +} from '../../test/sharedMutationLock.js' +import * as bootstrapStateNs from '../../bootstrap/state.js' +import * as memoryScanNs from '../../memdir/memoryScan.js' +import * as pathsNs from '../../memdir/paths.js' +import * as governanceNs from '../../utils/governancePolicy.js' +import * as debugNs from '../../utils/debug.js' +import * as forkedAgentNs from '../../utils/forkedAgent.js' +import * as growthbookNs from '../analytics/growthbook.js' +import * as analyticsNs from '../analytics/index.js' + +const realBootstrapState = { ...bootstrapStateNs } +const realMemoryScan = { ...memoryScanNs } +const realPaths = { ...pathsNs } +const realGovernance = { ...governanceNs } +const realDebug = { ...debugNs } +const realForkedAgent = { ...forkedAgentNs } +const realGrowthbook = { ...growthbookNs } +const realAnalytics = { ...analyticsNs } + +type ForkCall = { + abortController?: AbortController +} + +function createContext(messageUuid: string): any { + return { + messages: [ + { + type: 'user', + uuid: messageUuid, + timestamp: '2026-07-05T00:00:00.000Z', + message: { role: 'user', content: 'remember this' }, + }, + ], + systemPrompt: [], + userContext: {}, + systemContext: {}, + toolUseContext: {}, + } +} + +describe('extractMemories cancellation', () => { + beforeEach(async () => { + await acquireSharedMutationLock('extractMemories.abort.test.ts') + }) + + afterEach(() => { + try { + mock.restore() + mock.module('../../bootstrap/state.js', () => realBootstrapState) + mock.module('../../memdir/memoryScan.js', () => realMemoryScan) + mock.module('../../memdir/paths.js', () => realPaths) + mock.module('../../utils/governancePolicy.js', () => realGovernance) + mock.module('../../utils/debug.js', () => realDebug) + mock.module('../../utils/forkedAgent.js', () => realForkedAgent) + mock.module('../analytics/growthbook.js', () => realGrowthbook) + mock.module('../analytics/index.js', () => realAnalytics) + } finally { + releaseSharedMutationLock() + } + }) + + test('superseding an active extraction aborts the stale fork with an expected reason', async () => { + const debugLog = mock((_message: string) => {}) + const events: string[] = [] + const forkCalls: ForkCall[] = [] + const runForkedAgent = mock(({ overrides }: any) => { + const abortController = overrides?.abortController as + | AbortController + | undefined + forkCalls.push({ abortController }) + + if (forkCalls.length > 1) { + return Promise.resolve({ + messages: [], + totalUsage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }) + } + + return new Promise((_resolve, reject) => { + abortController?.signal.addEventListener( + 'abort', + () => reject(new APIUserAbortError()), + { once: true }, + ) + }) + }) + + mock.module('../../bootstrap/state.js', () => ({ + ...realBootstrapState, + getIsRemoteMode: () => false, + })) + mock.module('../../memdir/memoryScan.js', () => ({ + ...realMemoryScan, + formatMemoryManifest: () => 'No memories yet.', + scanMemoryFiles: async () => [], + })) + mock.module('../../memdir/paths.js', () => ({ + ...realPaths, + getAutoMemPath: () => '/tmp/memory-extraction-test', + isAutoMemoryEnabled: () => true, + isAutoMemPath: () => false, + })) + mock.module('../../utils/governancePolicy.js', () => ({ + ...realGovernance, + isMemoryWriteApprovalRequired: () => false, + })) + mock.module('../../utils/debug.js', () => ({ + ...realDebug, + logForDebugging: debugLog, + })) + mock.module('../../utils/forkedAgent.js', () => ({ + ...realForkedAgent, + createCacheSafeParams: (context: any) => ({ + systemPrompt: context.systemPrompt, + userContext: context.userContext, + systemContext: context.systemContext, + toolUseContext: context.toolUseContext, + forkContextMessages: context.messages, + }), + runForkedAgent, + })) + mock.module('../analytics/growthbook.js', () => ({ + ...realGrowthbook, + getFeatureValue_CACHED_MAY_BE_STALE: ( + key: string, + defaultValue: unknown, + ) => (key === 'tengu_passport_quail' ? true : defaultValue), + })) + mock.module('../analytics/index.js', () => ({ + ...realAnalytics, + logEvent: (name: string) => { + events.push(name) + }, + })) + + const { executeExtractMemories, initExtractMemories } = await import( + `./extractMemories.js?ts=${Date.now()}-${Math.random()}` + ) + initExtractMemories() + + const firstExtraction = executeExtractMemories(createContext('m1')) + await Promise.resolve() + await Promise.resolve() + + await executeExtractMemories(createContext('m2')) + + expect(runForkedAgent.mock.calls.length).toBeGreaterThanOrEqual(1) + expect(forkCalls[0]?.abortController).toBeDefined() + expect(forkCalls[0]!.abortController!.signal.aborted).toBe(true) + expect(forkCalls[0]!.abortController!.signal.reason).toBe( + 'memory-extraction-superseded', + ) + + await firstExtraction + + expect(events).toContain('tengu_extract_memories_coalesced') + expect(events).not.toContain('tengu_extract_memories_error') + expect( + debugLog.mock.calls.some(([message]) => + String(message).includes('[extractMemories] expected cancellation'), + ), + ).toBe(true) + }) +}) diff --git a/src/services/extractMemories/extractMemories.ts b/src/services/extractMemories/extractMemories.ts index b878d32780..5c38e010bd 100644 --- a/src/services/extractMemories/extractMemories.ts +++ b/src/services/extractMemories/extractMemories.ts @@ -41,7 +41,7 @@ import type { SystemLocalCommandMessage, SystemMessage, } from '../../types/message.js' -import { createAbortController } from '../../utils/abortController.js' +import { isExpectedSideTaskAbortReason } from '../../utils/abortReasons.js' import { count, uniq } from '../../utils/array.js' import { isMemoryWriteApprovalRequired } from '../../utils/governancePolicy.js' import { logForDebugging } from '../../utils/debug.js' @@ -62,6 +62,9 @@ import { buildExtractCombinedPrompt, } from './prompts.js' +const MEMORY_EXTRACTION_SUPERSEDED_ABORT_REASON = + 'memory-extraction-superseded' + /* eslint-disable @typescript-eslint/no-require-imports */ const teamMemPaths = feature('TEAMMEM') ? (require('../../memdir/teamMemPaths.js') as typeof import('../../memdir/teamMemPaths.js')) @@ -313,6 +316,9 @@ export function initExtractMemories(): void { /** True while runExtraction is executing — prevents overlapping runs. */ let inProgress = false + /** Abort controller for the currently running forked memory extraction. */ + let activeExtractionAbortController: AbortController | undefined + /** Counts eligible turns since the last extraction run. Resets to 0 after each run. */ let turnsSinceLastExtraction = 0 @@ -388,6 +394,9 @@ export function initExtractMemories(): void { inProgress = true const startTime = Date.now() + const extractionAbortController = new AbortController() + activeExtractionAbortController = extractionAbortController + try { logForDebugging( `[extractMemories] starting — ${newMessageCount} new messages, memoryDir=${memoryDir}`, @@ -397,7 +406,7 @@ export function initExtractMemories(): void { // a turn on `ls`. Reuses findRelevantMemories' frontmatter scan. // Placed after the throttle gate so skipped turns don't pay the scan cost. const existingMemories = formatMemoryManifest( - await scanMemoryFiles(memoryDir, createAbortController().signal), + await scanMemoryFiles(memoryDir, extractionAbortController.signal), ) const userPrompt = @@ -425,6 +434,7 @@ export function initExtractMemories(): void { // Well-behaved extractions complete in 2-4 turns (read → write). // A hard cap prevents verification rabbit-holes from burning turns. maxTurns: 5, + overrides: { abortController: extractionAbortController }, }) // Advance the cursor only after a successful run. If the agent errors @@ -496,12 +506,24 @@ export function initExtractMemories(): void { appendSystemMessage?.(msg) } } catch (error) { + if ( + extractionAbortController.signal.aborted && + isExpectedSideTaskAbortReason(extractionAbortController.signal.reason) + ) { + logForDebugging( + `[extractMemories] expected cancellation: ${extractionAbortController.signal.reason}`, + ) + return + } // Extraction is best-effort — log but don't notify on error logForDebugging(`[extractMemories] error: ${error}`) logEvent('tengu_extract_memories_error', { duration_ms: Date.now() - startTime, }) } finally { + if (activeExtractionAbortController === extractionAbortController) { + activeExtractionAbortController = undefined + } inProgress = false // If a call arrived while we were running, run a trailing extraction @@ -565,6 +587,9 @@ export function initExtractMemories(): void { ) logEvent('tengu_extract_memories_coalesced', {}) pendingContext = { context, appendSystemMessage } + activeExtractionAbortController?.abort( + MEMORY_EXTRACTION_SUPERSEDED_ABORT_REASON, + ) return } diff --git a/src/utils/abortReasons.test.ts b/src/utils/abortReasons.test.ts index a088c09a74..4400e23165 100644 --- a/src/utils/abortReasons.test.ts +++ b/src/utils/abortReasons.test.ts @@ -20,6 +20,12 @@ describe('abort reason normalization', () => { expect(normalizeAbortReason('interrupt')).toBe('interrupt') expect(normalizeAbortReason('background')).toBe('background') expect(normalizeAbortReason('parent-ended')).toBe('parent-ended') + expect(normalizeAbortReason('agent-summary-superseded')).toBe( + 'agent-summary-superseded', + ) + expect(normalizeAbortReason('memory-extraction-superseded')).toBe( + 'memory-extraction-superseded', + ) expect(normalizeAbortReason('hard_max')).toBe('hard-max-query-timeout') expect(normalizeAbortReason('streaming_fallback')).toBe( 'side-task-cancelled', @@ -90,6 +96,22 @@ describe('abort reason normalization', () => { ).toBe( 'Streaming aborted because side task was cancelled: Request was aborted.', ) + expect( + getStreamingAbortMessage( + 'agent-summary-superseded', + 'Request was aborted.', + ), + ).toBe( + 'Streaming aborted because agent summary was superseded: Request was aborted.', + ) + expect( + getStreamingAbortMessage( + 'memory-extraction-superseded', + 'Request was aborted.', + ), + ).toBe( + 'Streaming aborted because memory extraction was superseded: Request was aborted.', + ) }) test('only actual user cancellation creates user interruption transcript text', () => { @@ -106,6 +128,12 @@ describe('abort reason normalization', () => { expect(shouldCreateUserInterruptionMessage('query-timeout')).toBe(false) expect(shouldCreateUserInterruptionMessage('hard_max')).toBe(false) expect(shouldCreateUserInterruptionMessage('background')).toBe(false) + expect(shouldCreateUserInterruptionMessage('agent-summary-superseded')).toBe( + false, + ) + expect( + shouldCreateUserInterruptionMessage('memory-extraction-superseded'), + ).toBe(false) expect(shouldCreateUserInterruptionMessage('streaming_fallback')).toBe( false, ) @@ -121,6 +149,8 @@ describe('abort reason normalization', () => { expect(getQueryAbortSystemMessage('streaming_fallback')).toBe( 'Query stopped because a side task was cancelled.', ) + expect(getQueryAbortSystemMessage('agent-summary-superseded')).toBeNull() + expect(getQueryAbortSystemMessage('memory-extraction-superseded')).toBeNull() expect(getQueryAbortSystemMessage('parent-ended')).toBe( 'Query stopped because the parent query ended.', ) @@ -153,6 +183,8 @@ describe('abort reason normalization', () => { interrupt: true, background: true, 'side-task-cancelled': true, + 'agent-summary-superseded': true, + 'memory-extraction-superseded': true, 'tool-timeout': true, 'parent-ended': true, 'unknown-abort': true, diff --git a/src/utils/abortReasons.ts b/src/utils/abortReasons.ts index ad21acc3ce..2764b8262a 100644 --- a/src/utils/abortReasons.ts +++ b/src/utils/abortReasons.ts @@ -5,6 +5,8 @@ export type AbortReason = | 'interrupt' | 'background' | 'side-task-cancelled' + | 'agent-summary-superseded' + | 'memory-extraction-superseded' | 'tool-timeout' | 'parent-ended' | 'unknown-abort' @@ -16,6 +18,8 @@ const KNOWN_ABORT_REASONS = new Set([ 'interrupt', 'background', 'side-task-cancelled', + 'agent-summary-superseded', + 'memory-extraction-superseded', 'tool-timeout', 'parent-ended', 'unknown-abort', @@ -59,6 +63,17 @@ export function isQueryLevelAbort(reason: AbortReason): boolean { return reason !== 'tool-timeout' } +export function isExpectedSideTaskAbortReason(reason: unknown): boolean { + switch (normalizeAbortReason(reason)) { + case 'side-task-cancelled': + case 'agent-summary-superseded': + case 'memory-extraction-superseded': + return true + default: + return false + } +} + export function getShellAbortMessage(reason: AbortReason): string { switch (reason) { case 'query-timeout': @@ -69,6 +84,10 @@ export function getShellAbortMessage(reason: AbortReason): string { return 'Command was interrupted because the enclosing query was backgrounded.' case 'side-task-cancelled': return 'Command was interrupted because a side task was cancelled.' + case 'agent-summary-superseded': + return 'Command was interrupted because an agent summary was superseded.' + case 'memory-extraction-superseded': + return 'Command was interrupted because memory extraction was superseded.' case 'tool-timeout': return 'Command timed out before completion.' case 'user-abort': @@ -99,6 +118,10 @@ export function getStreamingAbortMessage( return `Streaming aborted for backgrounding: ${errorText}` case 'side-task-cancelled': return `Streaming aborted because side task was cancelled: ${errorText}` + case 'agent-summary-superseded': + return `Streaming aborted because agent summary was superseded: ${errorText}` + case 'memory-extraction-superseded': + return `Streaming aborted because memory extraction was superseded: ${errorText}` case 'tool-timeout': return `Streaming aborted because tool timed out: ${errorText}` case 'parent-ended': @@ -122,6 +145,9 @@ export function getQueryAbortSystemMessage(reason: unknown): string | null { return 'Query was backgrounded before completion.' case 'side-task-cancelled': return 'Query stopped because a side task was cancelled.' + case 'agent-summary-superseded': + case 'memory-extraction-superseded': + return null case 'parent-ended': return 'Query stopped because the parent query ended.' default: @@ -143,6 +169,10 @@ export function getMissingToolResultAbortMessage(reason: unknown): string { return 'Tool use was interrupted because the query was backgrounded.' case 'side-task-cancelled': return 'Tool use was interrupted because a side task was cancelled.' + case 'agent-summary-superseded': + return 'Tool use was interrupted because an agent summary was superseded.' + case 'memory-extraction-superseded': + return 'Tool use was interrupted because memory extraction was superseded.' case 'tool-timeout': return 'Tool use timed out before completion.' case 'parent-ended': diff --git a/src/utils/queryLifecycle.test.ts b/src/utils/queryLifecycle.test.ts index 8c70eccdb8..15fb41414d 100644 --- a/src/utils/queryLifecycle.test.ts +++ b/src/utils/queryLifecycle.test.ts @@ -28,6 +28,22 @@ describe('query lifecycle log formatting', () => { expect(line).not.toContain('abortReason=query-timeout') expect(line.match(/\babortReason=/g)).toHaveLength(1) }) + + test('keeps exact expected side-task abort reason on grouped lifecycle end log', () => { + const context: QueryLifecycleContext = { + queryId: 'query-1', + queryGeneration: 1, + querySource: 'forked_agent', + startedAt: 1, + terminalReason: 'parent-ended', + abortReason: 'memory-extraction-superseded', + } + + const line = formatQueryLifecycleLogMessage('end', context) + + expect(line).toContain('terminalReason=parent-ended') + expect(line).toContain('abortReason=memory-extraction-superseded') + }) }) describe('query terminal reason classification', () => { @@ -92,5 +108,17 @@ describe('query terminal reason classification', () => { false, ), ).toBe('parent-ended') + expect( + getQueryTerminalReason( + { aborted: true, reason: 'agent-summary-superseded' }, + false, + ), + ).toBe('parent-ended') + expect( + getQueryTerminalReason( + { aborted: true, reason: 'memory-extraction-superseded' }, + false, + ), + ).toBe('parent-ended') }) }) diff --git a/src/utils/queryLifecycle.ts b/src/utils/queryLifecycle.ts index c6b573ac33..71953cb2a4 100644 --- a/src/utils/queryLifecycle.ts +++ b/src/utils/queryLifecycle.ts @@ -84,6 +84,8 @@ export function getQueryTerminalReason( case 'background': case 'parent-ended': case 'side-task-cancelled': + case 'agent-summary-superseded': + case 'memory-extraction-superseded': return 'parent-ended' default: return 'unknown' From c05ada7ba0cc056ee4a872c566fc466d35bc941f Mon Sep 17 00:00:00 2001 From: chioarub Date: Sun, 5 Jul 2026 21:54:18 +0300 Subject: [PATCH 2/3] fix(api): tighten expected abort handling --- src/services/api/claude.ts | 43 +++++--- .../extractMemories.abort.test.ts | 102 ++++++++++++++++++ .../extractMemories/extractMemories.ts | 6 +- 3 files changed, 136 insertions(+), 15 deletions(-) diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index 80ab8e16ca..32faf6ab93 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -831,6 +831,21 @@ export function getClaudeExpectedSideTaskApiAbortLogMessage( return `Expected side-task API abort (${normalizeAbortReason(signal.reason)}): ${errorMessage(error)}` } +function handleClaudeExpectedSideTaskApiAbort( + signal: Pick, + errorFromRetry: unknown, + releaseStreamResources: () => void, +): boolean { + const expectedSideTaskAbortLogMessage = + getClaudeExpectedSideTaskApiAbortLogMessage(signal, errorFromRetry) + if (!expectedSideTaskAbortLogMessage) return false + if (!(errorFromRetry instanceof CannotRetryError)) { + logForDebugging(expectedSideTaskAbortLogMessage) + } + releaseStreamResources() + return true +} + /** * Determines if an LSP tool should be deferred (tool appears with defer_loading: true) * because LSP initialization is not yet complete. @@ -2834,13 +2849,13 @@ async function* queryModel( throw errorFromRetry } - const expectedSideTaskAbortLogMessage = - getClaudeExpectedSideTaskApiAbortLogMessage(signal, errorFromRetry) - if (expectedSideTaskAbortLogMessage) { - if (!(errorFromRetry instanceof CannotRetryError)) { - logForDebugging(expectedSideTaskAbortLogMessage) - } - releaseStreamResources() + if ( + handleClaudeExpectedSideTaskApiAbort( + signal, + errorFromRetry, + releaseStreamResources, + ) + ) { return } @@ -2946,13 +2961,13 @@ async function* queryModel( throw fallbackError } - const expectedSideTaskAbortLogMessage = - getClaudeExpectedSideTaskApiAbortLogMessage(signal, fallbackError) - if (expectedSideTaskAbortLogMessage) { - if (!(fallbackError instanceof CannotRetryError)) { - logForDebugging(expectedSideTaskAbortLogMessage) - } - releaseStreamResources() + if ( + handleClaudeExpectedSideTaskApiAbort( + signal, + fallbackError, + releaseStreamResources, + ) + ) { return } diff --git a/src/services/extractMemories/extractMemories.abort.test.ts b/src/services/extractMemories/extractMemories.abort.test.ts index ca72f0bc95..46fb700134 100644 --- a/src/services/extractMemories/extractMemories.abort.test.ts +++ b/src/services/extractMemories/extractMemories.abort.test.ts @@ -171,4 +171,106 @@ describe('extractMemories cancellation', () => { ), ).toBe(true) }) + + test('superseded extraction still reports non-abort failures', async () => { + const debugLog = mock((_message: string) => {}) + const events: string[] = [] + const forkCalls: ForkCall[] = [] + const runForkedAgent = mock(({ overrides }: any) => { + const abortController = overrides?.abortController as + | AbortController + | undefined + forkCalls.push({ abortController }) + + if (forkCalls.length > 1) { + return Promise.resolve({ + messages: [], + totalUsage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }) + } + + return new Promise((_resolve, reject) => { + abortController?.signal.addEventListener( + 'abort', + () => reject(new Error('write failed after abort')), + { once: true }, + ) + }) + }) + + mock.module('../../bootstrap/state.js', () => ({ + ...realBootstrapState, + getIsRemoteMode: () => false, + })) + mock.module('../../memdir/memoryScan.js', () => ({ + ...realMemoryScan, + formatMemoryManifest: () => 'No memories yet.', + scanMemoryFiles: async () => [], + })) + mock.module('../../memdir/paths.js', () => ({ + ...realPaths, + getAutoMemPath: () => '/tmp/memory-extraction-test', + isAutoMemoryEnabled: () => true, + isAutoMemPath: () => false, + })) + mock.module('../../utils/governancePolicy.js', () => ({ + ...realGovernance, + isMemoryWriteApprovalRequired: () => false, + })) + mock.module('../../utils/debug.js', () => ({ + ...realDebug, + logForDebugging: debugLog, + })) + mock.module('../../utils/forkedAgent.js', () => ({ + ...realForkedAgent, + createCacheSafeParams: (context: any) => ({ + systemPrompt: context.systemPrompt, + userContext: context.userContext, + systemContext: context.systemContext, + toolUseContext: context.toolUseContext, + forkContextMessages: context.messages, + }), + runForkedAgent, + })) + mock.module('../analytics/growthbook.js', () => ({ + ...realGrowthbook, + getFeatureValue_CACHED_MAY_BE_STALE: ( + key: string, + defaultValue: unknown, + ) => (key === 'tengu_passport_quail' ? true : defaultValue), + })) + mock.module('../analytics/index.js', () => ({ + ...realAnalytics, + logEvent: (name: string) => { + events.push(name) + }, + })) + + const { executeExtractMemories, initExtractMemories } = await import( + `./extractMemories.js?ts=${Date.now()}-${Math.random()}` + ) + initExtractMemories() + + const firstExtraction = executeExtractMemories(createContext('m1')) + await Promise.resolve() + await Promise.resolve() + + await executeExtractMemories(createContext('m2')) + await firstExtraction + + expect(forkCalls[0]?.abortController?.signal.reason).toBe( + 'memory-extraction-superseded', + ) + expect(events).toContain('tengu_extract_memories_error') + expect( + debugLog.mock.calls.some(([message]) => + String(message).includes('[extractMemories] expected cancellation'), + ), + ).toBe(false) + }) }) diff --git a/src/services/extractMemories/extractMemories.ts b/src/services/extractMemories/extractMemories.ts index 5c38e010bd..2227180190 100644 --- a/src/services/extractMemories/extractMemories.ts +++ b/src/services/extractMemories/extractMemories.ts @@ -49,6 +49,7 @@ import { createCacheSafeParams, runForkedAgent, } from '../../utils/forkedAgent.js' +import { isAbortError } from '../../utils/errors.js' import type { REPLHookContext } from '../../utils/hooks/postSamplingHooks.js' import { createMemorySavedMessage, @@ -508,7 +509,10 @@ export function initExtractMemories(): void { } catch (error) { if ( extractionAbortController.signal.aborted && - isExpectedSideTaskAbortReason(extractionAbortController.signal.reason) + isExpectedSideTaskAbortReason( + extractionAbortController.signal.reason, + ) && + isAbortError(error) ) { logForDebugging( `[extractMemories] expected cancellation: ${extractionAbortController.signal.reason}`, From 8e1934c8c4cde13592646292b4a02177d6d87350 Mon Sep 17 00:00:00 2001 From: chioarub Date: Mon, 6 Jul 2026 02:36:35 +0300 Subject: [PATCH 3/3] fix(memory): keep cursor on superseded extraction --- .../extractMemories.abort.test.ts | 125 ++++++++++++++++-- .../extractMemories/extractMemories.ts | 19 +++ 2 files changed, 135 insertions(+), 9 deletions(-) diff --git a/src/services/extractMemories/extractMemories.abort.test.ts b/src/services/extractMemories/extractMemories.abort.test.ts index 46fb700134..86b38c435b 100644 --- a/src/services/extractMemories/extractMemories.abort.test.ts +++ b/src/services/extractMemories/extractMemories.abort.test.ts @@ -26,16 +26,15 @@ type ForkCall = { abortController?: AbortController } -function createContext(messageUuid: string): any { +function createContext(messageUuids: string | string[]): any { + const uuids = Array.isArray(messageUuids) ? messageUuids : [messageUuids] return { - messages: [ - { - type: 'user', - uuid: messageUuid, - timestamp: '2026-07-05T00:00:00.000Z', - message: { role: 'user', content: 'remember this' }, - }, - ], + messages: uuids.map(messageUuid => ({ + type: 'user', + uuid: messageUuid, + timestamp: '2026-07-05T00:00:00.000Z', + message: { role: 'user', content: 'remember this' }, + })), systemPrompt: [], userContext: {}, systemContext: {}, @@ -273,4 +272,112 @@ describe('extractMemories cancellation', () => { ), ).toBe(false) }) + + test('superseded extraction does not advance cursor when stale fork resolves after abort', async () => { + const debugLog = mock((_message: string) => {}) + const events: string[] = [] + const forkPrompts: string[] = [] + const forkCalls: ForkCall[] = [] + const emptyResult = { + messages: [], + totalUsage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + } + const runForkedAgent = mock(({ overrides, promptMessages }: any) => { + const abortController = overrides?.abortController as + | AbortController + | undefined + forkCalls.push({ abortController }) + forkPrompts.push(String(promptMessages[0]?.message.content ?? '')) + + if (forkCalls.length > 1) { + return Promise.resolve(emptyResult) + } + + return new Promise(resolve => { + abortController?.signal.addEventListener( + 'abort', + () => resolve(emptyResult), + { once: true }, + ) + }) + }) + + mock.module('../../bootstrap/state.js', () => ({ + ...realBootstrapState, + getIsRemoteMode: () => false, + })) + mock.module('../../memdir/memoryScan.js', () => ({ + ...realMemoryScan, + formatMemoryManifest: () => 'No memories yet.', + scanMemoryFiles: async () => [], + })) + mock.module('../../memdir/paths.js', () => ({ + ...realPaths, + getAutoMemPath: () => '/tmp/memory-extraction-test', + isAutoMemoryEnabled: () => true, + isAutoMemPath: () => false, + })) + mock.module('../../utils/governancePolicy.js', () => ({ + ...realGovernance, + isMemoryWriteApprovalRequired: () => false, + })) + mock.module('../../utils/debug.js', () => ({ + ...realDebug, + logForDebugging: debugLog, + })) + mock.module('../../utils/forkedAgent.js', () => ({ + ...realForkedAgent, + createCacheSafeParams: (context: any) => ({ + systemPrompt: context.systemPrompt, + userContext: context.userContext, + systemContext: context.systemContext, + toolUseContext: context.toolUseContext, + forkContextMessages: context.messages, + }), + runForkedAgent, + })) + mock.module('../analytics/growthbook.js', () => ({ + ...realGrowthbook, + getFeatureValue_CACHED_MAY_BE_STALE: ( + key: string, + defaultValue: unknown, + ) => (key === 'tengu_passport_quail' ? true : defaultValue), + })) + mock.module('../analytics/index.js', () => ({ + ...realAnalytics, + logEvent: (name: string) => { + events.push(name) + }, + })) + + const { executeExtractMemories, initExtractMemories } = await import( + `./extractMemories.js?ts=${Date.now()}-${Math.random()}` + ) + initExtractMemories() + + const firstExtraction = executeExtractMemories(createContext(['m1', 'm2'])) + await Promise.resolve() + await Promise.resolve() + + await executeExtractMemories(createContext(['m1', 'm2', 'm3'])) + await firstExtraction + + expect(runForkedAgent).toHaveBeenCalledTimes(2) + expect(forkCalls[0]?.abortController?.signal.reason).toBe( + 'memory-extraction-superseded', + ) + expect(forkPrompts[1]).toContain('most recent ~3 messages') + expect(events).toContain('tengu_extract_memories_coalesced') + expect(events).not.toContain('tengu_extract_memories_error') + expect( + debugLog.mock.calls.some(([message]) => + String(message).includes('[extractMemories] expected cancellation'), + ), + ).toBe(true) + }) }) diff --git a/src/services/extractMemories/extractMemories.ts b/src/services/extractMemories/extractMemories.ts index 2227180190..e556413e1a 100644 --- a/src/services/extractMemories/extractMemories.ts +++ b/src/services/extractMemories/extractMemories.ts @@ -438,6 +438,25 @@ export function initExtractMemories(): void { overrides: { abortController: extractionAbortController }, }) + if (extractionAbortController.signal.aborted) { + if ( + isExpectedSideTaskAbortReason(extractionAbortController.signal.reason) + ) { + logForDebugging( + `[extractMemories] expected cancellation: ${extractionAbortController.signal.reason}`, + ) + return + } + + logForDebugging( + `[extractMemories] error: aborted after fork returned: ${extractionAbortController.signal.reason}`, + ) + logEvent('tengu_extract_memories_error', { + duration_ms: Date.now() - startTime, + }) + return + } + // Advance the cursor only after a successful run. If the agent errors // out (caught below), the cursor stays put so those messages are // reconsidered on the next extraction.