mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-09 17:19:02 +00:00
feat(core): persist file history snapshots for cross-session /rewind (T2.1) (#4897)
* feat(core): persist file history snapshots to JSONL for cross-session /rewind (T2.1) File history snapshots were purely in-memory — lost on process exit, making /rewind unusable after session resume. This adds JSONL persistence so restored sessions can rewind to any pre-resume turn. Key changes: - Serialize/deserialize FileHistorySnapshot to/from JSONL system records - Record each snapshot after makeSnapshot succeeds (incremental per turn) - Re-record surviving snapshots after rewind (full batch on active branch) - Parse file_history_snapshot records in sessionService.loadSession() - Restore snapshot chain in config.getFileHistoryService() on resume - Copy backup files on session fork (hard link with copy fallback) - Add session_resume capability tag (stable alias for unstable_session_resume) - validateRestoredSnapshots with dedup + batched parallel stat 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix: address Copilot review — last-wins dedup + isEnabled guard - Change snapshot dedup from first-wins to last-wins so rewind batch records (which contain the most up-to-date snapshot state) override earlier incremental records for the same promptId. - Guard validateRestoredSnapshots behind isEnabled() to skip I/O when file checkpointing is disabled. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix: address wenshao review — ghost snapshots + test assertion - Session.ts: slice snapshots to targetTurnIndex+1 to exclude turns being discarded (fixes ghost-snapshot persistence) - AppContainer.tsx: only pass snapshots when file restore succeeded (avoids writing un-truncated snapshots for conversation-only rewind) - Session.test.ts: update rewindRecording assertion to expect 3 args 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix: address wenshao R2 — slice snapshots in AppContainer + add deserialize warning - AppContainer.tsx: use .slice(0, targetTurnIndex + 1) to match Session.ts behavior (prevents ghost snapshots for conversation-only rewind) - sessionService.ts: log warning instead of silent continue on malformed file_history_snapshot deserialization 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix: enable daemon file checkpointing + seed TUI promptCount on resume Two fixes from wenshao's local verification: 1. ACP daemon sessions had fileCheckpointingEnabled=false because stdin is a pipe (non-TTY). Add enableFileCheckpointing() to Config and call it in acpAgent.newSessionConfig so daemon /rewind works. 2. TUI prompt counter restarted at 0 on --resume, colliding with restored snapshot promptIds and corrupting the chain via last-wins dedup. Seed promptCount from the resumed conversation's user turn count so new promptIds don't overlap. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix: add makeSnapshot + recordFileHistorySnapshot to ACP prompt path ACP sessions drive the chat through Session.prompt → GeminiChat, bypassing GeminiClient.sendMessageStream where makeSnapshot lives. This meant daemon-created sessions never produced file history snapshots, leaving /rewind non-functional. Add makeSnapshot + recordFileHistorySnapshot at the start of each ACP prompt turn (mirroring client.ts:1488), using the existing sessionId########turn promptId format. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix: add session_resume to integration test capability assertion 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix: add debug logging to ACP makeSnapshot catch blocks 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix(acp): move makeSnapshot after slash-command and hook checks Locally handled slash commands (/help, /memory, etc.) previously created file history snapshots even though no model turn was added. This caused the snapshot index to drift from the real user turn count, breaking rewind in web-shell sessions that use slash commands frequently. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: address qwen-code-ci-bot review round 5 - Remove stale line number from comment (Session.ts) - Single cast instead of double cast for systemPayload (sessionService.ts) - Guard mkdir in copyFileHistoryBackups to prevent fork failure (sessionService.ts) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: address wenshao review — mock, ordering guard, JSDoc placement - Add makeSnapshot/rewind to FileHistoryService mock in Session.test.ts - Invalidate cached FileHistoryService on enableFileCheckpointing() to prevent stale enabled=false if service was lazily created first - Move copyFileHistoryBackups above class JSDoc to fix association Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: truncate in-memory snapshots on ACP rewind + deduplicate recordFileHistorySnapshot - Call restoreFromSnapshots(survivingSnapshots) in ACP rewindToTurn to prevent phantom snapshots from accumulating in the in-memory array after conversation-only rewind - Simplify recordFileHistorySnapshot to delegate to the batch variant Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: add restoreFromSnapshots to FileHistoryService mock Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jinye
GitHub
parent
963fc543d1
commit
246a0a1fc5
13 changed files with 375 additions and 19 deletions
|
|
@ -193,6 +193,7 @@ describe('qwen serve — capabilities envelope', () => {
|
|||
'session_create',
|
||||
'session_scope_override',
|
||||
'session_load',
|
||||
'session_resume',
|
||||
'unstable_session_resume',
|
||||
'session_list',
|
||||
'session_prompt',
|
||||
|
|
|
|||
|
|
@ -6768,6 +6768,12 @@ class QwenAgent implements Agent {
|
|||
// <available_skills> at cold start.
|
||||
buildDisabledSkillNamesProvider(this.settings),
|
||||
);
|
||||
// ACP sessions run with piped stdio (non-TTY), so the default
|
||||
// interactive-based gating disables file checkpointing. Enable it
|
||||
// explicitly so /rewind works across daemon session resume.
|
||||
if (typeof config.enableFileCheckpointing === 'function') {
|
||||
config.enableFileCheckpointing();
|
||||
}
|
||||
// Inject the workspace-shared MCP transport pool BEFORE
|
||||
// `config.initialize()` so the ToolRegistry picks it up.
|
||||
if (
|
||||
|
|
|
|||
|
|
@ -307,6 +307,12 @@ describe('Session', () => {
|
|||
.fn()
|
||||
.mockReturnValue(mockBackgroundShellRegistry),
|
||||
getMonitorRegistry: vi.fn().mockReturnValue(mockMonitorRegistry),
|
||||
getFileHistoryService: vi.fn().mockReturnValue({
|
||||
makeSnapshot: vi.fn().mockResolvedValue(undefined),
|
||||
getSnapshots: vi.fn().mockReturnValue([]),
|
||||
restoreFromSnapshots: vi.fn(),
|
||||
rewind: vi.fn(),
|
||||
}),
|
||||
} as unknown as Config;
|
||||
|
||||
mockClient = {
|
||||
|
|
@ -446,9 +452,11 @@ describe('Session', () => {
|
|||
expect(result).toEqual({ targetTurnIndex: 1, apiTruncateIndex: 2 });
|
||||
expect(mockChat.truncateHistory).toHaveBeenCalledWith(2);
|
||||
expect(mockChat.stripThoughtsFromHistory).toHaveBeenCalled();
|
||||
expect(mockChatRecordingService.rewindRecording).toHaveBeenCalledWith(1, {
|
||||
truncatedCount: 2,
|
||||
});
|
||||
expect(mockChatRecordingService.rewindRecording).toHaveBeenCalledWith(
|
||||
1,
|
||||
{ truncatedCount: 2 },
|
||||
[],
|
||||
);
|
||||
});
|
||||
|
||||
it('preserves startup context when rewinding to the first user turn', () => {
|
||||
|
|
|
|||
|
|
@ -614,9 +614,20 @@ export class Session implements SessionContext {
|
|||
chat.truncateHistory(apiTruncateIndex);
|
||||
chat.stripThoughtsFromHistory();
|
||||
|
||||
this.config.getChatRecordingService()?.rewindRecording(targetTurnIndex, {
|
||||
truncatedCount: Math.max(0, apiHistory.length - apiTruncateIndex),
|
||||
});
|
||||
const fileHistoryService = this.config.getFileHistoryService();
|
||||
const survivingSnapshots = fileHistoryService
|
||||
.getSnapshots()
|
||||
.slice(0, targetTurnIndex + 1);
|
||||
|
||||
fileHistoryService.restoreFromSnapshots(survivingSnapshots);
|
||||
|
||||
this.config
|
||||
.getChatRecordingService()
|
||||
?.rewindRecording(
|
||||
targetTurnIndex,
|
||||
{ truncatedCount: Math.max(0, apiHistory.length - apiTruncateIndex) },
|
||||
survivingSnapshots,
|
||||
);
|
||||
|
||||
return { targetTurnIndex, apiTruncateIndex };
|
||||
}
|
||||
|
|
@ -1054,6 +1065,27 @@ export class Session implements SessionContext {
|
|||
}
|
||||
}
|
||||
|
||||
// Snapshot file state before this turn (mirrors the makeSnapshot
|
||||
// block in GeminiClient.sendMessageStream). Placed after
|
||||
// slash-command and hook early-returns so locally handled commands
|
||||
// don't create phantom snapshots that desync the snapshot index.
|
||||
try {
|
||||
const fileHistoryService = this.config.getFileHistoryService();
|
||||
await fileHistoryService.makeSnapshot(promptId);
|
||||
try {
|
||||
const latestSnapshot = fileHistoryService.getSnapshots().at(-1);
|
||||
if (latestSnapshot) {
|
||||
this.config
|
||||
.getChatRecordingService()
|
||||
?.recordFileHistorySnapshot(latestSnapshot);
|
||||
}
|
||||
} catch (e) {
|
||||
debugLogger.error(`FileHistory: recordSnapshot failed: ${e}`);
|
||||
}
|
||||
} catch (e) {
|
||||
debugLogger.error(`FileHistory: makeSnapshot failed: ${e}`);
|
||||
}
|
||||
|
||||
// Prepend session-level system reminders (plan mode / subagent /
|
||||
// arena) so the model sees them, matching the behaviour of
|
||||
// `GeminiClient.sendMessageStream` in the CLI/TUI path. Without this,
|
||||
|
|
|
|||
|
|
@ -35,9 +35,9 @@ export const SERVE_CAPABILITY_REGISTRY = {
|
|||
session_create: { since: 'v1' },
|
||||
session_scope_override: { since: 'v1' },
|
||||
session_load: { since: 'v1' },
|
||||
// ACP backs this with `connection.unstable_resumeSession`. Surface
|
||||
// the unstable prefix so clients don't pin against a `v1` shape that
|
||||
// the underlying ACP method may still change.
|
||||
session_resume: { since: 'v1' },
|
||||
// Deprecated alias — kept until @agentclientprotocol/sdk graduates
|
||||
// the underlying ACP method from unstable_resumeSession to resumeSession.
|
||||
unstable_session_resume: { since: 'v1' },
|
||||
session_list: { since: 'v1' },
|
||||
session_prompt: { since: 'v1' },
|
||||
|
|
|
|||
|
|
@ -107,6 +107,7 @@ const EXPECTED_STAGE1_FEATURES = [
|
|||
'session_create',
|
||||
'session_scope_override',
|
||||
'session_load',
|
||||
'session_resume',
|
||||
'unstable_session_resume',
|
||||
'session_list',
|
||||
'session_prompt',
|
||||
|
|
|
|||
|
|
@ -441,7 +441,11 @@ export const AppContainer = (props: AppContainerProps) => {
|
|||
);
|
||||
|
||||
// Additional hooks moved from App.tsx
|
||||
const { stats: sessionStats, startNewSession } = useSessionStats();
|
||||
const {
|
||||
stats: sessionStats,
|
||||
startNewSession,
|
||||
seedPromptCount,
|
||||
} = useSessionStats();
|
||||
const logger = useLogger(config.storage, sessionStats.sessionId);
|
||||
const branchName = useGitBranchName(config.getTargetDir());
|
||||
const worktreeSession = useWorktreeSession(config);
|
||||
|
|
@ -549,6 +553,15 @@ export const AppContainer = (props: AppContainerProps) => {
|
|||
);
|
||||
historyManager.loadHistory(historyItems);
|
||||
|
||||
// Seed the prompt counter from the resumed conversation so new
|
||||
// promptIds don't collide with restored file history snapshots.
|
||||
const userTurnCount = resumedSessionData.conversation.messages.filter(
|
||||
(m) => m.type === 'user' && m.subtype !== 'mid_turn_user_message',
|
||||
).length;
|
||||
if (userTurnCount > 0) {
|
||||
seedPromptCount(userTurnCount);
|
||||
}
|
||||
|
||||
// Re-arm any `/goal` that was active when the prior session ended.
|
||||
try {
|
||||
restoreGoalFromHistory(historyItems, config, historyManager.addItem);
|
||||
|
|
@ -2622,9 +2635,16 @@ export const AppContainer = (props: AppContainerProps) => {
|
|||
Date.now(),
|
||||
);
|
||||
|
||||
config.getChatRecordingService()?.rewindRecording(targetTurnIndex, {
|
||||
truncatedCount: originalLength - truncatedUi.length,
|
||||
});
|
||||
config.getChatRecordingService()?.rewindRecording(
|
||||
targetTurnIndex,
|
||||
{ truncatedCount: originalLength - truncatedUi.length },
|
||||
!hasRestoreFailure
|
||||
? config
|
||||
.getFileHistoryService()
|
||||
.getSnapshots()
|
||||
.slice(0, targetTurnIndex + 1)
|
||||
: undefined,
|
||||
);
|
||||
}
|
||||
|
||||
// Show file restore result after conversation truncation so the
|
||||
|
|
|
|||
|
|
@ -191,6 +191,7 @@ interface SessionStatsContextValue {
|
|||
startNewSession: (sessionId: string) => void;
|
||||
startNewPrompt: () => void;
|
||||
getPromptCount: () => number;
|
||||
seedPromptCount: (count: number) => void;
|
||||
}
|
||||
|
||||
// --- Context Definition ---
|
||||
|
|
@ -271,14 +272,22 @@ export const SessionStatsProvider: React.FC<{
|
|||
[stats.promptCount],
|
||||
);
|
||||
|
||||
const seedPromptCount = useCallback((count: number) => {
|
||||
setStats((prevState) => ({
|
||||
...prevState,
|
||||
promptCount: Math.max(prevState.promptCount, count),
|
||||
}));
|
||||
}, []);
|
||||
|
||||
const value = useMemo(
|
||||
() => ({
|
||||
stats,
|
||||
startNewSession,
|
||||
startNewPrompt,
|
||||
getPromptCount,
|
||||
seedPromptCount,
|
||||
}),
|
||||
[stats, startNewSession, startNewPrompt, getPromptCount],
|
||||
[stats, startNewSession, startNewPrompt, getPromptCount, seedPromptCount],
|
||||
);
|
||||
|
||||
return (
|
||||
|
|
|
|||
|
|
@ -1158,7 +1158,7 @@ export class Config {
|
|||
private fileDiscoveryService: FileDiscoveryService | null = null;
|
||||
private sessionService: SessionService | undefined = undefined;
|
||||
private chatRecordingService: ChatRecordingService | undefined = undefined;
|
||||
private readonly fileCheckpointingEnabled: boolean;
|
||||
private fileCheckpointingEnabled: boolean;
|
||||
private fileHistoryService: FileHistoryService | undefined;
|
||||
private readonly proxy: string | undefined;
|
||||
private cwd: string;
|
||||
|
|
@ -3791,6 +3791,11 @@ export class Config {
|
|||
return this.fileCheckpointingEnabled;
|
||||
}
|
||||
|
||||
enableFileCheckpointing(): void {
|
||||
this.fileCheckpointingEnabled = true;
|
||||
this.fileHistoryService = undefined;
|
||||
}
|
||||
|
||||
getFileHistoryService(): FileHistoryService {
|
||||
if (!this.fileHistoryService) {
|
||||
this.fileHistoryService = new FileHistoryService(
|
||||
|
|
@ -3798,6 +3803,15 @@ export class Config {
|
|||
this.fileCheckpointingEnabled,
|
||||
this.cwd,
|
||||
);
|
||||
const snapshots = this.sessionData?.fileHistorySnapshots;
|
||||
if (snapshots?.length && this.fileHistoryService.isEnabled()) {
|
||||
this.fileHistoryService.restoreFromSnapshots(snapshots);
|
||||
void this.fileHistoryService.validateRestoredSnapshots().catch((e) => {
|
||||
this.debugLogger.error(
|
||||
`FileHistory: validateRestoredSnapshots failed: ${e}`,
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
return this.fileHistoryService;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1645,6 +1645,19 @@ export class GeminiClient {
|
|||
if (messageType === SendMessageType.UserQuery) {
|
||||
try {
|
||||
await this.config.getFileHistoryService().makeSnapshot(prompt_id);
|
||||
try {
|
||||
const latestSnapshot = this.config
|
||||
.getFileHistoryService()
|
||||
.getSnapshots()
|
||||
.at(-1);
|
||||
if (latestSnapshot) {
|
||||
this.config
|
||||
.getChatRecordingService()
|
||||
?.recordFileHistorySnapshot(latestSnapshot);
|
||||
}
|
||||
} catch (e) {
|
||||
debugLogger.error(`FileHistory: recordSnapshot failed: ${e}`);
|
||||
}
|
||||
} catch (e) {
|
||||
debugLogger.error(`FileHistory: makeSnapshot failed: ${e}`);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,6 +28,11 @@ import type {
|
|||
import type { Status } from '../core/coreToolScheduler.js';
|
||||
import type { AgentResultDisplay, FileDiff } from '../tools/tools.js';
|
||||
import type { UiEvent } from '../telemetry/uiTelemetry.js';
|
||||
import type {
|
||||
FileHistorySnapshot,
|
||||
SerializedFileHistorySnapshot,
|
||||
} from './fileHistoryService.js';
|
||||
import { serializeSnapshot } from './fileHistoryService.js';
|
||||
|
||||
const debugLogger = createDebugLogger('CHAT_RECORDING');
|
||||
|
||||
|
|
@ -235,7 +240,8 @@ export interface ChatRecord {
|
|||
| 'custom_title'
|
||||
| 'rewind'
|
||||
| 'agent_bootstrap'
|
||||
| 'agent_launch_prompt';
|
||||
| 'agent_launch_prompt'
|
||||
| 'file_history_snapshot';
|
||||
/** Working directory at time of message */
|
||||
cwd: string;
|
||||
/** CLI version for compatibility tracking */
|
||||
|
|
@ -281,7 +287,8 @@ export interface ChatRecord {
|
|||
| CustomTitleRecordPayload
|
||||
| NotificationRecordPayload
|
||||
| RewindRecordPayload
|
||||
| AgentBootstrapRecordPayload;
|
||||
| AgentBootstrapRecordPayload
|
||||
| FileHistorySnapshotRecordPayload;
|
||||
|
||||
/** Background subagent that produced this record (e.g. "explore-7f3c"). */
|
||||
agentId?: string;
|
||||
|
|
@ -428,6 +435,14 @@ export interface RewindRecordPayload {
|
|||
truncatedCount: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stored payload for file history snapshot persistence.
|
||||
* Each entry records one or more snapshots for session resume.
|
||||
*/
|
||||
export interface FileHistorySnapshotRecordPayload {
|
||||
snapshots: SerializedFileHistorySnapshot[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Service for recording the current chat session to disk.
|
||||
*
|
||||
|
|
@ -1150,7 +1165,11 @@ export class ChatRecordingService {
|
|||
* nothing before it), 1 means keep the first user turn, etc.
|
||||
* @param payload Additional metadata to persist with the rewind record.
|
||||
*/
|
||||
rewindRecording(targetTurnIndex: number, payload: RewindRecordPayload): void {
|
||||
rewindRecording(
|
||||
targetTurnIndex: number,
|
||||
payload: RewindRecordPayload,
|
||||
survivingFileHistorySnapshots?: FileHistorySnapshot[],
|
||||
): void {
|
||||
try {
|
||||
// Re-root: point back to the record just before the target user turn.
|
||||
this.lastRecordUuid = this.turnParentUuids[targetTurnIndex] ?? null;
|
||||
|
|
@ -1171,6 +1190,12 @@ export class ChatRecordingService {
|
|||
};
|
||||
|
||||
this.appendRecord(record);
|
||||
|
||||
// Re-record surviving file history snapshots on the active branch so
|
||||
// they are visible to reconstructHistory on resume.
|
||||
if (survivingFileHistorySnapshots?.length) {
|
||||
this.recordFileHistorySnapshotBatch(survivingFileHistorySnapshots);
|
||||
}
|
||||
} catch (error) {
|
||||
debugLogger.error('Error saving rewind record:', error);
|
||||
}
|
||||
|
|
@ -1356,4 +1381,23 @@ export class ChatRecordingService {
|
|||
debugLogger.error('Error saving attribution snapshot:', error);
|
||||
}
|
||||
}
|
||||
|
||||
recordFileHistorySnapshot(snapshot: FileHistorySnapshot): void {
|
||||
this.recordFileHistorySnapshotBatch([snapshot]);
|
||||
}
|
||||
|
||||
recordFileHistorySnapshotBatch(snapshots: FileHistorySnapshot[]): void {
|
||||
if (snapshots.length === 0) return;
|
||||
try {
|
||||
const record: ChatRecord = {
|
||||
...this.createBaseRecord('system'),
|
||||
type: 'system',
|
||||
subtype: 'file_history_snapshot',
|
||||
systemPayload: { snapshots: snapshots.map(serializeSnapshot) },
|
||||
};
|
||||
this.appendRecord(record);
|
||||
} catch (error) {
|
||||
debugLogger.error('Error saving file history snapshot batch:', error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -98,8 +98,70 @@ export interface TurnDiff {
|
|||
};
|
||||
}
|
||||
|
||||
const MAX_SNAPSHOTS = 100;
|
||||
export const MAX_SNAPSHOTS = 100;
|
||||
export const FILE_HISTORY_DIR = 'file-history';
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Serialization types for JSONL persistence
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface SerializedFileHistorySnapshot {
|
||||
promptId: string;
|
||||
trackedFileBackups: Record<string, SerializedFileHistoryBackup>;
|
||||
timestamp: string;
|
||||
}
|
||||
|
||||
export interface SerializedFileHistoryBackup {
|
||||
backupFileName: string | null;
|
||||
version: number;
|
||||
backupTime: string;
|
||||
failed?: boolean;
|
||||
}
|
||||
|
||||
export function serializeSnapshot(
|
||||
s: FileHistorySnapshot,
|
||||
): SerializedFileHistorySnapshot {
|
||||
return {
|
||||
promptId: s.promptId,
|
||||
timestamp: s.timestamp.toISOString(),
|
||||
trackedFileBackups: Object.fromEntries(
|
||||
Object.entries(s.trackedFileBackups).map(([path, backup]) => [
|
||||
path,
|
||||
{
|
||||
backupFileName: backup.backupFileName,
|
||||
version: backup.version,
|
||||
backupTime: backup.backupTime.toISOString(),
|
||||
failed: backup.failed || undefined,
|
||||
},
|
||||
]),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
function safeParseDate(iso: string): Date {
|
||||
const d = new Date(iso);
|
||||
return Number.isNaN(d.getTime()) ? new Date(0) : d;
|
||||
}
|
||||
|
||||
export function deserializeSnapshots(
|
||||
arr: SerializedFileHistorySnapshot[],
|
||||
): FileHistorySnapshot[] {
|
||||
return arr.map((s) => ({
|
||||
promptId: s.promptId,
|
||||
timestamp: safeParseDate(s.timestamp),
|
||||
trackedFileBackups: Object.fromEntries(
|
||||
Object.entries(s.trackedFileBackups).map(([path, backup]) => [
|
||||
path,
|
||||
{
|
||||
backupFileName: backup.backupFileName,
|
||||
version: backup.version,
|
||||
backupTime: safeParseDate(backup.backupTime),
|
||||
failed: backup.failed,
|
||||
},
|
||||
]),
|
||||
),
|
||||
}));
|
||||
}
|
||||
/** Per-turn read-fanout cap. Each candidate file may read up to two backups,
|
||||
* so 500 files ≈ 1000 concurrent opens — safely under the typical 4096 fd
|
||||
* ceiling and well below `ulimit -n` defaults on Linux/macOS. */
|
||||
|
|
@ -521,6 +583,52 @@ export class FileHistoryService {
|
|||
};
|
||||
}
|
||||
|
||||
async validateRestoredSnapshots(): Promise<void> {
|
||||
// Collect unique backup file names to stat (dedup: many snapshots share
|
||||
// the same backup file via the inheritance optimization in makeSnapshot).
|
||||
const uniqueNames = new Set<string>();
|
||||
for (const snapshot of this.state.snapshots) {
|
||||
for (const backup of Object.values(snapshot.trackedFileBackups)) {
|
||||
if (backup.backupFileName !== null && !backup.failed) {
|
||||
uniqueNames.add(backup.backupFileName);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (uniqueNames.size === 0) return;
|
||||
|
||||
// Parallel stat with bounded concurrency to avoid fd exhaustion.
|
||||
const BATCH_SIZE = 200;
|
||||
const missing = new Set<string>();
|
||||
const names = [...uniqueNames];
|
||||
for (let i = 0; i < names.length; i += BATCH_SIZE) {
|
||||
const batch = names.slice(i, i + BATCH_SIZE);
|
||||
const results = await Promise.all(
|
||||
batch.map((name) =>
|
||||
pathExists(resolveBackupPath(name, this.sessionId)),
|
||||
),
|
||||
);
|
||||
for (let j = 0; j < batch.length; j++) {
|
||||
if (!results[j]) missing.add(batch[j]);
|
||||
}
|
||||
}
|
||||
|
||||
if (missing.size === 0) return;
|
||||
|
||||
// Single synchronous pass to mark failures — minimizes the mutation
|
||||
// window so concurrent makeSnapshot/trackEdit see a consistent state.
|
||||
for (const snapshot of this.state.snapshots) {
|
||||
for (const backup of Object.values(snapshot.trackedFileBackups)) {
|
||||
if (backup.backupFileName && missing.has(backup.backupFileName)) {
|
||||
backup.failed = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
debugLogger.warn(
|
||||
`FileHistory: ${missing.size} restored backup file(s) missing on disk`,
|
||||
);
|
||||
}
|
||||
|
||||
async trackEdit(filePath: string): Promise<void> {
|
||||
if (!this.enabled) return;
|
||||
|
||||
|
|
|
|||
|
|
@ -15,9 +15,16 @@ import * as jsonl from '../utils/jsonl-utils.js';
|
|||
import type {
|
||||
ChatCompressionRecordPayload,
|
||||
ChatRecord,
|
||||
FileHistorySnapshotRecordPayload,
|
||||
TitleSource,
|
||||
UiTelemetryRecordPayload,
|
||||
} from './chatRecordingService.js';
|
||||
import type { FileHistorySnapshot } from './fileHistoryService.js';
|
||||
import {
|
||||
deserializeSnapshots,
|
||||
FILE_HISTORY_DIR,
|
||||
MAX_SNAPSHOTS,
|
||||
} from './fileHistoryService.js';
|
||||
import { uiTelemetryService } from '../telemetry/uiTelemetry.js';
|
||||
import { createDebugLogger } from '../utils/debugLogger.js';
|
||||
import { readRuntimeStatus } from '../utils/runtimeStatus.js';
|
||||
|
|
@ -130,6 +137,8 @@ export interface ResumedSessionData {
|
|||
filePath: string;
|
||||
/** UUID of the last completed message - new messages should use this as parentUuid */
|
||||
lastCompletedUuid: string | null;
|
||||
/** Deserialized file history snapshots for resume (enables /rewind across sessions) */
|
||||
fileHistorySnapshots?: FileHistorySnapshot[];
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -158,6 +167,57 @@ const MAX_PROMPT_SCAN_LINES = 10;
|
|||
*/
|
||||
const TAIL_READ_SIZE = 64 * 1024;
|
||||
|
||||
async function copyFileHistoryBackups(
|
||||
sourceSessionId: string,
|
||||
targetSessionId: string,
|
||||
): Promise<void> {
|
||||
const fsPromises = await import('node:fs/promises');
|
||||
const sourceDir = path.join(
|
||||
Storage.getGlobalQwenDir(),
|
||||
FILE_HISTORY_DIR,
|
||||
sourceSessionId,
|
||||
);
|
||||
const targetDir = path.join(
|
||||
Storage.getGlobalQwenDir(),
|
||||
FILE_HISTORY_DIR,
|
||||
targetSessionId,
|
||||
);
|
||||
|
||||
let entries: string[];
|
||||
try {
|
||||
entries = await fsPromises.readdir(sourceDir);
|
||||
} catch (e: unknown) {
|
||||
if ((e as NodeJS.ErrnoException).code === 'ENOENT') return;
|
||||
debugLogger.warn(`copyFileHistoryBackups: readdir failed: ${e}`);
|
||||
return;
|
||||
}
|
||||
if (entries.length === 0) return;
|
||||
|
||||
try {
|
||||
await fsPromises.mkdir(targetDir, { recursive: true });
|
||||
} catch (e) {
|
||||
debugLogger.warn(`copyFileHistoryBackups: mkdir failed: ${e}`);
|
||||
return;
|
||||
}
|
||||
await Promise.all(
|
||||
entries.map(async (name) => {
|
||||
const src = path.join(sourceDir, name);
|
||||
const dst = path.join(targetDir, name);
|
||||
try {
|
||||
await fsPromises.link(src, dst);
|
||||
} catch {
|
||||
try {
|
||||
await fsPromises.copyFile(src, dst);
|
||||
} catch (copyErr) {
|
||||
debugLogger.warn(
|
||||
`copyFileHistoryBackups: failed to copy ${name}: ${copyErr}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Service for managing chat sessions.
|
||||
*
|
||||
|
|
@ -750,10 +810,48 @@ export class SessionService {
|
|||
messages,
|
||||
};
|
||||
|
||||
// Extract file history snapshots for /rewind across resume
|
||||
const fileHistorySnapshots: FileHistorySnapshot[] = [];
|
||||
const seenPromptIds = new Map<string, number>();
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
const cappedSnapshots =
|
||||
fileHistorySnapshots.length > MAX_SNAPSHOTS
|
||||
? fileHistorySnapshots.slice(-MAX_SNAPSHOTS)
|
||||
: fileHistorySnapshots;
|
||||
|
||||
return {
|
||||
conversation,
|
||||
filePath,
|
||||
lastCompletedUuid: lastMessage.uuid,
|
||||
fileHistorySnapshots:
|
||||
cappedSnapshots.length > 0 ? cappedSnapshots : undefined,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -990,6 +1088,8 @@ export class SessionService {
|
|||
fs.closeSync(fd);
|
||||
}
|
||||
|
||||
await copyFileHistoryBackups(sourceSessionId, newSessionId);
|
||||
|
||||
return { filePath: targetPath, copiedCount: forked.length };
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue