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

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<T extends keyof WireRecordMap>(

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` (`<homeDir>/sessions/<ws>/<sid>/
@@ -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() {

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

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<ContextTranscript> {
-    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<PersistedRecord[]> {
-  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,

packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycle.ts

@@ -97,7 +97,20 @@ export interface ISessionLifecycleService {
   readonly onDidForkSession: Event<SessionForkedEvent>;
   readonly hooks: Hooks<SessionLifecycleHooks>;
   create(opts: CreateSessionOptions): Promise<ISessionScopeHandle>;
+  /**
+   * 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

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<string, Promise<ISessionScopeHandle | undefined>>();
 
   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<void> {

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<string, unknown>[];
+  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' });
+  });
+});

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<void>((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[] = [];

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', () => {