feat(claude): support conversation rewind via native session truncation

cli/src/claude/claudeRemote.ts

@@ -90,6 +90,28 @@ export async function claudeRemote(opts: {
     }
     const forkedFrom = forkSession ? startFrom : null;
 
+    // One-shot rewind flags (set by the RewindConversation handler) pass through
+    // claudeArgs; filterCatalogAffectingClaudeArgs strips them from additionalArgs,
+    // so they must be parsed here into first-class SDK options.
+    let resumeSessionAt: string | undefined;
+    const resumeDropsTurn: string[] = [];
+    if (opts.claudeArgs) {
+        for (let i = 0; i < opts.claudeArgs.length; i++) {
+            if (opts.claudeArgs[i] === '--resume-session-at' && i + 1 < opts.claudeArgs.length) {
+                resumeSessionAt = opts.claudeArgs[++i];
+            } else if (opts.claudeArgs[i] === '--resume-drops-turn' && i + 1 < opts.claudeArgs.length) {
+                resumeDropsTurn.push(opts.claudeArgs[++i]);
+            }
+        }
+    }
+    // A rewind restart must spawn Claude immediately (no user prompt yet):
+    // the native truncation only materializes once the new process starts with
+    // the resume flags, and the launcher waits for that before reporting
+    // success to the hub. Like --fork-session, start query() without waiting
+    // for an initial child prompt; stream-json input stays open for later turns.
+    let awaitingRewindInit = resumeSessionAt !== undefined;
+    const REWIND_READY_DELAY_MS = 4_000;
+
     // Mode starts from the persisted session for fork bootstrap; updated when
     // the first child prompt arrives. plan/auto must be present at process start.
     const bootstrapMode: EnhancedMode = opts.bootstrapMode ?? { permissionMode: 'default' };
@@ -163,6 +185,8 @@ export async function claudeRemote(opts: {
         cwd: opts.path,
         resume: startFrom ?? undefined,
         forkSession,
+        resumeSessionAt,
+        resumeDropsTurn: resumeDropsTurn.length > 0 ? resumeDropsTurn : undefined,
         mcpServers: opts.mcpServers,
         permissionMode: bootstrapMode.permissionMode,
         model: bootstrapMode.model,
@@ -185,7 +209,7 @@ export async function claudeRemote(opts: {
         additionalDirectories: [getHapiBlobsDir()],
     }
 
-    if (!awaitingForkInit) {
+    if (!awaitingForkInit && !awaitingRewindInit) {
         const first = await applyInitialTurn();
         if (!first) {
             return;
@@ -277,7 +301,23 @@ export async function claudeRemote(opts: {
         })();
     };
 
-    updateThinking(true);
+    // A rewind respawn starts with no running turn: booting into "thinking"
+    // would leave the session permanently generating. Report idle once the
+    // process has survived long enough to prove the resume flags were accepted
+    // (a rejected resume exits almost immediately).
+    if (awaitingRewindInit) {
+        setTimeout(() => {
+            awaitingRewindInit = false;
+            updateThinking(false);
+            void opts.onReady?.();
+            // No result message will ever arrive for the skipped initial turn,
+            // so the queue consumer must be started here or user messages
+            // sent after the rewind would never reach Claude.
+            scheduleNextMessage();
+        }, REWIND_READY_DELAY_MS);
+    } else {
+        updateThinking(true);
+    }
     try {
         logger.debug(`[claudeRemote] Starting to iterate over response`);
 
@@ -321,6 +361,17 @@ export async function claudeRemote(opts: {
                     }
                     initial = first;
                 }
+
+                // Rewind restart: no child prompt was fed, so nothing is running.
+                // Clear the boot-time thinking state and report ready — otherwise
+                // the session looks permanently "generating" until the next turn.
+                if (awaitingRewindInit) {
+                    awaitingRewindInit = false;
+                    updateThinking(false);
+                    if (opts.onReady) {
+                        await opts.onReady();
+                    }
+                }
             }
 
             // Capture the /compact outcome. Only a reported failure is recorded:

cli/src/claude/claudeRemoteLauncher.ts

@@ -36,6 +36,12 @@ interface PermissionsField {
 // is dropped, so a later unrelated failure gets its own fresh budget. Only
 // the one message is given up on -- the session/process itself is not ended.
 const MAX_IMMEDIATE_RESPAWN_FAILURES = 3;
+/**
+ * A rewind respawn emits no system/init until its first prompt, but a rejected
+ * resume exits within moments of spawn. If the new attempt survives this long,
+ * the resume flags were accepted and the truncation is treated as applied.
+ */
+const REWIND_CONFIRM_MS = 8_000;
 
 function getRespawnBackoffMs(): number {
     const raw = process.env.CLAUDE_REMOTE_RESPAWN_BACKOFF_MS;
@@ -49,6 +55,7 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
     private readonly session: Session;
     private abortController: AbortController | null = null;
     private abortFuture: Future<void> | null = null;
+    private restartRequested = false;
     private permissionHandler: PermissionHandler | null = null;
     private handleSessionFound: ((sessionId: string) => void) | null = null;
 
@@ -114,6 +121,17 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
         await this.handleSwitchRequest();
     }
 
+    /**
+     * Abort the current SDK attempt (without an exit reason) so the main loop
+     * respawns Claude with fresh one-shot args. Used by rewind, which needs a
+     * process restart to apply --resume-session-at.
+     */
+    public async requestRestart(): Promise<void> {
+        logger.debug('[remote]: doRestart');
+        this.restartRequested = true;
+        await this.abort();
+    }
+
     public async launch(): Promise<RemoteLauncherExitReason> {
         return this.start({
             onExit: () => this.handleExitFromUi(),
@@ -128,6 +146,8 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
         const session = this.session;
         const messageBuffer = this.messageBuffer;
 
+        session.requestRemoteRestart = () => this.requestRestart();
+
         this.setupAbortHandlers(session.client.rpcHandlerManager, {
             onAbort: () => this.handleAbortRequest(),
             onSwitch: () => this.handleSwitchRequest()
@@ -153,6 +173,13 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
 
         const handleSessionFound = (sessionId: string) => {
             sdkToLogConverter.updateSessionId(sessionId);
+            // Rewind restart: system/init means the respawned process started
+            // with the resume flags accepted — the native truncation is real.
+            if (session.rewindAck) {
+                const ack = session.rewindAck;
+                session.rewindAck = null;
+                ack(true);
+            }
         };
         this.handleSessionFound = handleSessionFound;
         session.addSessionFoundCallback(handleSessionFound);
@@ -336,6 +363,18 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                 }
 
                 previousSessionId = session.sessionId;
+                // Rewind confirmation: surviving this window means the resume
+                // flags were accepted (a rejected resume exits almost immediately).
+                let rewindConfirmTimer: ReturnType<typeof setTimeout> | null = null;
+                if (session.rewindAck) {
+                    rewindConfirmTimer = setTimeout(() => {
+                        if (session.rewindAck) {
+                            const ack = session.rewindAck;
+                            session.rewindAck = null;
+                            ack(true);
+                        }
+                    }, REWIND_CONFIRM_MS);
+                }
                 const controller = new AbortController();
                 this.abortController = controller;
                 this.abortFuture = new Future<void>();
@@ -428,6 +467,7 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                                 deliveredMessageThisAttempt = true;
                                 const deliveredText = session.expandSkillReference(p.message)
                                 inFlightMessage = { items: p.items, mode: p.mode, isolate: p.isolate, deliveredText };
+                                session.onUserTurnDelivered?.(p.items.flatMap((item) => item.localId ? [item.localId] : []))
                                 session.client.notePendingHubPromptEcho(
                                     deliveredText,
                                     p.items.flatMap((item) => item.localId ? [item.localId] : [])
@@ -463,6 +503,7 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                                 deliveredMessageThisAttempt = true;
                                 const deliveredText = session.expandSkillReference(msg.message)
                                 inFlightMessage = { items: msg.items, mode: msg.mode, isolate: msg.isolate, deliveredText };
+                                session.onUserTurnDelivered?.(msg.items.flatMap((item) => item.localId ? [item.localId] : []))
                                 session.client.notePendingHubPromptEcho(
                                     deliveredText,
                                     msg.items.flatMap((item) => item.localId ? [item.localId] : [])
@@ -536,8 +577,23 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                         signal: controller.signal,
                     });
 
+                    // Attempt finished cleanly: the resume flags were accepted.
+                    if (rewindConfirmTimer) {
+                        clearTimeout(rewindConfirmTimer);
+                        rewindConfirmTimer = null;
+                        if (session.rewindAck) {
+                            const ack = session.rewindAck;
+                            session.rewindAck = null;
+                            ack(true);
+                        }
+                    }
+
                     if (!this.exitReason && controller.signal.aborted) {
-                        session.client.sendSessionEvent({ type: 'message', message: 'Aborted by user' });
+                        if (this.restartRequested) {
+                            this.restartRequested = false;
+                        } else {
+                            session.client.sendSessionEvent({ type: 'message', message: 'Aborted by user' });
+                        }
                     }
 
                     // A full attempt completed without throwing. Clear the
@@ -567,6 +623,21 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                     }
                 } catch (e) {
                     logger.debug('[remote]: launch error', e);
+                    if (rewindConfirmTimer) {
+                        clearTimeout(rewindConfirmTimer);
+                        rewindConfirmTimer = null;
+                    }
+                    // A deterministic resume rejection means the native state is
+                    // unchanged — report failure immediately instead of letting
+                    // the rewind handler time out.
+                    if (session.rewindAck) {
+                        const detail0 = e instanceof Error ? e.message : String(e);
+                        if (/Resume rejected|resume-drops-turn|would discard/i.test(detail0)) {
+                            const ack = session.rewindAck;
+                            session.rewindAck = null;
+                            ack(false, detail0);
+                        }
+                    }
 
                     // Restores a message batch that was already
                     // dequeued+acked from the queue (see
@@ -627,6 +698,23 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                                 message: `Process exited unexpectedly ${MAX_IMMEDIATE_RESPAWN_FAILURES} times in a row: ${detail}. Dropping the queued message; resolve the issue and resend it.`
                             });
                             immediateFailureCount = 0;
+                        } else if (this.restartRequested) {
+                            // Intentional restart (rewind): the aborted in-flight
+                            // turn belongs to the pre-rewind history. Re-delivering
+                            // it into the respawned (truncated) process would
+                            // diverge native history from the hub transcript.
+                            for (const item of inFlightMessage?.items ?? []) {
+                                if (item.localId) {
+                                    session.client.discardPendingHubPromptEcho(item.localId)
+                                }
+                            }
+                            if (inFlightMessage?.deliveredText) {
+                                session.client.discardPendingHubPromptEchoText(inFlightMessage.deliveredText)
+                            }
+                            inFlightMessage = null;
+                            session.client.sendSessionEvent({ type: 'message', message: `Process exited unexpectedly: ${detail}` });
+                            await this.respawnBackoff(getRespawnBackoffMs(), controller.signal);
+                            continue;
                         } else {
                             restoreInFlightMessage();
                             session.client.sendSessionEvent({ type: 'message', message: `Process exited unexpectedly: ${detail}` });
@@ -675,6 +763,7 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
 
     protected async cleanup(): Promise<void> {
         this.clearAbortHandlers(this.session.client.rpcHandlerManager);
+        this.session.requestRemoteRestart = null;
 
         if (this.handleSessionFound) {
             this.session.removeSessionFoundCallback(this.handleSessionFound);