perf(serve): Restore large sessions selectively (#9055)

* feat(core): Add selective session restore projection

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* feat(serve): Use selective session restore

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(serve): Address selective restore feedback (#9055)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(serve): Keep Goal correction best effort

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(serve): Align selective restore lease gating

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

---------

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
jinye 2026-08-13 14:34:03 +00:00 committed by GitHub
parent 97ec96ec54
commit c0e649b53c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
44 changed files with 7454 additions and 1166 deletions

View file

@ -475,11 +475,13 @@ segments once:
malformed-context, turn-reentry, and truncation decisions without retaining
evidence content. Add only the selected evidence UUIDs to the union, then feed
their materialized records to the shared accumulator and retain the resulting
window in the projection. This two-stage selection must preserve both the
existing production helper's result and its fail-closed errors; it must not
select every active record, perform a second scan, or copy Goal precedence.
Deferred Goal activation consumes that window instead of reading the
transcript again.
window in the projection. This two-stage selection must preserve the existing
production helper's valid result. When its evidence source is unavailable or
invalid, omit the projected window so deferred Goal activation falls back to
the existing runtime path and its established degradation behavior instead of
rejecting the whole session restore. It must not select every active record,
perform a second scan, or copy Goal precedence. Deferred Goal activation
consumes a valid projected window instead of reading the transcript again.
5. **File history.** Read every active `file_history_snapshot` record in
chronological order and feed each batch through the existing whole-batch
deserializer. This preserves today's behavior where one malformed item skips

View file

@ -4273,6 +4273,7 @@ describe('createAcpSessionBridge', () => {
keep: 'state',
'qwen.session.loadReplay': {
v: 1,
anchorRecordId: 'record-anchor',
hasMore: true,
partial: true,
replayError: 'replay boom',
@ -4317,6 +4318,7 @@ describe('createAcpSessionBridge', () => {
expect(loaded.partial).toBe(true);
expect(loaded.replayError).toBe('replay boom');
expect(loaded.historyHasMore).toBe(true);
expect(loaded.historyAnchorRecordId).toBe('record-anchor');
expect(loaded.lastEventId).toBe(2);
expect(loaded.compactedReplay).toHaveLength(2);
expect(loaded.liveJournal).toEqual([]);

View file

@ -137,6 +137,7 @@ import {
DAEMON_PROMPT_DISPLAY_TEXT_META_KEY,
LOAD_REPLAY_BULK_MODE,
LOAD_REPLAY_HIDE_INHERITED_META_KEY,
LOAD_REPLAY_MAX_UPDATES,
LOAD_REPLAY_META_KEY,
LOAD_REPLAY_MODE_META_KEY,
LOAD_REPLAY_PAGE_SIZE_META_KEY,
@ -252,7 +253,6 @@ const KNOWN_SESSION_UPDATE_TYPES = new Set([
'session_info_update',
'usage_update',
]);
const MAX_BULK_REPLAY_UPDATES = 10_000;
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value);
@ -661,6 +661,7 @@ function describeLoadReplayValue(value: unknown): string {
function extractLoadReplayResponse(state: BridgeSessionState): {
state: BridgeSessionState;
updates: SessionUpdate[];
anchorRecordId?: string;
partial?: true;
replayError?: string;
hasMore?: boolean;
@ -682,10 +683,10 @@ function extractLoadReplayResponse(state: BridgeSessionState): {
`(version=${LOAD_REPLAY_VERSION}, count=not-array)`,
);
}
if (rawUpdates.length > MAX_BULK_REPLAY_UPDATES) {
if (rawUpdates.length > LOAD_REPLAY_MAX_UPDATES) {
throw new Error(
`qwen.session.loadReplay updates exceed limit ` +
`(${rawUpdates.length} > ${MAX_BULK_REPLAY_UPDATES})`,
`(${rawUpdates.length} > ${LOAD_REPLAY_MAX_UPDATES})`,
);
}
const partial = replay['partial'];
@ -709,6 +710,13 @@ function extractLoadReplayResponse(state: BridgeSessionState): {
`(version=${LOAD_REPLAY_VERSION}, hasMore=${describeLoadReplayValue(hasMore)})`,
);
}
const anchorRecordId = replay['anchorRecordId'];
if (anchorRecordId !== undefined && typeof anchorRecordId !== 'string') {
throw new Error(
`Invalid qwen.session.loadReplay anchorRecordId ` +
`(version=${LOAD_REPLAY_VERSION}, anchorRecordId=${describeLoadReplayValue(anchorRecordId)})`,
);
}
const invalidUpdateIndex = rawUpdates.findIndex(
(update) => !isBulkReplayUpdate(update),
);
@ -735,6 +743,7 @@ function extractLoadReplayResponse(state: BridgeSessionState): {
return {
state: cleanState,
updates: rawUpdates,
...(typeof anchorRecordId === 'string' ? { anchorRecordId } : {}),
...(partial === true ? { partial: true as const } : {}),
...(typeof replayError === 'string' ? { replayError } : {}),
...(hasMore === true ? { hasMore: true } : {}),
@ -1130,6 +1139,7 @@ interface SessionEntry {
restoreReplayPartial?: true;
restoreReplayError?: string;
restoreHistoryHasMore?: true;
restoreHistoryAnchorRecordId?: string;
/**
* Most recent heartbeat across any client on this session (Date.now()
* epoch ms). Set on every `recordHeartbeat` call regardless of whether
@ -5379,6 +5389,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
| 'restoreReplayPartial'
| 'restoreReplayError'
| 'restoreHistoryHasMore'
| 'restoreHistoryAnchorRecordId'
| 'activePromptId'
>,
action: 'load' | 'resume',
@ -5392,6 +5403,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
| 'partial'
| 'replayError'
| 'historyHasMore'
| 'historyAnchorRecordId'
> => {
const replayStatus =
action === 'load' && entry.restoreReplayPartial === true
@ -5412,6 +5424,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
lastEventId: entry.events.lastEventId,
eventEpoch,
...replayStatus,
...(action === 'load' &&
entry.restoreHistoryAnchorRecordId !== undefined
? { historyAnchorRecordId: entry.restoreHistoryAnchorRecordId }
: {}),
};
}
if (action === 'load') {
@ -5437,6 +5453,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
...(entry.restoreHistoryHasMore === true
? { historyHasMore: true }
: {}),
...(entry.restoreHistoryAnchorRecordId !== undefined
? { historyAnchorRecordId: entry.restoreHistoryAnchorRecordId }
: {}),
};
}
return { lastEventId: snapshot.lastEventId, eventEpoch, ...replayStatus };
@ -6108,6 +6127,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
let replayPartial: true | undefined;
let replayError: string | undefined;
let replayHasMore: true | undefined;
let replayAnchorRecordId: string | undefined;
try {
const rawRestore = telemetry.withSpan(
'session.restore',
@ -6232,6 +6252,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
replayPartial = extracted.partial;
replayError = extracted.replayError;
replayHasMore = extracted.hasMore === true ? true : undefined;
replayAnchorRecordId = extracted.anchorRecordId;
}
} catch (err) {
if (err instanceof SessionRestoreTimeoutError) throw err;
@ -6353,6 +6374,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
if (replayHasMore === true) {
entry.restoreHistoryHasMore = true;
}
if (replayAnchorRecordId !== undefined) {
entry.restoreHistoryAnchorRecordId = replayAnchorRecordId;
}
seedSnapshotCaches(entry, publicState);
const artifactRestoreWarnings = await entry.artifacts.restore(
restoredArtifactSnapshot,

View file

@ -188,6 +188,8 @@ export const LOAD_REPLAY_HIDE_INHERITED_META_KEY =
'qwen.session.loadReplayHideInherited';
export const LOAD_REPLAY_BULK_MODE = 'bulk';
export const LOAD_REPLAY_VERSION = 1 as const;
export const LOAD_REPLAY_MAX_BYTES = 32 * 1024 * 1024;
export const LOAD_REPLAY_MAX_UPDATES = 10_000;
export const REQUESTED_SESSION_ID_META_KEY = 'qwen-code/sessionId';
@ -338,6 +340,7 @@ export interface ChannelStartupProfileV1 {
export interface BridgeLoadReplayEnvelope {
v: typeof LOAD_REPLAY_VERSION;
updates: SessionUpdate[];
anchorRecordId?: string;
hasMore?: boolean;
partial?: true;
replayError?: string;

View file

@ -109,6 +109,7 @@ type TestableAcpBridge = AcpBridge & {
extMethod: ReturnType<typeof vi.fn>;
newSession?: ReturnType<typeof vi.fn>;
loadSession?: ReturnType<typeof vi.fn>;
unstable_resumeSession?: ReturnType<typeof vi.fn>;
prompt?: ReturnType<typeof vi.fn>;
};
knownSessionIds: Set<string>;
@ -418,6 +419,34 @@ describe('AcpBridge', () => {
expect(extMethod).toHaveBeenCalledOnce();
});
it('restores channel sessions through resume without replaying history', async () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
cwd: '/tmp',
}) as unknown as TestableAcpBridge;
const resumeSession = vi.fn().mockResolvedValue({});
bridge.child = { killed: false, exitCode: null };
bridge.connection = {
extMethod: vi.fn(),
unstable_resumeSession: resumeSession,
} as TestableAcpBridge['connection'];
const bindingToken = {};
await expect(
bridge.loadSession('restored-session', '/tmp', undefined, bindingToken),
).resolves.toBe('restored-session');
expect(resumeSession).toHaveBeenCalledWith({
sessionId: 'restored-session',
cwd: '/tmp',
mcpServers: [],
});
expect(bridge.knownSessionIds.has('restored-session')).toBe(true);
expect(bridge.sessionBindingTokens.get('restored-session')).toBe(
bindingToken,
);
});
it('returns only the final turn text after tool calls', async () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',

View file

@ -247,7 +247,7 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
): Promise<string> {
const conn = this.ensureConnection();
await this.registerChannelLoopMcpServer();
await conn.loadSession({
await conn.unstable_resumeSession({
sessionId,
cwd,
mcpServers: [],

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

View file

@ -769,6 +769,7 @@ describe('Session', () => {
getGeminiClient: vi.fn().mockReturnValue(mockGeminiClient),
getGoalRuntime: vi.fn().mockReturnValue(mockGoalRuntime),
getGoalRuntimeReady: vi.fn().mockResolvedValue(mockGoalRuntime),
getGoalRuntimePrepared: vi.fn().mockResolvedValue(mockGoalRuntime),
bindGoalTurnHost: vi.fn().mockImplementation((host) => {
boundGoalHost = host;
return () => {
@ -14583,6 +14584,64 @@ describe('Session', () => {
expect(mockClient.sessionUpdate).not.toHaveBeenCalled();
});
it('suppresses a hidden recovered Goal until a different Goal replaces it', async () => {
const listener = mockGoalRuntime.subscribe.mock.calls[0]?.[0] as (
snapshot: core.GoalSnapshotV2,
cause?: core.GoalStateCause,
) => void;
session.primeRecoveredGoalPublication(undefined, 'goal-hidden');
const hidden: core.GoalSnapshotV2 = {
v: 2,
activity: 'idle',
goal: {
...migratedSnapshot.goal!,
goalId: 'goal-hidden',
revision: 1,
objective: 'hidden inherited goal',
status: 'active',
},
};
listener(hidden, 'create');
listener({ ...hidden, activity: 'running' });
await new Promise((resolve) => setTimeout(resolve, 0));
expect(mockClient.sessionUpdate).not.toHaveBeenCalled();
const progressed = {
...hidden,
activity: 'idle' as const,
goal: {
...hidden.goal!,
revision: 2,
objective: 'still hidden',
},
};
listener(progressed, 'edit');
await new Promise((resolve) => setTimeout(resolve, 0));
expect(mockClient.sessionUpdate).not.toHaveBeenCalled();
const replacement = {
...progressed,
goal: {
...progressed.goal!,
goalId: 'goal-visible',
revision: 1,
objective: 'visible replacement',
},
};
listener(replacement, 'replace');
await vi.waitFor(() =>
expect(mockClient.sessionUpdate).toHaveBeenCalledOnce(),
);
expect(mockClient.sessionUpdate).toHaveBeenCalledWith({
sessionId: 'test-session-id',
update: expect.objectContaining({
_meta: expect.objectContaining({ goalState: replacement }),
}),
});
});
it('returns nothing when no Goal was recovered', async () => {
mockGoalRuntime.getRecoveryCause.mockReturnValue(undefined);
expect(await session.renderRecoveredGoalUpdates([])).toEqual([]);

View file

@ -182,6 +182,8 @@ import {
runWithInvocationContext,
truncateNotificationLabel,
buildBackgroundEntryLabel,
collectSessionTurnState,
computeInitialTurnFromHistory as computeInitialTurnFromHistoryCore,
} from '@qwen-code/qwen-code-core';
import { NOT_CURRENTLY_GENERATING_CANCEL_MESSAGE } from '@qwen-code/acp-bridge/bridgeErrors';
import { CHANNEL_PROMPT_META_KEY } from '@qwen-code/channel-base';
@ -297,12 +299,12 @@ import { projectAcpToolResultUpdate } from './acp-tool-result-text-projection.js
import { ToolCallEmitter } from './emitters/tool-call-emitter.js';
import { ToolCallPreparationTracker } from './tool-call-preparation-tracker.js';
import { PlanEmitter } from './emitters/PlanEmitter.js';
import {
MessageEmitter,
buildGoalStateUpdate,
buildGoalStatusUpdate,
} from './emitters/MessageEmitter.js';
import { MessageEmitter } from './emitters/MessageEmitter.js';
import type { HistoryItemGoalStatus } from '../../ui/types.js';
import {
goalPublicationKey,
renderPreparedGoalUpdate,
} from './recovered-goal-update.js';
import { SubAgentTracker } from './SubAgentTracker.js';
import {
buildPermissionRequestContent,
@ -1271,30 +1273,7 @@ export function computeInitialTurnFromHistory(
records: ChatRecord[],
sessionId: string,
): number {
let maxPromptTurn = 0;
let userMessageCount = 0;
const promptIdPrefix = `${sessionId}########`;
for (const record of records) {
if (record.sessionId === sessionId && isUserPromptRecord(record)) {
userMessageCount += 1;
}
for (const promptId of getRecordPromptIds(record)) {
if (!promptId.startsWith(promptIdPrefix)) {
continue;
}
const suffix = promptId.slice(promptIdPrefix.length);
if (!/^\d+$/.test(suffix)) {
continue;
}
maxPromptTurn = Math.max(maxPromptTurn, Number(suffix));
}
}
return maxPromptTurn > 0 ? maxPromptTurn : userMessageCount;
return computeInitialTurnFromHistoryCore(records, sessionId);
}
export async function fireSessionPermissionDeniedForAutoMode(
@ -1329,42 +1308,6 @@ export async function fireSessionPermissionDeniedForAutoMode(
}
}
function getRecordPromptIds(record: ChatRecord): string[] {
const promptIds: string[] = [];
const recordPromptId = (record as { promptId?: unknown }).promptId;
if (typeof recordPromptId === 'string') {
promptIds.push(recordPromptId);
}
const telemetryPromptId = readTelemetryPromptId(record.systemPayload);
if (telemetryPromptId) {
promptIds.push(telemetryPromptId);
}
return promptIds;
}
function readTelemetryPromptId(payload: unknown): string | undefined {
if (!payload || typeof payload !== 'object' || !('uiEvent' in payload)) {
return undefined;
}
const uiEvent = (payload as { uiEvent?: unknown }).uiEvent;
if (!uiEvent || typeof uiEvent !== 'object' || !('prompt_id' in uiEvent)) {
return undefined;
}
const promptId = (uiEvent as { prompt_id?: unknown }).prompt_id;
return typeof promptId === 'string' ? promptId : undefined;
}
function isUserPromptRecord(record: ChatRecord): boolean {
if (record.type !== 'user' || record.subtype === 'realtime_message') {
return false;
}
return (
record.message?.parts?.some(
(part) => typeof part.text === 'string' && part.text.trim().length > 0,
) ?? false
);
}
const AT_TOKEN_RE = /@([^\s,;!?()[\]{}]+)/g;
function collectExtensionMentionRefs(
@ -1633,6 +1576,9 @@ export class Session implements SessionContext {
private goalRuntimeUnsubscribe?: () => void;
private lastGoalSnapshot?: GoalSnapshotV2;
private lastGoalPublicationKey?: string;
// Set only when runtime recovery selected a Goal that initial replay hid.
// Keep that Goal private through activation and later progress updates.
private suppressedRecoveredGoalId?: string;
private goalPublicationTail: Promise<void> = Promise.resolve();
// Set true in dispose(). Guards #drainCronQueue and #drainNotificationQueue
@ -1811,6 +1757,7 @@ export class Session implements SessionContext {
this.goalHostUnbind = undefined;
this.lastGoalSnapshot = undefined;
this.lastGoalPublicationKey = undefined;
this.suppressedRecoveredGoalId = undefined;
this.#bindGoalRuntime();
}
@ -1853,53 +1800,46 @@ export class Session implements SessionContext {
await this.#queueGoalState(runtime.getSnapshot(), cause);
}
/**
* Render the recovered-Goal cards instead of streaming them.
*
* The bulk load-replay path (`historyReplay: 'response'`) does not stream
* its replay: `loadSession` collects the page into the `LOAD_REPLAY`
* envelope and the bridge seeds those updates onto the session's event bus
* *after* the ACP `session/load` call returns. A card streamed from inside
* that call therefore lands on the bus **before** the replayed
* pre-migration `set` card the reverse of the ordering
* {@link publishRecoveredGoalState} exists to produce, leaving the phantom
* running goal exactly as it was. Returning the cards lets the caller
* append them to the envelope, after the replay page.
*
* Appending after a truncated page (`hasMore`) is still correct: paging
* drops the oldest records, so the authoritative state belongs last either
* way.
*
* Marks the publication as delivered, so the runtime subscription cannot
* emit a duplicate card for the same `(cause, snapshot)` once the session
* goes live.
*/
async renderRecoveredGoalUpdates(
replayedRecords?: readonly ChatRecord[],
): Promise<SessionUpdate[]> {
if (this.disposed || this.closing) return [];
let runtime;
try {
runtime = await this.config.getGoalRuntimeReady();
} catch (error) {
if (!(error instanceof GoalPersistenceUnavailableError)) throw error;
const status = this.#unrestorableGoalStatus(replayedRecords);
return status ? [buildGoalStatusUpdate(status)] : [];
const rendered = await renderPreparedGoalUpdate(
() => this.config.getGoalRuntimeReady(),
{
...(replayedRecords ? { replayedRecords } : {}),
previousGoal: this.lastGoalSnapshot?.goal ?? null,
},
);
if (
rendered.publicationKey &&
rendered.publicationKey === this.lastGoalPublicationKey
) {
return [];
}
const cause = runtime.getRecoveryCause?.();
// Nothing was recovered, so the replay already told the whole story.
if (!cause) return [];
const snapshot = runtime.getSnapshot();
const publicationKey = this.#goalPublicationKey(snapshot, cause);
if (publicationKey === this.lastGoalPublicationKey) return [];
this.lastGoalPublicationKey = publicationKey;
return [
buildGoalStateUpdate(
snapshot,
cause,
this.lastGoalSnapshot?.goal ?? null,
),
];
this.primeRecoveredGoalPublication(rendered.publicationKey);
return rendered.updates;
}
primeRecoveredGoalPublication(
publicationKey: string | undefined,
suppressedGoalId?: string,
): void {
if (publicationKey) this.lastGoalPublicationKey = publicationKey;
this.suppressedRecoveredGoalId = suppressedGoalId;
}
#suppressRecoveredGoalUpdate(snapshot: GoalSnapshotV2): boolean {
const suppressedGoalId = this.suppressedRecoveredGoalId;
if (!suppressedGoalId) return false;
const goal = snapshot.goal;
if (goal?.goalId === suppressedGoalId) return true;
if (goal === null) {
this.suppressedRecoveredGoalId = undefined;
return true;
}
this.suppressedRecoveredGoalId = undefined;
return false;
}
/**
@ -1940,19 +1880,13 @@ export class Session implements SessionContext {
};
}
#goalPublicationKey(
snapshot: GoalSnapshotV2,
cause?: GoalStateCause,
): string | undefined {
return cause ? `${cause}:${JSON.stringify(snapshot)}` : undefined;
}
async #publishGoalState(
snapshot: GoalSnapshotV2,
cause?: GoalStateCause,
previousGoal: GoalRecord | null = this.lastGoalSnapshot?.goal ?? null,
): Promise<void> {
const publicationKey = this.#goalPublicationKey(snapshot, cause);
if (this.#suppressRecoveredGoalUpdate(snapshot)) return;
const publicationKey = goalPublicationKey(snapshot, cause);
if (publicationKey && publicationKey === this.lastGoalPublicationKey) {
return;
}
@ -3148,30 +3082,34 @@ export class Session implements SessionContext {
* Delegates to HistoryReplayer for consistent event emission.
*/
primeTurnFromHistory(records: ChatRecord[]): void {
for (const record of records) {
if (record.subtype !== 'notification') continue;
const backgroundTask = (
record.systemPayload as
| { backgroundTask?: { taskId?: unknown } }
| undefined
)?.backgroundTask;
if (typeof backgroundTask?.taskId === 'string') {
this.persistedBackgroundNotificationTaskIds.add(backgroundTask.taskId);
}
}
this.turn = Math.max(
this.turn,
computeInitialTurnFromHistory(records, this.config.getSessionId()),
const turnState = collectSessionTurnState(
records,
this.config.getSessionId(),
);
this.primeTurnState(
turnState.initialTurn,
turnState.backgroundNotificationTaskIds,
);
}
primeTurnState(
initialTurn: number,
backgroundNotificationTaskIds: readonly string[],
): void {
for (const taskId of backgroundNotificationTaskIds) {
this.persistedBackgroundNotificationTaskIds.add(taskId);
}
this.turn = Math.max(this.turn, initialTurn);
}
async replayHistory(
records: ChatRecord[],
gaps?: HistoryGap[],
options?: Parameters<HistoryReplayer['replay']>[2],
): Promise<void> {
this.primeTurnFromHistory(records);
try {
await this.historyReplayer.replay(records, gaps);
await this.historyReplayer.replay(records, gaps, options);
} finally {
// Replayed plan updates re-stamp the revision via sendUpdate, but they
// belong to finished cycles; only live updates may bind the next

View file

@ -121,8 +121,13 @@ export class MessageEmitter extends BaseEmitter {
async emitGoalStatus(
status: Omit<HistoryItemGoalStatus, 'id' | 'type'>,
goalState?: unknown,
): Promise<void> {
await this.sendUpdate(buildGoalStatusUpdate(status));
const update = buildGoalStatusUpdate(status);
if (goalState) {
update._meta = { ...update._meta, goalState };
}
await this.sendUpdate(update);
}
async emitGoalState(

View file

@ -241,6 +241,37 @@ describe('history replay page', () => {
]);
});
it('fails incrementally before collecting an update above the count limit', async () => {
await expect(
collectHistoryReplayUpdates({
sessionId: SESSION_ID,
records: [userRecord()],
cumulativeUsage: createReplayCumulativeUsage(),
limits: { maxBytes: Number.MAX_SAFE_INTEGER, maxUpdates: 0 },
}),
).rejects.toMatchObject({
name: 'HistoryReplayLimitError',
reason: 'updates',
observed: 1,
limit: 0,
});
});
it('fails incrementally before retaining serialized updates above the byte limit', async () => {
await expect(
collectHistoryReplayUpdates({
sessionId: SESSION_ID,
records: [userRecord()],
cumulativeUsage: createReplayCumulativeUsage(),
limits: { maxBytes: 2, maxUpdates: 1 },
}),
).rejects.toMatchObject({
name: 'HistoryReplayLimitError',
reason: 'bytes',
limit: 2,
});
});
it('filters malformed replay state before encoding the next cursor', async () => {
const logger = { warn: vi.fn() };
const encodeCursor = vi.fn(() => 'next-cursor');

View file

@ -17,6 +17,7 @@ import {
} from '@qwen-code/qwen-code-core';
import type { SessionUpdate } from '@agentclientprotocol/sdk';
import type { TranscriptReplayStateV1 } from '@qwen-code/acp-bridge/transcriptReplay';
import { Buffer } from 'node:buffer';
import { projectAcpToolResultUpdate } from './acp-tool-result-text-projection.js';
import { HistoryReplayer } from './history-replayer.js';
import type { PendingReplayToolCall } from './history-replayer.js';
@ -26,6 +27,25 @@ interface ReplayLogger {
warn(message: string, ...args: unknown[]): void;
}
export class HistoryReplayLimitError extends Error {
constructor(
readonly sessionId: string,
readonly reason: 'bytes' | 'updates',
readonly observed: number,
readonly limit: number,
) {
super(
`Transcript replay for session ${sessionId} exceeds the ${reason} limit (${observed}, max ${limit})`,
);
this.name = 'HistoryReplayLimitError';
}
}
export interface HistoryReplayLimits {
maxBytes: number;
maxUpdates: number;
}
export function createReplayCumulativeUsage(): CumulativeUsage {
return {
promptTokens: 0,
@ -164,22 +184,46 @@ function replayContext(
updates: SessionUpdate[],
cumulativeUsage: CumulativeUsage,
config?: Config,
limits?: HistoryReplayLimits,
): SessionEmitterContext {
let activeRecordId: string | null = null;
let serializedUpdateBytes = 2;
return {
sessionId,
sendUpdate: async (update) => {
const projectedUpdate = projectAcpToolResultUpdate(update);
if (activeRecordId === null) {
updates.push(projectedUpdate);
return;
const updateWithRecordId = (() => {
if (activeRecordId === null) return projectedUpdate;
const record = projectedUpdate as unknown as Record<string, unknown>;
const meta = isObjectRecord(record['_meta']) ? record['_meta'] : {};
return {
...record,
_meta: { ...meta, 'qwen.session.recordId': activeRecordId },
} as unknown as SessionUpdate;
})();
if (limits) {
const updateCount = updates.length + 1;
if (updateCount > limits.maxUpdates) {
throw new HistoryReplayLimitError(
sessionId,
'updates',
updateCount,
limits.maxUpdates,
);
}
serializedUpdateBytes +=
(updates.length === 0 ? 0 : 1) +
Buffer.byteLength(JSON.stringify(updateWithRecordId), 'utf8');
if (serializedUpdateBytes > limits.maxBytes) {
throw new HistoryReplayLimitError(
sessionId,
'bytes',
serializedUpdateBytes,
limits.maxBytes,
);
}
}
const record = projectedUpdate as unknown as Record<string, unknown>;
const meta = isObjectRecord(record['_meta']) ? record['_meta'] : {};
updates.push({
...record,
_meta: { ...meta, 'qwen.session.recordId': activeRecordId },
} as unknown as SessionUpdate);
updates.push(updateWithRecordId);
},
setActiveRecordId: (recordId: string | null) => {
activeRecordId = recordId;
@ -196,6 +240,9 @@ export async function collectHistoryReplayUpdates({
gaps,
cumulativeUsage,
logger,
replayState,
goalBootstrap,
limits,
}: {
sessionId: string;
config?: Config;
@ -203,13 +250,22 @@ export async function collectHistoryReplayUpdates({
gaps?: HistoryGap[];
cumulativeUsage: CumulativeUsage;
logger?: ReplayLogger;
replayState?: unknown;
goalBootstrap?: import('./history-replayer.js').HistoryReplayGoalBootstrap;
limits?: HistoryReplayLimits;
}): Promise<{ updates: SessionUpdate[]; replayError?: string }> {
const updates: SessionUpdate[] = [];
try {
const initial = parseTranscriptReplayState(replayState, logger);
await new HistoryReplayer(
replayContext(sessionId, updates, cumulativeUsage, config),
).replay(records, gaps);
replayContext(sessionId, updates, cumulativeUsage, config, limits),
).replay(records, gaps, {
...(initial.goalState ? { initialGoalState: initial.goalState } : {}),
...(initial.goalCause ? { initialGoalCause: initial.goalCause } : {}),
...(goalBootstrap ? { goalBootstrap } : {}),
});
} catch (error) {
if (error instanceof HistoryReplayLimitError) throw error;
const replayError = error instanceof Error ? error.message : String(error);
logger?.warn(
'[historyReplay] History replay failed for session %s (partial updates: %d):',

View file

@ -10,6 +10,11 @@ import type {
GoalStateCause,
HistoryGap,
} from '@qwen-code/qwen-code-core';
import {
parseGoalSnapshotV2,
parseGoalStateCause,
projectGoalStateToLegacy,
} from '@qwen-code/qwen-code-core';
import {
createTranscriptReplayMachine,
MISSING_TRANSCRIPT_TOOL_RESULT_MESSAGE,
@ -49,6 +54,18 @@ export interface HistoryReplayPageState {
replay: TranscriptReplayStateV1;
}
export interface HistoryReplayGoalBootstrap {
goalStatus: {
kind: 'set' | 'checking';
condition: string;
iterations?: number;
setAt?: number;
durationMs?: number;
lastReason?: string;
};
goalState?: GoalSnapshotV2;
}
/**
* Handles replaying session history on session load.
*
@ -65,17 +82,65 @@ export class HistoryReplayer {
this.machine = this.createMachine();
}
async replay(records: ChatRecord[], gaps?: HistoryGap[]): Promise<void> {
async replay(
records: ChatRecord[],
gaps?: HistoryGap[],
options: {
initialGoalState?: GoalSnapshotV2;
initialGoalCause?: GoalStateCause;
goalBootstrap?: HistoryReplayGoalBootstrap;
} = {},
): Promise<void> {
try {
if (options.goalBootstrap) {
const update = {
sessionUpdate: 'agent_message_chunk' as const,
content: { type: 'text' as const, text: '' },
_meta: {
...(options.goalBootstrap.goalState
? { goalState: options.goalBootstrap.goalState }
: {}),
goalStatus: options.goalBootstrap.goalStatus,
},
};
await this.sendUpdate(update);
}
await this.replayPage(records, {
finalizeDangling: true,
gaps,
...(options.initialGoalState
? { goalState: options.initialGoalState }
: {}),
...(options.initialGoalCause
? { goalCause: options.initialGoalCause }
: {}),
});
} finally {
this.setActiveRecordId(null);
}
}
static v2GoalBootstrap(
rawGoalState: unknown,
rawGoalCause: unknown,
): HistoryReplayGoalBootstrap | undefined {
const goalState = parseGoalSnapshotV2(rawGoalState);
const goalCause = parseGoalStateCause(rawGoalCause);
if (!goalState?.goal || goalState.goal.status !== 'active' || !goalCause) {
return undefined;
}
const projection = projectGoalStateToLegacy({
v: 2,
cause: goalCause,
snapshot: goalState,
});
const { type: _type, kind, ...goalStatus } = projection.goalStatus;
if (kind !== 'set' && kind !== 'checking') {
return undefined;
}
return { goalStatus: { ...goalStatus, kind }, goalState };
}
async replayPage(
records: ChatRecord[],
options: HistoryReplayPageOptions = {},

View file

@ -0,0 +1,217 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { describe, expect, it, vi } from 'vitest';
import {
GoalPersistenceUnavailableError,
type GoalRuntime,
type GoalSnapshotV2,
} from '@qwen-code/qwen-code-core';
import { renderPreparedGoalUpdate } from './recovered-goal-update.js';
const hiddenSnapshot: GoalSnapshotV2 = {
v: 2,
activity: 'idle',
goal: {
goalId: 'hidden-goal',
revision: 1,
objective: 'hidden objective',
status: 'active',
evidenceCursor: { recordId: 'hidden-record' },
turnCount: 1,
activeTimeMs: 10,
createdAt: 1,
updatedAt: 2,
},
};
function runtime(): GoalRuntime {
return {
getSnapshot: vi.fn(() => hiddenSnapshot),
getRecoveryCause: vi.fn(() => 'create'),
} as unknown as GoalRuntime;
}
describe('renderPreparedGoalUpdate', () => {
it('renders the prepared runtime state for an ordinary load', async () => {
const result = await renderPreparedGoalUpdate(async () => runtime());
expect(result.publicationKey).toContain('hidden-goal');
expect(result.updates).toEqual([
expect.objectContaining({
_meta: expect.objectContaining({ goalState: hiddenSnapshot }),
}),
]);
});
it('does not duplicate the visible bootstrap for hidden-inherited history', async () => {
const bootstrap = {
goalStatus: { kind: 'set' as const, condition: 'visible objective' },
};
const result = await renderPreparedGoalUpdate(async () => runtime(), {
hideRuntimeGoal: true,
bootstrap,
});
expect(result.publicationKey).toContain('hidden-goal');
expect(result.suppressedGoalId).toBe('hidden-goal');
expect(result.updates).toEqual([]);
});
it('does not duplicate a v2 bootstrap that matches the runtime', async () => {
const result = await renderPreparedGoalUpdate(async () => runtime(), {
bootstrap: {
goalStatus: { kind: 'set', condition: 'hidden objective' },
goalState: hiddenSnapshot,
},
});
expect(result.updates).toEqual([]);
});
it('appends the runtime correction after a legacy bootstrap', async () => {
const result = await renderPreparedGoalUpdate(async () => runtime(), {
bootstrap: {
goalStatus: { kind: 'set', condition: 'hidden objective' },
},
});
expect(result.updates).toEqual([
expect.objectContaining({
_meta: expect.objectContaining({ goalState: hiddenSnapshot }),
}),
]);
});
it('clears a visible legacy bootstrap when recovery is unavailable', async () => {
const result = await renderPreparedGoalUpdate(
async () => {
throw new GoalPersistenceUnavailableError('unsupported record');
},
{
bootstrap: {
goalStatus: {
kind: 'checking',
condition: 'visible objective',
iterations: 2,
setAt: 123,
},
},
},
);
expect(result.updates).toEqual([
expect.objectContaining({
_meta: {
goalStatus: expect.objectContaining({
kind: 'cleared',
condition: 'visible objective',
iterations: 2,
setAt: 123,
}),
},
}),
]);
});
it('clears a replayed legacy Goal when recovery is unavailable', async () => {
const result = await renderPreparedGoalUpdate(
async () => {
throw new GoalPersistenceUnavailableError('unsupported record');
},
{
replayedRecords: [
{
uuid: 'goal-result',
parentUuid: null,
sessionId: 'session-1',
timestamp: new Date(0).toISOString(),
type: 'system',
subtype: 'slash_command',
cwd: '/tmp',
version: 'test',
systemPayload: {
phase: 'result',
rawCommand: '/goal',
outputHistoryItems: [
{
type: 'goal_status',
kind: 'set',
condition: 'replayed objective',
iterations: 3,
setAt: 456,
},
],
},
},
],
},
);
expect(result.updates).toEqual([
expect.objectContaining({
_meta: {
goalStatus: expect.objectContaining({
kind: 'cleared',
condition: 'replayed objective',
iterations: 3,
setAt: 456,
}),
},
}),
]);
});
it('falls back to a page-out bootstrap when replay has no Goal card', async () => {
const result = await renderPreparedGoalUpdate(
async () => {
throw new GoalPersistenceUnavailableError('unsupported record');
},
{
replayedRecords: [
{
uuid: 'user-1',
parentUuid: null,
sessionId: 'session-1',
timestamp: new Date(0).toISOString(),
type: 'user',
cwd: '/tmp',
version: 'test',
message: { role: 'user', parts: [{ text: 'continue' }] },
},
],
bootstrap: {
goalStatus: {
kind: 'set',
condition: 'page-out objective',
iterations: 1,
},
},
},
);
expect(result.updates).toEqual([
expect.objectContaining({
_meta: {
goalStatus: expect.objectContaining({
kind: 'cleared',
condition: 'page-out objective',
iterations: 1,
}),
},
}),
]);
});
it('propagates unexpected runtime failures', async () => {
await expect(
renderPreparedGoalUpdate(async () => {
throw new Error('snapshot failed');
}),
).rejects.toThrow('snapshot failed');
});
});

View file

@ -0,0 +1,106 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { SessionUpdate } from '@agentclientprotocol/sdk';
import {
GoalPersistenceUnavailableError,
type ChatRecord,
type GoalRecord,
type GoalRuntime,
type GoalSnapshotV2,
type GoalStateCause,
} from '@qwen-code/qwen-code-core';
import type { HistoryItemGoalStatus } from '../../ui/types.js';
import {
collectGoalStatusItemsFromRecords,
findGoalToRestore,
} from '../../ui/utils/restoreGoal.js';
import type { HistoryReplayGoalBootstrap } from './history-replayer.js';
import {
buildGoalStateUpdate,
buildGoalStatusUpdate,
} from './emitters/MessageEmitter.js';
export interface RecoveredGoalUpdate {
publicationKey?: string;
suppressedGoalId?: string;
updates: SessionUpdate[];
}
export async function renderPreparedGoalUpdate(
getRuntime: () => Promise<GoalRuntime>,
options: {
replayedRecords?: readonly ChatRecord[];
hideRuntimeGoal?: boolean;
bootstrap?: HistoryReplayGoalBootstrap;
previousGoal?: GoalRecord | null;
} = {},
): Promise<RecoveredGoalUpdate> {
let runtime;
try {
runtime = await getRuntime();
} catch (error) {
if (!(error instanceof GoalPersistenceUnavailableError)) throw error;
const status = unrestorableGoalStatus(
options.replayedRecords,
options.bootstrap,
);
return { updates: status ? [buildGoalStatusUpdate(status)] : [] };
}
const cause = runtime.getRecoveryCause?.();
if (!cause) return { updates: [] };
const snapshot = runtime.getSnapshot();
const publicationKey = goalPublicationKey(snapshot, cause);
if (options.hideRuntimeGoal) {
return {
publicationKey,
...(snapshot.goal
? {
suppressedGoalId: snapshot.goal.goalId,
}
: {}),
updates: [],
};
}
const bootstrapGoal = options.bootstrap?.goalState?.goal;
const bootstrapMatchesRuntime =
bootstrapGoal != null &&
snapshot.goal?.goalId === bootstrapGoal.goalId &&
snapshot.goal?.revision === bootstrapGoal.revision;
return {
publicationKey,
updates:
options.bootstrap && bootstrapMatchesRuntime
? []
: [buildGoalStateUpdate(snapshot, cause, options.previousGoal ?? null)],
};
}
function unrestorableGoalStatus(
replayedRecords?: readonly ChatRecord[],
bootstrap?: HistoryReplayGoalBootstrap,
): Omit<HistoryItemGoalStatus, 'id' | 'type'> | undefined {
const active =
(replayedRecords?.length
? findGoalToRestore(collectGoalStatusItemsFromRecords(replayedRecords))
: undefined) ?? bootstrap?.goalStatus;
if (!active) return undefined;
return {
kind: 'cleared',
condition: active.condition,
iterations: active.iterations,
...(active.setAt !== undefined ? { setAt: active.setAt } : {}),
lastReason:
'Goal not restored: its saved state could not be read, so this session is not driving it.',
};
}
export function goalPublicationKey(
snapshot: GoalSnapshotV2,
cause?: GoalStateCause,
): string | undefined {
return cause ? `${cause}:${JSON.stringify(snapshot)}` : undefined;
}

View file

@ -121,7 +121,7 @@ const mockDefaultDaemonClient = vi.hoisted(() =>
);
const mockDefaultDaemonSessionClient = vi.hoisted(() => ({
createOrAttach: vi.fn(),
load: vi.fn(),
resume: vi.fn(),
}));
const mockBridgeStart = vi.hoisted(() => vi.fn());
@ -337,7 +337,7 @@ function createSdk() {
setModel: vi.fn(),
respondToPermission: vi.fn(),
}),
load: vi.fn().mockResolvedValue({
resume: vi.fn().mockResolvedValue({
sessionId: 'loaded-session',
workspaceCwd: '/workspace',
prompt: vi.fn(),
@ -437,7 +437,7 @@ describe('createDaemonSessionFactory', () => {
},
'qwen-channel-worker',
);
expect(sdk.DaemonSessionClient.load).toHaveBeenCalledWith(
expect(sdk.DaemonSessionClient.resume).toHaveBeenCalledWith(
sdk.client,
'existing-session',
{
@ -477,7 +477,7 @@ describe('createDaemonSessionFactory', () => {
},
'qwen-channel-worker',
);
expect(sdk.DaemonSessionClient.load).toHaveBeenCalledWith(
expect(sdk.DaemonSessionClient.resume).toHaveBeenCalledWith(
sdk.client,
'existing-session',
{
@ -516,7 +516,7 @@ describe('createDaemonSessionFactory', () => {
);
// The load branch never re-stamps creation attribution: no sourceId in the
// load request even when the factory request carried one.
expect(sdk.DaemonSessionClient.load).toHaveBeenCalledWith(
expect(sdk.DaemonSessionClient.resume).toHaveBeenCalledWith(
sdk.client,
'existing-session',
{

View file

@ -137,7 +137,7 @@ interface DaemonSessionClientStaticLike {
},
clientId?: string,
): Promise<DaemonChannelSessionClient>;
load(
resume(
client: DaemonClientLike,
sessionId: string,
req: {
@ -210,7 +210,7 @@ export function createDaemonSessionFactory({
sessionScope: 'thread' as const,
};
if (req.sessionId) {
return await DaemonSessionClient.load(
return await DaemonSessionClient.resume(
client,
req.sessionId,
daemonReq,

View file

@ -1742,6 +1742,107 @@ describe('loadCliConfig', () => {
);
});
it('rebinds a selective restore projection to the forked session', async () => {
const sourceSessionId = '123e4567-e89b-42d3-a456-426614174000';
const projectionSource = vi.fn(async (sessionId: string) => ({
sessionId,
filePath: `/mock/${sessionId}.jsonl`,
startTime: '2026-08-13T00:00:00.000Z',
lastUpdated: '2026-08-13T00:00:00.000Z',
runtime: {
apiHistory: [],
uiTelemetryEvents: [],
recording: { lastCompletedUuid: 'leaf', turnParentUuids: [] },
goalRecords: [],
initialTurn: 0,
backgroundNotificationTaskIds: [],
},
}));
const config = await loadCliConfig(
{},
{ resume: sourceSessionId, forkSession: true } as CliArgs,
undefined,
undefined,
undefined,
undefined,
undefined,
undefined,
false,
{ sessionRestore: { projectionSource } },
);
const forkedSessionId = config.getSessionId();
expect(mockSessionServiceInstance.forkSession).toHaveBeenCalledWith(
sourceSessionId,
forkedSessionId,
);
expect(projectionSource).toHaveBeenCalledOnce();
expect(projectionSource).toHaveBeenCalledWith(forkedSessionId);
expect(mockSessionServiceInstance.loadSession).not.toHaveBeenCalled();
const configParams = mockConfigConstructorParams.mock.calls.at(-1)?.[0];
expect(configParams).toEqual(
expect.objectContaining({
sessionId: forkedSessionId,
sessionData: undefined,
sessionRestoreProjection: expect.objectContaining({
sessionId: forkedSessionId,
}),
sessionRestoreProjectionSource: expect.any(Function),
}),
);
const deferredProjection =
await configParams.sessionRestoreProjectionSource();
expect(deferredProjection).toEqual(
expect.objectContaining({ sessionId: forkedSessionId }),
);
expect(projectionSource).toHaveBeenNthCalledWith(2, forkedSessionId);
});
it('preloads a selective projection when a non-ACP host cannot acquire a writer lease', async () => {
const sourceSessionId = '123e4567-e89b-42d3-a456-426614174000';
const projectionSource = vi.fn(async (sessionId: string) => ({
sessionId,
filePath: `/mock/${sessionId}.jsonl`,
startTime: '2026-08-13T00:00:00.000Z',
lastUpdated: '2026-08-13T00:00:00.000Z',
runtime: {
apiHistory: [],
uiTelemetryEvents: [],
recording: { lastCompletedUuid: 'leaf', turnParentUuids: [] },
goalRecords: [],
initialTurn: 0,
backgroundNotificationTaskIds: [],
},
}));
await loadCliConfig(
{ experimental: { sessionWriterLease: true } },
{ resume: sourceSessionId } as CliArgs,
undefined,
undefined,
undefined,
undefined,
undefined,
undefined,
false,
{ sessionRestore: { projectionSource } },
);
expect(projectionSource).toHaveBeenCalledOnce();
expect(projectionSource).toHaveBeenCalledWith(sourceSessionId);
expect(mockConfigConstructorParams).toHaveBeenLastCalledWith(
expect.objectContaining({
experimentalZedIntegration: false,
sessionWriterLeaseEnabled: true,
sessionRestoreProjection: expect.objectContaining({
sessionId: sourceSessionId,
}),
}),
);
});
it('should explain when --fork-session fails to copy the source session', async () => {
const sourceSessionId = '123e4567-e89b-42d3-a456-426614174000';
const sourceData = {

View file

@ -23,6 +23,7 @@ import {
SessionService,
ideContextStore,
type ResumedSessionData,
type SessionRestoreProjection,
type LspClient,
type ToolName,
type ToolInvocationGuard,
@ -42,6 +43,7 @@ import {
type SkillLevel,
type WebSearchSettings,
MAX_SUBAGENT_DEPTH_LIMIT,
addDaemonRequestAttribute,
} from '@qwen-code/qwen-code-core';
import { extensionsCommand } from '../commands/extensions.js';
import { hooksCommand } from '../commands/hooks.js';
@ -1557,6 +1559,11 @@ export async function loadCliConfig(
*/
hostPolicy?: {
toolInvocationGuard?: ToolInvocationGuard;
sessionRestore?: {
projectionSource: (
sessionId: string,
) => Promise<SessionRestoreProjection | undefined>;
};
},
): Promise<Config> {
const debugMode = isDebugMode(argv);
@ -1975,6 +1982,10 @@ export async function loadCliConfig(
let sessionId: string | undefined;
let sessionData: ResumedSessionData | undefined;
let sessionRestoreProjection: SessionRestoreProjection | undefined;
const sessionRestoreProjectionSource =
hostPolicy?.sessionRestore?.projectionSource;
let deferProjectionUntilWriterLease = false;
if (argv.continue || argv.resume) {
const sessionService = new SessionService(cwd);
@ -1995,8 +2006,24 @@ export async function loadCliConfig(
// session UUID by gemini.tsx (which handles custom title lookup and
// the interactive picker for ambiguous matches).
sessionId = argv.resume;
sessionData = await sessionService.loadSession(argv.resume);
if (!sessionData) {
deferProjectionUntilWriterLease =
sessionRestoreProjectionSource !== undefined &&
(argv.chatRecording ?? settings.general?.chatRecording ?? true) &&
isAcpMode === true &&
settings.experimental?.sessionWriterLease === true;
if (sessionRestoreProjectionSource) {
if (!deferProjectionUntilWriterLease && !argv.forkSession) {
addDaemonRequestAttribute(
'qwen-code.daemon.session_restore.projection_acquisition',
'preloaded',
);
sessionRestoreProjection =
await sessionRestoreProjectionSource(sessionId);
}
} else {
sessionData = await sessionService.loadSession(argv.resume);
}
if (!sessionRestoreProjectionSource && !sessionData) {
const message = `No saved session found with ID ${argv.resume}. Run \`qwen --resume\` without an ID to choose from existing sessions.`;
writeStderrLine(message);
process.exit(1);
@ -2015,10 +2042,22 @@ export async function loadCliConfig(
process.exit(1);
}
sessionId = forkedSessionId;
sessionData = await sessionService.loadSession(forkedSessionId);
if (!sessionData) {
writeStderrLine(`Failed to load forked session ${forkedSessionId}.`);
process.exit(1);
if (sessionRestoreProjectionSource) {
sessionData = undefined;
if (!deferProjectionUntilWriterLease) {
addDaemonRequestAttribute(
'qwen-code.daemon.session_restore.projection_acquisition',
'preloaded',
);
sessionRestoreProjection =
await sessionRestoreProjectionSource(forkedSessionId);
}
} else {
sessionData = await sessionService.loadSession(forkedSessionId);
if (!sessionData) {
writeStderrLine(`Failed to load forked session ${forkedSessionId}.`);
process.exit(1);
}
}
}
} else if (argv.sandboxSessionId) {
@ -2047,6 +2086,11 @@ export async function loadCliConfig(
const modelProvidersConfig = settings.modelProviders;
const providerProtocolConfig = settings.providerProtocol;
const restoreSessionId = sessionId;
const boundSessionRestoreProjectionSource =
sessionRestoreProjectionSource && restoreSessionId
? () => sessionRestoreProjectionSource(restoreSessionId)
: undefined;
// Assemble MCP servers across all sources in precedence order (user/default
// settings < project `.mcp.json` < workspace/system settings < `--mcp-config`)
@ -2083,6 +2127,8 @@ export async function loadCliConfig(
const configParams: ConfigParameters = {
sessionId,
sessionData,
sessionRestoreProjection,
sessionRestoreProjectionSource: boundSessionRestoreProjectionSource,
embeddingModel: DEFAULT_QWEN_EMBEDDING_MODEL,
sandbox: sandboxConfig,
targetDir: cwd,

View file

@ -45,6 +45,9 @@ describe('scheduled-task keepalive', () => {
loadSession: async (req: { sessionId: string }) => {
loads.push(req.sessionId);
},
resumeSession: async (req: { sessionId: string }) => {
loads.push(req.sessionId);
},
spawnOrAttach: async () => {
throw new Error('spawnOrAttach not mocked');
},
@ -131,7 +134,7 @@ describe('scheduled-task keepalive', () => {
}
beats.push(id);
},
loadSession: async (req: { sessionId: string }) => {
resumeSession: async (req: { sessionId: string }) => {
loads.push(req.sessionId);
},
spawnOrAttach: async () => {
@ -173,7 +176,7 @@ describe('scheduled-task keepalive', () => {
}
beats.push(id);
},
loadSession: async (req: { sessionId: string }) => {
resumeSession: async (req: { sessionId: string }) => {
loads.push(req.sessionId);
},
spawnOrAttach: async () => {
@ -282,7 +285,7 @@ describe('scheduled-task keepalive', () => {
if (id === 'sess-1') throw new Error('not resident');
beats.push(id);
},
loadSession: async (req: { sessionId: string }) => {
resumeSession: async (req: { sessionId: string }) => {
loads.push(req.sessionId);
loadRequests.push(req);
},
@ -309,7 +312,6 @@ describe('scheduled-task keepalive', () => {
{
sessionId: 'sess-1',
workspaceCwd: workspace,
historyReplay: 'response',
sourceType: 'scheduled_task',
sourceId: 'a',
},
@ -327,7 +329,7 @@ describe('scheduled-task keepalive', () => {
if (id === 'sess-1') throw new Error('not resident');
beats.push(id);
},
loadSession: async (req: { sessionId: string }) => {
resumeSession: async (req: { sessionId: string }) => {
loads.push(req.sessionId);
if (req.sessionId === 'sess-1') throw new Error('transcript gone');
},
@ -358,7 +360,7 @@ describe('scheduled-task keepalive', () => {
recordHeartbeat: () => {
throw new Error('not resident');
},
loadSession: async (req: { sessionId: string }) => {
resumeSession: async (req: { sessionId: string }) => {
loads.push(req.sessionId);
throw new Error('transcript gone');
},
@ -390,7 +392,7 @@ describe('scheduled-task keepalive', () => {
recordHeartbeat: () => {
throw new Error('not resident');
},
loadSession: async (req: { sessionId: string }) => {
resumeSession: async (req: { sessionId: string }) => {
loads.push(req.sessionId);
// Hang: loadSession isn't abortable, so it keeps running past the timeout.
await new Promise<void>((resolve) => {
@ -435,7 +437,7 @@ describe('scheduled-task keepalive', () => {
recordHeartbeat: () => {
throw new Error('not resident');
},
loadSession: async () => {
resumeSession: async () => {
markStarted?.();
await new Promise<void>((resolve) => {
releaseLoad = resolve;
@ -492,7 +494,7 @@ describe('scheduled-task keepalive', () => {
}> = [];
const res = await rehydrateScheduledTaskSessions({
bridge: {
loadSession: async (req) => {
resumeSession: async (req) => {
loaded.push(req);
},
},
@ -525,7 +527,7 @@ describe('scheduled-task keepalive', () => {
const errors: string[] = [];
const res = await rehydrateScheduledTaskSessions({
bridge: {
loadSession: async (req) => {
resumeSession: async (req) => {
if (req.sessionId === 'gone') throw new Error('missing transcript');
},
},
@ -540,7 +542,7 @@ describe('scheduled-task keepalive', () => {
it('rehydrate is a no-op when there are no tasks', async () => {
const res = await rehydrateScheduledTaskSessions({
bridge: {
loadSession: async () => {
resumeSession: async () => {
throw new Error('should not be called');
},
},
@ -561,7 +563,7 @@ describe('scheduled-task keepalive', () => {
let maxInFlight = 0;
const res = await rehydrateScheduledTaskSessions({
bridge: {
loadSession: async () => {
resumeSession: async () => {
inFlight++;
maxInFlight = Math.max(maxInFlight, inFlight);
await new Promise((r) => setTimeout(r, 5));
@ -590,7 +592,7 @@ describe('scheduled-task keepalive', () => {
const res = await rehydrateScheduledTaskSessions({
bridge: {
// Never resolves — a genuinely hung, non-abortable load.
loadSession: () => {
resumeSession: () => {
started++;
return new Promise<void>(() => {});
},
@ -616,7 +618,7 @@ describe('scheduled-task keepalive', () => {
});
const rehydrate = rehydrateScheduledTaskSessions({
bridge: {
loadSession: async () => {
resumeSession: async () => {
markStarted?.();
await new Promise<void>((resolve) => {
releaseLoad = resolve;
@ -648,7 +650,7 @@ describe('scheduled-task keepalive', () => {
]);
const res = await rehydrateScheduledTaskSessions({
bridge: {
loadSession: async () => {
resumeSession: async () => {
await new Promise((resolve) => setTimeout(resolve, 20));
},
},

View file

@ -71,17 +71,16 @@ function collectBoundSessionIds(tasks: readonly DurableCronTask[]): string[] {
}
/** The slice of the bridge the keepalive needs narrowed for testability.
* `recordHeartbeat` keeps a live session resident; `loadSession` revives one
* `recordHeartbeat` keeps a live session resident; `resumeSession` revives one
* the reaper already let go (a re-enabled task's session). `spawnOrAttach`
* and `updateSessionMetadata` bind unbound durable tasks to dedicated
* sessions the same flow the POST /scheduled-tasks route uses for
* UI-created tasks, applied retroactively to cron_create tool tasks. */
export interface KeepaliveBridge {
recordHeartbeat(sessionId: string): unknown;
loadSession(req: {
resumeSession(req: {
sessionId: string;
workspaceCwd: string;
historyReplay?: 'stream' | 'response';
sourceType?: string;
sourceId?: string;
}): Promise<unknown>;
@ -286,9 +285,9 @@ export function startScheduledTaskKeepalive(
string,
{ failures: number; nextAttemptAt: number }
>();
// Sessions with a revive in flight. loadSession isn't abortable, so a
// Sessions with a revive in flight. resumeSession isn't abortable, so a
// timed-out revive keeps running in the background; without this guard a later
// tick would spawn a SECOND loadSession (a duplicate child) for it. Cleared on
// tick would spawn a SECOND resumeSession (a duplicate child) for it. Cleared on
// the load's TRUE settlement, not the timeout.
const reviving = new Set<string>();
@ -339,21 +338,24 @@ export function startScheduledTaskKeepalive(
const metadata = await new SessionService(
boundWorkspace,
).readCreationMetadata(sessionId);
const load = bridge.loadSession({
const resume = bridge.resumeSession({
sessionId,
workspaceCwd: boundWorkspace,
historyReplay: 'response',
...metadata,
});
// Clear the in-flight guard on the load's TRUE settlement (not the
// Clear the in-flight guard on the resume's TRUE settlement (not the
// timeout below) so a still-running load keeps blocking a duplicate.
void load
void resume
.catch(() => {})
.finally(() => {
reviving.delete(sessionId);
});
try {
await withTimeout(load, reviveTimeoutMs, `loadSession(${sessionId})`);
await withTimeout(
resume,
reviveTimeoutMs,
`resumeSession(${sessionId})`,
);
log.debug('keepalive: revived non-resident session', sessionId);
reviveState.delete(sessionId);
} catch (loadErr) {
@ -408,7 +410,7 @@ export function startScheduledTaskKeepalive(
// In-flight guard: a pass can outlast the interval (each revive awaits up to
// the revive timeout), so skip a tick while the previous is still running —
// overlapping passes would issue duplicate concurrent loadSession spawns for
// overlapping passes would issue duplicate concurrent resumeSession spawns for
// the same dead sessions.
let running = false;
const timer: ReturnType<typeof setInterval> = setInterval(() => {
@ -476,10 +478,9 @@ export function startScheduledTaskKeepalive(
/** The slice of the bridge rehydration needs — narrowed for testability. */
export interface RehydrateBridge {
loadSession(req: {
resumeSession(req: {
sessionId: string;
workspaceCwd: string;
historyReplay?: 'stream' | 'response';
sourceType?: string;
sourceId?: string;
}): Promise<unknown>;
@ -497,12 +498,12 @@ export interface RehydrateResult {
* lock owner deliberately never fires a bound task) until something loaded it.
*
* Best-effort: a session whose transcript is gone (deleted out-of-band) fails
* its `loadSession` and is skipped rather than aborting the sweep. Distinct
* its `resumeSession` and is skipped rather than aborting the sweep. Distinct
* session ids only; unbound tasks are ignored (they fire via the lock owner).
*/
/** Default caller headroom above the bridge's 60-second restore deadline. */
const REHYDRATE_LOAD_TIMEOUT_MS = 70_000;
/** Max sessions rehydrated at once. Each `loadSession` forks a real agent
const REHYDRATE_RESUME_TIMEOUT_MS = 70_000;
/** Max sessions rehydrated at once. Each `resumeSession` forks a real agent
* child, so loading all of them (up to MAX_JOBS = 50) in one shot would spike
* CPU/memory on boot and, on constrained hosts, hit spawn failures
* (EAGAIN/ENOMEM) that strand healthy tasks. Load in small batches instead. */
@ -517,7 +518,7 @@ export async function rehydrateScheduledTaskSessions(deps: {
onTasksRead?: (tasks: readonly DurableCronTask[]) => void;
}): Promise<RehydrateResult> {
const { bridge, boundWorkspace } = deps;
const timeoutMs = deps.loadTimeoutMs ?? REHYDRATE_LOAD_TIMEOUT_MS;
const timeoutMs = deps.loadTimeoutMs ?? REHYDRATE_RESUME_TIMEOUT_MS;
let tasks;
try {
tasks = await readCronTasks(boundWorkspace);
@ -540,18 +541,17 @@ export async function rehydrateScheduledTaskSessions(deps: {
const metadata = await new SessionService(
boundWorkspace,
).readCreationMetadata(sessionId);
const load = bridge.loadSession({
const resume = bridge.resumeSession({
sessionId,
workspaceCwd: boundWorkspace,
historyReplay: 'response',
...metadata,
});
// loadSession isn't abortable, so a timed-out load keeps forking/replaying
// resumeSession isn't abortable, so a timed-out resume keeps running
// in the background. Swallow its eventual settlement up front so it can't
// raise an unhandled rejection once we've stopped awaiting it below.
void load.catch(() => {});
void resume.catch(() => {});
try {
await withTimeout(load, timeoutMs, `loadSession(${sessionId})`);
await withTimeout(resume, timeoutMs, `resumeSession(${sessionId})`);
loaded.push(sessionId);
} catch (err) {
// Timed out (or the load rejected). Do NOT await the raw `load` here: a
@ -559,7 +559,7 @@ export async function rehydrateScheduledTaskSessions(deps: {
// enough loads hang, the whole boot sweep never completes (`Promise.all`
// never settles) — later task sessions would then never rehydrate. Record
// it as failed and free the worker to pull the next queued session; the
// background load, if it ever settles, just warms that session late.
// background resume, if it ever settles, just warms that session late.
failed.push(sessionId);
// The onError callback must never abort the sweep: if it throws (e.g. a
// stderr EPIPE during log rotation) the rejection would escape loadOne,

View file

@ -9,8 +9,8 @@ import {
setGoalTerminalObserver,
setLastGoalTerminal,
unregisterGoalHook,
type ChatRecord,
type Config,
type GoalRecoveryRecord,
type GoalTerminalEvent,
type GoalTerminalKind,
type SlashCommandRecordPayload,
@ -178,7 +178,7 @@ export function parseGoalStatusItem(item: unknown): GoalStatusItem | null {
* exists, so `findGoalToRestore` / `findLastTerminalGoal` are fed from here.
*/
export function collectGoalStatusItemsFromRecords(
records: readonly ChatRecord[],
records: readonly GoalRecoveryRecord[],
): GoalStatusItem[] {
const items: GoalStatusItem[] = [];
for (const record of records) {

View file

@ -2522,6 +2522,83 @@ describe('Server Config (config.ts)', () => {
expect(replacement.getSnapshot().goal?.status).toBe('active');
});
it('holds selective Goal readiness and autonomous work until finalization', async () => {
const record = resumedGoalSession('active').conversation.messages[0]!;
const config = new Config({
...baseParams,
chatRecording: true,
sessionRestoreProjection: {
sessionId: 'resumed-session',
filePath: '/tmp/resumed-session.jsonl',
startTime: new Date(0).toISOString(),
lastUpdated: new Date(0).toISOString(),
runtime: {
apiHistory: [],
uiTelemetryEvents: [],
recording: {
lastCompletedUuid: record.uuid,
turnParentUuids: [],
},
goalRecords: [record],
initialTurn: 0,
backgroundNotificationTaskIds: [],
},
},
});
const started: string[] = [];
config.bindGoalTurnHost({
startGoalTurn: vi.fn(async ({ permit }) => {
started.push(permit.goalId);
}),
preemptGoalTurn: vi.fn(),
});
let ready = false;
void config.getGoalRuntimeReady().then(() => {
ready = true;
});
await Promise.resolve();
expect(ready).toBe(false);
expect(started).toEqual([]);
config.finalizeSessionRestore();
await expect(config.getGoalRuntimeReady()).resolves.toBe(
config.getGoalRuntime(),
);
await vi.waitFor(() => expect(started).toEqual(['g-resumed']));
});
it('rejects selective Goal readiness when restore is abandoned', async () => {
const record = resumedGoalSession('active').conversation.messages[0]!;
const config = new Config({
...baseParams,
chatRecording: true,
sessionRestoreProjection: {
sessionId: 'resumed-session',
filePath: '/tmp/resumed-session.jsonl',
startTime: new Date(0).toISOString(),
lastUpdated: new Date(0).toISOString(),
runtime: {
apiHistory: [],
uiTelemetryEvents: [],
recording: {
lastCompletedUuid: record.uuid,
turnParentUuids: [],
},
goalRecords: [record],
initialTurn: 0,
backgroundNotificationTaskIds: [],
},
},
});
const readiness = config.getGoalRuntimeReady();
config.startNewSession('replacement-session');
await expect(readiness).rejects.toThrow('Session restore was abandoned');
});
it('owns one durable Goal runtime per canonical session', async () => {
const config = new Config({ ...baseParams, chatRecording: true });
const first = config.getGoalRuntime();

View file

@ -130,6 +130,7 @@ import {
SENSITIVE_SPAN_ATTRIBUTE_MAX_LENGTH_LIMIT,
isValidSensitiveSpanAttributeMaxLength,
isTelemetrySdkInitialized,
addDaemonRequestAttribute,
initializeTelemetry,
shutdownTelemetry,
refreshSessionContext,
@ -172,6 +173,7 @@ import {
type GoalRuntime,
type GoalTurnHost,
} from '../goals/goal-runtime.js';
import type { GoalRecoveryRecord } from '../goals/goal-persistence.js';
import { createGoalCheckpointVerifier } from '../goals/goal-checkpoint-verifier.js';
import { createGoalVerifier } from '../goals/goal-verifier.js';
import type { ToolInvocationGuard } from '../core/tool-invocation-guard.js';
@ -210,7 +212,6 @@ import {
ChatRecordingService,
type ChatRecordingFailureEvent,
type ChatRecordingFailureListener,
type ChatRecord,
} from '../services/chatRecordingService.js';
import { CHARS_PER_TOKEN } from '../services/tokenEstimation.js';
import {
@ -221,6 +222,10 @@ import {
SessionService,
type ResumedSessionData,
} from '../services/sessionService.js';
import type {
SessionRestoreProjection,
SessionRuntimeResumeState,
} from '../services/session-transcript-reader.js';
import {
SessionTranscriptChangedError,
SessionWriterError,
@ -944,6 +949,10 @@ export interface AgentsCollabSettings {
export interface ConfigParameters {
sessionId?: string;
sessionData?: ResumedSessionData;
sessionRestoreProjection?: SessionRestoreProjection;
sessionRestoreProjectionSource?: () => Promise<
SessionRestoreProjection | undefined
>;
embeddingModel?: string;
sandbox?: SandboxConfig;
targetDir: string;
@ -1735,6 +1744,14 @@ export class Config {
private sessionSourceType?: string;
private sessionSourceId?: string;
private sessionData?: ResumedSessionData;
private pendingSessionRestoreProjection?: SessionRestoreProjection;
private sessionRestoreRuntime?: SessionRuntimeResumeState;
private readonly sessionRestoreProjectionSource?: () => Promise<
SessionRestoreProjection | undefined
>;
private restoredFileHistory = false;
private goalRestoreActivation?: () => Promise<void>;
private rejectGoalRestoreActivation?: (reason?: unknown) => void;
private readonly sessionRuntimeBaseDir: string;
private sessionProjectDirRegistered = false;
private pendingSessionWriterLease?: SessionWriterLease;
@ -2134,6 +2151,8 @@ export class Config {
sessionEnvClaimed = true;
}
this.sessionData = params.sessionData;
this.sessionRestoreProjectionSource = params.sessionRestoreProjectionSource;
this.setSessionRestoreProjection(params.sessionRestoreProjection);
setDebugLogSession(this);
this.debugLogger = createDebugLogger();
this.embeddingModel = params.embeddingModel ?? DEFAULT_QWEN_EMBEDDING_MODEL;
@ -2523,7 +2542,17 @@ export class Config {
this.chatRecordingService = this.chatRecordingEnabled
? this.createChatRecordingService()
: undefined;
this.initializeGoalRuntime(this.sessionData?.conversation.messages);
if (
!this.sessionRestoreProjectionSource ||
this.sessionRestoreRuntime ||
!this.sessionWriterLeaseEnabled
) {
this.initializeGoalRuntime(
this.sessionRestoreRuntime?.goalRecords ??
this.sessionData?.conversation.messages,
this.sessionRestoreRuntime,
);
}
this.extensionManager = new ExtensionManager({
workspaceDir: this.targetDir,
enabledExtensionOverrides: this.overrideExtensions,
@ -2621,6 +2650,7 @@ export class Config {
this.sessionProjectDirRegistered = true;
await this.initializeInternal(options);
} catch (error) {
this.clearSessionRestoreProjection();
if (this.sessionProjectDirRegistered) {
unregisterSessionProjectDir(this.sessionId);
this.sessionProjectDirRegistered = false;
@ -3183,7 +3213,15 @@ export class Config {
throw new SessionTranscriptChangedError();
}
let authoritative: ResumedSessionData | undefined;
if (this.sessionData || lease.transcriptExistedAtAcquire) {
let projection: SessionRestoreProjection | undefined;
if (this.sessionRestoreProjectionSource) {
addDaemonRequestAttribute(
'qwen-code.daemon.session_restore.projection_acquisition',
'after_writer_lease',
);
projection = await this.sessionRestoreProjectionSource();
this.setSessionRestoreProjection(projection);
} else if (this.sessionData || lease.transcriptExistedAtAcquire) {
authoritative = await this.getSessionService().loadSession(
this.sessionId,
);
@ -3199,7 +3237,18 @@ export class Config {
throw new SessionWriterShutdownError();
}
this.sessionData = authoritative;
recorder.activate(lease, authoritative, persistedTitleInfo);
recorder.activate(
lease,
authoritative,
persistedTitleInfo,
projection?.runtime.recording,
);
if (this.sessionRestoreProjectionSource) {
this.initializeGoalRuntime(
projection?.runtime.goalRecords,
projection?.runtime,
);
}
this.pendingSessionWriterLease = undefined;
lease = undefined;
// The recorder can take writes now, so the restore the constructor
@ -3778,6 +3827,77 @@ export class Config {
return this.sessionId;
}
getSessionRestoreRuntime(): SessionRuntimeResumeState | undefined {
return this.sessionRestoreRuntime;
}
consumeSessionRestoreProjection(): SessionRestoreProjection | undefined {
const projection = this.pendingSessionRestoreProjection;
this.pendingSessionRestoreProjection = undefined;
return projection;
}
hydrateSessionRestoreFileHistory(): void {
if (this.restoredFileHistory) return;
const snapshots = this.sessionRestoreRuntime?.fileHistorySnapshots;
if (!snapshots?.length) return;
const service = this.getFileHistoryService();
if (!service.isEnabled()) return;
service.restoreFromSnapshots(snapshots);
this.restoredFileHistory = true;
}
finalizeSessionRestore(): void {
const runtime = this.sessionRestoreRuntime;
if (!runtime) return;
this.sessionRestoreRuntime = undefined;
if (runtime.attributionSnapshot) {
try {
CommitAttributionService.getInstance().restoreFromSnapshot(
runtime.attributionSnapshot,
);
} catch (error) {
this.debugLogger.error(
`Session restore attribution activation failed: ${error}`,
);
}
}
const activateGoal = this.goalRestoreActivation;
this.goalRestoreActivation = undefined;
this.rejectGoalRestoreActivation = undefined;
if (activateGoal) {
try {
void activateGoal().catch((error) => {
this.debugLogger.error(
`Session restore goal activation failed: ${error}`,
);
});
} catch (error) {
this.debugLogger.error(
`Session restore goal activation failed: ${error}`,
);
}
}
if (this.restoredFileHistory && this.fileHistoryService) {
try {
void this.fileHistoryService
.validateRestoredSnapshots()
.catch((error) => {
this.debugLogger.error(
`FileHistory: validateRestoredSnapshots failed: ${error}`,
);
});
} catch (error) {
this.debugLogger.error(
`FileHistory: validateRestoredSnapshots failed: ${error}`,
);
}
}
}
setSessionSource(sourceType: string, sourceId?: string): void {
this.sessionSourceType = sourceType;
this.sessionSourceId = sourceId;
@ -3863,6 +3983,7 @@ export class Config {
unregisterSessionModel(previousSessionId);
this.publishModelEnv();
this.sessionData = sessionData;
this.clearSessionRestoreProjection();
this.pendingRecoveredAgentsNotice = null;
this.getOwnActiveTodoReminders().clear();
this.getOwnActiveTodoWorkChainOwners().clear();
@ -5097,6 +5218,7 @@ export class Config {
private async shutdownResourcesOnce(): Promise<void> {
try {
this.clearSessionRestoreProjection();
// Drop this session's project-dir registry entry. It is registered during
// initialization, so it is released here whenever that step completed —
// in daemon mode, where one process serves many sessions, an unreleased
@ -5111,6 +5233,11 @@ export class Config {
unregisterSessionModel(this.sessionId);
if (Object.hasOwn(this, 'goalRuntime')) {
this.rejectGoalRestoreActivation?.(
new GoalPersistenceUnavailableError('Goal runtime disposed'),
);
this.goalRestoreActivation = undefined;
this.rejectGoalRestoreActivation = undefined;
this.goalTurnHostUnbind?.();
this.goalTurnHostUnbind = undefined;
// Shutting down before the writer arrived: nothing will ever run
@ -7482,6 +7609,12 @@ export class Config {
return this.goalRuntimeReady.then(() => runtime);
}
getGoalRuntimePrepared(): Promise<GoalRuntime> {
const runtime = this.getGoalRuntime();
if (!this.sessionRestoreRuntime) return this.getGoalRuntimeReady();
return runtime.getPreparedRestore().then(() => runtime);
}
async rebaseGoalRuntimeFromActiveTranscript(): Promise<void> {
const runtime = this.getGoalRuntime();
const recordingService = this.chatRecordingService;
@ -7533,10 +7666,19 @@ export class Config {
this.notifyChatRecordingFailure(event);
},
this.sessionWriterLeaseEnabled,
this.sessionRestoreRuntime?.recording,
);
}
private initializeGoalRuntime(records?: readonly ChatRecord[]): void {
private initializeGoalRuntime(
records?: readonly GoalRecoveryRecord[],
restoreRuntime?: SessionRuntimeResumeState,
): void {
this.rejectGoalRestoreActivation?.(
new GoalPersistenceUnavailableError('Goal runtime replaced'),
);
this.goalRestoreActivation = undefined;
this.rejectGoalRestoreActivation = undefined;
this.goalTurnHostUnbind?.();
this.goalTurnHostUnbind = undefined;
// A runtime built here supersedes any restore still waiting on the
@ -7569,7 +7711,30 @@ export class Config {
// failure as `recoveryError` for the life of the runtime — the
// migrated goal is dropped and goal persistence is bricked for the
// whole resumed session. Wait for the writer instead.
if (this.sessionWriterLeaseEnabled && !recorder.hasWriteOwnership()) {
if (restoreRuntime) {
const preparation = runtime.prepareRestore(
records ?? [],
restoreRuntime.goalCheckpointWindow,
);
let resolveActivation!: () => void;
let rejectActivation!: (reason?: unknown) => void;
const activation = new Promise<void>((resolve, reject) => {
resolveActivation = resolve;
rejectActivation = reject;
});
this.rejectGoalRestoreActivation = rejectActivation;
this.goalRestoreActivation = () => {
const started = runtime.activateRestoredWork();
void started.then(resolveActivation, rejectActivation);
return started;
};
this.goalRuntimeReady = Promise.all([preparation, activation]).then(
() => runtime,
);
} else if (
this.sessionWriterLeaseEnabled &&
!recorder.hasWriteOwnership()
) {
const ready = new Promise<GoalRuntime>((resolve, reject) => {
this.pendingGoalRestore = { runtime, resolve, reject };
});
@ -7624,6 +7789,25 @@ export class Config {
pending.reject(error);
}
private setSessionRestoreProjection(
projection: SessionRestoreProjection | undefined,
): void {
this.pendingSessionRestoreProjection = projection;
this.sessionRestoreRuntime = projection?.runtime;
this.restoredFileHistory = false;
}
private clearSessionRestoreProjection(): void {
this.pendingSessionRestoreProjection = undefined;
this.sessionRestoreRuntime = undefined;
this.restoredFileHistory = false;
this.rejectGoalRestoreActivation?.(
new GoalPersistenceUnavailableError('Session restore was abandoned'),
);
this.goalRestoreActivation = undefined;
this.rejectGoalRestoreActivation = undefined;
}
private notifyChatRecordingFailure(event: ChatRecordingFailureEvent): void {
for (const listener of [...this.chatRecordingFailureListeners]) {
try {

View file

@ -633,6 +633,7 @@ describe('Gemini Client (client.ts)', () => {
getChatRecordingService: vi.fn().mockReturnValue(undefined),
getFileHistoryService: vi.fn().mockReturnValue(mockFileHistoryService),
getResumedSessionData: vi.fn().mockReturnValue(undefined),
getSessionRestoreRuntime: vi.fn().mockReturnValue(undefined),
getArenaAgentClient: vi.fn().mockReturnValue(null),
getManagedAutoMemoryEnabled: vi.fn().mockReturnValue(true),
isManagedMemoryAvailable: vi.fn().mockReturnValue(true),
@ -705,6 +706,46 @@ describe('Gemini Client (client.ts)', () => {
});
describe('initialize', () => {
it('initializes from the selective runtime projection without the full transcript', async () => {
const seedResumeTokenCountsSpy = vi.spyOn(
GeminiChat.prototype,
'seedResumeTokenCounts',
);
const apiHistory = [
{ role: 'user' as const, parts: [{ text: 'projected history' }] },
];
const uiEvent = { type: 'projected-event' };
vi.mocked(mockConfig.getSessionRestoreRuntime).mockReturnValue({
apiHistory,
resumeTokenCounts: {
promptTokenCount: 321,
outputTokenCount: 45,
isEstimated: false,
},
uiTelemetryEvents: [uiEvent],
recording: {
lastCompletedUuid: 'record-1',
turnParentUuids: [],
},
goalRecords: [],
initialTurn: 0,
backgroundNotificationTaskIds: [],
} as unknown as ReturnType<Config['getSessionRestoreRuntime']>);
const resumedClient = new GeminiClient(mockConfig);
await resumedClient.initialize();
expect(resumedClient.getHistory().at(-1)).toEqual(apiHistory[0]);
expect(uiTelemetryService.resetSession).toHaveBeenCalledWith(
'test-session-id',
);
expect(uiTelemetryService.addEvent).toHaveBeenCalledWith(
uiEvent,
'test-session-id',
);
expect(seedResumeTokenCountsSpy).toHaveBeenCalledWith(321, 45, false);
});
it('seeds resumed chat with replayed prompt token count', async () => {
vi.mocked(mockConfig.getResumedSessionData).mockReturnValue({
conversation: {

View file

@ -428,7 +428,32 @@ export class GeminiClient {
// Check if we're resuming from a previous session
const resumedSessionData = this.config.getResumedSessionData();
if (resumedSessionData) {
const restoreRuntime = this.config.getSessionRestoreRuntime?.();
if (restoreRuntime) {
uiTelemetryService.resetSession(sessionId);
for (const event of restoreRuntime.uiTelemetryEvents) {
uiTelemetryService.addEvent(event, sessionId);
}
this.seedRecentCompletedToolNamesFromHistory(restoreRuntime.apiHistory);
await this.startChat(
restoreRuntime.apiHistory,
sessionStartSource ?? SessionStartSource.Resume,
);
const chat = this.getChat();
if (restoreRuntime.resumeTokenCounts) {
const counts = restoreRuntime.resumeTokenCounts;
uiTelemetryService.setLastPromptTokenCount(counts.promptTokenCount);
chat.seedResumeTokenCounts(
counts.promptTokenCount,
counts.outputTokenCount,
counts.isEstimated,
);
} else {
chat.setLastPromptTokenCount(
uiTelemetryService.getLastPromptTokenCount(),
);
}
} else if (resumedSessionData) {
const resumeTokenCounts = replayUiTelemetryFromConversation(
resumedSessionData.conversation,
this.config.getSessionId(),

View file

@ -14,7 +14,10 @@ import {
type GoalTerminalProposal,
type GoalTurnPermit,
} from './goal-protocol.js';
import { projectUserTranscriptForDisplay } from '../utils/transcript-records.js';
import {
isUserPromptSubmitContextPartText,
projectUserTranscriptForDisplay,
} from '../utils/transcript-records.js';
const CATALOG_PREVIEW_LIMIT = 240;
const CATALOG_ENTRY_LIMIT = 100;
@ -158,6 +161,378 @@ interface ParsedGoalContext {
turnId: string;
}
export interface GoalEvidenceRecordIndexHint {
uuid: string;
parsedGoalContext?: {
goalId: string;
revision: number;
turnId: string;
};
claimedGoalId?: string;
claimedRevision?: number;
provenance?: GoalEvidenceProvenance;
hasCatalogEligibleContent: boolean;
hasRawEligibleContent: boolean;
catalogEntryBytes?: number;
}
export class GoalEvidenceRecordIndexAccumulator {
private readonly uuid: string;
private readonly parsedGoalContext?: ParsedGoalContext;
private readonly claimedGoalId?: string;
private readonly claimedRevision?: number;
private readonly provenance?: GoalEvidenceProvenance;
private readonly hasObjectSystemPayload: boolean;
private readonly displayText?: string;
private readonly hasHookContext: boolean;
private prefixPreview = '';
private lastPartPreviewValues: string[] = [];
private lastPartIsHookContext = false;
private partCount = 0;
private hasRawEligibleContent = false;
constructor(record: GoalEvidenceRecord) {
this.uuid = record.uuid;
this.parsedGoalContext = parseGoalContext(record.goalContext);
const claimed = isRecord(record.goalContext)
? record.goalContext
: undefined;
this.claimedGoalId =
typeof claimed?.['goalId'] === 'string' ? claimed['goalId'] : undefined;
this.claimedRevision =
typeof claimed?.['revision'] === 'number'
? claimed['revision']
: undefined;
this.provenance = this.parsedGoalContext
? coherentEvidenceProvenance(record)
: undefined;
const systemPayload = isRecord(record.systemPayload)
? record.systemPayload
: undefined;
this.hasObjectSystemPayload = systemPayload !== undefined;
this.displayText =
typeof systemPayload?.['displayText'] === 'string'
? systemPayload['displayText'].slice(0, CATALOG_PREVIEW_LIMIT)
: undefined;
this.hasHookContext = typeof systemPayload?.['hookContext'] === 'string';
this.addFragment(record);
}
addFragment(record: GoalEvidenceRecord): void {
if (!this.provenance) return;
for (const part of record.message?.parts ?? []) {
this.finishPreviousPart();
const previewValues: string[] = [];
if (part.thought !== true && typeof part.text === 'string') {
previewValues.push(part.text.slice(0, CATALOG_PREVIEW_LIMIT));
if (part.text.trim()) this.hasRawEligibleContent = true;
}
if (this.provenance === 'tool_result' && part.functionResponse) {
previewValues.push(renderToolResponsePreview(part.functionResponse));
if (part.functionResponse.response !== undefined) {
this.hasRawEligibleContent = true;
}
}
this.lastPartPreviewValues = previewValues;
this.lastPartIsHookContext =
typeof part.text === 'string' &&
isUserPromptSubmitContextPartText(part.text);
this.partCount++;
}
}
finish(): GoalEvidenceRecordIndexHint {
let preview: string;
const hasFinalHookContextPart =
this.partCount > 1 && this.lastPartIsHookContext;
if (
this.provenance === 'real_user' &&
(this.hasHookContext || hasFinalHookContextPart) &&
this.displayText !== undefined
) {
preview = this.displayText.slice(0, CATALOG_PREVIEW_LIMIT).trim();
} else if (
this.provenance === 'real_user' &&
!this.hasObjectSystemPayload &&
hasFinalHookContextPart
) {
preview = this.prefixPreview.trim();
} else {
preview = appendPreviewValues(
this.prefixPreview,
this.lastPartPreviewValues,
).trim();
}
const catalogEntry =
this.provenance && this.parsedGoalContext && preview
? {
uuid: this.uuid,
provenance: this.provenance,
turnId: this.parsedGoalContext.turnId,
preview,
proofKind: proofKindOf(this.provenance),
}
: undefined;
return {
uuid: this.uuid,
...(this.parsedGoalContext
? { parsedGoalContext: this.parsedGoalContext }
: {}),
...(this.claimedGoalId !== undefined
? { claimedGoalId: this.claimedGoalId }
: {}),
...(this.claimedRevision !== undefined
? { claimedRevision: this.claimedRevision }
: {}),
...(this.provenance ? { provenance: this.provenance } : {}),
hasCatalogEligibleContent: catalogEntry !== undefined,
hasRawEligibleContent: this.hasRawEligibleContent,
...(catalogEntry
? {
catalogEntryBytes: Buffer.byteLength(
JSON.stringify(catalogEntry),
'utf8',
),
}
: {}),
};
}
private finishPreviousPart(): void {
if (this.partCount === 0) return;
this.prefixPreview = appendPreviewValues(
this.prefixPreview,
this.lastPartPreviewValues,
);
}
}
function appendPreviewValues(
current: string,
values: readonly string[],
): string {
let preview = current;
for (const value of values) {
if (!value || preview.length >= CATALOG_PREVIEW_LIMIT) continue;
const separator = preview ? '\n' : '';
const remaining = CATALOG_PREVIEW_LIMIT - preview.length;
preview += `${separator}${value}`.slice(0, remaining);
}
return preview;
}
export class GoalEvidenceCheckpointAccumulator {
private readonly candidateUuids: string[] = [];
private readonly candidateUuidSet = new Set<string>();
private readonly captured = new Map<string, ValidatedGoalEvidenceRecord>();
private readonly checkpointEntries: GoalEvidenceCatalogEntry[];
private readonly truncated: boolean;
private readonly shouldCheckpoint: boolean;
constructor(
hints: readonly GoalEvidenceRecordIndexHint[],
private readonly goal: GoalRecord,
permit: GoalTurnPermit,
) {
if (
permit.goalId !== goal.goalId ||
permit.revision !== goal.revision ||
!isNonEmptyString(permit.turnId)
) {
throw new EvidenceSourceUnavailableError(
'permit_goal_mismatch',
'The current Goal permit does not match the Goal evidence revision.',
);
}
const indexByUuid = new Map<string, number>();
for (let index = 0; index < hints.length; index++) {
const uuid = hints[index]!.uuid;
if (indexByUuid.has(uuid)) {
throw new EvidenceSourceUnavailableError(
'duplicate_record_uuid',
`The active transcript chain contains duplicate record UUID ${uuid}.`,
);
}
indexByUuid.set(uuid, index);
}
const cursorId = goal.evidenceCursor.recordId;
if (cursorId === null) {
throw new EvidenceSourceUnavailableError(
'cursor_unset',
'The Goal evidence cursor is not available.',
);
}
const cursorIndex = indexByUuid.get(cursorId);
if (cursorIndex === undefined) {
throw new EvidenceSourceUnavailableError(
'cursor_not_found',
`The Goal evidence cursor ${cursorId} is not in the active transcript chain.`,
);
}
const lineageTurnIds: string[] = [];
const seenTurnIds = new Set<string>();
let currentTurnId: string | undefined;
for (let index = cursorIndex + 1; index < hints.length; index++) {
const hint = hints[index]!;
const context = hint.parsedGoalContext;
if (!context) {
if (
hint.claimedGoalId === goal.goalId &&
hint.claimedRevision === goal.revision
) {
throw new EvidenceSourceUnavailableError(
'malformed_turn_context',
`Goal-owned transcript record ${hint.uuid} has malformed turn context.`,
);
}
continue;
}
if (
context.goalId !== goal.goalId ||
context.revision !== goal.revision
) {
continue;
}
if (context.turnId === currentTurnId) continue;
if (seenTurnIds.has(context.turnId)) {
throw new EvidenceSourceUnavailableError(
'turn_reentry',
`Goal turn ${context.turnId} re-enters the active transcript lineage.`,
);
}
seenTurnIds.add(context.turnId);
lineageTurnIds.push(context.turnId);
currentTurnId = context.turnId;
}
if (lineageTurnIds.at(-1) !== permit.turnId) {
throw new EvidenceSourceUnavailableError(
'current_turn_not_tail',
'The current Goal permit is not the tail of the active transcript lineage.',
);
}
this.checkpointEntries = checkpointCatalogEntries(goal);
const checkpointBytes = this.checkpointEntries.reduce(
(total, entry) =>
total + Buffer.byteLength(JSON.stringify(entry), 'utf8'),
0,
);
let truncated =
this.checkpointEntries.length >= CATALOG_ENTRY_LIMIT ||
checkpointBytes > CATALOG_BYTE_LIMIT;
const rawEntryLimit = Math.max(
0,
CATALOG_ENTRY_LIMIT - this.checkpointEntries.length,
);
let catalogBytes = checkpointBytes;
for (
let index = hints.length - 1;
!truncated && index > cursorIndex;
index--
) {
const hint = hints[index]!;
const context = hint.parsedGoalContext;
if (
!hint.provenance ||
!context ||
context.goalId !== goal.goalId ||
context.revision !== goal.revision
) {
continue;
}
if (this.candidateUuids.length >= rawEntryLimit) {
if (hint.hasRawEligibleContent) {
truncated = true;
break;
}
continue;
}
if (!hint.hasCatalogEligibleContent) continue;
const entryBytes = hint.catalogEntryBytes;
if (
entryBytes === undefined ||
catalogBytes + entryBytes > CATALOG_BYTE_LIMIT
) {
truncated = true;
break;
}
this.candidateUuids.push(hint.uuid);
this.candidateUuidSet.add(hint.uuid);
catalogBytes += entryBytes;
}
this.truncated = truncated;
this.shouldCheckpoint =
!truncated &&
this.candidateUuids.length > 0 &&
(this.checkpointEntries.length + this.candidateUuids.length >=
CHECKPOINT_ENTRY_THRESHOLD ||
catalogBytes >= CHECKPOINT_BYTE_THRESHOLD);
}
getCandidateUuids(): readonly string[] {
return this.shouldCheckpoint ? this.candidateUuids : [];
}
capture(record: GoalEvidenceRecord): void {
if (!this.shouldCheckpoint || !this.candidateUuidSet.has(record.uuid)) {
return;
}
const provenance = coherentEvidenceProvenance(record);
if (!provenance) return;
const context = parseGoalContext(record.goalContext);
if (
!context ||
context.goalId !== this.goal.goalId ||
context.revision !== this.goal.revision
) {
return;
}
const preview = evidencePreview(record, provenance);
const content = evidenceContent(record, provenance);
if (!preview || !content) return;
this.captured.set(record.uuid, {
uuid: record.uuid,
provenance,
turnId: context.turnId,
preview,
proofKind: proofKindOf(provenance),
content: capCheckpointContent(content),
});
}
finish(): GoalEvidenceCheckpointWindow {
const selected = this.shouldCheckpoint
? this.candidateUuids.map((uuid) => {
const entry = this.captured.get(uuid);
if (!entry) {
throw new InvalidGoalEvidenceReferenceError(
'ineligible_reference',
`Transcript record ${uuid} has no eligible evidence content.`,
uuid,
);
}
return entry;
})
: [];
selected.reverse();
return {
previousClaims: structuredClone(
this.goal.evidenceCheckpoint?.claims ?? [],
),
evidence: selected,
truncated: this.truncated,
shouldCheckpoint: this.shouldCheckpoint,
};
}
}
export function getGoalEvidenceRecordIndexHint(
record: GoalEvidenceRecord,
): GoalEvidenceRecordIndexHint {
return new GoalEvidenceRecordIndexAccumulator(record).finish();
}
export function buildGoalEvidenceCatalog(
input: GoalEvidenceContext,
): GoalEvidenceCatalog {
@ -172,37 +547,19 @@ export function buildGoalEvidenceCatalog(
export function buildGoalEvidenceCheckpointWindow(
input: GoalEvidenceContext,
): GoalEvidenceCheckpointWindow {
const analysis = analyzeEvidence(input);
const rawEntries = analysis.catalog.filter(
(entry) => entry.provenance !== 'goal_checkpoint',
const accumulator = new GoalEvidenceCheckpointAccumulator(
input.records.map(getGoalEvidenceRecordIndexHint),
input.goal,
input.permit,
);
const shouldCheckpoint =
!analysis.catalogTruncated &&
rawEntries.length > 0 &&
(analysis.catalog.length >= CHECKPOINT_ENTRY_THRESHOLD ||
analysis.catalogBytes >= CHECKPOINT_BYTE_THRESHOLD);
const evidence = (shouldCheckpoint ? rawEntries : []).map((entry) => {
const recordIndex = analysis.indexByUuid.get(entry.uuid);
const record =
recordIndex === undefined ? undefined : input.records[recordIndex];
const content = record ? evidenceContent(record, entry.provenance) : '';
if (!content) {
throw new InvalidGoalEvidenceReferenceError(
'ineligible_reference',
`Transcript record ${entry.uuid} has no eligible evidence content.`,
entry.uuid,
);
}
return { ...entry, content: capCheckpointContent(content) };
});
return {
previousClaims: structuredClone(
input.goal.evidenceCheckpoint?.claims ?? [],
),
evidence,
truncated: analysis.catalogTruncated,
shouldCheckpoint,
};
const recordsByUuid = new Map(
input.records.map((record) => [record.uuid, record]),
);
for (const uuid of accumulator.getCandidateUuids()) {
const record = recordsByUuid.get(uuid);
if (record) accumulator.capture(record);
}
return accumulator.finish();
}
export function validateGoalEvidenceReferences(

View file

@ -25,6 +25,11 @@ export type GoalRecoveryRecord = Pick<ChatRecord, 'uuid' | 'type'> & {
systemPayload?: unknown;
};
export interface GoalRecoverySelection {
recovery: GoalRecovery;
sourceUuid?: string;
}
const LEGACY_ACTIVE_KINDS = new Set(['set', 'checking']);
const LEGACY_STOPPED_KINDS = new Set([
'achieved',
@ -37,7 +42,14 @@ const LEGACY_STOPPED_KINDS = new Set([
export function recoverGoalFromRecords(
records: readonly GoalRecoveryRecord[],
): GoalRecovery {
return selectGoalRecoveryFromRecords(records).recovery;
}
export function selectGoalRecoveryFromRecords(
records: readonly GoalRecoveryRecord[],
): GoalRecoverySelection {
let unsupported: GoalRecovery | undefined;
let unsupportedSourceUuid: string | undefined;
for (let index = records.length - 1; index >= 0; index -= 1) {
const record = records[index];
if (record?.subtype !== 'goal_state') continue;
@ -45,19 +57,26 @@ export function recoverGoalFromRecords(
record.type === 'system'
? parseGoalStateRecordPayloadV2(record.systemPayload)
: undefined;
if (payload) return { kind: 'v2', payload };
unsupported ??= {
kind: 'unsupported',
reason: `Goal lifecycle record ${record.uuid} is malformed or uses an unsupported version`,
};
if (payload) {
return { recovery: { kind: 'v2', payload }, sourceUuid: record.uuid };
}
if (!unsupported) {
unsupported = {
kind: 'unsupported',
reason: `Goal lifecycle record ${record.uuid} is malformed or uses an unsupported version`,
};
unsupportedSourceUuid = record.uuid;
}
}
return unsupported ?? recoverLegacyGoal(records);
return unsupported
? { recovery: unsupported, sourceUuid: unsupportedSourceUuid }
: recoverLegacyGoal(records);
}
function recoverLegacyGoal(
records: readonly GoalRecoveryRecord[],
): GoalRecovery {
): GoalRecoverySelection {
for (
let recordIndex = records.length - 1;
recordIndex >= 0;
@ -86,16 +105,81 @@ function recoverLegacyGoal(
const kind = value['kind'];
const condition = value['condition'];
if (typeof kind !== 'string' || typeof condition !== 'string') {
return unsupportedLegacy(record.uuid);
return {
recovery: unsupportedLegacy(record.uuid),
sourceUuid: record.uuid,
};
}
if (LEGACY_STOPPED_KINDS.has(kind)) {
return { recovery: { kind: 'none' }, sourceUuid: record.uuid };
}
if (LEGACY_STOPPED_KINDS.has(kind)) return { kind: 'none' };
if (!LEGACY_ACTIVE_KINDS.has(kind) || condition.trim().length === 0) {
return unsupportedLegacy(record.uuid);
return {
recovery: unsupportedLegacy(record.uuid),
sourceUuid: record.uuid,
};
}
return { kind: 'legacy', objective: condition.trim() };
return {
recovery: { kind: 'legacy', objective: condition.trim() },
sourceUuid: record.uuid,
};
}
}
return { kind: 'none' };
return { recovery: { kind: 'none' } };
}
export function normalizeGoalRecoveryRecord(
record: GoalRecoveryRecord,
): GoalRecoveryRecord | undefined {
if (record.subtype === 'goal_state') {
return {
uuid: record.uuid,
type: record.type,
subtype: record.subtype,
systemPayload:
record.type === 'system'
? (parseGoalStateRecordPayloadV2(record.systemPayload) ?? null)
: null,
};
}
if (record.type !== 'system' || record.subtype !== 'slash_command') {
return undefined;
}
const payload = record.systemPayload as SlashCommandRecordPayload | undefined;
if (
payload?.phase !== 'result' ||
!Array.isArray(payload.outputHistoryItems)
) {
return undefined;
}
const goalStatusItems = payload.outputHistoryItems.filter(
(value) => isObjectRecord(value) && value['type'] === 'goal_status',
);
if (goalStatusItems.length === 0) return undefined;
return {
uuid: record.uuid,
type: record.type,
subtype: record.subtype,
systemPayload: {
phase: 'result',
outputHistoryItems: goalStatusItems,
},
};
}
export function isGoalRecoveryCandidate(record: GoalRecoveryRecord): boolean {
if (record.subtype === 'goal_state') return true;
if (record.type !== 'system' || record.subtype !== 'slash_command') {
return false;
}
const payload = record.systemPayload as SlashCommandRecordPayload | undefined;
return (
payload?.phase === 'result' &&
Array.isArray(payload.outputHistoryItems) &&
payload.outputHistoryItems.some(
(value) => isObjectRecord(value) && value['type'] === 'goal_status',
)
);
}
function unsupportedLegacy(recordUuid: string): GoalRecovery {

View file

@ -3670,6 +3670,82 @@ describe('goal runtime', () => {
});
});
it('prepares an active restore without broadcasting or starting work', async () => {
const host = fakeGoalTurnHost();
const runtime = createGoalRuntime({ journal: fakeGoalJournal() });
const listener = vi.fn();
runtime.bindHost(host);
runtime.subscribe(listener);
const record = goalStateRecord({
v: 2,
activity: 'idle',
goal: {
goalId: 'g-selective',
revision: 1,
objective: 'resume selectively',
status: 'active',
evidenceCursor: { recordId: 'restore-record' },
turnCount: 1,
activeTimeMs: 10,
createdAt: 1,
updatedAt: 2,
},
});
await runtime.prepareRestore([record]);
expect(runtime.getSnapshot().goal?.status).toBe('active');
expect(listener).not.toHaveBeenCalled();
expect(host.started).toEqual([]);
await runtime.activateRestoredWork();
expect(listener).toHaveBeenCalledTimes(2);
expect(host.started).toHaveLength(1);
});
it('coalesces preparation and activation and rejects activation before preparation', async () => {
const runtime = createGoalRuntime({ journal: fakeGoalJournal() });
await expect(runtime.activateRestoredWork()).rejects.toThrow(
'preparation has not started',
);
const record = goalStateRecord({
v: 2,
activity: 'idle',
goal: null,
});
const firstPreparation = runtime.prepareRestore([record]);
const secondPreparation = runtime.prepareRestore([record]);
await Promise.all([firstPreparation, secondPreparation]);
const firstActivation = runtime.activateRestoredWork();
const secondActivation = runtime.activateRestoredWork();
await expect(
Promise.all([firstActivation, secondActivation]),
).resolves.toEqual([undefined, undefined]);
});
it('prevents unfinished restore preparation from committing after disposal', async () => {
let releaseAppend!: () => void;
const appendGate = new Promise<void>((resolve) => {
releaseAppend = resolve;
});
const runtime = createGoalRuntime({
journal: fakeGoalJournal({ beforeAppend: () => appendGate }),
});
const preparing = runtime.prepareRestore([legacyGoalRecord()]);
await Promise.resolve();
runtime.dispose();
releaseAppend();
await expect(preparing).rejects.toThrow('Goal runtime has been disposed');
await expect(runtime.activateRestoredWork()).rejects.toThrow(
'Goal runtime has been disposed',
);
});
it('commits paused legacy recovery before a reentrant resume', async () => {
const journal = fakeGoalJournal({
appendErrors: [new Error('migration write failed'), undefined],

View file

@ -12,6 +12,7 @@ import {
InvalidGoalEvidenceReferenceError,
validateGoalEvidenceReferences,
type GoalEvidenceCatalog,
type GoalEvidenceCheckpointWindow,
type GoalEvidenceRecord,
} from './goal-evidence.js';
import {
@ -126,6 +127,12 @@ export interface GoalRuntime {
listener: (snapshot: GoalSnapshotV2, cause?: GoalStateCause) => void,
): () => void;
restore(records: readonly GoalRecoveryRecord[]): Promise<void>;
prepareRestore(
records: readonly GoalRecoveryRecord[],
checkpointWindow?: GoalEvidenceCheckpointWindow,
): Promise<void>;
getPreparedRestore(): Promise<void>;
activateRestoredWork(): Promise<void>;
dispatch(request: GoalControlRequest): Promise<GoalStateResponse>;
bindHost(host: GoalTurnHost): () => void;
beginTurn(turnKey: string): GoalTurnPermit | undefined;
@ -214,6 +221,12 @@ export function createGoalRuntime(
let nextVerifierFeedback: string | undefined;
let currentTurnFeedback: string | undefined;
let restored = false;
let restoreActivationPending = false;
let restorePreparation: Promise<CheckpointAttempt | undefined> | undefined;
let restoreActivation: Promise<void> | undefined;
let preparedRestoreCause: GoalStateCause | undefined;
let preparedRestoreHasSnapshot = false;
let preparedCheckpointWindow: GoalEvidenceCheckpointWindow | undefined;
let disposed = false;
let recoveryError: Error | undefined;
/**
@ -349,6 +362,7 @@ export function createGoalRuntime(
const queueContinuation = (cause?: GoalStateCause) => {
if (
restoreActivationPending ||
snapshot.goal?.status !== 'active' ||
currentPermit ||
pendingProposal ||
@ -800,10 +814,13 @@ export function createGoalRuntime(
});
};
const runCheckpoint = async (attempt: CheckpointAttempt): Promise<void> => {
const runCheckpoint = async (
attempt: CheckpointAttempt,
preparedWindow?: GoalEvidenceCheckpointWindow,
): Promise<void> => {
const evidenceSource = options.evidenceSource;
const checkpointVerifier = options.checkpointVerifier;
if (!evidenceSource || !checkpointVerifier) {
if ((!preparedWindow && !evidenceSource) || !checkpointVerifier) {
await recordCheckpointFailure(
attempt,
'Goal checkpoint recovery dependencies are unavailable',
@ -812,15 +829,18 @@ export function createGoalRuntime(
}
try {
await evidenceSource.flush();
if (attempt.controller.signal.aborted) return;
const records = await evidenceSource.readActiveTranscriptChain();
if (attempt.controller.signal.aborted) return;
const window = buildGoalEvidenceCheckpointWindow({
records,
goal: attempt.goal,
permit: attempt.permit,
});
let window = preparedWindow;
if (!window) {
await evidenceSource!.flush();
if (attempt.controller.signal.aborted) return;
const records = await evidenceSource!.readActiveTranscriptChain();
if (attempt.controller.signal.aborted) return;
window = buildGoalEvidenceCheckpointWindow({
records,
goal: attempt.goal,
permit: attempt.permit,
});
}
if (window.truncated) {
await recordCheckpointFailure(
attempt,
@ -900,8 +920,14 @@ export function createGoalRuntime(
listeners.add(listener);
return () => listeners.delete(listener);
},
restore(records: readonly GoalRecoveryRecord[]): Promise<void> {
const restoring = enqueue(
prepareRestore(
records: readonly GoalRecoveryRecord[],
checkpointWindow?: GoalEvidenceCheckpointWindow,
): Promise<void> {
if (restorePreparation) return restorePreparation.then(() => undefined);
restoreActivationPending = true;
preparedCheckpointWindow = checkpointWindow;
const preparation = enqueue(
async (): Promise<CheckpointAttempt | undefined> => {
assertAvailable();
if (restored) return;
@ -961,14 +987,15 @@ export function createGoalRuntime(
recoveredSnapshot = structuredClone(payload.snapshot);
recoveredCause = payload.cause;
}
assertAvailable();
if (recoveredSnapshot) snapshot = recoveredSnapshot;
recoveryError = undefined;
restored = true;
if (recoveredSnapshot) {
recoveryCause = recoveredCause;
broadcast(recoveredCause);
}
if (!checkpointAttempt) queueContinuation();
preparedRestoreHasSnapshot = recoveredSnapshot !== undefined;
preparedRestoreCause = recoveredCause;
return checkpointAttempt;
} catch (error) {
if (!disposed) {
@ -979,10 +1006,57 @@ export function createGoalRuntime(
}
},
);
return restoring.then(async (attempt) => {
if (!attempt) return;
restorePreparation = preparation;
return preparation.then(
() => undefined,
(error) => {
if (!restored && restorePreparation === preparation) {
restorePreparation = undefined;
restoreActivation = undefined;
restoreActivationPending = false;
preparedCheckpointWindow = undefined;
}
throw error;
},
);
},
getPreparedRestore(): Promise<void> {
if (!restorePreparation) {
return Promise.reject(
new GoalPersistenceUnavailableError(
'Goal restore preparation has not started',
),
);
}
return restorePreparation.then(() => undefined);
},
activateRestoredWork(): Promise<void> {
try {
assertAvailable();
} catch (error) {
return Promise.reject(error);
}
if (!restorePreparation) {
return Promise.reject(
new GoalPersistenceUnavailableError(
'Goal restore preparation has not started',
),
);
}
if (restoreActivation) return restoreActivation;
restoreActivation = restorePreparation.then(async (attempt) => {
assertAvailable();
restoreActivationPending = false;
if (preparedRestoreHasSnapshot) broadcast(preparedRestoreCause);
if (!attempt) {
await enqueue(async () => {
assertAvailable();
queueContinuation();
});
return;
}
try {
await runCheckpoint(attempt);
await runCheckpoint(attempt, preparedCheckpointWindow);
} catch {
// Recovery committed before the replay began, so a failed replay
// degrades instead of bricking the runtime: drop the pending
@ -990,6 +1064,11 @@ export function createGoalRuntime(
await settleDanglingAttempt(attempt.permit);
}
});
return restoreActivation;
},
async restore(records: readonly GoalRecoveryRecord[]): Promise<void> {
await this.prepareRestore(records);
await this.activateRestoredWork();
},
bindHost(nextHost: GoalTurnHost): () => void {
assertOperational();

View file

@ -295,6 +295,10 @@ export * from './services/sessionRecap.js';
export * from './services/session-artifact-persistence.js';
export * from './services/session-reference-service.js';
export * from './services/sessionService.js';
export {
collectSessionTurnState,
computeInitialTurnFromHistory,
} from './services/session-turn-state.js';
export * from './services/session-writer-lease.js';
export {
decodeSessionTranscriptCursor,
@ -315,6 +319,12 @@ export {
SessionTranscriptTooLargeError,
} from './services/session-transcript-reader.js';
export type {
SelectiveSessionRestoreOptions,
SessionLiveRestoreProjection,
SessionRestoreProjection,
SessionRestoreReplayPage,
SessionRestoreReplaySelection,
SessionRuntimeResumeState,
SessionTranscriptCursorState,
SessionTranscriptReadPageOptions,
SessionTranscriptRecordPage,

View file

@ -1756,6 +1756,48 @@ describe('ChatRecordingService', () => {
});
describe('legacy recorder', () => {
it('restores reduced recorder state without the full conversation', async () => {
const service = new ChatRecordingService(mockConfig, undefined, false, {
lastCompletedUuid: 'projected-leaf',
turnParentUuids: [null, 'projected-parent'],
customTitle: 'Projected title',
titleSource: 'manual',
parentSessionId: 'parent-session',
sourceType: 'channel',
sourceId: 'channel-main',
});
service.recordUserMessage([{ text: 'next' }]);
await service.flush();
const record = vi.mocked(jsonl.writeLine).mock.calls[0][1] as ChatRecord;
expect(record.parentUuid).toBe('projected-leaf');
expect(service.getCurrentCustomTitle()).toBe('Projected title');
expect(service.getCurrentTitleSource()).toBe('manual');
vi.mocked(jsonl.writeLine).mockClear();
await expect(service.recordParentSession('parent-session')).resolves.toBe(
true,
);
await expect(
service.recordSessionSource('channel', 'channel-main'),
).resolves.toBe(true);
expect(jsonl.writeLine).not.toHaveBeenCalled();
});
it('activates a leased recorder from reduced state', async () => {
const service = new ChatRecordingService(mockConfig);
service.activate(mockLease, undefined, undefined, {
lastCompletedUuid: 'leased-projected-leaf',
turnParentUuids: [null],
});
service.recordUserMessage([{ text: 'next' }]);
await service.flush();
const record = vi.mocked(jsonl.writeLine).mock.calls[0][1] as ChatRecord;
expect(record.parentUuid).toBe('leased-projected-leaf');
});
it('uses the effective session writer lease gate by default', async () => {
mockConfig.getExperimentalZedIntegration = vi.fn().mockReturnValue(true);
mockConfig.isSessionWriterLeaseEnabled = vi.fn().mockReturnValue(false);

View file

@ -574,6 +574,16 @@ export type ChatRecordingFailureListener = (
event: ChatRecordingFailureEvent,
) => void | Promise<void>;
export interface ChatRecordingRestoreState {
lastCompletedUuid: string;
turnParentUuids: Array<string | null>;
customTitle?: string;
titleSource?: TitleSource;
parentSessionId?: string;
sourceType?: string;
sourceId?: string;
}
/**
* Service for recording the current chat session to disk.
*
@ -716,25 +726,31 @@ export class ChatRecordingService {
writerLeaseRequired = config.isSessionWriterLeaseEnabled?.() ??
config.getExperimentalZedIntegration?.() ??
true,
restoreState?: ChatRecordingRestoreState,
) {
this.config = config;
this.writerLeaseRequired = writerLeaseRequired;
const resumed = config.getResumedSessionData();
if (writerLeaseRequired) {
this.lastRecordUuid = resumed?.lastCompletedUuid ?? null;
this.lastRecordUuid =
restoreState?.lastCompletedUuid ?? resumed?.lastCompletedUuid ?? null;
this.lastPersistedRecordUuid = this.lastRecordUuid;
} else {
this.state = 'active';
this.acceptingWrites = true;
this.restoreSessionState(
resumed
? {
conversation: resumed.conversation ?? { messages: [] },
lastCompletedUuid: resumed.lastCompletedUuid,
}
: undefined,
resumed ? this.readPersistedTitleInfo() : undefined,
);
if (restoreState) {
this.restoreProjectedState(restoreState);
} else {
this.restoreSessionState(
resumed
? {
conversation: resumed.conversation ?? { messages: [] },
lastCompletedUuid: resumed.lastCompletedUuid,
}
: undefined,
resumed ? this.readPersistedTitleInfo() : undefined,
);
}
}
}
@ -866,6 +882,20 @@ export class ChatRecordingService {
}
}
private restoreProjectedState(state: ChatRecordingRestoreState): void {
this.lastRecordUuid = state.lastCompletedUuid;
this.lastPersistedRecordUuid = state.lastCompletedUuid;
this.turnParentUuids = [...state.turnParentUuids];
this.currentCustomTitle = state.customTitle;
this.currentTitleSource = state.titleSource;
this.currentParentSessionId = state.parentSessionId;
this.currentSourceType = state.sourceType;
this.currentSourceId = state.sourceId;
if (this.currentCustomTitle) {
this.bytesSinceTitleAnchor = TITLE_REANCHOR_BYTES;
}
}
activate(
lease: SessionWriterLease,
sessionData?: {
@ -873,6 +903,7 @@ export class ChatRecordingService {
lastCompletedUuid: string | null;
},
persistedTitleInfo?: { title?: string; source?: TitleSource },
restoreState?: ChatRecordingRestoreState,
): void {
if (
!this.writerLeaseRequired ||
@ -882,7 +913,11 @@ export class ChatRecordingService {
throw new SessionWriterUnavailableError();
}
this.binding = { sessionId: lease.sessionId, lease };
this.restoreSessionState(sessionData, persistedTitleInfo);
if (restoreState) {
this.restoreProjectedState(restoreState);
} else {
this.restoreSessionState(sessionData, persistedTitleInfo);
}
this.state = 'active';
this.acceptingWrites = true;
}

View file

@ -0,0 +1,130 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Content, Part } from '@google/genai';
import type {
ChatCompressionRecordPayload,
ChatRecord,
} from './chatRecordingService.js';
export interface BuildApiHistoryOptions {
/**
* Whether to strip thought parts from the history.
* Thought parts are content parts that have `thought: true`.
* Keeping thoughts ensures `reasoning_content` from reasoning models
* (e.g. DeepSeek) is properly passed back in subsequent API calls.
* @default false
*/
stripThoughtsFromHistory?: boolean;
}
function stripThoughtsFromContent(content: Content): Content | null {
if (!content.parts) return content;
const filteredParts = content.parts.filter((part) => !(part as Part).thought);
if (filteredParts.length === 0) return null;
return { ...content, parts: filteredParts };
}
function copyContentForApiHistory(content: Content): Content {
return {
...content,
parts: content.parts?.map((part) => {
if ('functionCall' in part && part.functionCall) {
return {
...part,
functionCall: {
...part.functionCall,
args: part.functionCall.args
? { ...part.functionCall.args }
: part.functionCall.args,
},
};
}
if ('functionResponse' in part && part.functionResponse) {
return {
...part,
functionResponse: { ...part.functionResponse },
};
}
return { ...part };
}),
};
}
function appendApiHistoryRecord(history: Content[], record: ChatRecord): void {
if (!record.message || record.subtype === 'realtime_message') return;
const message = copyContentForApiHistory(record.message);
if (record.subtype === 'mid_turn_user_message') {
const previous = history.at(-1);
if (previous?.role === 'user') {
previous.parts = [...(previous.parts ?? []), ...(message.parts ?? [])];
return;
}
}
history.push(message);
}
export class SessionApiHistoryAccumulator {
private history: Content[] = [];
private compressionCandidate: unknown;
add(record: ChatRecord): void {
if (record.type === 'system') {
if (!isApiHistoryCompressionCandidate(record)) return;
const payload = record.systemPayload as ChatCompressionRecordPayload;
this.compressionCandidate = payload.compressedHistory;
this.history = Array.isArray(payload.compressedHistory)
? payload.compressedHistory.map(copyContentForApiHistory)
: [];
return;
}
if (
this.compressionCandidate !== undefined &&
!Array.isArray(this.compressionCandidate)
) {
return;
}
appendApiHistoryRecord(this.history, record);
}
finish(options: BuildApiHistoryOptions = {}): Content[] {
if (
this.compressionCandidate !== undefined &&
!Array.isArray(this.compressionCandidate)
) {
return (this.compressionCandidate as Content[]).map(
copyContentForApiHistory,
);
}
if (!options.stripThoughtsFromHistory) return this.history;
return this.history
.map(stripThoughtsFromContent)
.filter((content): content is Content => content !== null);
}
}
export function isApiHistoryCompressionCandidate(record: ChatRecord): boolean {
if (record.type !== 'system' || record.subtype !== 'chat_compression') {
return false;
}
const payload = record.systemPayload as
| ChatCompressionRecordPayload
| undefined;
return Boolean(payload?.compressedHistory);
}
export function buildApiHistoryFromConversation(
conversation: { messages: readonly ChatRecord[] },
options: BuildApiHistoryOptions = {},
): Content[] {
const accumulator = new SessionApiHistoryAccumulator();
for (const record of conversation.messages) accumulator.add(record);
return accumulator.finish(options);
}

View file

@ -159,6 +159,84 @@ export function isSessionArtifactRecord(
return isTranscriptArtifactRecord(record);
}
export function selectActiveSideArtifactRecordUuids(
records: ReadonlyArray<
SessionArtifactChatRecordLike & {
uuid: string;
parentUuid: string | null;
}
>,
activeRecordUuids: readonly string[],
): string[] {
const activeUuids = new Set(activeRecordUuids);
const firstActiveUuid = activeRecordUuids[0];
const firstActiveIndex =
firstActiveUuid === undefined
? -1
: records.findIndex((record) => record.uuid === firstActiveUuid);
const nextActiveUuidByIndex = new Map<number, string>();
const nextBlockingUuidByIndex = new Map<number, string>();
let nextActiveUuid: string | undefined;
let nextBlockingUuid: string | undefined;
for (let index = records.length - 1; index >= 0; index--) {
if (nextActiveUuid !== undefined) {
nextActiveUuidByIndex.set(index, nextActiveUuid);
}
if (nextBlockingUuid !== undefined) {
nextBlockingUuidByIndex.set(index, nextBlockingUuid);
}
const record = records[index]!;
if (activeUuids.has(record.uuid)) {
nextActiveUuid = record.uuid;
nextBlockingUuid = undefined;
} else if (
!isSessionArtifactRecord(record) &&
!(record.type === 'system' && record.subtype === 'custom_title')
) {
nextBlockingUuid = record.uuid;
}
}
const selected: string[] = [];
const includedSideArtifactUuids = new Set<string>();
let previousActiveUuid: string | undefined;
for (let index = 0; index < records.length; index++) {
const record = records[index]!;
if (activeUuids.has(record.uuid)) {
previousActiveUuid = record.uuid;
continue;
}
if (!isSessionArtifactRecord(record)) continue;
const nextUuid = nextActiveUuidByIndex.get(index);
const isInActiveSegment =
!nextBlockingUuidByIndex.has(index) &&
(nextUuid !== undefined
? activeUuids.has(nextUuid)
: previousActiveUuid !== undefined &&
activeUuids.has(previousActiveUuid));
if (
record.parentUuid !== null &&
(activeUuids.has(record.parentUuid) ||
includedSideArtifactUuids.has(record.parentUuid)) &&
isInActiveSegment &&
(record.parentUuid === previousActiveUuid ||
includedSideArtifactUuids.has(record.parentUuid))
) {
selected.push(record.uuid);
includedSideArtifactUuids.add(record.uuid);
} else if (
record.parentUuid === null &&
index < firstActiveIndex &&
isInActiveSegment
) {
selected.push(record.uuid);
includedSideArtifactUuids.add(record.uuid);
}
}
return selected;
}
export function stableSessionArtifactId(
sessionId: string,
identityKey: string,
@ -181,80 +259,90 @@ export function sessionArtifactIdentityKey(
return undefined;
}
export function rebuildSessionArtifactSnapshot(
records: readonly SessionArtifactChatRecordLike[],
fallbackSessionId?: string,
): RebuiltSessionArtifactSnapshot | undefined {
const artifacts = new Map<string, PersistedSessionArtifact>();
const tombstonedIds = new Set<string>();
const stickyEphemeralIds = new Set<string>();
const markerArtifacts = new Map<string, PersistedSessionArtifact>();
const warnings: string[] = [];
let sequence = 0;
let lastSnapshotSequence = 0;
let sessionId = fallbackSessionId;
let sawRecord = false;
export class SessionArtifactSnapshotAccumulator {
private readonly artifacts = new Map<string, PersistedSessionArtifact>();
private readonly tombstonedIds = new Set<string>();
private readonly stickyEphemeralIds = new Set<string>();
private readonly markerArtifacts = new Map<
string,
PersistedSessionArtifact
>();
private readonly warnings: string[] = [];
private sequence = 0;
private lastSnapshotSequence = 0;
private sessionId: string | undefined;
private sawRecord = false;
for (const record of records) {
if (!isSessionArtifactRecord(record)) continue;
constructor(fallbackSessionId?: string) {
this.sessionId = fallbackSessionId;
}
add(record: SessionArtifactChatRecordLike): void {
if (!isSessionArtifactRecord(record)) return;
if (record.subtype === 'session_artifact_snapshot') {
const payload = normalizeSnapshotPayload(record.systemPayload, warnings);
if (!payload) continue;
sawRecord = true;
sessionId = payload.sessionId;
sequence = Math.max(sequence, payload.sequence);
lastSnapshotSequence = payload.sequence;
artifacts.clear();
tombstonedIds.clear();
stickyEphemeralIds.clear();
markerArtifacts.clear();
for (const id of payload.tombstonedIds ?? []) tombstonedIds.add(id);
const payload = normalizeSnapshotPayload(
record.systemPayload,
this.warnings,
);
if (!payload) return;
this.sawRecord = true;
this.sessionId = payload.sessionId;
this.sequence = Math.max(this.sequence, payload.sequence);
this.lastSnapshotSequence = payload.sequence;
this.artifacts.clear();
this.tombstonedIds.clear();
this.stickyEphemeralIds.clear();
this.markerArtifacts.clear();
for (const id of payload.tombstonedIds ?? []) this.tombstonedIds.add(id);
for (const id of payload.stickyEphemeralIds ?? []) {
stickyEphemeralIds.add(id);
this.stickyEphemeralIds.add(id);
}
const markerIds = new Set([...tombstonedIds, ...stickyEphemeralIds]);
const markerIds = new Set([
...this.tombstonedIds,
...this.stickyEphemeralIds,
]);
for (const artifact of payload.markerArtifacts ?? []) {
if (markerIds.has(artifact.id)) {
markerArtifacts.set(artifact.id, artifact);
this.markerArtifacts.set(artifact.id, artifact);
}
}
for (const artifact of payload.artifacts) {
if (artifact.retention === 'ephemeral') continue;
artifacts.set(artifact.id, artifact);
markerArtifacts.delete(artifact.id);
this.artifacts.set(artifact.id, artifact);
this.markerArtifacts.delete(artifact.id);
}
continue;
return;
}
const payload = normalizeEventPayload(record.systemPayload, warnings);
if (!payload) continue;
sawRecord = true;
sessionId = payload.sessionId;
if (payload.sequence <= lastSnapshotSequence) {
warnings.push(
`skipped stale event sequence ${payload.sequence} at or before snapshot sequence ${lastSnapshotSequence}`,
const payload = normalizeEventPayload(record.systemPayload, this.warnings);
if (!payload) return;
this.sawRecord = true;
this.sessionId = payload.sessionId;
if (payload.sequence <= this.lastSnapshotSequence) {
this.warnings.push(
`skipped stale event sequence ${payload.sequence} at or before snapshot sequence ${this.lastSnapshotSequence}`,
);
continue;
return;
}
sequence = Math.max(sequence, payload.sequence);
this.sequence = Math.max(this.sequence, payload.sequence);
for (const change of payload.changes) {
if (change.action === 'removed') {
artifacts.delete(change.artifactId);
this.artifacts.delete(change.artifactId);
if (change.reason === 'explicit') {
tombstonedIds.add(change.artifactId);
stickyEphemeralIds.delete(change.artifactId);
this.tombstonedIds.add(change.artifactId);
this.stickyEphemeralIds.delete(change.artifactId);
if (change.artifact) {
markerArtifacts.set(change.artifactId, change.artifact);
this.markerArtifacts.set(change.artifactId, change.artifact);
}
}
if (change.reason === 'eviction') {
stickyEphemeralIds.delete(change.artifactId);
markerArtifacts.delete(change.artifactId);
this.stickyEphemeralIds.delete(change.artifactId);
this.markerArtifacts.delete(change.artifactId);
}
if (change.reason === 'unpin_to_ephemeral') {
stickyEphemeralIds.add(change.artifactId);
this.stickyEphemeralIds.add(change.artifactId);
if (change.artifact) {
markerArtifacts.set(change.artifactId, change.artifact);
this.markerArtifacts.set(change.artifactId, change.artifact);
}
}
continue;
@ -262,29 +350,38 @@ export function rebuildSessionArtifactSnapshot(
if (!change.artifact || change.artifact.retention === 'ephemeral') {
continue;
}
artifacts.set(change.artifact.id, change.artifact);
tombstonedIds.delete(change.artifact.id);
stickyEphemeralIds.delete(change.artifact.id);
markerArtifacts.delete(change.artifact.id);
this.artifacts.set(change.artifact.id, change.artifact);
this.tombstonedIds.delete(change.artifact.id);
this.stickyEphemeralIds.delete(change.artifact.id);
this.markerArtifacts.delete(change.artifact.id);
}
}
if (!sawRecord || !sessionId) {
return undefined;
}
finish(): RebuiltSessionArtifactSnapshot | undefined {
if (!this.sawRecord || !this.sessionId) return undefined;
return {
v: SESSION_ARTIFACT_PERSISTENCE_VERSION,
sessionId,
sequence,
artifacts: Array.from(artifacts.values()),
tombstonedIds: Array.from(tombstonedIds),
stickyEphemeralIds: Array.from(stickyEphemeralIds),
...(markerArtifacts.size > 0
? { markerArtifacts: Array.from(markerArtifacts.values()) }
: {}),
warnings,
};
return {
v: SESSION_ARTIFACT_PERSISTENCE_VERSION,
sessionId: this.sessionId,
sequence: this.sequence,
artifacts: Array.from(this.artifacts.values()),
tombstonedIds: Array.from(this.tombstonedIds),
stickyEphemeralIds: Array.from(this.stickyEphemeralIds),
...(this.markerArtifacts.size > 0
? { markerArtifacts: Array.from(this.markerArtifacts.values()) }
: {}),
warnings: this.warnings,
};
}
}
export function rebuildSessionArtifactSnapshot(
records: readonly SessionArtifactChatRecordLike[],
fallbackSessionId?: string,
): RebuiltSessionArtifactSnapshot | undefined {
const accumulator = new SessionArtifactSnapshotAccumulator(fallbackSessionId);
for (const record of records) accumulator.add(record);
return accumulator.finish();
}
export function remapSessionArtifactPayloadForFork(

View file

@ -0,0 +1,90 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { describe, expect, it } from 'vitest';
import type {
ChatRecord,
FileHistorySnapshotRecordPayload,
} from './chatRecordingService.js';
import { MAX_SNAPSHOTS } from './fileHistoryService.js';
import { SessionFileHistoryAccumulator } from './session-file-history-state.js';
function snapshotRecord(
snapshots: FileHistorySnapshotRecordPayload['snapshots'],
): Pick<ChatRecord, 'type' | 'subtype' | 'systemPayload'> {
return {
type: 'system',
subtype: 'file_history_snapshot',
systemPayload: { snapshots },
};
}
function snapshot(promptId: string, timestamp: string) {
return {
promptId,
timestamp,
trackedFileBackups: {},
};
}
describe('SessionFileHistoryAccumulator', () => {
it('keeps the final 100 first-insertion slots with retained replacements', () => {
const accumulator = new SessionFileHistoryAccumulator();
accumulator.add(
snapshotRecord(
Array.from({ length: MAX_SNAPSHOTS + 1 }, (_, index) =>
snapshot(
`prompt-${index}`,
`2026-01-01T00:00:${String(index % 60).padStart(2, '0')}.000Z`,
),
),
),
);
accumulator.add(
snapshotRecord([
snapshot('prompt-0', '2026-02-01T00:00:00.000Z'),
snapshot('prompt-50', '2026-03-01T00:00:00.000Z'),
]),
);
const restored = accumulator.finish();
expect(restored).toHaveLength(MAX_SNAPSHOTS);
expect(restored?.map((item) => item.promptId)).toEqual(
Array.from(
{ length: MAX_SNAPSHOTS },
(_, index) => `prompt-${index + 1}`,
),
);
expect(
restored?.find((item) => item.promptId === 'prompt-50')?.timestamp,
).toEqual(new Date('2026-03-01T00:00:00.000Z'));
});
it('does not partially apply a malformed snapshot batch', () => {
const accumulator = new SessionFileHistoryAccumulator();
accumulator.add(
snapshotRecord([snapshot('before', '2026-01-01T00:00:00.000Z')]),
);
expect(() =>
accumulator.add(
snapshotRecord([
snapshot('partial', '2026-01-02T00:00:00.000Z'),
{
promptId: 'malformed',
timestamp: '2026-01-03T00:00:00.000Z',
trackedFileBackups: null,
} as never,
]),
),
).toThrow();
expect(accumulator.finish()?.map((item) => item.promptId)).toEqual([
'before',
]);
});
});

View file

@ -0,0 +1,56 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type {
ChatRecord,
FileHistorySnapshotRecordPayload,
} from './chatRecordingService.js';
import {
deserializeSnapshots,
MAX_SNAPSHOTS,
type FileHistorySnapshot,
} from './fileHistoryService.js';
export class SessionFileHistoryAccumulator {
private readonly seenPromptIds = new Set<string>();
private readonly retainedPromptIds: string[] = [];
private readonly snapshotsByPromptId = new Map<string, FileHistorySnapshot>();
add(record: Pick<ChatRecord, 'type' | 'subtype' | 'systemPayload'>): void {
if (
record.type !== 'system' ||
record.subtype !== 'file_history_snapshot' ||
!record.systemPayload
) {
return;
}
const payload = record.systemPayload as FileHistorySnapshotRecordPayload;
if (!Array.isArray(payload.snapshots)) return;
const deserialized = deserializeSnapshots(payload.snapshots);
for (const snapshot of deserialized) {
if (this.seenPromptIds.has(snapshot.promptId)) {
if (this.snapshotsByPromptId.has(snapshot.promptId)) {
this.snapshotsByPromptId.set(snapshot.promptId, snapshot);
}
continue;
}
this.seenPromptIds.add(snapshot.promptId);
this.retainedPromptIds.push(snapshot.promptId);
this.snapshotsByPromptId.set(snapshot.promptId, snapshot);
if (this.retainedPromptIds.length > MAX_SNAPSHOTS) {
const evictedPromptId = this.retainedPromptIds.shift()!;
this.snapshotsByPromptId.delete(evictedPromptId);
}
}
}
finish(): FileHistorySnapshot[] | undefined {
const snapshots = this.retainedPromptIds.map(
(promptId) => this.snapshotsByPromptId.get(promptId)!,
);
return snapshots.length > 0 ? snapshots : undefined;
}
}

View file

@ -0,0 +1,81 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type {
ChatCompressionRecordPayload,
ChatRecord,
} from './chatRecordingService.js';
import { getUsageOutputTokenCountForPromptEstimate } from './tokenEstimation.js';
export interface ResumeTokenCounts {
promptTokenCount: number;
outputTokenCount: number;
isEstimated: boolean;
}
export class ResumeTokenCountsAccumulator {
private value: ResumeTokenCounts | undefined;
add(record: ChatRecord): void {
if (record.type === 'assistant') {
const usage = record.usageMetadata;
const candidate = usage?.promptTokenCount ?? usage?.totalTokenCount;
if (candidate) {
this.value = {
promptTokenCount: candidate,
outputTokenCount: getUsageOutputTokenCountForPromptEstimate(usage),
isEstimated: false,
};
}
return;
}
if (record.type === 'system' && record.subtype === 'chat_compression') {
const payload = record.systemPayload as
| ChatCompressionRecordPayload
| undefined;
if (payload?.info) {
this.value = {
promptTokenCount: payload.info.newTokenCount,
outputTokenCount: 0,
isEstimated: payload.info.newTokenCountIsEstimated ?? true,
};
}
}
}
finish(): ResumeTokenCounts | undefined {
return this.value;
}
}
export function isResumeTokenCountsCandidate(record: ChatRecord): boolean {
if (record.type === 'assistant') {
const usage = record.usageMetadata;
return Boolean(usage?.promptTokenCount ?? usage?.totalTokenCount);
}
if (record.type !== 'system' || record.subtype !== 'chat_compression') {
return false;
}
const payload = record.systemPayload as
| ChatCompressionRecordPayload
| undefined;
return payload?.info !== undefined;
}
export function getResumeTokenCounts(conversation: {
messages: readonly ChatRecord[];
}): ResumeTokenCounts | undefined {
const accumulator = new ResumeTokenCountsAccumulator();
for (const record of conversation.messages) accumulator.add(record);
return accumulator.finish();
}
export function getResumePromptTokenCount(conversation: {
messages: readonly ChatRecord[];
}): number | undefined {
return getResumeTokenCounts(conversation)?.promptTokenCount;
}

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,154 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { ChatRecord } from './chatRecordingService.js';
export interface SessionTurnState {
initialTurn: number;
turnParentUuids: Array<string | null>;
backgroundNotificationTaskIds: string[];
}
export interface SessionTurnRecordHint {
promptTurn?: number;
countsAsUserPrompt: boolean;
turnParentUuid?: string | null;
backgroundNotificationTaskId?: string;
}
export class SessionTurnStateAccumulator {
private maxPromptTurn = 0;
private userMessageCount = 0;
private readonly turnParentUuids: Array<string | null> = [];
private readonly backgroundNotificationTaskIds = new Set<string>();
constructor(private readonly sessionId: string) {}
add(record: ChatRecord): void {
this.addHint(getSessionTurnRecordHint(record, this.sessionId));
}
addHint(hint: SessionTurnRecordHint): void {
if (hint.countsAsUserPrompt) {
this.userMessageCount += 1;
}
if (hint.promptTurn !== undefined) {
this.maxPromptTurn = Math.max(this.maxPromptTurn, hint.promptTurn);
}
if (hint.turnParentUuid !== undefined) {
this.turnParentUuids.push(hint.turnParentUuid);
}
if (hint.backgroundNotificationTaskId !== undefined) {
this.backgroundNotificationTaskIds.add(hint.backgroundNotificationTaskId);
}
}
finish(): SessionTurnState {
return {
initialTurn:
this.maxPromptTurn > 0 ? this.maxPromptTurn : this.userMessageCount,
turnParentUuids: [...this.turnParentUuids],
backgroundNotificationTaskIds: [...this.backgroundNotificationTaskIds],
};
}
}
export function getSessionTurnRecordHint(
record: ChatRecord,
sessionId: string,
): SessionTurnRecordHint {
let promptTurn: number | undefined;
for (const promptId of getRecordPromptIds(record)) {
const candidate = parseSessionPromptTurn(promptId, sessionId);
if (candidate !== undefined) {
promptTurn = Math.max(promptTurn ?? 0, candidate);
}
}
const turnParentUuid =
record.type === 'user' &&
record.subtype !== 'goal_runtime' &&
record.subtype !== 'notification' &&
record.subtype !== 'cron' &&
record.subtype !== 'mid_turn_user_message' &&
record.subtype !== 'realtime_message'
? (record.parentUuid ?? null)
: undefined;
const backgroundTask =
record.subtype === 'notification'
? (
record.systemPayload as
| { backgroundTask?: { taskId?: unknown } }
| undefined
)?.backgroundTask
: undefined;
return {
...(promptTurn !== undefined ? { promptTurn } : {}),
countsAsUserPrompt:
record.sessionId === sessionId && isUserPromptRecord(record),
...(turnParentUuid !== undefined ? { turnParentUuid } : {}),
...(typeof backgroundTask?.taskId === 'string'
? { backgroundNotificationTaskId: backgroundTask.taskId }
: {}),
};
}
export function collectSessionTurnState(
records: readonly ChatRecord[],
sessionId: string,
): SessionTurnState {
const accumulator = new SessionTurnStateAccumulator(sessionId);
for (const record of records) accumulator.add(record);
return accumulator.finish();
}
export function computeInitialTurnFromHistory(
records: readonly ChatRecord[],
sessionId: string,
): number {
return collectSessionTurnState(records, sessionId).initialTurn;
}
function getRecordPromptIds(record: ChatRecord): string[] {
const promptIds: string[] = [];
const recordPromptId = (record as { promptId?: unknown }).promptId;
if (typeof recordPromptId === 'string') promptIds.push(recordPromptId);
const telemetryPromptId = readTelemetryPromptId(record.systemPayload);
if (telemetryPromptId) promptIds.push(telemetryPromptId);
return promptIds;
}
function readTelemetryPromptId(payload: unknown): string | undefined {
if (!payload || typeof payload !== 'object' || !('uiEvent' in payload)) {
return undefined;
}
const uiEvent = (payload as { uiEvent?: unknown }).uiEvent;
if (!uiEvent || typeof uiEvent !== 'object' || !('prompt_id' in uiEvent)) {
return undefined;
}
const promptId = (uiEvent as { prompt_id?: unknown }).prompt_id;
return typeof promptId === 'string' ? promptId : undefined;
}
function parseSessionPromptTurn(
promptId: string,
sessionId: string,
): number | undefined {
const promptIdPrefix = `${sessionId}########`;
if (!promptId.startsWith(promptIdPrefix)) return undefined;
const suffix = promptId.slice(promptIdPrefix.length);
return /^\d+$/.test(suffix) ? Number(suffix) : undefined;
}
function isUserPromptRecord(record: ChatRecord): boolean {
if (record.type !== 'user' || record.subtype === 'realtime_message') {
return false;
}
return (
record.message?.parts?.some(
(part) => typeof part.text === 'string' && part.text.trim().length > 0,
) ?? false
);
}

View file

@ -16,7 +16,6 @@ import * as jsonl from '../utils/jsonl-utils.js';
import type { HistoryGap } from '../utils/conversation-chain.js';
import { prepareTranscriptRecords } from '../utils/transcript-records.js';
import type {
ChatCompressionRecordPayload,
ChatRecord,
FileHistorySnapshotRecordPayload,
TitleSource,
@ -27,31 +26,50 @@ import type { FileHistorySnapshot } from './fileHistoryService.js';
import {
deserializeSnapshots,
FILE_HISTORY_DIR,
MAX_SNAPSHOTS,
serializeSnapshot,
} from './fileHistoryService.js';
import { SessionFileHistoryAccumulator } from './session-file-history-state.js';
import { uiTelemetryService } from '../telemetry/uiTelemetry.js';
import { createDebugLogger } from '../utils/debugLogger.js';
import { readRuntimeStatus } from '../utils/runtimeStatus.js';
import {
LITE_READ_BUF_SIZE,
readLastJsonStringFieldSync,
readLastJsonStringFieldsSync,
readSessionTitleInfoFromFileSync,
} from '../utils/sessionStorageUtils.js';
import { getUsageOutputTokenCountForPromptEstimate } from './tokenEstimation.js';
import {
isSessionArtifactRecord,
rebuildSessionArtifactSnapshot,
remapSessionArtifactPayloadForFork,
selectActiveSideArtifactRecordUuids,
type RebuiltSessionArtifactSnapshot,
} from './session-artifact-persistence.js';
import { SessionOrganizationService } from './session-organization-service.js';
import { SessionTranscriptTooLargeError } from './session-transcript-reader.js';
import {
SessionTranscriptReader,
SessionTranscriptTooLargeError,
type SelectiveSessionRestoreOptions,
type SessionLiveRestoreProjection,
type SessionRestoreProjection,
} from './session-transcript-reader.js';
import {
SessionWriterLease,
SessionWriterUnavailableError,
type SessionWriterProcessKind,
} from './session-writer-lease.js';
export {
buildApiHistoryFromConversation,
type BuildApiHistoryOptions,
} from './session-api-history.js';
import {
getResumeTokenCounts,
type ResumeTokenCounts,
} from './session-resume-token-counts.js';
export {
getResumePromptTokenCount,
getResumeTokenCounts,
type ResumeTokenCounts,
} from './session-resume-token-counts.js';
const debugLogger = createDebugLogger('SESSION');
@ -332,12 +350,18 @@ export class SessionService {
private readonly projectHash: string;
private readonly projectRoot: string;
private readonly onWarning: ((message: string) => void) | undefined;
private readonly transcriptReader: SessionTranscriptReader;
constructor(cwd: string, options: SessionServiceOptions = {}) {
this.storage = new Storage(cwd, options.runtimeBaseDir);
this.projectRoot = cwd;
this.projectHash = getProjectHash(cwd);
this.onWarning = options.onWarning;
this.transcriptReader = new SessionTranscriptReader(
cwd,
undefined,
options.runtimeBaseDir,
);
}
/** The workspace root this service is bound to (the cwd it was constructed
@ -696,19 +720,7 @@ export class SessionService {
title?: string;
source?: TitleSource;
} {
const hit = readLastJsonStringFieldsSync(
filePath,
'customTitle',
['titleSource'],
'"subtype":"custom_title"',
tailBuffer,
);
const title = hit['customTitle'];
if (!title) return {};
const rawSource = hit['titleSource'];
const source =
rawSource === 'auto' || rawSource === 'manual' ? rawSource : undefined;
return { title, source };
return readSessionTitleInfoFromFileSync(filePath, tailBuffer);
}
/**
@ -1273,6 +1285,26 @@ export class SessionService {
return this.loadSessionFromState(sessionId, 'active');
}
async readRestoreProjection(
sessionId: string,
options: SelectiveSessionRestoreOptions,
): Promise<SessionRestoreProjection | undefined> {
return this.transcriptReader.readRestoreProjection(sessionId, options, {
validateFirstRecord: (record) =>
this.sessionBelongsToCurrentProject(record.sessionId, record.cwd),
});
}
async readLiveRestoreProjection(
sessionId: string,
options: SelectiveSessionRestoreOptions,
): Promise<SessionLiveRestoreProjection | undefined> {
return this.transcriptReader.readLiveRestoreProjection(sessionId, options, {
validateFirstRecord: (record) =>
this.sessionBelongsToCurrentProject(record.sessionId, record.cwd),
});
}
/**
* Reads an archived session without changing its archive state.
* Daemon load/resume paths must continue to use {@link loadSession}.
@ -1362,40 +1394,17 @@ export class SessionService {
};
// Extract file history snapshots for /rewind across resume
const fileHistorySnapshots: FileHistorySnapshot[] = [];
const seenPromptIds = new Map<string, number>();
const fileHistoryAccumulator = new SessionFileHistoryAccumulator();
for (const msg of messages) {
if (
msg.type === 'system' &&
msg.subtype === 'file_history_snapshot' &&
msg.systemPayload
) {
const payload = msg.systemPayload as FileHistorySnapshotRecordPayload;
if (!Array.isArray(payload?.snapshots)) continue;
let deserialized: FileHistorySnapshot[];
try {
deserialized = deserializeSnapshots(payload.snapshots);
} catch (e) {
debugLogger.warn(
`loadSession: skipping malformed file_history_snapshot: ${e}`,
);
continue;
}
for (const s of deserialized) {
const existingIdx = seenPromptIds.get(s.promptId);
if (existingIdx !== undefined) {
fileHistorySnapshots[existingIdx] = s;
} else {
seenPromptIds.set(s.promptId, fileHistorySnapshots.length);
fileHistorySnapshots.push(s);
}
}
try {
fileHistoryAccumulator.add(msg);
} catch (e) {
debugLogger.warn(
`loadSession: skipping malformed file_history_snapshot: ${e}`,
);
}
}
const cappedSnapshots =
fileHistorySnapshots.length > MAX_SNAPSHOTS
? fileHistorySnapshots.slice(-MAX_SNAPSHOTS)
: fileHistorySnapshots;
const fileHistorySnapshots = fileHistoryAccumulator.finish();
const activeBranchRecords = includeActiveSideArtifactRecords(
records,
messages,
@ -1409,8 +1418,7 @@ export class SessionService {
conversation,
filePath,
lastCompletedUuid: lastMessage.uuid,
fileHistorySnapshots:
cappedSnapshots.length > 0 ? cappedSnapshots : undefined,
fileHistorySnapshots,
...(artifactSnapshot ? { artifactSnapshot } : {}),
historyGaps: gaps.length > 0 ? gaps : undefined,
};
@ -2192,151 +2200,6 @@ export class SessionService {
}
}
/**
* Options for building API history from conversation.
*/
export interface BuildApiHistoryOptions {
/**
* Whether to strip thought parts from the history.
* Thought parts are content parts that have `thought: true`.
* Keeping thoughts ensures `reasoning_content` from reasoning models
* (e.g. DeepSeek) is properly passed back in subsequent API calls.
* @default false
*/
stripThoughtsFromHistory?: boolean;
}
/**
* Strips thought parts from a Content object.
* Thought parts are identified by having `thought: true`.
* Returns null if the content only contained thought parts.
*/
function stripThoughtsFromContent(content: Content): Content | null {
if (!content.parts) return content;
const filteredParts = content.parts.filter((part) => !(part as Part).thought);
// If all parts were thoughts, remove the entire content
if (filteredParts.length === 0) {
return null;
}
return {
...content,
parts: filteredParts,
};
}
function copyContentForApiHistory(content: Content): Content {
return {
...content,
parts: content.parts?.map((part) => {
if ('functionCall' in part && part.functionCall) {
return {
...part,
functionCall: {
...part.functionCall,
args: part.functionCall.args
? { ...part.functionCall.args }
: part.functionCall.args,
},
};
}
if ('functionResponse' in part && part.functionResponse) {
return {
...part,
functionResponse: {
...part.functionResponse,
},
};
}
return { ...part };
}),
};
}
function appendApiHistoryRecord(history: Content[], record: ChatRecord): void {
if (!record.message || record.subtype === 'realtime_message') return;
const message = copyContentForApiHistory(record.message as Content);
if (record.subtype === 'mid_turn_user_message') {
const previous = history.at(-1);
if (previous?.role === 'user') {
previous.parts = [...(previous.parts ?? []), ...(message.parts ?? [])];
return;
}
}
history.push(message);
}
/**
* Builds the model-facing chat history (Content[]) from a reconstructed
* conversation. This keeps UI history intact while applying chat compression
* checkpoints for the API history used on resume.
*
* Strategy:
* - Find the latest system/chat_compression record (if any).
* - Use its compressedHistory snapshot as the base history.
* - Append all messages after that checkpoint (skipping system records).
* - If no checkpoint exists, return the linear message list (message field only).
*/
export function buildApiHistoryFromConversation(
conversation: ConversationRecord,
options: BuildApiHistoryOptions = {},
): Content[] {
const { stripThoughtsFromHistory = false } = options;
const { messages } = conversation;
let lastCompressionIndex = -1;
let compressedHistory: Content[] | undefined;
messages.forEach((record, index) => {
if (record.type === 'system' && record.subtype === 'chat_compression') {
const payload = record.systemPayload as
| ChatCompressionRecordPayload
| undefined;
if (payload?.compressedHistory) {
lastCompressionIndex = index;
compressedHistory = payload.compressedHistory;
}
}
});
if (compressedHistory && lastCompressionIndex >= 0) {
const baseHistory: Content[] = compressedHistory.map(
copyContentForApiHistory,
);
// Append everything after the compression record (newer turns)
for (let i = lastCompressionIndex + 1; i < messages.length; i++) {
const record = messages[i];
if (record.type === 'system') continue;
appendApiHistoryRecord(baseHistory, record);
}
if (stripThoughtsFromHistory) {
return baseHistory
.map(stripThoughtsFromContent)
.filter((content): content is Content => content !== null);
}
return baseHistory;
}
// Fallback: return linear messages as Content[]
const result: Content[] = [];
for (const record of messages) {
appendApiHistoryRecord(result, record);
}
if (stripThoughtsFromHistory) {
return result
.map(stripThoughtsFromContent)
.filter((content): content is Content => content !== null);
}
return result;
}
function remapSnapshotPromptId(
snapshot: FileHistorySnapshot,
sourceSessionId: string,
@ -2408,83 +2271,25 @@ function includeActiveSideArtifactRecords(
const activeByUuid = new Map(
activeRecords.map((record) => [record.uuid, record]),
);
const activeUuids = new Set(activeByUuid.keys());
const firstActiveUuid = activeRecords[0]?.uuid;
const firstActiveIndex =
firstActiveUuid === undefined
? -1
: records.findIndex((record) => record.uuid === firstActiveUuid);
const nextActiveUuidByIndex = new Map<number, string>();
const nextBlockingUuidByIndex = new Map<number, string>();
let nextActiveUuid: string | undefined;
let nextBlockingUuid: string | undefined;
for (let index = records.length - 1; index >= 0; index--) {
if (nextActiveUuid !== undefined) {
nextActiveUuidByIndex.set(index, nextActiveUuid);
}
if (nextBlockingUuid !== undefined) {
nextBlockingUuidByIndex.set(index, nextBlockingUuid);
}
if (activeUuids.has(records[index]!.uuid)) {
nextActiveUuid = records[index]!.uuid;
nextBlockingUuid = undefined;
} else if (
!isSessionArtifactRecord(records[index]!) &&
!isTailNeutralSideRecord(records[index]!)
) {
nextBlockingUuid = records[index]!.uuid;
}
}
const artifactUuids = new Set(
selectActiveSideArtifactRecordUuids(
records,
activeRecords.map((record) => record.uuid),
),
);
const selected: ChatRecord[] = [];
const includedSideArtifactUuids = new Set<string>();
let previousActiveUuid: string | undefined;
for (let index = 0; index < records.length; index++) {
const record = records[index]!;
for (const record of records) {
const activeRecord = activeByUuid.get(record.uuid);
if (activeRecord) {
selected.push(activeRecord);
activeByUuid.delete(record.uuid);
previousActiveUuid = record.uuid;
continue;
}
if (!isSessionArtifactRecord(record)) {
continue;
}
const nextUuid = nextActiveUuidByIndex.get(index);
const hasBlockingRecordBeforeNextActive =
nextBlockingUuidByIndex.has(index);
const isInActiveSegment =
!hasBlockingRecordBeforeNextActive &&
(nextUuid !== undefined
? activeUuids.has(nextUuid)
: previousActiveUuid !== undefined &&
activeUuids.has(previousActiveUuid));
if (
record.parentUuid !== null &&
(activeUuids.has(record.parentUuid) ||
includedSideArtifactUuids.has(record.parentUuid)) &&
isInActiveSegment &&
(record.parentUuid === previousActiveUuid ||
includedSideArtifactUuids.has(record.parentUuid))
) {
selected.push(record);
includedSideArtifactUuids.add(record.uuid);
} else if (
record.parentUuid === null &&
index < firstActiveIndex &&
isInActiveSegment
) {
selected.push(record);
includedSideArtifactUuids.add(record.uuid);
}
if (artifactUuids.has(record.uuid)) selected.push(record);
}
return selected;
}
function isTailNeutralSideRecord(record: ChatRecord): boolean {
return record.type === 'system' && record.subtype === 'custom_title';
}
function collectFileHistorySnapshotPromptIds(
records: ChatRecord[],
): Set<string> {
@ -2546,68 +2351,6 @@ export function replayUiTelemetryFromConversation(
return resumeTokenCounts;
}
export interface ResumeTokenCounts {
promptTokenCount: number;
outputTokenCount: number;
isEstimated: boolean;
}
/**
* Returns the best available prompt token count for resuming telemetry.
* Walks backward through messages and returns the first valid value:
* - The latest assistant's non-zero usage (promptTokenCount ?? totalTokenCount).
* - The most recent chat compression checkpoint's newTokenCount.
*/
export function getResumePromptTokenCount(
conversation: ConversationRecord,
): number | undefined {
return getResumeTokenCounts(conversation)?.promptTokenCount;
}
/**
* Returns the prompt and previous-response output token counts used to seed a
* resumed chat. The prompt count restores the context anchor; the output
* count preserves the output tokens appended after that prompt count was
* reported, matching steady-state prompt estimation on the next send.
*/
export function getResumeTokenCounts(
conversation: ConversationRecord,
): ResumeTokenCounts | undefined {
for (let i = conversation.messages.length - 1; i >= 0; i--) {
const record = conversation.messages[i];
if (record.type === 'assistant') {
const usage = record.usageMetadata;
const candidate = usage?.promptTokenCount ?? usage?.totalTokenCount;
if (candidate) {
return {
promptTokenCount: candidate,
outputTokenCount: getUsageOutputTokenCountForPromptEstimate(usage),
isEstimated: false,
};
}
}
if (record.type === 'system' && record.subtype === 'chat_compression') {
const payload = record.systemPayload as
| ChatCompressionRecordPayload
| undefined;
if (payload?.info) {
return {
promptTokenCount: payload.info.newTokenCount,
outputTokenCount: 0,
// Checkpoints created before provenance was persisted are safest to
// treat as estimates: this keeps the output clamp's overhead pad on
// and prevents an optimistic resume from overflowing the window.
isEstimated: payload.info.newTokenCountIsEstimated ?? true,
};
}
}
}
return undefined;
}
const MAX_BRANCH_COLLISION_SCAN = 99;
export async function computeUniqueBranchTitle(

View file

@ -486,3 +486,22 @@ export function readLastJsonStringFieldsSync(
}
}
}
export function readSessionTitleInfoFromFileSync(
filePath: string,
scratchBuffer?: Buffer,
): { title?: string; source?: 'auto' | 'manual' } {
const hit = readLastJsonStringFieldsSync(
filePath,
'customTitle',
['titleSource'],
'"subtype":"custom_title"',
scratchBuffer,
);
const title = hit['customTitle'];
if (!title) return {};
const rawSource = hit['titleSource'];
const source =
rawSource === 'auto' || rawSource === 'manual' ? rawSource : undefined;
return { title, source };
}