diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index a65a0d1d73d9adcc6a315ef9a837a5d83f45ef5c..16bcf42bcbe80f52111657226b20e3d9e168806a 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -6,21 +6,20 @@ import { execFileSync } from "node:child_process"; import { NodeServices } from "@effect/platform-node"; import { ApprovalRequestId, + CheckpointRef, type OrchestrationEvent, type OrchestrationThread, - type ThreadId, } from "@t3tools/contracts"; import { Effect, Exit, Layer, ManagedRuntime, Option, Schedule, Scope, Stream } from "effect"; import { CheckpointStoreLive } from "../src/checkpointing/Layers/CheckpointStore.ts"; -import { CheckpointDiffBlobRepositoryLive } from "../src/persistence/Layers/CheckpointDiffBlobs.ts"; +import { CheckpointStore } from "../src/checkpointing/Services/CheckpointStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../src/persistence/Layers/OrchestrationCommandReceipts.ts"; import { OrchestrationEventStoreLive } from "../src/persistence/Layers/OrchestrationEventStore.ts"; import { ProjectionCheckpointRepositoryLive } from "../src/persistence/Layers/ProjectionCheckpoints.ts"; import { ProjectionPendingApprovalRepositoryLive } from "../src/persistence/Layers/ProjectionPendingApprovals.ts"; import { ProviderSessionRuntimeRepositoryLive } from "../src/persistence/Layers/ProviderSessionRuntime.ts"; import { makeSqlitePersistenceLive } from "../src/persistence/Layers/Sqlite.ts"; -import { CheckpointDiffBlobRepository } from "../src/persistence/Services/CheckpointDiffBlobs.ts"; import { ProjectionCheckpointRepository } from "../src/persistence/Services/ProjectionCheckpoints.ts"; import { ProjectionPendingApprovalRepository } from "../src/persistence/Services/ProjectionPendingApprovals.ts"; import { ProviderUnsupportedError } from "../src/provider/Errors.ts"; @@ -64,9 +63,9 @@ function initializeGitWorkspace(cwd: string) { runGit(cwd, ["commit", "-m", "Initial"]); } -export function checkpointRefForTurn(threadId: string, turnCount: number): string { +export function checkpointRefForTurn(threadId: string, turnCount: number): CheckpointRef { const encodedThreadId = Buffer.from(threadId, "utf8").toString("base64url"); - return `refs/t3/checkpoints/${encodedThreadId}/turn/${turnCount}`; + return CheckpointRef.makeUnsafe(`refs/t3/checkpoints/${encodedThreadId}/turn/${turnCount}`); } export function gitRefExists(cwd: string, ref: string): boolean { @@ -127,7 +126,7 @@ export interface OrchestrationIntegrationHarness { readonly engine: OrchestrationEngineShape; readonly snapshotQuery: ProjectionSnapshotQuery["Service"]; readonly providerService: ProviderService["Service"]; - readonly diffBlobRepository: CheckpointDiffBlobRepository["Service"]; + readonly checkpointStore: CheckpointStore["Service"]; readonly checkpointRepository: ProjectionCheckpointRepository["Service"]; readonly pendingApprovalRepository: ProjectionPendingApprovalRepository["Service"]; readonly waitForThread: ( @@ -155,23 +154,6 @@ export interface OrchestrationIntegrationHarness { }, never >; - readonly waitForDiffBlob: ( - input: { - readonly threadId: ThreadId; - readonly fromTurnCount: number; - readonly toTurnCount: number; - }, - timeoutMs?: number, - ) => Effect.Effect< - { - readonly threadId: ThreadId; - readonly fromTurnCount: number; - readonly toTurnCount: number; - readonly diff: string; - readonly createdAt: string; - }, - never - >; readonly dispose: Effect.Effect; } @@ -214,7 +196,6 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () { const runtimeServicesLayer = Layer.mergeAll( orchestrationLayer, OrchestrationProjectionSnapshotQueryLive, - CheckpointDiffBlobRepositoryLive, ProjectionCheckpointRepositoryLive, ProjectionPendingApprovalRepositoryLive, checkpointStoreLayer, @@ -252,8 +233,8 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () { const providerService = yield* Effect.promise(() => runtime.runPromise(Effect.service(ProviderService)), ); - const diffBlobRepository = yield* Effect.promise(() => - runtime.runPromise(Effect.service(CheckpointDiffBlobRepository)), + const checkpointStore = yield* Effect.promise(() => + runtime.runPromise(Effect.service(CheckpointStore)), ); const checkpointRepository = yield* Effect.promise(() => runtime.runPromise(Effect.service(ProjectionCheckpointRepository)), @@ -335,38 +316,6 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () { never >; - const waitForDiffBlob: OrchestrationIntegrationHarness["waitForDiffBlob"] = (input, timeoutMs) => - waitFor( - diffBlobRepository.get(input).pipe( - Effect.map((blob) => - Option.match(blob, { - onNone: () => null, - onSome: (value) => value, - }), - ), - ), - ( - blob, - ): blob is { - readonly threadId: ThreadId; - readonly fromTurnCount: number; - readonly toTurnCount: number; - readonly diff: string; - readonly createdAt: string; - } => blob !== null, - `checkpoint diff blob ${input.threadId}:${input.fromTurnCount}->${input.toTurnCount}`, - timeoutMs, - ) as Effect.Effect< - { - readonly threadId: ThreadId; - readonly fromTurnCount: number; - readonly toTurnCount: number; - readonly diff: string; - readonly createdAt: string; - }, - never - >; - let disposed = false; const dispose = Effect.gen(function* () { if (disposed) { @@ -390,13 +339,12 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () { engine, snapshotQuery, providerService, - diffBlobRepository, + checkpointStore, checkpointRepository, pendingApprovalRepository, waitForThread, waitForDomainEvent, waitForPendingApproval, - waitForDiffBlob, dispose, } satisfies OrchestrationIntegrationHarness; }); diff --git a/apps/server/integration/orchestrationEngine.integration.test.ts b/apps/server/integration/orchestrationEngine.integration.test.ts index 434457c9c661b6f18de0369d7b925b8e07cf9281..75a2811ecbd9980c468da24e18f70b7e2de71d28 100644 --- a/apps/server/integration/orchestrationEngine.integration.test.ts +++ b/apps/server/integration/orchestrationEngine.integration.test.ts @@ -202,7 +202,7 @@ it.live("runs a single turn end-to-end and persists checkpoint state in sqlite + ), ); -it.live("runs multi-turn file edits and persists checkpoint diffs + blobs", () => +it.live("runs multi-turn file edits and persists checkpoint diffs", () => withHarness((harness) => Effect.gen(function* () { yield* seedProjectAndThread(harness); @@ -327,12 +327,21 @@ it.live("runs multi-turn file edits and persists checkpoint diffs + blobs", () = [1, 2], ); - const diffBlob = yield* harness.waitForDiffBlob({ - threadId: THREAD_ID, - fromTurnCount: 1, - toTurnCount: 2, + const incrementalDiff = yield* harness.checkpointStore.diffCheckpoints({ + cwd: harness.workspaceDir, + fromCheckpointRef: checkpointRefForTurn(THREAD_ID, 1), + toCheckpointRef: checkpointRefForTurn(THREAD_ID, 2), + fallbackFromToHead: false, }); - assert.equal(diffBlob.diff.includes("README.md"), true); + assert.equal(incrementalDiff.includes("README.md"), true); + + const fullDiff = yield* harness.checkpointStore.diffCheckpoints({ + cwd: harness.workspaceDir, + fromCheckpointRef: checkpointRefForTurn(THREAD_ID, 0), + toCheckpointRef: checkpointRefForTurn(THREAD_ID, 2), + fallbackFromToHead: false, + }); + assert.equal(fullDiff.includes("README.md"), true); assert.equal( gitShowFileAtRef(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 1), "README.md"), @@ -511,6 +520,31 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, + { + type: "tool.started", + ...runtimeBase("evt-revert-1-tool-started", "2026-02-24T10:05:00.025Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + toolKind: "command", + title: "Edit file", + detail: "README.md", + }, + { + type: "tool.completed", + ...runtimeBase("evt-revert-1-tool-completed", "2026-02-24T10:05:00.035Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + toolKind: "command", + title: "Edit file", + detail: "README.md", + }, + { + type: "message.delta", + ...runtimeBase("evt-revert-1a", "2026-02-24T10:05:00.050Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + delta: "Updated README to v2.\n", + }, { type: "turn.completed", ...runtimeBase("evt-revert-2", "2026-02-24T10:05:00.100Z"), @@ -549,6 +583,31 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, + { + type: "tool.started", + ...runtimeBase("evt-revert-3-tool-started", "2026-02-24T10:05:01.025Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + toolKind: "command", + title: "Edit file", + detail: "README.md", + }, + { + type: "tool.completed", + ...runtimeBase("evt-revert-3-tool-completed", "2026-02-24T10:05:01.035Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + toolKind: "command", + title: "Edit file", + detail: "README.md", + }, + { + type: "message.delta", + ...runtimeBase("evt-revert-3a", "2026-02-24T10:05:01.050Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + delta: "Updated README to v3.\n", + }, { type: "turn.completed", ...runtimeBase("evt-revert-4", "2026-02-24T10:05:01.100Z"), @@ -571,7 +630,10 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git yield* harness.waitForThread( THREAD_ID, - (entry) => entry.latestTurnId === "turn-2" && entry.checkpoints.length === 2, + (entry) => + entry.latestTurnId === "turn-2" && + entry.checkpoints.length === 2 && + entry.activities.some((activity) => activity.turnId === "turn-2"), 8000, ); @@ -590,6 +652,29 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git entry.checkpoints.length === 1 && entry.checkpoints[0]?.checkpointTurnCount === 1, ); assert.equal(revertedThread.checkpoints[0]?.checkpointTurnCount, 1); + assert.deepEqual( + revertedThread.messages.map((message) => ({ role: message.role, text: message.text })), + [ + { role: "user", text: "First edit" }, + { role: "assistant", text: "Updated README to v2.\n" }, + ], + ); + assert.equal( + revertedThread.activities.some((activity) => activity.turnId === "turn-2"), + false, + ); + assert.equal( + revertedThread.activities.some( + (activity) => activity.turnId === "turn-1" && activity.kind === "tool.started", + ), + true, + ); + assert.equal( + revertedThread.activities.some( + (activity) => activity.turnId === "turn-1" && activity.kind === "tool.completed", + ), + true, + ); assert.equal(fs.readFileSync(path.join(harness.workspaceDir, "README.md"), "utf8"), "v2\n"); assert.equal(gitRefExists(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 2)), false); assert.deepEqual(harness.adapterHarness.getRollbackCalls(sessionId), [1]); diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 61c3813c17e3ffbc4526e3dc16776d10771bc72a..c2ce1b3449454edaeefcdb80b0995d6494a068cd 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -118,6 +118,7 @@ async function main() { port, host: mode === "desktop" ? "127.0.0.1" : undefined, cwd, + autoBootstrapProjectFromCwd: mode === "web", stateDir, staticDir, devUrl, diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 87619f26b690195d749e54f133b53745f39e3908..338a64e874c56b38048f6aeb30d42b336ef21066 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -24,11 +24,9 @@ import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts import { CheckpointReactorLive } from "./CheckpointReactor.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; -import { CheckpointDiffBlobRepositoryLive } from "../../persistence/Layers/CheckpointDiffBlobs.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; -import { CheckpointDiffBlobRepository } from "../../persistence/Services/CheckpointDiffBlobs.ts"; import { OrchestrationEngineService, type OrchestrationEngineShape, @@ -145,26 +143,6 @@ async function waitForEvent( return poll(); } -async function waitForDiffBlob( - repository: CheckpointDiffBlobRepository["Service"], - input: { threadId: ThreadId; fromTurnCount: number; toTurnCount: number }, - timeoutMs = 2000, -) { - const deadline = Date.now() + timeoutMs; - const poll = async () => { - const blob = await Effect.runPromise(repository.get(input)); - if (blob._tag === "Some") { - return blob.value; - } - if (Date.now() >= deadline) { - throw new Error("Timed out waiting for checkpoint diff blob persistence."); - } - await new Promise((resolve) => setTimeout(resolve, 10)); - return poll(); - }; - return poll(); -} - function runGit(cwd: string, args: ReadonlyArray) { return execFileSync("git", args, { cwd, @@ -214,7 +192,7 @@ async function waitForGitRefExists(cwd: string, ref: string, timeoutMs = 2000) { describe("CheckpointReactor", () => { let runtime: ManagedRuntime.ManagedRuntime< - OrchestrationEngineService | CheckpointReactor | CheckpointDiffBlobRepository | CheckpointStore, + OrchestrationEngineService | CheckpointReactor | CheckpointStore, unknown > | null = null; let scope: Scope.Closeable | null = null; @@ -250,23 +228,16 @@ describe("CheckpointReactor", () => { Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), ); - const diffBlobLayer = CheckpointDiffBlobRepositoryLive.pipe( - Layer.provide(SqlitePersistenceMemory), - ); const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); const layer = CheckpointReactorLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), - Layer.provideMerge(diffBlobLayer), Layer.provideMerge(checkpointStoreLayer), ); runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const reactor = await runtime.runPromise(Effect.service(CheckpointReactor)); - const diffBlobRepository = await runtime.runPromise( - Effect.service(CheckpointDiffBlobRepository), - ); const checkpointStore = await runtime.runPromise(Effect.service(CheckpointStore)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(reactor.start.pipe(Scope.provide(scope))); @@ -325,7 +296,6 @@ describe("CheckpointReactor", () => { engine, provider, cwd, - diffBlobRepository, }; } @@ -561,50 +531,4 @@ describe("CheckpointReactor", () => { expect(harness.provider.rollbackConversation).not.toHaveBeenCalled(); }); - it("persists full checkpoint diff blobs on thread.turn-diff-completed", async () => { - const harness = await createHarness(); - const createdAt = new Date().toISOString(); - - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.session.set", - commandId: asCommandId("cmd-session-set-diff"), - threadId: asThreadId("thread-1"), - session: { - threadId: asThreadId("thread-1"), - status: "ready", - providerName: "codex", - providerSessionId: asSessionId("sess-1"), - providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), - activeTurnId: null, - lastError: null, - updatedAt: createdAt, - }, - createdAt, - }), - ); - - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.turn.diff.complete", - commandId: asCommandId("cmd-diff-persist"), - threadId: asThreadId("thread-1"), - turnId: asTurnId("turn-1"), - completedAt: createdAt, - checkpointRef: checkpointRefForTurn(asThreadId("thread-1"), 1), - status: "ready", - files: [], - checkpointTurnCount: 1, - createdAt, - }), - ); - - const persistedBlob = await waitForDiffBlob(harness.diffBlobRepository, { - threadId: asThreadId("thread-1"), - fromTurnCount: 0, - toTurnCount: 1, - }); - expect(persistedBlob.diff).toContain("diff --git"); - expect(persistedBlob.diff).toContain("README.md"); - }); }); diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 3c787967d78fd9a564f6530ac63bd8003c94719f..974670dd3ec64056223f535919636aa2c4b75b3e 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -13,7 +13,6 @@ import { Effect, Layer, Option, Queue, Stream } from "effect"; import { parseTurnDiffFilesFromUnifiedDiff } from "../../checkpointing/Diffs.ts"; import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts"; -import { CheckpointDiffBlobRepository } from "../../persistence/Services/CheckpointDiffBlobs.ts"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { CheckpointReactor, type CheckpointReactorShape } from "../Services/CheckpointReactor.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; @@ -67,7 +66,6 @@ const make = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngineService; const providerService = yield* ProviderService; const checkpointStore = yield* CheckpointStore; - const checkpointDiffBlobRepository = yield* CheckpointDiffBlobRepository; const appendRevertFailureActivity = (input: { readonly threadId: ThreadId; @@ -160,62 +158,6 @@ const make = Effect.gen(function* () { return findSessionWithCwd(matchedSession); }); - const persistDiffBlobFromTurnCompletion = Effect.fnUntraced(function* ( - event: Extract, - ) { - const toTurnCount = event.payload.checkpointTurnCount; - const fromTurnCount = Math.max(0, toTurnCount - 1); - - const readModel = yield* orchestrationEngine.getReadModel(); - const thread = readModel.threads.find((entry) => entry.id === event.payload.threadId); - if (!thread) { - return; - } - - const sessionRuntime = yield* resolveSessionRuntimeForThread(event.payload.threadId); - if (Option.isNone(sessionRuntime)) { - yield* Effect.logWarning( - "checkpoint diff blob skipped: no active provider session cwd for thread", - { - threadId: event.payload.threadId, - fromTurnCount, - toTurnCount, - }, - ); - return; - } - - const fromCheckpointRef = - fromTurnCount === 0 - ? checkpointRefForThreadTurn(event.payload.threadId, 0) - : thread.checkpoints.find((checkpoint) => checkpoint.checkpointTurnCount === fromTurnCount) - ?.checkpointRef; - - if (!fromCheckpointRef) { - yield* Effect.logWarning("checkpoint diff blob skipped: missing from checkpoint ref", { - threadId: event.payload.threadId, - fromTurnCount, - toTurnCount, - }); - return; - } - - const diff = yield* checkpointStore.diffCheckpoints({ - cwd: sessionRuntime.value.cwd, - fromCheckpointRef, - toCheckpointRef: event.payload.checkpointRef, - fallbackFromToHead: false, - }); - - yield* checkpointDiffBlobRepository.upsert({ - threadId: event.payload.threadId, - fromTurnCount, - toTurnCount, - diff, - createdAt: event.payload.completedAt, - }); - }); - const captureCheckpointFromTurnCompletion = Effect.fnUntraced(function* ( event: Extract, ) { @@ -565,20 +507,6 @@ const make = Effect.gen(function* () { return; } - if (event.type === "thread.turn-diff-completed") { - yield* persistDiffBlobFromTurnCompletion(event).pipe( - Effect.catch((error) => - Effect.logWarning("checkpoint diff blob persistence failed", { - threadId: event.payload.threadId, - turnId: event.payload.turnId, - checkpointTurnCount: event.payload.checkpointTurnCount, - detail: error.message, - }), - ), - ); - return; - } - if (event.type === "thread.checkpoint-revert-requested") { yield* handleRevertRequested(event).pipe( Effect.catch((error) => @@ -714,7 +642,6 @@ const make = Effect.gen(function* () { if ( event.type !== "thread.turn-start-requested" && event.type !== "thread.message-sent" && - event.type !== "thread.turn-diff-completed" && event.type !== "thread.checkpoint-revert-requested" ) { return Effect.void; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 9ef91c23560d9cdcf282bb4d6168ac6a701ef3f9..03b2a6d2c69d9296bb5019622d1cc0cb15897af8 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -269,6 +269,137 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { } }), ); + + it.effect("keeps accumulated assistant text when completion payload text is empty", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const now = new Date().toISOString(); + + yield* eventStore.append({ + type: "project.created", + eventId: EventId.makeUnsafe("evt-empty-1"), + aggregateKind: "project", + aggregateId: ProjectId.makeUnsafe("project-empty"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-empty-1"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-empty-1"), + metadata: {}, + payload: { + projectId: ProjectId.makeUnsafe("project-empty"), + title: "Project Empty", + workspaceRoot: "/tmp/project-empty", + defaultModel: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.created", + eventId: EventId.makeUnsafe("evt-empty-2"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-empty"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-empty-2"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-empty-2"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-empty"), + projectId: ProjectId.makeUnsafe("project-empty"), + title: "Thread Empty", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-empty-3"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-empty"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-empty-3"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-empty-3"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-empty"), + messageId: MessageId.makeUnsafe("assistant-empty"), + role: "assistant", + text: "Hello", + turnId: null, + streaming: true, + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-empty-4"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-empty"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-empty-4"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-empty-4"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-empty"), + messageId: MessageId.makeUnsafe("assistant-empty"), + role: "assistant", + text: " world", + turnId: null, + streaming: true, + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-empty-5"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-empty"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-empty-5"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-empty-5"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-empty"), + messageId: MessageId.makeUnsafe("assistant-empty"), + role: "assistant", + text: "", + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + + yield* projectionPipeline.bootstrap; + + const messageRows = yield* sql<{ readonly text: string; readonly isStreaming: unknown }>` + SELECT + text, + is_streaming AS "isStreaming" + FROM projection_thread_messages + WHERE message_id = 'assistant-empty' + `; + assert.equal(messageRows.length, 1); + assert.equal(messageRows[0]?.text, "Hello world"); + assert.isFalse(Boolean(messageRows[0]?.isStreaming)); + }), + ); }); const engineLayer = it.layer( diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 2280d15f27f4be927d688244b0e3872d2dbb00ff..351ec9bba4ce2b79a9f5fe54994f2e2c114f8c12 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -12,7 +12,11 @@ import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts"; import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts"; import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts"; -import { ProjectionThreadMessageRepository } from "../../persistence/Services/ProjectionThreadMessages.ts"; +import { type ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts"; +import { + type ProjectionThreadMessage, + ProjectionThreadMessageRepository, +} from "../../persistence/Services/ProjectionThreadMessages.ts"; import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts"; import { type ProjectionThreadTurn } from "../../persistence/Services/ProjectionThreadTurns.ts"; import { ProjectionThreadTurnRepository } from "../../persistence/Services/ProjectionThreadTurns.ts"; @@ -62,6 +66,91 @@ function nextTurnCount(turns: ReadonlyArray): number { return turns.reduce((maxTurnCount, turn) => Math.max(maxTurnCount, turn.turnCount), 0) + 1; } +function retainProjectionMessagesAfterRevert( + messages: ReadonlyArray, + turns: ReadonlyArray, + turnCount: number, +): ReadonlyArray { + const retainedMessageIds = new Set(); + const retainedTurnIds = new Set(); + const keptTurns = turns.filter((turn) => turn.turnCount <= turnCount); + for (const turn of keptTurns) { + retainedTurnIds.add(turn.turnId); + if (turn.userMessageId !== null) { + retainedMessageIds.add(turn.userMessageId); + } + if (turn.assistantMessageId !== null) { + retainedMessageIds.add(turn.assistantMessageId); + } + } + + for (const message of messages) { + if (message.role === "system") { + retainedMessageIds.add(message.messageId); + continue; + } + if (message.turnId !== null && retainedTurnIds.has(message.turnId)) { + retainedMessageIds.add(message.messageId); + } + } + + 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)) + .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); + } + } + + 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), + ) + .toSorted( + (left, right) => + left.createdAt.localeCompare(right.createdAt) || + left.messageId.localeCompare(right.messageId), + ) + .slice(0, missingAssistantCount); + for (const message of fallbackAssistantMessages) { + retainedMessageIds.add(message.messageId); + } + } + + return messages.filter((message) => retainedMessageIds.has(message.messageId)); +} + +function retainProjectionActivitiesAfterRevert( + activities: ReadonlyArray, + turns: ReadonlyArray, + turnCount: number, +): ReadonlyArray { + const retainedTurnIds = new Set( + turns + .filter((turn) => turn.turnCount <= turnCount) + .map((turn) => turn.turnId), + ); + return activities.filter( + (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId), + ); +} + const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const eventStore = yield* OrchestrationEventStore; @@ -260,44 +349,111 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const applyThreadMessagesProjection: ProjectorDefinition["apply"] = (event) => Effect.gen(function* () { - if (event.type !== "thread.message-sent") { - return; + switch (event.type) { + case "thread.message-sent": { + const existingRows = yield* projectionThreadMessageRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + const existingMessage = existingRows.find((row) => row.messageId === event.payload.messageId); + const nextText = + existingMessage && event.payload.streaming + ? `${existingMessage.text}${event.payload.text}` + : existingMessage && event.payload.text.length === 0 + ? existingMessage.text + : event.payload.text; + yield* projectionThreadMessageRepository.upsert({ + messageId: event.payload.messageId, + threadId: event.payload.threadId, + turnId: event.payload.turnId, + role: event.payload.role, + text: nextText, + isStreaming: event.payload.streaming, + createdAt: existingMessage?.createdAt ?? event.payload.createdAt, + updatedAt: event.payload.updatedAt, + }); + return; + } + + case "thread.reverted": { + const existingRows = yield* projectionThreadMessageRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + if (existingRows.length === 0) { + return; + } + + const existingTurns = yield* projectionThreadTurnRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + const keptRows = retainProjectionMessagesAfterRevert( + existingRows, + existingTurns, + event.payload.turnCount, + ); + if (keptRows.length === existingRows.length) { + return; + } + + yield* projectionThreadMessageRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + yield* Effect.forEach(keptRows, projectionThreadMessageRepository.upsert, { + concurrency: 1, + }).pipe(Effect.asVoid); + return; + } + + default: + return; } - const existingRows = yield* projectionThreadMessageRepository.listByThreadId({ - threadId: event.payload.threadId, - }); - const existingMessage = existingRows.find((row) => row.messageId === event.payload.messageId); - const nextText = - existingMessage && event.payload.streaming - ? `${existingMessage.text}${event.payload.text}` - : event.payload.text; - yield* projectionThreadMessageRepository.upsert({ - messageId: event.payload.messageId, - threadId: event.payload.threadId, - turnId: event.payload.turnId, - role: event.payload.role, - text: nextText, - isStreaming: event.payload.streaming, - createdAt: existingMessage?.createdAt ?? event.payload.createdAt, - updatedAt: event.payload.updatedAt, - }); }); const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = (event) => Effect.gen(function* () { - if (event.type !== "thread.activity-appended") { - return; + switch (event.type) { + case "thread.activity-appended": + yield* projectionThreadActivityRepository.upsert({ + activityId: event.payload.activity.id, + threadId: event.payload.threadId, + turnId: event.payload.activity.turnId, + tone: event.payload.activity.tone, + kind: event.payload.activity.kind, + summary: event.payload.activity.summary, + payload: event.payload.activity.payload, + createdAt: event.payload.activity.createdAt, + }); + return; + + case "thread.reverted": { + const existingRows = yield* projectionThreadActivityRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + if (existingRows.length === 0) { + return; + } + const existingTurns = yield* projectionThreadTurnRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + const keptRows = retainProjectionActivitiesAfterRevert( + existingRows, + existingTurns, + event.payload.turnCount, + ); + if (keptRows.length === existingRows.length) { + return; + } + yield* projectionThreadActivityRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + yield* Effect.forEach(keptRows, projectionThreadActivityRepository.upsert, { + concurrency: 1, + }).pipe(Effect.asVoid); + return; + } + + default: + return; } - yield* projectionThreadActivityRepository.upsert({ - activityId: event.payload.activity.id, - threadId: event.payload.threadId, - turnId: event.payload.activity.turnId, - tone: event.payload.activity.tone, - kind: event.payload.activity.kind, - summary: event.payload.activity.summary, - payload: event.payload.activity.payload, - createdAt: event.payload.activity.createdAt, - }); }); const applyThreadSessionsProjection: ProjectorDefinition["apply"] = (event) => diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index bf54b31a7e92e2208ca2a75616315af0d5f98a68..45802c2b98d3cf050890723fdceeb153b5862d00 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -282,4 +282,213 @@ describe("orchestration projector", () => { expect(message?.streaming).toBe(false); expect(message?.updatedAt).toBe(completeAt); }); + + it("prunes reverted turn messages from in-memory thread snapshot", async () => { + const createdAt = "2026-02-23T10:00:00.000Z"; + const model = createEmptyReadModel(createdAt); + + const afterCreate = await Effect.runPromise( + projectEvent( + model, + makeEvent({ + sequence: 1, + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: createdAt, + commandId: "cmd-create", + payload: { + threadId: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5.3-codex", + branch: null, + worktreePath: null, + createdAt, + updatedAt: createdAt, + }, + }), + ), + ); + + const events: ReadonlyArray = [ + makeEvent({ + sequence: 2, + type: "thread.message-sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: "2026-02-23T10:00:01.000Z", + commandId: "cmd-user-1", + payload: { + threadId: "thread-1", + messageId: "user-msg-1", + role: "user", + text: "First edit", + turnId: null, + streaming: false, + createdAt: "2026-02-23T10:00:01.000Z", + updatedAt: "2026-02-23T10:00:01.000Z", + }, + }), + makeEvent({ + sequence: 3, + type: "thread.message-sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: "2026-02-23T10:00:02.000Z", + commandId: "cmd-assistant-1", + payload: { + threadId: "thread-1", + messageId: "assistant-msg-1", + role: "assistant", + text: "Updated README to v2.\n", + turnId: "turn-1", + streaming: false, + createdAt: "2026-02-23T10:00:02.000Z", + updatedAt: "2026-02-23T10:00:02.000Z", + }, + }), + makeEvent({ + sequence: 4, + type: "thread.turn-diff-completed", + aggregateType: "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", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: "2026-02-23T10:00:02.750Z", + commandId: "cmd-activity-1", + payload: { + threadId: "thread-1", + activity: { + id: "activity-1", + tone: "tool", + kind: "tool.started", + summary: "Edit file started", + payload: { toolKind: "command" }, + turnId: "turn-1", + createdAt: "2026-02-23T10:00:02.750Z", + }, + }, + }), + makeEvent({ + sequence: 6, + type: "thread.message-sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: "2026-02-23T10:00:03.000Z", + commandId: "cmd-user-2", + payload: { + threadId: "thread-1", + messageId: "user-msg-2", + role: "user", + text: "Second edit", + turnId: null, + streaming: false, + createdAt: "2026-02-23T10:00:03.000Z", + updatedAt: "2026-02-23T10:00:03.000Z", + }, + }), + makeEvent({ + sequence: 7, + type: "thread.message-sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: "2026-02-23T10:00:04.000Z", + commandId: "cmd-assistant-2", + payload: { + threadId: "thread-1", + messageId: "assistant-msg-2", + role: "assistant", + text: "Updated README to v3.\n", + turnId: "turn-2", + streaming: false, + createdAt: "2026-02-23T10:00:04.000Z", + updatedAt: "2026-02-23T10:00:04.000Z", + }, + }), + makeEvent({ + sequence: 8, + type: "thread.turn-diff-completed", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: "2026-02-23T10:00:04.500Z", + commandId: "cmd-turn-2-complete", + payload: { + threadId: "thread-1", + turnId: "turn-2", + checkpointTurnCount: 2, + checkpointRef: "refs/t3/checkpoints/thread-1/turn/2", + status: "ready", + files: [], + assistantMessageId: "assistant-msg-2", + completedAt: "2026-02-23T10:00:04.500Z", + }, + }), + makeEvent({ + sequence: 9, + type: "thread.activity-appended", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: "2026-02-23T10:00:04.750Z", + commandId: "cmd-activity-2", + payload: { + threadId: "thread-1", + activity: { + id: "activity-2", + tone: "tool", + kind: "tool.completed", + summary: "Edit file complete", + payload: { toolKind: "command" }, + turnId: "turn-2", + createdAt: "2026-02-23T10:00:04.750Z", + }, + }, + }), + makeEvent({ + sequence: 10, + type: "thread.reverted", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: "2026-02-23T10:00:05.000Z", + commandId: "cmd-revert", + payload: { + threadId: "thread-1", + turnCount: 1, + }, + }), + ]; + + const afterRevert = await events.reduce>>( + (statePromise, event) => + statePromise.then((state) => Effect.runPromise(projectEvent(state, event))), + Promise.resolve(afterCreate), + ); + + const thread = afterRevert.threads[0]; + expect(thread?.messages.map((message) => ({ role: message.role, text: message.text }))).toEqual([ + { role: "user", text: "First edit" }, + { role: "assistant", text: "Updated README to v2.\n" }, + ]); + 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?.latestTurnId).toBe("turn-1"); + }); }); diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 6be9b130d09414bc02db7fbede765047893b7078..1b6d4d6cf75f91742788f215aaae0e1a3e3f72f7 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -47,6 +47,68 @@ function decodeForEvent( ) as Effect.Effect; } +function retainThreadMessagesAfterRevert( + messages: ReadonlyArray, + retainedTurnIds: ReadonlySet, + turnCount: number, +): ReadonlyArray { + const retainedMessageIds = new Set(); + for (const message of messages) { + if (message.role === "system") { + retainedMessageIds.add(message.id); + continue; + } + if (message.turnId !== null && retainedTurnIds.has(message.turnId)) { + retainedMessageIds.add(message.id); + } + } + + 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)) + .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); + } + } + + 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)) + .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)); +} + +function retainThreadActivitiesAfterRevert( + activities: ReadonlyArray, + retainedTurnIds: ReadonlySet, +): ReadonlyArray { + return activities.filter( + (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId), + ); +} + export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { return { snapshotSequence: 0, @@ -347,6 +409,16 @@ export function projectEvent( const checkpoints = thread.checkpoints .filter((entry) => entry.checkpointTurnCount <= payload.turnCount) .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount); + const retainedTurnIds = new Set(checkpoints.map((checkpoint) => checkpoint.turnId)); + const messages = retainThreadMessagesAfterRevert( + thread.messages, + retainedTurnIds, + payload.turnCount, + ); + const activities = retainThreadActivitiesAfterRevert( + thread.activities, + retainedTurnIds, + ); const latestTurnId = checkpoints.length > 0 ? checkpoints[checkpoints.length - 1]?.turnId ?? null : null; @@ -355,6 +427,8 @@ export function projectEvent( ...nextBase, threads: updateThread(nextBase.threads, payload.threadId, { checkpoints, + messages, + activities, latestTurnId, updatedAt: event.occurredAt, }), diff --git a/apps/server/src/persistence/Errors.ts b/apps/server/src/persistence/Errors.ts index 86cfbf914d9c2ece851f7de4d5de836b5cc85b7c..a2b3b57c2ac6184900b7a6066488fe4211e00d57 100644 --- a/apps/server/src/persistence/Errors.ts +++ b/apps/server/src/persistence/Errors.ts @@ -205,8 +205,6 @@ export type OrchestrationCommandReceiptRepositoryError = | PersistenceSqlError | PersistenceDecodeError; -export type CheckpointDiffBlobRepositoryError = PersistenceSqlError | PersistenceDecodeError; - export type ProviderSessionRuntimeRepositoryError = | PersistenceSqlError | PersistenceDecodeError diff --git a/apps/server/src/persistence/Layers/CheckpointDiffBlobs.ts b/apps/server/src/persistence/Layers/CheckpointDiffBlobs.ts deleted file mode 100644 index 1267f70923c325295c2716448ddf24679fab41d8..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Layers/CheckpointDiffBlobs.ts +++ /dev/null @@ -1,79 +0,0 @@ -import * as SqlClient from "effect/unstable/sql/SqlClient"; -import * as SqlSchema from "effect/unstable/sql/SqlSchema"; -import { Effect, Layer } from "effect"; - -import { toPersistenceSqlError } from "../Errors.ts"; - -import { - CheckpointDiffBlobRepository, - CheckpointDiffBlob, - GetCheckpointDiffBlobInput, - type CheckpointDiffBlobRepositoryShape, -} from "../Services/CheckpointDiffBlobs.ts"; - -const makeCheckpointDiffBlobRepository = Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - const upsertDiffBlobRow = SqlSchema.void({ - Request: CheckpointDiffBlob, - execute: (blob) => - sql` - INSERT INTO checkpoint_diff_blobs ( - thread_id, - from_turn_count, - to_turn_count, - diff, - created_at - ) - VALUES ( - ${blob.threadId}, - ${blob.fromTurnCount}, - ${blob.toTurnCount}, - ${blob.diff}, - ${blob.createdAt} - ) - ON CONFLICT (thread_id, from_turn_count, to_turn_count) - DO UPDATE SET - diff = excluded.diff, - created_at = excluded.created_at - `, - }); - - const getDiffBlobRow = SqlSchema.findOneOption({ - Request: GetCheckpointDiffBlobInput, - Result: CheckpointDiffBlob, - execute: (request) => - sql` - SELECT - thread_id AS "threadId", - from_turn_count AS "fromTurnCount", - to_turn_count AS "toTurnCount", - diff, - created_at AS "createdAt" - FROM checkpoint_diff_blobs - WHERE thread_id = ${request.threadId} - AND from_turn_count = ${request.fromTurnCount} - AND to_turn_count = ${request.toTurnCount} - `, - }); - - const upsert: CheckpointDiffBlobRepositoryShape["upsert"] = (blob) => - upsertDiffBlobRow(blob).pipe( - Effect.mapError(toPersistenceSqlError("CheckpointDiffBlobRepository.upsert:query")), - ); - - const get: CheckpointDiffBlobRepositoryShape["get"] = (input) => - getDiffBlobRow(input).pipe( - Effect.mapError(toPersistenceSqlError("CheckpointDiffBlobRepository.get:query")), - ); - - return { - upsert, - get, - } satisfies CheckpointDiffBlobRepositoryShape; -}); - -export const CheckpointDiffBlobRepositoryLive = Layer.effect( - CheckpointDiffBlobRepository, - makeCheckpointDiffBlobRepository, -); diff --git a/apps/server/src/persistence/Services/CheckpointDiffBlobs.ts b/apps/server/src/persistence/Services/CheckpointDiffBlobs.ts deleted file mode 100644 index ca0875f6d4a49b954fe2ec0d8c8fe834cd065f8c..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Services/CheckpointDiffBlobs.ts +++ /dev/null @@ -1,36 +0,0 @@ -import { IsoDateTime, NonNegativeInt, ThreadId } from "@t3tools/contracts"; -import { Option, Schema, ServiceMap } from "effect"; -import type { Effect } from "effect"; - -import type { CheckpointDiffBlobRepositoryError } from "../Errors.ts"; - -export const CheckpointDiffBlob = Schema.Struct({ - threadId: ThreadId, - fromTurnCount: NonNegativeInt, - toTurnCount: NonNegativeInt, - diff: Schema.String, - createdAt: IsoDateTime, -}); -export type CheckpointDiffBlob = typeof CheckpointDiffBlob.Type; - -export const GetCheckpointDiffBlobInput = Schema.Struct({ - threadId: ThreadId, - fromTurnCount: NonNegativeInt, - toTurnCount: NonNegativeInt, -}); -export type GetCheckpointDiffBlobInput = typeof GetCheckpointDiffBlobInput.Type; - -export interface CheckpointDiffBlobRepositoryShape { - readonly upsert: ( - blob: CheckpointDiffBlob, - ) => Effect.Effect; - - readonly get: ( - input: GetCheckpointDiffBlobInput, - ) => Effect.Effect, CheckpointDiffBlobRepositoryError>; -} - -export class CheckpointDiffBlobRepository extends ServiceMap.Service< - CheckpointDiffBlobRepository, - CheckpointDiffBlobRepositoryShape ->()("persistence/CheckpointDiffBlobRepository") {} diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index 38b19e621989ac917a0957af456d15386553e63a..75c02ca4079de6fd5aa5384ac4b860f381edfd62 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -273,6 +273,7 @@ describe("WebSocket Server", () => { options: { persistenceLayer?: Layer.Layer; cwd?: string; + autoBootstrapProjectFromCwd?: boolean; devUrl?: string; authToken?: string; stateDir?: string; @@ -290,6 +291,9 @@ describe("WebSocket Server", () => { cwd: options.cwd ?? "/test/project", stateDir, persistenceLayer: options.persistenceLayer ?? SqlitePersistenceMemory, + ...(options.autoBootstrapProjectFromCwd !== undefined + ? { autoBootstrapProjectFromCwd: options.autoBootstrapProjectFromCwd } + : {}), ...(options.devUrl ? { devUrl: options.devUrl } : {}), ...(options.authToken ? { authToken: options.authToken } : {}), ...(options.providerLayer ? { providerLayer: options.providerLayer } : {}), @@ -333,6 +337,37 @@ describe("WebSocket Server", () => { }); }); + it("bootstraps the cwd project on startup when enabled", async () => { + server = createTestServer({ + cwd: "/test/bootstrap-workspace", + autoBootstrapProjectFromCwd: true, + }); + await server.start(); + const addr = server.httpServer.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + expect(port).toBeGreaterThan(0); + + const ws = await connectWs(port); + connections.push(ws); + await waitForMessage(ws); // welcome + + const snapshotResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.getSnapshot); + expect(snapshotResponse.error).toBeUndefined(); + const snapshot = snapshotResponse.result as { + projects: Array<{ workspaceRoot: string; title: string; defaultModel: string | null }>; + }; + + expect(snapshot.projects).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + workspaceRoot: "/test/bootstrap-workspace", + title: "bootstrap-workspace", + defaultModel: "gpt-5-codex", + }), + ]), + ); + }); + it("logs outbound websocket push events in dev mode", async () => { const logSpy = vi.spyOn(console, "log").mockImplementation(() => { // Keep test output clean while verifying websocket logs. @@ -670,6 +705,86 @@ describe("WebSocket Server", () => { expect(response.error!.message).toContain("Invalid request format"); }); + it("returns error when requesting turn diff for unknown thread", async () => { + server = createTestServer({ cwd: "/test" }); + await server.start(); + const addr = server.httpServer.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + + const ws = await connectWs(port); + connections.push(ws); + await waitForMessage(ws); + + const response = await sendRequest(ws, ORCHESTRATION_WS_METHODS.getTurnDiff, { + threadId: "thread-missing", + fromTurnCount: 1, + toTurnCount: 2, + }); + expect(response.result).toBeUndefined(); + expect(response.error?.message).toContain("Thread 'thread-missing' not found."); + }); + + it("returns error when requesting full thread diff for unknown thread", async () => { + server = createTestServer({ cwd: "/test" }); + await server.start(); + const addr = server.httpServer.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + + const ws = await connectWs(port); + connections.push(ws); + await waitForMessage(ws); + + const response = await sendRequest(ws, ORCHESTRATION_WS_METHODS.getFullThreadDiff, { + threadId: "thread-missing", + toTurnCount: 2, + }); + expect(response.result).toBeUndefined(); + expect(response.error?.message).toContain("Thread 'thread-missing' not found."); + }); + + it("returns retryable error when requested turn exceeds current checkpoint turn count", async () => { + server = createTestServer({ cwd: "/test" }); + await server.start(); + const addr = server.httpServer.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + + const ws = await connectWs(port); + connections.push(ws); + await waitForMessage(ws); + + const createdAt = new Date().toISOString(); + const createProjectResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.dispatchCommand, { + type: "project.create", + commandId: "cmd-diff-project-create", + projectId: "project-diff", + title: "Diff Project", + workspaceRoot: "/tmp/ws-diff-project", + defaultModel: "gpt-5-codex", + createdAt, + }); + expect(createProjectResponse.error).toBeUndefined(); + const createThreadResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.dispatchCommand, { + type: "thread.create", + commandId: "cmd-diff-thread-create", + threadId: "thread-diff", + projectId: "project-diff", + title: "Diff Thread", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }); + expect(createThreadResponse.error).toBeUndefined(); + + const response = await sendRequest(ws, ORCHESTRATION_WS_METHODS.getTurnDiff, { + threadId: "thread-diff", + fromTurnCount: 0, + toTurnCount: 1, + }); + expect(response.result).toBeUndefined(); + expect(response.error?.message).toContain("exceeds current turn count"); + }); + it("keeps orchestration domain push behavior for provider runtime events", async () => { const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); const sessionId = "sess-test"; @@ -1011,21 +1126,23 @@ describe("WebSocket Server", () => { expect(pullResponse.error?.message).not.toContain("Unknown method"); }); - it("returns error for git.status", async () => { + it("supports git.status over websocket", async () => { + const statusResult = { + branch: "feature/test", + hasWorkingTreeChanges: true, + workingTree: { + files: [{ path: "src/index.ts", insertions: 7, deletions: 2 }], + insertions: 7, + deletions: 2, + }, + hasUpstream: false, + aheadCount: 0, + behindCount: 0, + openPr: null, + }; + const gitManager = { - status: vi.fn().mockResolvedValue({ - branch: "feature/test", - hasWorkingTreeChanges: true, - workingTree: { - files: [{ path: "src/index.ts", insertions: 7, deletions: 2 }], - insertions: 7, - deletions: 2, - }, - hasUpstream: false, - aheadCount: 0, - behindCount: 0, - openPr: null, - }), + status: vi.fn().mockResolvedValue(statusResult), runStackedAction: vi.fn(), }; @@ -1041,9 +1158,9 @@ describe("WebSocket Server", () => { const response = await sendRequest(ws, WS_METHODS.gitStatus, { cwd: "/test", }); - expect(response.result).toBeUndefined(); - expect(response.error?.message).toContain("Unknown method"); - expect(gitManager.status).not.toHaveBeenCalled(); + expect(response.error).toBeUndefined(); + expect(response.result).toEqual(statusResult); + expect(gitManager.status).toHaveBeenCalledWith({ cwd: "/test" }); }); it("returns errors from git.runStackedAction", async () => { diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index 0fbfa82d37366ebd5e527f4c04fff865a86ba8c6..bb2bb8aeb1ad8cee46e762caa7940976a25c4449 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -6,6 +6,7 @@ import path from "node:path"; import type { Duplex } from "node:stream"; import { + CheckpointRef, EDITORS, ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS, @@ -17,7 +18,7 @@ import { WsResponse, } from "@t3tools/contracts"; import { NodeServices } from "@effect/platform-node"; -import { Effect, Exit, Layer, ManagedRuntime, Option, Schema, Scope, Stream } from "effect"; +import { Effect, Exit, Layer, ManagedRuntime, Schema, Scope, Stream } from "effect"; import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger"; @@ -52,7 +53,6 @@ import { import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor"; import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite"; import assert from "node:assert"; -import { CheckpointDiffBlobRepositoryLive } from "./persistence/Layers/CheckpointDiffBlobs"; import { OrchestrationEventStoreLive } from "./persistence/Layers/OrchestrationEventStore"; import { OrchestrationCommandReceiptRepositoryLive } from "./persistence/Layers/OrchestrationCommandReceipts"; import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter"; @@ -62,10 +62,7 @@ import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionD import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService"; import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore"; import { ProviderSessionRuntimeRepositoryLive } from "./persistence/Layers/ProviderSessionRuntime"; -import { - CheckpointDiffBlobRepository, - type CheckpointDiffBlobRepositoryShape, -} from "./persistence/Services/CheckpointDiffBlobs"; +import { CheckpointStore, type CheckpointStoreShape } from "./checkpointing/Services/CheckpointStore"; import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as SqlError from "effect/unstable/sql/SqlError"; @@ -91,6 +88,7 @@ export interface ServerOptions { port: number; host?: string | undefined; cwd: string; + autoBootstrapProjectFromCwd?: boolean | undefined; stateDir?: string | undefined; persistenceLayer?: | Layer.Layer @@ -171,11 +169,18 @@ function stripRequestTag( return rest; } +const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints"; + +function checkpointRefForThreadTurn(threadId: string, turnCount: number): CheckpointRef { + const encodedThreadId = Buffer.from(threadId, "utf8").toString("base64url"); + return CheckpointRef.makeUnsafe(`${CHECKPOINT_REFS_PREFIX}/${encodedThreadId}/turn/${turnCount}`); +} + type EffectRuntime = ManagedRuntime.ManagedRuntime< | OrchestrationEngineService | ProjectionSnapshotQuery | ProviderService - | CheckpointDiffBlobRepository + | CheckpointStore | OrchestrationReactor, unknown >; @@ -185,6 +190,7 @@ export function createServer(options: ServerOptions) { port, host, cwd, + autoBootstrapProjectFromCwd = false, stateDir, persistenceLayer: customPersistenceLayer, staticDir, @@ -201,7 +207,7 @@ export function createServer(options: ServerOptions) { let orchestrationEngine: OrchestrationEngineShape | null = null; let projectionSnapshotQuery: ProjectionSnapshotQueryShape | null = null; let providerService: ProviderServiceShape | null = null; - let checkpointDiffBlobRepository: CheckpointDiffBlobRepositoryShape | null = null; + let checkpointStore: CheckpointStoreShape | null = null; const clients = new Set(); const logger = createLogger("ws"); const logWebSocketEvents = @@ -235,11 +241,40 @@ export function createServer(options: ServerOptions) { return projectionSnapshotQuery; }; - const getCheckpointDiffBlobRepository = () => { - assert(checkpointDiffBlobRepository, "Checkpoint diff blob repository is not started"); - return checkpointDiffBlobRepository; + const getCheckpointStore = () => { + assert(checkpointStore, "Checkpoint store is not started"); + return checkpointStore; }; + async function bootstrapProjectForCwd() { + if (!autoBootstrapProjectFromCwd) { + return; + } + + const runtime = getEffectRuntime(); + const snapshotQuery = getProjectionSnapshotQuery(); + const engine = getOrchestrationEngine(); + const snapshot = await runtime.runPromise(snapshotQuery.getSnapshot()); + const existing = snapshot.projects.find((project) => project.workspaceRoot === cwd); + if (existing) { + return; + } + + const createdAt = new Date().toISOString(); + const projectName = path.basename(cwd) || "project"; + await runtime.runPromise( + engine.dispatchUnknownCommand({ + type: "project.create", + commandId: crypto.randomUUID(), + projectId: crypto.randomUUID(), + title: projectName, + workspaceRoot: cwd, + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + } + const attachSubscriptionss = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngineService; const orchestrationReactor = yield* OrchestrationReactor; @@ -439,7 +474,114 @@ export function createServer(options: ServerOptions) { const orchestrationEngine = getOrchestrationEngine(); const projectionReadModelQuery = getProjectionSnapshotQuery(); const effectRuntime = getEffectRuntime(); - const diffBlobRepository = getCheckpointDiffBlobRepository(); + const liveCheckpointStore = getCheckpointStore(); + + const computeTurnDiff = async (input: { + readonly threadId: string; + readonly fromTurnCount: number; + readonly toTurnCount: number; + }) => { + if (input.fromTurnCount > input.toTurnCount) { + throw new Error( + `Invalid turn diff range for thread '${input.threadId}': from ${input.fromTurnCount} exceeds to ${input.toTurnCount}.`, + ); + } + + if (input.fromTurnCount === input.toTurnCount) { + return { + threadId: input.threadId, + fromTurnCount: input.fromTurnCount, + toTurnCount: input.toTurnCount, + diff: "", + }; + } + + const snapshot = await effectRuntime.runPromise(projectionReadModelQuery.getSnapshot()); + const thread = snapshot.threads.find((entry) => entry.id === input.threadId); + if (!thread) { + throw new Error(`Thread '${input.threadId}' not found.`); + } + + const maxTurnCount = thread.checkpoints.reduce( + (max, checkpoint) => Math.max(max, checkpoint.checkpointTurnCount), + 0, + ); + if (input.toTurnCount > maxTurnCount) { + throw new Error( + `Turn diff range exceeds current turn count for thread '${input.threadId}': requested ${input.toTurnCount}, current ${maxTurnCount}.`, + ); + } + + const workspaceCwd = + thread.worktreePath?.trim() || + snapshot.projects.find((project) => project.id === thread.projectId)?.workspaceRoot?.trim(); + if (!workspaceCwd) { + throw new Error( + `Invariant violation: workspace path missing for thread '${input.threadId}' when computing turn diff.`, + ); + } + + const fromCheckpointRef = + input.fromTurnCount === 0 + ? checkpointRefForThreadTurn(input.threadId, 0) + : thread.checkpoints.find( + (checkpoint) => checkpoint.checkpointTurnCount === input.fromTurnCount, + )?.checkpointRef; + if (!fromCheckpointRef) { + throw new Error( + `Checkpoint is unavailable for turn ${input.fromTurnCount} in thread ${input.threadId}.`, + ); + } + + const toCheckpointRef = thread.checkpoints.find( + (checkpoint) => checkpoint.checkpointTurnCount === input.toTurnCount, + )?.checkpointRef; + if (!toCheckpointRef) { + throw new Error( + `Checkpoint is unavailable for turn ${input.toTurnCount} in thread ${input.threadId}.`, + ); + } + + const [fromExists, toExists] = await Promise.all([ + effectRuntime.runPromise( + liveCheckpointStore.hasCheckpointRef({ + cwd: workspaceCwd, + checkpointRef: fromCheckpointRef, + }), + ), + effectRuntime.runPromise( + liveCheckpointStore.hasCheckpointRef({ + cwd: workspaceCwd, + checkpointRef: toCheckpointRef, + }), + ), + ]); + if (!fromExists) { + throw new Error( + `Filesystem checkpoint is unavailable for turn ${input.fromTurnCount} in thread ${input.threadId}.`, + ); + } + if (!toExists) { + throw new Error( + `Filesystem checkpoint is unavailable for turn ${input.toTurnCount} in thread ${input.threadId}.`, + ); + } + + const diff = await effectRuntime.runPromise( + liveCheckpointStore.diffCheckpoints({ + cwd: workspaceCwd, + fromCheckpointRef, + toCheckpointRef, + fallbackFromToHead: false, + }), + ); + return { + threadId: input.threadId, + fromTurnCount: input.fromTurnCount, + toTurnCount: input.toTurnCount, + diff, + }; + }; switch (request.body._tag) { case ORCHESTRATION_WS_METHODS.getSnapshot: { @@ -456,19 +598,16 @@ export function createServer(options: ServerOptions) { case ORCHESTRATION_WS_METHODS.getTurnDiff: { const body = stripRequestTag(request.body); - const blob = await effectRuntime.runPromise(diffBlobRepository.get(body)); - if (Option.isNone(blob)) { - throw new Error( - `Turn diff not found for thread '${body.threadId}' from ${body.fromTurnCount} to ${body.toTurnCount}.`, - ); - } + return computeTurnDiff(body); + } - return { + case ORCHESTRATION_WS_METHODS.getFullThreadDiff: { + const body = stripRequestTag(request.body); + return computeTurnDiff({ threadId: body.threadId, - fromTurnCount: body.fromTurnCount, + fromTurnCount: 0, toTurnCount: body.toTurnCount, - diff: blob.value?.diff, - }; + }); } case ORCHESTRATION_WS_METHODS.replayEvents: @@ -526,28 +665,8 @@ export function createServer(options: ServerOptions) { return undefined; } - // TODO: Bring back - // case WS_METHODS.gitStatus: { - // const params = request.params as { cwd: string }; - // const status = await gitManager.status(request.params as never); - // const projects = await effectRuntime.runPromise(projectRepository.list()); - // const project = projects.find((entry) => entry.cwd === params.cwd); - // if (project) { - // await effectRuntime.runPromise( - // orchestrationEngine.dispatch({ - // type: "git.readModel.upsert", - // commandId: crypto.randomUUID(), - // projectId: project.id, - // branch: status.branch, - // hasWorkingTreeChanges: status.hasWorkingTreeChanges, - // aheadCount: status.aheadCount, - // behindCount: status.behindCount, - // createdAt: new Date().toISOString(), - // }), - // ); - // } - // return status; - // } + case WS_METHODS.gitStatus: + return gitManager.status(stripRequestTag(request.body)); case WS_METHODS.gitPull: return pullGitBranch(stripRequestTag(request.body)); @@ -606,8 +725,10 @@ export function createServer(options: ServerOptions) { keybindings: keybindingsConfig, }; - default: - throw new Error(`Unknown method: ${request.body._tag}`); + default: { + const _exhaustiveCheck: never = request.body; + throw new Error(`Unknown method: ${String(_exhaustiveCheck)}`); + } } } @@ -651,7 +772,6 @@ export function createServer(options: ServerOptions) { const runtimeServicesLayer = Layer.mergeAll( orchestrationLayer, OrchestrationProjectionSnapshotQueryLive, - CheckpointDiffBlobRepositoryLive, checkpointStoreLayer, providerLayer, ); @@ -679,17 +799,17 @@ export function createServer(options: ServerOptions) { const [ nextOrchestrationEngine, nextProjectionSnapshotQuery, - nextCheckpointDiffBlobRepository, + nextCheckpointStore, nextProviderService, ] = await Promise.all([ runtime.runPromise(Effect.service(OrchestrationEngineService)), runtime.runPromise(Effect.service(ProjectionSnapshotQuery)), - runtime.runPromise(Effect.service(CheckpointDiffBlobRepository)), + runtime.runPromise(Effect.service(CheckpointStore)), runtime.runPromise(Effect.service(ProviderService)), ]); orchestrationEngine = nextOrchestrationEngine; projectionSnapshotQuery = nextProjectionSnapshotQuery; - checkpointDiffBlobRepository = nextCheckpointDiffBlobRepository; + checkpointStore = nextCheckpointStore; providerService = nextProviderService; await runtime.runPromise(attachSubscriptionss); } catch (error) { @@ -715,7 +835,7 @@ export function createServer(options: ServerOptions) { effectRuntime = null; orchestrationEngine = null; projectionSnapshotQuery = null; - checkpointDiffBlobRepository = null; + checkpointStore = null; providerService = null; if (!runtime) { @@ -731,6 +851,7 @@ export function createServer(options: ServerOptions) { async function start() { effectRuntime = await createEffectRuntime(); + await bootstrapProjectForCwd(); return new Promise((resolve, reject) => { const onError = (error: Error) => { diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 7f45829815b1943b2413dee18dec36bf7b295650..6b4029cb5255d0d9af4bc8cd9a7ef0ae2f9e9108 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -132,28 +132,39 @@ export default function Sidebar() { branch?: string | null; worktreePath?: string | null; }, - ) => { - if (!api) return; + ): Promise => { + if (!api) return Promise.resolve(); const threadId = newThreadId(); const createdAt = new Date().toISOString(); const model = state.projects.find((project) => project.id === projectId)?.model ?? DEFAULT_MODEL; - void api.orchestration.dispatchCommand({ - type: "thread.create", - commandId: newCommandId(), - threadId, - projectId: asProjectId(projectId), - title: "New thread", - model, - branch: options?.branch ?? null, - worktreePath: options?.worktreePath ?? null, - createdAt, - }); - void navigate({ - to: "/$threadId", - params: { threadId }, - }); + return (async () => { + await api.orchestration.dispatchCommand({ + type: "thread.create", + commandId: newCommandId(), + threadId, + projectId: asProjectId(projectId), + title: "New thread", + model, + branch: options?.branch ?? null, + worktreePath: options?.worktreePath ?? null, + createdAt, + }); + + // Ensure route guards can see the new thread before navigating. + try { + const snapshot = await api.orchestration.getSnapshot(); + dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot }); + } catch { + // Event stream can still hydrate the thread shortly after dispatch. + } + + await navigate({ + to: "/$threadId", + params: { threadId }, + }); + })(); }, - [api, navigate, state.projects], + [api, dispatch, navigate, state.projects], ); const focusMostRecentThreadForProject = useCallback( @@ -201,7 +212,7 @@ export default function Sidebar() { defaultModel: DEFAULT_MODEL, createdAt, }); - handleNewThread(projectId); + await handleNewThread(projectId); } finally { setIsAddingProject(false); setNewCwd(""); @@ -399,7 +410,7 @@ export default function Sidebar() { const projectId = activeThread?.projectId ?? state.projects[0]?.id; if (!projectId) return; event.preventDefault(); - handleNewThread(projectId); + void handleNewThread(projectId); return; } @@ -407,7 +418,7 @@ export default function Sidebar() { const projectId = activeThread?.projectId ?? state.projects[0]?.id; if (!projectId) return; event.preventDefault(); - handleNewThread(projectId, { + void handleNewThread(projectId, { branch: activeThread?.branch ?? null, worktreePath: activeThread?.worktreePath ?? null, }); @@ -446,7 +457,7 @@ export default function Sidebar() { return; } const firstProject = state.projects[0]; - if (firstProject) handleNewThread(firstProject.id); + if (firstProject) void handleNewThread(firstProject.id); }} > + @@ -561,7 +572,9 @@ export default function Sidebar() { diff --git a/apps/web/src/lib/providerReactQuery.test.ts b/apps/web/src/lib/providerReactQuery.test.ts index 91edd765677fc6cd67fa111037bb7b4dea082ee7..96e12f0a1cfe32e4cf60cdf0ede5394980f23d4e 100644 --- a/apps/web/src/lib/providerReactQuery.test.ts +++ b/apps/web/src/lib/providerReactQuery.test.ts @@ -28,9 +28,11 @@ describe("providerQueryKeys.checkpointDiff", () => { describe("checkpointDiffQueryOptions", () => { it("forwards checkpoint range to the provider API", async () => { const getTurnDiff = vi.fn().mockResolvedValue({ diff: "patch" }); + const getFullThreadDiff = vi.fn().mockResolvedValue({ diff: "patch" }); const api = { orchestration: { getTurnDiff, + getFullThreadDiff, }, } as unknown as NativeApi; @@ -49,6 +51,34 @@ describe("checkpointDiffQueryOptions", () => { fromTurnCount: 3, toTurnCount: 4, }); + expect(getFullThreadDiff).not.toHaveBeenCalled(); + }); + + it("uses explicit full thread diff API when range starts from zero", async () => { + const getTurnDiff = vi.fn().mockResolvedValue({ diff: "patch" }); + const getFullThreadDiff = vi.fn().mockResolvedValue({ diff: "patch" }); + const api = { + orchestration: { + getTurnDiff, + getFullThreadDiff, + }, + } as unknown as NativeApi; + + const options = checkpointDiffQueryOptions(api, { + threadId: "thread-id", + fromTurnCount: 0, + toTurnCount: 2, + cacheScope: "thread:all", + }); + + const queryClient = new QueryClient(); + await queryClient.fetchQuery(options); + + expect(getFullThreadDiff).toHaveBeenCalledWith({ + threadId: "thread-id", + toTurnCount: 2, + }); + expect(getTurnDiff).not.toHaveBeenCalled(); }); it("retries checkpoint-not-ready errors longer than generic failures", () => { diff --git a/apps/web/src/lib/providerReactQuery.ts b/apps/web/src/lib/providerReactQuery.ts index bb69cc9e56a4517b6057d2a1a773c5b08df92445..a457ca4e8c9183c88eb000bdd718912ac3c8e3b3 100644 --- a/apps/web/src/lib/providerReactQuery.ts +++ b/apps/web/src/lib/providerReactQuery.ts @@ -64,6 +64,12 @@ export function checkpointDiffQueryOptions( if (typeof fromTurnCount !== "number" || typeof toTurnCount !== "number") { throw new Error("Checkpoint diff range is invalid."); } + if (fromTurnCount === 0) { + return api.orchestration.getFullThreadDiff({ + threadId: asThreadId(input.threadId), + toTurnCount, + }); + } return api.orchestration.getTurnDiff({ threadId: asThreadId(input.threadId), fromTurnCount, diff --git a/apps/web/src/routes/__root.tsx b/apps/web/src/routes/__root.tsx index eb9d06cbc38d7f8f3d17e72902db049776685c73..eb2de5aca04720d797bfd5f8bbcd5b04ce58380a 100644 --- a/apps/web/src/routes/__root.tsx +++ b/apps/web/src/routes/__root.tsx @@ -3,18 +3,16 @@ import { createRootRouteWithContext, type ErrorComponentProps, } from "@tanstack/react-router"; -import { useEffect, useRef } from "react"; -import { QueryClient } from "@tanstack/react-query"; +import { useEffect } from "react"; +import { QueryClient, useQueryClient } from "@tanstack/react-query"; import { APP_DISPLAY_NAME } from "../branding"; import { Button } from "../components/ui/button"; import { AnchoredToastProvider, ToastProvider } from "../components/ui/toast"; -import { isElectron } from "../env"; import { useNativeApi } from "../hooks/useNativeApi"; -import { DEFAULT_MODEL } from "../model-logic"; import { useStore } from "../store"; -import { newCommandId, newProjectId, newThreadId } from "../lib/orchestrationIds"; import { onServerWelcome } from "../wsNativeApi"; +import { providerQueryKeys } from "../lib/providerReactQuery"; export const Route = createRootRouteWithContext<{ queryClient: QueryClient; @@ -42,7 +40,6 @@ function RootRouteView() { - @@ -124,6 +121,7 @@ function errorDetails(error: unknown): string { function EventRouter() { const api = useNativeApi(); const { dispatch } = useStore(); + const queryClient = useQueryClient(); useEffect(() => { if (!api) return; @@ -166,6 +164,9 @@ function EventRouter() { return; } latestSequence = event.sequence; + if (event.type === "thread.turn-diff-completed" || event.type === "thread.reverted") { + void queryClient.invalidateQueries({ queryKey: providerQueryKeys.all }); + } void syncSnapshot(); }); const unsubWelcome = onServerWelcome(() => { @@ -176,63 +177,7 @@ function EventRouter() { unsubDomainEvent(); unsubWelcome(); }; - }, [api, dispatch]); - - return null; -} - -function AutoProjectBootstrap() { - const api = useNativeApi(); - const { state, dispatch } = useStore(); - const bootstrappedRef = useRef(false); - - useEffect(() => { - if (!api) return; - // Browser mode auto-adds the current cwd project via server welcome. - // Desktop skips this because projects are already loaded from the repository. - if (isElectron) return; - - return onServerWelcome((payload) => { - if (bootstrappedRef.current) return; - - // Don't create duplicate projects for the same cwd - const existing = state.projects.find((project) => project.cwd === payload.cwd); - if (existing) { - bootstrappedRef.current = true; - dispatch({ type: "SET_THREADS_HYDRATED", hydrated: true }); - return; - } - - bootstrappedRef.current = true; - const now = new Date().toISOString(); - const projectId = newProjectId(); - const threadId = newThreadId(); - void api.orchestration - .dispatchCommand({ - type: "project.create", - commandId: newCommandId(), - projectId, - title: payload.projectName, - workspaceRoot: payload.cwd, - defaultModel: DEFAULT_MODEL, - createdAt: now, - }) - .then(() => - api.orchestration.dispatchCommand({ - type: "thread.create", - commandId: newCommandId(), - threadId, - projectId, - title: "New thread", - model: DEFAULT_MODEL, - branch: null, - worktreePath: null, - createdAt: now, - }), - ) - .catch(() => undefined); - }); - }, [api, state.projects, dispatch]); + }, [api, dispatch, queryClient]); return null; } diff --git a/apps/web/src/wsNativeApi.test.ts b/apps/web/src/wsNativeApi.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..1197faa6ea191044ee00eafdc7d1cbc60c4b207e --- /dev/null +++ b/apps/web/src/wsNativeApi.test.ts @@ -0,0 +1,53 @@ +import { CommandId, ORCHESTRATION_WS_METHODS, ProjectId, ThreadId } from "@t3tools/contracts"; +import { describe, expect, it, vi } from "vitest"; + +const requestMock = vi.fn<(...args: Array) => Promise>(); +const subscribeMock = vi.fn<(...args: Array) => () => void>(() => () => undefined); + +vi.mock("./wsTransport", () => { + return { + WsTransport: class MockWsTransport { + request = requestMock; + subscribe = subscribeMock; + }, + }; +}); + +describe("wsNativeApi", () => { + it("wraps orchestration dispatch commands in the command envelope", async () => { + requestMock.mockResolvedValue(undefined); + const { createWsNativeApi } = await import("./wsNativeApi"); + + const api = createWsNativeApi(); + const command = { + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-1"), + projectId: ProjectId.makeUnsafe("project-1"), + title: "Project", + workspaceRoot: "/tmp/project", + defaultModel: "gpt-5-codex", + createdAt: "2026-02-24T00:00:00.000Z", + } as const; + await api.orchestration.dispatchCommand(command); + + expect(requestMock).toHaveBeenCalledWith(ORCHESTRATION_WS_METHODS.dispatchCommand, { + command, + }); + }); + + it("forwards full-thread diff requests to the orchestration websocket method", async () => { + requestMock.mockResolvedValue({ diff: "patch" }); + const { createWsNativeApi } = await import("./wsNativeApi"); + + const api = createWsNativeApi(); + await api.orchestration.getFullThreadDiff({ + threadId: ThreadId.makeUnsafe("thread-1"), + toTurnCount: 1, + }); + + expect(requestMock).toHaveBeenCalledWith(ORCHESTRATION_WS_METHODS.getFullThreadDiff, { + threadId: "thread-1", + toTurnCount: 1, + }); + }); +}); diff --git a/apps/web/src/wsNativeApi.ts b/apps/web/src/wsNativeApi.ts index 88a11e1424077be293de3a10361f0709ba77da50..88ab2f3089b37139eaddee2dc94805aab2656525 100644 --- a/apps/web/src/wsNativeApi.ts +++ b/apps/web/src/wsNativeApi.ts @@ -128,8 +128,10 @@ export function createWsNativeApi(): NativeApi { orchestration: { getSnapshot: () => transport.request(ORCHESTRATION_WS_METHODS.getSnapshot), dispatchCommand: (command) => - transport.request(ORCHESTRATION_WS_METHODS.dispatchCommand, command), + transport.request(ORCHESTRATION_WS_METHODS.dispatchCommand, { command }), getTurnDiff: (input) => transport.request(ORCHESTRATION_WS_METHODS.getTurnDiff, input), + getFullThreadDiff: (input) => + transport.request(ORCHESTRATION_WS_METHODS.getFullThreadDiff, input), replayEvents: (fromSequenceExclusive) => transport.request(ORCHESTRATION_WS_METHODS.replayEvents, { fromSequenceExclusive }), onDomainEvent: (callback) => diff --git a/packages/contracts/src/ipc.ts b/packages/contracts/src/ipc.ts index 154b7adab4feac1806f982974076b1fd5ce90e8e..ce1c0305a54a918515205b55ba98a2fcfad11c0b 100644 --- a/packages/contracts/src/ipc.ts +++ b/packages/contracts/src/ipc.ts @@ -28,6 +28,8 @@ import type { import type { ServerUpsertKeybindingInput, ServerUpsertKeybindingResult } from "./server"; import type { ClientOrchestrationCommand, + OrchestrationGetFullThreadDiffInput, + OrchestrationGetFullThreadDiffResult, OrchestrationGetTurnDiffInput, OrchestrationGetTurnDiffResult, OrchestrationEvent, @@ -95,6 +97,9 @@ export interface NativeApi { getSnapshot: () => Promise; dispatchCommand: (command: ClientOrchestrationCommand) => Promise<{ sequence: number }>; getTurnDiff: (input: OrchestrationGetTurnDiffInput) => Promise; + getFullThreadDiff: ( + input: OrchestrationGetFullThreadDiffInput, + ) => Promise; replayEvents: (fromSequenceExclusive: number) => Promise; onDomainEvent: (callback: (event: OrchestrationEvent) => void) => () => void; }; diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index b8012890a999fc1585d985abedde414c3da9eab7..8a840217985064761eade2285f0e554bb2f8aa71 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -4,6 +4,7 @@ export const ORCHESTRATION_WS_METHODS = { getSnapshot: "orchestration.getSnapshot", dispatchCommand: "orchestration.dispatchCommand", getTurnDiff: "orchestration.getTurnDiff", + getFullThreadDiff: "orchestration.getFullThreadDiff", replayEvents: "orchestration.replayEvents", } as const; @@ -837,6 +838,15 @@ export type OrchestrationGetTurnDiffInput = typeof OrchestrationGetTurnDiffInput export const OrchestrationGetTurnDiffResult = ThreadTurnDiff; export type OrchestrationGetTurnDiffResult = typeof OrchestrationGetTurnDiffResult.Type; +export const OrchestrationGetFullThreadDiffInput = Schema.Struct({ + threadId: ThreadId, + toTurnCount: NonNegativeInt, +}); +export type OrchestrationGetFullThreadDiffInput = typeof OrchestrationGetFullThreadDiffInput.Type; + +export const OrchestrationGetFullThreadDiffResult = ThreadTurnDiff; +export type OrchestrationGetFullThreadDiffResult = typeof OrchestrationGetFullThreadDiffResult.Type; + export const OrchestrationReplayEventsInput = Schema.Struct({ fromSequenceExclusive: NonNegativeInt, }); @@ -858,6 +868,10 @@ export const OrchestrationRpcSchemas = { input: OrchestrationGetTurnDiffInput, output: OrchestrationGetTurnDiffResult, }, + getFullThreadDiff: { + input: OrchestrationGetFullThreadDiffInput, + output: OrchestrationGetFullThreadDiffResult, + }, replayEvents: { input: OrchestrationReplayEventsInput, output: OrchestrationReplayEventsResult, diff --git a/packages/contracts/src/ws.ts b/packages/contracts/src/ws.ts index 5f5d26b866ff2c4d94975556ce903c117ee622b3..57dbf8899f220a62ecc55239cff48338b49c2d2a 100644 --- a/packages/contracts/src/ws.ts +++ b/packages/contracts/src/ws.ts @@ -2,6 +2,7 @@ import { Schema } from "effect"; import { ClientOrchestrationCommand, + OrchestrationGetFullThreadDiffInput, ORCHESTRATION_WS_METHODS, OrchestrationGetSnapshotInput, OrchestrationGetTurnDiffInput, @@ -80,6 +81,7 @@ export const WebSocketRequestBody = Schema.TaggedUnion({ [ORCHESTRATION_WS_METHODS.dispatchCommand]: { command: ClientOrchestrationCommand }, [ORCHESTRATION_WS_METHODS.getSnapshot]: OrchestrationGetSnapshotInput.fields, [ORCHESTRATION_WS_METHODS.getTurnDiff]: OrchestrationGetTurnDiffInput.fields, + [ORCHESTRATION_WS_METHODS.getFullThreadDiff]: OrchestrationGetFullThreadDiffInput.fields, [ORCHESTRATION_WS_METHODS.replayEvents]: OrchestrationReplayEventsInput.fields, // Project Search