mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 14:38:23 -10:00
fix(memory): preserve migration progress when index rebuild fails (#12913)
* fix(memory): preserve partial migration results * test(memory): cover index rebuild retry limit * fix(memory): migrate after tool-completing turns * fix(memory): preserve migration across continuations * fix(memory): log swallowed rebuild failure and carry failure_reason in telemetry Address three review findings on the migration failure path: - metadata-migration: log the swallowed index-rebuild failure to the AUTO_MEMORY_MIGRATION debug channel (matching the sibling catch at the post-commit re-entry site) instead of recording it only on the in-process task record, which dies with the session. - telemetry: add an optional failure_reason field to MemoryMigrationEvent and populate it (index rebuild error message, or 'stalled') so a failed event no longer sits next to success-looking counters with no cause. - manager.test: spy logMemoryMigration through the existing hoisted telemetry mock factory and assert the event agrees with the task record (status failed, committed counts intact, reason attached). Each new assertion is red under its corresponding revert. Co-authored-by: Qwen-Coder <[email protected]> * fix(memory): retain user interaction type across tool turns * test(memory): ignore background requests in tool E2E * fix(core): clear stale memory interaction type --------- Co-authored-by: Qwen-Coder <[email protected]>
This commit is contained in:
易良andQwen-Coder authored and GitHub committed
1 parent
25d6b702c4
commit
b1eb94daa8
10 files changed
+345
-46
No files matched your search
@@ -149,7 +149,7 @@ describe('headless extension workflow Skill entry', () => {
|
||||
const requests = fakeServer.requests.filter(
|
||||
({ body }) => body['stream'] === true,
|
||||
);
|
||||
expect(requests).toHaveLength(2);
|
||||
expect(requests.length).toBeGreaterThanOrEqual(2);
|
||||
|
||||
const firstMessages = requests[0].body['messages'] as Message[];
|
||||
const availableSkills = firstMessages
|
||||
@@ -160,8 +160,16 @@ describe('headless extension workflow Skill entry', () => {
|
||||
expect(availableSkills).toContain(WORKFLOW_NAME);
|
||||
expect(availableSkills).toContain(WHEN_TO_USE);
|
||||
|
||||
const secondMessages = requests[1].body['messages'] as Message[];
|
||||
const skillResult = secondMessages.find(
|
||||
const secondMessages = requests
|
||||
.map(({ body }) => body['messages'] as Message[])
|
||||
.find((messages) =>
|
||||
messages.some(
|
||||
(message) =>
|
||||
message.role === 'tool' && message.tool_call_id === CALL_ID,
|
||||
),
|
||||
);
|
||||
expect(secondMessages).toBeDefined();
|
||||
const skillResult = secondMessages!.find(
|
||||
(message) =>
|
||||
message.role === 'tool' && message.tool_call_id === CALL_ID,
|
||||
);
|
||||
|
||||
@@ -681,12 +681,19 @@ describe('Permission Control (E2E)', () => {
|
||||
}> = [];
|
||||
const firstFile = 'first.txt';
|
||||
const secondFile = 'second.txt';
|
||||
const fakeServer = await startFakeOpenAIServer(({ requestIndex }) => {
|
||||
// Turn 1 consumes requests 0 (tool call) and 1 ('Done.'), so turn 2's
|
||||
// tool call is request 2. Absolute indexing is deliberate: the fake
|
||||
// server never retries, so an extra model request means the protocol
|
||||
// sequence changed and this test should fail loudly rather than drift.
|
||||
if (requestIndex === 0) {
|
||||
let firstToolServed = false;
|
||||
let secondToolServed = false;
|
||||
const fakeServer = await startFakeOpenAIServer(({ body }) => {
|
||||
const tools = body['tools'];
|
||||
const advertisesWriteFile =
|
||||
Array.isArray(tools) &&
|
||||
tools.some(
|
||||
(entry) =>
|
||||
(entry as { function?: { name?: unknown } }).function?.name ===
|
||||
'write_file',
|
||||
);
|
||||
if (!firstToolServed && advertisesWriteFile) {
|
||||
firstToolServed = true;
|
||||
return {
|
||||
toolCalls: [
|
||||
fakeToolCall('write_file', {
|
||||
@@ -696,7 +703,12 @@ describe('Permission Control (E2E)', () => {
|
||||
],
|
||||
};
|
||||
}
|
||||
if (requestIndex === 2) {
|
||||
if (
|
||||
!secondToolServed &&
|
||||
advertisesWriteFile &&
|
||||
userMessageContains(body, `Create ${secondFile}.`)
|
||||
) {
|
||||
secondToolServed = true;
|
||||
return {
|
||||
toolCalls: [
|
||||
fakeToolCall('write_file', {
|
||||
@@ -793,7 +805,6 @@ describe('Permission Control (E2E)', () => {
|
||||
]);
|
||||
|
||||
expect(secondResponseReceived).toBe(true);
|
||||
expect(fakeServer.requests).toHaveLength(4);
|
||||
expect(toolCalls).toHaveLength(1);
|
||||
await expect(helper.readFile(secondFile)).resolves.toBe('second');
|
||||
} finally {
|
||||
|
||||
@@ -256,10 +256,20 @@ describe('SDK MCP Server Integration (E2E)', () => {
|
||||
tools: [calculatorTool],
|
||||
});
|
||||
let streamingRequestIndex = 0;
|
||||
let resumedToolCallServed = false;
|
||||
let resumed = false;
|
||||
fakeResponse = ({ body }) => {
|
||||
if (body['stream'] !== true) {
|
||||
return { content: '{"selected_memories":[]}' };
|
||||
}
|
||||
const tools = body['tools'];
|
||||
const advertisedNames = Array.isArray(tools)
|
||||
? tools.flatMap((entry): string[] => {
|
||||
const name = (entry as { function?: { name?: unknown } }).function
|
||||
?.name;
|
||||
return typeof name === 'string' ? [name] : [];
|
||||
})
|
||||
: [];
|
||||
const requestIndex = streamingRequestIndex++;
|
||||
if (requestIndex === 0) {
|
||||
return {
|
||||
@@ -275,7 +285,12 @@ describe('SDK MCP Server Integration (E2E)', () => {
|
||||
toolCalls: [fakeToolCall(MCP_CALCULATE_SUM, { a: 25, b: 17 })],
|
||||
};
|
||||
}
|
||||
if (requestIndex === 3) {
|
||||
if (
|
||||
resumed &&
|
||||
!resumedToolCallServed &&
|
||||
advertisedNames.includes(MCP_CALCULATE_SUM)
|
||||
) {
|
||||
resumedToolCallServed = true;
|
||||
// The resumed model calls the historical tool directly, without a
|
||||
// second tool_search request.
|
||||
return {
|
||||
@@ -308,6 +323,10 @@ describe('SDK MCP Server Integration (E2E)', () => {
|
||||
await firstQuery.close();
|
||||
}
|
||||
|
||||
const firstQueryStreamingRequests = fakeServer.requests.filter(
|
||||
({ body }) => body['stream'] === true,
|
||||
).length;
|
||||
resumed = true;
|
||||
const resumedQuery = query({
|
||||
prompt: 'Now calculate 8 + 5 with the same tool.',
|
||||
options: {
|
||||
@@ -324,7 +343,22 @@ describe('SDK MCP Server Integration (E2E)', () => {
|
||||
resumedMessages.push(message);
|
||||
}
|
||||
|
||||
expect(advertisedToolNames(fakeServer, 3)).toContain(MCP_CALCULATE_SUM);
|
||||
const resumedRequests = fakeServer.requests
|
||||
.filter(({ body }) => body['stream'] === true)
|
||||
.slice(firstQueryStreamingRequests);
|
||||
expect(
|
||||
resumedRequests.some(({ body }) => {
|
||||
const tools = body['tools'];
|
||||
return (
|
||||
Array.isArray(tools) &&
|
||||
tools.some(
|
||||
(entry) =>
|
||||
(entry as { function?: { name?: unknown } }).function
|
||||
?.name === MCP_CALCULATE_SUM,
|
||||
)
|
||||
);
|
||||
}),
|
||||
).toBe(true);
|
||||
const resumedResults = findToolResults(
|
||||
resumedMessages,
|
||||
MCP_CALCULATE_SUM,
|
||||
|
||||
@@ -8717,6 +8717,11 @@ hello
|
||||
mockInteractionTelemetry.endInteractionSpan,
|
||||
).not.toHaveBeenCalled();
|
||||
|
||||
// MockTurn does not copy emitted tool calls into pendingToolCalls.
|
||||
mockMemoryManager.scheduleMetadataMigration.mockClear();
|
||||
mockMemoryManager.scheduleExtract.mockClear();
|
||||
mockMemoryManager.scheduleDream.mockClear();
|
||||
|
||||
mockTurnRunFn.mockReturnValueOnce(
|
||||
(async function* () {
|
||||
yield { type: LlmEventType.Content, value: 'done' };
|
||||
@@ -8743,6 +8748,66 @@ hello
|
||||
'ok',
|
||||
{ promptId },
|
||||
);
|
||||
expect(mockMemoryManager.scheduleMetadataMigration).toHaveBeenCalledTimes(
|
||||
2,
|
||||
);
|
||||
expect(mockMemoryManager.scheduleExtract).toHaveBeenCalledOnce();
|
||||
expect(mockMemoryManager.scheduleDream).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it('schedules memory work after a tool-result completion without telemetry', async () => {
|
||||
const promptId = 'prompt-tool-loop-without-telemetry';
|
||||
mockInteractionTelemetry.getActiveInteractionSpan.mockReturnValue(
|
||||
undefined,
|
||||
);
|
||||
mockTurnRunFn.mockReturnValueOnce(
|
||||
(async function* () {
|
||||
yield {
|
||||
type: LlmEventType.ToolCallRequest,
|
||||
value: {
|
||||
callId: 'call-1',
|
||||
name: 'read_file',
|
||||
args: {},
|
||||
isClientInitiated: false,
|
||||
prompt_id: promptId,
|
||||
},
|
||||
};
|
||||
})(),
|
||||
);
|
||||
|
||||
await fromAsync(
|
||||
client.sendMessageStream(
|
||||
[{ text: 'use a tool' }],
|
||||
new AbortController().signal,
|
||||
promptId,
|
||||
{ type: SendMessageType.UserQuery },
|
||||
),
|
||||
);
|
||||
|
||||
// MockTurn does not copy emitted tool calls into pendingToolCalls.
|
||||
mockMemoryManager.scheduleMetadataMigration.mockClear();
|
||||
mockMemoryManager.scheduleExtract.mockClear();
|
||||
mockMemoryManager.scheduleDream.mockClear();
|
||||
mockTurnRunFn.mockReturnValueOnce(
|
||||
(async function* () {
|
||||
yield { type: LlmEventType.Content, value: 'done' };
|
||||
})(),
|
||||
);
|
||||
|
||||
await fromAsync(
|
||||
client.sendMessageStream(
|
||||
[{ functionResponse: { name: 'read_file', response: { ok: true } } }],
|
||||
new AbortController().signal,
|
||||
promptId,
|
||||
{ type: SendMessageType.ToolResult },
|
||||
),
|
||||
);
|
||||
|
||||
expect(mockMemoryManager.scheduleMetadataMigration).toHaveBeenCalledTimes(
|
||||
2,
|
||||
);
|
||||
expect(mockMemoryManager.scheduleExtract).toHaveBeenCalledOnce();
|
||||
expect(mockMemoryManager.scheduleDream).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it('starts Retry as a fresh agent invocation', async () => {
|
||||
@@ -11003,6 +11068,58 @@ hello
|
||||
finishMigration({ status: 'skipped', skippedReason: 'complete' });
|
||||
});
|
||||
|
||||
it('runs only metadata migration after a completed tool-result turn', () => {
|
||||
const runBackgroundTasks = (
|
||||
client as unknown as {
|
||||
runManagedAutoMemoryBackgroundTasks: (type: SendMessageType) => void;
|
||||
}
|
||||
).runManagedAutoMemoryBackgroundTasks.bind(client);
|
||||
|
||||
runBackgroundTasks(SendMessageType.ToolResult);
|
||||
|
||||
expect(mockMemoryManager.scheduleMetadataMigration).toHaveBeenCalledTimes(
|
||||
2,
|
||||
);
|
||||
expect(mockMemoryManager.scheduleExtract).not.toHaveBeenCalled();
|
||||
expect(mockMemoryManager.scheduleDream).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('runs tool-result migration after a next-speaker continuation', async () => {
|
||||
const { checkNextSpeaker } = await import(
|
||||
'../utils/nextSpeakerChecker.js'
|
||||
);
|
||||
vi.mocked(checkNextSpeaker)
|
||||
.mockResolvedValueOnce({
|
||||
reasoning: 'continue',
|
||||
next_speaker: 'model',
|
||||
})
|
||||
.mockResolvedValue(null);
|
||||
mockTurnRunFn.mockImplementation(() =>
|
||||
(async function* () {
|
||||
yield { type: LlmEventType.Content, value: 'Done' };
|
||||
})(),
|
||||
);
|
||||
client['chat'] = {
|
||||
addHistory: vi.fn(),
|
||||
getHistory: vi.fn().mockReturnValue([]),
|
||||
} as unknown as LlmChat;
|
||||
|
||||
await fromAsync(
|
||||
client.sendMessageStream(
|
||||
[{ text: 'Tool finished' }],
|
||||
new AbortController().signal,
|
||||
'prompt-id-tool-result-continuation',
|
||||
{ type: SendMessageType.ToolResult },
|
||||
),
|
||||
);
|
||||
|
||||
expect(mockMemoryManager.scheduleMetadataMigration).toHaveBeenCalledTimes(
|
||||
2,
|
||||
);
|
||||
expect(mockMemoryManager.scheduleExtract).not.toHaveBeenCalled();
|
||||
expect(mockMemoryManager.scheduleDream).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('activates a prepared memory protocol before starting UserQuery recall', async () => {
|
||||
let mode: 'legacy' | 'structured' = 'legacy';
|
||||
const setHistory = vi.fn();
|
||||
|
||||
@@ -469,10 +469,7 @@ export class LlmClient {
|
||||
private readonly surfacedRelevantAutoMemoryPaths = new Set<string>();
|
||||
private shutdownRequested = false;
|
||||
private readonly settledSteerInputs = new WeakSet<SteerInput>();
|
||||
private readonly interactionStartTypeByOwner = new WeakMap<
|
||||
object,
|
||||
SendMessageType
|
||||
>();
|
||||
private readonly interactionStartTypes = new Map<string, SendMessageType>();
|
||||
|
||||
private readonly loopDetector: LoopDetectionService;
|
||||
private lastPromptId: string | undefined = undefined;
|
||||
@@ -589,6 +586,7 @@ export class LlmClient {
|
||||
if (this.isInitialized() && this.initializedSessionId === sessionId) {
|
||||
return;
|
||||
}
|
||||
this.interactionStartTypes.clear();
|
||||
|
||||
// Check if we're resuming from a previous session
|
||||
const resumedSessionData = this.config.getResumedSessionData();
|
||||
@@ -1845,6 +1843,7 @@ export class LlmClient {
|
||||
this.lastApiCompletionTimestamp = null;
|
||||
this.lastHookMicrocompactionTimestamp = null;
|
||||
this.recentCompletedToolNames = [];
|
||||
this.interactionStartTypes.clear();
|
||||
// startChat() rewrites the chat to its initial state. Any prior
|
||||
// read_file tool results the FileReadCache still tracks are no
|
||||
// longer in history, so a follow-up Read would serve a placeholder
|
||||
@@ -3058,15 +3057,14 @@ export class LlmClient {
|
||||
}
|
||||
}
|
||||
|
||||
// extract and dream keep the original UserQuery-only gate to preserve
|
||||
// the existing "once per user turn" semantics and avoid redundant work.
|
||||
if (messageType !== SendMessageType.UserQuery) {
|
||||
if (
|
||||
messageType !== SendMessageType.UserQuery &&
|
||||
messageType !== SendMessageType.ToolResult
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
const projectRoot = this.config.getProjectRoot();
|
||||
const sessionId = this.config.getSessionId();
|
||||
const history = this.getHistoryShallow();
|
||||
const mgr = this.config.getMemoryManager();
|
||||
|
||||
if (!this.config.getManagedAutoMemoryEnabled()) {
|
||||
@@ -3088,6 +3086,14 @@ export class LlmClient {
|
||||
});
|
||||
}
|
||||
|
||||
// Extract and Dream stay once per user query; migration also needs the
|
||||
// completed ToolResult path so tool-using turns can activate recall.
|
||||
if (messageType !== SendMessageType.UserQuery) {
|
||||
return;
|
||||
}
|
||||
|
||||
const sessionId = this.config.getSessionId();
|
||||
const history = this.getHistoryShallow();
|
||||
const extractPromise = mgr
|
||||
.scheduleExtract({
|
||||
projectRoot,
|
||||
@@ -3385,6 +3391,8 @@ export class LlmClient {
|
||||
prompt_id: string,
|
||||
options?: SendMessageOptions,
|
||||
turns: number = MAX_TURNS,
|
||||
managedMemoryType: SendMessageType = options?.type ??
|
||||
SendMessageType.UserQuery,
|
||||
): AsyncGenerator<ServerLlmStreamEvent, Turn> {
|
||||
const messageType = options?.type ?? SendMessageType.UserQuery;
|
||||
const startsInteraction =
|
||||
@@ -3407,16 +3415,17 @@ export class LlmClient {
|
||||
errorType?: string,
|
||||
) => {
|
||||
if (
|
||||
!interactionOwner ||
|
||||
interactionOwner &&
|
||||
getActiveInteractionSpan(prompt_id) !== interactionOwner
|
||||
) {
|
||||
return;
|
||||
}
|
||||
const interactionStartType =
|
||||
this.interactionStartTypeByOwner.get(interactionOwner);
|
||||
const interactionStartType = this.interactionStartTypes.get(prompt_id);
|
||||
this.interactionStartTypes.delete(prompt_id);
|
||||
const ownsStructuredOutputContract =
|
||||
interactionStartType === SendMessageType.UserQuery ||
|
||||
interactionStartType === SendMessageType.Retry;
|
||||
if (!interactionOwner) return;
|
||||
if (
|
||||
status === 'ok' &&
|
||||
ownsStructuredOutputContract &&
|
||||
@@ -3736,12 +3745,8 @@ export class LlmClient {
|
||||
messageType,
|
||||
});
|
||||
interactionOwner = getActiveInteractionSpan(prompt_id);
|
||||
if (
|
||||
interactionOwner &&
|
||||
!this.interactionStartTypeByOwner.has(interactionOwner)
|
||||
) {
|
||||
this.interactionStartTypeByOwner.set(interactionOwner, messageType);
|
||||
}
|
||||
this.interactionStartTypes.clear();
|
||||
this.interactionStartTypes.set(prompt_id, messageType);
|
||||
if (
|
||||
interactionOwner &&
|
||||
messageType === SendMessageType.UserQuery &&
|
||||
@@ -3754,6 +3759,10 @@ export class LlmClient {
|
||||
);
|
||||
}
|
||||
}
|
||||
const interactionStartType = this.interactionStartTypes.get(prompt_id);
|
||||
if (interactionStartType !== undefined) {
|
||||
managedMemoryType = interactionStartType;
|
||||
}
|
||||
let userPromptRecordPayload: UserPromptRecordPayload | undefined;
|
||||
let hooksEnabled: boolean;
|
||||
let messageBus: ReturnType<Config['getMessageBus']>;
|
||||
@@ -4999,6 +5008,7 @@ export class LlmClient {
|
||||
steerInput,
|
||||
},
|
||||
steerTurnBudget,
|
||||
managedMemoryType,
|
||||
);
|
||||
} finally {
|
||||
settleSteerInput(steerInput, pushCountBefore);
|
||||
@@ -5148,6 +5158,7 @@ export class LlmClient {
|
||||
steerInput: pendingSteer,
|
||||
},
|
||||
hookTurnBudget,
|
||||
managedMemoryType,
|
||||
);
|
||||
} finally {
|
||||
settleSteerInput(pendingSteer, pushCountBefore);
|
||||
@@ -5246,6 +5257,7 @@ export class LlmClient {
|
||||
steerInput: pendingSteer,
|
||||
},
|
||||
hookTurnBudget,
|
||||
managedMemoryType,
|
||||
);
|
||||
} finally {
|
||||
settleSteerInput(pendingSteer, pushCountBefore);
|
||||
@@ -5307,7 +5319,7 @@ export class LlmClient {
|
||||
|
||||
if (this.config.getSkipNextSpeakerCheck()) {
|
||||
if (!isGoalRuntimeTurn) {
|
||||
this.runManagedAutoMemoryBackgroundTasks(messageType);
|
||||
this.runManagedAutoMemoryBackgroundTasks(managedMemoryType);
|
||||
}
|
||||
if (arenaAgentClient) {
|
||||
await arenaAgentClient.reportCompleted();
|
||||
@@ -5362,6 +5374,7 @@ export class LlmClient {
|
||||
steerInput: pendingSteer,
|
||||
},
|
||||
continueTurnBudget,
|
||||
managedMemoryType,
|
||||
);
|
||||
} finally {
|
||||
settleSteerInput(pendingSteer, pushCountBefore);
|
||||
@@ -5389,7 +5402,7 @@ export class LlmClient {
|
||||
}
|
||||
|
||||
if (!isGoalRuntimeTurn) {
|
||||
this.runManagedAutoMemoryBackgroundTasks(messageType);
|
||||
this.runManagedAutoMemoryBackgroundTasks(managedMemoryType);
|
||||
}
|
||||
|
||||
if (arenaAgentClient) {
|
||||
|
||||
@@ -31,11 +31,13 @@ import { ToolNames } from '../tools/tool-names.js';
|
||||
|
||||
const telemetryMocks = vi.hoisted(() => ({
|
||||
logMemoryExtract: vi.fn(),
|
||||
logMemoryMigration: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock('../telemetry/index.js', async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import('../telemetry/index.js')>()),
|
||||
logMemoryExtract: telemetryMocks.logMemoryExtract,
|
||||
logMemoryMigration: telemetryMocks.logMemoryMigration,
|
||||
}));
|
||||
|
||||
vi.mock('./extract.js', () => ({
|
||||
@@ -829,6 +831,73 @@ describe('MemoryManager', () => {
|
||||
}
|
||||
});
|
||||
|
||||
it('preserves committed counts when the index rebuild fails', async () => {
|
||||
telemetryMocks.logMemoryMigration.mockClear();
|
||||
await writeLegacy(getAutoMemoryRoot(projectRoot), 'project.md');
|
||||
vi.spyOn(
|
||||
metadataMigration,
|
||||
'runMemoryMetadataMigration',
|
||||
).mockResolvedValue({
|
||||
filesScanned: 1,
|
||||
legacyFiles: 1,
|
||||
remainingLegacyFiles: 0,
|
||||
attempted: 1,
|
||||
committed: 1,
|
||||
conflicts: 0,
|
||||
failed: 0,
|
||||
agentDurationMs: 1,
|
||||
inputTokens: 1,
|
||||
outputTokens: 1,
|
||||
totalTokens: 2,
|
||||
indexRebuildError: 'Index rebuild failed',
|
||||
});
|
||||
const manager = new MemoryManager();
|
||||
const params = {
|
||||
projectRoot,
|
||||
scope: 'project' as const,
|
||||
config: makeMockConfig(),
|
||||
};
|
||||
const scheduled = await manager.scheduleMetadataMigration(params);
|
||||
|
||||
expect(scheduled.status).toBe('scheduled');
|
||||
if (scheduled.status !== 'scheduled') return;
|
||||
const record = await scheduled.promise;
|
||||
expect(record).toMatchObject({
|
||||
status: 'failed',
|
||||
error: 'Index rebuild failed',
|
||||
metadata: {
|
||||
attempted: 1,
|
||||
committed: 1,
|
||||
remainingLegacyFiles: 0,
|
||||
indexRebuildError: 'Index rebuild failed',
|
||||
},
|
||||
});
|
||||
// The telemetry event must agree with the task record: a failed run
|
||||
// with the committed counts intact and the reason attached.
|
||||
expect(telemetryMocks.logMemoryMigration).toHaveBeenCalledTimes(1);
|
||||
expect(
|
||||
telemetryMocks.logMemoryMigration.mock.calls[0]?.[1],
|
||||
).toMatchObject({
|
||||
status: 'failed',
|
||||
failure_reason: 'Index rebuild failed',
|
||||
committed: 1,
|
||||
failed: 0,
|
||||
remaining_legacy_files: 0,
|
||||
});
|
||||
for (let attempt = 1; attempt < 3; attempt++) {
|
||||
const retry = await manager.scheduleMetadataMigration(params);
|
||||
expect(retry.status).toBe('scheduled');
|
||||
await retry.promise;
|
||||
}
|
||||
await expect(manager.scheduleMetadataMigration(params)).resolves.toEqual({
|
||||
status: 'skipped',
|
||||
skippedReason: 'stalled',
|
||||
});
|
||||
expect(
|
||||
metadataMigration.runMemoryMetadataMigration,
|
||||
).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it('tells the manual dream gate that the migration has stalled', async () => {
|
||||
// dream.ts owns the manual migration gate but cannot read the stall
|
||||
// counter, so the manager has to hand it over. Without the flag a
|
||||
|
||||
@@ -971,22 +971,29 @@ export class MemoryManager {
|
||||
);
|
||||
return record;
|
||||
}
|
||||
if (result.committed > 0 || result.remainingLegacyFiles === 0) {
|
||||
this.migrationStallCountByDomain.delete(domain);
|
||||
} else {
|
||||
const indexRebuildFailed = result.indexRebuildError !== undefined;
|
||||
if (
|
||||
indexRebuildFailed ||
|
||||
(result.committed === 0 && result.remainingLegacyFiles > 0)
|
||||
) {
|
||||
this.migrationStallCountByDomain.set(
|
||||
domain,
|
||||
(this.migrationStallCountByDomain.get(domain) ?? 0) + 1,
|
||||
);
|
||||
} else {
|
||||
this.migrationStallCountByDomain.delete(domain);
|
||||
}
|
||||
const stalled =
|
||||
(this.migrationStallCountByDomain.get(domain) ?? 0) >=
|
||||
MIGRATION_STALL_LIMIT;
|
||||
const failed = stalled || indexRebuildFailed;
|
||||
this.update(record, {
|
||||
status: stalled ? 'failed' : 'completed',
|
||||
...(stalled
|
||||
status: failed ? 'failed' : 'completed',
|
||||
...(failed
|
||||
? {
|
||||
error: `Migration stalled: ${result.remainingLegacyFiles} legacy file(s) could not be migrated in ${MIGRATION_STALL_LIMIT} consecutive runs; giving up for this session.`,
|
||||
error:
|
||||
result.indexRebuildError ??
|
||||
`Migration stalled: ${result.remainingLegacyFiles} legacy file(s) could not be migrated in ${MIGRATION_STALL_LIMIT} consecutive runs; giving up for this session.`,
|
||||
}
|
||||
: {}),
|
||||
progressText: `Migrated ${result.committed} memory file(s).`,
|
||||
@@ -996,7 +1003,9 @@ export class MemoryManager {
|
||||
params.config,
|
||||
new MemoryMigrationEvent({
|
||||
scope: params.scope,
|
||||
status: stalled ? 'failed' : 'completed',
|
||||
status: failed ? 'failed' : 'completed',
|
||||
failure_reason:
|
||||
result.indexRebuildError ?? (stalled ? 'stalled' : undefined),
|
||||
files_scanned: result.filesScanned,
|
||||
legacy_files: result.legacyFiles,
|
||||
remaining_legacy_files: result.remainingLegacyFiles,
|
||||
|
||||
@@ -32,6 +32,18 @@ import { parseAutoMemoryTopicDocument } from './structured-scan.js';
|
||||
|
||||
vi.mock('../agents/forkedAgent.js', () => ({ runForkedAgent: vi.fn() }));
|
||||
|
||||
const debugLogger = vi.hoisted(() => ({
|
||||
debug: vi.fn(),
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock('../utils/debugLogger.js', async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import('../utils/debugLogger.js')>()),
|
||||
createDebugLogger: () => debugLogger,
|
||||
}));
|
||||
|
||||
vi.mock('node:fs/promises', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import('node:fs/promises')>();
|
||||
return { ...actual };
|
||||
@@ -1244,9 +1256,12 @@ describe('memory metadata migration', () => {
|
||||
return readdir(...args);
|
||||
});
|
||||
try {
|
||||
await expect(runMemoryMetadataMigration(params)).rejects.toThrow(
|
||||
'incomplete',
|
||||
);
|
||||
await expect(runMemoryMetadataMigration(params)).resolves.toMatchObject({
|
||||
attempted: 1,
|
||||
committed: 1,
|
||||
remainingLegacyFiles: 0,
|
||||
indexRebuildError: expect.stringContaining('incomplete'),
|
||||
});
|
||||
expect(generateMetadata).toHaveBeenCalledTimes(1);
|
||||
expect(await fs.readFile(filePath, 'utf-8')).toContain(
|
||||
'name: Migrated memory',
|
||||
@@ -1288,13 +1303,26 @@ describe('memory metadata migration', () => {
|
||||
generateMetadata,
|
||||
};
|
||||
|
||||
await expect(runMemoryMetadataMigration(params)).rejects.toThrow();
|
||||
await fs.rmdir(index);
|
||||
await expect(runMemoryMetadataMigration(params)).resolves.toMatchObject({
|
||||
attempted: 1,
|
||||
committed: 1,
|
||||
remainingLegacyFiles: 0,
|
||||
indexRebuildError: expect.any(String),
|
||||
});
|
||||
// The swallowed rebuild failure must still reach the debug channel —
|
||||
// the in-memory record is gone at process exit.
|
||||
expect(debugLogger.error).toHaveBeenCalledWith(
|
||||
'Memory index rebuild failed:',
|
||||
expect.any(Error),
|
||||
);
|
||||
await fs.rmdir(index);
|
||||
const repaired = await runMemoryMetadataMigration(params);
|
||||
expect(repaired).toMatchObject({
|
||||
attempted: 0,
|
||||
committed: 0,
|
||||
remainingLegacyFiles: 0,
|
||||
});
|
||||
expect(repaired).not.toHaveProperty('indexRebuildError');
|
||||
expect(generateMetadata).toHaveBeenCalledTimes(1);
|
||||
await expect(fs.readFile(index, 'utf-8')).resolves.toContain('legacy.md');
|
||||
});
|
||||
|
||||
@@ -100,6 +100,7 @@ interface MemoryMetadataMigrationResult {
|
||||
inputTokens: number;
|
||||
outputTokens: number;
|
||||
totalTokens: number;
|
||||
indexRebuildError?: string;
|
||||
}
|
||||
|
||||
interface GeneratedMemoryMetadataWithUsage {
|
||||
@@ -758,7 +759,13 @@ export async function runMemoryMetadataMigration(params: {
|
||||
}
|
||||
}
|
||||
// A previous run may have committed metadata before its index write failed.
|
||||
await rebuildIndexes(roots);
|
||||
try {
|
||||
await rebuildIndexes(roots);
|
||||
} catch (error) {
|
||||
debugLogger.error('Memory index rebuild failed:', error);
|
||||
result.indexRebuildError =
|
||||
error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
result.remainingLegacyFiles = (
|
||||
await Promise.all(
|
||||
roots.map((root) =>
|
||||
|
||||
@@ -1869,6 +1869,8 @@ export class MemoryMigrationEvent implements BaseTelemetryEvent {
|
||||
output_tokens: number;
|
||||
total_tokens: number;
|
||||
duration_ms: number;
|
||||
/** Optional: why a failed run failed (index rebuild error message, 'stalled'). */
|
||||
failure_reason?: string;
|
||||
|
||||
constructor(
|
||||
params: Omit<MemoryMigrationEvent, 'event.name' | 'event.timestamp'>,
|
||||
@@ -1889,6 +1891,7 @@ export class MemoryMigrationEvent implements BaseTelemetryEvent {
|
||||
this.output_tokens = params.output_tokens;
|
||||
this.total_tokens = params.total_tokens;
|
||||
this.duration_ms = params.duration_ms;
|
||||
this.failure_reason = params.failure_reason;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user