mirror of
https://github.com/MoonshotAI/kimi-code.git
synced 2026-07-10 01:39:25 +00:00
Merge branch 'kimi-code-v2' of https://github.com/MoonshotAI/kimi-code into kimi-code-v2
This commit is contained in:
_Kerman
commit
bc84b2e90b
17 changed files with 783 additions and 235 deletions
|
|
@ -0,0 +1,317 @@
|
|||
/**
|
||||
* `contextMemory` transcript reducer — rebuilds the FULL message history of an
|
||||
* agent from its `context.*` wire records for UI display (snapshot / messages).
|
||||
*
|
||||
* The live `ContextModel` (`ContextMemoryService`) rewrites the model-facing
|
||||
* context on `context.apply_compaction` into `[...keptUserMessages,
|
||||
* compaction_summary]`, so reading the live context after a compaction loses
|
||||
* everything before the fold. The wire log keeps every record, though, so this
|
||||
* reducer re-reduces the `context.*` records with the same semantics as the
|
||||
* live `ContextMemory` restore, EXCEPT that `context.apply_compaction` KEEPS
|
||||
* the full history and appends a user-role summary marker — the same view the
|
||||
* v1 transcript / TUI shows after resume. `foldedLength` tracks what the live
|
||||
* (folded) `context.history.length` would be, so a caller can detect and
|
||||
* append an unflushed live tail.
|
||||
*
|
||||
* Mirrors v1 `reduceWireRecords`
|
||||
* (`packages/agent-core/src/services/message/transcript.ts`):
|
||||
* - `context.append_message` → append (deferred while a tool exchange is open)
|
||||
* - `context.append_loop_event` → step.begin/content.part/tool.call mutate the
|
||||
* open assistant; tool.result appends a tool
|
||||
* message with the raw output
|
||||
* - `context.apply_compaction` → keep the full history, append the user-role
|
||||
* summary marker, recover `foldedLength` from
|
||||
* the recorded kept-count fields
|
||||
* - `context.undo` → remove tail messages (skip injections, stop
|
||||
* at compaction summaries / clear floor)
|
||||
* - `context.clear` → keep prior transcript entries but reset the
|
||||
* folded view
|
||||
* - `context.splice` → array splice (legacy 1.5 / internal)
|
||||
*/
|
||||
|
||||
import { type ContentPart, type ToolCall } from '#/app/llmProtocol/message';
|
||||
import type { PersistedRecord } from '#/wire/wireService';
|
||||
|
||||
import {
|
||||
COMPACT_USER_MESSAGE_MAX_TOKENS,
|
||||
collectCompactableUserMessages,
|
||||
isRealUserInput,
|
||||
selectRecentUserMessages,
|
||||
} from './compactionHandoff';
|
||||
import type { LoopRecordedEvent } from './loopEventFold';
|
||||
import type { ContextMessage } from './types';
|
||||
|
||||
const TOOL_INTERRUPTED_ON_RESUME_OUTPUT =
|
||||
'Tool execution was interrupted before its result was recorded. Do not assume the tool completed successfully.';
|
||||
|
||||
export interface ContextTranscript {
|
||||
/** Full message history, compacted prefixes included. */
|
||||
readonly entries: readonly ContextMessage[];
|
||||
/** Length the live (folded) `context.history` would have after these records. */
|
||||
readonly foldedLength: number;
|
||||
}
|
||||
|
||||
interface MutableMessage {
|
||||
id?: string;
|
||||
role: ContextMessage['role'];
|
||||
content: ContentPart[];
|
||||
toolCalls: ToolCall[];
|
||||
toolCallId?: string;
|
||||
isError?: boolean;
|
||||
origin?: ContextMessage['origin'];
|
||||
}
|
||||
|
||||
interface MutableEntry {
|
||||
message: MutableMessage;
|
||||
}
|
||||
|
||||
/** Reduce `context.*` wire records into the full transcript. Pure (no I/O). */
|
||||
export function reduceContextTranscript(records: Iterable<PersistedRecord>): ContextTranscript {
|
||||
const transcript: MutableEntry[] = [];
|
||||
/** What `context.history.length` would be right now (post-folding). */
|
||||
let foldedLength = 0;
|
||||
/** Transcript index `context.undo` may not cross (set by `context.clear`). */
|
||||
let clearFloor = 0;
|
||||
const openSteps = new Map<string, MutableEntry>();
|
||||
const pendingToolResultIds = new Set<string>();
|
||||
let deferred: MutableEntry[] = [];
|
||||
|
||||
const push = (...entries: MutableEntry[]): void => {
|
||||
transcript.push(...entries);
|
||||
foldedLength += entries.length;
|
||||
};
|
||||
const flushDeferredIfToolExchangeClosed = (): void => {
|
||||
if (pendingToolResultIds.size > 0 || deferred.length === 0) return;
|
||||
push(...deferred);
|
||||
deferred = [];
|
||||
};
|
||||
const closePendingToolResults = (): void => {
|
||||
if (pendingToolResultIds.size === 0) return;
|
||||
const interruptedToolCallIds = [...pendingToolResultIds];
|
||||
for (const toolCallId of interruptedToolCallIds) {
|
||||
push({
|
||||
message: {
|
||||
role: 'tool',
|
||||
content: [{ type: 'text', text: TOOL_INTERRUPTED_ON_RESUME_OUTPUT }],
|
||||
toolCalls: [],
|
||||
toolCallId,
|
||||
isError: true,
|
||||
},
|
||||
});
|
||||
pendingToolResultIds.delete(toolCallId);
|
||||
}
|
||||
flushDeferredIfToolExchangeClosed();
|
||||
};
|
||||
const resetOpenState = (): void => {
|
||||
openSteps.clear();
|
||||
pendingToolResultIds.clear();
|
||||
deferred = [];
|
||||
};
|
||||
|
||||
const applyLoopEvent = (event: LoopRecordedEvent): void => {
|
||||
switch (event.type) {
|
||||
case 'step.begin': {
|
||||
closePendingToolResults();
|
||||
const entry: MutableEntry = {
|
||||
message: { role: 'assistant', content: [], toolCalls: [] },
|
||||
};
|
||||
push(entry);
|
||||
openSteps.set(event.uuid, entry);
|
||||
return;
|
||||
}
|
||||
case 'step.end': {
|
||||
openSteps.delete(event.uuid);
|
||||
flushDeferredIfToolExchangeClosed();
|
||||
return;
|
||||
}
|
||||
case 'content.part': {
|
||||
// Lenient where the live reducer throws: a dangling part in a damaged
|
||||
// file should not take the whole transcript down.
|
||||
openSteps.get(event.stepUuid)?.message.content.push(event.part);
|
||||
return;
|
||||
}
|
||||
case 'tool.call': {
|
||||
const openStep = openSteps.get(event.stepUuid);
|
||||
if (openStep === undefined) return;
|
||||
const call: ToolCall = {
|
||||
type: 'function',
|
||||
id: event.toolCallId,
|
||||
name: event.name,
|
||||
arguments: event.args === undefined ? null : JSON.stringify(event.args),
|
||||
...(event.extras !== undefined ? { extras: event.extras } : {}),
|
||||
};
|
||||
openStep.message.toolCalls.push(call);
|
||||
pendingToolResultIds.add(event.toolCallId);
|
||||
return;
|
||||
}
|
||||
case 'tool.result': {
|
||||
if (!pendingToolResultIds.has(event.toolCallId)) return;
|
||||
push({
|
||||
message: {
|
||||
role: 'tool',
|
||||
content: rawToolResultContent(event.result.output),
|
||||
toolCalls: [],
|
||||
toolCallId: event.toolCallId,
|
||||
isError: event.result.isError,
|
||||
},
|
||||
});
|
||||
pendingToolResultIds.delete(event.toolCallId);
|
||||
flushDeferredIfToolExchangeClosed();
|
||||
return;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const applyUndo = (count: number): void => {
|
||||
if (count <= 0) return;
|
||||
let removedUserCount = 0;
|
||||
for (let i = transcript.length - 1; i >= clearFloor; i--) {
|
||||
const message = transcript[i]!.message;
|
||||
if (message.origin?.kind === 'injection') continue;
|
||||
if (message.origin?.kind === 'compaction_summary') break;
|
||||
transcript.splice(i, 1);
|
||||
foldedLength = Math.max(0, foldedLength - 1);
|
||||
if (isRealUserInput(message)) {
|
||||
removedUserCount++;
|
||||
if (removedUserCount >= count) break;
|
||||
}
|
||||
}
|
||||
resetOpenState();
|
||||
};
|
||||
|
||||
for (const record of records) {
|
||||
switch (record.type) {
|
||||
case 'context.append_message': {
|
||||
const message = record['message'] as ContextMessage;
|
||||
const entry: MutableEntry = {
|
||||
message: {
|
||||
...(message.id !== undefined ? { id: message.id } : {}),
|
||||
role: message.role,
|
||||
content: [...message.content],
|
||||
toolCalls: [...message.toolCalls],
|
||||
...(message.toolCallId !== undefined ? { toolCallId: message.toolCallId } : {}),
|
||||
...(message.isError !== undefined ? { isError: message.isError } : {}),
|
||||
...(message.origin !== undefined ? { origin: message.origin } : {}),
|
||||
},
|
||||
};
|
||||
if (pendingToolResultIds.size > 0) deferred.push(entry);
|
||||
else push(entry);
|
||||
break;
|
||||
}
|
||||
case 'context.append_loop_event':
|
||||
applyLoopEvent(record['event'] as LoopRecordedEvent);
|
||||
break;
|
||||
case 'context.splice': {
|
||||
const start = record['start'] as number;
|
||||
const deleteCount = record['deleteCount'] as number;
|
||||
const messages = record['messages'] as readonly ContextMessage[];
|
||||
if (deleteCount === 0 && messages.length === 0) break;
|
||||
// The transcript is append-only (matches the TUI / replay-builder
|
||||
// transcript): a splice's inserted messages land at the tail and the
|
||||
// deleted messages stay for display, so legacy compaction-via-splice
|
||||
// keeps the pre-compaction prefix. `foldedLength` tracks the literal
|
||||
// live-context length after the splice.
|
||||
for (const message of messages) transcript.push(toMutableEntry(message));
|
||||
const clampedStart = Math.min(Math.max(start, 0), foldedLength);
|
||||
const removed = Math.min(deleteCount, foldedLength - clampedStart);
|
||||
foldedLength = foldedLength - removed + messages.length;
|
||||
resetOpenState();
|
||||
break;
|
||||
}
|
||||
case 'context.apply_compaction': {
|
||||
// The live context folds into `[...keptUserMessages, summary]`; the
|
||||
// transcript keeps the full history and appends the summary marker.
|
||||
transcript.push({
|
||||
message: {
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: readCompactionSummaryText(record) }],
|
||||
toolCalls: [],
|
||||
origin: { kind: 'compaction_summary' },
|
||||
},
|
||||
});
|
||||
const keptUserMessageCount = readNumber(record, 'keptUserMessageCount');
|
||||
const keptHeadUserMessageCount = readNumber(record, 'keptHeadUserMessageCount');
|
||||
const compactedCount = readNumber(record, 'compactedCount');
|
||||
if (keptUserMessageCount !== undefined) {
|
||||
// +1 for the summary message; +1 more when the selection split into
|
||||
// head + tail (the live context then also holds an elision marker).
|
||||
foldedLength = keptUserMessageCount + (keptHeadUserMessageCount === undefined ? 1 : 2);
|
||||
} else if (compactedCount !== undefined && compactedCount < foldedLength) {
|
||||
// Legacy record that kept `history.slice(compactedCount)` verbatim.
|
||||
foldedLength = 1 + (foldedLength - compactedCount);
|
||||
} else {
|
||||
// Legacy record covering the whole live history: re-derive from the
|
||||
// post-clear transcript only (the live context rebuilds from the
|
||||
// post-`/clear` messages).
|
||||
const keptUserMessages = selectRecentUserMessages(
|
||||
collectCompactableUserMessages(transcript.slice(clearFloor).map((e) => e.message)),
|
||||
COMPACT_USER_MESSAGE_MAX_TOKENS,
|
||||
);
|
||||
foldedLength = keptUserMessages.length + 1;
|
||||
}
|
||||
resetOpenState();
|
||||
break;
|
||||
}
|
||||
case 'context.undo':
|
||||
applyUndo(record['count'] as number);
|
||||
break;
|
||||
case 'context.clear':
|
||||
clearFloor = transcript.length;
|
||||
foldedLength = 0;
|
||||
resetOpenState();
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return { entries: transcript.map((e) => e.message), foldedLength };
|
||||
}
|
||||
|
||||
function toMutableEntry(message: ContextMessage): MutableEntry {
|
||||
return {
|
||||
message: {
|
||||
...(message.id !== undefined ? { id: message.id } : {}),
|
||||
role: message.role,
|
||||
content: [...message.content],
|
||||
toolCalls: [...message.toolCalls],
|
||||
...(message.toolCallId !== undefined ? { toolCallId: message.toolCallId } : {}),
|
||||
...(message.isError !== undefined ? { isError: message.isError } : {}),
|
||||
...(message.origin !== undefined ? { origin: message.origin } : {}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function readCompactionSummaryText(record: PersistedRecord): string {
|
||||
const summary = record['summary'];
|
||||
if (typeof summary === 'string') return summary;
|
||||
const contextSummary = record['contextSummary'];
|
||||
if (typeof contextSummary === 'string') return contextSummary;
|
||||
// Legacy record whose `summary` is a whole ContextMessage — flatten its text.
|
||||
if (isContextMessageLike(summary)) return textOfParts(summary.content);
|
||||
return '';
|
||||
}
|
||||
|
||||
function isContextMessageLike(value: unknown): value is ContextMessage {
|
||||
if (value === null || typeof value !== 'object' || Array.isArray(value)) return false;
|
||||
const message = value as { role?: unknown; content?: unknown };
|
||||
return typeof message.role === 'string' && Array.isArray(message.content);
|
||||
}
|
||||
|
||||
function textOfParts(content: readonly ContentPart[]): string {
|
||||
let text = '';
|
||||
for (const part of content) {
|
||||
if (part.type === 'text') text += part.text;
|
||||
}
|
||||
return text;
|
||||
}
|
||||
|
||||
function readNumber(record: PersistedRecord, key: string): number | undefined {
|
||||
const value = record[key];
|
||||
return typeof value === 'number' ? value : undefined;
|
||||
}
|
||||
|
||||
/** Raw output verbatim — status text is added only at LLM projection, never in the transcript. */
|
||||
function rawToolResultContent(output: string | readonly ContentPart[]): ContentPart[] {
|
||||
return typeof output === 'string' ? [{ type: 'text', text: output }] : [...output];
|
||||
}
|
||||
|
|
@ -3,25 +3,44 @@
|
|||
*
|
||||
* Stateless App-scope dispatcher: each call resolves the target session (and
|
||||
* its main agent), sources the transcript, and projects it into the v1 wire
|
||||
* shape. Both live and cold sessions are read from the main agent's
|
||||
* `IAgentContextMemoryService`: live sessions already hold the folded history in
|
||||
* memory, and cold sessions are resumed (restoring the main agent's wire log and
|
||||
* replaying it into the `ContextModel`) before the read, so the same `get()`
|
||||
* yields the full transcript. Pagination, id derivation, and the role filter
|
||||
* mirror v1's `MessageService`
|
||||
* 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`
|
||||
* (`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 IAgentScopeHandle, LifecycleScope, registerScopedService } from '#/_base/di/scope';
|
||||
import { ensureMainAgent } from '#/session/agentLifecycle/mainAgent';
|
||||
import { type ISessionScopeHandle, LifecycleScope, registerScopedService } from '#/_base/di/scope';
|
||||
import { ensureMainAgent, MAIN_AGENT_ID } from '#/session/agentLifecycle/mainAgent';
|
||||
import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory';
|
||||
import {
|
||||
reduceContextTranscript,
|
||||
type ContextTranscript,
|
||||
} from '#/agent/contextMemory/contextTranscript';
|
||||
import { toProtocolMessage } from '#/agent/contextMemory/messageProjection';
|
||||
import type { ContextMessage } from '#/agent/contextMemory/types';
|
||||
import { ErrorCodes, KimiError } from '#/errors';
|
||||
import { ISessionIndex } from '#/app/sessionIndex/sessionIndex';
|
||||
import { ISessionLifecycleService } from '#/app/sessionLifecycle/sessionLifecycle';
|
||||
import { ISessionContext } from '#/session/sessionContext/sessionContext';
|
||||
import type { PersistedRecord } from '#/wire/wireService';
|
||||
|
||||
import { IMessageLegacyService, type MessageListQuery } from './messageLegacy';
|
||||
|
||||
|
|
@ -97,33 +116,85 @@ export class MessageLegacyService implements IMessageLegacyService {
|
|||
throw new KimiError(ErrorCodes.SESSION_NOT_FOUND, `session ${sessionId} does not exist`);
|
||||
}
|
||||
|
||||
const agent = await this.resolveMainAgent(sessionId);
|
||||
if (agent === undefined) return [];
|
||||
|
||||
// The transcript is the main agent's `ContextModel`: live sessions already
|
||||
// hold it in memory, and cold sessions have been resumed (wire log restored
|
||||
// + replayed into the Model) by `resolveMainAgent` before we get here, so a
|
||||
// single `get()` covers both. The legacy replay read model
|
||||
// (`IAgentRecordService.buildReplay`) is empty on every path and is gone.
|
||||
const source = agent.accessor.get(IAgentContextMemoryService).get();
|
||||
|
||||
return source.map((msg, index) => toProtocolMessage(sessionId, index, msg, summary.createdAt));
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the session's main agent, loading + restoring it from the persisted
|
||||
* wire log when the session is cold (delegated to `ISessionLifecycleService.resume`).
|
||||
* Returns `undefined` only when a cold session's workspace is gone and the
|
||||
* session directory cannot be reconstructed (mirrors the `fork` limitation).
|
||||
*/
|
||||
private async resolveMainAgent(sessionId: string): Promise<IAgentScopeHandle | undefined> {
|
||||
const session = await this.lifecycle.resume(sessionId);
|
||||
if (session === undefined) return undefined;
|
||||
// Live session whose main agent has not been materialized yet: create it
|
||||
// fresh. No restore here — the session was already live, so any persisted
|
||||
// wire is already reflected in memory; re-restoring would re-apply splices.
|
||||
return ensureMainAgent(session);
|
||||
if (session === undefined) return [];
|
||||
// Materialize the main agent so the live context is available for the
|
||||
// unflushed-tail merge below. `resume` already restored + replayed the
|
||||
// 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);
|
||||
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);
|
||||
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.
|
||||
*/
|
||||
function mergeLiveTail(
|
||||
transcript: ContextTranscript,
|
||||
contextMessages: readonly ContextMessage[],
|
||||
): readonly ContextMessage[] {
|
||||
if (contextMessages.length <= transcript.foldedLength) return transcript.entries;
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -82,7 +82,7 @@ export class SessionLegacyService implements ISessionLegacyService {
|
|||
sessionId: string,
|
||||
body: UpdateSessionProfileRequest,
|
||||
): Promise<SessionWireFields> {
|
||||
const session = this.lifecycle.get(sessionId);
|
||||
const session = await this.lifecycle.resume(sessionId);
|
||||
if (session === undefined) {
|
||||
throw new KimiError(ErrorCodes.SESSION_NOT_FOUND, `session ${sessionId} does not exist`);
|
||||
}
|
||||
|
|
@ -274,7 +274,10 @@ export class SessionLegacyService implements ISessionLegacyService {
|
|||
// Native `ISessionLifecycleService.archive` is a no-op for sessions that
|
||||
// are not live, so gate on the live handle (matches the previous route
|
||||
// behaviour): a missing live session is reported as `session.not_found`.
|
||||
if (this.lifecycle.get(sessionId) === undefined) {
|
||||
// `resume` (not `get`) so archiving a freshly-opened cold session still
|
||||
// works; `resume` returns undefined only when the session is unknown or its
|
||||
// workspace is gone, which is reported as `session.not_found`.
|
||||
if ((await this.lifecycle.resume(sessionId)) === undefined) {
|
||||
throw new KimiError(ErrorCodes.SESSION_NOT_FOUND, `session ${sessionId} does not exist`);
|
||||
}
|
||||
await this.lifecycle.archive(sessionId);
|
||||
|
|
@ -315,11 +318,6 @@ export class SessionLegacyService implements ISessionLegacyService {
|
|||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the session's main agent, creating it on demand (mirrors v1's
|
||||
* `resumeSession`; delegates to the `agentLifecycle` domain's
|
||||
* `ensureMainAgent` bootstrap helper).
|
||||
*/
|
||||
/**
|
||||
* Apply the v1 `agent_config` patch onto the main agent. Mirrors v1's
|
||||
* `IPromptService.applyAgentState` (`promptService.ts:650-743`) in both order
|
||||
|
|
@ -383,7 +381,13 @@ export class SessionLegacyService implements ISessionLegacyService {
|
|||
}
|
||||
|
||||
private async resolveMainAgent(sessionId: string): Promise<IAgentScopeHandle> {
|
||||
const session = this.lifecycle.get(sessionId);
|
||||
// `resume` (not `get`) so a persisted-but-cold session — freshly opened in
|
||||
// the web UI before any prompt, or created by a previous process — is loaded
|
||||
// from disk instead of being reported as `session.not_found`. Mirrors v1's
|
||||
// `SessionService.undo`/`compact`, which call `resumeSession` first; `resume`
|
||||
// returns `undefined` only when the session is unknown or its workspace is
|
||||
// gone, so a genuinely missing session still 404s.
|
||||
const session = await this.lifecycle.resume(sessionId);
|
||||
if (session === undefined) {
|
||||
throw new KimiError(ErrorCodes.SESSION_NOT_FOUND, `session ${sessionId} does not exist`);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -302,6 +302,7 @@ export * from '#/agent/contextMemory/compactionHandoff';
|
|||
export * from '#/agent/contextMemory/loopEventFold';
|
||||
export * from '#/agent/contextMemory/messageId';
|
||||
export * from '#/agent/contextMemory/messageProjection';
|
||||
export * from '#/agent/contextMemory/contextTranscript';
|
||||
export * from '#/agent/contextMemory/types';
|
||||
export * from '#/agent/systemReminder/systemReminder';
|
||||
export * from '#/agent/systemReminder/systemReminderService';
|
||||
|
|
|
|||
212
packages/agent-core-v2/test/contextMemory/transcript.test.ts
Normal file
212
packages/agent-core-v2/test/contextMemory/transcript.test.ts
Normal file
|
|
@ -0,0 +1,212 @@
|
|||
/**
|
||||
* Tests for `reduceContextTranscript` — the wire-transcript reducer used by the
|
||||
* snapshot and messages endpoints. Mirrors v1 `reduceWireRecords` expectations:
|
||||
* compaction keeps the prefix and appends a summary marker; undo removes the
|
||||
* tail but stops at compaction summaries / clear floors; clear keeps the
|
||||
* transcript but resets the folded view.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import {
|
||||
reduceContextTranscript,
|
||||
type ContextTranscript,
|
||||
} from '../../src/agent/contextMemory/contextTranscript';
|
||||
import type { LoopRecordedEvent } from '../../src/agent/contextMemory/loopEventFold';
|
||||
import type { ContextMessage, PromptOrigin } from '../../src/agent/contextMemory/types';
|
||||
import type { PersistedRecord } from '../../src/wire/wireService';
|
||||
|
||||
function userMessage(text: string, origin?: PromptOrigin): ContextMessage {
|
||||
return {
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text }],
|
||||
toolCalls: [],
|
||||
...(origin === undefined ? {} : { origin }),
|
||||
};
|
||||
}
|
||||
|
||||
function assistantMessage(text: string): ContextMessage {
|
||||
return { role: 'assistant', content: [{ type: 'text', text }], toolCalls: [] };
|
||||
}
|
||||
|
||||
function appendMessage(message: ContextMessage): PersistedRecord {
|
||||
return { type: 'context.append_message', message };
|
||||
}
|
||||
|
||||
function loopEvent(event: LoopRecordedEvent): PersistedRecord {
|
||||
return { type: 'context.append_loop_event', event };
|
||||
}
|
||||
|
||||
function assistantStep(uuid: string, text: string): PersistedRecord[] {
|
||||
return [
|
||||
loopEvent({ type: 'step.begin', uuid }),
|
||||
loopEvent({ type: 'content.part', stepUuid: uuid, part: { type: 'text', text } }),
|
||||
loopEvent({ type: 'step.end', uuid }),
|
||||
];
|
||||
}
|
||||
|
||||
function compaction(
|
||||
summary: string,
|
||||
compactedCount: number,
|
||||
keptUserMessageCount?: number,
|
||||
keptHeadUserMessageCount?: number,
|
||||
): PersistedRecord {
|
||||
return {
|
||||
type: 'context.apply_compaction',
|
||||
summary,
|
||||
contextSummary: `prefixed ${summary}`,
|
||||
compactedCount,
|
||||
tokensBefore: 1000,
|
||||
tokensAfter: 100,
|
||||
...(keptUserMessageCount === undefined ? {} : { keptUserMessageCount }),
|
||||
...(keptHeadUserMessageCount === undefined ? {} : { keptHeadUserMessageCount }),
|
||||
};
|
||||
}
|
||||
|
||||
function undo(count: number): PersistedRecord {
|
||||
return { type: 'context.undo', count };
|
||||
}
|
||||
|
||||
function texts(result: ContextTranscript): string[] {
|
||||
return result.entries.map((m) =>
|
||||
m.content.map((p) => (p.type === 'text' ? p.text : `[${p.type}]`)).join(''),
|
||||
);
|
||||
}
|
||||
|
||||
describe('reduceContextTranscript', () => {
|
||||
it('builds the transcript from append_message and loop events', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('u1')),
|
||||
...assistantStep('s1', 'a1'),
|
||||
]);
|
||||
expect(texts(result)).toEqual(['u1', 'a1']);
|
||||
expect(result.entries.map((m) => m.role)).toEqual(['user', 'assistant']);
|
||||
expect(result.foldedLength).toBe(2);
|
||||
});
|
||||
|
||||
it('compaction keeps the prefix and appends a user-role summary marker', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('u1')),
|
||||
...assistantStep('s1', 'a1'),
|
||||
appendMessage(userMessage('u2')),
|
||||
...assistantStep('s2', 'a2'),
|
||||
compaction('SUM', 4),
|
||||
appendMessage(userMessage('u3')),
|
||||
]);
|
||||
expect(texts(result)).toEqual(['u1', 'a1', 'u2', 'a2', 'SUM', 'u3']);
|
||||
expect(result.entries[4]!.origin).toEqual({ kind: 'compaction_summary' });
|
||||
expect(result.entries[4]!.role).toBe('user');
|
||||
// live folded view would be [u1, u2, SUM, u3]
|
||||
expect(result.foldedLength).toBe(4);
|
||||
});
|
||||
|
||||
it('uses the recorded kept-user count for foldedLength when present', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('u1')),
|
||||
appendMessage(userMessage('u2')),
|
||||
appendMessage(userMessage('u3')),
|
||||
compaction('SUM', 3, 1),
|
||||
appendMessage(userMessage('u4')),
|
||||
]);
|
||||
// 1 kept user message + summary + u4 appended after compaction.
|
||||
expect(result.foldedLength).toBe(3);
|
||||
});
|
||||
|
||||
it('accounts for the elision marker when the record kept a head segment', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('u1')),
|
||||
appendMessage(userMessage('u2')),
|
||||
...assistantStep('s1', 'a1'),
|
||||
compaction('SUM', 3, 2, 1),
|
||||
]);
|
||||
// Live context: head user + elision marker + tail user + summary.
|
||||
expect(result.foldedLength).toBe(4);
|
||||
});
|
||||
|
||||
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
|
||||
// context; the transcript keeps the full history).
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('message A')),
|
||||
appendMessage(assistantMessage('reply A')),
|
||||
compaction('summary text', 2, 1),
|
||||
appendMessage(userMessage('message B')),
|
||||
appendMessage(assistantMessage('reply B')),
|
||||
undo(1),
|
||||
]);
|
||||
expect(texts(result)).toEqual(['message A', 'reply A', 'summary text']);
|
||||
expect(result.entries.map((m) => m.role)).toEqual(['user', 'assistant', 'user']);
|
||||
expect(result.foldedLength).toBe(2);
|
||||
});
|
||||
|
||||
it('undo without compaction keeps the earlier exchange intact', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('message A')),
|
||||
appendMessage(assistantMessage('reply A')),
|
||||
appendMessage(userMessage('message B')),
|
||||
appendMessage(assistantMessage('reply B')),
|
||||
undo(1),
|
||||
]);
|
||||
expect(texts(result)).toEqual(['message A', 'reply A']);
|
||||
});
|
||||
|
||||
it('undo stops at a compaction summary', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('old')),
|
||||
compaction('SUM', 1, 1),
|
||||
appendMessage(userMessage('recent')),
|
||||
appendMessage(assistantMessage('answer')),
|
||||
undo(2),
|
||||
]);
|
||||
// Only the post-compaction exchange is removed; the summary blocks further undo.
|
||||
expect(texts(result)).toEqual(['old', 'SUM']);
|
||||
});
|
||||
|
||||
it('clear keeps prior transcript entries but resets the folded view', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('u1')),
|
||||
appendMessage(userMessage('u2')),
|
||||
{ type: 'context.clear' },
|
||||
appendMessage(userMessage('u3')),
|
||||
]);
|
||||
expect(texts(result)).toEqual(['u1', 'u2', 'u3']);
|
||||
expect(result.foldedLength).toBe(1);
|
||||
});
|
||||
|
||||
it('undo does not cross a clear floor', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('u1')),
|
||||
{ type: 'context.clear' },
|
||||
appendMessage(userMessage('u2')),
|
||||
appendMessage(assistantMessage('a2')),
|
||||
undo(1),
|
||||
]);
|
||||
// The post-clear exchange (u2 + a2) is removed; pre-clear u1 stays in the
|
||||
// transcript and the clear floor blocks undo from reaching it.
|
||||
expect(texts(result)).toEqual(['u1']);
|
||||
expect(result.foldedLength).toBe(0);
|
||||
});
|
||||
|
||||
it('folds tool calls and results from loop events', () => {
|
||||
const result = reduceContextTranscript([
|
||||
appendMessage(userMessage('q')),
|
||||
loopEvent({ type: 'step.begin', uuid: 's1' }),
|
||||
loopEvent({ type: 'content.part', stepUuid: 's1', part: { type: 'text', text: 'hi' } }),
|
||||
loopEvent({
|
||||
type: 'tool.call',
|
||||
stepUuid: 's1',
|
||||
toolCallId: 'call_1',
|
||||
name: 'Bash',
|
||||
args: { command: 'echo hi' },
|
||||
}),
|
||||
loopEvent({ type: 'tool.result', toolCallId: 'call_1', result: { output: 'hi' } }),
|
||||
loopEvent({ type: 'step.end', uuid: 's1' }),
|
||||
]);
|
||||
expect(result.entries.map((m) => m.role)).toEqual(['user', 'assistant', 'tool']);
|
||||
expect(result.entries[1]!.toolCalls).toHaveLength(1);
|
||||
expect(result.entries[1]!.toolCalls[0]!.id).toBe('call_1');
|
||||
expect(result.entries[2]!.toolCallId).toBe('call_1');
|
||||
expect(result.foldedLength).toBe(3);
|
||||
});
|
||||
});
|
||||
|
|
@ -100,7 +100,7 @@ export function registerApprovalsRoutes(app: ApprovalRouteHost, core: Scope): vo
|
|||
},
|
||||
async (req, reply) => {
|
||||
const { session_id } = req.params;
|
||||
const handle = core.accessor.get(ISessionLifecycleService).get(session_id);
|
||||
const handle = await core.accessor.get(ISessionLifecycleService).resume(session_id);
|
||||
if (handle === undefined) {
|
||||
reply.send(
|
||||
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
|
||||
|
|
@ -134,7 +134,7 @@ export function registerApprovalsRoutes(app: ApprovalRouteHost, core: Scope): vo
|
|||
},
|
||||
async (req, reply) => {
|
||||
const { session_id, approval_id } = req.params;
|
||||
const handle = core.accessor.get(ISessionLifecycleService).get(session_id);
|
||||
const handle = await core.accessor.get(ISessionLifecycleService).resume(session_id);
|
||||
if (handle === undefined) {
|
||||
reply.send(
|
||||
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
|
||||
|
|
|
|||
|
|
@ -145,6 +145,18 @@ export function registerFsRoutes(app: FsRouteHost, core: Scope): void {
|
|||
}
|
||||
const fsAction = action as FsAction;
|
||||
|
||||
// Cold-load a persisted-but-not-live session so fs actions (which only
|
||||
// need the work dir) do not 404 on a freshly-opened session. Matches v1,
|
||||
// which reads the persisted cwd. `resume` returns undefined only when the
|
||||
// session is unknown or its workspace is gone.
|
||||
const session = await core.accessor.get(ISessionLifecycleService).resume(session_id);
|
||||
if (session === undefined) {
|
||||
reply.send(
|
||||
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
switch (fsAction) {
|
||||
case 'list':
|
||||
|
|
@ -232,6 +244,16 @@ export function registerFsRoutes(app: FsRouteHost, core: Scope): void {
|
|||
return;
|
||||
}
|
||||
|
||||
// Cold-load so a freshly-opened (persisted but not live) session can still
|
||||
// serve downloads; `resume` only returns undefined for unknown / workspace-gone.
|
||||
const session = await core.accessor.get(ISessionLifecycleService).resume(session_id);
|
||||
if (session === undefined) {
|
||||
reply.send(
|
||||
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
let resolved: Awaited<ReturnType<ISessionFsService['resolveDownload']>>;
|
||||
try {
|
||||
resolved = await resolveFs(core, session_id).resolveDownload(relPath);
|
||||
|
|
|
|||
|
|
@ -124,7 +124,7 @@ export function registerQuestionsRoutes(app: QuestionRouteHost, core: Scope): vo
|
|||
},
|
||||
async (req, reply) => {
|
||||
const { session_id } = req.params;
|
||||
const handle = core.accessor.get(ISessionLifecycleService).get(session_id);
|
||||
const handle = await core.accessor.get(ISessionLifecycleService).resume(session_id);
|
||||
if (handle === undefined) {
|
||||
reply.send(
|
||||
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
|
||||
|
|
@ -173,7 +173,7 @@ export function registerQuestionsRoutes(app: QuestionRouteHost, core: Scope): vo
|
|||
const questionId = parsed.id;
|
||||
const action: 'resolve' | 'dismiss' = parsed.kind === 'bare' ? 'resolve' : parsed.action;
|
||||
|
||||
const handle = core.accessor.get(ISessionLifecycleService).get(session_id);
|
||||
const handle = await core.accessor.get(ISessionLifecycleService).resume(session_id);
|
||||
if (handle === undefined) {
|
||||
reply.send(
|
||||
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
|
||||
|
|
|
|||
|
|
@ -622,7 +622,9 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
|
|||
}
|
||||
|
||||
if (parsed.action === 'btw') {
|
||||
const session = core.accessor.get(ISessionLifecycleService).get(parsed.id);
|
||||
// `resume` (not `get`) so a freshly-opened cold session can start a
|
||||
// side-channel agent; matches v1's `startBtw` which resumes first.
|
||||
const session = await core.accessor.get(ISessionLifecycleService).resume(parsed.id);
|
||||
if (session === undefined) {
|
||||
throw new KimiError(
|
||||
ErrorCodes.SESSION_NOT_FOUND,
|
||||
|
|
@ -772,7 +774,9 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
|
|||
},
|
||||
async (req, reply) => {
|
||||
const { session_id } = req.params;
|
||||
const session = core.accessor.get(ISessionLifecycleService).get(session_id);
|
||||
// `resume` (not `get`) so a freshly-opened cold session still computes its
|
||||
// warnings; matches v1's best-effort `resumeSession` before reading them.
|
||||
const session = await core.accessor.get(ISessionLifecycleService).resume(session_id);
|
||||
if (session === undefined) {
|
||||
reply.send(
|
||||
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
|
||||
|
|
|
|||
|
|
@ -141,7 +141,11 @@ async function resolveActivatedSession(
|
|||
sessionId: string,
|
||||
requestId: string,
|
||||
): Promise<ResolvedSession> {
|
||||
const handle = core.accessor.get(ISessionLifecycleService).get(sessionId);
|
||||
// `resume` (not `get`) so listing/activating skills on a freshly-opened cold
|
||||
// session cold-loads it instead of reporting "not activated"; matches v1's
|
||||
// `resumeSession` in SkillService. `resume` returns undefined only when the
|
||||
// session is unknown or its workspace is gone.
|
||||
const handle = await core.accessor.get(ISessionLifecycleService).resume(sessionId);
|
||||
if (handle !== undefined) return { handle };
|
||||
|
||||
const summary = await core.accessor.get(ISessionIndex).get(sessionId);
|
||||
|
|
|
|||
|
|
@ -77,12 +77,13 @@ const sessionAndTailParamSchema = z.object({
|
|||
const detailsSchema = z.array(z.object({ path: z.string(), message: z.string() }));
|
||||
|
||||
/**
|
||||
* Resolve the session's `ISessionTerminalService` from the URL session id. Throws a
|
||||
* coded `session.not_found` when the session is not live — terminals are
|
||||
* inherently live PTY state, so a persisted-but-not-live session is `40401`.
|
||||
* Resolve the session's `ISessionTerminalService` from the URL session id,
|
||||
* cold-loading a persisted-but-not-live session first (matches v1, which spawns
|
||||
* from the persisted cwd). Throws `session.not_found` only when the session is
|
||||
* unknown or its workspace is gone.
|
||||
*/
|
||||
function resolveTerminal(core: Scope, sessionId: string): ISessionTerminalService {
|
||||
const session = core.accessor.get(ISessionLifecycleService).get(sessionId);
|
||||
async function resolveTerminal(core: Scope, sessionId: string): Promise<ISessionTerminalService> {
|
||||
const session = await core.accessor.get(ISessionLifecycleService).resume(sessionId);
|
||||
if (session === undefined) {
|
||||
throw new KimiError(ErrorCodes.SESSION_NOT_FOUND, `session ${sessionId} does not exist`);
|
||||
}
|
||||
|
|
@ -106,7 +107,7 @@ export function registerTerminalsRoutes(app: TerminalsRouteHost, core: Scope): v
|
|||
async (req, reply) => {
|
||||
try {
|
||||
const { session_id } = req.params;
|
||||
const items = await resolveTerminal(core, session_id).list();
|
||||
const items = await (await resolveTerminal(core, session_id)).list();
|
||||
reply.send(okEnvelope({ items }, req.id));
|
||||
} catch (err) {
|
||||
sendMappedError(reply, req.id, err);
|
||||
|
|
@ -137,7 +138,7 @@ export function registerTerminalsRoutes(app: TerminalsRouteHost, core: Scope): v
|
|||
async (req, reply) => {
|
||||
try {
|
||||
const { session_id } = req.params;
|
||||
const terminal = await resolveTerminal(core, session_id).create(req.body);
|
||||
const terminal = await (await resolveTerminal(core, session_id)).create(req.body);
|
||||
reply.send(okEnvelope(terminal, req.id));
|
||||
} catch (err) {
|
||||
sendMappedError(reply, req.id, err);
|
||||
|
|
@ -167,7 +168,7 @@ export function registerTerminalsRoutes(app: TerminalsRouteHost, core: Scope): v
|
|||
async (req, reply) => {
|
||||
try {
|
||||
const { session_id, terminal_id } = req.params;
|
||||
const terminal = await resolveTerminal(core, session_id).get(terminal_id);
|
||||
const terminal = await (await resolveTerminal(core, session_id)).get(terminal_id);
|
||||
reply.send(okEnvelope(terminal, req.id));
|
||||
} catch (err) {
|
||||
sendMappedError(reply, req.id, err);
|
||||
|
|
@ -209,7 +210,7 @@ export function registerTerminalsRoutes(app: TerminalsRouteHost, core: Scope): v
|
|||
reply.send(errEnvelope(ErrorCode.VALIDATION_FAILED, message, req.id));
|
||||
return;
|
||||
}
|
||||
const result = await resolveTerminal(core, session_id).close(parsed.id);
|
||||
const result = await (await resolveTerminal(core, session_id)).close(parsed.id);
|
||||
reply.send(okEnvelope(result, req.id));
|
||||
} catch (err) {
|
||||
sendMappedError(reply, req.id, err);
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ export type { ISnapshotReader } from './snapshot';
|
|||
export { SnapshotNotFoundError, SnapshotTimeoutError } from './snapshot';
|
||||
export {
|
||||
SnapshotReader,
|
||||
reduceContextRecords,
|
||||
readWireRecords,
|
||||
type SnapshotReaderDeps,
|
||||
type SnapshotReaderLogger,
|
||||
|
|
|
|||
|
|
@ -5,11 +5,13 @@
|
|||
* Reads `<homeDir>/sessions/<workspaceId>/<sid>/state.json` and
|
||||
* `…/agents/main/wire.jsonl` directly, bypassing
|
||||
* `ISessionLifecycleService.resume` (DI-scope materialization, MCP connect,
|
||||
* full wire replay). The transcript is reduced from the folded `context.*`
|
||||
* records with the exact `contextOps` apply semantics, so the result is
|
||||
* byte-identical to the live `IAgentContextMemoryService.get()` view used by
|
||||
* the `legacy` (resume) path. `(size, mtimeMs)` transcript cache and the
|
||||
* watermark both come from in-memory state, keeping warm reads sub-ms.
|
||||
* full wire replay). The transcript is reduced from the `context.*` records
|
||||
* with `reduceContextTranscript`, which mirrors the live reducers EXCEPT that
|
||||
* `context.apply_compaction` keeps the full history and appends a summary
|
||||
* marker instead of dropping the compacted prefix — the same full-transcript
|
||||
* view v1 serves (so compacted-away assistant replies stay visible after a
|
||||
* later undo). `(size, mtimeMs)` transcript cache and the watermark both come
|
||||
* from in-memory state, keeping warm reads sub-ms.
|
||||
*
|
||||
* Pending approvals/questions, the live status, and `current_prompt_id` are
|
||||
* only available while the session is live; for a cold session they correctly
|
||||
|
|
@ -20,9 +22,6 @@ import { readFile, stat as fsStat } from 'node:fs/promises';
|
|||
import { join } from 'node:path';
|
||||
|
||||
import {
|
||||
computeUndoCut,
|
||||
foldAppendMessage,
|
||||
foldLoopEvent,
|
||||
IAgentLifecycleService,
|
||||
IAgentPromptLegacyService,
|
||||
ISessionActivity,
|
||||
|
|
@ -31,11 +30,9 @@ import {
|
|||
ISessionLifecycleService,
|
||||
IWorkspaceRegistry,
|
||||
normalizeSessionMeta,
|
||||
applyContextCompactionRecord,
|
||||
resetFold,
|
||||
reduceContextTranscript,
|
||||
toProtocolMessage,
|
||||
type ContextMessage,
|
||||
type LoopRecordedEvent,
|
||||
type Scope,
|
||||
type SessionMeta,
|
||||
} from '@moonshot-ai/agent-core-v2';
|
||||
|
|
@ -211,7 +208,7 @@ export class SnapshotReader implements ISnapshotReader {
|
|||
if (cached !== undefined) this.transcriptCache.delete(sid);
|
||||
|
||||
const records = await readWireRecords(wirePath);
|
||||
const messages = reduceContextRecords(records);
|
||||
const messages = [...reduceContextTranscript(records).entries];
|
||||
this.transcriptCache.set(sid, { size: info.size, mtimeMs: info.mtimeMs, messages });
|
||||
while (this.transcriptCache.size > this.deps.config.cacheLimit) {
|
||||
const oldest = this.transcriptCache.keys().next().value;
|
||||
|
|
@ -283,61 +280,6 @@ interface ContextRecord {
|
|||
readonly [key: string]: unknown;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reduce folded `context.*` wire records into the live `ContextMessage[]` view,
|
||||
* routing through the exact `contextOps.ts` reducers (`foldAppendMessage` /
|
||||
* `foldLoopEvent` / `resetFold`) so the result is byte-identical to the live
|
||||
* `IAgentContextMemoryService.get()` view:
|
||||
* append_message → push (deferred while a tool exchange is open) ·
|
||||
* append_loop_event → v1 fold (step/content/tool) · splice → array splice ·
|
||||
* clear → drop all · apply_compaction → `[summary, ...state.slice(compactedCount)]` ·
|
||||
* undo → trailing cut.
|
||||
*/
|
||||
export function reduceContextRecords(records: Iterable<ContextRecord>): ContextMessage[] {
|
||||
let state: readonly ContextMessage[] = [];
|
||||
for (const record of records) {
|
||||
switch (record.type) {
|
||||
case 'context.append_message': {
|
||||
state = foldAppendMessage(state, record['message'] as ContextMessage);
|
||||
break;
|
||||
}
|
||||
case 'context.append_loop_event': {
|
||||
state = foldLoopEvent(state, record['event'] as LoopRecordedEvent);
|
||||
break;
|
||||
}
|
||||
case 'context.splice': {
|
||||
const start = record['start'] as number;
|
||||
const deleteCount = record['deleteCount'] as number;
|
||||
const messages = record['messages'] as readonly ContextMessage[];
|
||||
if (deleteCount === 0 && messages.length === 0) break;
|
||||
const next = state.slice();
|
||||
next.splice(start, deleteCount, ...messages);
|
||||
state = resetFold(next);
|
||||
break;
|
||||
}
|
||||
case 'context.clear': {
|
||||
if (state.length !== 0) state = resetFold([]);
|
||||
break;
|
||||
}
|
||||
case 'context.apply_compaction': {
|
||||
state = applyContextCompactionRecord(state, record);
|
||||
break;
|
||||
}
|
||||
case 'context.undo': {
|
||||
const count = record['count'] as number;
|
||||
if (count <= 0 || state.length === 0) break;
|
||||
const { cutIndex, removedCount } = computeUndoCut(state, count);
|
||||
if (cutIndex < 0 || removedCount < count) break;
|
||||
state = resetFold(state.slice(0, cutIndex));
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
return state as ContextMessage[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a `wire.jsonl` file. A torn final line (crash mid-flush) is dropped;
|
||||
* corruption anywhere else throws so the route surfaces 50001. The leading
|
||||
|
|
|
|||
|
|
@ -302,10 +302,11 @@ describe('server-v2 /api/v1/sessions/{sid}/messages', () => {
|
|||
]);
|
||||
await agent.accessor.get(IAgentWireRecordService).flush();
|
||||
|
||||
// Live (post-compaction) context exposes only the folded summary; capture
|
||||
// its id so we can assert it survives the restart unchanged.
|
||||
// The live read already serves the full transcript (pre-compaction prefix
|
||||
// + summary), matching v1's `/messages`. Capture the summary id so we can
|
||||
// assert it survives the restart unchanged.
|
||||
const livePage = await getJson<PageWire>(`/api/v1/sessions/${id}/messages?page_size=100`);
|
||||
expect(livePage.body.data.items).toHaveLength(1);
|
||||
expect(livePage.body.data.items).toHaveLength(4);
|
||||
const liveSummaryId = livePage.body.data.items[0]!.id;
|
||||
|
||||
// Restart the server on the same homeDir → the session is cold for the next
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import { join } from 'node:path';
|
|||
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
|
||||
|
||||
import { IEventService } from '@moonshot-ai/agent-core-v2';
|
||||
import { IEventService, ISessionLifecycleService } from '@moonshot-ai/agent-core-v2';
|
||||
import { sessionWarningsResponseSchema } from '@moonshot-ai/protocol';
|
||||
|
||||
import { type RunningServer, startServer } from '../src/start';
|
||||
|
|
@ -323,6 +323,24 @@ describe('server-v2 /api/v1/sessions', () => {
|
|||
expect(got.body.data.archived).toBe(true);
|
||||
});
|
||||
|
||||
it('cold-loads a persisted session on :undo instead of 40401', async () => {
|
||||
const cwd = home as string;
|
||||
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
|
||||
const id = created.body.data.id;
|
||||
|
||||
// Drop the live handle so the session is persisted-but-cold (index + disk
|
||||
// only) — the state right after opening a session in the web UI before any
|
||||
// prompt has been sent. Before the fix, `:undo` resolved the main agent via
|
||||
// `lifecycle.get` (memory only) and reported 40401 "session does not exist".
|
||||
await (server as RunningServer).core.accessor.get(ISessionLifecycleService).close(id);
|
||||
|
||||
const res = await postJson<{ messages: unknown }>(`/api/v1/sessions/${id}:undo`, { count: 1 });
|
||||
// Cold-loaded successfully: the empty history yields "nothing to undo"
|
||||
// (40911), not the pre-fix "session does not exist" (40401).
|
||||
expect(res.body.code).toBe(40911);
|
||||
expect(res.body.msg).toMatch(/nothing to undo/i);
|
||||
});
|
||||
|
||||
it('rejects an unsupported action suffix (40001)', async () => {
|
||||
const cwd = home as string;
|
||||
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
|
||||
|
|
|
|||
|
|
@ -146,16 +146,18 @@ describe('server-v2 /api/v1 skills', () => {
|
|||
expect(body.msg).toMatch(/does not exist/);
|
||||
});
|
||||
|
||||
it('returns 40401 with an activation hint for a persisted but not activated session', async () => {
|
||||
it('cold-loads a persisted but not live (archived) session and lists skills', async () => {
|
||||
const id = await createSession();
|
||||
// Archiving removes the session from the live map but keeps it in the index.
|
||||
const archived = await postJson<{ archived: boolean }>(`/api/v1/sessions/${id}:archive`);
|
||||
expect(archived.body.code).toBe(0);
|
||||
|
||||
const { body } = await getJson<null>(`/api/v1/sessions/${id}/skills`);
|
||||
expect(body.code).toBe(40401);
|
||||
expect(body.msg).toMatch(/not activated/);
|
||||
expect(body.msg).toMatch(/activate it first/);
|
||||
// Cold-loaded on demand (matches v1's `resumeSession` in SkillService):
|
||||
// listing skills succeeds instead of reporting "not activated".
|
||||
const { body } = await getJson<{ skills: SkillWire[] }>(`/api/v1/sessions/${id}/skills`);
|
||||
expect(body.code).toBe(0);
|
||||
const skills = listSkillsResponseSchema.parse(body.data).skills;
|
||||
expect(skills.some((s) => s.name === 'update-config')).toBe(true);
|
||||
});
|
||||
|
||||
it('lists builtin skills projected to the wire shape', async () => {
|
||||
|
|
|
|||
|
|
@ -24,7 +24,6 @@ import { afterEach, describe, expect, it } from 'vitest';
|
|||
|
||||
import {
|
||||
loadSnapshotConfig,
|
||||
reduceContextRecords,
|
||||
readWireRecords,
|
||||
SnapshotNotFoundError,
|
||||
SnapshotReader,
|
||||
|
|
@ -242,7 +241,7 @@ describe('SnapshotReader.read', () => {
|
|||
expect((tool.content[0] as { tool_call_id: string }).tool_call_id).toBe('call_1');
|
||||
});
|
||||
|
||||
it('reduces context.apply_compaction to [summary, ...tail]', async () => {
|
||||
it('keeps the full history across context.apply_compaction and appends a summary marker', async () => {
|
||||
const f = await makeFixtureAsync();
|
||||
await seedSession(f, 'sess_compact');
|
||||
await writeWire(f.sessionDir('sess_compact'), [
|
||||
|
|
@ -257,10 +256,11 @@ describe('SnapshotReader.read', () => {
|
|||
]);
|
||||
const snap = await f.reader.read('sess_compact');
|
||||
const texts = snap.messages.items.map((m) => (m.content[0] as { text: string }).text);
|
||||
expect(texts).toEqual(['summary', 'after']);
|
||||
expect(texts).toEqual(['old-1', 'old-2', 'summary', 'after']);
|
||||
expect(snap.messages.items[2]?.metadata).toEqual({ origin: { kind: 'compaction_summary' } });
|
||||
});
|
||||
|
||||
it('reduces v1-shaped string summary compaction records', async () => {
|
||||
it('keeps the full history across v1-shaped string summary compaction records', async () => {
|
||||
const f = await makeFixtureAsync();
|
||||
await seedSession(f, 'sess_compact_v1');
|
||||
await writeWire(f.sessionDir('sess_compact_v1'), [
|
||||
|
|
@ -278,11 +278,11 @@ describe('SnapshotReader.read', () => {
|
|||
const snap = await f.reader.read('sess_compact_v1');
|
||||
const messages = snap.messages.items;
|
||||
const texts = messages.map((m) => (m.content[0] as { text: string }).text);
|
||||
expect(texts).toEqual(['summary', 'after']);
|
||||
expect(messages[0]?.metadata).toEqual({ origin: { kind: 'compaction_summary' } });
|
||||
expect(texts).toEqual(['old-1', 'old-2', 'summary', 'after']);
|
||||
expect(messages[2]?.metadata).toEqual({ origin: { kind: 'compaction_summary' } });
|
||||
});
|
||||
|
||||
it('reduces new compaction records to kept user messages followed by contextSummary', async () => {
|
||||
it('keeps compacted-away assistant messages and uses the raw summary as the marker', async () => {
|
||||
const f = await makeFixtureAsync();
|
||||
await seedSession(f, 'sess_compact_kept_users');
|
||||
await writeWire(f.sessionDir('sess_compact_kept_users'), [
|
||||
|
|
@ -309,12 +309,49 @@ describe('SnapshotReader.read', () => {
|
|||
const snap = await f.reader.read('sess_compact_kept_users');
|
||||
const messages = snap.messages.items;
|
||||
const texts = messages.map((m) => (m.content[0] as { text: string }).text);
|
||||
expect(messages.map((m) => m.role)).toEqual(['user', 'user', 'user']);
|
||||
expect(texts).toEqual(['old user', 'recent user', 'model-facing summary']);
|
||||
expect(messages[2]?.metadata).toEqual({ origin: { kind: 'compaction_summary' } });
|
||||
expect(messages.map((m) => m.role)).toEqual(['user', 'assistant', 'user', 'user']);
|
||||
expect(texts).toEqual(['old user', 'old assistant', 'recent user', 'raw summary']);
|
||||
expect(messages[3]?.metadata).toEqual({ origin: { kind: 'compaction_summary' } });
|
||||
});
|
||||
|
||||
it('honors context.clear and context.undo', async () => {
|
||||
it('preserves the pre-compaction assistant reply after a later undo', async () => {
|
||||
// Regression: send A, /compact, send B, undo. The snapshot must still show
|
||||
// A's assistant reply (compaction folds only the live context; the
|
||||
// transcript keeps the full history).
|
||||
const f = await makeFixtureAsync();
|
||||
await seedSession(f, 'sess_compact_undo');
|
||||
const assistant = (text: string): ContextMessage => ({
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text }],
|
||||
toolCalls: [],
|
||||
});
|
||||
await writeWire(f.sessionDir('sess_compact_undo'), [
|
||||
{ type: 'context.append_message', message: userMessage('message A') },
|
||||
{ type: 'context.append_message', message: assistant('reply A') },
|
||||
{
|
||||
type: 'context.apply_compaction',
|
||||
summary: 'summary text',
|
||||
contextSummary: 'model-facing summary',
|
||||
compactedCount: 2,
|
||||
tokensBefore: 100,
|
||||
tokensAfter: 20,
|
||||
keptUserMessageCount: 1,
|
||||
},
|
||||
{ type: 'context.append_message', message: userMessage('message B') },
|
||||
{ type: 'context.append_message', message: assistant('reply B') },
|
||||
{ type: 'context.undo', count: 1 },
|
||||
]);
|
||||
const snap = await f.reader.read('sess_compact_undo');
|
||||
const messages = snap.messages.items;
|
||||
expect(messages.map((m) => m.role)).toEqual(['user', 'assistant', 'user']);
|
||||
expect(messages.map((m) => (m.content[0] as { text: string }).text)).toEqual([
|
||||
'message A',
|
||||
'reply A',
|
||||
'summary text',
|
||||
]);
|
||||
});
|
||||
|
||||
it('keeps pre-clear messages in the transcript and lets undo remove the tail', async () => {
|
||||
const f = await makeFixtureAsync();
|
||||
await seedSession(f, 'sess_ops');
|
||||
await writeWire(f.sessionDir('sess_ops'), [
|
||||
|
|
@ -322,8 +359,10 @@ describe('SnapshotReader.read', () => {
|
|||
{ type: 'context.append_message', message: userMessage('b') },
|
||||
{ type: 'context.clear' },
|
||||
{ type: 'context.append_message', message: userMessage('c') },
|
||||
{ type: 'context.undo', count: 1 },
|
||||
]);
|
||||
expect((await f.reader.read('sess_ops')).messages.items.map((m) => (m.content[0] as { text: string }).text)).toEqual(['c']);
|
||||
// /clear keeps prior messages for display; undo removes the post-clear tail (c).
|
||||
expect((await f.reader.read('sess_ops')).messages.items.map((m) => (m.content[0] as { text: string }).text)).toEqual(['a', 'b']);
|
||||
});
|
||||
|
||||
it('caps the page at 100 and flags has_more', async () => {
|
||||
|
|
@ -395,95 +434,6 @@ describe('SnapshotReader.read', () => {
|
|||
});
|
||||
});
|
||||
|
||||
// ─── pure helpers ─────────────────────────────────────────────────────────
|
||||
|
||||
describe('reduceContextRecords', () => {
|
||||
it('ignores unknown records and the metadata envelope', () => {
|
||||
const out = reduceContextRecords([
|
||||
{ type: 'metadata', protocol_version: '1.4' },
|
||||
{ type: 'turn.started', turnId: 1 },
|
||||
{ type: 'context.append_message', message: userMessage('x') },
|
||||
]);
|
||||
expect(out).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('applies context.splice', () => {
|
||||
const out = reduceContextRecords([
|
||||
{ type: 'context.append_message', message: userMessage('a') },
|
||||
{ type: 'context.splice', start: 1, deleteCount: 0, messages: [userMessage('b')] },
|
||||
]);
|
||||
expect(out).toHaveLength(2);
|
||||
expect((out[1]!.content[0] as { text: string }).text).toBe('b');
|
||||
});
|
||||
|
||||
it('folds context.append_loop_event (step/content/tool) into messages', () => {
|
||||
const out = reduceContextRecords([
|
||||
{ type: 'context.append_message', message: userMessage('q') },
|
||||
{ type: 'context.append_loop_event', event: { type: 'step.begin', uuid: 's1' } },
|
||||
{
|
||||
type: 'context.append_loop_event',
|
||||
event: { type: 'content.part', stepUuid: 's1', part: { type: 'text', text: 'hello' } },
|
||||
},
|
||||
{
|
||||
type: 'context.append_loop_event',
|
||||
event: {
|
||||
type: 'tool.call',
|
||||
stepUuid: 's1',
|
||||
toolCallId: 'call_1',
|
||||
name: 'Bash',
|
||||
args: { command: 'echo hi' },
|
||||
},
|
||||
},
|
||||
{
|
||||
type: 'context.append_loop_event',
|
||||
event: { type: 'tool.result', toolCallId: 'call_1', result: { output: 'hi' } },
|
||||
},
|
||||
{ type: 'context.append_loop_event', event: { type: 'step.end', uuid: 's1' } },
|
||||
]);
|
||||
expect(out.map((m) => m.role)).toEqual(['user', 'assistant', 'tool']);
|
||||
expect((out[1]!.content[0] as { text: string }).text).toBe('hello');
|
||||
expect(out[1]!.toolCalls).toHaveLength(1);
|
||||
expect(out[1]!.toolCalls[0]!.id).toBe('call_1');
|
||||
expect(out[1]!.partial).toBeUndefined();
|
||||
expect(out[2]!.toolCallId).toBe('call_1');
|
||||
});
|
||||
|
||||
it('defers context.append_message until the open tool exchange closes', () => {
|
||||
const out = reduceContextRecords([
|
||||
{ type: 'context.append_loop_event', event: { type: 'step.begin', uuid: 's1' } },
|
||||
{
|
||||
type: 'context.append_loop_event',
|
||||
event: { type: 'tool.call', stepUuid: 's1', toolCallId: 'call_1', name: 'Bash', args: {} },
|
||||
},
|
||||
// Arrives while the tool exchange is still open — must not land between
|
||||
// the assistant and its tool result.
|
||||
{ type: 'context.append_message', message: userMessage('injected') },
|
||||
{
|
||||
type: 'context.append_loop_event',
|
||||
event: { type: 'tool.result', toolCallId: 'call_1', result: { output: 'ok' } },
|
||||
},
|
||||
{ type: 'context.append_loop_event', event: { type: 'step.end', uuid: 's1' } },
|
||||
]);
|
||||
expect(out.map((m) => m.role)).toEqual(['assistant', 'tool', 'user']);
|
||||
expect((out[2]!.content[0] as { text: string }).text).toBe('injected');
|
||||
});
|
||||
|
||||
it('synthesizes an interrupted result for a tool call left open at a step boundary', () => {
|
||||
const out = reduceContextRecords([
|
||||
{ type: 'context.append_loop_event', event: { type: 'step.begin', uuid: 's1' } },
|
||||
{
|
||||
type: 'context.append_loop_event',
|
||||
event: { type: 'tool.call', stepUuid: 's1', toolCallId: 'call_1', name: 'Bash', args: {} },
|
||||
},
|
||||
// No tool.result before the next step begins — the dangling call is closed.
|
||||
{ type: 'context.append_loop_event', event: { type: 'step.begin', uuid: 's2' } },
|
||||
{ type: 'context.append_loop_event', event: { type: 'step.end', uuid: 's2' } },
|
||||
]);
|
||||
expect(out.map((m) => m.role)).toEqual(['assistant', 'tool', 'assistant']);
|
||||
expect(out[1]!.isError).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('readWireRecords', () => {
|
||||
it('drops a torn final line but throws on mid-file corruption', async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'kimi-wire-'));
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue