From f0e671f2ab2303b757f08e95a6cd630ffdf89191 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Fri, 10 Jul 2026 17:25:32 +0800 Subject: [PATCH] fix(agent-core-v2): resume-safe session reads and in-memory transcript - sessionLifecycle: get/list no longer return a session whose cold resume is still in flight, so callers never observe a half-initialized handle; resume remains the way to await a fully restored handle - messageLegacy: reduce the transcript from the main agent's in-memory wire journal instead of re-reading wire.jsonl; AgentWireRecordService now keeps the journal current with live dispatch so cold and live sessions both read a consistent, full transcript --- .../src/agent/wireRecord/wireRecord.ts | 10 +- .../src/agent/wireRecord/wireRecordService.ts | 19 +++ .../src/app/messageLegacy/messageLegacy.ts | 21 +-- .../app/messageLegacy/messageLegacyService.ts | 104 +++++---------- .../app/sessionLifecycle/sessionLifecycle.ts | 13 ++ .../sessionLifecycleService.ts | 23 +++- .../app/messageLegacy/messageLegacy.test.ts | 121 ++++++++++++++++++ .../sessionLifecycle/sessionLifecycle.test.ts | 29 +++++ packages/agent-core-v2/test/index.test.ts | 19 +++ 9 files changed, 268 insertions(+), 91 deletions(-) create mode 100644 packages/agent-core-v2/test/app/messageLegacy/messageLegacy.test.ts diff --git a/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts b/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts index f0f8cb7a9..58f3ae1cb 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts @@ -72,10 +72,12 @@ export interface IAgentWireRecordService { readonly restoring: WireRecordRestoringContext | null; readonly postRestoring: boolean; /** - * Snapshot of every restored record currently held in memory, in order, - * excluding the leading `metadata` envelope record. Intended for callers that - * need to replay the same history into another agent via {@link restore} - * (e.g. session fork). + * Snapshot of every record held in memory, in order, excluding the leading + * `metadata` envelope: the records seeded by {@link restore} plus every record + * persisted by live dispatch afterwards (appended in dispatch order). Intended + * for callers that need to replay or reduce the same history without + * re-reading `wire.jsonl` (e.g. session fork, the messages/snapshot + * transcript). */ getRecords(): readonly PersistedWireRecord[]; register( diff --git a/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts b/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts index 323e273b1..87763d372 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts @@ -7,6 +7,8 @@ import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { IBootstrapService } from '#/app/bootstrap/bootstrap'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { OrderedHookSlot } from '#/hooks'; +import { IAgentWireService } from '#/wire/tokens'; +import type { IWireService } from '#/wire/wireService'; import type { WireRecord, WireRecordMap } from './wireRecord'; import { AGENT_WIRE_PROTOCOL_VERSION, @@ -52,6 +54,7 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco @IBootstrapService bootstrap: IBootstrapService, @IAgentBlobService private readonly blobStore?: IAgentBlobService, @IAppendLogStore private readonly log?: IAppendLogStore, + @IAgentWireService wire?: IWireService, ) { super(); // Each agent scope seeds its own `homedir` (`/sessions/// @@ -63,6 +66,22 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco if (this.log !== undefined) { this._register(this.log.acquire(this.wireScope, WIRE_RECORD_FILENAME)); } + // Keep the in-memory journal current with live dispatch: `restore()` seeds + // it from disk and every persisted record afterwards is appended here in + // dispatch order, so transcript readers reduce memory instead of re-reading + // `wire.jsonl`. Metadata envelopes are excluded to honor `getRecords()`. + // `wire` is optional so direct construction (tests, migration round-trips) + // keeps the restore-only journal; live tracking is active whenever DI + // supplies the agent wire service. + if (wire !== undefined) { + this._register( + wire.onEmission((emission) => { + if (emission.type === 'record' && emission.record.type !== 'metadata') { + this.records.push(emission.record as WireRecord); + } + }), + ); + } } get restoring() { diff --git a/packages/agent-core-v2/src/app/messageLegacy/messageLegacy.ts b/packages/agent-core-v2/src/app/messageLegacy/messageLegacy.ts index cf448275a..ec20db95f 100644 --- a/packages/agent-core-v2/src/app/messageLegacy/messageLegacy.ts +++ b/packages/agent-core-v2/src/app/messageLegacy/messageLegacy.ts @@ -4,16 +4,17 @@ * Implements the legacy `GET /api/v1/sessions/{sid}/messages[/{mid}]` contract * (`packages/server/src/routes/messages.ts`) on top of the native v2 services. * - * The native `IAgentContextMemoryService` (Agent scope, serving `/api/v2` `messages:*`) - * holds the model's CURRENT, folded context and is the transcript source here - * too. For a live session this adapter reads that folded history directly (its - * transcript is in memory by definition); for a cold session it resumes the - * session — restoring the main agent's wire log and replaying it into the - * `ContextModel` — and reads the rebuilt full transcript from the same - * `IAgentContextMemoryService`. The `ContextMessage → Message` projection is - * shared with the `snapshot` and `:undo` edges via - * `contextMemory/messageProjection`. Bound at App scope — a stateless dispatcher - * that resolves the target session/agent per call. + * The native `IAgentContextMemoryService` (Agent scope, serving `/api/v2` + * `messages:*`) holds the model's CURRENT, folded context and is NOT the full + * transcript: after a compaction it collapses into `[...keptUserMessages, + * compaction_summary]`. The full transcript is reduced from the main agent's + * in-memory record journal (`IAgentWireRecordService.getRecords()`), which + * `ISessionLifecycleService.resume` seeds from `wire.jsonl` and live dispatch + * then keeps current — so neither a live nor a cold session is read back from + * disk here. The `ContextMessage → Message` projection is shared with the + * `snapshot` and `:undo` edges via `contextMemory/messageProjection`. Bound at + * App scope — a stateless dispatcher that resolves the target session/agent per + * call. * * Error contract (mapped at the route layer): * - `session.not_found` → 40401 diff --git a/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts b/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts index 0780c07ea..5f18d6c36 100644 --- a/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts +++ b/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts @@ -5,30 +5,26 @@ * its main agent), sources the transcript, and projects it into the v1 wire * shape. * - * History source is the main agent's `wire.jsonl` record log, NOT the live - * `IAgentContextMemoryService.get()`: that live history is the model's CURRENT - * context — after a compaction it collapses into `[...keptUserMessages, - * compaction_summary]`, which made `GET /sessions/{sid}/messages` lose - * everything before the fold. The wire log keeps every record, so - * `reduceContextTranscript` rebuilds the full transcript (compaction inserts a - * summary marker instead of dropping the prefix) — the same view v1's - * `MessageService` serves. Records reach disk through an async flush queue, so - * a request on a live session may find the wire a few records behind memory: - * `foldedLength` is what the live history length WOULD be from the file's - * records, and anything beyond it in the real live context is appended as the - * unflushed tail. Pagination, id derivation, and the role filter mirror v1's - * `MessageService` + * History source is the main agent's in-memory record journal + * (`IAgentWireRecordService.getRecords()`), seeded from `wire.jsonl` by + * `ISessionLifecycleService.resume` and then kept current as live dispatch + * appends each record — so a transcript read never re-reads the file. The + * journal is reduced by `reduceContextTranscript` (the same reducer v1's + * `MessageService` uses), which keeps the full history across compactions + * (inserting a summary marker instead of folding) — unlike the live + * `IAgentContextMemoryService.get()`, whose folded context collapses into + * `[...keptUserMessages, compaction_summary]` and would lose the prefix. + * `foldedLength` is what the live history length WOULD be from the journal's + * records; because the journal can trail the live context by a record within a + * single dispatch, anything beyond it is appended as the unflushed tail. + * Pagination, id derivation, and the role filter mirror v1's `MessageService` * (`packages/agent-core/src/services/message/messageService.ts`). */ -import { readFile } from 'node:fs/promises'; -import { join } from 'node:path'; - import type { Message, PageResponse } from '@moonshot-ai/protocol'; import { InstantiationType } from '#/_base/di/extensions'; -import { type ISessionScopeHandle, LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { ensureMainAgent, MAIN_AGENT_ID } from '#/session/agentLifecycle/mainAgent'; +import { type IAgentScopeHandle, LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; import { reduceContextTranscript, @@ -36,10 +32,11 @@ import { } from '#/agent/contextMemory/contextTranscript'; import { toProtocolMessage } from '#/agent/contextMemory/messageProjection'; import type { ContextMessage } from '#/agent/contextMemory/types'; -import { ErrorCodes, KimiError } from '#/errors'; +import { IAgentWireRecordService } from '#/agent/wireRecord/wireRecord'; import { ISessionIndex } from '#/app/sessionIndex/sessionIndex'; import { ISessionLifecycleService } from '#/app/sessionLifecycle/sessionLifecycle'; -import { ISessionContext } from '#/session/sessionContext/sessionContext'; +import { ErrorCodes, KimiError } from '#/errors'; +import { ensureMainAgent } from '#/session/agentLifecycle/mainAgent'; import type { PersistedRecord } from '#/wire/wireService'; import { IMessageLegacyService, type MessageListQuery } from './messageLegacy'; @@ -123,30 +120,32 @@ export class MessageLegacyService implements IMessageLegacyService { // wire for a cold session; a live session is already current. const agent = await ensureMainAgent(session); - // Read the wire file BEFORE the live context so the in-memory history is - // always at least as new as the file snapshot and the tail merge can only - // append (mirrors v1 `MessageService`). - const transcript = await this.readTranscript(session); + // Reduce the transcript from the main agent's in-memory record journal + // (seeded by `resume` from disk and kept current by live dispatch) instead + // of re-reading `wire.jsonl`. The journal is always at least as new as the + // 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); return entries.map((msg, index) => toProtocolMessage(sessionId, index, msg, summary.createdAt)); } - /** Reduce the main agent's persisted wire log into the full transcript. */ - private async readTranscript(session: ISessionScopeHandle): Promise { - const ctx = session.accessor.get(ISessionContext); - const wirePath = join(ctx.sessionDir, 'agents', MAIN_AGENT_ID, 'wire.jsonl'); - const records = await readWireRecords(wirePath); + /** Reduce the main agent's in-memory record journal into the full transcript. */ + private readTranscript(agent: IAgentScopeHandle): ContextTranscript { + const records = agent + .accessor.get(IAgentWireRecordService) + .getRecords() as readonly PersistedRecord[]; return reduceContextTranscript(records); } } /** * Append the unflushed live tail: when the in-memory (folded) context is - * longer than the wire-derived `foldedLength`, the surplus is records that - * have not reached disk yet and must be appended so a read on a live session - * does not trail memory. + * 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. */ function mergeLiveTail( transcript: ContextTranscript, @@ -156,47 +155,6 @@ function mergeLiveTail( return [...transcript.entries, ...contextMessages.slice(transcript.foldedLength)]; } -/** - * Parse a `wire.jsonl` file. A torn final line (crash mid-flush) is dropped; - * corruption anywhere else throws. A missing file yields an empty record list - * (a brand-new session whose context has not been flushed yet). - */ -async function readWireRecords(wirePath: string): Promise { - let raw: string; - try { - raw = await readFile(wirePath, 'utf8'); - } catch (error) { - if (isEnoent(error)) return []; - throw error; - } - const lines = raw.split('\n'); - const records: PersistedRecord[] = []; - for (let i = 0; i < lines.length; i++) { - let line = lines[i]!; - if (line.endsWith('\r')) line = line.slice(0, -1); - if (line.length === 0) continue; - try { - records.push(JSON.parse(line) as PersistedRecord); - } catch (parseError) { - if (i === lines.length - 1) break; - throw new Error( - `wire.jsonl: corrupted line ${i + 1} in ${wirePath}: ${String(parseError)}`, - { cause: parseError }, - ); - } - } - return records; -} - -function isEnoent(error: unknown): boolean { - return ( - typeof error === 'object' && - error !== null && - 'code' in error && - (error as NodeJS.ErrnoException).code === 'ENOENT' - ); -} - registerScopedService( LifecycleScope.App, IMessageLegacyService, diff --git a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycle.ts b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycle.ts index 1a13bda6e..a62bc4584 100644 --- a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycle.ts +++ b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycle.ts @@ -97,7 +97,20 @@ export interface ISessionLifecycleService { readonly onDidForkSession: Event; readonly hooks: Hooks; create(opts: CreateSessionOptions): Promise; + /** + * Return the live handle for `sessionId`, or `undefined` when it is not open. + * A session whose cold {@link resume} is still in flight is intentionally NOT + * returned — its main agent has not finished restore + replay, so the handle + * is half-initialized. Callers that must obtain the handle should + * `await resume(sessionId)` instead. This invisibility is a service + * invariant, not caller discipline: every read path (`get` / {@link list} / + * {@link resume}) agrees a resuming session is not yet observable. + */ get(sessionId: string): ISessionScopeHandle | undefined; + /** + * Snapshot of every fully-initialized live session. Excludes sessions still + * mid-{@link resume} for the same reason as {@link get}. + */ list(): readonly ISessionScopeHandle[]; /** * Load a persisted session into the live scope tree and restore its main diff --git a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts index b09acff17..42f698d04 100644 --- a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts +++ b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts @@ -94,9 +94,12 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec 'onDidCreateSession', 'onWillCloseSession', ]); - /** In-flight `resume` promises, keyed by session id — de-dupes concurrent - * cold loads so a hot read path (e.g. snapshot retry) cannot materialize - * the same session twice and leak a handle. */ + /** In-flight `resume` promises, keyed by session id. De-dupes concurrent cold + * loads so a hot read path (e.g. snapshot retry) cannot materialize the same + * session twice and leak a handle — and doubles as the visibility gate for + * `get` / `list`: while an id is present here its materialized handle is + * half-initialized (main agent not yet restored + replayed) and must not be + * observable. */ private readonly resuming = new Map>(); constructor( @@ -193,6 +196,12 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec } get(sessionId: string): ISessionScopeHandle | undefined { + // A session mid-resume is already materialized in `this.sessions` (so + // close/archive can still find it) but its main agent has not finished + // restore + replay — exposing it would hand callers a half-initialized + // handle. Hide it until `resume` settles; callers that need the handle + // should `await resume(sessionId)`. + if (this.resuming.has(sessionId)) return undefined; return this.sessions.get(sessionId); } @@ -247,7 +256,13 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec } list(): readonly ISessionScopeHandle[] { - return [...this.sessions.values()]; + // Exclude sessions still mid-resume for the same reason as `get`: the handle + // exists but is not yet restored, so it must not be observable. + const ready: ISessionScopeHandle[] = []; + for (const [id, handle] of this.sessions) { + if (!this.resuming.has(id)) ready.push(handle); + } + return ready; } async close(sessionId: string): Promise { diff --git a/packages/agent-core-v2/test/app/messageLegacy/messageLegacy.test.ts b/packages/agent-core-v2/test/app/messageLegacy/messageLegacy.test.ts new file mode 100644 index 000000000..d99230ab6 --- /dev/null +++ b/packages/agent-core-v2/test/app/messageLegacy/messageLegacy.test.ts @@ -0,0 +1,121 @@ +import { describe, expect, it } from 'vitest'; + +import { type IAgentScopeHandle, type ISessionScopeHandle, LifecycleScope } from '#/_base/di/scope'; +import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; +import type { ContextMessage } from '#/agent/contextMemory/types'; +import { IAgentWireRecordService } from '#/agent/wireRecord/wireRecord'; +import { ISessionIndex, type SessionSummary } from '#/app/sessionIndex/sessionIndex'; +import { ISessionLifecycleService } from '#/app/sessionLifecycle/sessionLifecycle'; +import { IAgentLifecycleService } from '#/session/agentLifecycle/agentLifecycle'; +import { MAIN_AGENT_ID } from '#/session/agentLifecycle/mainAgent'; +import { ISessionCronService } from '#/session/cron/sessionCronService'; + +import { MessageLegacyService } from '#/app/messageLegacy/messageLegacyService'; + +function textMessage(role: ContextMessage['role'], text: string): ContextMessage { + return { role, content: [{ type: 'text', text }], toolCalls: [] }; +} + +function buildService(opts: { + readonly summary: SessionSummary; + readonly records: readonly Record[]; + readonly contextMessages: readonly ContextMessage[]; +}): MessageLegacyService { + const mainHandle = { + id: MAIN_AGENT_ID, + kind: LifecycleScope.Agent, + accessor: { + get: (token: unknown): unknown => { + if (token === IAgentWireRecordService) { + return { getRecords: () => opts.records }; + } + if (token === IAgentContextMemoryService) { + return { get: () => opts.contextMessages }; + } + throw new Error('unexpected main agent service access'); + }, + }, + dispose: () => {}, + } as unknown as IAgentScopeHandle; + + const sessionHandle = { + id: opts.summary.id, + kind: LifecycleScope.Session, + accessor: { + get: (token: unknown): unknown => { + if (token === IAgentLifecycleService) { + return { getHandle: (id: string) => (id === MAIN_AGENT_ID ? mainHandle : undefined) }; + } + if (token === ISessionCronService) return {}; + throw new Error('unexpected session service access'); + }, + }, + dispose: () => {}, + } as unknown as ISessionScopeHandle; + + const lifecycle = { + resume: (sessionId: string) => + Promise.resolve(sessionId === opts.summary.id ? sessionHandle : undefined), + } as unknown as ISessionLifecycleService; + + const index = { + get: (sessionId: string) => Promise.resolve(sessionId === opts.summary.id ? opts.summary : undefined), + } as unknown as ISessionIndex; + + return new MessageLegacyService(lifecycle, index); +} + +describe('MessageLegacyService', () => { + const summary: SessionSummary = { + id: 's1', + workspaceId: 'wd', + createdAt: 1_000, + updatedAt: 1_000, + archived: false, + }; + + it('reduces the transcript from the in-memory record journal (no disk read)', async () => { + const user = textMessage('user', 'hi'); + const assistant = textMessage('assistant', 'hello'); + const svc = buildService({ + summary, + records: [ + { type: 'context.append_message', message: user }, + { type: 'context.append_message', message: assistant }, + ], + // Folded context length matches the journal-derived foldedLength, so the + // live-tail merge is a no-op and the output is purely the journal. + contextMessages: [user, assistant], + }); + + const page = await svc.list('s1', {}); + + // Newest first; both entries come from the journal, not from wire.jsonl. + expect(page.items.map((m) => m.role)).toEqual(['assistant', 'user']); + expect(page.items[1]?.content[0]).toEqual({ type: 'text', text: 'hi' }); + expect(page.has_more).toBe(false); + }); + + it('throws session.not_found for an unknown session id', async () => { + const svc = buildService({ summary, records: [], contextMessages: [] }); + await expect(svc.list('missing', {})).rejects.toMatchObject({ code: 'session.not_found' }); + }); + + it('resolves a single message by derived id', async () => { + const user = textMessage('user', 'hi'); + const assistant = textMessage('assistant', 'hello'); + const svc = buildService({ + summary, + records: [ + { type: 'context.append_message', message: user }, + { type: 'context.append_message', message: assistant }, + ], + contextMessages: [user, assistant], + }); + + const message = await svc.get('s1', 'msg_s1_000001'); + + expect(message.role).toBe('assistant'); + expect(message.content[0]).toEqual({ type: 'text', text: 'hello' }); + }); +}); diff --git a/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts b/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts index 084c30249..7559ea4e8 100644 --- a/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts +++ b/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts @@ -618,6 +618,35 @@ describe('SessionLifecycleService', () => { expect(settled).toBe(true); }); + it('hides a session from get/list until its resume finishes', async () => { + let resolveMcpReady: (() => void) | undefined; + const mcpReady = new Promise((resolve) => { + resolveMcpReady = resolve; + }); + const svc = build([ + stubPair(ISessionIndex, sessionIndexWithSummary('s1', '/tmp/proj')), + stubPair(IAgentLifecycleService, { + ...agentLifecycleWithMainStub(), + ensureMcpReady: () => mcpReady, + }), + ]); + + const resumed = svc.resume('s1'); + await tick(); + + // materialize has registered the handle in `sessions` and is now blocked on + // ensureMcpReady with `resuming` set — the handle must not be observable yet. + expect(svc.get('s1')).toBeUndefined(); + expect(svc.list()).toEqual([]); + + resolveMcpReady?.(); + const handle = await resumed; + + expect(handle?.id).toBe('s1'); + expect(svc.get('s1')).toBe(handle); + expect(svc.list()).toEqual([handle]); + }); + it('fires onDidCloseSession when a session is closed', async () => { const svc = build(); const closed: string[] = []; diff --git a/packages/agent-core-v2/test/index.test.ts b/packages/agent-core-v2/test/index.test.ts index d3e61777d..9e0044f7f 100644 --- a/packages/agent-core-v2/test/index.test.ts +++ b/packages/agent-core-v2/test/index.test.ts @@ -18,6 +18,7 @@ import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; import { setRuntimePhase } from '#/agent/runtime/runtimeOps'; +import { contextAppendMessage } from '#/agent/contextMemory/contextOps'; import { wireMetadata } from '#/agent/wireRecord/metadataOps'; import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore'; import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; @@ -446,6 +447,24 @@ describe('IAgentWireRecordService.records()', () => { (snapshot as unknown as PersistedWireRecord[]).pop(); expect(records.getRecords()).toHaveLength(lengthBefore); }); + + it('appends live-dispatched records after the restored journal', async () => { + const persistence = new InMemoryWireRecordPersistence([ + { type: 'metadata', protocol_version: AGENT_WIRE_PROTOCOL_VERSION, created_at: 1 }, + { type: 'context.append_message', message: userMessage('restored') }, + ]); + const ctx = createTestAgent({ persistence, autoConfigure: false }); + await ctx.wireRecord.restore(); + const restoredLength = ctx.wireRecord.getRecords().length; + + ctx.get(IAgentWireService).dispatch(contextAppendMessage({ message: userMessage('live') })); + + const after = ctx.wireRecord.getRecords(); + expect(after).toHaveLength(restoredLength + 1); + const last = after[after.length - 1] as { type: string; message?: ContextMessage }; + expect(last.type).toBe('context.append_message'); + expect(last.message?.content[0]).toEqual({ type: 'text', text: 'live' }); + }); }); describe.skip('agent replay range build', () => {