From 394493a801b9857a76a2a212726e01c24812cb6f Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Mon, 13 Jul 2026 23:24:04 +0800 Subject: [PATCH] fix(agent-core-v2): repair messages API history after resume - rehydrate restored `blobref:` media URLs to inline `data:` URIs from the agent's blob store, matching the shape of live emissions - pass tool results carrying media (e.g. ReadMediaFile) through as raw content parts instead of flattening them to empty text - use wire record times for `created_at`, nudged to stay strictly increasing, instead of a synthesized session-start offset - carry per-entry record times through `ContextTranscript` and the live-tail merge --- .changeset/v2-messages-rehydrate-media.md | 5 + .../agent/contextMemory/contextTranscript.ts | 32 ++++- .../agent/contextMemory/messageProjection.ts | 14 +- .../app/messageLegacy/messageLegacyService.ts | 53 ++++++- .../contextMemory/contextTranscript.test.ts | 27 ++++ .../app/messageLegacy/messageLegacy.test.ts | 135 ++++++++++++++++++ 6 files changed, 249 insertions(+), 17 deletions(-) create mode 100644 .changeset/v2-messages-rehydrate-media.md diff --git a/.changeset/v2-messages-rehydrate-media.md b/.changeset/v2-messages-rehydrate-media.md new file mode 100644 index 0000000000..86577f46d2 --- /dev/null +++ b/.changeset/v2-messages-rehydrate-media.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/agent-core-v2": patch +--- + +Fix the v2 messages API serving broken history after resume: restored `blobref:` media URLs are rehydrated to inline `data:` URIs from the agent's blob store (matching live emissions), tool results carrying media (e.g. ReadMediaFile) pass their content parts through instead of being flattened to empty text, and `created_at` uses the wire record time instead of a synthesized session-start offset. diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextTranscript.ts b/packages/agent-core-v2/src/agent/contextMemory/contextTranscript.ts index 0c74d2ab38..6fc661f93c 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextTranscript.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextTranscript.ts @@ -46,6 +46,14 @@ const TOOL_INTERRUPTED_ON_RESUME_OUTPUT = export interface ContextTranscript { /** Full message history, compacted prefixes included. */ readonly entries: readonly ContextMessage[]; + /** + * Wall-clock time (ms) of the originating wire record per entry, when + * present — v1 `TranscriptEntry.time`. Synthesized entries (e.g. + * interrupted-on-resume tool results) carry the time of the record that + * triggered their synthesis; entries without a source record (live tail) + * have `undefined`. + */ + readonly times: readonly (number | undefined)[]; /** Length the live (folded) `context.history` would have after these records. */ readonly foldedLength: number; } @@ -62,6 +70,7 @@ interface MutableMessage { interface MutableEntry { message: MutableMessage; + time?: number; } /** Reduce `context.*` wire records into the full transcript. Pure (no I/O). */ @@ -84,7 +93,7 @@ export function reduceContextTranscript(records: Iterable): Con push(...deferred); deferred = []; }; - const closePendingToolResults = (): void => { + const closePendingToolResults = (time: number | undefined): void => { if (pendingToolResultIds.size === 0) return; const interruptedToolCallIds = [...pendingToolResultIds]; for (const toolCallId of interruptedToolCallIds) { @@ -96,6 +105,7 @@ export function reduceContextTranscript(records: Iterable): Con toolCallId, isError: true, }, + time, }); pendingToolResultIds.delete(toolCallId); } @@ -107,12 +117,13 @@ export function reduceContextTranscript(records: Iterable): Con deferred = []; }; - const applyLoopEvent = (event: LoopRecordedEvent): void => { + const applyLoopEvent = (event: LoopRecordedEvent, time: number | undefined): void => { switch (event.type) { case 'step.begin': { - closePendingToolResults(); + closePendingToolResults(time); const entry: MutableEntry = { message: { role: 'assistant', content: [], toolCalls: [] }, + time, }; push(entry); openSteps.set(event.uuid, entry); @@ -153,6 +164,7 @@ export function reduceContextTranscript(records: Iterable): Con toolCallId: event.toolCallId, isError: event.result.isError, }, + time, }); pendingToolResultIds.delete(event.toolCallId); flushDeferredIfToolExchangeClosed(); @@ -181,13 +193,13 @@ export function reduceContextTranscript(records: Iterable): Con for (const record of records) { switch (record.type) { case 'context.append_message': { - const entry = toMutableEntry(record['message'] as ContextMessage); + const entry = toMutableEntry(record['message'] as ContextMessage, record.time); if (pendingToolResultIds.size > 0) deferred.push(entry); else push(entry); break; } case 'context.append_loop_event': - applyLoopEvent(record['event'] as LoopRecordedEvent); + applyLoopEvent(record['event'] as LoopRecordedEvent, record.time); break; case 'context.apply_compaction': { // The live context folds into `[...keptUserMessages, summary]`; the @@ -199,6 +211,7 @@ export function reduceContextTranscript(records: Iterable): Con toolCalls: [], origin: { kind: 'compaction_summary' }, }, + time: record.time, }); foldedLength = recoverFoldedLength(record, transcript, clearFloor, foldedLength); resetOpenState(); @@ -217,10 +230,14 @@ export function reduceContextTranscript(records: Iterable): Con } } - return { entries: transcript.map((e) => e.message), foldedLength }; + return { + entries: transcript.map((e) => e.message), + times: transcript.map((e) => e.time), + foldedLength, + }; } -function toMutableEntry(message: ContextMessage): MutableEntry { +function toMutableEntry(message: ContextMessage, time: number | undefined): MutableEntry { return { message: { ...(message.id !== undefined ? { id: message.id } : {}), @@ -231,6 +248,7 @@ function toMutableEntry(message: ContextMessage): MutableEntry { ...(message.isError !== undefined ? { isError: message.isError } : {}), ...(message.origin !== undefined ? { origin: message.origin } : {}), }, + time, }; } diff --git a/packages/agent-core-v2/src/agent/contextMemory/messageProjection.ts b/packages/agent-core-v2/src/agent/contextMemory/messageProjection.ts index ac4e7a0928..8f86a5758e 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/messageProjection.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/messageProjection.ts @@ -55,7 +55,11 @@ function mapContentPart(part: ContextMessage['content'][number]): MessageContent /** * Build the protocol-shaped `Message.content[]` for one history entry: - * 1. `tool` role → a single `tool_result` part. + * 1. `tool` role → a single `tool_result` part. A result carrying media + * parts (e.g. ReadMediaFile) passes the raw kosong content-part array + * through — the same shape the live `tool.result` event stream carries — + * so REST consumers can still render the media; other results flatten + * to joined text. `is_error` mirrors `ContextMessage.isError`. * 2. other roles → each mapped content part, then one `tool_use` part per * `ToolCall` (assistant only). */ @@ -113,19 +117,21 @@ function buildProtocolContent(msg: ContextMessage): MessageContent[] { /** * Convert one history entry into the protocol's `Message` shape. `created_at` - * is synthesized from the session's `createdAt` plus the entry index so it - * increases monotonically across the array. + * defaults to the session's `createdAt` plus the entry index; callers that + * know the real record time pass `createdAtMsOverride` (v1: the wire record + * time, nudged to stay strictly increasing). */ export function toProtocolMessage( sessionId: string, index: number, msg: ContextMessage, sessionCreatedAtMs: number, + createdAtMsOverride?: number, ): Message { const id = msg.id ?? deriveMessageId(sessionId, index); const role = toProtocolRole(msg.role); const content = buildProtocolContent(msg); - const createdAtMs = sessionCreatedAtMs + index; + const createdAtMs = createdAtMsOverride ?? sessionCreatedAtMs + index; const metadata = msg.origin !== undefined ? { origin: msg.origin } : undefined; return { id, diff --git a/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts b/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts index a9d4a3001c..020d7db1f6 100644 --- a/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts +++ b/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts @@ -25,6 +25,7 @@ import type { Message, PageResponse } from '@moonshot-ai/protocol'; import { InstantiationType } from '#/_base/di/extensions'; import { type IAgentScopeHandle, LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; import { reduceContextTranscript, @@ -126,9 +127,40 @@ export class MessageLegacyService implements IMessageLegacyService { // live context, so the tail merge below can only append (mirrors v1). const transcript = this.readTranscript(agent); const contextMessages = agent.accessor.get(IAgentContextMemoryService).get(); - const entries = mergeLiveTail(transcript, contextMessages); + const merged = mergeLiveTail(transcript, contextMessages); + const entries = await this.rehydrate(agent, merged.messages); + + let previousMs = Number.NEGATIVE_INFINITY; + return entries.map((msg, index) => { + const baseMs = merged.times[index] ?? summary.createdAt + index; + const createdAtMs = Math.max(previousMs + 1, baseMs); + previousMs = createdAtMs; + return toProtocolMessage(sessionId, index, msg, summary.createdAt, createdAtMs); + }); + } - return entries.map((msg, index) => toProtocolMessage(sessionId, index, msg, summary.createdAt)); + /** + * Replace `blobref:` media URLs with `data:` URIs read from the agent's + * blob store (v1's `rehydrateBlobRefs`); unresolvable refs become the + * `[media missing]` placeholder, same as v1 and live replay. + */ + private async rehydrate( + agent: IAgentScopeHandle, + messages: readonly ContextMessage[], + ): Promise { + const blobs = agent.accessor.get(IAgentBlobService); + let changed = false; + const out: ContextMessage[] = []; + for (const msg of messages) { + const content = await blobs.loadParts(msg.content); + if (content === msg.content) { + out.push(msg); + continue; + } + changed = true; + out.push({ ...msg, content: [...content] }); + } + return changed ? out : messages; } /** Reduce the main agent's in-memory record journal into the full transcript. */ @@ -145,14 +177,23 @@ export class MessageLegacyService implements IMessageLegacyService { * longer than the journal-derived `foldedLength`, the surplus is records that * have landed in the live context within the same dispatch but not yet in the * journal, and must be appended so a read on a live session does not trail - * memory. + * memory. Tail entries have no source wire record, hence no record time. */ function mergeLiveTail( transcript: ContextTranscript, contextMessages: readonly ContextMessage[], -): readonly ContextMessage[] { - if (contextMessages.length <= transcript.foldedLength) return transcript.entries; - return [...transcript.entries, ...contextMessages.slice(transcript.foldedLength)]; +): { + readonly messages: readonly ContextMessage[]; + readonly times: readonly (number | undefined)[]; +} { + if (contextMessages.length <= transcript.foldedLength) { + return { messages: transcript.entries, times: transcript.times }; + } + const tail = contextMessages.slice(transcript.foldedLength); + return { + messages: [...transcript.entries, ...tail], + times: [...transcript.times, ...tail.map(() => undefined)], + }; } registerScopedService( diff --git a/packages/agent-core-v2/test/agent/contextMemory/contextTranscript.test.ts b/packages/agent-core-v2/test/agent/contextMemory/contextTranscript.test.ts index 7bd1423dcc..0591ad1959 100644 --- a/packages/agent-core-v2/test/agent/contextMemory/contextTranscript.test.ts +++ b/packages/agent-core-v2/test/agent/contextMemory/contextTranscript.test.ts @@ -123,6 +123,33 @@ describe('reduceContextTranscript', () => { expect(result.foldedLength).toBe(4); }); + it('carries the originating wire record time per entry', () => { + const result = reduceContextTranscript([ + { type: 'context.append_message', message: userMessage('u1'), time: 100 }, + { type: 'context.append_loop_event', event: { type: 'step.begin', uuid: 'st1' }, time: 200 }, + { + type: 'context.append_loop_event', + event: { type: 'tool.call', stepUuid: 'st1', toolCallId: 'c1', name: 'Bash' }, + time: 210, + }, + { + type: 'context.append_loop_event', + event: { + type: 'tool.result', + toolCallId: 'c1', + result: { output: 'ok', isError: false }, + }, + time: 220, + }, + { type: 'context.append_loop_event', event: { type: 'step.end', uuid: 'st1' }, time: 230 }, + // No record time → undefined (falls back to session createdAt + index). + { type: 'context.append_message', message: userMessage('u2') }, + ]); + + expect(result.entries.map((m) => m.role)).toEqual(['user', 'assistant', 'tool', 'user']); + expect(result.times).toEqual([100, 200, 220, undefined]); + }); + it('preserves the pre-compaction assistant reply after a later undo', () => { // The reported regression: send A, /compact, send B, undo. The snapshot // must still show A's assistant reply (compaction only folds the live diff --git a/packages/agent-core-v2/test/app/messageLegacy/messageLegacy.test.ts b/packages/agent-core-v2/test/app/messageLegacy/messageLegacy.test.ts index 557dc90bf6..7170484736 100644 --- a/packages/agent-core-v2/test/app/messageLegacy/messageLegacy.test.ts +++ b/packages/agent-core-v2/test/app/messageLegacy/messageLegacy.test.ts @@ -1,8 +1,10 @@ import { describe, expect, it } from 'vitest'; import { type IAgentScopeHandle, type ISessionScopeHandle, LifecycleScope } from '#/_base/di/scope'; +import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; import type { ContextMessage } from '#/agent/contextMemory/types'; +import type { ContentPart } from '#/app/llmProtocol/message'; import { IAgentWireRecordService } from '#/agent/wireRecord/wireRecord'; import { ISessionIndex, type SessionSummary } from '#/app/sessionIndex/sessionIndex'; import { ISessionLifecycleService } from '#/app/sessionLifecycle/sessionLifecycle'; @@ -20,6 +22,7 @@ function buildService(opts: { readonly summary: SessionSummary; readonly records: readonly Record[]; readonly contextMessages: readonly ContextMessage[]; + readonly loadParts?: (parts: readonly ContentPart[]) => Promise; }): MessageLegacyService { const mainHandle = { id: MAIN_AGENT_ID, @@ -32,6 +35,13 @@ function buildService(opts: { if (token === IAgentContextMemoryService) { return { get: () => opts.contextMessages }; } + if (token === IAgentBlobService) { + return { + loadParts: + opts.loadParts ?? + ((parts: readonly ContentPart[]) => Promise.resolve(parts)), + }; + } throw new Error('unexpected main agent service access'); }, }, @@ -121,4 +131,129 @@ describe('MessageLegacyService', () => { expect(message.role).toBe('assistant'); expect(message.content[0]).toEqual({ type: 'text', text: 'hello' }); }); + + it('rehydrates blobref media URLs from restored journal records', async () => { + const blobRefPart = { + type: 'image_url', + imageUrl: { url: 'blobref:image/png;deadbeef' }, + } as unknown as ContentPart; + const hydratedPart = { + type: 'image_url', + imageUrl: { url: 'data:image/png;base64,AAAA' }, + } as unknown as ContentPart; + const svc = buildService({ + summary, + records: [ + { + type: 'context.append_message', + message: { role: 'user', content: [blobRefPart], toolCalls: [] }, + }, + ], + contextMessages: [], + loadParts: async (parts) => parts.map((p) => (p === blobRefPart ? hydratedPart : p)), + }); + + const page = await svc.list('s1', {}); + + // The restored `blobref:` URL is served inline, the same shape live + // emissions carry — not as a broken blobref: link. + expect(page.items[0]?.content[0]).toEqual({ + type: 'image', + source: { kind: 'url', url: 'data:image/png;base64,AAAA' }, + }); + }); + + it('passes media tool results through as raw content parts instead of flattening', async () => { + const mediaPart = { + type: 'image_url', + imageUrl: { url: 'data:image/png;base64,AAAA' }, + } as unknown as ContentPart; + const svc = buildService({ + summary, + records: [ + { type: 'context.append_loop_event', event: { type: 'step.begin', uuid: 'st1' } }, + { + type: 'context.append_loop_event', + event: { + type: 'tool.call', + stepUuid: 'st1', + toolCallId: 'call_1', + name: 'ReadMediaFile', + args: {}, + }, + }, + { + type: 'context.append_loop_event', + event: { + type: 'tool.result', + toolCallId: 'call_1', + result: { output: [mediaPart], isError: false }, + }, + }, + { type: 'context.append_loop_event', event: { type: 'step.end', uuid: 'st1' } }, + ], + contextMessages: [], + }); + + const page = await svc.list('s1', {}); + + const toolMessage = page.items.find((m) => m.role === 'tool'); + expect(toolMessage?.content[0]).toEqual({ + type: 'tool_result', + tool_call_id: 'call_1', + output: [mediaPart], + }); + }); + + it('flattens text-only tool results to joined text', async () => { + const svc = buildService({ + summary, + records: [ + { type: 'context.append_loop_event', event: { type: 'step.begin', uuid: 'st1' } }, + { + type: 'context.append_loop_event', + event: { type: 'tool.call', stepUuid: 'st1', toolCallId: 'call_1', name: 'Bash' }, + }, + { + type: 'context.append_loop_event', + event: { + type: 'tool.result', + toolCallId: 'call_1', + result: { output: 'command output', isError: false }, + }, + }, + { type: 'context.append_loop_event', event: { type: 'step.end', uuid: 'st1' } }, + ], + contextMessages: [], + }); + + const page = await svc.list('s1', {}); + + const toolMessage = page.items.find((m) => m.role === 'tool'); + expect(toolMessage?.content[0]).toEqual({ + type: 'tool_result', + tool_call_id: 'call_1', + output: 'command output', + }); + }); + + it('uses wire record times for created_at, nudged to stay strictly increasing', async () => { + const svc = buildService({ + summary, // createdAt: 1000 + records: [ + { type: 'context.append_message', message: textMessage('user', 'u1'), time: 5000 }, + // Same record time → the second is nudged one ms forward. + { type: 'context.append_message', message: textMessage('assistant', 'a1'), time: 5000 }, + // No record time → falls back to session createdAt + index, then nudged. + { type: 'context.append_message', message: textMessage('user', 'u2') }, + ], + contextMessages: [], + }); + + const page = await svc.list('s1', {}); + + // Newest first: u2 (nudged 5002), a1 (nudged 5001), u1 (5000). + const created = page.items.map((m) => new Date(m.created_at).getTime()); + expect(created).toEqual([5002, 5001, 5000]); + }); });