diff --git a/apps/server/integration/orchestrationEngine.integration.test.ts b/apps/server/integration/orchestrationEngine.integration.test.ts index a577fc59ebc..abe00f16bb8 100644 --- a/apps/server/integration/orchestrationEngine.integration.test.ts +++ b/apps/server/integration/orchestrationEngine.integration.test.ts @@ -888,7 +888,7 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git ); it.live( - "appends checkpoint.revert.failed activity when revert is requested without an active session", + "appends checkpoint.revert.failed activity when revert is requested without a provider binding", () => withHarness((harness) => Effect.gen(function* () { @@ -917,7 +917,7 @@ it.live( assert.equal( String( (failureActivity?.payload as { readonly detail?: string } | undefined)?.detail, - ).includes("No active provider session"), + ).includes("no persisted provider binding exists"), true, ); }), diff --git a/apps/server/src/claudeHistoryWorker.ts b/apps/server/src/claudeHistoryWorker.ts new file mode 100644 index 00000000000..d00282bb77f --- /dev/null +++ b/apps/server/src/claudeHistoryWorker.ts @@ -0,0 +1,25 @@ +import { forkSession, getSessionMessages } from "@anthropic-ai/claude-agent-sdk"; +import * as Schema from "effect/Schema"; + +// A separate process gives SDK history helpers the provider's environment without +// mutating the server's environment. This entry is bundled alongside the server. +const [method, sessionId, rawOptions] = process.argv.slice(2); +const options = Schema.decodeSync( + Schema.fromJsonString( + Schema.Struct({ + dir: Schema.optionalKey(Schema.String), + includeSystemMessages: Schema.optionalKey(Schema.Boolean), + upToMessageId: Schema.optionalKey(Schema.String), + }), + ), +)(rawOptions ?? "{}"); +if (!sessionId) throw new Error("Claude history session id is required."); +const result = + method === "getSessionMessages" + ? await getSessionMessages(sessionId, options) + : method === "forkSession" + ? await forkSession(sessionId, options) + : (() => { + throw new Error("Unknown Claude history operation."); + })(); +process.stdout.write(JSON.stringify(result)); diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index b88f7f012d4..01430a20128 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -1916,27 +1916,45 @@ describe("CheckpointReactor", () => { }); }); - it("appends an error activity when revert is requested without an active session", async () => { - const harness = await createHarness({ hasSession: false }); - const createdAt = "2026-01-01T00:00:00.000Z"; - - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.checkpoint.revert", - commandId: CommandId.make("cmd-revert-no-session"), - threadId: ThreadId.make("thread-1"), - turnCount: 1, - createdAt, - }), - ); + it.each([false, true])( + "reverts without an active session using project cwd fallback: %s", + async (useProjectCwd) => { + const harness = await createHarness({ + hasSession: false, + ...(useProjectCwd ? { threadWorktreePath: null } : {}), + }); + const createdAt = "2026-01-01T00:00:00.000Z"; - const thread = await waitForThread(harness.readModel, (entry) => - entry.activities.some((activity) => activity.kind === "checkpoint.revert.failed"), - ); + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.diff.complete", + commandId: CommandId.make("cmd-diff-before-session-recovery"), + threadId: ThreadId.make("thread-1"), + turnId: asTurnId("turn-1"), + completedAt: createdAt, + checkpointRef: checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1), + status: "ready", + files: [], + checkpointTurnCount: 1, + createdAt, + }), + ); + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.checkpoint.revert", + commandId: CommandId.make("cmd-revert-no-session"), + threadId: ThreadId.make("thread-1"), + turnCount: 0, + createdAt, + }), + ); - expect(thread.activities.some((activity) => activity.kind === "checkpoint.revert.failed")).toBe( - true, - ); - expect(harness.provider.rollbackConversation).not.toHaveBeenCalled(); - }); + await waitForEvent(harness.engine, (event) => event.type === "thread.reverted"); + expect(harness.provider.rollbackConversation).toHaveBeenCalledWith({ + threadId: ThreadId.make("thread-1"), + numTurns: 1, + }); + expect(NodeFS.readFileSync(NodePath.join(harness.cwd, "README.md"), "utf8")).toBe("v1\n"); + }, + ); }); diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index d4d6b940980..fc1a3f74034 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -699,21 +699,17 @@ const make = Effect.gen(function* () { return; } - const sessionRuntime = yield* resolveSessionRuntimeForThread(event.payload.threadId); - if (Option.isNone(sessionRuntime)) { - yield* appendRevertFailureActivity({ - threadId: event.payload.threadId, - turnCount: event.payload.turnCount, - detail: "No active provider session with workspace cwd is bound to this thread.", - createdAt: now, - }).pipe(Effect.catch(() => Effect.void)); - return; - } - if (!(yield* checkpointStore.isGitRepository(sessionRuntime.value.cwd))) { + const checkpointCwd = yield* resolveCheckpointCwd({ + threadId: event.payload.threadId, + thread, + projects: yield* resolveThreadProjects(thread.projectId), + preferSessionRuntime: true, + }); + if (!checkpointCwd) { yield* appendRevertFailureActivity({ threadId: event.payload.threadId, turnCount: event.payload.turnCount, - detail: "Checkpoints are unavailable because this project is not a git repository.", + detail: "Checkpoint workspace is unavailable or is not a git repository.", createdAt: now, }).pipe(Effect.catch(() => Effect.void)); return; @@ -754,7 +750,7 @@ const make = Effect.gen(function* () { yield* providerService.assertConversationRollbackSupported(event.payload.threadId); const restored = yield* checkpointStore.restoreCheckpoint({ - cwd: sessionRuntime.value.cwd, + cwd: checkpointCwd, checkpointRef: targetCheckpointRef, fallbackToHead: event.payload.turnCount === 0, }); @@ -770,12 +766,12 @@ const make = Effect.gen(function* () { // Refresh the workspace entry index so the @-mention file picker // reflects the reverted filesystem state. - yield* workspaceEntries.refresh(sessionRuntime.value.cwd); + yield* workspaceEntries.refresh(checkpointCwd); const rolledBackTurns = Math.max(0, currentTurnCount - event.payload.turnCount); if (rolledBackTurns > 0) { yield* providerService.rollbackConversation({ - threadId: sessionRuntime.value.threadId, + threadId: event.payload.threadId, numTurns: rolledBackTurns, }); } @@ -789,7 +785,7 @@ const make = Effect.gen(function* () { if (staleCheckpointRefs.length > 0) { yield* checkpointStore.deleteCheckpointRefs({ - cwd: sessionRuntime.value.cwd, + cwd: checkpointCwd, checkpointRefs: staleCheckpointRefs, }); } diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index c4136156df1..38bdb7f1f2e 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -170,8 +170,11 @@ function makeHarness(config?: { readonly instanceId?: ProviderInstanceId; readonly scopedLimitNames?: ClaudeAdapterLiveOptions["scopedLimitNames"]; readonly environment?: ClaudeAdapterLiveOptions["environment"]; + readonly getSessionMessages?: ClaudeAdapterLiveOptions["getSessionMessages"]; + readonly forkSession?: ClaudeAdapterLiveOptions["forkSession"]; }) { const query = new FakeClaudeQuery(); + const queries = [query]; let createInput: | { readonly prompt: AsyncIterable; @@ -184,9 +187,12 @@ function makeHarness(config?: { ...(config?.instanceId ? { instanceId: config.instanceId } : {}), ...(config?.scopedLimitNames ? { scopedLimitNames: config.scopedLimitNames } : {}), modelCatalog: Effect.succeed(SYNTHETIC_CLAUDE_MODEL_CATALOG), + ...(config?.getSessionMessages ? { getSessionMessages: config.getSessionMessages } : {}), + ...(config?.forkSession ? { forkSession: config.forkSession } : {}), createQuery: (input) => { + if (createInput && config?.getSessionMessages) queries.push(new FakeClaudeQuery()); createInput = input; - return query; + return queries.at(-1)!; }, ...(config?.nativeEventLogger ? { @@ -218,6 +224,7 @@ function makeHarness(config?: { Layer.provideMerge(NodeServices.layer), ), query, + queries, getLastCreateQueryInput: () => createInput, }; } @@ -6256,87 +6263,226 @@ describe("ClaudeAdapterLive", () => { ); }); - it.effect( - "supports rollbackThread by trimming in-memory turns and preserving earlier turns", - () => { - const harness = makeHarness(); - return Effect.gen(function* () { - const adapter = yield* ClaudeAdapter; - - const session = yield* adapter.startSession({ - threadId: THREAD_ID, - provider: ProviderDriverKind.make("claudeAgent"), - runtimeMode: "full-access", - }); + it.effect("rewinds a steered Claude turn after recovery and preserves fork boundaries", () => { + const forkCalls: Array>> = []; + let firstTurnId = ""; + let secondTurnId = ""; + let missingBoundary = false; + let legacyHistory = false; + const harness = makeHarness({ + forkSession: async (...args) => { + forkCalls.push(args); + return { sessionId: "550e8400-e29b-41d4-a716-446655440020" }; + }, + getSessionMessages: async (sessionId) => { + const history: Awaited< + ReturnType> + > = [ + { + type: "user", + uuid: firstTurnId, + session_id: "550e8400-e29b-41d4-a716-446655440010", + parent_tool_use_id: null, + parent_agent_id: null, + message: { content: "first" }, + }, + { + type: "assistant", + uuid: "assistant-1", + session_id: "550e8400-e29b-41d4-a716-446655440010", + parent_tool_use_id: null, + parent_agent_id: null, + message: { content: [] }, + }, + { + type: "user", + uuid: "tool-result-1", + session_id: "550e8400-e29b-41d4-a716-446655440010", + parent_tool_use_id: null, + parent_agent_id: null, + message: { content: [{ type: "tool_result" }] }, + }, + { + type: "assistant", + uuid: "assistant-1-final", + session_id: "550e8400-e29b-41d4-a716-446655440010", + parent_tool_use_id: null, + parent_agent_id: null, + message: { content: [] }, + }, + { + type: "user", + uuid: secondTurnId, + session_id: "550e8400-e29b-41d4-a716-446655440010", + parent_tool_use_id: null, + parent_agent_id: null, + message: { content: "second" }, + }, + { + type: "assistant", + uuid: "assistant-2", + session_id: "550e8400-e29b-41d4-a716-446655440010", + parent_tool_use_id: null, + parent_agent_id: null, + message: { content: [] }, + }, + { + type: "user", + uuid: "steer", + session_id: sessionId, + parent_tool_use_id: null, + parent_agent_id: null, + message: { content: "steer the second turn" }, + }, + { + type: "assistant", + uuid: "assistant-steer", + session_id: sessionId, + parent_tool_use_id: null, + parent_agent_id: null, + message: { content: [] }, + }, + ]; + return sessionId.endsWith("0020") + ? history.slice(0, 4).map((message) => ({ ...message, uuid: `fork-${message.uuid}` })) + : legacyHistory + ? history.slice(0, 6) + : missingBoundary + ? history.filter((message) => message.uuid !== secondTurnId) + : history; + }, + }); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; - const firstTurn = yield* adapter.sendTurn({ - threadId: session.threadId, - input: "first", - attachments: [], - }); + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); - const firstCompletedFiber = yield* Stream.filter( - adapter.streamEvents, - (event) => event.type === "turn.completed", - ).pipe(Stream.runHead, Effect.forkChild); + const firstTurn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "first", + attachments: [], + }); + firstTurnId = firstTurn.turnId; - harness.query.emit({ - type: "result", - subtype: "success", - is_error: false, - errors: [], - session_id: "sdk-session-rollback", - uuid: "result-first", - } as unknown as SDKMessage); + const firstCompletedFiber = yield* Stream.filter( + adapter.streamEvents, + (event) => event.type === "turn.completed", + ).pipe(Stream.runHead, Effect.forkChild); - const firstCompleted = yield* Fiber.join(firstCompletedFiber); - assert.equal(firstCompleted._tag, "Some"); - if (firstCompleted._tag === "Some" && firstCompleted.value.type === "turn.completed") { - assert.equal(String(firstCompleted.value.turnId), String(firstTurn.turnId)); - } + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "550e8400-e29b-41d4-a716-446655440010", + uuid: "result-first", + } as unknown as SDKMessage); - const secondTurn = yield* adapter.sendTurn({ - threadId: session.threadId, - input: "second", - attachments: [], - }); + const firstCompleted = yield* Fiber.join(firstCompletedFiber); + assert.equal(firstCompleted._tag, "Some"); + if (firstCompleted._tag === "Some" && firstCompleted.value.type === "turn.completed") { + assert.equal(String(firstCompleted.value.turnId), String(firstTurn.turnId)); + } - const secondCompletedFiber = yield* Stream.filter( - adapter.streamEvents, - (event) => event.type === "turn.completed", - ).pipe(Stream.runHead, Effect.forkChild); + const secondTurn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "second", + attachments: [], + }); + secondTurnId = secondTurn.turnId; + const steer = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "steer the second turn", + attachments: [], + }); + assert.equal(steer.turnId, secondTurn.turnId); - harness.query.emit({ - type: "result", - subtype: "success", - is_error: false, - errors: [], - session_id: "sdk-session-rollback", - uuid: "result-second", - } as unknown as SDKMessage); + const secondCompletedFiber = yield* Stream.filter( + adapter.streamEvents, + (event) => event.type === "turn.completed", + ).pipe(Stream.runHead, Effect.forkChild); - const secondCompleted = yield* Fiber.join(secondCompletedFiber); - assert.equal(secondCompleted._tag, "Some"); - if (secondCompleted._tag === "Some" && secondCompleted.value.type === "turn.completed") { - assert.equal(String(secondCompleted.value.turnId), String(secondTurn.turnId)); - } + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "550e8400-e29b-41d4-a716-446655440010", + uuid: "result-second", + } as unknown as SDKMessage); - const threadBeforeRollback = yield* adapter.readThread(session.threadId); - assert.equal(threadBeforeRollback.turns.length, 2); + const secondCompleted = yield* Fiber.join(secondCompletedFiber); + assert.equal(secondCompleted._tag, "Some"); + if (secondCompleted._tag === "Some" && secondCompleted.value.type === "turn.completed") { + assert.equal(String(secondCompleted.value.turnId), String(secondTurn.turnId)); + } - const rolledBack = yield* adapter.rollbackThread(session.threadId, 1); - assert.equal(rolledBack.turns.length, 1); - assert.equal(rolledBack.turns[0]?.id, firstTurn.turnId); + const threadBeforeRollback = yield* adapter.readThread(session.threadId); + assert.equal(threadBeforeRollback.turns.length, 2); + const cursor = (yield* adapter.listSessions())[0]?.resumeCursor; + yield* adapter.stopSession(session.threadId); + legacyHistory = true; + yield* adapter.startSession({ + threadId: session.threadId, + runtimeMode: "full-access", + resumeCursor: { + threadId: session.threadId, + resume: "550e8400-e29b-41d4-a716-446655440010", + turnCount: 1, + }, + }); + const legacyOptions = harness.getLastCreateQueryInput(); + const ambiguousLegacy = yield* adapter.rollbackThread(session.threadId, 1).pipe(Effect.flip); + assert.match(ambiguousLegacy.message, /exact Claude turn boundary is unavailable/); + assert.equal(forkCalls.length, 0); + assert.equal(harness.getLastCreateQueryInput(), legacyOptions); + assert.equal((yield* adapter.listSessions()).length, 1); + yield* adapter.stopSession(session.threadId); + legacyHistory = false; + yield* adapter.startSession({ + threadId: session.threadId, + runtimeMode: "full-access", + resumeCursor: cursor, + }); + missingBoundary = true; + const unavailable = yield* adapter.rollbackThread(session.threadId, 1).pipe(Effect.flip); + assert.match(unavailable.message, /exact Claude turn boundary is unavailable/); + assert.equal(forkCalls.length, 0); + missingBoundary = false; + + const recoveredQuery = harness.queries.at(-1)!; + assert.equal(recoveredQuery.closeCalls, 0); + yield* adapter.rollbackThread(session.threadId, 1); + assert.equal(recoveredQuery.closeCalls, 1); + const forkOptions = harness.getLastCreateQueryInput()?.options; + assert.deepEqual(forkCalls, [ + ["550e8400-e29b-41d4-a716-446655440010", { upToMessageId: "assistant-1-final" }], + ]); + assert.equal(forkOptions?.resume, "550e8400-e29b-41d4-a716-446655440020"); + assert.equal(forkOptions?.resumeSessionAt, undefined); + assert.equal(forkOptions?.forkSession, undefined); + assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, { + threadId: session.threadId, + resume: "550e8400-e29b-41d4-a716-446655440020", + turnCount: 1, + turnStartMessageIds: [`fork-${firstTurnId}`], + }); - const threadAfterRollback = yield* adapter.readThread(session.threadId); - assert.equal(threadAfterRollback.turns.length, 1); - assert.equal(threadAfterRollback.turns[0]?.id, firstTurn.turnId); - }).pipe( - Effect.provideService(Random.Random, makeDeterministicRandomService()), - Effect.provide(harness.layer), - ); - }, - ); + yield* adapter.rollbackThread(session.threadId, 2); + const resetOptions = harness.getLastCreateQueryInput()?.options; + assert.equal(resetOptions?.resume, undefined); + assert.equal(resetOptions?.resumeSessionAt, undefined); + assert.equal(resetOptions?.forkSession, undefined); + assert.ok(resetOptions?.sessionId); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); it.effect("updates model on sendTurn when model override is provided", () => { const harness = makeHarness(); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 981e183414f..fc52fe38dc6 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -9,6 +9,8 @@ import { type CanUseTool, query, + getSessionMessages, + forkSession, type Options as ClaudeQueryOptions, type PermissionMode, type PermissionResult, @@ -78,6 +80,7 @@ import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; @@ -108,9 +111,26 @@ import { type ProviderAdapterError, } from "../Errors.ts"; import { type ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; +import { spawnAndCollect } from "../providerSnapshot.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); +const encodeHistoryArgs = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); +const decodeHistoryFork = Schema.decodeSync( + Schema.fromJsonString(Schema.Struct({ sessionId: Schema.String })), +); +const decodeSessionMessages = Schema.decodeSync( + Schema.fromJsonString( + Schema.Array( + Schema.Struct({ + type: Schema.Literals(["user", "assistant", "system"]), + uuid: Schema.String, + parent_tool_use_id: Schema.NullOr(Schema.String), + message: Schema.Unknown, + }), + ), + ), +); const PROVIDER = ProviderDriverKind.make("claudeAgent"); type ClaudeTextStreamKind = Extract; @@ -139,6 +159,7 @@ interface ClaudeResumeState { readonly resume?: string; readonly resumeSessionAt?: string; readonly turnCount?: number; + readonly turnStartMessageIds?: ReadonlyArray; } interface ClaudeTurnState { @@ -290,6 +311,8 @@ function rememberPendingTaskModel( interface ClaudeSessionContext { session: ProviderSession; + startInput: Parameters[0]; + readonly turnStartMessageIds: Array; readonly promptQueue: Queue.Queue; readonly query: ClaudeQueryRuntime; streamFiber: Fiber.Fiber | undefined; @@ -350,6 +373,8 @@ export interface ClaudeAdapterLiveOptions { readonly prompt: AsyncIterable; readonly options: ClaudeQueryOptions; }) => ClaudeQueryRuntime; + readonly getSessionMessages?: typeof getSessionMessages; + readonly forkSession?: typeof forkSession; readonly nativeEventLogPath?: string; readonly nativeEventLogger?: EventNdjsonLogger; readonly modelCatalog?: Effect.Effect; @@ -854,6 +879,7 @@ function readClaudeResumeState(resumeCursor: unknown): ClaudeResumeState | undef sessionId?: unknown; resumeSessionAt?: unknown; turnCount?: unknown; + turnStartMessageIds?: unknown; }; const threadIdCandidate = typeof cursor.threadId === "string" ? cursor.threadId : undefined; @@ -871,11 +897,17 @@ function readClaudeResumeState(resumeCursor: unknown): ClaudeResumeState | undef const resumeSessionAt = typeof cursor.resumeSessionAt === "string" ? cursor.resumeSessionAt : undefined; const turnCountValue = typeof cursor.turnCount === "number" ? cursor.turnCount : undefined; + const turnStartMessageIds = + Array.isArray(cursor.turnStartMessageIds) && + cursor.turnStartMessageIds.every((id: unknown) => id === null || typeof id === "string") + ? (cursor.turnStartMessageIds as Array) + : undefined; return { ...(threadId ? { threadId } : {}), ...(resume ? { resume } : {}), ...(resumeSessionAt ? { resumeSessionAt } : {}), + ...(turnStartMessageIds ? { turnStartMessageIds } : {}), ...(turnCountValue !== undefined && Number.isInteger(turnCountValue) && turnCountValue >= 0 ? { turnCount: turnCountValue } : {}), @@ -1927,6 +1959,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const path = yield* Path.Path; const serverConfig = yield* ServerConfig; const crypto = yield* Crypto.Crypto; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const claudeEnvironment = yield* makeClaudeEnvironment(claudeSettings, options?.environment).pipe( Effect.provideService(Path.Path, path), ); @@ -2043,7 +2076,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( threadId, ...(context.resumeSessionId ? { resume: context.resumeSessionId } : {}), ...(context.lastAssistantUuid ? { resumeSessionAt: context.lastAssistantUuid } : {}), - turnCount: context.turns.length, + turnCount: context.turnStartMessageIds.length, + turnStartMessageIds: [...context.turnStartMessageIds], }; context.session = { @@ -3155,6 +3189,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( if (!context.turnState) { const turnId = TurnId.make(yield* randomUUIDv4); const startedAt = yield* nowIso; + context.turnStartMessageIds.push(message.uuid); context.turnState = { turnId, startedAt, @@ -3177,6 +3212,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( activeTurnId: turnId, updatedAt: startedAt, }; + yield* updateResumeCursor(context); const turnStartedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "turn.started", @@ -4779,6 +4815,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(sessionId ? { resume: sessionId } : {}), ...(resumeState?.resumeSessionAt ? { resumeSessionAt: resumeState.resumeSessionAt } : {}), turnCount: resumeState?.turnCount ?? 0, + ...(resumeState?.turnStartMessageIds + ? { turnStartMessageIds: resumeState.turnStartMessageIds } + : {}), }, createdAt: startedAt, updatedAt: startedAt, @@ -4786,6 +4825,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const context: ClaudeSessionContext = { session, + startInput: input, + turnStartMessageIds: resumeState?.turnStartMessageIds + ? [...resumeState.turnStartMessageIds] + : Array.from({ length: resumeState?.turnCount ?? 0 }, () => null), promptQueue, query: queryRuntime, streamFiber: undefined, @@ -4899,6 +4942,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const modelSelection = selectedModel ? { ...selectedModel, model: resolveClaudeModelSlug(modelCatalog, selectedModel.model) } : undefined; + if (modelSelection) { + context.startInput = { ...context.startInput, modelSelection }; + } // A sendTurn while a real turn is running is a steer: the message is // queued into the live SDK agent loop and the work continues as the same @@ -5015,9 +5061,14 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ), }); + if (steeringTurnState === null) context.turnStartMessageIds.push(turnId); + yield* updateResumeCursor(context); yield* Queue.offer(context.promptQueue, { type: "message", - message, + message: + steeringTurnState === null + ? { ...message, uuid: turnId as NonNullable } + : message, }).pipe(Effect.mapError((cause) => toRequestError(input.threadId, "turn/start", cause))); return { @@ -5049,10 +5100,190 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const rollbackThread: ClaudeAdapterShape["rollbackThread"] = Effect.fn("rollbackThread")( function* (threadId, numTurns) { const context = yield* requireSession(threadId); - const nextLength = Math.max(0, context.turns.length - numTurns); - context.turns.splice(nextLength); - yield* updateResumeCursor(context); - return yield* snapshotThread(context); + if (!Number.isInteger(numTurns) || numTurns < 1) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThread", + issue: "numTurns must be an integer >= 1.", + }); + } + if ( + context.turnStartMessageIds.length > 0 && + context.turnStartMessageIds.every((id) => id !== null) && + numTurns >= context.turnStartMessageIds.length + ) { + yield* stopSessionInternal(context, { emitExitEvent: false }); + yield* startSession({ + ...context.startInput, + runtimeMode: context.session.runtimeMode, + resumeCursor: undefined, + }); + return yield* snapshotThread(yield* requireSession(threadId)); + } + const sessionId = context.resumeSessionId; + if (!sessionId) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback", + detail: "Claude session id is unavailable.", + }); + } + const historyWorkerPath = yield* path + .fromFileUrl( + new URL( + import.meta.url.endsWith(".ts") + ? "../../claudeHistoryWorker.ts" + : "./claudeHistoryWorker.mjs", + import.meta.url, + ), + ) + .pipe(Effect.mapError((cause) => toRequestError(threadId, "thread/rollback", cause))); + const runScopedHistoryCommand = async ( + method: "getSessionMessages" | "forkSession", + args: object, + historySessionId = sessionId, + ) => { + // SDK history helpers read process.env. Isolate the provider's home instead + // of changing the server's environment while other providers are running. + const result = await Effect.runPromise( + spawnAndCollect( + process.execPath, + ChildProcess.make( + process.execPath, + [historyWorkerPath, method, historySessionId, encodeHistoryArgs(args)], + { env: { ...claudeEnvironment, ELECTRON_RUN_AS_NODE: "1" } }, + ), + ).pipe( + Effect.timeout("30 seconds"), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ), + ); + if (result.code !== 0) throw new Error(result.stderr || "Claude history command failed."); + return result.stdout; + }; + const readHistory = (historySessionId: string) => + Effect.tryPromise({ + try: async () => { + const readOptions = { + ...(context.session.cwd ? { dir: context.session.cwd } : {}), + includeSystemMessages: true, + }; + if (options?.getSessionMessages) + return options.getSessionMessages(historySessionId, readOptions); + if (claudeEnvironment.CLAUDE_CONFIG_DIR === process.env.CLAUDE_CONFIG_DIR) { + return getSessionMessages(historySessionId, readOptions); + } + return decodeSessionMessages( + await runScopedHistoryCommand("getSessionMessages", readOptions, historySessionId), + ); + }, + catch: (cause) => toRequestError(threadId, "thread/rollback", cause), + }); + const messages = yield* readHistory(sessionId); + // Tool results are user-role messages too. Only human prompts begin a turn. + const turnStarts = messages.flatMap((message, index) => { + if (message.type !== "user" || message.parent_tool_use_id !== null) return []; + const body = message.message; + if (typeof body !== "object" || body === null || !("content" in body)) return []; + const content = body.content; + return typeof content === "string" || + (Array.isArray(content) && + content.some( + (part: unknown) => + typeof part === "object" && + part !== null && + "type" in part && + part.type !== "tool_result", + )) + ? [index] + : []; + }); + if (messages.length === 0) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback", + detail: "Claude session history is unavailable.", + }); + } + const boundaries = [...context.turnStartMessageIds]; + // Older cursors did not record native boundaries. Infer them only when + // their T3 turn count agrees; steers must never be treated as extra turns. + if ( + boundaries.every((id): boolean => id === null) && + boundaries.length === turnStarts.length + ) { + boundaries.splice( + 0, + boundaries.length, + ...turnStarts.map((index) => messages[index]!.uuid), + ); + } + const retainedCount = Math.max(0, boundaries.length - numTurns); + const firstRemovedId = boundaries[retainedCount]; + const firstRemoved = messages.findIndex((message) => message.uuid === firstRemovedId); + if ( + boundaries.length === 0 || + boundaries.some((id) => id === null) || + (retainedCount > 0 && firstRemoved < 1) + ) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback", + detail: + "The exact Claude turn boundary is unavailable, possibly after compaction or recovery of older history. Start a new thread instead.", + }); + } + const rollbackAt = retainedCount > 0 ? messages[firstRemoved - 1]?.uuid : undefined; + const retainedTurns = context.turns.slice(0, Math.max(0, context.turns.length - numTurns)); + const fork = rollbackAt + ? yield* Effect.tryPromise({ + try: async () => { + const forkOptions = { + ...(context.session.cwd ? { dir: context.session.cwd } : {}), + upToMessageId: rollbackAt, + }; + if (options?.forkSession) return options.forkSession(sessionId, forkOptions); + if (claudeEnvironment.CLAUDE_CONFIG_DIR === process.env.CLAUDE_CONFIG_DIR) { + return forkSession(sessionId, forkOptions); + } + return decodeHistoryFork(await runScopedHistoryCommand("forkSession", forkOptions)); + }, + catch: (cause) => toRequestError(threadId, "thread/rollback", cause), + }) + : undefined; + const retainedBoundaries = boundaries.slice(0, retainedCount); + if (fork) { + const forkMessages = yield* readHistory(fork.sessionId); + if (forkMessages.length !== firstRemoved) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback", + detail: "Claude fork history did not preserve the retained turn boundaries.", + }); + } + // Native forks replace every UUID while preserving transcript order. + for (let index = 0; index < retainedBoundaries.length; index++) { + const messageIndex = messages.findIndex( + (message) => message.uuid === retainedBoundaries[index], + ); + retainedBoundaries[index] = forkMessages[messageIndex]?.uuid ?? null; + } + } + yield* stopSessionInternal(context, { emitExitEvent: false }); + yield* startSession({ + ...context.startInput, + runtimeMode: context.session.runtimeMode, + resumeCursor: fork + ? { + resume: fork.sessionId, + turnCount: retainedCount, + turnStartMessageIds: retainedBoundaries, + } + : undefined, + }); + const restarted = yield* requireSession(threadId); + restarted.turns.push(...retainedTurns); + return yield* snapshotThread(restarted); }, ); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 3385137a2da..c6257221c73 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -18,10 +18,101 @@ import { isRecoverableThreadResumeError, makeMemoryConsolidationNotificationFilter, openCodexThread, + readCodexThread, + rollbackCodexThread, toMcpElicitationResponse, } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); +describe("Codex thread history", () => { + for (const numTurns of [1, 2, 3, 5]) { + it.effect(`reverts ${numTurns} paginated turns at the durable boundary`, () => + Effect.gen(function* () { + let retained = ["turn-1", "turn-2", "turn-3"]; + const client: Parameters[0] = { + request: () => Effect.die("Legacy history API must not be used for paginated threads"), + raw: { + request: (method, params) => + Effect.sync(() => { + if (method === "thread/read") return { thread: { historyMode: "paginated" } }; + if (method === "thread/turns/list") { + const { cursor } = params as { cursor: string | null }; + const start = cursor === null ? 0 : Number(cursor); + const ids = retained.slice(start, start + 2); + return { + data: ids.map((id) => ({ id, items: [], status: "completed" })), + nextCursor: start + 2 < retained.length ? String(start + 2) : null, + }; + } + NodeAssert.equal(method, "thread/revert"); + const { beforeTurnId } = params as { beforeTurnId: string }; + retained = retained.slice(0, retained.indexOf(beforeTurnId)); + return { thread: { id: "thread-1", turns: [] } }; + }), + }, + }; + const result = yield* rollbackCodexThread(client, "thread-1", numTurns); + const expected = ["turn-1", "turn-2", "turn-3"].slice(0, Math.max(0, 3 - numTurns)); + NodeAssert.deepEqual( + result.turns.map((turn) => turn.id), + expected, + ); + NodeAssert.deepEqual( + (yield* readCodexThread(client, "thread-1")).turns.map((turn) => turn.id), + expected, + ); + }), + ); + } + + for (const cursors of [ + ["next", "next"], + ["first", "second", "first"], + ]) { + it.effect(`rejects a pagination cursor cycle: ${cursors.join(", ")}`, () => + Effect.gen(function* () { + let pageCount = 0; + const client: Parameters[0] = { + request: () => Effect.die("Unexpected legacy request"), + raw: { + request: (method) => + Effect.sync(() => { + if (method === "thread/read") return { thread: { historyMode: "paginated" } }; + NodeAssert.ok(pageCount < cursors.length, "Repeated cursor was requested"); + return { data: [], nextCursor: cursors[pageCount++] }; + }), + }, + }; + const error = yield* Effect.flip(readCodexThread(client, "thread-1")); + NodeAssert.ok(isCodexAppServerRequestError(error)); + NodeAssert.equal(pageCount, cursors.length); + }), + ); + } + + it.effect("keeps the count-based rollback API for older threads", () => + Effect.gen(function* () { + const client: Parameters[0] = { + raw: { request: () => Effect.succeed({ thread: {} }) }, + request: ( + method: M, + params: CodexRpc.ClientRequestParamsByMethod[M], + ) => { + NodeAssert.equal(method, "thread/rollback"); + NodeAssert.deepEqual(params, { threadId: "legacy-thread", numTurns: 2 }); + return Effect.succeed({ + thread: { id: "legacy-thread", turns: [] }, + } as unknown as CodexRpc.ClientRequestResponsesByMethod[M]); + }, + }; + NodeAssert.deepEqual(yield* rollbackCodexThread(client, "legacy-thread", 2), { + threadId: "legacy-thread", + turns: [], + }); + }), + ); +}); + describe("CodexSessionRuntimeIdentifierGenerationError", () => { it("retains identifier purpose and the random source failure", () => { const cause = new Error("random source unavailable"); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index bd9a6b6f34a..a04db140591 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1192,6 +1192,97 @@ function parseThreadSnapshot( }; } +const CodexThreadHistoryMetadata = Schema.Struct({ + thread: Schema.Struct({ + historyMode: Schema.optionalKey(Schema.Literals(["legacy", "paginated"])), + }), +}); +const CodexTurnsPage = Schema.Struct({ + data: Schema.Array(EffectCodexSchema.V2ThreadReadResponse__Turn), + nextCursor: Schema.NullOr(Schema.String), +}); +const decodeCodexHistoryMetadata = Schema.decodeUnknownEffect(CodexThreadHistoryMetadata); +const decodeCodexTurnsPage = Schema.decodeUnknownEffect(CodexTurnsPage); +type CodexHistoryClient = { + readonly raw: Pick; + readonly request: CodexClient.CodexAppServerClient["Service"]["request"]; +}; + +const readCodexHistoryMode = Effect.fn("readCodexHistoryMode")(function* ( + client: CodexHistoryClient, + threadId: string, +) { + const response = yield* client.raw.request("thread/read", { threadId, includeTurns: false }); + const metadata = yield* decodeCodexHistoryMetadata(response).pipe( + Effect.mapError((error) => + CodexErrors.CodexAppServerRequestError.invalidPayload("thread/read", "decode-payload", error), + ), + ); + return metadata.thread.historyMode; +}); + +export const readCodexThread = Effect.fn("readCodexThread")(function* ( + client: CodexHistoryClient, + threadId: string, +): Effect.fn.Return { + if ((yield* readCodexHistoryMode(client, threadId)) !== "paginated") { + return parseThreadSnapshot( + yield* client.request("thread/read", { threadId, includeTurns: true }), + ); + } + const turns: Array = []; + const requestedCursors = new Set(); + let cursor: string | null = null; + do { + if (requestedCursors.has(cursor)) { + return yield* CodexErrors.CodexAppServerRequestError.internalError( + "Thread history pagination repeated a cursor.", + undefined, + { method: "thread/turns/list", operation: "decode-payload" }, + ); + } + requestedCursors.add(cursor); + const response: unknown = yield* client.raw.request("thread/turns/list", { + threadId, + cursor, + limit: 100, + sortDirection: "asc", + itemsView: "full", + }); + const page = yield* decodeCodexTurnsPage(response).pipe( + Effect.mapError((error) => + CodexErrors.CodexAppServerRequestError.invalidPayload( + "thread/turns/list", + "decode-payload", + error, + ), + ), + ); + turns.push(...page.data.map((turn) => ({ id: TurnId.make(turn.id), items: turn.items }))); + cursor = page.nextCursor; + } while (cursor !== null); + return { threadId, turns }; +}); + +export const rollbackCodexThread = Effect.fn("rollbackCodexThread")(function* ( + client: CodexHistoryClient, + threadId: string, + numTurns: number, +): Effect.fn.Return { + if ((yield* readCodexHistoryMode(client, threadId)) !== "paginated") { + return parseThreadSnapshot(yield* client.request("thread/rollback", { threadId, numTurns })); + } + // Paginated threads replace history at a turn boundary instead of supporting + // the legacy count-based rollback endpoint. + const snapshot = yield* readCodexThread(client, threadId); + const retainedCount = Math.max(0, snapshot.turns.length - numTurns); + const firstRemoved = snapshot.turns[retainedCount]; + if (firstRemoved) { + yield* client.raw.request("thread/revert", { threadId, beforeTurnId: firstRemoved.id }); + } + return { threadId, turns: snapshot.turns.slice(0, retainedCount) }; +}); + export const makeCodexSessionRuntime = ( options: CodexSessionRuntimeOptions, ): Effect.Effect< @@ -2435,24 +2526,17 @@ export const makeCodexSessionRuntime = ( }), readThread: Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; - const response = yield* client.request("thread/read", { - threadId: providerThreadId, - includeTurns: true, - }); - return parseThreadSnapshot(response); + return yield* readCodexThread(client, providerThreadId); }), rollbackThread: (numTurns) => Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; - const response = yield* client.request("thread/rollback", { - threadId: providerThreadId, - numTurns, - }); + const snapshot = yield* rollbackCodexThread(client, providerThreadId, numTurns); yield* updateSession(sessionRef, { status: "ready", activeTurnId: undefined, }); - return parseThreadSnapshot(response); + return snapshot; }), uploadFeedback: (reason) => Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/CursorAdapter.test.ts b/apps/server/src/provider/Layers/CursorAdapter.test.ts index 0c192ce7e11..bdc818994a9 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.test.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.test.ts @@ -162,6 +162,28 @@ const cursorAdapterTestLayer = it.layer( ); cursorAdapterTestLayer("CursorAdapterLive", (it) => { + it.effect("rejects rollback without discarding the provider conversation", () => + Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const settings = yield* ServerSettingsService; + const threadId = ThreadId.make("cursor-unsupported-rollback"); + const wrapperPath = yield* Effect.promise(() => makeMockAgentWrapper()); + yield* settings.updateSettings({ providers: { cursor: { binaryPath: wrapperPath } } }); + yield* adapter.startSession({ + threadId, + cwd: process.cwd(), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ threadId, input: "Remember this turn", attachments: [] }); + const originalTurns = [...(yield* adapter.readThread(threadId)).turns]; + assert.isFalse(adapter.capabilities.supportsConversationRollback); + const error = yield* adapter.rollbackThread(threadId, 1).pipe(Effect.flip); + assert.equal(error._tag, "ProviderAdapterRequestError"); + assert.deepStrictEqual((yield* adapter.readThread(threadId)).turns, originalTurns); + yield* adapter.stopSession(threadId); + }), + ); + it.effect("rejects a Cursor transport error returned as a successful assistant answer", () => Effect.gen(function* () { const adapter = yield* CursorAdapter; diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index b19d46112fd..925d585e583 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -1187,7 +1187,7 @@ export function makeCursorAdapter( const rollbackThread: CursorAdapterShape["rollbackThread"] = (threadId, numTurns) => Effect.gen(function* () { - const ctx = yield* requireSession(threadId); + yield* requireSession(threadId); if (!Number.isInteger(numTurns) || numTurns < 1) { return yield* new ProviderAdapterValidationError({ provider: PROVIDER, @@ -1195,9 +1195,11 @@ export function makeCursorAdapter( issue: "numTurns must be an integer >= 1.", }); } - const nextLength = Math.max(0, ctx.turns.length - numTurns); - ctx.turns.splice(nextLength); - return { threadId, turns: ctx.turns }; + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback", + detail: "Cursor ACP sessions do not support provider-side rollback.", + }); }); const stopSession: CursorAdapterShape["stopSession"] = (threadId) => @@ -1235,7 +1237,7 @@ export function makeCursorAdapter( return { provider: PROVIDER, - capabilities: { sessionModelSwitch: "in-session" }, + capabilities: { sessionModelSwitch: "in-session", supportsConversationRollback: false }, compaction: { type: "slash-command", command: "/compress" }, startSession, sendTurn, diff --git a/apps/server/src/provider/Layers/CursorProvider.test.ts b/apps/server/src/provider/Layers/CursorProvider.test.ts index adda9f44d46..5b71c64244b 100644 --- a/apps/server/src/provider/Layers/CursorProvider.test.ts +++ b/apps/server/src/provider/Layers/CursorProvider.test.ts @@ -492,6 +492,7 @@ describe("buildCursorProviderSnapshot", () => { status: "warning", message: "Cursor ACP model discovery timed out after 15000ms.", models: [], + supportsConversationRollback: false, }); }); diff --git a/apps/server/src/provider/Layers/CursorProvider.ts b/apps/server/src/provider/Layers/CursorProvider.ts index cf4f00ac967..7a18eef55e6 100644 --- a/apps/server/src/provider/Layers/CursorProvider.ts +++ b/apps/server/src/provider/Layers/CursorProvider.ts @@ -55,6 +55,7 @@ const decodeCursorListAvailableModelsResponse = Schema.decodeUnknownEffect( ); const CURSOR_PRESENTATION = { displayName: "Cursor", + supportsConversationRollback: false, badgeLabel: "Early Access", showInteractionModeToggle: true, } as const; diff --git a/apps/server/src/provider/Layers/GrokAdapter.test.ts b/apps/server/src/provider/Layers/GrokAdapter.test.ts index 41fa6ed0f60..9efb28d8062 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.test.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.test.ts @@ -212,6 +212,26 @@ it("requires a settlement to match the live Grok turn", () => { }); it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { + it.effect("rejects rollback without discarding the provider conversation", () => + Effect.gen(function* () { + const threadId = ThreadId.make("grok-unsupported-rollback"); + const wrapperPath = yield* Effect.promise(() => makeMockGrokWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + yield* adapter.startSession({ + threadId, + cwd: process.cwd(), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ threadId, input: "Remember this turn" }); + const originalTurns = [...(yield* adapter.readThread(threadId)).turns]; + assert.isFalse(adapter.capabilities.supportsConversationRollback); + const error = yield* adapter.rollbackThread(threadId, 1).pipe(Effect.flip); + assert.equal(error._tag, "ProviderAdapterRequestError"); + assert.deepStrictEqual((yield* adapter.readThread(threadId)).turns, originalTurns); + yield* adapter.stopSession(threadId); + }), + ); + it.effect("sends runtime context with the current model without changing saved prompts", () => Effect.gen(function* () { const threadId = ThreadId.make("grok-runtime-context"); diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index a2f78a0d72d..395d7e546f9 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -2134,7 +2134,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte return { provider: PROVIDER, - capabilities: { sessionModelSwitch: "in-session" }, + capabilities: { sessionModelSwitch: "in-session", supportsConversationRollback: false }, compaction: { type: "slash-command", command: "/compact" }, startSession, sendTurn, diff --git a/apps/server/src/provider/Layers/GrokProvider.test.ts b/apps/server/src/provider/Layers/GrokProvider.test.ts index c751919b189..d0e5010bc2b 100644 --- a/apps/server/src/provider/Layers/GrokProvider.test.ts +++ b/apps/server/src/provider/Layers/GrokProvider.test.ts @@ -282,6 +282,7 @@ describe("buildInitialGrokProviderSnapshot", () => { expect(snapshot.version).toBeNull(); expect(snapshot.message).toContain("Checking Grok"); expect(snapshot.requiresNewThreadForModelChange).toBeUndefined(); + expect(snapshot.supportsConversationRollback).toBe(false); }), ); }); diff --git a/apps/server/src/provider/Layers/GrokProvider.ts b/apps/server/src/provider/Layers/GrokProvider.ts index 493e46d4435..18a77334b9a 100644 --- a/apps/server/src/provider/Layers/GrokProvider.ts +++ b/apps/server/src/provider/Layers/GrokProvider.ts @@ -44,6 +44,7 @@ import { discoverGrokSkills } from "../Drivers/GrokSkills.ts"; const GROK_PRESENTATION = { displayName: "Grok", + supportsConversationRollback: false, badgeLabel: "Early Access", showInteractionModeToggle: false, } as const; diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 17f49355a90..1d86e6d943a 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -1654,6 +1654,9 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("routes provider operations and rollback conversation", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; + const modelSelection = createModelSelection(codexInstanceId, "gpt-5.6-sol", [ + { id: "reasoningEffort", value: "high" }, + ]); const session = yield* provider.startSession(asThreadId("thread-1"), { provider: ProviderDriverKind.make("codex"), @@ -1671,6 +1674,7 @@ routing.layer("ProviderServiceLive routing", (it) => { threadId: session.threadId, input: "hello", attachments: [], + modelSelection, }); assert.equal(routing.codex.sendTurn.mock.calls.length, 1); @@ -1708,6 +1712,21 @@ routing.layer("ProviderServiceLive routing", (it) => { numTurns: 0, }); + const rewindCursor = { threadId: "rewound-provider-thread" }; + routing.codex.updateSession(session.threadId, (session) => ({ + ...session, + resumeCursor: rewindCursor, + })); + yield* provider.rollbackConversation({ threadId: session.threadId, numTurns: 1 }); + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const rewoundBinding = yield* directory.getBinding(session.threadId); + assert(Option.isSome(rewoundBinding)); + assert.deepEqual(rewoundBinding.value.resumeCursor, rewindCursor); + assert.deepEqual( + (rewoundBinding.value.runtimePayload as { modelSelection?: unknown }).modelSelection, + modelSelection, + ); + yield* provider.stopSession({ threadId: session.threadId }); routing.codex.startSession.mockClear(); routing.codex.sendTurn.mockClear(); @@ -1727,16 +1746,93 @@ routing.layer("ProviderServiceLive routing", (it) => { cwd?: string; resumeCursor?: unknown; threadId?: string; + modelSelection?: unknown; }; assert.equal(startPayload.provider, "codex"); assert.equal(startPayload.cwd, fixtureCwd("project")); - assert.deepEqual(startPayload.resumeCursor, session.resumeCursor); + assert.deepEqual(startPayload.resumeCursor, rewindCursor); + assert.deepEqual(startPayload.modelSelection, modelSelection); assert.equal(startPayload.threadId, session.threadId); } assert.equal(routing.codex.sendTurn.mock.calls.length, 1); }), ); + it.effect("preserves background turn boundaries when stopping before rollback recovery", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const threadId = asThreadId("thread-background-rewind"); + const initial = yield* provider.startSession(threadId, { + provider: CLAUDE_AGENT_DRIVER, + providerInstanceId: claudeAgentInstanceId, + threadId, + runtimeMode: "full-access", + }); + const cursor = { + resume: "550e8400-e29b-41d4-a716-446655440010", + turnCount: 2, + turnStartMessageIds: ["user-prompt", "background-assistant"], + }; + routing.claude.updateSession(threadId, (session) => ({ ...session, resumeCursor: cursor })); + const completed = yield* provider.streamEvents.pipe( + Stream.filter((event) => event.eventId === "evt-background-rewind"), + Stream.take(1), + Stream.runDrain, + Effect.forkChild, + ); + yield* Effect.yieldNow; + routing.claude.emit({ + type: "turn.completed", + eventId: asEventId("evt-background-rewind"), + provider: CLAUDE_AGENT_DRIVER, + createdAt: "2026-01-01T00:00:00.000Z", + threadId, + turnId: asTurnId("background-turn"), + payload: { state: "completed" }, + }); + yield* Fiber.join(completed); + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const binding = yield* directory.getBinding(threadId); + assert(Option.isSome(binding)); + assert.deepEqual(binding.value.resumeCursor, cursor); + yield* provider.stopSession({ threadId }); + routing.claude.startSession.mockClear(); + yield* provider.rollbackConversation({ threadId, numTurns: 1 }); + assert.deepEqual(routing.claude.startSession.mock.calls[0]?.[0].resumeCursor, cursor); + + const replacement = yield* provider.startSession(threadId, { + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + threadId, + runtimeMode: "full-access", + }); + routing.claude.listSessions.mockReturnValueOnce( + Effect.succeed([{ ...initial, resumeCursor: cursor }]), + ); + const staleCompleted = yield* provider.streamEvents.pipe( + Stream.filter((event) => event.eventId === "evt-stale-background-rewind"), + Stream.take(1), + Stream.runDrain, + Effect.forkChild, + ); + yield* Effect.yieldNow; + routing.claude.emit({ + type: "turn.completed", + eventId: asEventId("evt-stale-background-rewind"), + provider: CLAUDE_AGENT_DRIVER, + createdAt: "2026-01-01T00:00:01.000Z", + threadId, + turnId: asTurnId("old-background-turn"), + payload: { state: "completed" }, + }); + yield* Fiber.join(staleCompleted); + const replacementBinding = yield* directory.getBinding(threadId); + assert(Option.isSome(replacementBinding)); + assert.equal(replacementBinding.value.providerInstanceId, codexInstanceId); + assert.deepEqual(replacementBinding.value.resumeCursor, replacement.resumeCursor); + }), + ); + it.effect("marks a successful fallback compaction as compacted", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 7ffbb113d54..17ece376779 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -1100,6 +1100,35 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( canonicalEvent.type === "turn.aborted" ) { yield* recordTurnCompletedAnalytics(source, canonicalEvent); + if (source.provider === "claudeAgent") { + // Background Claude turns have no sendTurn response to persist their + // new native boundary. Save it before clients can checkpoint the turn. + yield* Effect.gen(function* () { + const adapter = yield* registry.getByInstance(source.instanceId); + const session = (yield* adapter.listSessions()).find( + (session) => session.threadId === canonicalEvent.threadId, + ); + if (session?.resumeCursor !== undefined) { + const binding = yield* directory.getBinding(session.threadId); + if ( + Option.isNone(binding) || + binding.value.providerInstanceId !== source.instanceId + ) { + return; + } + yield* directory.upsert({ + threadId: session.threadId, + provider: source.provider, + providerInstanceId: source.instanceId, + resumeCursor: session.resumeCursor, + }); + } + }).pipe( + Effect.catch((cause) => + Effect.logWarning("failed to persist Claude turn resume state", { cause }), + ), + ); + } } else if (canonicalEvent.type === "session.exited") { yield* clearTurnAnalyticsSession(source.instanceId, canonicalEvent.threadId); } @@ -2006,6 +2035,15 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( "provider.thread_id": input.threadId, }); if (routed.isActive) { + const session = (yield* routed.adapter.listSessions()).find( + (session) => session.threadId === routed.threadId, + ); + if (session) { + yield* upsertSessionBinding( + { ...session, providerInstanceId: routed.instanceId }, + input.threadId, + ); + } yield* routed.adapter.stopSession(routed.threadId); } const pendingCompaction = pendingCompactions.get(input.threadId); @@ -2176,6 +2214,15 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( "provider.rollback_turns": input.numTurns, }); yield* routed.adapter.rollbackThread(routed.threadId, input.numTurns); + const session = (yield* routed.adapter.listSessions()).find( + (session) => session.threadId === routed.threadId, + ); + if (session) { + yield* upsertSessionBinding( + { ...session, providerInstanceId: routed.instanceId }, + input.threadId, + ); + } yield* analytics.record("provider.conversation.rolled_back", { provider: routed.adapter.provider, turns: input.numTurns, diff --git a/apps/server/src/provider/providerSnapshot.ts b/apps/server/src/provider/providerSnapshot.ts index d53367a9064..40ae0eeefde 100644 --- a/apps/server/src/provider/providerSnapshot.ts +++ b/apps/server/src/provider/providerSnapshot.ts @@ -66,6 +66,7 @@ export interface ServerProviderPresentation { readonly showInteractionModeToggle?: boolean; readonly reportsContextWindow?: boolean; readonly requiresNewThreadForModelChange?: boolean; + readonly supportsConversationRollback?: boolean; } export type ServerProviderDraft = Omit; @@ -209,6 +210,9 @@ export function buildServerProvider(input: { : undefined; return { displayName: input.presentation.displayName, + ...(typeof input.presentation.supportsConversationRollback === "boolean" + ? { supportsConversationRollback: input.presentation.supportsConversationRollback } + : {}), ...(input.presentation.badgeLabel ? { badgeLabel: input.presentation.badgeLabel } : {}), ...(typeof input.presentation.showInteractionModeToggle === "boolean" ? { showInteractionModeToggle: input.presentation.showInteractionModeToggle } diff --git a/apps/server/src/vcs/GitVcsDriver.test.ts b/apps/server/src/vcs/GitVcsDriver.test.ts index 031055a3b6c..5c7e2e1360a 100644 --- a/apps/server/src/vcs/GitVcsDriver.test.ts +++ b/apps/server/src/vcs/GitVcsDriver.test.ts @@ -7,7 +7,7 @@ import * as PlatformError from "effect/PlatformError"; import { ChildProcessSpawner } from "effect/unstable/process"; import { assert, it } from "@effect/vitest"; -import { GitCommandError } from "@t3tools/contracts"; +import { CheckpointRef, GitCommandError } from "@t3tools/contracts"; import * as ServerConfig from "../config.ts"; import * as GitVcsDriver from "./GitVcsDriver.ts"; import * as VcsProcess from "./VcsProcess.ts"; @@ -65,6 +65,70 @@ runVcsDriverContractSuite({ }, }); +it.effect("restores empty checkpoints without changing paths outside the workspace", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const driver = yield* GitVcsDriver.makeVcsDriverShape(); + for (const nested of [false, true]) { + const root = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-empty-checkpoint-" }); + yield* runGit(root, ["init"]); + yield* runGit(root, ["config", "user.email", "test@test.com"]); + yield* runGit(root, ["config", "user.name", "Test"]); + if (nested) { + yield* fileSystem.writeFileString(path.join(root, "outside.txt"), "original\n"); + yield* runGit(root, ["add", "."]); + } + yield* runGit(root, ["commit", "--allow-empty", "-m", "initial"]); + const cwd = nested ? path.join(root, "nested") : root; + yield* fileSystem.makeDirectory(cwd, { recursive: true }); + const checkpointRef = CheckpointRef.make("refs/t3/checkpoints/empty"); + yield* driver.checkpoints.captureCheckpoint({ cwd, checkpointRef }); + if (nested) { + yield* fileSystem.writeFileString(path.join(root, "outside.txt"), "changed\n"); + yield* runGit(root, ["add", "outside.txt"]); + } + for (const staged of [false, true]) { + const addedPath = path.join(cwd, "added.txt"); + yield* fileSystem.writeFileString(addedPath, "new\n"); + if (staged) yield* runGit(cwd, ["add", "added.txt"]); + assert.isTrue( + yield* driver.checkpoints.restoreCheckpoint({ + cwd, + checkpointRef, + fallbackToHead: false, + }), + ); + assert.isFalse(yield* fileSystem.exists(addedPath)); + } + yield* fileSystem.writeFileString( + path.join(root, ".git", "info", "exclude"), + "ignored.txt\n", + ); + yield* fileSystem.writeFileString(path.join(cwd, "ignored.txt"), "keep\n"); + yield* fileSystem.makeDirectory(path.join(cwd, "untracked")); + yield* fileSystem.writeFileString(path.join(cwd, "untracked", "file.txt"), "remove\n"); + assert.isTrue( + yield* driver.checkpoints.restoreCheckpoint({ cwd, checkpointRef, fallbackToHead: false }), + ); + assert.strictEqual(yield* fileSystem.readFileString(path.join(cwd, "ignored.txt")), "keep\n"); + assert.isFalse(yield* fileSystem.exists(path.join(cwd, "untracked"))); + if (nested) { + assert.strictEqual( + yield* fileSystem.readFileString(path.join(root, "outside.txt")), + "changed\n", + ); + const staged = yield* driver.execute({ + operation: "test", + cwd: root, + args: ["diff", "--cached", "--name-only"], + }); + assert.strictEqual(staged.stdout.trim(), "outside.txt"); + } + } + }).pipe(Effect.scoped, Effect.provide(GitContractLayer)), +); + it.effect("GitVcsDriver forwards execute env to the VCS process", () => { let observedEnv: NodeJS.ProcessEnv | undefined; let observedAppendTruncationMarker: boolean | undefined; diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index f1e48a24d6f..9b25e915973 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -811,16 +811,56 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( return false; } - yield* execute({ + const tracked = yield* execute({ operation, cwd: input.cwd, - args: ["restore", "--source", commitOid, "--worktree", "--staged", "--", "."], + args: ["ls-files", "--cached", `--with-tree=${commitOid}`, "-z", "--", "."], }); - yield* execute({ + // An empty index and checkpoint have nothing for git restore's pathspec to match. + if (tracked.stdout.length > 0) { + yield* execute({ + operation, + cwd: input.cwd, + args: ["restore", "--source", commitOid, "--worktree", "--staged", "--", "."], + }); + } + // Restoring away the last tracked file can remove a nested workspace directory. + yield* fileSystem.makeDirectory(input.cwd, { recursive: true }).pipe( + Effect.mapError( + (cause) => + new VcsProcessExitError({ + operation, + command: "git restore", + cwd: input.cwd, + exitCode: 0, + detail: `Could not recreate the checkpoint workspace: ${cause.message}`, + }), + ), + ); + const cleaned = yield* execute({ operation, cwd: input.cwd, args: ["clean", "-fd", "--", "."], + allowNonZeroExit: true, }); + if (cleaned.exitCode !== 0) { + // Git can remove every child, then fail trying to remove './' itself. + const emptiedWorkspace = + cleaned.exitCode === 1 && + /^warning: failed to remove \.\/: [^\n]+$/.test(cleaned.stderr.trim()) && + (yield* fileSystem.readDirectory(input.cwd).pipe( + Effect.map((entries) => entries.length === 0), + Effect.catch(() => Effect.succeed(false)), + )); + if (!emptiedWorkspace) + return yield* new VcsProcessExitError({ + operation, + command: "git clean", + cwd: input.cwd, + exitCode: cleaned.exitCode, + detail: cleaned.stderr.trim() || "Could not clean the checkpoint workspace.", + }); + } const headExists = yield* hasHeadCommit(input.cwd); if (headExists) { diff --git a/apps/server/vite.config.ts b/apps/server/vite.config.ts index 88c8c2f4d37..621a1f7bf66 100644 --- a/apps/server/vite.config.ts +++ b/apps/server/vite.config.ts @@ -36,7 +36,7 @@ export default mergeConfig( }, }, pack: { - entry: ["src/bin.ts"], + entry: ["src/bin.ts", "src/claudeHistoryWorker.ts"], outDir: "dist", sourcemap: true, clean: true, diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index ae0cdc8581f..713f1c462a4 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -1,6 +1,8 @@ import { ANTIGRAVITY_DEFAULT_MODEL, + CheckpointRef, EnvironmentId, + EventId, MessageId, ProjectId, ProviderDriverKind, @@ -10,6 +12,9 @@ import { TurnId, } from "@t3tools/contracts"; import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; +import { Atom, AsyncResult } from "effect/unstable/reactivity"; +import { appAtomRegistry } from "../rpc/atomRegistry"; +import { environmentThreadDetails } from "../state/threads"; import type { Thread, ThreadShell, TurnDiffSummary } from "../types"; import { deriveProviderInstanceEntries, NO_PROVIDER_MODEL_SELECTION } from "../providerInstances"; @@ -80,6 +85,8 @@ import { shouldShowPlanFollowUpPrompt, shouldWriteThreadErrorToCurrentServerThread, toolGroupConsumesUpwardNavigation, + waitForRevertedMessage, + prepareRevertedMessageAttachments, } from "./ChatView.logic"; describe("agent browser close confirmation", () => { @@ -2242,3 +2249,134 @@ describe("threadShellHasStarted", () => { expect(threadShellHasStarted(null)).toBe(false); }); }); + +describe("rewind draft recovery", () => { + const message = { + id: MessageId.make("rewound-message"), + role: "user" as const, + text: "edit this question", + turnId: TurnId.make("rewound-turn"), + createdAt: now, + updatedAt: now, + streaming: false, + }; + + afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + vi.useRealTimers(); + }); + + it("waits past command acceptance until the exact message disappears", async () => { + const atom = Atom.make(makeThread({ messages: [message] })); + vi.spyOn(environmentThreadDetails, "detailAtom").mockReturnValue(atom); + let accepted = false; + const result = waitForRevertedMessage({ environmentId, threadId }, message.id, 0, async () => { + accepted = true; + }); + let completed = false; + void result.then(() => { + completed = true; + }); + await Promise.resolve(); + await Promise.resolve(); + expect(accepted).toBe(true); + expect(completed).toBe(false); + appAtomRegistry.set( + atom, + makeThread({ + messages: [], + latestTurn: completedTurn, + checkpoints: [ + { + turnId: completedTurn.turnId, + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make("refs/t3/checkpoints/1"), + status: "ready", + files: [], + assistantMessageId: null, + completedAt: now, + }, + ], + }), + ); + await Promise.resolve(); + expect(completed).toBe(false); + appAtomRegistry.set(atom, makeThread({ messages: [] })); + await result; + }); + + it("rejects a new provider rewind failure without restoring a draft", async () => { + const atom = Atom.make(makeThread({ messages: [message] })); + vi.spyOn(environmentThreadDetails, "detailAtom").mockReturnValue(atom); + const result = waitForRevertedMessage({ environmentId, threadId }, message.id, 0, async () => { + appAtomRegistry.set( + atom, + makeThread({ + messages: [message], + activities: [ + { + id: EventId.make("rewind-failed"), + kind: "checkpoint.revert.failed", + tone: "error", + summary: "Checkpoint revert failed", + payload: { detail: "Native history unavailable", turnCount: 0 }, + turnId: null, + createdAt: now, + }, + ], + }), + ); + }); + await expect(result).rejects.toThrow("Native history unavailable"); + }); + + it("bounds waits when a provider never finishes", async () => { + vi.useFakeTimers(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const clearTimeoutSpy = vi.spyOn(globalThis, "clearTimeout"); + const atom = Atom.make(makeThread({ messages: [message] })); + vi.spyOn(environmentThreadDetails, "detailAtom").mockReturnValue(atom); + const result = waitForRevertedMessage( + { environmentId, threadId }, + message.id, + 0, + async () => {}, + 20, + ); + const timeoutIndex = setTimeoutSpy.mock.calls.findIndex(([, delay]) => delay === 20); + const rewindTimeout = setTimeoutSpy.mock.results[timeoutIndex]?.value; + expect(rewindTimeout).toBeDefined(); + const rejection = expect(result).rejects.toThrow("Timed out waiting"); + await vi.advanceTimersByTimeAsync(20); + await rejection; + expect(clearTimeoutSpy).toHaveBeenCalledWith(rewindTimeout); + }); + + it("copies attachment bytes before rewind into a fresh file", async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response("original bytes")); + vi.stubGlobal("fetch", fetchMock); + const files = await prepareRevertedMessageAttachments({ + message: { + ...message, + attachments: [ + { + type: "file", + id: "old-attachment", + name: "notes.txt", + mimeType: "text/plain", + sizeBytes: 14, + }, + ], + }, + environmentId, + httpBaseUrl: "https://server.test", + createAssetUrl: async () => + AsyncResult.success({ relativeUrl: "/asset/signed", expiresAt: Date.now() + 60_000 }), + }); + expect(files[0]).toBeInstanceOf(File); + expect(files[0]?.name).toBe("notes.txt"); + expect(await files[0]?.text()).toBe("original bytes"); + expect(fetchMock.mock.calls[0]?.[0]).toBe("https://server.test/asset/signed"); + }); +}); diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index b6bb98b4c50..b916862ef59 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -737,6 +737,38 @@ export async function resolveFileAttachmentUrl(input: { return url; } +export async function prepareRevertedMessageAttachments(input: { + message: ChatMessage; + environmentId: EnvironmentId; + httpBaseUrl: string; + createAssetUrl: Parameters[0]["createAssetUrl"]; +}): Promise { + return Promise.all( + (input.message.attachments ?? []).map(async (attachment) => { + if (attachment.type !== "image" && attachment.type !== "file") { + throw new Error("This message has an attachment that cannot be restored."); + } + const result = await input.createAssetUrl({ + environmentId: input.environmentId, + input: { + resource: { + _tag: "attachment", + attachmentId: attachment.id, + fileName: attachment.name, + mimeType: attachment.mimeType, + }, + }, + }); + if (result._tag === "Failure") throw squashAtomCommandFailure(result); + const url = resolveAssetUrl(input.httpBaseUrl, result.value.relativeUrl); + if (url === null) throw new Error("The environment returned an invalid attachment URL."); + const response = await fetch(url, { signal: AbortSignal.timeout(30_000) }); + if (!response.ok) throw new Error(`Could not restore attachment: ${attachment.name}`); + return new File([await response.blob()], attachment.name, { type: attachment.mimeType }); + }), + ); +} + export function revokeUserMessagePreviewUrls(message: ChatMessage): void { if (message.role !== "user" || !message.attachments) { return; @@ -1105,6 +1137,81 @@ export async function waitForStartedServerThread( }); } +export async function waitForRevertedMessage( + threadRef: ScopedThreadRef, + messageId: MessageId, + turnCount: number, + revert: () => Promise, + timeoutMs = 120_000, +): Promise { + const threadAtom = environmentThreadDetails.detailAtom(threadRef); + const initial = appAtomRegistry.get(threadAtom); + if (!initial?.messages.some((message) => message.id === messageId)) { + throw new Error("The message to rewind is no longer available."); + } + const previousFailures = new Set( + initial.activities + .filter((activity) => activity.kind === "checkpoint.revert.failed") + .map((activity) => activity.id), + ); + return new Promise((resolve, reject) => { + let settled = false; + let accepted = false; + let unsubscribe = () => {}; + let timeout: ReturnType | undefined; + const finish = (error?: unknown) => { + if (settled) return; + settled = true; + if (timeout !== undefined) globalThis.clearTimeout(timeout); + unsubscribe(); + if (error !== undefined) reject(error); + else resolve(); + }; + const inspect = () => { + const thread = appAtomRegistry.get(threadAtom); + if (!thread) return; + const failure = thread.activities.findLast( + (activity) => + activity.kind === "checkpoint.revert.failed" && !previousFailures.has(activity.id), + ); + if (failure) { + const payload = failure.payload; + finish( + new Error( + typeof payload === "object" && + payload !== null && + "detail" in payload && + typeof payload.detail === "string" + ? payload.detail + : failure.summary, + ), + ); + } else if ( + accepted && + !thread.messages.some((message) => message.id === messageId) && + thread.checkpoints.every((checkpoint) => checkpoint.checkpointTurnCount <= turnCount) && + (turnCount === 0 + ? thread.latestTurn === null + : thread.checkpoints.some( + (checkpoint) => checkpoint.turnId === thread.latestTurn?.turnId, + )) + ) { + finish(); + } + }; + unsubscribe = appAtomRegistry.subscribe(threadAtom, inspect); + timeout = globalThis.setTimeout(() => { + finish(new Error("Timed out waiting for the thread to rewind.")); + }, timeoutMs); + Promise.resolve() + .then(revert) + .then(() => { + accepted = true; + inspect(); + }, finish); + }); +} + export interface LocalDispatchSnapshot { startedAt: string; preparingWorktree: boolean; diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 212c23ecdb3..0fd1a56639c 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -228,7 +228,7 @@ import { PaperclipIcon, WifiOffIcon, } from "lucide-react"; -import { cn, randomHex } from "~/lib/utils"; +import { cn, randomHex, randomUUID } from "~/lib/utils"; import { stackedThreadToast, toastManager } from "./ui/toast"; import { decodeProjectScriptKeybindingRule } from "~/lib/projectScriptKeybindings"; import { type NewProjectScriptInput } from "./ProjectScriptsControl"; @@ -410,6 +410,8 @@ import { deriveLockedProvider, readFileAsDataUrl, resolveFileAttachmentUrl, + prepareRevertedMessageAttachments, + waitForRevertedMessage, reconcileMountedTerminalThreadIds, recallCheckoutIsRepo, rememberCheckoutIsRepo, @@ -485,7 +487,10 @@ import { supportsServerUpdateThreadContinuation, } from "../versionSkew"; import { useAssetUrls } from "../assets/assetUrls"; -import { ATTACHMENT_ONLY_BOOTSTRAP_PROMPT } from "./chat/composerPromptHistory"; +import { + ATTACHMENT_ONLY_BOOTSTRAP_PROMPT, + recallableComposerPrompt, +} from "./chat/composerPromptHistory"; const EMPTY_ACTIVITIES: OrchestrationThreadActivity[] = []; const EMPTY_PROVIDERS: ServerProvider[] = []; @@ -1427,6 +1432,13 @@ export default function ChatView(props: ChatViewProps) { [environmentId, threadId], ); const routeThreadKey = useMemo(() => scopedThreadKey(routeThreadRef), [routeThreadRef]); + const currentRouteThreadKeyRef = useRef(routeThreadKey); + useLayoutEffect(() => { + currentRouteThreadKeyRef.current = routeThreadKey; + return () => { + currentRouteThreadKeyRef.current = null; + }; + }, [routeThreadKey]); const updateProjectScriptSettings = useAtomCommand(serverEnvironment.updateSettings, { reportFailure: false, }); @@ -1641,7 +1653,9 @@ export default function ChatView(props: ChatViewProps) { Record >({}); const [isConnecting, _setIsConnecting] = useState(false); - const [isRevertingCheckpoint, setIsRevertingCheckpoint] = useState(false); + const isRevertingCheckpoint = useComposerDraftStore((store) => + store.rewindingThreadKeys.has(routeThreadKey), + ); const [maximizedRightPanelThreadKey, setMaximizedRightPanelThreadKey] = useState( null, ); @@ -5342,10 +5356,6 @@ export default function ChatView(props: ChatViewProps) { // activeThreadRef resets transitively with the active thread. }, [activeThread?.id]); - useEffect(() => { - setIsRevertingCheckpoint(false); - }, [activeThread?.id]); - useEffect(() => { if (!activeThread?.id || terminalUiState.terminalOpen) return; const frame = window.requestAnimationFrame(() => { @@ -6514,9 +6524,11 @@ export default function ChatView(props: ChatViewProps) { }, [activeThreadId, composerRef]); const onRevertToTurnCount = useCallback( - async (turnCount: number) => { + async (turnCount: number, messageId: MessageId) => { const localApi = readLocalApi(); if (!localApi || !activeThread || isRevertingCheckpoint) return; + const message = activeThread.messages.find((message) => message.id === messageId); + if (!message || message.role !== "user") return; if (!supportsConversationRollback) { setThreadError( @@ -6538,9 +6550,9 @@ export default function ChatView(props: ChatViewProps) { } const confirmed = await localApi.dialogs.confirm( [ - `Revert this thread to checkpoint ${turnCount}?`, - "This will discard newer messages and turn diffs in this thread.", - "This action cannot be undone.", + "Edit from here?", + "Rewind files and chat to before this message.", + "Your prompt and attachments return to the composer.", ].join("\n"), { variant: "destructive" }, ); @@ -6548,34 +6560,102 @@ export default function ChatView(props: ChatViewProps) { return; } - setIsRevertingCheckpoint(true); + useComposerDraftStore.setState((store) => ({ + rewindingThreadKeys: new Set(store.rewindingThreadKeys).add(routeThreadKey), + })); setThreadError(activeThread.id, null); - const result = await revertThreadCheckpoint({ - environmentId, - input: { - threadId: activeThread.id, - turnCount, - }, - }); - if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { - const error = squashAtomCommandFailure(result); + try { + if (composerRef.current?.hasPendingAttachments()) { + throw new Error("Wait for attachments to finish preparing before rewinding."); + } + const connection = readPreparedConnection(environmentId); + if (!connection) throw new Error("The environment is not connected."); + const files = await prepareRevertedMessageAttachments({ + message, + environmentId, + httpBaseUrl: connection.httpBaseUrl, + createAssetUrl: createAttachmentAssetUrl, + }); + const store = useComposerDraftStore.getState(); + const draft = store.getComposerDraft(composerDraftTarget); + if ( + (draft?.images.length ?? 0) + (draft?.files.length ?? 0) + files.length > + PROVIDER_SEND_TURN_MAX_ATTACHMENTS + ) { + throw new Error( + "Make room for this message's attachments in the composer before rewinding.", + ); + } + await waitForRevertedMessage(routeThreadRef, messageId, turnCount, async () => { + const result = await revertThreadCheckpoint({ + environmentId, + input: { threadId: activeThread.id, turnCount }, + }); + if (result._tag === "Failure") throw squashAtomCommandFailure(result); + }); + const currentPrompt = store.getComposerDraft(composerDraftTarget)?.prompt ?? ""; + const restoredPrompt = recallableComposerPrompt(message.text); + const nextPrompt = + restoredPrompt.length === 0 + ? currentPrompt + : currentPrompt.length > 0 + ? `${currentPrompt}\n\n${restoredPrompt}` + : restoredPrompt; + store.setPrompt(composerDraftTarget, nextPrompt); + const images: ComposerImageAttachment[] = []; + const restoredFiles: ComposerFileAttachment[] = []; + files.forEach((file, index) => { + const attachment = { + id: randomUUID(), + name: file.name, + mimeType: file.type, + sizeBytes: file.size, + file, + }; + if (message.attachments?.[index]?.type === "image") { + images.push({ ...attachment, type: "image", previewUrl: URL.createObjectURL(file) }); + } else { + restoredFiles.push({ ...attachment, type: "file" }); + } + }); + store.addImages(composerDraftTarget, images, { allowDuplicates: true }); + store.addFiles(composerDraftTarget, restoredFiles, { allowDuplicates: true }); + if (currentRouteThreadKeyRef.current === routeThreadKey) { + promptRef.current = nextPrompt; + composerRef.current?.resetCursorState({ prompt: nextPrompt, cursor: nextPrompt.length }); + requestAnimationFrame(() => { + if (currentRouteThreadKeyRef.current === routeThreadKey) + composerRef.current?.focusAtEnd(); + }); + } + } catch (error) { setThreadError( activeThread.id, error instanceof Error ? error.message : "Failed to revert thread state.", ); + } finally { + useComposerDraftStore.setState((store) => { + const remaining = new Set(store.rewindingThreadKeys); + remaining.delete(routeThreadKey); + return { rewindingThreadKeys: remaining }; + }); } - setIsRevertingCheckpoint(false); }, [ activeThread, activeEnvironmentUnavailable, activeEnvironmentUnavailableLabel, + composerDraftTarget, + composerRef, + createAttachmentAssetUrl, environmentId, isConnecting, isRevertingCheckpoint, isSendBusy, phase, revertThreadCheckpoint, + routeThreadKey, + routeThreadRef, setThreadError, supportsConversationRollback, ], @@ -6693,6 +6773,7 @@ export default function ChatView(props: ChatViewProps) { !activeThread || isSendBusy || isConnecting || + isRevertingCheckpoint || !clientSettingsHydrated || threadDetailLoading || sendInFlightRef.current || @@ -8145,8 +8226,8 @@ export default function ChatView(props: ChatViewProps) { // reference is fully stable and never busts TimelineRowCtx identity. const onRevertToTurnCountRef = useRef(onRevertToTurnCount); onRevertToTurnCountRef.current = onRevertToTurnCount; - const onRevertTimelineTurn = useCallback((targetTurnCount: number) => { - void onRevertToTurnCountRef.current(targetTurnCount); + const onRevertTimelineTurn = useCallback((targetTurnCount: number, messageId: MessageId) => { + void onRevertToTurnCountRef.current(targetTurnCount, messageId); }, []); // Files dropped on a sidebar row land here once the dropped-on thread is @@ -8640,6 +8721,7 @@ export default function ChatView(props: ChatViewProps) { {/* Input bar — centered hero while a draft has no messages, docked at the bottom otherwise */}
void; collapseForTimelineScrollKey: (key: string) => void; addDroppedFiles: (files: File[]) => void; + hasPendingAttachments: () => boolean; insertTextAtEnd: (text: string, options?: { ensureLeadingBoundary?: boolean }) => boolean; citeAssistantText: ( citation: AssistantCitation, @@ -1266,6 +1267,7 @@ export interface ChatComposerProps { phase: SessionPhase; isConnecting: boolean; isSendBusy: boolean; + isRevertingCheckpoint?: boolean; sendDisabledReason: string | null; isPreparingWorktree: boolean; bannerItems: readonly ComposerBannerStackItem[]; @@ -1414,6 +1416,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) phase, isConnecting, isSendBusy, + isRevertingCheckpoint = false, sendDisabledReason: externalSendDisabledReason, isPreparingWorktree, environmentUnavailable, @@ -2001,6 +2004,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) * the next draft. */ const pendingImageCompressionsRef = useRef>(new Map()); + const isRevertingCheckpointRef = useRef(isRevertingCheckpoint); + isRevertingCheckpointRef.current = isRevertingCheckpoint; // ------------------------------------------------------------------ // Derived: composer send state @@ -3324,7 +3329,10 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) // A thread switch during the verify await would mix the new thread's // prompt with this invocation's captured target. Nothing was taken yet, // so abort and leave the entry restorable where the user now is. - if (composerTargetKey(composerDraftTarget) !== composerDraftTargetKeyRef.current) { + if ( + isRevertingCheckpointRef.current || + composerTargetKey(composerDraftTarget) !== composerDraftTargetKeyRef.current + ) { return; } @@ -4301,7 +4309,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) // even when the composer is in a state that can't stash. event.preventDefault(); event.stopPropagation(); - if (isCommandPaletteOpen()) { + if (isCommandPaletteOpen() || isRevertingCheckpoint) { return; } if (pendingUserInputs.length > 0 && !isComposerApprovalState) { @@ -4323,6 +4331,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) pendingUserInputs.length, projectSelectionRequired, stashCurrentPrompt, + isRevertingCheckpoint, terminalOpen, ]); @@ -4330,7 +4339,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) // Callbacks: attachments // ------------------------------------------------------------------ const addComposerAttachments = async (files: File[]) => { - if (!activeThreadId || files.length === 0) return; + if (!activeThreadId || files.length === 0 || isRevertingCheckpointRef.current) return; if ( pendingUserInputs.length > 0 && (!supportsQuestionAttachments || @@ -4745,6 +4754,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) void addComposerAttachments(files); focusComposer(); }, + hasPendingAttachments: () => + (pendingImageCompressionsRef.current.get(attachmentTargetKey) ?? 0) > 0, insertTextAtEnd: insertComposerTextAtEnd, citeAssistantText: (citation, sourceAnchor) => insertComposerText( diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index e2c3b6520de..87e11d6280f 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -221,7 +221,7 @@ interface TimelineRowSharedState { workspaceRoot: string | undefined; skills: ReadonlyArray>; activeThreadEnvironmentId: EnvironmentId; - onRevertToTurnCount: (targetTurnCount: number) => void; + onRevertToTurnCount: (targetTurnCount: number, messageId: MessageId) => void; onUseArtifactTemplate: (template: CodexArtifactTemplate) => void; onImageExpand: (preview: ExpandedImagePreview) => void; onFileOpen: (attachment: ChatFileAttachment) => void; @@ -339,7 +339,7 @@ interface MessagesTimelineProps { displayThreadKey?: string; onOpenTurnDiff: (turnId: TurnId, filePath?: string) => void; supportsConversationRollback: boolean; - onRevertToTurnCount: (targetTurnCount: number) => void; + onRevertToTurnCount: (targetTurnCount: number, messageId: MessageId) => void; onUseArtifactTemplate?: (template: CodexArtifactTemplate) => void; isRevertingCheckpoint: boolean; onImageExpand: (preview: ExpandedImagePreview) => void; @@ -1580,7 +1580,7 @@ function UserTimelineRow({ row }: { row: Extract
{typeof revertTurnCount === "number" && ( - + )} {displayedUserMessage.copyText && ( @@ -1592,7 +1592,13 @@ function UserTimelineRow({ row }: { row: Extract ctx.onRevertToTurnCount(turnCount)} - aria-label="Revert to this message" + onClick={() => ctx.onRevertToTurnCount(turnCount, messageId)} + aria-label="Edit from here" /> } > - Revert to this message + Edit from here ); } diff --git a/apps/web/src/composerDraftStore.test.ts b/apps/web/src/composerDraftStore.test.ts index 919342adbfc..66978daf141 100644 --- a/apps/web/src/composerDraftStore.test.ts +++ b/apps/web/src/composerDraftStore.test.ts @@ -145,6 +145,7 @@ function makeTerminalContext(input: { function resetComposerDraftStore() { useComposerDraftStore.setState({ + rewindingThreadKeys: new Set(), draftsByThreadKey: {}, draftThreadsByThreadKey: {}, logicalProjectDraftThreadKeyByLogicalProjectKey: {}, @@ -275,6 +276,28 @@ describe("composerDraftStore addImages", () => { expect(revokeSpy).toHaveBeenCalledWith("blob:duplicate"); }); + it("restores images with matching metadata without replacing unsent bytes", () => { + const store = useComposerDraftStore.getState(); + const unsent = makeImage({ id: "unsent", previewUrl: "blob:unsent" }); + const restored = { + ...unsent, + id: "restored", + previewUrl: "blob:restored", + file: new File([new Uint8Array(unsent.sizeBytes).fill(2)], unsent.name, { + type: unsent.mimeType, + lastModified: unsent.file.lastModified, + }), + }; + store.addImages(threadRef, [unsent]); + store.addImages(threadRef, [restored, restored], { allowDuplicates: true }); + + const images = store.getComposerDraft(threadRef)?.images; + expect(images?.map((image) => image.id)).toEqual(["unsent", "restored"]); + expect(images?.[0]?.file).toBe(unsent.file); + expect(images?.[1]?.file).toBe(restored.file); + expect(revokeSpy).not.toHaveBeenCalled(); + }); + it("deduplicates against existing images across calls by file signature", () => { const first = makeImage({ id: "img-a", @@ -388,6 +411,16 @@ describe("composerDraftStore unsent draft marker", () => { useComposerDraftStore.getState().clearComposerContent(threadRef); expect(hasDraft()).toBe(false); }); + + it("does not persist active rewind locks", () => { + const threadKey = threadKeyFor(threadId, TEST_ENVIRONMENT_ID); + useComposerDraftStore.setState({ rewindingThreadKeys: new Set([threadKey]) }); + + expect(useComposerDraftStore.getState().rewindingThreadKeys.has(threadKey)).toBe(true); + expect(partializeComposerDraftStoreState(useComposerDraftStore.getState())).not.toHaveProperty( + "rewindingThreadKeys", + ); + }); }); describe("composerDraftStore file attachments", () => { @@ -664,6 +697,23 @@ describe("composerDraftStore file attachments", () => { ]); }); + it("restores files with matching metadata without replacing unsent bytes", () => { + const store = useComposerDraftStore.getState(); + const unsent = makeFile("unsent"); + const restored = { + ...unsent, + id: "restored", + file: new File(["edited"], unsent.name, { type: unsent.mimeType }), + }; + store.addFiles(threadRef, [unsent]); + store.addFiles(threadRef, [restored, restored], { allowDuplicates: true }); + + const files = store.getComposerDraft(threadRef)?.files; + expect(files?.map((file) => file.id)).toEqual(["unsent", "restored"]); + expect(files?.[0]?.file).toBe(unsent.file); + expect(files?.[1]?.file).toBe(restored.file); + }); + it("keeps same-name videos with different MIME types", () => { const store = useComposerDraftStore.getState(); const mp4 = new File(["report"], "clip", { type: "video/mp4" }); @@ -1136,10 +1186,19 @@ describe("composerDraftStore project draft thread mapping", () => { store.addImage(localDraftId, makeImage({ id: "img-local", previewUrl: "blob:local-draft" })); store.setPrompt(localThreadRef, "local thread draft"); store.setPrompt(remoteThreadRef, "remote thread draft"); + useComposerDraftStore.setState({ + rewindingThreadKeys: new Set([ + threadKeyFor(threadId, TEST_ENVIRONMENT_ID), + threadKeyFor(otherThreadId, OTHER_TEST_ENVIRONMENT_ID), + ]), + }); clearComposerDraftsEnvironment(TEST_ENVIRONMENT_ID); const next = useComposerDraftStore.getState(); + expect([...next.rewindingThreadKeys]).toEqual([ + threadKeyFor(otherThreadId, OTHER_TEST_ENVIRONMENT_ID), + ]); expect(next.getDraftThreadByProjectRef(projectRef)).toBeNull(); expect(next.getDraftThreadByProjectRef(remoteProjectRef)).not.toBeNull(); expect(next.getComposerDraft(localDraftId)).toBeNull(); diff --git a/apps/web/src/composerDraftStore.ts b/apps/web/src/composerDraftStore.ts index 8f4ad48e16a..1b457258649 100644 --- a/apps/web/src/composerDraftStore.ts +++ b/apps/web/src/composerDraftStore.ts @@ -475,6 +475,7 @@ interface ComposerDraftStoreState { draftThreadsByThreadKey: Record; logicalProjectDraftThreadKeyByLogicalProjectKey: Record; backgroundSubmissionThreadKeys: Record; + rewindingThreadKeys: ReadonlySet; stickyModelSelectionByProvider: Partial>; stickyActiveProvider: ProviderInstanceId | null; /** Returns the editable composer content for a draft session or server thread. */ @@ -602,9 +603,17 @@ interface ComposerDraftStoreState { interactionMode: ProviderInteractionMode | null | undefined, ) => void; addImage: (threadRef: ComposerThreadTarget, image: ComposerImageAttachment) => boolean; - addImages: (threadRef: ComposerThreadTarget, images: ComposerImageAttachment[]) => void; + addImages: ( + threadRef: ComposerThreadTarget, + images: ComposerImageAttachment[], + options?: { allowDuplicates?: boolean }, + ) => void; removeImage: (threadRef: ComposerThreadTarget, imageId: string) => void; - addFiles: (threadRef: ComposerThreadTarget, files: ComposerFileAttachment[]) => void; + addFiles: ( + threadRef: ComposerThreadTarget, + files: ComposerFileAttachment[], + options?: { allowDuplicates?: boolean }, + ) => void; removeFile: (threadRef: ComposerThreadTarget, fileId: string) => void; setFileUpload: ( threadRef: ComposerThreadTarget, @@ -2522,6 +2531,7 @@ const composerDraftStore = create()( draftThreadsByThreadKey: {}, logicalProjectDraftThreadKeyByLogicalProjectKey: {}, backgroundSubmissionThreadKeys: {}, + rewindingThreadKeys: new Set(), stickyModelSelectionByProvider: {}, stickyActiveProvider: null, getComposerDraft: (target) => getComposerDraftState(get(), target), @@ -3289,7 +3299,7 @@ const composerDraftStore = create()( (get().draftsByThreadKey[threadKey]?.images.some(({ id }) => id === image.id) ?? false) ); }, - addImages: (threadRef, images) => { + addImages: (threadRef, images, options) => { const threadKey = resolveComposerDraftKey(get(), threadRef) ?? ""; if (threadKey.length === 0 || images.length === 0) { return; @@ -3304,7 +3314,10 @@ const composerDraftStore = create()( const dedupedIncoming: ComposerImageAttachment[] = []; for (const image of images) { const dedupKey = composerImageDedupKey(image); - if (existingIds.has(image.id) || existingDedupKeys.has(dedupKey)) { + if ( + existingIds.has(image.id) || + (!options?.allowDuplicates && existingDedupKeys.has(dedupKey)) + ) { // Avoid revoking a blob URL that's still referenced by an accepted image. if (!acceptedPreviewUrls.has(image.previewUrl)) { revokeObjectPreviewUrl(image.previewUrl); @@ -3374,7 +3387,7 @@ const composerDraftStore = create()( return { draftsByThreadKey: nextDraftsByThreadKey }; }); }, - addFiles: (threadRef, files) => { + addFiles: (threadRef, files, options) => { const threadKey = resolveComposerDraftKey(get(), threadRef) ?? ""; if (threadKey.length === 0 || files.length === 0) { return; @@ -3394,14 +3407,15 @@ const composerDraftStore = create()( if (knownIds.has(file.id)) { continue; } - const duplicate = - knownFiles.get(key) ?? - existing.files.find( - (candidate) => - composerFileNeedsReattach(candidate) && - !replacements.has(candidate.id) && - composerFileMatchesReattachMarker(candidate, file), - ); + const duplicate = options?.allowDuplicates + ? undefined + : (knownFiles.get(key) ?? + existing.files.find( + (candidate) => + composerFileNeedsReattach(candidate) && + !replacements.has(candidate.id) && + composerFileMatchesReattachMarker(candidate, file), + )); if (duplicate) { // A needs-reattach marker is not a usable duplicate. Replace // it so the upload restarts. @@ -4105,6 +4119,11 @@ export function clearComposerDraftsEnvironment(environmentId: EnvironmentId): vo draftThreadsByThreadKey: nextDraftThreads, logicalProjectDraftThreadKeyByLogicalProjectKey: nextLogicalMappings, backgroundSubmissionThreadKeys: nextBackgroundSubmissionThreadKeys, + rewindingThreadKeys: new Set( + [...state.rewindingThreadKeys].filter( + (threadKey) => parseScopedThreadKey(threadKey)?.environmentId !== environmentId, + ), + ), }; }); composerDebouncedStorage.flush(); diff --git a/docs/user/composer.md b/docs/user/composer.md index 4a8df533366..befdd656bc2 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -69,6 +69,17 @@ the text is an unedited recalled prompt, with the caret on the first visual line the last visual line for `ArrowDown`, counting wrapped lines. Editing a recalled prompt turns it into a normal draft. +## Edit an earlier prompt + +On web and desktop, choose **Edit from here** beneath a sent message to rewind +the conversation and workspace files to before that message. The selected prompt +and its attachments return to the composer for editing and resending. Any unsent +draft stays above the restored prompt. + +This removes the selected message and later conversation from the active thread +and provider history. It does not undo external actions or separate provider +memory. The action is available only when the provider supports rewind. + ## Prompt stash On web and desktop, press `Cmd+S` on macOS or `Ctrl+S` on Windows and Linux to save diff --git a/knip.jsonc b/knip.jsonc index 859c309ab7c..d00bd5f46d6 100644 --- a/knip.jsonc +++ b/knip.jsonc @@ -21,6 +21,7 @@ "entry": [ "src/bin.ts!", "src/service-launcher.ts!", + "src/claudeHistoryWorker.ts!", "scripts/cli.ts", "src/provider/testFixtures/*.mjs", ],