From 0b15b9b116e36628af8b4701651866010d6a69f3 Mon Sep 17 00:00:00 2001 From: Aarya2004 Date: Mon, 31 Aug 2026 23:47:29 -0700 Subject: [PATCH 1/7] Rename recovery staging helpers to staging helpers The staging lock, staging root, staged-session start, promotion, and discard helpers are generic. Session fork will call them too, so the `recovery` prefix would misdescribe them. The on-disk directory and lock file names are unchanged. Claude-Session: https://claude.ai/code/session_01Hjm7J6N3SL5Y62TJ3bPxwD --- src/core/session/session_store.zig | 74 +++++++++++++++--------------- 1 file changed, 38 insertions(+), 36 deletions(-) diff --git a/src/core/session/session_store.zig b/src/core/session/session_store.zig index 18b667445..a3debbe6f 100644 --- a/src/core/session/session_store.zig +++ b/src/core/session/session_store.zig @@ -69,8 +69,10 @@ const latestCachePublish = latest_pointer.latestCachePublish; const latestCacheWriteDeferred = latest_pointer.latestCacheWriteDeferred; const latest_sessions_lock_file = latest_pointer.latest_sessions_lock_file; const latest_sessions_dir = latest_pointer.latest_sessions_dir; -const recovery_staging_dir = "recovery+staging"; -const recovery_staging_lock_file = "recovery-staging.lock"; +// The on-disk names stay `recovery+staging` so staged directories written by +// earlier versions are still found and cleaned up. +const staging_dir = "recovery+staging"; +const staging_lock_file = "recovery-staging.lock"; const usage_recovery_dir = profile_paths.usage_recovery_dir_name; const usage_recovery_marker_prefix = "v1 "; const max_usage_recovery_marker_bytes = @@ -657,7 +659,7 @@ pub const Store = struct { return loaded; } - fn startRecoveryStagedSession( + fn startStagedSession( self: Store, alloc: Allocator, staging_root: *session_log.Root, @@ -675,7 +677,7 @@ pub const Store = struct { return loaded; } - fn initRecoveryStagingRoot( + fn initStagingRoot( self: Store, alloc: Allocator, ) !session_log.Root { @@ -684,20 +686,20 @@ pub const Store = struct { return error.SessionStoreUnavailable); var staging = try io_mod.openOrCreateVerifiedPrivateDir( sessions, - recovery_staging_dir, + staging_dir, ); errdefer staging.close(); return .{ .sessions = staging, .display_root = try std.fs.path.join( alloc, - &.{ self.sessions_dir, recovery_staging_dir }, + &.{ self.sessions_dir, staging_dir }, ), .mode = .writable, }; } - fn deinitRecoveryStagingRoot( + fn deinitStagingRoot( self: Store, alloc: Allocator, staging_root: *session_log.Root, @@ -706,18 +708,18 @@ pub const Store = struct { const sessions = &(self.canonical_root.sessions orelse return); sessions.dir.deleteDir( io_mod.getIo(), - recovery_staging_dir, + staging_dir, ) catch return; io_mod.syncVerifiedDir(sessions.dir) catch |err| { debug_trace.logf( "session", - "event=recovery_staging_cleanup disposition=indeterminate err={s}", + "event=session_staging_cleanup disposition=indeterminate err={s}", .{@errorName(err)}, ); }; } - fn acquireRecoveryStagingLock( + fn acquireStagingLock( self: Store, deadline_ms: u64, ) !io_mod.TimedAdvisoryLock { @@ -726,7 +728,7 @@ pub const Store = struct { return error.SessionStoreUnavailable); return io_mod.acquireTimedAdvisoryLock( sessions, - recovery_staging_lock_file, + staging_lock_file, deadline_ms, ) catch |err| switch (err) { error.LockBusy => error.SessionBusy, @@ -735,7 +737,7 @@ pub const Store = struct { }; } - fn cleanupAbandonedRecoveryStages( + fn cleanupAbandonedStages( staging_root: *session_log.Root, ) !void { const staging = &(staging_root.sessions orelse @@ -749,18 +751,18 @@ pub const Store = struct { changed = true; debug_trace.logf( "session", - "event=recovery_staging_abandoned_target_removed", + "event=session_staging_abandoned_target_removed", .{}, ); } if (changed) try io_mod.syncVerifiedDir(staging.dir); } - fn promoteRecoveryStagedSession( + fn promoteStagedSession( self: Store, staging_root: *session_log.Root, session_id: []const u8, - ) !RecoveryPromotionStatus { + ) !StagingPromotionStatus { const staging = &(staging_root.sessions orelse return error.SessionStoreUnavailable); const sessions = &(self.canonical_root.sessions orelse @@ -779,7 +781,7 @@ pub const Store = struct { } /// Consumes `loaded` on every return. - fn discardRecoveryStagedSession( + fn discardStagedSession( staging_root: *session_log.Root, alloc: Allocator, loaded: *LoadedWritableSession, @@ -790,7 +792,7 @@ pub const Store = struct { { debug_trace.logf( "session", - "event=recovery_staging_discard disposition=retained reason=guard_failed", + "event=session_staging_discard disposition=retained reason=guard_failed", .{}, ); return .retained; @@ -802,7 +804,7 @@ pub const Store = struct { ) catch |err| { debug_trace.logf( "session", - "event=recovery_staging_discard disposition=retained reason=store_root_unverified err={s}", + "event=session_staging_discard disposition=retained reason=store_root_unverified err={s}", .{@errorName(err)}, ); return .retained; @@ -810,7 +812,7 @@ pub const Store = struct { if (!writer_belongs_to_store) { debug_trace.logf( "session", - "event=recovery_staging_discard disposition=retained reason=store_root_mismatch", + "event=session_staging_discard disposition=retained reason=store_root_mismatch", .{}, ); return .retained; @@ -818,7 +820,7 @@ pub const Store = struct { const sessions = &(staging_root.sessions orelse { debug_trace.logf( "session", - "event=recovery_staging_discard disposition=indeterminate stage=sessions_root", + "event=session_staging_discard disposition=indeterminate stage=sessions_root", .{}, ); return .indeterminate; @@ -826,7 +828,7 @@ pub const Store = struct { sessions.dir.deleteTree(io_mod.getIo(), loaded.active_id) catch |err| { debug_trace.logf( "session", - "event=recovery_staging_discard disposition=indeterminate stage=delete err={s}", + "event=session_staging_discard disposition=indeterminate stage=delete err={s}", .{@errorName(err)}, ); return .indeterminate; @@ -834,14 +836,14 @@ pub const Store = struct { io_mod.syncVerifiedDir(sessions.dir) catch |err| { debug_trace.logf( "session", - "event=recovery_staging_discard disposition=indeterminate stage=sync err={s}", + "event=session_staging_discard disposition=indeterminate stage=sync err={s}", .{@errorName(err)}, ); return .indeterminate; }; debug_trace.logf( "session", - "event=recovery_staging_discard disposition=discarded", + "event=session_staging_discard disposition=discarded", .{}, ); return .discarded; @@ -3945,14 +3947,14 @@ pub const Store = struct { var initial = try recoveryInitialState(alloc, recovered); defer initial.deinit(alloc); - var staging_lock = try self.acquireRecoveryStagingLock( + var staging_lock = try self.acquireStagingLock( options.session_lock_deadline_ms, ); defer staging_lock.release(); - var staging_root = try self.initRecoveryStagingRoot(alloc); - defer self.deinitRecoveryStagingRoot(alloc, &staging_root); - try cleanupAbandonedRecoveryStages(&staging_root); - var target = try self.startRecoveryStagedSession( + var staging_root = try self.initStagingRoot(alloc); + defer self.deinitStagingRoot(alloc, &staging_root); + try cleanupAbandonedStages(&staging_root); + var target = try self.startStagedSession( alloc, &staging_root, initial, @@ -3964,7 +3966,7 @@ pub const Store = struct { if (target_promoted) { target.deinit(alloc); } else { - const disposition = discardRecoveryStagedSession( + const disposition = discardStagedSession( &staging_root, alloc, &target, @@ -4038,7 +4040,7 @@ pub const Store = struct { @errorName(resolve_err), }, ); - const disposition = discardRecoveryStagedSession( + const disposition = discardStagedSession( &staging_root, alloc, &target, @@ -4055,7 +4057,7 @@ pub const Store = struct { }; }; if (!try session_log.durableStatesEqual(target.state, recovered)) { - const disposition = discardRecoveryStagedSession( + const disposition = discardStagedSession( &staging_root, alloc, &target, @@ -4076,7 +4078,7 @@ pub const Store = struct { "event=session_recovery_target_indeterminate target={s} validation_err={s}", .{ recovered_id, @errorName(err) }, ); - const disposition = discardRecoveryStagedSession( + const disposition = discardStagedSession( &staging_root, alloc, &target, @@ -4091,7 +4093,7 @@ pub const Store = struct { } return error.SessionRecoveryIndeterminate; }; - const promotion = self.promoteRecoveryStagedSession( + const promotion = self.promoteStagedSession( &staging_root, recovered_id, ) catch |err| { @@ -4187,7 +4189,7 @@ pub const PristineDiscardDisposition = enum { indeterminate, }; -const RecoveryPromotionStatus = enum { +const StagingPromotionStatus = enum { promoted, indeterminate, }; @@ -9678,14 +9680,14 @@ test "abandoned recovery staging stays invisible and does not replace unrelated ctx.workspace, ); defer staged_state.deinit(alloc); - var staging_root = try ctx.store.initRecoveryStagingRoot(alloc); + var staging_root = try ctx.store.initStagingRoot(alloc); const abandoned_path = try sessionDirPath( alloc, staging_root.display_root, staged_state.id, ); defer alloc.free(abandoned_path); - var abandoned = try ctx.store.startRecoveryStagedSession( + var abandoned = try ctx.store.startStagedSession( alloc, &staging_root, staged_state, From c62953f4441726b31b10d03e579de954e0fe6f4a Mon Sep 17 00:00:00 2001 From: Aarya2004 Date: Mon, 31 Aug 2026 23:59:30 -0700 Subject: [PATCH 2/7] Add fork and rewind state replacement reasons A fork writes its history into a brand-new session, so its cache publication cannot be deferred. A rewind rewrites the same session in place and follows compaction. Claude-Session: https://claude.ai/code/session_01Hjm7J6N3SL5Y62TJ3bPxwD --- src/core/session/session_event.zig | 2 ++ src/core/session/session_log.zig | 4 ++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/core/session/session_event.zig b/src/core/session/session_event.zig index 4a63af688..eb5d849d5 100644 --- a/src/core/session/session_event.zig +++ b/src/core/session/session_event.zig @@ -31,6 +31,8 @@ pub const ReplacementReason = enum { migration, recovery, log_compaction, + fork, + rewind, }; pub const SessionStarted = struct { diff --git a/src/core/session/session_log.zig b/src/core/session/session_log.zig index 04600db38..a65c7afee 100644 --- a/src/core/session/session_log.zig +++ b/src/core/session/session_log.zig @@ -706,8 +706,8 @@ pub const LoadedWritableSession = struct { state.workspace_root, ); const may_defer_cache = same_workspace and switch (reason) { - .compaction, .log_compaction => true, - .migration, .recovery => false, + .compaction, .log_compaction, .rewind => true, + .migration, .recovery, .fork => false, }; const cache_deferred = if (may_defer_cache) try self.prepareCommitLifecycleOpportunistic(alloc, options) From a1352458397e0b1a65c95c308df2304d7cf27864 Mon Sep 17 00:00:00 2001 From: Aarya2004 Date: Tue, 1 Sep 2026 00:56:43 -0700 Subject: [PATCH 3/7] Add durable session fork and rewind primitives `forkSessionCopy` stages a new session holding the first N turns of a healthy source, copies only the artifacts those turns reach, and commits the branch history as a `fork` state replacement. The source is opened under its writer lock and read through the read-only replay, so it is never rewritten. `rewindSession` truncates the same session in place under a `rewind` state replacement. Artifacts for the dropped turns are left on disk. Claude-Session: https://claude.ai/code/session_01Hjm7J6N3SL5Y62TJ3bPxwD --- src/core/session/session_store.zig | 731 ++++++++++++++++++++++- src/core/session/session_store_types.zig | 39 ++ 2 files changed, 768 insertions(+), 2 deletions(-) diff --git a/src/core/session/session_store.zig b/src/core/session/session_store.zig index a3debbe6f..8472001ff 100644 --- a/src/core/session/session_store.zig +++ b/src/core/session/session_store.zig @@ -146,8 +146,12 @@ pub const ResumeTarget = store_types.ResumeTarget; pub const ResumeViewAdmission = session_log.ResumeViewAdmission; pub const SessionMigrationResult = store_types.SessionMigrationResult; pub const SessionMigrationStatus = store_types.SessionMigrationStatus; +pub const SessionForkResult = store_types.SessionForkResult; +pub const SessionForkStatus = store_types.SessionForkStatus; pub const SessionRecoveryResult = store_types.SessionRecoveryResult; pub const SessionRecoveryStatus = store_types.SessionRecoveryStatus; +pub const SessionRewindResult = store_types.SessionRewindResult; +pub const SessionRewindStatus = store_types.SessionRewindStatus; pub const SessionSummary = store_types.SessionSummary; pub const HistoryPage = store_types.HistoryPage; @@ -3945,7 +3949,7 @@ pub const Store = struct { alloc.free(recovered.id); recovered.id = replacement_id; - var initial = try recoveryInitialState(alloc, recovered); + var initial = try initialSeedState(alloc, recovered); defer initial.deinit(alloc); var staging_lock = try self.acquireStagingLock( options.session_lock_deadline_ms, @@ -4181,6 +4185,364 @@ pub const Store = struct { .recovered, }; } + + /// Creates a new session holding the first `retained_turns` turns of a + /// healthy source. The source is locked for the read and never modified. + pub fn forkSessionCopy( + self: Store, + alloc: Allocator, + session_id: []const u8, + retained_turns: usize, + options: session_log.Options, + ) !SessionForkResult { + try validateSessionId(session_id); + var source = try self.openWritableSessionDir( + alloc, + session_id, + options.session_lock_deadline_ms, + ); + defer source.deinit(alloc); + const authority = try classifyAuthority( + alloc, + &source.dir, + session_id, + ); + if (authority != .schema_v3) { + return error.SessionForkRequiresCurrentSchema; + } + + var forked = blk: { + var root = self.canonical_root; + var state = root.loadReadOnly( + alloc, + session_id, + options, + ) catch |err| return mapReplayError(err); + errdefer state.deinit(alloc); + try resolveSessionSnapshotLocators( + alloc, + state.history, + self.sessions_dir, + session_id, + ); + break :blk state; + }; + defer forked.deinit(alloc); + + const source_history_len = forked.history.len; + if (retained_turns == 0 or retained_turns > source_history_len) { + return error.SessionTurnOutOfRange; + } + try truncateSessionHistory(alloc, &forked, retained_turns); + // The branch inherits the source token totals. There is no per-turn + // usage ledger to recompute a truncated total from, and zeroing would + // misreport what producing this history actually cost. + if (forked.recovery_checkpoint) |*checkpoint| { + checkpoint.deinit(alloc); + forked.recovery_checkpoint = null; + } + const forked_at_ms = io_mod.milliTimestamp(); + forked.created_at_ms = forked_at_ms; + forked.updated_at_ms = forked_at_ms; + + const source_dir_path = try sessionDirPath( + alloc, + self.sessions_dir, + session_id, + ); + defer alloc.free(source_dir_path); + var source_children = try session_child_store.SessionChildCapability.init( + alloc, + source.dir.dir, + source_dir_path, + .read_only, + ); + defer source_children.deinit(); + + const source_id = try alloc.dupe(u8, session_id); + errdefer alloc.free(source_id); + const forked_id = try generateSessionId(alloc); + errdefer alloc.free(forked_id); + const replacement_id = try alloc.dupe(u8, forked_id); + alloc.free(forked.id); + forked.id = replacement_id; + + var initial = try initialSeedState(alloc, forked); + defer initial.deinit(alloc); + var staging_lock = try self.acquireStagingLock( + options.session_lock_deadline_ms, + ); + defer staging_lock.release(); + var staging_root = try self.initStagingRoot(alloc); + defer self.deinitStagingRoot(alloc, &staging_root); + try cleanupAbandonedStages(&staging_root); + var target = try self.startStagedSession( + alloc, + &staging_root, + initial, + options, + ); + var target_owned = true; + var target_promoted = false; + errdefer if (target_owned) { + if (target_promoted) { + target.deinit(alloc); + } else { + const disposition = discardStagedSession( + &staging_root, + alloc, + &target, + ); + if (disposition != .discarded) { + debug_trace.logf( + "session", + "event=session_fork_unpublished_target_cleanup disposition={s}", + .{@tagName(disposition)}, + ); + } + } + target_owned = false; + }; + + const staged_target_dir = try sessionDirPath( + alloc, + staging_root.display_root, + forked_id, + ); + defer alloc.free(staged_target_dir); + const staged_target_images = try std.fs.path.join( + alloc, + &.{ staged_target_dir, "images" }, + ); + defer alloc.free(staged_target_images); + const target_dir = try sessionDirPath( + alloc, + self.sessions_dir, + forked_id, + ); + defer alloc.free(target_dir); + const target_images = try std.fs.path.join( + alloc, + &.{ target_dir, "images" }, + ); + defer alloc.free(target_images); + copyRecoveredImageSnapshots( + alloc, + forked.history, + staged_target_images, + ) catch |err| return mapForkArtifactError(err); + rebaseRecoveredImageSnapshots( + alloc, + forked.history, + staged_target_images, + target_images, + ) catch |err| return mapForkArtifactError(err); + const contains_unverified_artifacts = copyRecoveredManagedChildren( + alloc, + forked.history, + &source_children, + target.child_capability orelse + return error.SessionChildStoreFailed, + ) catch |err| return mapForkArtifactError(err); + _ = target.commitStateReplacement( + alloc, + forked, + .fork, + .rollback_before_adapter_continue, + options, + ) catch |commit_err| { + target.namespace_confirmation_required = true; + target.validateResumeBoundary(alloc, options) catch |resolve_err| { + debug_trace.logf( + "session", + "event=session_fork_target_indeterminate target={s} commit_err={s} resolve_err={s}", + .{ + forked_id, + @errorName(commit_err), + @errorName(resolve_err), + }, + ); + discardForkTarget(&staging_root, alloc, &target); + target_owned = false; + return error.SessionForkIndeterminate; + }; + }; + if (!try session_log.durableStatesEqual(target.state, forked)) { + discardForkTarget(&staging_root, alloc, &target); + target_owned = false; + return error.SessionForkIndeterminate; + } + target.validateResumeBoundary(alloc, options) catch |err| { + debug_trace.logf( + "session", + "event=session_fork_target_indeterminate target={s} validation_err={s}", + .{ forked_id, @errorName(err) }, + ); + discardForkTarget(&staging_root, alloc, &target); + target_owned = false; + return error.SessionForkIndeterminate; + }; + const promotion = self.promoteStagedSession( + &staging_root, + forked_id, + ) catch |err| { + debug_trace.logf( + "session", + "event=session_fork_target_promotion_failed target={s} err={s}", + .{ forked_id, @errorName(err) }, + ); + return error.SessionForkIndeterminate; + }; + target_promoted = true; + const indeterminate = SessionForkResult{ + .source_session_id = source_id, + .forked_session_id = forked_id, + .history_len = forked.history.len, + .source_history_len = source_history_len, + .status = .indeterminate, + }; + if (promotion == .indeterminate) { + target.deinit(alloc); + target_owned = false; + return indeterminate; + } + self.publishRecoveredLatestPointer( + alloc, + forked, + target.position, + source_id, + options, + ) catch |err| { + debug_trace.logf( + "session", + "event=session_fork_target_indeterminate target={s} latest_err={s}", + .{ forked_id, @errorName(err) }, + ); + target.deinit(alloc); + target_owned = false; + return indeterminate; + }; + target.deinit(alloc); + target_owned = false; + + var verified = self.resumeExactForWrite( + alloc, + forked_id, + forked.workspace_root, + false, + .{ .log = options }, + ) catch |err| { + debug_trace.logf( + "session", + "event=session_fork_target_indeterminate target={s} verify_err={s}", + .{ forked_id, @errorName(err) }, + ); + return indeterminate; + }; + defer verified.deinit(alloc); + if (!try session_log.durableStatesEqual(verified.state, forked)) { + return indeterminate; + } + + return .{ + .source_session_id = source_id, + .forked_session_id = forked_id, + .history_len = forked.history.len, + .source_history_len = source_history_len, + .status = if (contains_unverified_artifacts) + .forked_with_unverified_artifacts + else + .forked, + }; + } + + /// Drops every turn after `retained_turns` from a session in place, keeping + /// its id. The rewind is recorded as a state replacement, so the dropped + /// turns stay in the event log until it is compacted and their artifacts + /// are left on disk. + pub fn rewindSession( + self: Store, + alloc: Allocator, + session_id: []const u8, + retained_turns: usize, + options: session_log.Options, + ) !SessionRewindResult { + var loaded = try self.openSchemaV3ForWriteInPlace( + alloc, + session_id, + options, + ); + defer loaded.deinit(alloc); + + const history_len = loaded.state.history.len; + if (retained_turns > history_len) return error.SessionTurnOutOfRange; + if (retained_turns == history_len) { + return .{ + .session_id = try alloc.dupe(u8, loaded.active_id), + .history_len = history_len, + .removed_turn_count = 0, + .status = .already_at_target, + }; + } + + var rewound = try loaded.state.dupe(alloc); + defer rewound.deinit(alloc); + try truncateSessionHistory(alloc, &rewound, retained_turns); + rewound.updated_at_ms = io_mod.milliTimestamp(); + _ = try loaded.commitStateReplacement( + alloc, + rewound, + .rewind, + .rollback_before_adapter_continue, + options, + ); + + return .{ + .session_id = try alloc.dupe(u8, loaded.active_id), + .history_len = retained_turns, + .removed_turn_count = history_len - retained_turns, + .status = .rewound, + }; + } + + /// Opens a current-schema session for writing without touching its + /// workspace binding, so a session saved from another workspace can still + /// be rewritten in place. + fn openSchemaV3ForWriteInPlace( + self: Store, + alloc: Allocator, + session_id: []const u8, + options: session_log.Options, + ) !LoadedWritableSession { + try validateSessionId(session_id); + var session_dir = try self.openSessionDir(session_id); + defer session_dir.close(); + const authority = try classifyAuthority(alloc, &session_dir, session_id); + if (authority != .schema_v3) { + return error.SessionRewindRequiresCurrentSchema; + } + var root = self.canonical_root; + var loaded = root.resumeForWrite( + alloc, + session_id, + options, + ) catch |err| return mapReplayError(err); + errdefer loaded.deinit(alloc); + try resolveSessionSnapshotLocators( + alloc, + loaded.state.history, + self.sessions_dir, + loaded.active_id, + ); + if (loaded.commit_lifecycle == null) { + try self.installLatestCacheLifecycle( + alloc, + &loaded, + options.test_controls, + ); + } + return loaded; + } }; pub const PristineDiscardDisposition = enum { @@ -4194,7 +4556,51 @@ const StagingPromotionStatus = enum { indeterminate, }; -fn recoveryInitialState( +/// Frees the turns after `retained_turns` and clamps the model-context cursor +/// so it can never point past the shortened history. +fn truncateSessionHistory( + alloc: Allocator, + state: *session_codec.DurableSessionState, + retained_turns: usize, +) !void { + std.debug.assert(retained_turns <= state.history.len); + state.context_history_start = @min( + state.context_history_start, + retained_turns, + ); + if (retained_turns == state.history.len) return; + const dropped = state.history; + const retained: []session.HistoryTurn = if (retained_turns == 0) + &.{} + else + try alloc.dupe(session.HistoryTurn, dropped[0..retained_turns]); + for (dropped[retained_turns..]) |turn| session.freeHistoryTurn(alloc, turn); + alloc.free(dropped); + state.history = retained; +} + +fn mapForkArtifactError(err: anyerror) anyerror { + return switch (err) { + error.SessionRecoveryBoundaryInvalid => error.SessionForkArtifactsUnavailable, + else => err, + }; +} + +fn discardForkTarget( + staging_root: *session_log.Root, + alloc: Allocator, + target: *LoadedWritableSession, +) void { + const disposition = Store.discardStagedSession(staging_root, alloc, target); + if (disposition == .discarded) return; + debug_trace.logf( + "session", + "event=session_fork_staged_target_cleanup disposition={s}", + .{@tagName(disposition)}, + ); +} + +fn initialSeedState( alloc: Allocator, recovered: session_codec.DurableSessionState, ) !session_codec.DurableSessionState { @@ -13871,3 +14277,324 @@ test "history page allocation failure sweep frees replay and page ownership" { try std.testing.expectEqual(failing.allocated_bytes, failing.freed_bytes); } } + +fn writeForkSourceFixture( + alloc: Allocator, + store: Store, + id: []const u8, + workspace_root: []const u8, + count: usize, + context_history_start: usize, +) !void { + var state = try testDurableState(alloc, id, workspace_root); + defer state.deinit(alloc); + var created = try store.startWritableSession(alloc, state); + created.deinit(alloc); + if (count == 0 and context_history_start == 0) return; + + var writable = try store.resumeForWrite(alloc, id); + defer writable.deinit(alloc); + const history = try makeTaggedHistoryPageTurns(alloc, count, "turn"); + defer session.freeHistoryTurnSlice(alloc, history); + const desired = session_codec.DurableSessionState{ + .id = writable.state.id, + .origin_workspace_root = writable.state.origin_workspace_root, + .workspace_root = writable.state.workspace_root, + .created_at_ms = writable.state.created_at_ms, + .updated_at_ms = 20, + .conversation_language = writable.state.conversation_language, + .preferences = writable.state.preferences, + .history = history, + .context_history_start = context_history_start, + .total_input_tokens = 4321, + .total_output_tokens = 8765, + }; + _ = try writable.commitStateReplacement( + alloc, + desired, + .recovery, + .retry_expected_tail, + .{}, + ); +} + +fn expectHistoryPrompts( + state: session_codec.DurableSessionState, + expected: []const []const u8, +) !void { + try std.testing.expectEqual(expected.len, state.history.len); + for (state.history, expected) |turn, prompt| { + try std.testing.expectEqualStrings(prompt, turn.assistant.user.text); + } +} + +fn readSessionDurableBytes( + alloc: Allocator, + store: Store, + id: []const u8, +) ![2][]u8 { + const events = try readFixtureFile( + alloc, + store, + id, + "events.jsonl", + 1024 * 1024, + ); + errdefer alloc.free(events); + const manifest = try readFixtureFile( + alloc, + store, + id, + "session.json", + session_projection.manifest_max_bytes, + ); + return .{ events, manifest }; +} + +test "fork copies a turn prefix into a new session and leaves the source unchanged" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const source_id = "fork-prefix-source"; + try writeForkSourceFixture(alloc, ctx.store, source_id, ctx.workspace, 5, 1); + + const before = try readSessionDurableBytes(alloc, ctx.store, source_id); + defer alloc.free(before[0]); + defer alloc.free(before[1]); + + var result = try ctx.store.forkSessionCopy(alloc, source_id, 3, .{}); + defer result.deinit(alloc); + try std.testing.expectEqualStrings(source_id, result.source_session_id); + try std.testing.expect(!std.mem.eql( + u8, + source_id, + result.forked_session_id, + )); + try std.testing.expectEqual(@as(usize, 3), result.history_len); + try std.testing.expectEqual(@as(usize, 5), result.source_history_len); + try std.testing.expectEqual( + store_types.SessionForkStatus.forked, + result.status, + ); + + const after = try readSessionDurableBytes(alloc, ctx.store, source_id); + defer alloc.free(after[0]); + defer alloc.free(after[1]); + try std.testing.expectEqualSlices(u8, before[0], after[0]); + try std.testing.expectEqualSlices(u8, before[1], after[1]); + + var forked = try ctx.store.loadReadOnly(alloc, result.forked_session_id); + defer forked.deinit(alloc); + try expectHistoryPrompts(forked, &.{ "turn-0", "turn-1", "turn-2" }); + try std.testing.expectEqual(@as(usize, 1), forked.context_history_start); + try std.testing.expectEqual(@as(u64, 4321), forked.total_input_tokens); + try std.testing.expectEqual(@as(u64, 8765), forked.total_output_tokens); + try std.testing.expect(forked.recovery_checkpoint == null); + try std.testing.expectEqualStrings(ctx.workspace, forked.workspace_root); +} + +test "fork clamps a model context cursor that outruns the retained turns" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const source_id = "fork-cursor-source"; + try writeForkSourceFixture(alloc, ctx.store, source_id, ctx.workspace, 5, 4); + + var result = try ctx.store.forkSessionCopy(alloc, source_id, 2, .{}); + defer result.deinit(alloc); + + var forked = try ctx.store.loadReadOnly(alloc, result.forked_session_id); + defer forked.deinit(alloc); + try std.testing.expectEqual(@as(usize, 2), forked.history.len); + try std.testing.expectEqual(@as(usize, 2), forked.context_history_start); +} + +test "fork rejects turn counts outside the source history" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const source_id = "fork-range-source"; + try writeForkSourceFixture(alloc, ctx.store, source_id, ctx.workspace, 3, 0); + try std.testing.expectError( + error.SessionTurnOutOfRange, + ctx.store.forkSessionCopy(alloc, source_id, 0, .{}), + ); + try std.testing.expectError( + error.SessionTurnOutOfRange, + ctx.store.forkSessionCopy(alloc, source_id, 4, .{}), + ); + + const empty_id = "fork-empty-source"; + try writeForkSourceFixture(alloc, ctx.store, empty_id, ctx.workspace, 0, 0); + try std.testing.expectError( + error.SessionTurnOutOfRange, + ctx.store.forkSessionCopy(alloc, empty_id, 1, .{}), + ); +} + +test "rewind drops trailing turns in place and keeps the session id" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const session_id = "rewind-in-place"; + try writeForkSourceFixture(alloc, ctx.store, session_id, ctx.workspace, 5, 4); + + var result = try ctx.store.rewindSession(alloc, session_id, 2, .{}); + defer result.deinit(alloc); + try std.testing.expectEqualStrings(session_id, result.session_id); + try std.testing.expectEqual(@as(usize, 2), result.history_len); + try std.testing.expectEqual(@as(usize, 3), result.removed_turn_count); + try std.testing.expectEqual( + store_types.SessionRewindStatus.rewound, + result.status, + ); + + var state = try ctx.store.loadReadOnly(alloc, session_id); + defer state.deinit(alloc); + try expectHistoryPrompts(state, &.{ "turn-0", "turn-1" }); + try std.testing.expectEqual(@as(usize, 2), state.context_history_start); +} + +test "rewind to an empty conversation is allowed" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const session_id = "rewind-to-empty"; + try writeForkSourceFixture(alloc, ctx.store, session_id, ctx.workspace, 3, 3); + + var result = try ctx.store.rewindSession(alloc, session_id, 0, .{}); + defer result.deinit(alloc); + try std.testing.expectEqual(@as(usize, 0), result.history_len); + try std.testing.expectEqual(@as(usize, 3), result.removed_turn_count); + + var state = try ctx.store.loadReadOnly(alloc, session_id); + defer state.deinit(alloc); + try std.testing.expectEqual(@as(usize, 0), state.history.len); + try std.testing.expectEqual(@as(usize, 0), state.context_history_start); +} + +test "rewind reports already_at_target without writing the log" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const session_id = "rewind-noop"; + try writeForkSourceFixture(alloc, ctx.store, session_id, ctx.workspace, 3, 0); + + const before = try readFixtureFile( + alloc, + ctx.store, + session_id, + "events.jsonl", + 1024 * 1024, + ); + defer alloc.free(before); + + var result = try ctx.store.rewindSession(alloc, session_id, 3, .{}); + defer result.deinit(alloc); + try std.testing.expectEqual( + store_types.SessionRewindStatus.already_at_target, + result.status, + ); + try std.testing.expectEqual(@as(usize, 3), result.history_len); + try std.testing.expectEqual(@as(usize, 0), result.removed_turn_count); + + const after = try readFixtureFile( + alloc, + ctx.store, + session_id, + "events.jsonl", + 1024 * 1024, + ); + defer alloc.free(after); + try std.testing.expectEqualSlices(u8, before, after); +} + +test "rewind rejects retaining more turns than the session holds" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const session_id = "rewind-range"; + try writeForkSourceFixture(alloc, ctx.store, session_id, ctx.workspace, 2, 0); + try std.testing.expectError( + error.SessionTurnOutOfRange, + ctx.store.rewindSession(alloc, session_id, 3, .{}), + ); +} + +test "rewind appends to the log and keeps the frames that hold the dropped turns" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const session_id = "rewind-log-retention"; + try writeForkSourceFixture(alloc, ctx.store, session_id, ctx.workspace, 3, 0); + + const before = try readFixtureFile( + alloc, + ctx.store, + session_id, + "events.jsonl", + 1024 * 1024, + ); + defer alloc.free(before); + + var result = try ctx.store.rewindSession(alloc, session_id, 1, .{}); + defer result.deinit(alloc); + + const after = try readFixtureFile( + alloc, + ctx.store, + session_id, + "events.jsonl", + 1024 * 1024, + ); + defer alloc.free(after); + try std.testing.expect(after.len > before.len); + try std.testing.expectEqualSlices(u8, before, after[0..before.len]); +} + +test "rewind refuses a session that is already open for writing" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const session_id = "rewind-busy"; + try writeForkSourceFixture(alloc, ctx.store, session_id, ctx.workspace, 3, 0); + + var holder = try ctx.store.resumeForWrite(alloc, session_id); + defer holder.deinit(alloc); + try std.testing.expectError( + error.SessionBusy, + ctx.store.rewindSession( + alloc, + session_id, + 1, + .{ .session_lock_deadline_ms = 0 }, + ), + ); +} diff --git a/src/core/session/session_store_types.zig b/src/core/session/session_store_types.zig index 0fdb3fe22..d49dd0c11 100644 --- a/src/core/session/session_store_types.zig +++ b/src/core/session/session_store_types.zig @@ -218,6 +218,45 @@ pub const SessionRecoveryResult = struct { } }; +pub const SessionForkStatus = enum { + forked, + forked_with_unverified_artifacts, + indeterminate, +}; + +/// Result of branching a session at an absolute turn boundary. Owns both ids. +pub const SessionForkResult = struct { + source_session_id: []u8, + forked_session_id: []u8, + history_len: usize, + source_history_len: usize, + status: SessionForkStatus = .forked, + + pub fn deinit(self: *SessionForkResult, alloc: Allocator) void { + alloc.free(self.source_session_id); + alloc.free(self.forked_session_id); + self.* = undefined; + } +}; + +pub const SessionRewindStatus = enum { + rewound, + already_at_target, +}; + +/// Result of dropping trailing turns from a session in place. Owns its id. +pub const SessionRewindResult = struct { + session_id: []u8, + history_len: usize, + removed_turn_count: usize, + status: SessionRewindStatus = .rewound, + + pub fn deinit(self: *SessionRewindResult, alloc: Allocator) void { + alloc.free(self.session_id); + self.* = undefined; + } +}; + /// One class of integrity problem `doctor` can report for a session. pub const DoctorIssueKind = enum { authority_less_creation_orphan, From 330e492c62acafb6a0f58dd31b824a1d6219cb5c Mon Sep 17 00:00:00 2001 From: Aarya2004 Date: Tue, 1 Sep 2026 01:03:28 -0700 Subject: [PATCH 4/7] Render session fork and rewind snapshots Both verbs render text and JSON from one snapshot, following the session recovery contract. Claude-Session: https://claude.ai/code/session_01Hjm7J6N3SL5Y62TJ3bPxwD --- src/core/output/output_contracts.zig | 238 +++++++++++++++++++++++++++ 1 file changed, 238 insertions(+) diff --git a/src/core/output/output_contracts.zig b/src/core/output/output_contracts.zig index 271346b91..a4be70946 100644 --- a/src/core/output/output_contracts.zig +++ b/src/core/output/output_contracts.zig @@ -1302,6 +1302,136 @@ pub const SessionRecoverySnapshot = struct { } }; +pub const SessionForkSnapshot = struct { + result: session_store.SessionForkResult, + + pub fn render( + self: SessionForkSnapshot, + alloc: Allocator, + format: OutputFormat, + ) ![]u8 { + return switch (format) { + .text => self.renderText(alloc), + .json => self.renderJson(alloc), + }; + } + + pub fn renderText(self: SessionForkSnapshot, alloc: Allocator) ![]u8 { + if (self.result.status == .indeterminate) { + return std.fmt.allocPrint( + alloc, + "[session fork] could not confirm branch {s}\nsource: {s} (unchanged)\nresolve: fx --resume {s}\ninspect: fx doctor\n", + .{ + self.result.forked_session_id, + self.result.source_session_id, + self.result.forked_session_id, + }, + ); + } + if (self.result.status == .forked_with_unverified_artifacts) { + return std.fmt.allocPrint( + alloc, + "[session fork] forked {s} to {s}\nhistory_turns: {d} of {d}\nwarning: legacy command artifacts could not be authenticated\nresume: fx --resume {s}\n", + .{ + self.result.source_session_id, + self.result.forked_session_id, + self.result.history_len, + self.result.source_history_len, + self.result.forked_session_id, + }, + ); + } + return std.fmt.allocPrint( + alloc, + "[session fork] forked {s} to {s}\nhistory_turns: {d} of {d}\nresume: fx --resume {s}\n", + .{ + self.result.source_session_id, + self.result.forked_session_id, + self.result.history_len, + self.result.source_history_len, + self.result.forked_session_id, + }, + ); + } + + pub fn renderJson(self: SessionForkSnapshot, alloc: Allocator) ![]u8 { + var out: std.Io.Writer.Allocating = .init(alloc); + defer out.deinit(); + try out.writer.writeAll("{\"kind\":\"session_fork\",\"source_id\":"); + try std.json.Stringify.value( + self.result.source_session_id, + .{}, + &out.writer, + ); + try out.writer.writeAll(",\"forked_id\":"); + try std.json.Stringify.value( + self.result.forked_session_id, + .{}, + &out.writer, + ); + try out.writer.writeAll(",\"status\":"); + try std.json.Stringify.value( + @tagName(self.result.status), + .{}, + &out.writer, + ); + try out.writer.print( + ",\"history_turns\":{d},\"source_history_turns\":{d}}}", + .{ self.result.history_len, self.result.source_history_len }, + ); + return try out.toOwnedSlice(); + } +}; + +pub const SessionRewindSnapshot = struct { + result: session_store.SessionRewindResult, + + pub fn render( + self: SessionRewindSnapshot, + alloc: Allocator, + format: OutputFormat, + ) ![]u8 { + return switch (format) { + .text => self.renderText(alloc), + .json => self.renderJson(alloc), + }; + } + + pub fn renderText(self: SessionRewindSnapshot, alloc: Allocator) ![]u8 { + return std.fmt.allocPrint( + alloc, + "[session rewind] {s}\nhistory_turns: {d}\nremoved_turns: {d}\n", + .{ + self.result.session_id, + self.result.history_len, + self.result.removed_turn_count, + }, + ); + } + + pub fn renderJson(self: SessionRewindSnapshot, alloc: Allocator) ![]u8 { + var out: std.Io.Writer.Allocating = .init(alloc); + defer out.deinit(); + try out.writer.writeAll("{\"kind\":\"session_rewind\",\"id\":"); + try std.json.Stringify.value( + self.result.session_id, + .{}, + &out.writer, + ); + try out.writer.writeAll(",\"status\":"); + try std.json.Stringify.value( + @tagName(self.result.status), + .{}, + &out.writer, + ); + try out.writer.print( + ",\"history_turns\":{d},\"removed_turns\":{d}}}", + .{ self.result.history_len, self.result.removed_turn_count }, + ); + return try out.toOwnedSlice(); + } +}; + pub const DoctorSnapshot = struct { workspace_root: []const u8, model: []const u8, @@ -2896,6 +3026,114 @@ test "core session recovery snapshot text and json stay stable" { ); } +test "core session fork snapshot text and json stay stable" { + const result = session_store.SessionForkResult{ + .source_session_id = @constCast("source-session"), + .forked_session_id = @constCast("forked-session"), + .history_len = 3, + .source_history_len = 7, + }; + + const text = try (SessionForkSnapshot{ .result = result }).renderText( + std.testing.allocator, + ); + defer std.testing.allocator.free(text); + try std.testing.expectEqualStrings( + "[session fork] forked source-session to forked-session\nhistory_turns: 3 of 7\nresume: fx --resume forked-session\n", + text, + ); + + const json = try (SessionForkSnapshot{ .result = result }).renderJson( + std.testing.allocator, + ); + defer std.testing.allocator.free(json); + try std.testing.expectEqualStrings( + "{\"kind\":\"session_fork\",\"source_id\":\"source-session\",\"forked_id\":\"forked-session\",\"status\":\"forked\",\"history_turns\":3,\"source_history_turns\":7}", + json, + ); + + const partial = session_store.SessionForkResult{ + .source_session_id = @constCast("source-session"), + .forked_session_id = @constCast("partial-session"), + .history_len = 3, + .source_history_len = 7, + .status = .forked_with_unverified_artifacts, + }; + const partial_text = try (SessionForkSnapshot{ + .result = partial, + }).renderText(std.testing.allocator); + defer std.testing.allocator.free(partial_text); + try std.testing.expectEqualStrings( + "[session fork] forked source-session to partial-session\nhistory_turns: 3 of 7\nwarning: legacy command artifacts could not be authenticated\nresume: fx --resume partial-session\n", + partial_text, + ); + + const indeterminate = session_store.SessionForkResult{ + .source_session_id = @constCast("source-session"), + .forked_session_id = @constCast("target-session"), + .history_len = 3, + .source_history_len = 7, + .status = .indeterminate, + }; + const warning = try (SessionForkSnapshot{ + .result = indeterminate, + }).renderText(std.testing.allocator); + defer std.testing.allocator.free(warning); + try std.testing.expectEqualStrings( + "[session fork] could not confirm branch target-session\nsource: source-session (unchanged)\nresolve: fx --resume target-session\ninspect: fx doctor\n", + warning, + ); + const indeterminate_json = try (SessionForkSnapshot{ + .result = indeterminate, + }).renderJson(std.testing.allocator); + defer std.testing.allocator.free(indeterminate_json); + try std.testing.expectEqualStrings( + "{\"kind\":\"session_fork\",\"source_id\":\"source-session\",\"forked_id\":\"target-session\",\"status\":\"indeterminate\",\"history_turns\":3,\"source_history_turns\":7}", + indeterminate_json, + ); +} + +test "core session rewind snapshot text and json stay stable" { + const result = session_store.SessionRewindResult{ + .session_id = @constCast("rewound-session"), + .history_len = 4, + .removed_turn_count = 2, + }; + + const text = try (SessionRewindSnapshot{ .result = result }).renderText( + std.testing.allocator, + ); + defer std.testing.allocator.free(text); + try std.testing.expectEqualStrings( + "[session rewind] rewound-session\nhistory_turns: 4\nremoved_turns: 2\n", + text, + ); + + const json = try (SessionRewindSnapshot{ .result = result }).renderJson( + std.testing.allocator, + ); + defer std.testing.allocator.free(json); + try std.testing.expectEqualStrings( + "{\"kind\":\"session_rewind\",\"id\":\"rewound-session\",\"status\":\"rewound\",\"history_turns\":4,\"removed_turns\":2}", + json, + ); + + const unchanged = session_store.SessionRewindResult{ + .session_id = @constCast("rewound-session"), + .history_len = 6, + .removed_turn_count = 0, + .status = .already_at_target, + }; + const unchanged_json = try (SessionRewindSnapshot{ + .result = unchanged, + }).renderJson(std.testing.allocator); + defer std.testing.allocator.free(unchanged_json); + try std.testing.expectEqualStrings( + "{\"kind\":\"session_rewind\",\"id\":\"rewound-session\",\"status\":\"already_at_target\",\"history_turns\":6,\"removed_turns\":0}", + unchanged_json, + ); +} + test "core doctor snapshot text and json stay stable" { const checks = [_]doctor_runtime.Check{ .{ .name = @constCast("auth"), .status = .ok, .detail = @constCast("AI_GATEWAY_API_KEY is configured") }, From aac16391e5033010cebe4542953ea8c2025fbb0d Mon Sep 17 00:00:00 2001 From: Aarya2004 Date: Tue, 1 Sep 2026 01:06:59 -0700 Subject: [PATCH 5/7] Add fx session fork and fx session rewind `fork --at ` names an absolute boundary read straight off the `[turn N]` labels `fx session ` prints. `rewind --by ` names a relative one, which is how undoing the last turns is asked for. The CLI resolves both to the absolute retained-turn count the store takes, and reports an out-of-range request with the session's real turn count. Claude-Session: https://claude.ai/code/session_01Hjm7J6N3SL5Y62TJ3bPxwD --- src/builtins/commands.zig | 8 +- src/core/cli/cli_surface.zig | 375 +++++++++++++++++++++++++++++++++-- 2 files changed, 369 insertions(+), 14 deletions(-) diff --git a/src/builtins/commands.zig b/src/builtins/commands.zig index adcdaa9dd..9a97cab8b 100644 --- a/src/builtins/commands.zig +++ b/src/builtins/commands.zig @@ -193,14 +193,16 @@ pub const top_level_specs = [_]TopLevelSpec{ .{ .kind = .session, .token = "session", - .usage = "session |--id [--json] | session resume [last|] | session resume --id | session migrate |--id [--allow-large] [--json] | session recover |--id [--json]", - .summary = "Inspect, resume, migrate, or recover saved sessions", + .usage = "session |--id [--json] | session resume [last|] | session resume --id | session migrate |--id [--allow-large] [--json] | session recover |--id [--json] | session fork |--id --at [--json] | session rewind |--id --by [--json]", + .summary = "Inspect, resume, migrate, recover, fork, or rewind saved sessions", .options = &.{ .{ .flag = "last", .description = "Inspect the current workspace session" }, .{ .flag = "--id ", .description = "Inspect a saved session by exact id" }, .{ .flag = "resume [last|]", .description = "Resume the latest workspace session or a session by id" }, .{ .flag = "migrate ", .description = "Migrate a saved session to the current format" }, .{ .flag = "recover ", .description = "Copy a recoverable corrupt session into a new session" }, + .{ .flag = "fork --at ", .description = "Branch a session into a new session holding its first turns" }, + .{ .flag = "rewind --by ", .description = "Drop the last turns from a session in place" }, .{ .flag = "--allow-large", .description = "Permit migrating an oversized session" }, json_option, }, @@ -312,6 +314,8 @@ pub const top_level_help_groups = [_]TopLevelHelpGroup{ .{ .usage = "session resume [last|id]", .summary = "Resume the latest workspace session or a session by id" }, .{ .usage = "session migrate ", .summary = "Migrate a saved session to the current format" }, .{ .usage = "session recover ", .summary = "Copy a recoverable corrupt session" }, + .{ .usage = "session fork --at ", .summary = "Branch a session at a turn into a new session" }, + .{ .usage = "session rewind --by ", .summary = "Drop the last turns from a session in place" }, } }, .{ .entries = &.{ .{ .kind = .login, .usage = "login [vercel|codex|grok]" }, diff --git a/src/core/cli/cli_surface.zig b/src/core/cli/cli_surface.zig index de99b8da0..8c167bc48 100644 --- a/src/core/cli/cli_surface.zig +++ b/src/core/cli/cli_surface.zig @@ -291,6 +291,28 @@ const SessionRecoveryOptions = struct { } }; +const SessionForkOptions = struct { + format: output_contracts.OutputFormat = .text, + session_id: []u8, + at_turn: usize, + + fn deinit(self: *SessionForkOptions, alloc: Allocator) void { + alloc.free(self.session_id); + self.* = undefined; + } +}; + +const SessionRewindOptions = struct { + format: output_contracts.OutputFormat = .text, + session_id: []u8, + by_turns: usize, + + fn deinit(self: *SessionRewindOptions, alloc: Allocator) void { + alloc.free(self.session_id); + self.* = undefined; + } +}; + const AcpOptions = struct { model: ?[]const u8 = null, log_file: ?[]const u8 = null, @@ -1346,6 +1368,143 @@ fn runNonInteractiveWithDeps( .handled_failure; } + if (rest.len > 0 and std.mem.eql(u8, rest[0], "fork")) { + var fork = parseSessionForkArgs( + alloc, + rest[1..], + ) catch |err| { + try writeUsageOrJsonError( + alloc, + cfg.command_catalog, + deps, + .session, + "session", + err, + rest[1..], + ); + return .handled_failure; + }; + defer fork.deinit(alloc); + + const workspace_root = try io_mod.realpathAlloc(alloc, "."); + defer alloc.free(workspace_root); + var store = session_store.Store.init( + alloc, + workspace_root, + ) catch |err| { + try writeLookupFailure(alloc, deps, "session", err, fork.format); + return .handled_failure; + }; + defer store.deinit(alloc); + + const history_len = readSessionHistoryLen( + alloc, + store, + fork.session_id, + ) catch |err| { + try writeLookupFailure(alloc, deps, "session", err, fork.format); + return .handled_failure; + }; + if (fork.at_turn == 0 or fork.at_turn > history_len) { + try writeSessionForkRangeFailure( + alloc, + deps, + fork.at_turn, + history_len, + fork.format, + ); + return .handled_failure; + } + + var result = store.forkSessionCopy( + alloc, + fork.session_id, + fork.at_turn, + .{}, + ) catch |err| { + try writeLookupFailure(alloc, deps, "session", err, fork.format); + return .handled_failure; + }; + defer result.deinit(alloc); + + const text = try (output_contracts.SessionForkSnapshot{ + .result = result, + }).render(alloc, fork.format); + defer alloc.free(text); + try writeFormattedOutput(deps, text, fork.format); + return if (result.status == .indeterminate) + .handled_failure + else + .handled_success; + } + + if (rest.len > 0 and std.mem.eql(u8, rest[0], "rewind")) { + var rewind = parseSessionRewindArgs( + alloc, + rest[1..], + ) catch |err| { + try writeUsageOrJsonError( + alloc, + cfg.command_catalog, + deps, + .session, + "session", + err, + rest[1..], + ); + return .handled_failure; + }; + defer rewind.deinit(alloc); + + const workspace_root = try io_mod.realpathAlloc(alloc, "."); + defer alloc.free(workspace_root); + var store = session_store.Store.init( + alloc, + workspace_root, + ) catch |err| { + try writeLookupFailure(alloc, deps, "session", err, rewind.format); + return .handled_failure; + }; + defer store.deinit(alloc); + + const history_len = readSessionHistoryLen( + alloc, + store, + rewind.session_id, + ) catch |err| { + try writeLookupFailure(alloc, deps, "session", err, rewind.format); + return .handled_failure; + }; + if (rewind.by_turns > history_len) { + try writeSessionRewindRangeFailure( + alloc, + deps, + rewind.by_turns, + history_len, + rewind.format, + ); + return .handled_failure; + } + + var result = store.rewindSession( + alloc, + rewind.session_id, + history_len - rewind.by_turns, + .{}, + ) catch |err| { + try writeLookupFailure(alloc, deps, "session", err, rewind.format); + return .handled_failure; + }; + defer result.deinit(alloc); + + const text = try (output_contracts.SessionRewindSnapshot{ + .result = result, + }).render(alloc, rewind.format); + defer alloc.free(text); + try writeFormattedOutput(deps, text, rewind.format); + return .handled_success; + } + if (rest.len > 0 and std.mem.eql(u8, rest[0], "migrate")) { var migration = parseSessionMigrationArgs(alloc, rest[1..]) catch |err| { try writeUsageOrJsonError(alloc, cfg.command_catalog, deps, .session, "session", err, rest[1..]); @@ -3157,6 +3316,36 @@ fn writeLookupFailure( "fx session: the recovery copy could not be confirmed; the source was left unchanged\n", ); }, + error.SessionTurnOutOfRange => { + try writeStderr( + deps, + "fx session: the requested turn is out of range for this session\n", + ); + }, + error.SessionForkRequiresCurrentSchema => { + try writeStderr( + deps, + "fx session: fork only applies to current schema-v3 sessions; migrate legacy sessions first\n", + ); + }, + error.SessionRewindRequiresCurrentSchema => { + try writeStderr( + deps, + "fx session: rewind only applies to current schema-v3 sessions; migrate legacy sessions first\n", + ); + }, + error.SessionForkArtifactsUnavailable => { + try writeStderr( + deps, + "fx session: the retained turn artifacts could not be copied; the source was left unchanged\n", + ); + }, + error.SessionForkIndeterminate => { + try writeStderr( + deps, + "fx session: the forked session could not be confirmed; the source was left unchanged\n", + ); + }, error.SessionAuthorityBoundaryUnavailable, error.SessionCommitBoundaryUnavailable, => { @@ -3198,6 +3387,84 @@ fn writeLookupFailure( } } +fn writeSessionMessageFailure( + alloc: Allocator, + deps: RunDeps, + err: anyerror, + message: []const u8, + format: output_contracts.OutputFormat, +) !void { + if (format == .json) { + return writeJsonCommandFailure(alloc, deps, "session", err, message); + } + try writeStderr(deps, "fx session: "); + try writeStderr(deps, message); + try writeStderr(deps, "\n"); +} + +fn turnPlural(count: usize) []const u8 { + return if (count == 1) "" else "s"; +} + +fn writeSessionForkRangeFailure( + alloc: Allocator, + deps: RunDeps, + at_turn: usize, + history_len: usize, + format: output_contracts.OutputFormat, +) !void { + const message = try std.fmt.allocPrint( + alloc, + "turn {d} is out of range; session has {d} turn{s}", + .{ at_turn, history_len, turnPlural(history_len) }, + ); + defer alloc.free(message); + return writeSessionMessageFailure( + alloc, + deps, + error.SessionTurnOutOfRange, + message, + format, + ); +} + +fn writeSessionRewindRangeFailure( + alloc: Allocator, + deps: RunDeps, + by_turns: usize, + history_len: usize, + format: output_contracts.OutputFormat, +) !void { + const message = try std.fmt.allocPrint( + alloc, + "cannot rewind {d} turn{s}; session has {d} turn{s}", + .{ + by_turns, + turnPlural(by_turns), + history_len, + turnPlural(history_len), + }, + ); + defer alloc.free(message); + return writeSessionMessageFailure( + alloc, + deps, + error.SessionTurnOutOfRange, + message, + format, + ); +} + +fn readSessionHistoryLen( + alloc: Allocator, + store: session_store.Store, + session_id: []const u8, +) !usize { + var state = try store.loadReadOnly(alloc, session_id); + defer state.deinit(alloc); + return state.history.len; +} + fn writeSessionDetailFailure( alloc: Allocator, deps: RunDeps, @@ -3225,18 +3492,7 @@ fn writeSessionDetailFailure( ), }; defer alloc.free(message); - if (format == .json) { - return writeJsonCommandFailure( - alloc, - deps, - "session", - err, - message, - ); - } - try writeStderr(deps, "fx session: "); - try writeStderr(deps, message); - try writeStderr(deps, "\n"); + return writeSessionMessageFailure(alloc, deps, err, message, format); } fn commandFailureMessage(err: anyerror) ?[]const u8 { @@ -3248,6 +3504,8 @@ fn commandFailureMessage(err: anyerror) ?[]const u8 { error.InvalidSessionDetailArgs, error.InvalidSessionMigrationArgs, error.InvalidSessionRecoveryArgs, + error.InvalidSessionForkArgs, + error.InvalidSessionRewindArgs, error.InvalidResumeArgs, => "invalid arguments", else => null, @@ -3275,6 +3533,11 @@ fn lookupFailureMessage(err: anyerror) ?[]const u8 { error.SessionRecoveryUnsupportedSchema => "recovery is unavailable for this unsupported session version", error.SessionRecoveryBoundaryInvalid => "no exact trustworthy recovery boundary was found; the source was left unchanged", error.SessionRecoveryIndeterminate => "the recovery copy could not be confirmed; the source was left unchanged", + error.SessionTurnOutOfRange => "the requested turn is out of range for this session", + error.SessionForkRequiresCurrentSchema => "fork only applies to current schema-v3 sessions; migrate legacy sessions first", + error.SessionRewindRequiresCurrentSchema => "rewind only applies to current schema-v3 sessions; migrate legacy sessions first", + error.SessionForkArtifactsUnavailable => "the retained turn artifacts could not be copied; the source was left unchanged", + error.SessionForkIndeterminate => "the forked session could not be confirmed; the source was left unchanged", error.SessionAuthorityBoundaryUnavailable, error.SessionCommitBoundaryUnavailable, => "session authority is temporarily unavailable while an incomplete commit is resolved", @@ -3788,6 +4051,94 @@ fn parseSessionRecoveryArgs( }; } +fn parseSessionForkArgs( + alloc: Allocator, + args: []const [:0]const u8, +) !SessionForkOptions { + var format: output_contracts.OutputFormat = .text; + var session_id: ?[]u8 = null; + var at_turn: ?usize = null; + errdefer if (session_id) |id| alloc.free(id); + + var i: usize = 0; + while (i < args.len) : (i += 1) { + const arg = args[i]; + if (std.mem.eql(u8, arg, "--json")) { + format = .json; + continue; + } + if (std.mem.eql(u8, arg, "--at")) { + if (at_turn != null) return error.InvalidSessionForkArgs; + i += 1; + if (i >= args.len) return error.InvalidSessionForkArgs; + at_turn = parseTurnCount(args[i]) catch + return error.InvalidSessionForkArgs; + continue; + } + if (session_id != null) return error.InvalidSessionForkArgs; + const exact_id = std.mem.eql(u8, arg, "--id"); + if (exact_id) { + i += 1; + if (i >= args.len) return error.InvalidSessionForkArgs; + } + const trimmed = std.mem.trim(u8, args[i], " \t\r\n"); + if (trimmed.len == 0) return error.InvalidSessionForkArgs; + session_id = try alloc.dupe(u8, trimmed); + } + return .{ + .format = format, + .session_id = session_id orelse return error.InvalidSessionForkArgs, + .at_turn = at_turn orelse return error.InvalidSessionForkArgs, + }; +} + +fn parseSessionRewindArgs( + alloc: Allocator, + args: []const [:0]const u8, +) !SessionRewindOptions { + var format: output_contracts.OutputFormat = .text; + var session_id: ?[]u8 = null; + var by_turns: ?usize = null; + errdefer if (session_id) |id| alloc.free(id); + + var i: usize = 0; + while (i < args.len) : (i += 1) { + const arg = args[i]; + if (std.mem.eql(u8, arg, "--json")) { + format = .json; + continue; + } + if (std.mem.eql(u8, arg, "--by")) { + if (by_turns != null) return error.InvalidSessionRewindArgs; + i += 1; + if (i >= args.len) return error.InvalidSessionRewindArgs; + by_turns = parseTurnCount(args[i]) catch + return error.InvalidSessionRewindArgs; + continue; + } + if (session_id != null) return error.InvalidSessionRewindArgs; + const exact_id = std.mem.eql(u8, arg, "--id"); + if (exact_id) { + i += 1; + if (i >= args.len) return error.InvalidSessionRewindArgs; + } + const trimmed = std.mem.trim(u8, args[i], " \t\r\n"); + if (trimmed.len == 0) return error.InvalidSessionRewindArgs; + session_id = try alloc.dupe(u8, trimmed); + } + return .{ + .format = format, + .session_id = session_id orelse return error.InvalidSessionRewindArgs, + .by_turns = by_turns orelse return error.InvalidSessionRewindArgs, + }; +} + +fn parseTurnCount(raw: []const u8) !usize { + const trimmed = std.mem.trim(u8, raw, " \t\r\n"); + if (trimmed.len == 0) return error.InvalidTurnCount; + return std.fmt.parseInt(usize, trimmed, 10) catch error.InvalidTurnCount; +} + fn parseResumeArgs( alloc: Allocator, command_catalog: CommandCatalog, From 82718e3c80331bf4c60cc6e079362aa9e0d1fbd5 Mon Sep 17 00:00:00 2001 From: Aarya2004 Date: Tue, 1 Sep 2026 01:16:38 -0700 Subject: [PATCH 6/7] Document session fork and rewind and cover them end to end The e2e suite drives the built binary against a fake gateway: it seeds a multi-turn session, forks it, resumes the branch, and checks the source directory is byte-identical afterward. Classified verification-only in the PGSO corpus, since branching and undo are deliberate rare operations that must stay correct without being made hot. Claude-Session: https://claude.ai/code/session_01Hjm7J6N3SL5Y62TJ3bPxwD --- README.md | 11 + scripts/pgso/corpus.json | 1 + tests/e2e/ci-shard-weights.json | 1 + tests/e2e/cli.test.ts | 6 +- tests/e2e/session-fork.test.ts | 398 ++++++++++++++++++++++++++++++++ 5 files changed, 415 insertions(+), 2 deletions(-) create mode 100644 tests/e2e/session-fork.test.ts diff --git a/README.md b/README.md index 37f120eb0..e67ba3ada 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,17 @@ fx session resume last fx session resume --id ``` +`fx session ` prints the saved conversation with a `[turn N]` label above every turn. Those labels are the boundaries the branch and undo commands take: + +```bash +fx session fork --at 7 +fx session rewind --by 2 +``` + +`fork` copies the first 7 turns into a brand-new session and prints the new ID to resume. The source session is left exactly as it was. `rewind` drops the last 2 turns from the session in place, keeping its ID. Both accept `--id ` and `--json`. + +Rewound turns are not erased. The rewind is recorded as a new revision in the session's event log, and the files those turns referenced stay on disk. + Each interactive session names its terminal tab. The title prefers the session name, falls back to the workspace name, and keeps the active model as secondary context. Renaming or resuming a session updates the tab, and exiting clears the fx-owned title. Noninteractive commands do not emit terminal-title controls. Run `/feedback` to open the feedback form at `fx.sh/feedback`. It does not create a diagnostic or change the clipboard. diff --git a/scripts/pgso/corpus.json b/scripts/pgso/corpus.json index cc37f1c68..738964002 100644 --- a/scripts/pgso/corpus.json +++ b/scripts/pgso/corpus.json @@ -113,6 +113,7 @@ ], "verification_scenarios": [ {"name": "verify-auto-mode-reliability", "argv": ["bun", "test", "--max-concurrency", "1", "./auto-mode-reliability.test.ts"], "test_file": "auto-mode-reliability.test.ts"}, + {"name": "verify-session-fork", "argv": ["bun", "test", "--max-concurrency", "1", "./session-fork.test.ts"], "test_file": "session-fork.test.ts", "requires_tmux": false}, {"name": "verify-oauth-keychain-migration", "argv": ["bun", "test", "--max-concurrency", "1", "./oauth-keychain-migration.test.ts"], "test_file": "oauth-keychain-migration.test.ts", "allow_keychain": true}, {"name": "verify-tui-auth-source-selection", "argv": ["bun", "test", "--max-concurrency", "1", "./tui-auth-source-selection.test.ts"], "test_file": "tui-auth-source-selection.test.ts"}, {"name": "verify-tui-composer-edit-contracts", "argv": ["bun", "test", "--max-concurrency", "1", "./tui-composer-edit-contracts.test.ts"], "test_file": "tui-composer-edit-contracts.test.ts"}, diff --git a/tests/e2e/ci-shard-weights.json b/tests/e2e/ci-shard-weights.json index 157ac392c..19d4444b6 100644 --- a/tests/e2e/ci-shard-weights.json +++ b/tests/e2e/ci-shard-weights.json @@ -18,6 +18,7 @@ { "file": "oauth-keychain-migration.test.ts", "weight": 30 }, { "file": "permission-errors.test.ts", "weight": 1 }, { "file": "prompt-history.test.ts", "weight": 14 }, + { "file": "session-fork.test.ts", "weight": 2 }, { "file": "session-recovery.test.ts", "weight": 13 }, { "file": "terminal-host.test.ts", "weight": 273 }, { "file": "tmux-helpers.test.ts", "weight": 2 }, diff --git a/tests/e2e/cli.test.ts b/tests/e2e/cli.test.ts index e221b5bb7..eafb7ec0d 100644 --- a/tests/e2e/cli.test.ts +++ b/tests/e2e/cli.test.ts @@ -350,7 +350,7 @@ With --prompt-permissions, JSON and quiet requests may prompt on stderr only whe ); test( - "fx session help documents inspect resume migrate and recover", + "fx session help documents inspect resume migrate recover fork and rewind", async () => { for (const args of [ ["session", "--help"], @@ -359,11 +359,13 @@ With --prompt-permissions, JSON and quiet requests may prompt on stderr only whe const r = await runFx(args); expect(r.code).toBe(0); expect(r.stderr).toBe(""); - expect(r.stdout).toContain("Inspect, resume, migrate, or recover saved sessions"); + expect(r.stdout).toContain("Inspect, resume, migrate, recover, fork, or rewind saved sessions"); expect(r.stdout).toContain("session |--id "); expect(r.stdout).toContain("session resume [last|]"); expect(r.stdout).toContain("session migrate |--id "); expect(r.stdout).toContain("session recover |--id "); + expect(r.stdout).toContain("session fork |--id --at "); + expect(r.stdout).toContain("session rewind |--id --by "); } }, TIMEOUT, diff --git a/tests/e2e/session-fork.test.ts b/tests/e2e/session-fork.test.ts new file mode 100644 index 000000000..13a4115fa --- /dev/null +++ b/tests/e2e/session-fork.test.ts @@ -0,0 +1,398 @@ +import { describe, expect, test } from "bun:test"; +import { + mkdirSync, + mkdtempSync, + readFileSync, + readdirSync, + realpathSync, + rmSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { runFx } from "../evals/eval-helpers"; +import { fakeGatewayFinalText, startFakeGateway } from "./tmux-helpers"; + +const TIMEOUT = 60_000; + +function sessionFileSnapshot(sessionDir: string): Record { + return Object.fromEntries( + readdirSync(sessionDir, { withFileTypes: true }) + .filter((entry) => entry.isFile()) + .sort((left, right) => left.name.localeCompare(right.name)) + .map((entry) => [ + entry.name, + readFileSync(join(sessionDir, entry.name)).toString("base64"), + ]), + ); +} + +function makeWorkspace(prefix: string) { + const root = mkdtempSync(join(tmpdir(), prefix)); + const home = join(root, "home"); + const workspace = join(root, "workspace"); + mkdirSync(home); + mkdirSync(workspace); + return { root, home, workspaceRoot: realpathSync(workspace) }; +} + +async function seedSession( + workspaceRoot: string, + home: string, + answers: string[], +): Promise { + const gateway = startFakeGateway( + answers.map((answer) => fakeGatewayFinalText(answer)), + ); + try { + let sessionId = ""; + for (const [index, answer] of answers.entries()) { + const args = + index === 0 + ? ["ask", "--json", "--auto", `prompt for ${answer}`] + : [ + "ask", + "--json", + "--auto", + "--resume", + "last", + `prompt for ${answer}`, + ]; + const run = await runFx(args, { + cwd: workspaceRoot, + env: { + HOME: home, + AI_GATEWAY_API_KEY: "e2e-placeholder", + VERCEL_OIDC_TOKEN: "", + FX_GATEWAY_BASE_URL: gateway.baseUrl, + FX_GATEWAY_CHAT_URL: gateway.chatUrl, + }, + }); + expect(run.code).toBe(0); + sessionId = JSON.parse(run.stdout).session_id; + } + return sessionId; + } finally { + gateway.stop?.(); + } +} + +async function sessionDetail( + workspaceRoot: string, + home: string, + sessionId: string, +) { + const detail = await runFx(["session", "--id", sessionId, "--json"], { + cwd: workspaceRoot, + env: { HOME: home }, + }); + expect(detail.code).toBe(0); + return JSON.parse(detail.stdout); +} + +function promptTexts(detail: { history: { user: { text: string } }[] }) { + return detail.history.map((turn) => turn.user.text); +} + +describe("session fork", () => { + test( + "fork branches a turn prefix into a new session and leaves the source unchanged", + async () => { + const { root, home, workspaceRoot } = makeWorkspace("fx-session-fork-"); + try { + const sessionId = await seedSession(workspaceRoot, home, [ + "ONE", + "TWO", + "THREE", + "FOUR", + ]); + const sessionDir = join(home, ".fx", "sessions", sessionId); + const sourceBefore = sessionFileSnapshot(sessionDir); + + const fork = await runFx( + ["session", "fork", sessionId, "--at", "2"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(fork.code).toBe(0); + expect(fork.stderr).toBe(""); + const forkedId = fork.stdout.match(/ to (\S+)/)![1]!; + expect(forkedId).not.toBe(sessionId); + expect(fork.stdout).toBe( + `[session fork] forked ${sessionId} to ${forkedId}\n` + + "history_turns: 2 of 4\n" + + `resume: fx --resume ${forkedId}\n`, + ); + + expect(sessionFileSnapshot(sessionDir)).toEqual(sourceBefore); + + const forkedDetail = await sessionDetail(workspaceRoot, home, forkedId); + expect(forkedDetail.history_len).toBe(2); + expect(promptTexts(forkedDetail)).toEqual([ + "prompt for ONE", + "prompt for TWO", + ]); + + const sourceDetail = await sessionDetail( + workspaceRoot, + home, + sessionId, + ); + expect(sourceDetail.history_len).toBe(4); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + + test( + "a forked session resumes and continues from its branch point", + async () => { + const { root, home, workspaceRoot } = makeWorkspace( + "fx-session-fork-resume-", + ); + try { + const sessionId = await seedSession(workspaceRoot, home, [ + "ONE", + "TWO", + "THREE", + ]); + const fork = await runFx( + ["session", "fork", "--id", sessionId, "--at", "1", "--json"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(fork.code).toBe(0); + const forkedId = JSON.parse(fork.stdout).forked_id; + + const gateway = startFakeGateway([ + fakeGatewayFinalText("BRANCH_CONTINUED"), + ]); + try { + const resumed = await runFx( + ["ask", "--json", "--auto", "--resume", forkedId, "keep going"], + { + cwd: workspaceRoot, + env: { + HOME: home, + AI_GATEWAY_API_KEY: "e2e-placeholder", + VERCEL_OIDC_TOKEN: "", + FX_GATEWAY_BASE_URL: gateway.baseUrl, + FX_GATEWAY_CHAT_URL: gateway.chatUrl, + }, + }, + ); + expect(resumed.code).toBe(0); + expect(JSON.parse(resumed.stdout).session_id).toBe(forkedId); + } finally { + gateway.stop?.(); + } + + const detail = await sessionDetail(workspaceRoot, home, forkedId); + expect(promptTexts(detail)).toEqual(["prompt for ONE", "keep going"]); + expect( + (await sessionDetail(workspaceRoot, home, sessionId)).history_len, + ).toBe(3); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + + test( + "fork --json reports both turn counts", + async () => { + const { root, home, workspaceRoot } = makeWorkspace( + "fx-session-fork-json-", + ); + try { + const sessionId = await seedSession(workspaceRoot, home, [ + "ONE", + "TWO", + "THREE", + ]); + const fork = await runFx( + ["session", "fork", sessionId, "--at", "2", "--json"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(fork.code).toBe(0); + expect(fork.stderr).toBe(""); + const payload = JSON.parse(fork.stdout); + expect(payload.kind).toBe("session_fork"); + expect(payload.source_id).toBe(sessionId); + expect(payload.forked_id).not.toBe(sessionId); + expect(payload.status).toBe("forked"); + expect(payload.history_turns).toBe(2); + expect(payload.source_history_turns).toBe(3); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + + test( + "fork rejects a turn past the end of the session in text and json", + async () => { + const { root, home, workspaceRoot } = makeWorkspace( + "fx-session-fork-range-", + ); + try { + const sessionId = await seedSession(workspaceRoot, home, [ + "ONE", + "TWO", + ]); + + const text = await runFx( + ["session", "fork", sessionId, "--at", "12"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(text.code).toBe(1); + expect(text.stdout).toBe(""); + expect(text.stderr).toBe( + "fx session: turn 12 is out of range; session has 2 turns\n", + ); + + const json = await runFx( + ["session", "fork", sessionId, "--at", "12", "--json"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(json.code).toBe(1); + expect(json.stderr).toBe(""); + const payload = JSON.parse(json.stdout); + expect(payload.kind).toBe("session"); + expect(payload.code).toBe("SessionTurnOutOfRange"); + expect(payload.error).toBe( + "turn 12 is out of range; session has 2 turns", + ); + + expect( + (await sessionDetail(workspaceRoot, home, sessionId)).history_len, + ).toBe(2); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); +}); + +describe("session rewind", () => { + test( + "rewind drops the last turns in place and keeps the session id", + async () => { + const { root, home, workspaceRoot } = makeWorkspace("fx-session-rewind-"); + try { + const sessionId = await seedSession(workspaceRoot, home, [ + "ONE", + "TWO", + "THREE", + ]); + + const rewind = await runFx( + ["session", "rewind", sessionId, "--by", "2"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(rewind.code).toBe(0); + expect(rewind.stderr).toBe(""); + expect(rewind.stdout).toBe( + `[session rewind] ${sessionId}\n` + + "history_turns: 1\n" + + "removed_turns: 2\n", + ); + + const detail = await sessionDetail(workspaceRoot, home, sessionId); + expect(detail.id).toBe(sessionId); + expect(promptTexts(detail)).toEqual(["prompt for ONE"]); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + + test( + "rewind --json reports the removed turn count and a no-op rewind", + async () => { + const { root, home, workspaceRoot } = makeWorkspace( + "fx-session-rewind-json-", + ); + try { + const sessionId = await seedSession(workspaceRoot, home, [ + "ONE", + "TWO", + "THREE", + ]); + + const rewind = await runFx( + ["session", "rewind", "--id", sessionId, "--by", "1", "--json"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(rewind.code).toBe(0); + expect(rewind.stderr).toBe(""); + expect(JSON.parse(rewind.stdout)).toEqual({ + kind: "session_rewind", + id: sessionId, + status: "rewound", + history_turns: 2, + removed_turns: 1, + }); + + const noop = await runFx( + ["session", "rewind", sessionId, "--by", "0", "--json"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(noop.code).toBe(0); + expect(JSON.parse(noop.stdout)).toEqual({ + kind: "session_rewind", + id: sessionId, + status: "already_at_target", + history_turns: 2, + removed_turns: 0, + }); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + + test( + "rewind rejects dropping more turns than the session holds", + async () => { + const { root, home, workspaceRoot } = makeWorkspace( + "fx-session-rewind-range-", + ); + try { + const sessionId = await seedSession(workspaceRoot, home, [ + "ONE", + "TWO", + ]); + + const text = await runFx( + ["session", "rewind", sessionId, "--by", "9"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(text.code).toBe(1); + expect(text.stdout).toBe(""); + expect(text.stderr).toBe( + "fx session: cannot rewind 9 turns; session has 2 turns\n", + ); + + const json = await runFx( + ["session", "rewind", sessionId, "--by", "9", "--json"], + { cwd: workspaceRoot, env: { HOME: home } }, + ); + expect(json.code).toBe(1); + expect(json.stderr).toBe(""); + expect(JSON.parse(json.stdout).code).toBe("SessionTurnOutOfRange"); + + expect( + (await sessionDetail(workspaceRoot, home, sessionId)).history_len, + ).toBe(2); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); +}); From ac1feeef77566c5cf586554ceefd31775e10fc4a Mon Sep 17 00:00:00 2001 From: Aarya2004 Date: Tue, 1 Sep 2026 09:00:10 -0700 Subject: [PATCH 7/7] Clear the paused response when a rewind drops turns A recovery checkpoint describes an in-flight turn that sits past the last committed one, so any rewind that drops turns leaves it describing work the session no longer has. Keeping it let `--continue-recovery` resume a response for a turn that was removed. `forkSessionCopy` already cleared it. A no-op rewind still keeps it, because that path drops nothing and commits nothing. Claude-Session: https://claude.ai/code/session_01Hjm7J6N3SL5Y62TJ3bPxwD --- src/core/session/session_store.zig | 84 +++++++++++++++++++++++++++++- 1 file changed, 83 insertions(+), 1 deletion(-) diff --git a/src/core/session/session_store.zig b/src/core/session/session_store.zig index 8472001ff..e209e6065 100644 --- a/src/core/session/session_store.zig +++ b/src/core/session/session_store.zig @@ -4459,7 +4459,8 @@ pub const Store = struct { /// Drops every turn after `retained_turns` from a session in place, keeping /// its id. The rewind is recorded as a state replacement, so the dropped /// turns stay in the event log until it is compacted and their artifacts - /// are left on disk. + /// are left on disk. Any paused response is cleared too, since it describes + /// a turn past the retained tail. pub fn rewindSession( self: Store, alloc: Allocator, @@ -4488,6 +4489,10 @@ pub const Store = struct { var rewound = try loaded.state.dupe(alloc); defer rewound.deinit(alloc); try truncateSessionHistory(alloc, &rewound, retained_turns); + if (rewound.recovery_checkpoint) |*checkpoint| { + checkpoint.deinit(alloc); + rewound.recovery_checkpoint = null; + } rewound.updated_at_ms = io_mod.milliTimestamp(); _ = try loaded.commitStateReplacement( alloc, @@ -14598,3 +14603,80 @@ test "rewind refuses a session that is already open for writing" { ), ); } + +fn testRecoveryCheckpoint() session_codec.RecoveryCheckpoint { + return .{ + .turn_id = 1, + .user = .{ .text = @constCast("prompt") }, + .assistant_source = @constCast(""), + .cause = .network_interrupted, + .action = .retrying_request, + .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, + .requested_fast_mode = false, + .fast_mode = false, + .max_provider_attempts = 10, + .consumed_provider_attempts = 1, + }; +} + +test "rewind clears a paused response because it belongs to the dropped tail" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const session_id = "rewind-clears-checkpoint"; + try writeForkSourceFixture(alloc, ctx.store, session_id, ctx.workspace, 3, 0); + + var writable = try ctx.store.resumeForWrite(alloc, session_id); + _ = try writable.appendEvent( + alloc, + .{ .recovery_checkpoint_set = .{ .checkpoint = testRecoveryCheckpoint() } }, + 20, + .retry_expected_tail, + .{}, + ); + writable.deinit(alloc); + + var result = try ctx.store.rewindSession(alloc, session_id, 1, .{}); + defer result.deinit(alloc); + try std.testing.expectEqual(@as(usize, 2), result.removed_turn_count); + + var state = try ctx.store.loadReadOnly(alloc, session_id); + defer state.deinit(alloc); + try std.testing.expect(state.recovery_checkpoint == null); +} + +test "rewind leaves a paused response untouched when already at target" { + const alloc = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var ctx = try initTempStore(alloc, &tmp); + defer ctx.deinit(alloc); + + const session_id = "rewind-noop-keeps-checkpoint"; + try writeForkSourceFixture(alloc, ctx.store, session_id, ctx.workspace, 3, 0); + + var writable = try ctx.store.resumeForWrite(alloc, session_id); + _ = try writable.appendEvent( + alloc, + .{ .recovery_checkpoint_set = .{ .checkpoint = testRecoveryCheckpoint() } }, + 20, + .retry_expected_tail, + .{}, + ); + writable.deinit(alloc); + + var result = try ctx.store.rewindSession(alloc, session_id, 3, .{}); + defer result.deinit(alloc); + try std.testing.expectEqual( + store_types.SessionRewindStatus.already_at_target, + result.status, + ); + + var state = try ctx.store.loadReadOnly(alloc, session_id); + defer state.deinit(alloc); + try std.testing.expect(state.recovery_checkpoint != null); + try std.testing.expectEqual(@as(u64, 1), state.recovery_checkpoint.?.turn_id); +}