diff --git a/apps/server/src/checkpointing/CheckpointStore.test.ts b/apps/server/src/checkpointing/CheckpointStore.test.ts index bf332d20d0d..978327d7ee5 100644 --- a/apps/server/src/checkpointing/CheckpointStore.test.ts +++ b/apps/server/src/checkpointing/CheckpointStore.test.ts @@ -1,4 +1,5 @@ // @effect-diagnostics nodeBuiltinImport:off +import * as NodeCrypto from "node:crypto"; import * as NodePath from "node:path"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -112,9 +113,247 @@ it.layer(TestLayer)("CheckpointStore.layer", (it) => { expect(yield* checkpointStore.isGitRepository(tmp)).toBe(true); }), ); + + it.effect("recognizes an owned Git repository through a symbolic path alias", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir(); + const fileSystem = yield* FileSystem.FileSystem; + const repositoryPath = NodePath.join(tmp, "repository"); + const aliasPath = NodePath.join(tmp, "repository-alias"); + yield* fileSystem.makeDirectory(repositoryPath); + yield* initRepoWithCommit(repositoryPath); + yield* fileSystem.symlink(repositoryPath, aliasPath); + const checkpointStore = yield* CheckpointStore.CheckpointStore; + + expect(yield* checkpointStore.isGitRepository(aliasPath)).toBe(true); + }), + ); + + it.effect("treats a standalone project nested under another repository as non-Git", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir(); + const fileSystem = yield* FileSystem.FileSystem; + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const project = NodePath.join(tmp, "project"); + yield* git(tmp, ["init"]); + yield* fileSystem.makeDirectory(project); + + expect(yield* checkpointStore.isGitRepository(project)).toBe(false); + }), + ); }); describe("diffCheckpoints", () => { + it.effect("captures and restores an initialized repository before its first commit", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir("checkpoint-store-unborn-git-test-"); + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const threadId = ThreadId.make("thread-unborn-git-checkpoint-store"); + const baselineRef = checkpointRefForThreadTurn(threadId, 0); + const sourcePath = NodePath.join(tmp, "source.txt"); + + yield* git(tmp, ["init"]); + yield* writeTextFile(sourcePath, "before\n"); + yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: baselineRef }); + yield* writeTextFile(sourcePath, "after\n"); + + expect( + yield* checkpointStore.restoreCheckpoint({ + cwd: tmp, + checkpointRef: baselineRef, + }), + ).toBe(true); + const fileSystem = yield* FileSystem.FileSystem; + expect(yield* fileSystem.readFileString(sourcePath)).toBe("before\n"); + }), + ); + + it.effect("restores an empty checkpoint in a repository before its first commit", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir("checkpoint-store-unborn-empty-test-"); + const fileSystem = yield* FileSystem.FileSystem; + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const baselineRef = checkpointRefForThreadTurn( + ThreadId.make("thread-unborn-empty-checkpoint-store"), + 0, + ); + const createdPath = NodePath.join(tmp, "created.txt"); + const postCheckpointIgnorePath = NodePath.join(tmp, ".gitignore"); + const ignoredCreatedPath = NodePath.join(tmp, "state.cache"); + + yield* git(tmp, ["init"]); + yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: baselineRef }); + yield* writeTextFile(createdPath, "created after baseline\n"); + yield* writeTextFile(postCheckpointIgnorePath, "*.cache\n"); + yield* writeTextFile(ignoredCreatedPath, "ignored after baseline\n"); + + expect( + yield* checkpointStore.restoreCheckpoint({ + cwd: tmp, + checkpointRef: baselineRef, + }), + ).toBe(true); + expect(yield* fileSystem.exists(createdPath)).toBe(false); + expect(yield* fileSystem.exists(postCheckpointIgnorePath)).toBe(false); + expect(yield* fileSystem.exists(ignoredCreatedPath)).toBe(false); + }), + ); + + it.effect("preserves files ignored by the restored checkpoint", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir("checkpoint-store-unborn-ignored-test-"); + const fileSystem = yield* FileSystem.FileSystem; + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const baselineRef = checkpointRefForThreadTurn( + ThreadId.make("thread-unborn-ignored-checkpoint-store"), + 0, + ); + const ignorePath = NodePath.join(tmp, ".gitignore"); + const ignoredPath = NodePath.join(tmp, "local.cache"); + + yield* git(tmp, ["init"]); + yield* writeTextFile(ignorePath, "*.cache\n"); + yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: baselineRef }); + yield* writeTextFile(ignoredPath, "local-only data\n"); + + expect( + yield* checkpointStore.restoreCheckpoint({ + cwd: tmp, + checkpointRef: baselineRef, + }), + ).toBe(true); + expect(yield* fileSystem.readFileString(ignorePath)).toBe("*.cache\n"); + expect(yield* fileSystem.readFileString(ignoredPath)).toBe("local-only data\n"); + }), + ); + + it.effect("uses shadow checkpoints for a project nested under an unrelated repository", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir("checkpoint-store-nested-project-test-"); + const fileSystem = yield* FileSystem.FileSystem; + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const project = NodePath.join(tmp, "project"); + const threadId = ThreadId.make("thread-nested-project-checkpoint-store"); + const baselineRef = checkpointRefForThreadTurn(threadId, 0); + const changedRef = checkpointRefForThreadTurn(threadId, 1); + const createdPath = NodePath.join(project, "created.txt"); + + yield* git(tmp, ["init"]); + yield* fileSystem.makeDirectory(project); + yield* checkpointStore.captureCheckpoint({ cwd: project, checkpointRef: baselineRef }); + yield* writeTextFile(createdPath, "created\n"); + yield* checkpointStore.captureCheckpoint({ cwd: project, checkpointRef: changedRef }); + + expect( + yield* checkpointStore.restoreCheckpoint({ + cwd: project, + checkpointRef: baselineRef, + }), + ).toBe(true); + expect(yield* fileSystem.exists(createdPath)).toBe(false); + expect(yield* checkpointStore.isGitRepository(project)).toBe(false); + }), + ); + + it.effect("captures and restores files without initializing Git in the workspace", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir("checkpoint-store-shadow-test-"); + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const threadId = ThreadId.make("thread-shadow-checkpoint-store"); + const baselineRef = checkpointRefForThreadTurn(threadId, 0); + const changedRef = checkpointRefForThreadTurn(threadId, 1); + const sourcePath = NodePath.join(tmp, "source.txt"); + const createdPath = NodePath.join(tmp, "created.txt"); + + yield* writeTextFile(sourcePath, "before\n"); + yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: baselineRef }); + yield* writeTextFile(sourcePath, "after\n"); + yield* writeTextFile(createdPath, "new\n"); + yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: changedRef }); + + const diff = yield* checkpointStore.diffCheckpoints({ + cwd: tmp, + fromCheckpointRef: baselineRef, + toCheckpointRef: changedRef, + ignoreWhitespace: false, + }); + expect(diff).toContain("+after"); + expect(diff).toContain("created.txt"); + + expect( + yield* checkpointStore.restoreCheckpoint({ + cwd: tmp, + checkpointRef: baselineRef, + }), + ).toBe(true); + const fileSystem = yield* FileSystem.FileSystem; + expect(yield* fileSystem.readFileString(sourcePath)).toBe("before\n"); + expect(yield* fileSystem.exists(createdPath)).toBe(false); + expect(yield* checkpointStore.isGitRepository(tmp)).toBe(false); + }), + ); + + it.effect("fails shadow checkpoint diffs when a ref is unavailable", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir("checkpoint-store-shadow-invalid-ref-test-"); + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const threadId = ThreadId.make("thread-shadow-invalid-ref-checkpoint-store"); + const baselineRef = checkpointRefForThreadTurn(threadId, 0); + const missingRef = checkpointRefForThreadTurn(threadId, 1); + + yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef: baselineRef }); + const error = yield* Effect.flip( + checkpointStore.diffCheckpoints({ + cwd: tmp, + fromCheckpointRef: baselineRef, + toCheckpointRef: missingRef, + ignoreWhitespace: false, + }), + ); + + expect(error._tag).toBe("VcsProcessExitError"); + expect(error.message).toContain("git diff"); + }), + ); + + it.effect("surfaces corrupted shadow checkpoint refs instead of treating them as missing", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir("checkpoint-store-shadow-corrupt-ref-test-"); + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const fileSystem = yield* FileSystem.FileSystem; + const config = yield* ServerConfig.ServerConfig; + const checkpointRef = checkpointRefForThreadTurn( + ThreadId.make("thread-shadow-corrupt-ref-checkpoint-store"), + 0, + ); + + yield* checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef }); + const shadowRepositoryPath = NodePath.join( + config.stateDir, + "checkpoints", + NodeCrypto.createHash("sha256").update(NodePath.resolve(tmp)).digest("hex"), + ); + yield* fileSystem.writeFileString( + NodePath.join(shadowRepositoryPath, checkpointRef), + "not-a-commit\n", + ); + + const error = yield* Effect.flip( + checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef }), + ); + expect(error._tag).toBe("VcsProcessExitError"); + expect(error.message).toContain("git for-each-ref"); + + const captureError = yield* Effect.flip( + checkpointStore.captureCheckpoint({ cwd: tmp, checkpointRef }), + ); + expect(captureError._tag).toBe("VcsProcessExitError"); + expect( + yield* Effect.flip(checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef })), + ).toMatchObject({ _tag: "VcsProcessExitError" }); + }), + ); + it.effect("returns full oversized checkpoint diffs without truncation", () => Effect.gen(function* () { const tmp = yield* makeTmpDir(); diff --git a/apps/server/src/checkpointing/CheckpointStore.ts b/apps/server/src/checkpointing/CheckpointStore.ts index f13aa4572c1..e84ecfd34c5 100644 --- a/apps/server/src/checkpointing/CheckpointStore.ts +++ b/apps/server/src/checkpointing/CheckpointStore.ts @@ -1,9 +1,9 @@ /** * CheckpointStore - Repository interface for filesystem-backed workspace checkpoints. * - * Owns hidden Git-ref checkpoint capture/restore and diff computation for a - * workspace thread timeline. It does not store user-facing checkpoint metadata - * and does not coordinate provider conversation rollback. + * Owns hidden checkpoint capture/restore and diff computation for a workspace + * thread timeline. Git workspaces use their repository's hidden refs; other + * workspaces use a T3-managed shadow Git store outside the project directory. * * The live adapter resolves the active VCS driver once per checkpoint operation * and delegates to the driver's optional checkpoint capability. @@ -13,14 +13,21 @@ * * @module CheckpointStore */ -import { VcsUnsupportedOperationError, type CheckpointRef } from "@t3tools/contracts"; +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeCrypto from "node:crypto"; +import * as NodePath from "node:path"; + +import { type CheckpointRef, VcsProcessExitError } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import type { CheckpointStoreError } from "./Errors.ts"; import type { VcsCheckpointOps } from "../vcs/VcsDriver.ts"; +import { ServerConfig } from "../config.ts"; import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; +import * as VcsProcess from "../vcs/VcsProcess.ts"; export interface CaptureCheckpointInput { readonly cwd: string; @@ -50,7 +57,7 @@ export interface DeleteCheckpointRefsInput { export class CheckpointStore extends Context.Service< CheckpointStore, { - /** Check whether cwd is inside a Git worktree. */ + /** Check whether cwd is inside a user-managed Git worktree. */ readonly isGitRepository: (cwd: string) => Effect.Effect; /** @@ -98,26 +105,258 @@ export class CheckpointStore extends Context.Service< export const make = Effect.gen(function* () { const vcsRegistry = yield* VcsDriverRegistry.VcsDriverRegistry; + const vcsProcess = yield* VcsProcess.VcsProcess; + const fileSystem = yield* FileSystem.FileSystem; + const config = yield* ServerConfig; + + const shadowRepositoryPath = (cwd: string) => + NodePath.join( + config.stateDir, + "checkpoints", + NodeCrypto.createHash("sha256").update(NodePath.resolve(cwd)).digest("hex"), + ); + const canonicalizeWorkspacePath = (value: string) => + fileSystem.realPath(value).pipe(Effect.orElseSucceed(() => NodePath.resolve(value))); + + const runShadowGit = Effect.fn("CheckpointStore.runShadowGit")(function* (input: { + readonly operation: string; + readonly cwd: string; + readonly args: ReadonlyArray; + readonly env?: NodeJS.ProcessEnv; + readonly allowNonZeroExit?: boolean; + readonly maxOutputBytes?: number; + }) { + const gitDir = shadowRepositoryPath(input.cwd); + yield* vcsProcess.run({ + operation: `${input.operation}.init`, + command: "git", + cwd: input.cwd, + args: ["init", "--bare", "--quiet", gitDir], + }); + return yield* vcsProcess.run({ + operation: input.operation, + command: "git", + cwd: input.cwd, + args: [`--git-dir=${gitDir}`, `--work-tree=${NodePath.resolve(input.cwd)}`, ...input.args], + ...(input.env ? { env: input.env } : {}), + ...(input.allowNonZeroExit !== undefined ? { allowNonZeroExit: input.allowNonZeroExit } : {}), + ...(input.maxOutputBytes !== undefined ? { maxOutputBytes: input.maxOutputBytes } : {}), + }); + }); + + const resolveShadowCommit = Effect.fn("CheckpointStore.resolveShadowCommit")(function* (input: { + readonly cwd: string; + readonly checkpointRef: CheckpointRef; + }) { + const refResult = yield* runShadowGit({ + operation: "CheckpointStore.shadow.resolveCheckpointRef", + cwd: input.cwd, + args: ["for-each-ref", "--format=%(objectname)", "--", input.checkpointRef], + allowNonZeroExit: true, + }); + const refStdout = refResult.stdout.trim(); + const refStderr = refResult.stderr.trim(); + if (refResult.exitCode !== 0 || (refStdout.length === 0 && refStderr.length > 0)) { + return yield* new VcsProcessExitError({ + operation: "CheckpointStore.shadow.resolveCheckpointRef", + command: "git for-each-ref", + cwd: input.cwd, + exitCode: refResult.exitCode === 0 ? 1 : refResult.exitCode, + detail: "Failed to resolve shadow checkpoint ref.", + stderrLength: refStderr.length, + stderrTruncated: refResult.stderrTruncated, + }); + } + if (refStdout.length === 0) { + return null; + } + + const result = yield* runShadowGit({ + operation: "CheckpointStore.shadow.resolveCheckpointCommit", + cwd: input.cwd, + args: ["rev-parse", "--verify", "--quiet", `${input.checkpointRef}^{commit}`], + allowNonZeroExit: true, + }); + const stdout = result.stdout.trim(); + const stderr = result.stderr.trim(); + if (result.exitCode !== 0) { + return yield* new VcsProcessExitError({ + operation: "CheckpointStore.shadow.resolveCheckpointCommit", + command: "git rev-parse", + cwd: input.cwd, + exitCode: result.exitCode, + detail: `Checkpoint ref ${input.checkpointRef} does not resolve to a commit.`, + stderrLength: stderr.length, + stderrTruncated: result.stderrTruncated, + }); + } + return stdout.length > 0 ? stdout : null; + }); + + const shadowCheckpoints: VcsCheckpointOps = { + captureCheckpoint: Effect.fn("CheckpointStore.shadow.captureCheckpoint")(function* (input) { + const gitDir = shadowRepositoryPath(input.cwd); + const tempIndexPath = NodePath.join(gitDir, `t3-checkpoint-index-${NodeCrypto.randomUUID()}`); + const env: NodeJS.ProcessEnv = { + ...process.env, + GIT_INDEX_FILE: tempIndexPath, + GIT_AUTHOR_NAME: "T3 Code", + GIT_AUTHOR_EMAIL: "t3code@users.noreply.github.com", + GIT_COMMITTER_NAME: "T3 Code", + GIT_COMMITTER_EMAIL: "t3code@users.noreply.github.com", + }; + const cleanup = fileSystem.remove(tempIndexPath, { force: true }).pipe(Effect.ignore); + + yield* Effect.gen(function* () { + yield* runShadowGit({ + operation: "CheckpointStore.shadow.captureCheckpoint.readTree", + cwd: input.cwd, + args: ["read-tree", "--empty"], + env, + }); + yield* runShadowGit({ + operation: "CheckpointStore.shadow.captureCheckpoint.add", + cwd: input.cwd, + args: ["add", "-A", "--", "."], + env, + }); + const tree = yield* runShadowGit({ + operation: "CheckpointStore.shadow.captureCheckpoint.writeTree", + cwd: input.cwd, + args: ["write-tree"], + env, + }); + const commit = yield* runShadowGit({ + operation: "CheckpointStore.shadow.captureCheckpoint.commitTree", + cwd: input.cwd, + args: [ + "commit-tree", + tree.stdout.trim(), + "-m", + `t3 checkpoint ref=${input.checkpointRef}`, + ], + env, + }); + yield* runShadowGit({ + operation: "CheckpointStore.shadow.captureCheckpoint.updateRef", + cwd: input.cwd, + args: ["update-ref", input.checkpointRef, commit.stdout.trim()], + }); + }).pipe(Effect.ensuring(cleanup)); + }), + + hasCheckpointRef: (input) => + resolveShadowCommit(input).pipe(Effect.map((commit) => commit !== null)), + + restoreCheckpoint: Effect.fn("CheckpointStore.shadow.restoreCheckpoint")(function* (input) { + const commit = yield* resolveShadowCommit(input); + if (!commit) { + // Shadow repositories intentionally have no HEAD: every restorable + // state must be addressed by its explicit checkpoint ref. Fail closed + // so callers never roll conversation history back without matching files. + return false; + } + const tempIndexPath = NodePath.join( + shadowRepositoryPath(input.cwd), + `t3-checkpoint-index-${NodeCrypto.randomUUID()}`, + ); + const env: NodeJS.ProcessEnv = { ...process.env, GIT_INDEX_FILE: tempIndexPath }; + const cleanup = fileSystem.remove(tempIndexPath, { force: true }).pipe(Effect.ignore); + yield* Effect.gen(function* () { + yield* runShadowGit({ + operation: "CheckpointStore.shadow.restoreCheckpoint.readTree", + cwd: input.cwd, + args: ["read-tree", commit], + env, + }); + yield* runShadowGit({ + operation: "CheckpointStore.shadow.restoreCheckpoint.checkout", + cwd: input.cwd, + args: ["checkout-index", "--all", "--force"], + env, + }); + yield* runShadowGit({ + operation: "CheckpointStore.shadow.restoreCheckpoint.clean", + cwd: input.cwd, + args: ["clean", "-fd", "--", "."], + env, + }); + }).pipe(Effect.ensuring(cleanup)); + return true; + }), + + diffCheckpoints: Effect.fn("CheckpointStore.shadow.diffCheckpoints")(function* (input) { + const result = yield* runShadowGit({ + operation: "CheckpointStore.shadow.diffCheckpoints", + cwd: input.cwd, + args: [ + "diff", + "--patch", + "--no-color", + "--no-ext-diff", + "--no-textconv", + ...(input.ignoreWhitespace ? ["--ignore-all-space"] : []), + `${input.fromCheckpointRef}^{commit}`, + `${input.toCheckpointRef}^{commit}`, + ], + allowNonZeroExit: true, + maxOutputBytes: 50 * 1024 * 1024, + }); + if (result.exitCode !== 0) { + return yield* new VcsProcessExitError({ + operation: "CheckpointStore.shadow.diffCheckpoints", + command: "git diff", + cwd: input.cwd, + exitCode: result.exitCode, + detail: "Checkpoint ref is unavailable for diff operation.", + stderrLength: result.stderr.trim().length, + stderrTruncated: result.stderrTruncated, + }); + } + return result.stdout; + }), + + deleteCheckpointRefs: Effect.fn("CheckpointStore.shadow.deleteCheckpointRefs")( + function* (input) { + yield* Effect.forEach( + input.checkpointRefs, + (checkpointRef) => + runShadowGit({ + operation: "CheckpointStore.shadow.deleteCheckpointRefs", + cwd: input.cwd, + args: ["update-ref", "-d", checkpointRef], + allowNonZeroExit: true, + }), + { concurrency: 1, discard: true }, + ); + }, + ), + }; const resolveCheckpoints = Effect.fn("CheckpointStore.resolveCheckpoints")(function* ( - operation: string, + _operation: string, cwd: string, ) { - const handle = yield* vcsRegistry.resolve({ cwd }); - if (!handle.driver.checkpoints) { - return yield* new VcsUnsupportedOperationError({ - operation, - kind: handle.kind, - detail: `${handle.kind} driver does not implement checkpoint operations.`, - }); - } - return handle.driver.checkpoints satisfies VcsCheckpointOps; + const handle = yield* vcsRegistry.detect({ cwd }); + const workspaceOwnsRepository = + handle !== null && + (yield* canonicalizeWorkspacePath(handle.repository.rootPath)) === + (yield* canonicalizeWorkspacePath(cwd)); + return workspaceOwnsRepository + ? (handle.driver.checkpoints ?? shadowCheckpoints) + : shadowCheckpoints; }); - const isGitRepository: CheckpointStore["Service"]["isGitRepository"] = (cwd) => - vcsRegistry - .detect({ cwd, requestedKind: "git" }) - .pipe(Effect.map((repository) => repository !== null)); + const isGitRepository: CheckpointStore["Service"]["isGitRepository"] = Effect.fn( + "CheckpointStore.isGitRepository", + )(function* (cwd) { + const repository = yield* vcsRegistry.detect({ cwd, requestedKind: "git" }); + return ( + repository !== null && + (yield* canonicalizeWorkspacePath(repository.repository.rootPath)) === + (yield* canonicalizeWorkspacePath(cwd)) + ); + }); const captureCheckpoint: CheckpointStore["Service"]["captureCheckpoint"] = Effect.fn( "captureCheckpoint", diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 707c87c43c9..1fe198f18a9 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -35,7 +35,7 @@ import * as VcsDriverRegistry from "../../vcs/VcsDriverRegistry.ts"; import * as VcsProcess from "../../vcs/VcsProcess.ts"; import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts"; import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts"; -import { CheckpointReactorLive } from "./CheckpointReactor.ts"; +import { CheckpointReactorLive, conversationTurnCountForTurn } from "./CheckpointReactor.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; @@ -229,6 +229,24 @@ function gitShowFileAtRef(cwd: string, ref: string, filePath: string): string { return runGit(cwd, ["show", `${ref}:${filePath}`]); } +describe("conversationTurnCountForTurn", () => { + it("counts steering prompts through the last assistant message for the same turn", () => { + const steeredTurnId = asTurnId("steered-turn"); + const laterTurnId = asTurnId("later-turn"); + const messages = [ + { role: "user", turnId: null }, + { role: "assistant", turnId: steeredTurnId }, + { role: "user", turnId: null }, + { role: "assistant", turnId: steeredTurnId }, + { role: "user", turnId: null }, + { role: "assistant", turnId: laterTurnId }, + ]; + + expect(conversationTurnCountForTurn(messages, steeredTurnId)).toBe(2); + expect(conversationTurnCountForTurn(messages, laterTurnId)).toBe(3); + }); +}); + async function waitForGitRefExists(cwd: string, ref: string, timeoutMs = 15_000) { const deadline = (await Effect.runPromise(Clock.currentTimeMillis)) + timeoutMs; const poll = async (): Promise => { @@ -413,6 +431,7 @@ describe("CheckpointReactor", () => { return { engine, + runPromise: (effect: Effect.Effect) => runtime!.runPromise(effect), readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()), provider, cwd, @@ -424,7 +443,7 @@ describe("CheckpointReactor", () => { const harness = await createHarness({ seedFilesystemCheckpoints: false }); const createdAt = "2026-01-01T00:00:00.000Z"; - await Effect.runPromise( + await harness.runPromise( harness.engine.dispatch({ type: "thread.session.set", commandId: CommandId.make("cmd-session-set-capture"), @@ -786,6 +805,81 @@ describe("CheckpointReactor", () => { ).toBe("v2\n"); }); + it("keeps the authoritative turn count when a placeholder is processed late", async () => { + const harness = await createHarness({ seedFilesystemCheckpoints: false }); + const threadId = ThreadId.make("thread-1"); + const createdAt = "2026-01-01T00:00:00.000Z"; + + for (const turnNumber of [1, 2]) { + const turnId = asTurnId(`late-placeholder-turn-${turnNumber}`); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`cmd-late-placeholder-user-${turnNumber}`), + threadId, + message: { + messageId: MessageId.make(`late-placeholder-user-${turnNumber}`), + role: "user", + text: `User ${turnNumber}`, + attachments: [], + }, + runtimeMode: "approval-required", + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + createdAt, + }), + ); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.message.assistant.delta", + commandId: CommandId.make(`cmd-late-placeholder-assistant-delta-${turnNumber}`), + threadId, + messageId: MessageId.make(`late-placeholder-assistant-${turnNumber}`), + delta: `Assistant ${turnNumber}`, + turnId, + createdAt, + }), + ); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.message.assistant.complete", + commandId: CommandId.make(`cmd-late-placeholder-assistant-complete-${turnNumber}`), + threadId, + messageId: MessageId.make(`late-placeholder-assistant-${turnNumber}`), + turnId, + createdAt, + }), + ); + } + + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.turn.diff.complete", + commandId: CommandId.make("cmd-late-placeholder-diff"), + threadId, + turnId: asTurnId("late-placeholder-turn-1"), + completedAt: createdAt, + checkpointRef: checkpointRefForThreadTurn(threadId, 1), + status: "missing", + files: [], + checkpointTurnCount: 1, + createdAt, + }), + ); + + const thread = await waitForThread( + harness.readModel, + (entry) => + entry.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 1) && + entry.activities.some((activity) => activity.kind === "checkpoint.captured"), + ); + expect(thread.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 1)).toBe( + true, + ); + expect(thread.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 2)).toBe( + false, + ); + }); + it("ignores non-v2 checkpoint.captured runtime events", async () => { const harness = await createHarness(); const createdAt = "2026-01-01T00:00:00.000Z"; @@ -828,7 +922,7 @@ describe("CheckpointReactor", () => { ); }); - it("continues processing runtime events after a single checkpoint runtime failure", async () => { + it("captures completed turns in a non-Git workspace and continues processing", async () => { const nonRepositorySessionCwd = NodeFS.mkdtempSync( NodePath.join(NodeOS.tmpdir(), "t3-checkpoint-runtime-non-repo-"), ); @@ -879,12 +973,12 @@ describe("CheckpointReactor", () => { turnId: asTurnId("turn-after-runtime-failure"), }); - await waitForGitRefExists( - harness.cwd, - checkpointRefForThreadTurn(ThreadId.make("thread-1"), 0), + const thread = await waitForThread(harness.readModel, (entry) => + entry.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 1), ); + expect(thread.checkpoints).toHaveLength(1); expect( - gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.make("thread-1"), 0)), + thread.activities.some((activity) => activity.kind === "checkpoint.capture.failed"), ).toBe(true); }); @@ -969,6 +1063,247 @@ describe("CheckpointReactor", () => { ).toBe(false); }); + it("reverts legacy predecessor turns in mixed checkpoint history", async () => { + const harness = await createHarness(); + const threadId = ThreadId.make("thread-1"); + const createdAt = "2026-01-01T00:00:00.000Z"; + + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-legacy-session-set"), + threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }), + ); + + for (const turnNumber of [1, 2, 3]) { + const turnId = asTurnId(`legacy-turn-${turnNumber}`); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`cmd-legacy-user-${turnNumber}`), + threadId, + message: { + messageId: MessageId.make(`legacy-user-${turnNumber}`), + role: "user", + text: `User ${turnNumber}`, + attachments: [], + }, + runtimeMode: "approval-required", + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + createdAt, + }), + ); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.message.assistant.delta", + commandId: CommandId.make(`cmd-legacy-assistant-delta-${turnNumber}`), + threadId, + messageId: MessageId.make(`legacy-assistant-${turnNumber}`), + delta: `Assistant ${turnNumber}`, + turnId, + createdAt, + }), + ); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.message.assistant.complete", + commandId: CommandId.make(`cmd-legacy-assistant-complete-${turnNumber}`), + threadId, + messageId: MessageId.make(`legacy-assistant-${turnNumber}`), + turnId, + createdAt, + }), + ); + } + + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.turn.diff.complete", + commandId: CommandId.make("cmd-mixed-checkpoint"), + threadId, + turnId: asTurnId("legacy-turn-3"), + completedAt: createdAt, + checkpointRef: checkpointRefForThreadTurn(threadId, 1), + status: "ready", + files: [], + checkpointTurnCount: 1, + createdAt, + }), + ); + + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.checkpoint.revert", + commandId: CommandId.make("cmd-legacy-revert-request"), + threadId, + turnCount: 2, + createdAt, + }), + ); + + await waitForEvent(harness.engine, (event) => event.type === "thread.reverted"); + expect(harness.provider.rollbackConversation).toHaveBeenCalledWith({ + threadId, + numTurns: 1, + }); + }); + + it("rewinds legacy conversation history when no filesystem checkpoints exist", async () => { + const harness = await createHarness({ seedFilesystemCheckpoints: false }); + const threadId = ThreadId.make("thread-1"); + const createdAt = "2026-01-01T00:00:00.000Z"; + + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-legacy-only-session-set"), + threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }), + ); + + for (const turnNumber of [1, 2]) { + const turnId = asTurnId(`legacy-only-turn-${turnNumber}`); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`cmd-legacy-only-user-${turnNumber}`), + threadId, + message: { + messageId: MessageId.make(`legacy-only-user-${turnNumber}`), + role: "user", + text: `User ${turnNumber}`, + attachments: [], + }, + runtimeMode: "approval-required", + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + createdAt, + }), + ); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.message.assistant.delta", + commandId: CommandId.make(`cmd-legacy-only-assistant-delta-${turnNumber}`), + threadId, + messageId: MessageId.make(`legacy-only-assistant-${turnNumber}`), + delta: `Assistant ${turnNumber}`, + turnId, + createdAt, + }), + ); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.message.assistant.complete", + commandId: CommandId.make(`cmd-legacy-only-assistant-complete-${turnNumber}`), + threadId, + messageId: MessageId.make(`legacy-only-assistant-${turnNumber}`), + turnId, + createdAt, + }), + ); + } + + runGit(harness.cwd, ["update-ref", "-d", checkpointRefForThreadTurn(threadId, 1)]); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.checkpoint.revert", + commandId: CommandId.make("cmd-legacy-only-revert"), + threadId, + turnCount: 1, + createdAt, + }), + ); + + await waitForEvent(harness.engine, (event) => event.type === "thread.reverted"); + expect(harness.provider.rollbackConversation).toHaveBeenCalledWith({ + threadId, + numTurns: 1, + }); + }); + + it("does not rewind conversation when a shadow filesystem checkpoint is missing", async () => { + const workspace = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-shadow-revert-")); + tempDirs.push(workspace); + NodeFS.writeFileSync(NodePath.join(workspace, "keep.txt"), "unchanged\n", "utf8"); + const harness = await createHarness({ + seedFilesystemCheckpoints: false, + projectWorkspaceRoot: workspace, + threadWorktreePath: workspace, + providerSessionCwd: workspace, + }); + const threadId = ThreadId.make("thread-1"); + const createdAt = "2026-01-01T00:00:00.000Z"; + + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-shadow-missing-session-set"), + threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }), + ); + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.turn.diff.complete", + commandId: CommandId.make("cmd-shadow-missing-checkpoint-summary"), + threadId, + turnId: asTurnId("shadow-missing-turn"), + completedAt: createdAt, + checkpointRef: checkpointRefForThreadTurn(threadId, 1), + status: "ready", + files: [], + checkpointTurnCount: 1, + createdAt, + }), + ); + + await harness.runPromise( + harness.engine.dispatch({ + type: "thread.checkpoint.revert", + commandId: CommandId.make("cmd-shadow-missing-revert"), + threadId, + turnCount: 0, + createdAt, + }), + ); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some((activity) => activity.kind === "checkpoint.revert.failed"), + ); + expect(thread.checkpoints).toHaveLength(1); + expect(harness.provider.rollbackConversation).not.toHaveBeenCalled(); + expect(NodeFS.readFileSync(NodePath.join(workspace, "keep.txt"), "utf8")).toBe("unchanged\n"); + }); + it("executes provider revert and emits thread.reverted for claude sessions", async () => { const harness = await createHarness({ providerName: ProviderDriverKind.make("claudeAgent") }); const createdAt = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 3ba244ddf2c..c73d159143d 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -32,7 +32,6 @@ import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts" import { RuntimeReceiptBus } from "../Services/RuntimeReceiptBus.ts"; import type { CheckpointStoreError } from "../../checkpointing/Errors.ts"; import type { OrchestrationDispatchError } from "../Errors.ts"; -import { isGitRepository } from "../../git/Utils.ts"; import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts"; import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts"; @@ -59,6 +58,39 @@ function sameId(left: string | null | undefined, right: string | null | undefine return left === right; } +export function countConversationTurns(messages: ReadonlyArray<{ readonly role: string }>): number { + return messages.reduce((count, message) => count + (message.role === "user" ? 1 : 0), 0); +} + +export function conversationTurnCountForTurn( + messages: ReadonlyArray<{ + readonly role: string; + readonly turnId: TurnId | null; + }>, + turnId: TurnId, +): number | undefined { + let turnCount = 0; + let matchedTurnCount: number | undefined; + for (const message of messages) { + if (message.role === "user") { + turnCount += 1; + } + if (message.role === "assistant" && sameId(message.turnId, turnId)) { + matchedTurnCount = turnCount > 0 ? turnCount : undefined; + } + } + return matchedTurnCount; +} + +function maxCheckpointTurnCount( + checkpoints: ReadonlyArray<{ readonly checkpointTurnCount: number }>, +): number { + return checkpoints.reduce( + (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount), + 0, + ); +} + function checkpointStatusFromRuntime(status: string | undefined): "ready" | "missing" | "error" { switch (status) { case "failed": @@ -174,12 +206,10 @@ const make = Effect.gen(function* () { return project ? [project] : []; }); - const isGitWorkspace = (cwd: string) => isGitRepository(cwd); - // Resolves the workspace CWD for checkpoint operations, preferring the // active provider session CWD and falling back to the thread/project config. - // Returns undefined when no CWD can be determined or the workspace is not - // a git repository. + // CheckpointStore supports both user Git repositories and T3-managed shadow + // repositories, so a project does not need to initialize Git first. const resolveCheckpointCwd = Effect.fn("resolveCheckpointCwd")(function* (input: { readonly threadId: ThreadId; readonly thread: { readonly projectId: ProjectId; readonly worktreePath: string | null }; @@ -206,9 +236,6 @@ const make = Effect.gen(function* () { if (!cwd) { return undefined; } - if (!isGitWorkspace(cwd)) { - return undefined; - } return cwd; }); @@ -235,10 +262,21 @@ const make = Effect.gen(function* () { const fromCheckpointRef = checkpointRefForThreadTurn(input.threadId, fromTurnCount); const targetCheckpointRef = checkpointRefForThreadTurn(input.threadId, input.turnCount); - const fromCheckpointExists = yield* checkpointStore.hasCheckpointRef({ - cwd: input.cwd, - checkpointRef: fromCheckpointRef, - }); + const fromCheckpointExists = yield* checkpointStore + .hasCheckpointRef({ + cwd: input.cwd, + checkpointRef: fromCheckpointRef, + }) + .pipe( + Effect.catchTags({ + VcsProcessExitError: (error) => + Effect.logWarning("previous checkpoint ref probe failed", { + checkpointRef: fromCheckpointRef, + cwd: input.cwd, + detail: error.message, + }).pipe(Effect.as(false)), + }), + ); if (!fromCheckpointExists) { yield* Effect.logWarning("checkpoint capture missing pre-turn baseline", { threadId: input.threadId, @@ -393,13 +431,10 @@ const make = Effect.gen(function* () { const existingPlaceholder = thread.checkpoints.find( (checkpoint) => checkpoint.turnId === turnId && checkpoint.status === "missing", ); - const currentTurnCount = thread.checkpoints.reduce( - (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount), - 0, - ); + const currentTurnCount = maxCheckpointTurnCount(thread.checkpoints); const nextTurnCount = existingPlaceholder ? existingPlaceholder.checkpointTurnCount - : currentTurnCount + 1; + : (conversationTurnCountForTurn(thread.messages, turnId) ?? currentTurnCount + 1); yield* captureAndDispatchCheckpoint({ threadId: thread.id, @@ -499,9 +534,9 @@ const make = Effect.gen(function* () { return; } - const currentTurnCount = thread.checkpoints.reduce( - (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount), - 0, + const currentTurnCount = Math.max( + maxCheckpointTurnCount(thread.checkpoints), + Math.max(0, countConversationTurns(thread.messages) - 1), ); const baselineCheckpointRef = checkpointRefForThreadTurn(thread.id, currentTurnCount); const baselineExists = yield* checkpointStore.hasCheckpointRef({ @@ -581,9 +616,13 @@ const make = Effect.gen(function* () { return; } - const currentTurnCount = thread.checkpoints.reduce( - (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount), - 0, + const projectedConversationTurnCount = countConversationTurns(thread.messages); + const currentTurnCount = Math.max( + maxCheckpointTurnCount(thread.checkpoints), + event.type === "thread.message-sent" && + thread.messages.some((message) => message.id === event.payload.messageId) + ? Math.max(0, projectedConversationTurnCount - 1) + : projectedConversationTurnCount, ); const baselineCheckpointRef = checkpointRefForThreadTurn(threadId, currentTurnCount); const baselineExists = yield* checkpointStore.hasCheckpointRef({ @@ -633,20 +672,8 @@ const make = Effect.gen(function* () { }).pipe(Effect.catch(() => Effect.void)); return; } - if (!isGitWorkspace(sessionRuntime.value.cwd)) { - yield* appendRevertFailureActivity({ - threadId: event.payload.threadId, - turnCount: event.payload.turnCount, - detail: "Checkpoints are unavailable because this project is not a git repository.", - createdAt: now, - }).pipe(Effect.catch(() => Effect.void)); - return; - } - - const currentTurnCount = thread.checkpoints.reduce( - (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount), - 0, - ); + const checkpointTurnCount = maxCheckpointTurnCount(thread.checkpoints); + const currentTurnCount = Math.max(checkpointTurnCount, countConversationTurns(thread.messages)); if (event.payload.turnCount > currentTurnCount) { yield* appendRevertFailureActivity({ @@ -658,42 +685,37 @@ const make = Effect.gen(function* () { return; } - const targetCheckpointRef = - event.payload.turnCount === 0 - ? checkpointRefForThreadTurn(event.payload.threadId, 0) - : thread.checkpoints.find( - (checkpoint) => checkpoint.checkpointTurnCount === event.payload.turnCount, - )?.checkpointRef; - - if (!targetCheckpointRef) { - yield* appendRevertFailureActivity({ - threadId: event.payload.threadId, - turnCount: event.payload.turnCount, - detail: `Checkpoint ref for turn ${event.payload.turnCount} is unavailable in read model.`, - createdAt: now, - }).pipe(Effect.catch(() => Effect.void)); - return; - } + // Legacy/imported threads can predate filesystem checkpoints entirely. They + // still support conversation-only rewind. Once a thread has checkpoint + // summaries, however, fail closed if the matching filesystem state cannot + // be restored so conversation and workspace never diverge silently. + if (checkpointTurnCount > 0) { + const summarizedCheckpoint = thread.checkpoints.find( + (checkpoint) => checkpoint.checkpointTurnCount === event.payload.turnCount, + ); + const targetCheckpointRef = + summarizedCheckpoint?.checkpointRef ?? + checkpointRefForThreadTurn(event.payload.threadId, event.payload.turnCount); + const restored = yield* checkpointStore.restoreCheckpoint({ + cwd: sessionRuntime.value.cwd, + checkpointRef: targetCheckpointRef, + fallbackToHead: event.payload.turnCount === 0, + }); + if (!restored) { + yield* appendRevertFailureActivity({ + threadId: event.payload.threadId, + turnCount: event.payload.turnCount, + detail: `Filesystem checkpoint is unavailable for turn ${event.payload.turnCount}. Conversation history was not changed.`, + createdAt: now, + }).pipe(Effect.catch(() => Effect.void)); + return; + } - const restored = yield* checkpointStore.restoreCheckpoint({ - cwd: sessionRuntime.value.cwd, - checkpointRef: targetCheckpointRef, - fallbackToHead: event.payload.turnCount === 0, - }); - if (!restored) { - yield* appendRevertFailureActivity({ - threadId: event.payload.threadId, - turnCount: event.payload.turnCount, - detail: `Filesystem checkpoint is unavailable for turn ${event.payload.turnCount}.`, - createdAt: now, - }).pipe(Effect.catch(() => Effect.void)); - return; + // Refresh the workspace entry index so the @-mention file picker reflects + // the reverted filesystem state before conversation history is rewound. + yield* workspaceEntries.refresh(sessionRuntime.value.cwd); } - // Refresh the workspace entry index so the @-mention file picker - // reflects the reverted filesystem state. - yield* workspaceEntries.refresh(sessionRuntime.value.cwd); - const rolledBackTurns = Math.max(0, currentTurnCount - event.payload.turnCount); if (rolledBackTurns > 0) { yield* providerService.rollbackConversation({ diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 0999000ed4f..d5f43edec88 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -788,6 +788,28 @@ it.layer( }, }); + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.make("evt-revert-files-user-keep"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-01-01T00:00:00.100Z", + commandId: CommandId.make("cmd-revert-files-user-keep"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-files-user-keep"), + metadata: {}, + payload: { + threadId, + messageId: MessageId.make("message-user-keep"), + role: "user", + text: "Keep this turn", + turnId: null, + streaming: false, + createdAt: "2026-01-01T00:00:00.100Z", + updatedAt: "2026-01-01T00:00:00.100Z", + }, + }); + yield* appendAndProject({ type: "thread.turn-diff-completed", eventId: EventId.make("evt-revert-files-3"), @@ -836,8 +858,8 @@ it.layer( ], turnId: TurnId.make("turn-keep"), streaming: false, - createdAt: now, - updatedAt: now, + createdAt: "2026-01-01T00:00:00.200Z", + updatedAt: "2026-01-01T00:00:00.200Z", }, }); @@ -863,6 +885,28 @@ it.layer( }, }); + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.make("evt-revert-files-user-remove"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-01-01T00:00:00.300Z", + commandId: CommandId.make("cmd-revert-files-user-remove"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-files-user-remove"), + metadata: {}, + payload: { + threadId, + messageId: MessageId.make("message-user-remove"), + role: "user", + text: "Revert here", + turnId: null, + streaming: false, + createdAt: "2026-01-01T00:00:00.300Z", + updatedAt: "2026-01-01T00:00:00.300Z", + }, + }); + yield* appendAndProject({ type: "thread.message-sent", eventId: EventId.make("evt-revert-files-6"), @@ -889,8 +933,8 @@ it.layer( ], turnId: TurnId.make("turn-remove"), streaming: false, - createdAt: now, - updatedAt: now, + createdAt: "2026-01-01T00:00:00.400Z", + updatedAt: "2026-01-01T00:00:00.400Z", }, }); @@ -2193,7 +2237,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); - it.effect("does not fallback-retain messages whose turnId is removed by revert", () => + it.effect("preserves an uncheckpointed predecessor when reverting mixed history", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; @@ -2251,24 +2295,24 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }); yield* appendAndProject({ - type: "thread.turn-diff-completed", - eventId: EventId.make("evt-revert-3"), + type: "thread.message-sent", + eventId: EventId.make("evt-revert-user-keep"), aggregateKind: "thread", aggregateId: ThreadId.make("thread-revert"), - occurredAt: "2026-02-26T12:00:02.000Z", - commandId: CommandId.make("cmd-revert-3"), + occurredAt: "2026-02-26T12:00:02.050Z", + commandId: CommandId.make("cmd-revert-user-keep"), causationEventId: null, - correlationId: CorrelationId.make("cmd-revert-3"), + correlationId: CorrelationId.make("cmd-revert-user-keep"), metadata: {}, payload: { threadId: ThreadId.make("thread-revert"), - turnId: TurnId.make("turn-1"), - checkpointTurnCount: 1, - checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-revert/turn/1"), - status: "ready", - files: [], - assistantMessageId: MessageId.make("assistant-keep"), - completedAt: "2026-02-26T12:00:02.000Z", + messageId: MessageId.make("user-keep"), + role: "user", + text: "start", + turnId: null, + streaming: false, + createdAt: "2026-02-26T12:00:02.050Z", + updatedAt: "2026-02-26T12:00:02.050Z", }, }); @@ -2307,15 +2351,63 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { payload: { threadId: ThreadId.make("thread-revert"), turnId: TurnId.make("turn-2"), - checkpointTurnCount: 2, + checkpointTurnCount: 1, checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-revert/turn/2"), status: "ready", files: [], - assistantMessageId: MessageId.make("assistant-remove"), + assistantMessageId: null, completedAt: "2026-02-26T12:00:03.000Z", }, }); + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-revert-checkpoint-activity"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-revert"), + occurredAt: "2026-02-26T12:00:03.010Z", + commandId: CommandId.make("cmd-revert-checkpoint-activity"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-checkpoint-activity"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-revert"), + activity: { + id: EventId.make("checkpoint-only-activity"), + tone: "tool", + kind: "command", + summary: "Retained checkpoint activity", + payload: {}, + turnId: TurnId.make("turn-2"), + createdAt: "2026-02-26T12:00:03.010Z", + }, + }, + }); + + yield* appendAndProject({ + type: "thread.proposed-plan-upserted", + eventId: EventId.make("evt-revert-checkpoint-plan"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-revert"), + occurredAt: "2026-02-26T12:00:03.020Z", + commandId: CommandId.make("cmd-revert-checkpoint-plan"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-checkpoint-plan"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-revert"), + proposedPlan: { + id: "checkpoint-only-plan", + turnId: TurnId.make("turn-2"), + planMarkdown: "## Retained checkpoint plan", + implementedAt: null, + implementationThreadId: null, + createdAt: "2026-02-26T12:00:03.020Z", + updatedAt: "2026-02-26T12:00:03.020Z", + }, + }, + }); + yield* appendAndProject({ type: "thread.message-sent", eventId: EventId.make("evt-revert-6"), @@ -2331,7 +2423,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { messageId: MessageId.make("user-remove"), role: "user", text: "removed", - turnId: TurnId.make("turn-2"), + turnId: null, streaming: false, createdAt: "2026-02-26T12:00:03.050Z", updatedAt: "2026-02-26T12:00:03.050Z", @@ -2353,7 +2445,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { messageId: MessageId.make("assistant-remove"), role: "assistant", text: "removed", - turnId: TurnId.make("turn-2"), + turnId: TurnId.make("turn-1"), streaming: false, createdAt: "2026-02-26T12:00:03.100Z", updatedAt: "2026-02-26T12:00:03.100Z", @@ -2390,12 +2482,44 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { ORDER BY created_at ASC, message_id ASC `; assert.deepEqual(messageRows, [ + { + messageId: "user-keep", + turnId: null, + role: "user", + }, { messageId: "assistant-keep", turnId: "turn-1", role: "assistant", }, + { + messageId: "user-remove", + turnId: null, + role: "user", + }, ]); + + const turnRows = yield* sql<{ readonly turnId: string }>` + SELECT turn_id AS "turnId" + FROM projection_turns + WHERE thread_id = 'thread-revert' + ORDER BY requested_at ASC, turn_id ASC + `; + assert.deepEqual(turnRows, [{ turnId: "turn-1" }, { turnId: "turn-2" }]); + + const activityRows = yield* sql<{ readonly activityId: string }>` + SELECT activity_id AS "activityId" + FROM projection_thread_activities + WHERE thread_id = 'thread-revert' + `; + assert.deepEqual(activityRows, [{ activityId: "checkpoint-only-activity" }]); + + const planRows = yield* sql<{ readonly planId: string }>` + SELECT plan_id AS "planId" + FROM projection_thread_proposed_plans + WHERE thread_id = 'thread-revert' + `; + assert.deepEqual(planRows, [{ planId: "checkpoint-only-plan" }]); }), ); }); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index f12df850941..8f041cab2f6 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -205,86 +205,47 @@ function deriveHasActionableProposedPlan(input: { function retainProjectionMessagesAfterRevert( messages: ReadonlyArray, - turns: ReadonlyArray, turnCount: number, ): ReadonlyArray { - const retainedMessageIds = new Set(); - const retainedTurnIds = new Set(); - const keptTurns = turns.filter( - (turn) => - turn.turnId !== null && - turn.checkpointTurnCount !== null && - turn.checkpointTurnCount <= turnCount, - ); - for (const turn of keptTurns) { - if (turn.turnId !== null) { - retainedTurnIds.add(turn.turnId); - } - if (turn.pendingMessageId !== null) { - retainedMessageIds.add(turn.pendingMessageId); - } - if (turn.assistantMessageId !== null) { - retainedMessageIds.add(turn.assistantMessageId); - } - } - - for (const message of messages) { + let userMessageIndex = 0; + let reachedTargetUser = false; + return messages.filter((message) => { if (message.role === "system") { - retainedMessageIds.add(message.messageId); - continue; + return true; } - if (message.turnId !== null && retainedTurnIds.has(message.turnId)) { - retainedMessageIds.add(message.messageId); + if (reachedTargetUser) { + return false; } - } - - const retainedUserCount = messages.filter( - (message) => message.role === "user" && retainedMessageIds.has(message.messageId), - ).length; - const missingUserCount = Math.max(0, turnCount - retainedUserCount); - if (missingUserCount > 0) { - const fallbackUserMessages = messages - .filter( - (message) => - message.role === "user" && - !retainedMessageIds.has(message.messageId) && - (message.turnId === null || retainedTurnIds.has(message.turnId)), - ) - .toSorted( - (left, right) => - left.createdAt.localeCompare(right.createdAt) || - left.messageId.localeCompare(right.messageId), - ) - .slice(0, missingUserCount); - for (const message of fallbackUserMessages) { - retainedMessageIds.add(message.messageId); + if (message.role === "user") { + reachedTargetUser = userMessageIndex === turnCount; + userMessageIndex += 1; } - } + return true; + }); +} - const retainedAssistantCount = messages.filter( - (message) => message.role === "assistant" && retainedMessageIds.has(message.messageId), - ).length; - const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount); - if (missingAssistantCount > 0) { - const fallbackAssistantMessages = messages - .filter( - (message) => - message.role === "assistant" && - !retainedMessageIds.has(message.messageId) && - (message.turnId === null || retainedTurnIds.has(message.turnId)), - ) - .toSorted( - (left, right) => - left.createdAt.localeCompare(right.createdAt) || - left.messageId.localeCompare(right.messageId), +function retainProjectionTurnsAfterRevert( + turns: ReadonlyArray, + turnCount: number, +): ReadonlyArray { + const sortedTurns = turns + .filter((turn) => turn.turnId !== null) + .toSorted( + (left, right) => + left.requestedAt.localeCompare(right.requestedAt) || + (left.turnId ?? "").localeCompare(right.turnId ?? ""), + ); + const retainedTurnIds = new Set( + sortedTurns + .slice(0, turnCount) + .concat( + sortedTurns.filter( + (turn) => turn.checkpointTurnCount !== null && turn.checkpointTurnCount <= turnCount, + ), ) - .slice(0, missingAssistantCount); - for (const message of fallbackAssistantMessages) { - retainedMessageIds.add(message.messageId); - } - } - - return messages.filter((message) => retainedMessageIds.has(message.messageId)); + .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])), + ); + return sortedTurns.filter((turn) => turn.turnId !== null && retainedTurnIds.has(turn.turnId)); } function retainProjectionActivitiesAfterRevert( @@ -293,14 +254,9 @@ function retainProjectionActivitiesAfterRevert( turnCount: number, ): ReadonlyArray { const retainedTurnIds = new Set( - turns - .filter( - (turn) => - turn.turnId !== null && - turn.checkpointTurnCount !== null && - turn.checkpointTurnCount <= turnCount, - ) - .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])), + retainProjectionTurnsAfterRevert(turns, turnCount).flatMap((turn) => + turn.turnId === null ? [] : [turn.turnId], + ), ); return activities.filter( (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId), @@ -313,14 +269,9 @@ function retainProjectionProposedPlansAfterRevert( turnCount: number, ): ReadonlyArray { const retainedTurnIds = new Set( - turns - .filter( - (turn) => - turn.turnId !== null && - turn.checkpointTurnCount !== null && - turn.checkpointTurnCount <= turnCount, - ) - .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])), + retainProjectionTurnsAfterRevert(turns, turnCount).flatMap((turn) => + turn.turnId === null ? [] : [turn.turnId], + ), ); return proposedPlans.filter( (proposedPlan) => proposedPlan.turnId === null || retainedTurnIds.has(proposedPlan.turnId), @@ -772,19 +723,22 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } - const retainedTurns = yield* projectionTurnRepository.listByThreadId({ + const existingTurns = yield* projectionTurnRepository.listByThreadId({ threadId: event.payload.threadId, }); + const retainedTurns = retainProjectionTurnsAfterRevert( + existingTurns, + event.payload.turnCount, + ); let latestTurnId: ProjectionTurn["turnId"] = null; let latestCheckpointTurnCount = -1; for (let index = 0; index < retainedTurns.length; index += 1) { const turn = retainedTurns[index]; - if ( - !turn || - turn.turnId === null || - turn.checkpointTurnCount === null || - turn.checkpointTurnCount > event.payload.turnCount - ) { + if (!turn || turn.turnId === null) { + continue; + } + if (turn.checkpointTurnCount === null) { + latestTurnId = turn.turnId; continue; } if (turn.checkpointTurnCount > latestCheckpointTurnCount) { @@ -856,12 +810,8 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } - const existingTurns = yield* projectionTurnRepository.listByThreadId({ - threadId: event.payload.threadId, - }); const keptRows = retainProjectionMessagesAfterRevert( existingRows, - existingTurns, event.payload.turnCount, ); if (keptRows.length === existingRows.length) { @@ -1305,11 +1255,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const existingTurns = yield* projectionTurnRepository.listByThreadId({ threadId: event.payload.threadId, }); - const keptTurns = existingTurns.filter( - (turn) => - turn.turnId !== null && - turn.checkpointTurnCount !== null && - turn.checkpointTurnCount <= event.payload.turnCount, + const keptTurns = retainProjectionTurnsAfterRevert( + existingTurns, + event.payload.turnCount, ); yield* projectionTurnRepository.deleteByThreadId({ threadId: event.payload.threadId, diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index fadd5078026..dc1358b7dd6 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -484,7 +484,7 @@ describe("orchestration projector", () => { expect(message?.updatedAt).toBe(completeAt); }); - it("prunes reverted turn messages from in-memory thread snapshot", async () => { + it("preserves an uncheckpointed predecessor when reverting mixed history", async () => { const createdAt = "2026-02-23T10:00:00.000Z"; const model = createEmptyReadModel(createdAt); @@ -555,24 +555,6 @@ describe("orchestration projector", () => { }), makeEvent({ sequence: 4, - type: "thread.turn-diff-completed", - aggregateKind: "thread", - aggregateId: "thread-1", - occurredAt: "2026-02-23T10:00:02.500Z", - commandId: "cmd-turn-1-complete", - payload: { - threadId: "thread-1", - turnId: "turn-1", - checkpointTurnCount: 1, - checkpointRef: "refs/t3/checkpoints/thread-1/turn/1", - status: "ready", - files: [], - assistantMessageId: "assistant-msg-1", - completedAt: "2026-02-23T10:00:02.500Z", - }, - }), - makeEvent({ - sequence: 5, type: "thread.activity-appended", aggregateKind: "thread", aggregateId: "thread-1", @@ -592,7 +574,7 @@ describe("orchestration projector", () => { }, }), makeEvent({ - sequence: 6, + sequence: 5, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", @@ -610,7 +592,7 @@ describe("orchestration projector", () => { }, }), makeEvent({ - sequence: 7, + sequence: 6, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-1", @@ -628,7 +610,7 @@ describe("orchestration projector", () => { }, }), makeEvent({ - sequence: 8, + sequence: 7, type: "thread.turn-diff-completed", aggregateKind: "thread", aggregateId: "thread-1", @@ -646,7 +628,7 @@ describe("orchestration projector", () => { }, }), makeEvent({ - sequence: 9, + sequence: 8, type: "thread.activity-appended", aggregateKind: "thread", aggregateId: "thread-1", @@ -666,7 +648,7 @@ describe("orchestration projector", () => { }, }), makeEvent({ - sequence: 10, + sequence: 9, type: "thread.reverted", aggregateKind: "thread", aggregateId: "thread-1", @@ -690,16 +672,17 @@ describe("orchestration projector", () => { [ { role: "user", text: "First edit" }, { role: "assistant", text: "Updated README to v2.\n" }, + { role: "user", text: "Second edit" }, ], ); expect( thread?.activities.map((activity) => ({ id: activity.id, turnId: activity.turnId })), ).toEqual([{ id: "activity-1", turnId: "turn-1" }]); - expect(thread?.checkpoints.map((checkpoint) => checkpoint.checkpointTurnCount)).toEqual([1]); - expect(thread?.latestTurn?.turnId).toBe("turn-1"); + expect(thread?.checkpoints).toEqual([]); + expect(thread?.latestTurn).toBeNull(); }); - it("does not fallback-retain messages tied to removed turn IDs", async () => { + it("removes assistant output produced after the selected steering message", async () => { const createdAt = "2026-02-26T12:00:00.000Z"; const model = createEmptyReadModel(createdAt); @@ -734,6 +717,24 @@ describe("orchestration projector", () => { const events: ReadonlyArray = [ makeEvent({ sequence: 2, + type: "thread.message-sent", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:00.500Z", + commandId: "cmd-user-keep", + payload: { + threadId: "thread-revert", + messageId: "user-keep", + role: "user", + text: "start", + turnId: null, + streaming: false, + createdAt: "2026-02-26T12:00:00.500Z", + updatedAt: "2026-02-26T12:00:00.500Z", + }, + }), + makeEvent({ + sequence: 3, type: "thread.turn-diff-completed", aggregateKind: "thread", aggregateId: "thread-revert", @@ -741,17 +742,17 @@ describe("orchestration projector", () => { commandId: "cmd-turn-1", payload: { threadId: "thread-revert", - turnId: "turn-1", + turnId: "checkpoint-only-turn", checkpointTurnCount: 1, checkpointRef: "refs/t3/checkpoints/thread-revert/turn/1", status: "ready", files: [], - assistantMessageId: "assistant-keep", + assistantMessageId: null, completedAt: "2026-02-26T12:00:01.000Z", }, }), makeEvent({ - sequence: 3, + sequence: 4, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-revert", @@ -769,7 +770,47 @@ describe("orchestration projector", () => { }, }), makeEvent({ - sequence: 4, + sequence: 5, + type: "thread.activity-appended", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:01.200Z", + commandId: "cmd-checkpoint-only-activity", + payload: { + threadId: "thread-revert", + activity: { + id: "checkpoint-only-activity", + tone: "tool", + kind: "command", + summary: "Retained activity", + payload: {}, + turnId: "checkpoint-only-turn", + createdAt: "2026-02-26T12:00:01.200Z", + }, + }, + }), + makeEvent({ + sequence: 6, + type: "thread.proposed-plan-upserted", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:01.300Z", + commandId: "cmd-checkpoint-only-plan", + payload: { + threadId: "thread-revert", + proposedPlan: { + id: "checkpoint-only-plan", + turnId: "checkpoint-only-turn", + planMarkdown: "## Retained plan", + implementedAt: null, + implementationThreadId: null, + createdAt: "2026-02-26T12:00:01.300Z", + updatedAt: "2026-02-26T12:00:01.300Z", + }, + }, + }), + makeEvent({ + sequence: 7, type: "thread.turn-diff-completed", aggregateKind: "thread", aggregateId: "thread-revert", @@ -782,12 +823,12 @@ describe("orchestration projector", () => { checkpointRef: "refs/t3/checkpoints/thread-revert/turn/2", status: "ready", files: [], - assistantMessageId: "assistant-remove", + assistantMessageId: null, completedAt: "2026-02-26T12:00:02.000Z", }, }), makeEvent({ - sequence: 5, + sequence: 8, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-revert", @@ -798,14 +839,14 @@ describe("orchestration projector", () => { messageId: "user-remove", role: "user", text: "removed", - turnId: "turn-2", + turnId: null, streaming: false, createdAt: "2026-02-26T12:00:02.050Z", updatedAt: "2026-02-26T12:00:02.050Z", }, }), makeEvent({ - sequence: 6, + sequence: 9, type: "thread.message-sent", aggregateKind: "thread", aggregateId: "thread-revert", @@ -816,14 +857,14 @@ describe("orchestration projector", () => { messageId: "assistant-remove", role: "assistant", text: "removed", - turnId: "turn-2", + turnId: "turn-1", streaming: false, createdAt: "2026-02-26T12:00:02.100Z", updatedAt: "2026-02-26T12:00:02.100Z", }, }), makeEvent({ - sequence: 7, + sequence: 10, type: "thread.reverted", aggregateKind: "thread", aggregateId: "thread-revert", @@ -849,7 +890,13 @@ describe("orchestration projector", () => { role: message.role, turnId: message.turnId, })), - ).toEqual([{ id: "assistant-keep", role: "assistant", turnId: "turn-1" }]); + ).toEqual([ + { id: "user-keep", role: "user", turnId: null }, + { id: "assistant-keep", role: "assistant", turnId: "turn-1" }, + { id: "user-remove", role: "user", turnId: null }, + ]); + expect(thread?.activities.map((activity) => activity.id)).toEqual(["checkpoint-only-activity"]); + expect(thread?.proposedPlans.map((plan) => plan.id)).toEqual(["checkpoint-only-plan"]); }); it("caps message and checkpoint retention for long-lived threads", async () => { diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index fc6ab8f6fcf..75f98094bae 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -82,65 +82,23 @@ function decodeForEvent( function retainThreadMessagesAfterRevert( messages: ReadonlyArray, - retainedTurnIds: ReadonlySet, turnCount: number, ): ReadonlyArray { - const retainedMessageIds = new Set(); - for (const message of messages) { + let userMessageIndex = 0; + let reachedTargetUser = false; + return messages.filter((message) => { if (message.role === "system") { - retainedMessageIds.add(message.id); - continue; + return true; } - if (message.turnId !== null && retainedTurnIds.has(message.turnId)) { - retainedMessageIds.add(message.id); + if (reachedTargetUser) { + return false; } - } - - const retainedUserCount = messages.filter( - (message) => message.role === "user" && retainedMessageIds.has(message.id), - ).length; - const missingUserCount = Math.max(0, turnCount - retainedUserCount); - if (missingUserCount > 0) { - const fallbackUserMessages = messages - .filter( - (message) => - message.role === "user" && - !retainedMessageIds.has(message.id) && - (message.turnId === null || retainedTurnIds.has(message.turnId)), - ) - .toSorted( - (left, right) => - left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id), - ) - .slice(0, missingUserCount); - for (const message of fallbackUserMessages) { - retainedMessageIds.add(message.id); + if (message.role === "user") { + reachedTargetUser = userMessageIndex === turnCount; + userMessageIndex += 1; } - } - - const retainedAssistantCount = messages.filter( - (message) => message.role === "assistant" && retainedMessageIds.has(message.id), - ).length; - const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount); - if (missingAssistantCount > 0) { - const fallbackAssistantMessages = messages - .filter( - (message) => - message.role === "assistant" && - !retainedMessageIds.has(message.id) && - (message.turnId === null || retainedTurnIds.has(message.turnId)), - ) - .toSorted( - (left, right) => - left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id), - ) - .slice(0, missingAssistantCount); - for (const message of fallbackAssistantMessages) { - retainedMessageIds.add(message.id); - } - } - - return messages.filter((message) => retainedMessageIds.has(message.id)); + return true; + }); } function retainThreadActivitiesAfterRevert( @@ -623,12 +581,14 @@ export function projectEvent( .filter((entry) => entry.checkpointTurnCount <= payload.turnCount) .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount) .slice(-MAX_THREAD_CHECKPOINTS); - const retainedTurnIds = new Set(checkpoints.map((checkpoint) => checkpoint.turnId)); const messages = retainThreadMessagesAfterRevert( thread.messages, - retainedTurnIds, payload.turnCount, ).slice(-MAX_THREAD_MESSAGES); + const retainedTurnIds = new Set([ + ...checkpoints.map((checkpoint) => checkpoint.turnId), + ...messages.flatMap((message) => (message.turnId === null ? [] : [message.turnId])), + ]); const proposedPlans = retainThreadProposedPlansAfterRevert( thread.proposedPlans, retainedTurnIds, diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 55aa8f38835..01c40bfbf32 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -747,6 +747,62 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( return false; } + const headExists = yield* hasHeadCommit(input.cwd); + if (!headExists) { + const gitCommonDir = yield* resolveGitCommonDir(input.cwd); + const tempIndexPath = path.join( + gitCommonDir, + `t3-checkpoint-restore-index-${NodeCrypto.randomUUID()}`, + ); + const restoreEnv: NodeJS.ProcessEnv = { + ...process.env, + GIT_INDEX_FILE: tempIndexPath, + }; + const cleanupTempIndex = fileSystem + .remove(tempIndexPath, { force: true }) + .pipe(Effect.ignore); + + yield* Effect.gen(function* () { + // The real index is empty in an unborn repository. Clean before + // restoring so untracked paths cannot block the checkout. + yield* execute({ + operation, + cwd: input.cwd, + args: ["clean", "-fd", "--", "."], + }); + const snapshotEntries = yield* execute({ + operation, + cwd: input.cwd, + args: ["ls-tree", "-r", "--name-only", commitOid], + }); + if (snapshotEntries.stdout.trim().length > 0) { + yield* execute({ + operation, + cwd: input.cwd, + args: ["restore", "--source", commitOid, "--worktree", "--", "."], + }); + } + + // Load the checkpoint tree into a temporary index before the final + // clean. Snapshot files are then protected as tracked, while paths + // ignored only by post-checkpoint rules are removed safely without -x. + yield* execute({ + operation, + cwd: input.cwd, + args: ["read-tree", commitOid], + env: restoreEnv, + }); + yield* execute({ + operation, + cwd: input.cwd, + args: ["clean", "-fd", "--", "."], + env: restoreEnv, + }); + }).pipe(Effect.ensuring(cleanupTempIndex)); + + return true; + } + yield* execute({ operation, cwd: input.cwd, @@ -757,15 +813,11 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( cwd: input.cwd, args: ["clean", "-fd", "--", "."], }); - - const headExists = yield* hasHeadCommit(input.cwd); - if (headExists) { - yield* execute({ - operation, - cwd: input.cwd, - args: ["reset", "--quiet", "--", "."], - }); - } + yield* execute({ + operation, + cwd: input.cwd, + args: ["reset", "--quiet", "--", "."], + }); return true; }), diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index bc7487cee29..d21f97a6cb6 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -1,4 +1,5 @@ import { + CheckpointRef, EnvironmentId, MessageId, ProjectId, @@ -16,6 +17,7 @@ import { buildThreadTurnInterruptInput, createLocalDispatchSnapshot, deriveComposerSendState, + deriveRevertTurnCountByUserMessageId, getStartedThreadModelChangeBlockReason, hasServerAcknowledgedLocalDispatch, reconcileMountedTerminalThreadIds, @@ -24,6 +26,101 @@ import { shouldWriteThreadErrorToCurrentServerThread, } from "./ChatView.logic"; +function timelineMessage( + id: string, + role: "user" | "assistant", + turnId: TurnId | null, + createdAt: string, +) { + return { + id: `entry-${id}`, + kind: "message" as const, + createdAt, + message: { + id: MessageId.make(id), + role, + text: id, + turnId, + createdAt, + updatedAt: createdAt, + streaming: false, + }, + }; +} + +describe("deriveRevertTurnCountByUserMessageId", () => { + const user = timelineMessage("user-1", "user", null, "2026-03-29T00:00:00.000Z"); + const assistant = timelineMessage( + "assistant-1", + "assistant", + TurnId.make("turn-1"), + "2026-03-29T00:00:01.000Z", + ); + + it("falls back to the stable turn id when the assistant message id is absent", () => { + const result = deriveRevertTurnCountByUserMessageId({ + timelineEntries: [user, assistant], + checkpoints: [ + { + turnId: TurnId.make("turn-1"), + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-1/turn/1"), + status: "ready", + files: [], + assistantMessageId: null, + completedAt: assistant.createdAt, + }, + ], + }); + + expect(result.get(user.message.id)).toBe(0); + }); + + it("uses conversation order for completed legacy turns without checkpoints", () => { + const secondUser = timelineMessage("user-2", "user", null, "2026-03-29T00:00:02.000Z"); + const secondAssistant = timelineMessage( + "assistant-2", + "assistant", + TurnId.make("turn-2"), + "2026-03-29T00:00:03.000Z", + ); + const result = deriveRevertTurnCountByUserMessageId({ + timelineEntries: [user, assistant, secondUser, secondAssistant], + checkpoints: [], + }); + + expect(result.get(user.message.id)).toBe(0); + expect(result.get(secondUser.message.id)).toBe(1); + }); + + it("uses conversation order for unmatched legacy turns in mixed checkpoint histories", () => { + const secondUser = timelineMessage("user-2", "user", null, "2026-03-29T00:00:02.000Z"); + const secondAssistant = timelineMessage( + "assistant-2", + "assistant", + TurnId.make("turn-2"), + "2026-03-29T00:00:03.000Z", + ); + const result = deriveRevertTurnCountByUserMessageId({ + timelineEntries: [user, assistant, secondUser, secondAssistant], + checkpoints: [ + { + turnId: TurnId.make("turn-2"), + checkpointTurnCount: 2, + checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-1/turn/2"), + status: "ready", + files: [], + assistantMessageId: secondAssistant.message.id, + completedAt: secondAssistant.createdAt, + }, + ], + }); + + expect(result.get(user.message.id)).toBe(0); + expect(result.get(secondUser.message.id)).toBe(1); + }); +}); + const environmentId = EnvironmentId.make("environment-local"); const projectId = ProjectId.make("project-1"); const threadId = ThreadId.make("thread-1"); diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 325f9afa90a..5dd98d6d445 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -9,7 +9,8 @@ import { type ThreadId, type TurnId, } from "@t3tools/contracts"; -import { type ChatMessage, type SessionPhase, type Thread } from "../types"; +import { type ChatMessage, type SessionPhase, type Thread, type TurnDiffSummary } from "../types"; +import type { TimelineEntry } from "../session-logic"; import { type ComposerImageAttachment, type DraftThreadState } from "../composerDraftStore"; import * as Schema from "effect/Schema"; import { appAtomRegistry } from "../rpc/atomRegistry"; @@ -27,6 +28,62 @@ export const MAX_HIDDEN_MOUNTED_PREVIEW_THREADS = 3; export const LastInvokedScriptByProjectSchema = Schema.Record(ProjectId, Schema.String); +export function deriveRevertTurnCountByUserMessageId(input: { + timelineEntries: ReadonlyArray; + checkpoints: ReadonlyArray; +}): Map { + const result = new Map(); + const checkpointByAssistantMessageId = new Map( + input.checkpoints.flatMap((checkpoint) => + checkpoint.assistantMessageId === null + ? [] + : ([[checkpoint.assistantMessageId, checkpoint]] as const), + ), + ); + const checkpointByTurnId = new Map( + input.checkpoints.map((checkpoint) => [checkpoint.turnId, checkpoint] as const), + ); + let userTurnIndex = 0; + + for (let index = 0; index < input.timelineEntries.length; index += 1) { + const entry = input.timelineEntries[index]; + if (!entry || entry.kind !== "message" || entry.message.role !== "user") { + continue; + } + + for (let nextIndex = index + 1; nextIndex < input.timelineEntries.length; nextIndex += 1) { + const nextEntry = input.timelineEntries[nextIndex]; + if (!nextEntry || nextEntry.kind !== "message") { + continue; + } + if (nextEntry.message.role === "user") { + break; + } + if (nextEntry.message.role !== "assistant") { + continue; + } + + const checkpoint = + checkpointByAssistantMessageId.get(nextEntry.message.id) ?? + (nextEntry.message.turnId === null + ? undefined + : checkpointByTurnId.get(nextEntry.message.turnId)); + if (checkpoint) { + result.set(entry.message.id, userTurnIndex); + break; + } + if (!nextEntry.message.streaming) { + result.set(entry.message.id, userTurnIndex); + break; + } + } + + userTurnIndex += 1; + } + + return result; +} + export function buildLocalDraftThread( threadId: ThreadId, draftThread: DraftThreadState, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index c296c717066..e573090a2dd 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -231,6 +231,7 @@ import { collectUserMessageBlobPreviewUrls, createLocalDispatchSnapshot, deriveComposerSendState, + deriveRevertTurnCountByUserMessageId, hasServerAcknowledgedLocalDispatch, getStartedThreadModelChangeBlockReason, LAST_INVOKED_SCRIPT_BY_PROJECT_KEY, @@ -2208,8 +2209,7 @@ function ChatViewContent(props: ChatViewProps) { attachDraftHeroComposerAnchorRef, captureDraftHeroComposerRect, ] = useDraftHeroLayoutTransition(isDraftHeroState); - const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = - useTurnDiffSummaries(activeThread); + const { turnDiffSummaries } = useTurnDiffSummaries(activeThread); const turnDiffSummaryByAssistantMessageId = useMemo(() => { const byMessageId = new Map(); for (const summary of turnDiffSummaries) { @@ -2219,37 +2219,11 @@ function ChatViewContent(props: ChatViewProps) { return byMessageId; }, [turnDiffSummaries]); const revertTurnCountByUserMessageId = useMemo(() => { - const byUserMessageId = new Map(); - for (let index = 0; index < timelineEntries.length; index += 1) { - const entry = timelineEntries[index]; - if (!entry || entry.kind !== "message" || entry.message.role !== "user") { - continue; - } - - for (let nextIndex = index + 1; nextIndex < timelineEntries.length; nextIndex += 1) { - const nextEntry = timelineEntries[nextIndex]; - if (!nextEntry || nextEntry.kind !== "message") { - continue; - } - if (nextEntry.message.role === "user") { - break; - } - const summary = turnDiffSummaryByAssistantMessageId.get(nextEntry.message.id); - if (!summary) { - continue; - } - const turnCount = - summary.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[summary.turnId]; - if (typeof turnCount !== "number") { - break; - } - byUserMessageId.set(entry.message.id, Math.max(0, turnCount - 1)); - break; - } - } - - return byUserMessageId; - }, [inferredCheckpointTurnCountByTurnId, timelineEntries, turnDiffSummaryByAssistantMessageId]); + return deriveRevertTurnCountByUserMessageId({ + timelineEntries, + checkpoints: turnDiffSummaries, + }); + }, [timelineEntries, turnDiffSummaries]); const gitCwd = activeProject ? projectScriptCwd({ @@ -3972,18 +3946,21 @@ function ChatViewContent(props: ChatViewProps) { if (activeEnvironmentUnavailable && activeEnvironmentUnavailableLabel) { setThreadError( activeThread.id, - `Reconnect ${activeEnvironmentUnavailableLabel} before reverting checkpoints.`, + `Reconnect ${activeEnvironmentUnavailableLabel} before reverting this conversation.`, ); return; } if (phase === "running" || isSendBusy || isConnecting) { - setThreadError(activeThread.id, "Interrupt the current turn before reverting checkpoints."); + setThreadError( + activeThread.id, + "Interrupt the current turn before reverting this conversation.", + ); return; } const confirmed = await localApi.dialogs.confirm( [ - `Revert this thread to checkpoint ${turnCount}?`, - "This will discard newer messages and turn diffs in this thread.", + "Revert this conversation to this message?", + "This will discard newer messages and restore workspace files when a snapshot is available.", "This action cannot be undone.", ].join("\n"), ); diff --git a/apps/web/src/components/chat/MessagesTimeline.test.tsx b/apps/web/src/components/chat/MessagesTimeline.test.tsx index b340a248fbe..7f2bc52bcd7 100644 --- a/apps/web/src/components/chat/MessagesTimeline.test.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.test.tsx @@ -271,6 +271,21 @@ describe("MessagesTimeline", () => { expect(markup).toContain("1 changed file"); }); + it("renders a disabled revert action while a user turn has no checkpoint yet", async () => { + const { MessagesTimeline } = await import("./MessagesTimeline"); + const markup = renderToStaticMarkup( + , + ); + + expect(markup).toContain('aria-label="Revert to this message"'); + expect(markup).toContain("disabled"); + }); + it("uses LegendList isNearEnd when deciding whether the live edge is visible", async () => { const { resolveTimelineIsAtEnd, diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index f759aa150be..87dee9f3a9d 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -328,6 +328,8 @@ export const MessagesTimeline = memo(function MessagesTimeline({ const [timelineViewportElement, setTimelineViewportElement] = useState( null, ); + const timelineViewportWidthRef = useRef(null); + const [timelineLayoutGeneration, setTimelineLayoutGeneration] = useState(0); const [minimapHasPersistentGutter, setMinimapHasPersistentGutter] = useState(false); const [minimapHitStripWidth, setMinimapHitStripWidth] = useState(0); const handleAnchorReady = useCallback( @@ -395,25 +397,62 @@ export const MessagesTimeline = memo(function MessagesTimeline({ return; } - const measure = () => { - const viewportWidth = timelineViewportElement.getBoundingClientRect().width; + let layoutRefreshTimeout: number | undefined; + + const measure = (forceLayoutRefresh = false) => { + const viewportWidth = Math.round(timelineViewportElement.getBoundingClientRect().width); const nextHasPersistentGutter = resolveTimelineMinimapHasPersistentGutter(viewportWidth); setMinimapHasPersistentGutter((current) => current === nextHasPersistentGutter ? current : nextHasPersistentGutter, ); setMinimapHitStripWidth(resolveTimelineMinimapHitStripWidth(viewportWidth)); + + if (viewportWidth <= 0) { + return; + } + if (timelineViewportWidthRef.current === null) { + timelineViewportWidthRef.current = + viewportWidth > 0 ? viewportWidth : timelineViewportWidthRef.current; + return; + } + if (!forceLayoutRefresh && viewportWidth === timelineViewportWidthRef.current) { + return; + } + + timelineViewportWidthRef.current = viewportWidth; + if (layoutRefreshTimeout !== undefined) { + window.clearTimeout(layoutRefreshTimeout); + } + layoutRefreshTimeout = window.setTimeout(() => { + // LegendList caches both row measurements and recycled container widths. A desktop + // window resize can briefly make the chat only a few pixels wide. Clearing sizes alone + // leaves some recycled rows one-character-wide, so remount the list after resize settles. + flushSync(() => { + setTimelineLayoutGeneration((generation) => generation + 1); + }); + }, 120); }; + const handleViewportResize = () => measure(true); - const frame = requestAnimationFrame(measure); + const frame = requestAnimationFrame(() => measure()); - const observer = new ResizeObserver(measure); + const observer = new ResizeObserver(() => measure()); observer.observe(timelineViewportElement); + window.addEventListener("resize", handleViewportResize); + window.visualViewport?.addEventListener("resize", handleViewportResize); + const pollId = window.setInterval(() => measure(), 250); return () => { cancelAnimationFrame(frame); + if (layoutRefreshTimeout !== undefined) { + window.clearTimeout(layoutRefreshTimeout); + } observer.disconnect(); + window.removeEventListener("resize", handleViewportResize); + window.visualViewport?.removeEventListener("resize", handleViewportResize); + window.clearInterval(pollId); }; - }, [timelineViewportElement, rows.length]); + }, [timelineViewportElement]); const sharedState = useMemo( () => ({ @@ -482,8 +521,13 @@ export const MessagesTimeline = memo(function MessagesTimeline({ return ( -
+
+ key={timelineLayoutGeneration} ref={listRef} data={rows} keyExtractor={keyExtractor} @@ -955,7 +999,7 @@ function UserTimelineRow({ row }: { row: Extract
- {canRevertAgentWork && } + {displayedUserMessage.copyText && ( )} @@ -966,27 +1010,37 @@ function UserTimelineRow({ row }: { row: Extract - ctx.onRevertUserMessage(messageId)} - aria-label="Revert to this message" - /> - } - > - + }> + - Revert to this message + {tooltip} ); } diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 94eb1c65370..9eeb9315b4b 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -599,6 +599,320 @@ describe("applyThreadDetailEvent", () => { }); describe("thread.reverted", () => { + it("retains earlier turns and the selected legacy user prompt", () => { + const legacyThread: OrchestrationThread = { + ...baseThread, + messages: [ + { + id: MessageId.make("legacy-user-1"), + role: "user", + text: "First", + turnId: null, + streaming: false, + createdAt: "2026-04-01T01:00:00.000Z", + updatedAt: "2026-04-01T01:00:00.000Z", + }, + { + id: MessageId.make("legacy-assistant-1"), + role: "assistant", + text: "Response 1", + turnId: TurnId.make("legacy-turn-1"), + streaming: false, + createdAt: "2026-04-01T02:00:00.000Z", + updatedAt: "2026-04-01T02:00:00.000Z", + }, + { + id: MessageId.make("legacy-user-2"), + role: "user", + text: "Second", + turnId: null, + streaming: false, + createdAt: "2026-04-01T03:00:00.000Z", + updatedAt: "2026-04-01T03:00:00.000Z", + }, + { + id: MessageId.make("legacy-assistant-2"), + role: "assistant", + text: "Response 2", + turnId: TurnId.make("legacy-turn-2"), + streaming: false, + createdAt: "2026-04-01T04:00:00.000Z", + updatedAt: "2026-04-01T04:00:00.000Z", + }, + ], + checkpoints: [], + }; + + const result = applyThreadDetailEvent(legacyThread, { + ...baseEventFields, + sequence: 14, + occurredAt: "2026-04-01T05:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.reverted", + payload: { + threadId: ThreadId.make("thread-1"), + turnCount: 1, + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.messages.map((message) => message.id)).toEqual([ + MessageId.make("legacy-user-1"), + MessageId.make("legacy-assistant-1"), + MessageId.make("legacy-user-2"), + ]); + } + }); + + it("retains uncheckpointed predecessor responses in mixed histories", () => { + const mixedThread: OrchestrationThread = { + ...baseThread, + messages: [ + { + id: MessageId.make("mixed-user-1"), + role: "user", + text: "First", + turnId: null, + streaming: false, + createdAt: "2026-04-01T01:00:00.000Z", + updatedAt: "2026-04-01T01:00:00.000Z", + }, + { + id: MessageId.make("mixed-assistant-1"), + role: "assistant", + text: "Response 1", + turnId: TurnId.make("mixed-turn-1"), + streaming: false, + createdAt: "2026-04-01T02:00:00.000Z", + updatedAt: "2026-04-01T02:00:00.000Z", + }, + { + id: MessageId.make("mixed-user-2"), + role: "user", + text: "Second", + turnId: null, + streaming: false, + createdAt: "2026-04-01T03:00:00.000Z", + updatedAt: "2026-04-01T03:00:00.000Z", + }, + { + id: MessageId.make("mixed-assistant-2"), + role: "assistant", + text: "Response 2", + turnId: TurnId.make("mixed-turn-2"), + streaming: false, + createdAt: "2026-04-01T04:00:00.000Z", + updatedAt: "2026-04-01T04:00:00.000Z", + }, + { + id: MessageId.make("mixed-user-3"), + role: "user", + text: "Third", + turnId: null, + streaming: false, + createdAt: "2026-04-01T05:00:00.000Z", + updatedAt: "2026-04-01T05:00:00.000Z", + }, + { + id: MessageId.make("mixed-assistant-3"), + role: "assistant", + text: "Response 3", + turnId: TurnId.make("mixed-turn-3"), + streaming: false, + createdAt: "2026-04-01T06:00:00.000Z", + updatedAt: "2026-04-01T06:00:00.000Z", + }, + ], + checkpoints: [ + { + turnId: TurnId.make("mixed-turn-2"), + checkpointTurnCount: 2, + checkpointRef: CheckpointRef.make("mixed-ref-2"), + status: "ready", + files: [], + assistantMessageId: MessageId.make("mixed-assistant-2"), + completedAt: "2026-04-01T04:00:00.000Z", + }, + ], + }; + + const result = applyThreadDetailEvent(mixedThread, { + ...baseEventFields, + sequence: 14, + occurredAt: "2026-04-01T07:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.reverted", + payload: { + threadId: ThreadId.make("thread-1"), + turnCount: 2, + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.messages.map((message) => message.id)).toEqual([ + MessageId.make("mixed-user-1"), + MessageId.make("mixed-assistant-1"), + MessageId.make("mixed-user-2"), + MessageId.make("mixed-assistant-2"), + MessageId.make("mixed-user-3"), + ]); + } + }); + + it("removes assistant output produced after the selected steering message", () => { + const steeringThread: OrchestrationThread = { + ...baseThread, + messages: [ + { + id: MessageId.make("steering-user-initial"), + role: "user", + text: "Start the task", + turnId: null, + streaming: false, + createdAt: "2026-04-01T01:00:00.000Z", + updatedAt: "2026-04-01T01:00:00.000Z", + }, + { + id: MessageId.make("steering-assistant-before"), + role: "assistant", + text: "Working on it", + turnId: TurnId.make("steering-turn"), + streaming: false, + createdAt: "2026-04-01T02:00:00.000Z", + updatedAt: "2026-04-01T02:00:00.000Z", + }, + { + id: MessageId.make("steering-user-target"), + role: "user", + text: "Actually, change direction", + turnId: null, + streaming: false, + createdAt: "2026-04-01T03:00:00.000Z", + updatedAt: "2026-04-01T03:00:00.000Z", + }, + { + id: MessageId.make("steering-assistant-after"), + role: "assistant", + text: "Changed direction", + turnId: TurnId.make("steering-turn"), + streaming: false, + createdAt: "2026-04-01T04:00:00.000Z", + updatedAt: "2026-04-01T04:00:00.000Z", + }, + ], + }; + + const result = applyThreadDetailEvent(steeringThread, { + ...baseEventFields, + sequence: 14, + occurredAt: "2026-04-01T05:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.reverted", + payload: { + threadId: ThreadId.make("thread-1"), + turnCount: 1, + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.messages.map((message) => message.id)).toEqual([ + MessageId.make("steering-user-initial"), + MessageId.make("steering-assistant-before"), + MessageId.make("steering-user-target"), + ]); + } + }); + + it("removes every message after the selected user prompt when checkpoints exist", () => { + const threadWithData: OrchestrationThread = { + ...baseThread, + messages: [ + { + id: MessageId.make("user-hello"), + role: "user", + text: "hello", + turnId: null, + streaming: false, + createdAt: "2026-04-01T01:00:00.000Z", + updatedAt: "2026-04-01T01:00:00.000Z", + }, + { + id: MessageId.make("assistant-hello"), + role: "assistant", + text: "Hello!", + turnId: TurnId.make("turn-1"), + streaming: false, + createdAt: "2026-04-01T02:00:00.000Z", + updatedAt: "2026-04-01T02:00:00.000Z", + }, + { + id: MessageId.make("user-create-file"), + role: "user", + text: "create a file", + turnId: null, + streaming: false, + createdAt: "2026-04-01T03:00:00.000Z", + updatedAt: "2026-04-01T03:00:00.000Z", + }, + { + id: MessageId.make("assistant-create-file"), + role: "assistant", + text: "Created it.", + turnId: TurnId.make("turn-2"), + streaming: false, + createdAt: "2026-04-01T04:00:00.000Z", + updatedAt: "2026-04-01T04:00:00.000Z", + }, + ], + checkpoints: [ + { + turnId: TurnId.make("turn-1"), + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make("ref-1"), + status: "ready", + files: [], + assistantMessageId: MessageId.make("assistant-hello"), + completedAt: "2026-04-01T02:00:00.000Z", + }, + { + turnId: TurnId.make("turn-2"), + checkpointTurnCount: 2, + checkpointRef: CheckpointRef.make("ref-2"), + status: "ready", + files: [], + assistantMessageId: MessageId.make("assistant-create-file"), + completedAt: "2026-04-01T04:00:00.000Z", + }, + ], + }; + + const result = applyThreadDetailEvent(threadWithData, { + ...baseEventFields, + sequence: 14, + occurredAt: "2026-04-01T05:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.reverted", + payload: { + threadId: ThreadId.make("thread-1"), + turnCount: 0, + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.messages.map((message) => message.id)).toEqual([ + MessageId.make("user-hello"), + ]); + } + }); + it("filters entities to retained turns", () => { const threadWithData: OrchestrationThread = { ...baseThread, @@ -623,6 +937,15 @@ describe("applyThreadDetailEvent", () => { }, { id: MessageId.make("msg-3"), + role: "user", + text: "Second", + turnId: null, + streaming: false, + createdAt: "2026-04-01T02:30:00.000Z", + updatedAt: "2026-04-01T02:30:00.000Z", + }, + { + id: MessageId.make("msg-4"), role: "assistant", text: "Response 2", turnId: TurnId.make("turn-2"), @@ -647,7 +970,7 @@ describe("applyThreadDetailEvent", () => { checkpointRef: CheckpointRef.make("ref-2"), status: "ready", files: [], - assistantMessageId: MessageId.make("msg-3"), + assistantMessageId: MessageId.make("msg-4"), completedAt: "2026-04-01T03:00:00.000Z", }, ], @@ -671,11 +994,94 @@ describe("applyThreadDetailEvent", () => { // turn-2 checkpoint is filtered out (turnCount 2 > revert target 1) expect(result.thread.checkpoints).toHaveLength(1); expect(result.thread.checkpoints[0]?.turnId).toBe("turn-1"); - // msg-3 (turn-2) is filtered, msg-1 (no turn) and msg-2 (turn-1) remain - expect(result.thread.messages).toHaveLength(2); + // The selected second user prompt remains, but its turn-2 response is removed. + expect(result.thread.messages).toHaveLength(3); expect(result.thread.latestTurn?.turnId).toBe("turn-1"); } }); + + it("retains checkpoint turn entities when no assistant message exists", () => { + const retainedTurnId = TurnId.make("checkpoint-only-turn"); + const checkpointOnlyThread: OrchestrationThread = { + ...baseThread, + messages: [ + { + id: MessageId.make("checkpoint-only-user"), + role: "user", + text: "Start", + turnId: null, + streaming: false, + createdAt: "2026-04-01T01:00:00.000Z", + updatedAt: "2026-04-01T01:00:00.000Z", + }, + { + id: MessageId.make("checkpoint-only-next-user"), + role: "user", + text: "Try again", + turnId: null, + streaming: false, + createdAt: "2026-04-01T02:00:00.000Z", + updatedAt: "2026-04-01T02:00:00.000Z", + }, + ], + checkpoints: [ + { + turnId: retainedTurnId, + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make("checkpoint-only-ref"), + status: "ready", + files: [], + assistantMessageId: null, + completedAt: "2026-04-01T01:30:00.000Z", + }, + ], + proposedPlans: [ + { + id: "checkpoint-only-plan", + turnId: retainedTurnId, + planMarkdown: "## Retained plan", + implementedAt: null, + implementationThreadId: null, + createdAt: "2026-04-01T01:10:00.000Z", + updatedAt: "2026-04-01T01:10:00.000Z", + }, + ], + activities: [ + { + id: EventId.make("checkpoint-only-activity"), + tone: "tool", + kind: "command", + summary: "Retained activity", + payload: {}, + turnId: retainedTurnId, + createdAt: "2026-04-01T01:20:00.000Z", + }, + ], + }; + + const result = applyThreadDetailEvent(checkpointOnlyThread, { + ...baseEventFields, + sequence: 14, + occurredAt: "2026-04-01T03:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.reverted", + payload: { + threadId: ThreadId.make("thread-1"), + turnCount: 1, + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.proposedPlans.map((plan) => plan.id)).toEqual([ + "checkpoint-only-plan", + ]); + expect(result.thread.activities.map((activity) => activity.id)).toEqual([ + EventId.make("checkpoint-only-activity"), + ]); + } + }); }); describe("no-op events", () => { diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index 670540fee70..90be24a6114 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -417,8 +417,11 @@ export function applyThreadDetailEvent( Arr.sort(checkpointOrder), ); - const retainedTurnIds = new Set(Arr.map(checkpoints, (entry) => entry.turnId)); - const messages = retainMessagesAfterRevert(thread.messages, retainedTurnIds); + const messages = retainMessagesAfterRevert(thread.messages, event.payload.turnCount); + const retainedTurnIds = new Set([ + ...Arr.map(checkpoints, (checkpoint) => checkpoint.turnId), + ...messages.flatMap((message) => (message.turnId === null ? [] : [message.turnId])), + ]); const proposedPlans = pipe( thread.proposedPlans, Arr.filter((plan) => plan.turnId === null || retainedTurnIds.has(plan.turnId)), @@ -531,17 +534,21 @@ function rebindCheckpointAssistantMessage( function retainMessagesAfterRevert( messages: ReadonlyArray, - retainedTurnIds: ReadonlySet, + turnCount: number, ): OrchestrationMessage[] { - // Keep messages that belong to a retained turn, plus system messages and - // messages without a turn binding (pre-turn-0 user messages). + let userMessageIndex = 0; + let reachedTargetUser = false; return Arr.filter(messages, (message) => { if (message.role === "system") { return true; } - if (message.turnId === null) { - return true; + if (reachedTargetUser) { + return false; + } + if (message.role === "user") { + reachedTargetUser = userMessageIndex === turnCount; + userMessageIndex += 1; } - return retainedTurnIds.has(message.turnId); + return true; }); }