From c2b9680a5efc0855d24fb5a4bdcf9320469354d4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 25 Feb 2026 15:05:58 -0800 Subject: [PATCH] buuuunch of cleanup --- .../OrchestrationEngineHarness.integration.ts | 6 - .../orchestrationEngine.integration.test.ts | 52 +++-- .../Layers/CheckpointDiffQuery.test.ts | 190 +++++++++++++++++ .../Layers/CheckpointDiffQuery.ts | 193 +++++++++++++++++ apps/server/src/checkpointing/Refs.ts | 10 + .../Services/CheckpointDiffQuery.ts | 24 +++ apps/server/src/git.ts | 8 +- apps/server/src/keybindings.test.ts | 2 +- apps/server/src/keybindings.ts | 4 +- apps/server/src/orchestration/Layer.test.ts | 68 +++--- .../Layers/CheckpointReactor.test.ts | 115 +++++----- .../orchestration/Layers/CheckpointReactor.ts | 14 +- .../Layers/OrchestrationEngine.ts | 24 +-- .../Layers/ProjectionSnapshotQuery.test.ts | 5 +- .../Layers/ProviderCommandReactor.test.ts | 46 ++-- .../Layers/ProviderRuntimeIngestion.test.ts | 16 +- .../Services/OrchestrationEngine.ts | 14 +- .../decider.projectScripts.test.ts | 26 ++- .../src/provider/Layers/CodexAdapter.test.ts | 11 +- .../provider/Layers/ProviderService.test.ts | 11 +- apps/server/src/serverLayers.ts | 8 + apps/server/src/terminalManager.test.ts | 7 +- apps/server/src/terminalManager.ts | 6 +- apps/server/src/wsServer.test.ts | 22 +- apps/server/src/wsServer.ts | 188 +++-------------- .../components/BranchToolbar.logic.test.ts | 11 +- .../web/src/components/BranchToolbar.logic.ts | 4 +- apps/web/src/components/BranchToolbar.tsx | 26 +-- apps/web/src/components/ChatMarkdown.tsx | 14 +- apps/web/src/components/ChatView.tsx | 199 +++++++++--------- apps/web/src/components/DiffPanel.tsx | 19 +- apps/web/src/components/GitActionsControl.tsx | 36 ++-- apps/web/src/components/Sidebar.tsx | 74 +++---- .../src/components/ThreadTerminalDrawer.tsx | 34 ++- apps/web/src/components/ui/command.tsx | 5 +- apps/web/src/components/ui/sheet.tsx | 1 + apps/web/src/components/ui/toast.tsx | 10 +- apps/web/src/diffRouteSearch.ts | 7 +- apps/web/src/historyBootstrap.test.ts | 17 +- apps/web/src/hooks/useNativeApi.ts | 6 - apps/web/src/lib/gitReactQuery.ts | 89 ++++---- apps/web/src/lib/orchestrationIds.ts | 1 - apps/web/src/lib/projectReactQuery.ts | 11 +- apps/web/src/lib/providerReactQuery.test.ts | 80 ++++--- apps/web/src/lib/providerReactQuery.ts | 64 +++--- apps/web/src/lib/serverReactQuery.ts | 9 +- apps/web/src/model-logic.test.ts | 33 --- apps/web/src/model-logic.ts | 10 - apps/web/src/nativeApi.ts | 8 + apps/web/src/routes/__root.tsx | 10 +- apps/web/src/routes/_chat.$threadId.tsx | 14 +- apps/web/src/routes/_chat.tsx | 2 +- apps/web/src/session-logic.test.ts | 36 +--- apps/web/src/session-logic.ts | 80 ++----- apps/web/src/store.ts | 39 ++-- apps/web/src/threadFactory.ts | 42 ---- apps/web/src/types.ts | 34 +-- apps/web/src/worktreeCleanup.test.ts | 35 ++- apps/web/src/worktreeCleanup.ts | 2 +- apps/web/src/wsTransport.ts | 7 +- packages/contracts/src/editor.ts | 5 +- packages/contracts/src/git.ts | 96 ++++----- packages/contracts/src/keybindings.ts | 26 ++- packages/contracts/src/model.ts | 4 + packages/contracts/src/orchestration.test.ts | 46 ++++ packages/contracts/src/orchestration.ts | 33 ++- packages/contracts/src/project.ts | 19 +- packages/contracts/src/ws.test.ts | 39 ++++ packages/contracts/src/ws.ts | 82 +++++--- 69 files changed, 1387 insertions(+), 1102 deletions(-) create mode 100644 apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts create mode 100644 apps/server/src/checkpointing/Layers/CheckpointDiffQuery.ts create mode 100644 apps/server/src/checkpointing/Refs.ts create mode 100644 apps/server/src/checkpointing/Services/CheckpointDiffQuery.ts delete mode 100644 apps/web/src/hooks/useNativeApi.ts delete mode 100644 apps/web/src/model-logic.test.ts delete mode 100644 apps/web/src/model-logic.ts delete mode 100644 apps/web/src/threadFactory.ts create mode 100644 packages/contracts/src/orchestration.test.ts create mode 100644 packages/contracts/src/ws.test.ts diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index 16bcf42bcbe80f52111657226b20e3d9e168806a..5fa6fead39193dcfb9c2fcc22613828778ba3fd4 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -6,7 +6,6 @@ import { execFileSync } from "node:child_process"; import { NodeServices } from "@effect/platform-node"; import { ApprovalRequestId, - CheckpointRef, type OrchestrationEvent, type OrchestrationThread, } from "@t3tools/contracts"; @@ -63,11 +62,6 @@ function initializeGitWorkspace(cwd: string) { runGit(cwd, ["commit", "-m", "Initial"]); } -export function checkpointRefForTurn(threadId: string, turnCount: number): CheckpointRef { - const encodedThreadId = Buffer.from(threadId, "utf8").toString("base64url"); - return CheckpointRef.makeUnsafe(`refs/t3/checkpoints/${encodedThreadId}/turn/${turnCount}`); -} - export function gitRefExists(cwd: string, ref: string): boolean { try { runGit(cwd, ["show-ref", "--verify", "--quiet", ref]); diff --git a/apps/server/integration/orchestrationEngine.integration.test.ts b/apps/server/integration/orchestrationEngine.integration.test.ts index a2d8aeff4a629456c06f8735abe7e26b79c3433b..511beb566195e49fec05abc59daf9d5892856ba6 100644 --- a/apps/server/integration/orchestrationEngine.integration.test.ts +++ b/apps/server/integration/orchestrationEngine.integration.test.ts @@ -17,17 +17,15 @@ import { Effect, Option } from "effect"; import type { TestTurnResponse } from "./TestProviderAdapter.integration.ts"; import { - checkpointRefForTurn, gitRefExists, gitShowFileAtRef, makeOrchestrationIntegrationHarness, type OrchestrationIntegrationHarness, } from "./OrchestrationEngineHarness.integration.ts"; +import { checkpointRefForThreadTurn } from "../src/checkpointing/Refs.ts"; -const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); -const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); const asProviderSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); @@ -37,7 +35,7 @@ const asApprovalRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); const PROJECT_ID = asProjectId("project-1"); -const THREAD_ID = asThreadId("thread-1"); +const THREAD_ID = ThreadId.makeUnsafe("thread-1"); const FIXTURE_SESSION_ID = asProviderSessionId("fixture-session"); const FIXTURE_THREAD_ID = asProviderThreadId("fixture-thread"); const FIXTURE_TURN_ID = asProviderTurnId("fixture-turn"); @@ -95,7 +93,7 @@ const seedProjectAndThread = (harness: OrchestrationIntegrationHarness) => yield* harness.engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-create"), + commandId: CommandId.makeUnsafe("cmd-project-create"), projectId: PROJECT_ID, title: "Integration Project", workspaceRoot: harness.workspaceDir, @@ -105,7 +103,7 @@ const seedProjectAndThread = (harness: OrchestrationIntegrationHarness) => yield* harness.engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-create"), + commandId: CommandId.makeUnsafe("cmd-thread-create"), threadId: THREAD_ID, projectId: PROJECT_ID, title: "Integration Thread", @@ -124,7 +122,7 @@ const startTurn = (input: { }) => input.harness.engine.dispatch({ type: "thread.turn.start", - commandId: asCommandId(input.commandId), + commandId: CommandId.makeUnsafe(input.commandId), threadId: THREAD_ID, message: { messageId: asMessageId(input.messageId), @@ -195,8 +193,8 @@ it.live("runs a single turn end-to-end and persists checkpoint state in sqlite + yield* harness.waitForDomainEvent((event) => event.type === "thread.turn-diff-completed"); - const ref0 = checkpointRefForTurn(THREAD_ID, 0); - const ref1 = checkpointRefForTurn(THREAD_ID, 1); + const ref0 = checkpointRefForThreadTurn(THREAD_ID, 0); + const ref1 = checkpointRefForThreadTurn(THREAD_ID, 1); assert.equal(gitRefExists(harness.workspaceDir, ref0), true); assert.equal(gitRefExists(harness.workspaceDir, ref1), true); assert.equal(gitShowFileAtRef(harness.workspaceDir, ref0, "README.md"), "v1\n"); @@ -335,26 +333,34 @@ it.live("runs multi-turn file edits and persists checkpoint diffs", () => const incrementalDiff = yield* harness.checkpointStore.diffCheckpoints({ cwd: harness.workspaceDir, - fromCheckpointRef: checkpointRefForTurn(THREAD_ID, 1), - toCheckpointRef: checkpointRefForTurn(THREAD_ID, 2), + fromCheckpointRef: checkpointRefForThreadTurn(THREAD_ID, 1), + toCheckpointRef: checkpointRefForThreadTurn(THREAD_ID, 2), fallbackFromToHead: false, }); 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), + fromCheckpointRef: checkpointRefForThreadTurn(THREAD_ID, 0), + toCheckpointRef: checkpointRefForThreadTurn(THREAD_ID, 2), fallbackFromToHead: false, }); assert.equal(fullDiff.includes("README.md"), true); assert.equal( - gitShowFileAtRef(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 1), "README.md"), + gitShowFileAtRef( + harness.workspaceDir, + checkpointRefForThreadTurn(THREAD_ID, 1), + "README.md", + ), "v2\n", ); assert.equal( - gitShowFileAtRef(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 2), "README.md"), + gitShowFileAtRef( + harness.workspaceDir, + checkpointRefForThreadTurn(THREAD_ID, 2), + "README.md", + ), "v3\n", ); }), @@ -416,7 +422,7 @@ it.live("tracks approval requests and resolves pending approvals on user respons yield* harness.engine.dispatch({ type: "thread.approval.respond", - commandId: asCommandId("cmd-approval-respond"), + commandId: CommandId.makeUnsafe("cmd-approval-respond"), threadId: THREAD_ID, requestId: APPROVAL_REQUEST_ID, decision: "accept", @@ -507,7 +513,10 @@ it.live("records failed turn runtime state and checkpoint status as error", () = if (Option.isSome(checkpointRow)) { assert.equal(checkpointRow.value.status, "error"); } - assert.equal(gitRefExists(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 1)), true); + assert.equal( + gitRefExists(harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 1)), + true, + ); }), ), ); @@ -644,7 +653,7 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git yield* harness.engine.dispatch({ type: "thread.checkpoint.revert", - commandId: asCommandId("cmd-checkpoint-revert"), + commandId: CommandId.makeUnsafe("cmd-checkpoint-revert"), threadId: THREAD_ID, turnCount: 1, createdAt: nowIso(), @@ -681,7 +690,10 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git 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.equal( + gitRefExists(harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 2)), + false, + ); assert.deepEqual(harness.adapterHarness.getRollbackCalls(sessionId), [1]); const checkpointRows = yield* harness.checkpointRepository.listByThreadId({ @@ -701,7 +713,7 @@ it.live( yield* harness.engine.dispatch({ type: "thread.checkpoint.revert", - commandId: asCommandId("cmd-checkpoint-revert-no-session"), + commandId: CommandId.makeUnsafe("cmd-checkpoint-revert-no-session"), threadId: THREAD_ID, turnCount: 0, createdAt: nowIso(), diff --git a/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..d61103cd2b3bc7ce8bcf661604344746893935b6 --- /dev/null +++ b/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts @@ -0,0 +1,190 @@ +import { + CheckpointRef, + ProjectId, + ThreadId, + TurnId, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import { Effect, Layer } from "effect"; +import { describe, expect, it } from "vitest"; + +import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { checkpointRefForThreadTurn } from "../Refs.ts"; +import { CheckpointDiffQueryLive } from "./CheckpointDiffQuery.ts"; +import { CheckpointStore, type CheckpointStoreShape } from "../Services/CheckpointStore.ts"; +import { CheckpointDiffQuery } from "../Services/CheckpointDiffQuery.ts"; + +const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); +const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); +const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value); + +function makeSnapshot(input: { + readonly projectId: ProjectId; + readonly threadId: ThreadId; + readonly workspaceRoot: string; + readonly worktreePath: string | null; + readonly checkpointTurnCount: number; + readonly checkpointRef: CheckpointRef; +}): OrchestrationReadModel { + return { + snapshotSequence: 0, + updatedAt: "2026-01-01T00:00:00.000Z", + projects: [ + { + id: input.projectId, + title: "Project", + workspaceRoot: input.workspaceRoot, + defaultModel: null, + scripts: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + deletedAt: null, + }, + ], + threads: [ + { + id: input.threadId, + projectId: input.projectId, + title: "Thread", + model: "gpt-5-codex", + branch: null, + worktreePath: input.worktreePath, + latestTurnId: asTurnId("turn-1"), + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + deletedAt: null, + messages: [], + activities: [], + checkpoints: [ + { + turnId: asTurnId("turn-1"), + checkpointTurnCount: input.checkpointTurnCount, + checkpointRef: input.checkpointRef, + status: "ready", + files: [], + assistantMessageId: null, + completedAt: "2026-01-01T00:00:00.000Z", + }, + ], + session: null, + }, + ], + }; +} + +describe("CheckpointDiffQueryLive", () => { + it("computes diffs using canonical turn-0 checkpoint refs", async () => { + const projectId = asProjectId("project-1"); + const threadId = ThreadId.makeUnsafe("thread-1"); + const toCheckpointRef = asCheckpointRef("refs/t3/checkpoints/thread-1/turn/1"); + const hasCheckpointRefCalls: Array = []; + const diffCheckpointsCalls: Array<{ + readonly fromCheckpointRef: CheckpointRef; + readonly toCheckpointRef: CheckpointRef; + readonly cwd: string; + }> = []; + + const snapshot = makeSnapshot({ + projectId, + threadId, + workspaceRoot: "/tmp/workspace", + worktreePath: null, + checkpointTurnCount: 1, + checkpointRef: toCheckpointRef, + }); + + const checkpointStore: CheckpointStoreShape = { + isGitRepository: () => Effect.succeed(true), + captureCheckpoint: () => Effect.void, + hasCheckpointRef: ({ checkpointRef }) => + Effect.sync(() => { + hasCheckpointRefCalls.push(checkpointRef); + return true; + }), + restoreCheckpoint: () => Effect.succeed(true), + diffCheckpoints: ({ fromCheckpointRef, toCheckpointRef, cwd }) => + Effect.sync(() => { + diffCheckpointsCalls.push({ fromCheckpointRef, toCheckpointRef, cwd }); + return "diff patch"; + }), + deleteCheckpointRefs: () => Effect.void, + }; + + const layer = CheckpointDiffQueryLive.pipe( + Layer.provideMerge(Layer.succeed(CheckpointStore, checkpointStore)), + Layer.provideMerge( + Layer.succeed(ProjectionSnapshotQuery, { + getSnapshot: () => Effect.succeed(snapshot), + }), + ), + ); + + const result = await Effect.runPromise( + Effect.gen(function* () { + const query = yield* CheckpointDiffQuery; + return yield* query.getTurnDiff({ + threadId, + fromTurnCount: 0, + toTurnCount: 1, + }); + }).pipe(Effect.provide(layer)), + ); + + const expectedFromRef = checkpointRefForThreadTurn(threadId, 0); + expect(hasCheckpointRefCalls).toEqual([expectedFromRef, toCheckpointRef]); + expect(diffCheckpointsCalls).toEqual([ + { + cwd: "/tmp/workspace", + fromCheckpointRef: expectedFromRef, + toCheckpointRef, + }, + ]); + expect(result).toEqual({ + threadId, + fromTurnCount: 0, + toTurnCount: 1, + diff: "diff patch", + }); + }); + + it("fails when the thread is missing from the snapshot", async () => { + const threadId = ThreadId.makeUnsafe("thread-missing"); + + const checkpointStore: CheckpointStoreShape = { + isGitRepository: () => Effect.succeed(true), + captureCheckpoint: () => Effect.void, + hasCheckpointRef: () => Effect.succeed(true), + restoreCheckpoint: () => Effect.succeed(true), + diffCheckpoints: () => Effect.succeed(""), + deleteCheckpointRefs: () => Effect.void, + }; + + const layer = CheckpointDiffQueryLive.pipe( + Layer.provideMerge(Layer.succeed(CheckpointStore, checkpointStore)), + Layer.provideMerge( + Layer.succeed(ProjectionSnapshotQuery, { + getSnapshot: () => + Effect.succeed({ + snapshotSequence: 0, + projects: [], + threads: [], + updatedAt: "2026-01-01T00:00:00.000Z", + } satisfies OrchestrationReadModel), + }), + ), + ); + + await expect( + Effect.runPromise( + Effect.gen(function* () { + const query = yield* CheckpointDiffQuery; + return yield* query.getTurnDiff({ + threadId, + fromTurnCount: 0, + toTurnCount: 1, + }); + }).pipe(Effect.provide(layer)), + ), + ).rejects.toThrow("Thread 'thread-missing' not found."); + }); +}); diff --git a/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.ts b/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.ts new file mode 100644 index 0000000000000000000000000000000000000000..a18c9dc8cec016f9e69753ae33e0f64f7734174e --- /dev/null +++ b/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.ts @@ -0,0 +1,193 @@ +import { + OrchestrationGetTurnDiffResult, + type ProjectId, + type OrchestrationGetFullThreadDiffInput, + type OrchestrationGetFullThreadDiffResult, + type OrchestrationGetTurnDiffResult as OrchestrationGetTurnDiffResultType, +} from "@t3tools/contracts"; +import { Effect, Layer, Schema } from "effect"; + +import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { CheckpointInvariantError, CheckpointUnavailableError } from "../Errors.ts"; +import { checkpointRefForThreadTurn } from "../Refs.ts"; +import { CheckpointStore } from "../Services/CheckpointStore.ts"; +import { CheckpointDiffQuery, type CheckpointDiffQueryShape } from "../Services/CheckpointDiffQuery.ts"; + +function trimToNonEmpty(value: string | null | undefined): string | undefined { + const normalized = value?.trim(); + return normalized && normalized.length > 0 ? normalized : undefined; +} + +function resolveThreadWorkspaceCwd(input: { + readonly thread: { + readonly projectId: ProjectId; + readonly worktreePath: string | null; + }; + readonly projects: ReadonlyArray<{ + readonly id: ProjectId; + readonly workspaceRoot: string; + }>; +}): string | undefined { + const worktreeCwd = trimToNonEmpty(input.thread.worktreePath); + if (worktreeCwd) { + return worktreeCwd; + } + + const workspaceRoot = input.projects.find( + (project) => project.id === input.thread.projectId, + )?.workspaceRoot; + return trimToNonEmpty(workspaceRoot); +} + +const isTurnDiffResult = Schema.is(OrchestrationGetTurnDiffResult); + +const make = Effect.gen(function* () { + const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + const checkpointStore = yield* CheckpointStore; + + const getTurnDiff: CheckpointDiffQueryShape["getTurnDiff"] = (input) => + Effect.gen(function* () { + const operation = "CheckpointDiffQuery.getTurnDiff"; + + if (input.fromTurnCount === input.toTurnCount) { + const emptyDiff: OrchestrationGetTurnDiffResultType = { + threadId: input.threadId, + fromTurnCount: input.fromTurnCount, + toTurnCount: input.toTurnCount, + diff: "", + }; + if (!isTurnDiffResult(emptyDiff)) { + return yield* new CheckpointInvariantError({ + operation, + detail: "Computed turn diff result does not satisfy contract schema.", + }); + } + return emptyDiff; + } + + const snapshot = yield* projectionSnapshotQuery.getSnapshot(); + const thread = snapshot.threads.find((entry) => entry.id === input.threadId); + if (!thread) { + return yield* new CheckpointInvariantError({ + operation, + detail: `Thread '${input.threadId}' not found.`, + }); + } + + const maxTurnCount = thread.checkpoints.reduce( + (max, checkpoint) => Math.max(max, checkpoint.checkpointTurnCount), + 0, + ); + if (input.toTurnCount > maxTurnCount) { + return yield* new CheckpointUnavailableError({ + threadId: input.threadId, + turnCount: input.toTurnCount, + detail: `Turn diff range exceeds current turn count: requested ${input.toTurnCount}, current ${maxTurnCount}.`, + }); + } + + const workspaceCwd = resolveThreadWorkspaceCwd({ + thread, + projects: snapshot.projects, + }); + if (!workspaceCwd) { + return yield* new CheckpointInvariantError({ + operation, + detail: `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) { + return yield* new CheckpointUnavailableError({ + threadId: input.threadId, + turnCount: input.fromTurnCount, + detail: `Checkpoint ref is unavailable for turn ${input.fromTurnCount}.`, + }); + } + + const toCheckpointRef = thread.checkpoints.find( + (checkpoint) => checkpoint.checkpointTurnCount === input.toTurnCount, + )?.checkpointRef; + if (!toCheckpointRef) { + return yield* new CheckpointUnavailableError({ + threadId: input.threadId, + turnCount: input.toTurnCount, + detail: `Checkpoint ref is unavailable for turn ${input.toTurnCount}.`, + }); + } + + const [fromExists, toExists] = yield* Effect.all( + [ + checkpointStore.hasCheckpointRef({ + cwd: workspaceCwd, + checkpointRef: fromCheckpointRef, + }), + checkpointStore.hasCheckpointRef({ + cwd: workspaceCwd, + checkpointRef: toCheckpointRef, + }), + ], + { concurrency: "unbounded" }, + ); + + if (!fromExists) { + return yield* new CheckpointUnavailableError({ + threadId: input.threadId, + turnCount: input.fromTurnCount, + detail: `Filesystem checkpoint is unavailable for turn ${input.fromTurnCount}.`, + }); + } + + if (!toExists) { + return yield* new CheckpointUnavailableError({ + threadId: input.threadId, + turnCount: input.toTurnCount, + detail: `Filesystem checkpoint is unavailable for turn ${input.toTurnCount}.`, + }); + } + + const diff = yield* checkpointStore.diffCheckpoints({ + cwd: workspaceCwd, + fromCheckpointRef, + toCheckpointRef, + fallbackFromToHead: false, + }); + + const turnDiff: OrchestrationGetTurnDiffResultType = { + threadId: input.threadId, + fromTurnCount: input.fromTurnCount, + toTurnCount: input.toTurnCount, + diff, + }; + if (!isTurnDiffResult(turnDiff)) { + return yield* new CheckpointInvariantError({ + operation, + detail: "Computed turn diff result does not satisfy contract schema.", + }); + } + + return turnDiff; + }); + + const getFullThreadDiff: CheckpointDiffQueryShape["getFullThreadDiff"] = ( + input: OrchestrationGetFullThreadDiffInput, + ) => + getTurnDiff({ + threadId: input.threadId, + fromTurnCount: 0, + toTurnCount: input.toTurnCount, + }).pipe(Effect.map((result): OrchestrationGetFullThreadDiffResult => result)); + + return { + getTurnDiff, + getFullThreadDiff, + } satisfies CheckpointDiffQueryShape; +}); + +export const CheckpointDiffQueryLive = Layer.effect(CheckpointDiffQuery, make); diff --git a/apps/server/src/checkpointing/Refs.ts b/apps/server/src/checkpointing/Refs.ts new file mode 100644 index 0000000000000000000000000000000000000000..3da640f5457bd3e2840a078f87ad02638b9c143d --- /dev/null +++ b/apps/server/src/checkpointing/Refs.ts @@ -0,0 +1,10 @@ +import { Encoding } from "effect"; +import { CheckpointRef, type ThreadId } from "@t3tools/contracts"; + +export const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints"; + +export function checkpointRefForThreadTurn(threadId: ThreadId, turnCount: number): CheckpointRef { + return CheckpointRef.makeUnsafe( + `${CHECKPOINT_REFS_PREFIX}/${Encoding.encodeBase64Url(threadId)}/turn/${turnCount}`, + ); +} diff --git a/apps/server/src/checkpointing/Services/CheckpointDiffQuery.ts b/apps/server/src/checkpointing/Services/CheckpointDiffQuery.ts new file mode 100644 index 0000000000000000000000000000000000000000..ea6ef17c032bd95cd5d71a1bcd6579b8f08f648a --- /dev/null +++ b/apps/server/src/checkpointing/Services/CheckpointDiffQuery.ts @@ -0,0 +1,24 @@ +import type { + OrchestrationGetFullThreadDiffInput, + OrchestrationGetFullThreadDiffResult, + OrchestrationGetTurnDiffInput, + OrchestrationGetTurnDiffResult, +} from "@t3tools/contracts"; +import { ServiceMap } from "effect"; +import type { Effect } from "effect"; + +import type { CheckpointServiceError } from "../Errors.ts"; + +export interface CheckpointDiffQueryShape { + readonly getTurnDiff: ( + input: OrchestrationGetTurnDiffInput, + ) => Effect.Effect; + readonly getFullThreadDiff: ( + input: OrchestrationGetFullThreadDiffInput, + ) => Effect.Effect; +} + +export class CheckpointDiffQuery extends ServiceMap.Service< + CheckpointDiffQuery, + CheckpointDiffQueryShape +>()("checkpointing/CheckpointDiffQuery") {} diff --git a/apps/server/src/git.ts b/apps/server/src/git.ts index d41c3577300fa3a746f164773bedc72df759c8ce..908088112c6773376238ac06993c5f5877796c61 100644 --- a/apps/server/src/git.ts +++ b/apps/server/src/git.ts @@ -216,7 +216,7 @@ export class GitCoreService { async status(input: GitStatusInput): Promise { const details = await this.statusDetails(input.cwd); - return GitStatusResult.makeUnsafe({ + return { branch: details.branch, hasWorkingTreeChanges: details.hasWorkingTreeChanges, workingTree: details.workingTree, @@ -224,7 +224,7 @@ export class GitCoreService { aheadCount: details.aheadCount, behindCount: details.behindCount, openPr: null, - }); + }; } async statusDetails(cwd: string): Promise { @@ -385,11 +385,11 @@ export class GitCoreService { }); const afterSha = trimStdout(await this.gitStdout(cwd, ["rev-parse", "HEAD"], true)); const refreshed = await this.statusDetails(cwd); - return GitPullResult.makeUnsafe({ + return { status: beforeSha.length > 0 && beforeSha === afterSha ? "skipped_up_to_date" : "pulled", branch, upstreamBranch: refreshed.upstreamRef, - }); + }; } async readRangeContext(cwd: string, baseBranch: string): Promise { diff --git a/apps/server/src/keybindings.test.ts b/apps/server/src/keybindings.test.ts index 95f6cdb50cb9859cf9b52b1277be5307b2af0286..690e7d7d08e84feb48b798bdb7cabbae2f5c62fa 100644 --- a/apps/server/src/keybindings.test.ts +++ b/apps/server/src/keybindings.test.ts @@ -40,7 +40,7 @@ const writeKeybindingsConfig = (configPath: string, rules: readonly KeybindingRu Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const encoded = yield* Schema.encodeEffect(KeybindingsConfigJson)( - rules.map((rule) => KeybindingRule.makeUnsafe(rule)), + rules, ); yield* fileSystem.writeFileString(configPath, encoded); }); diff --git a/apps/server/src/keybindings.ts b/apps/server/src/keybindings.ts index 9942b5c61274412f1ab45a970dc65610e67725c5..f591cb983c8d1e0e91cabd63331e2990d6f6d585 100644 --- a/apps/server/src/keybindings.ts +++ b/apps/server/src/keybindings.ts @@ -313,7 +313,7 @@ export const ResolvedKeybindingFromConfig = KeybindingRule.pipe( title: "Invalid keybinding rule", }), ), - Effect.map((resolved) => ResolvedKeybindingRule.makeUnsafe(resolved)), + Effect.map((resolved) => resolved), ), encode: (resolved) => @@ -456,7 +456,7 @@ const makeKeybindings = Effect.gen(function* () { const tempPath = `${keybindingsConfigPath}.${process.pid}.${Date.now()}.tmp`; return Schema.encodeEffect(KeybindingsConfigJson)( - rules.map((rule) => KeybindingRule.makeUnsafe(rule)), + rules, ).pipe( Effect.tap(() => fs.makeDirectory(path.dirname(keybindingsConfigPath), { recursive: true })), Effect.tap((encoded) => fs.writeFileString(tempPath, encoded)), diff --git a/apps/server/src/orchestration/Layer.test.ts b/apps/server/src/orchestration/Layer.test.ts index 242f2c8661eae6e8282f5a110fb8bd4402bacfc8..1ccf75579fa93ef7cae818baeafe2ff3d22ef5a0 100644 --- a/apps/server/src/orchestration/Layer.test.ts +++ b/apps/server/src/orchestration/Layer.test.ts @@ -22,9 +22,7 @@ import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./Layers/ProjectionPipeline.ts"; import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; -const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); -const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value); @@ -58,7 +56,7 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-1-create"), + commandId: CommandId.makeUnsafe("cmd-project-1-create"), projectId: asProjectId("project-1"), title: "Project 1", workspaceRoot: "/tmp/project-1", @@ -69,8 +67,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-1-create"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-thread-1-create"), + threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", @@ -82,8 +80,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.turn.start", - commandId: asCommandId("cmd-turn-start-1"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-turn-start-1"), + threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("msg-1"), role: "user", @@ -108,7 +106,7 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-replay-create"), + commandId: CommandId.makeUnsafe("cmd-project-replay-create"), projectId: asProjectId("project-replay"), title: "Replay Project", workspaceRoot: "/tmp/project-replay", @@ -119,8 +117,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-replay-create"), - threadId: asThreadId("thread-replay"), + commandId: CommandId.makeUnsafe("cmd-thread-replay-create"), + threadId: ThreadId.makeUnsafe("thread-replay"), projectId: asProjectId("project-replay"), title: "replay", model: "gpt-5-codex", @@ -132,8 +130,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.delete", - commandId: asCommandId("cmd-thread-replay-delete"), - threadId: asThreadId("thread-replay"), + commandId: CommandId.makeUnsafe("cmd-thread-replay-delete"), + threadId: ThreadId.makeUnsafe("thread-replay"), }), ); @@ -158,7 +156,7 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-stream-create"), + commandId: CommandId.makeUnsafe("cmd-project-stream-create"), projectId: asProjectId("project-stream"), title: "Stream Project", workspaceRoot: "/tmp/project-stream", @@ -179,8 +177,8 @@ describe("OrchestrationEngine", () => { yield* Effect.sleep("10 millis"); yield* engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-stream-thread-create"), - threadId: asThreadId("thread-stream"), + commandId: CommandId.makeUnsafe("cmd-stream-thread-create"), + threadId: ThreadId.makeUnsafe("thread-stream"), projectId: asProjectId("project-stream"), title: "domain-stream", model: "gpt-5-codex", @@ -190,8 +188,8 @@ describe("OrchestrationEngine", () => { }); yield* engine.dispatch({ type: "thread.meta.update", - commandId: asCommandId("cmd-stream-thread-update"), - threadId: asThreadId("thread-stream"), + commandId: CommandId.makeUnsafe("cmd-stream-thread-update"), + threadId: ThreadId.makeUnsafe("thread-stream"), title: "domain-stream-updated", }); eventTypes.push((yield* Queue.take(eventQueue)).type); @@ -211,7 +209,7 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-turn-diff-create"), + commandId: CommandId.makeUnsafe("cmd-project-turn-diff-create"), projectId: asProjectId("project-turn-diff"), title: "Turn Diff Project", workspaceRoot: "/tmp/project-turn-diff", @@ -222,8 +220,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-turn-diff-create"), - threadId: asThreadId("thread-turn-diff"), + commandId: CommandId.makeUnsafe("cmd-thread-turn-diff-create"), + threadId: ThreadId.makeUnsafe("thread-turn-diff"), projectId: asProjectId("project-turn-diff"), title: "Turn diff thread", model: "gpt-5-codex", @@ -235,8 +233,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.turn.diff.complete", - commandId: asCommandId("cmd-turn-diff-complete"), - threadId: asThreadId("thread-turn-diff"), + commandId: CommandId.makeUnsafe("cmd-turn-diff-complete"), + threadId: ThreadId.makeUnsafe("thread-turn-diff"), turnId: asTurnId("turn-1"), completedAt: createdAt, checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"), @@ -275,7 +273,7 @@ describe("OrchestrationEngine", () => { const flakyStore: OrchestrationEventStoreShape = { append(event) { - if (shouldFailFirstAppend && event.commandId === asCommandId("cmd-flaky-1")) { + if (shouldFailFirstAppend && event.commandId === CommandId.makeUnsafe("cmd-flaky-1")) { shouldFailFirstAppend = false; return Effect.fail( new PersistenceSqlError({ @@ -314,7 +312,7 @@ describe("OrchestrationEngine", () => { await runtime.runPromise( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-flaky-create"), + commandId: CommandId.makeUnsafe("cmd-project-flaky-create"), projectId: asProjectId("project-flaky"), title: "Flaky Project", workspaceRoot: "/tmp/project-flaky", @@ -327,8 +325,8 @@ describe("OrchestrationEngine", () => { runtime.runPromise( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-flaky-1"), - threadId: asThreadId("thread-flaky-fail"), + commandId: CommandId.makeUnsafe("cmd-flaky-1"), + threadId: ThreadId.makeUnsafe("thread-flaky-fail"), projectId: asProjectId("project-flaky"), title: "flaky-fail", model: "gpt-5-codex", @@ -342,8 +340,8 @@ describe("OrchestrationEngine", () => { const result = await runtime.runPromise( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-flaky-2"), - threadId: asThreadId("thread-flaky-ok"), + commandId: CommandId.makeUnsafe("cmd-flaky-2"), + threadId: ThreadId.makeUnsafe("thread-flaky-ok"), projectId: asProjectId("project-flaky"), title: "flaky-ok", model: "gpt-5-codex", @@ -366,8 +364,8 @@ describe("OrchestrationEngine", () => { system.run( engine.dispatch({ type: "thread.turn.start", - commandId: asCommandId("cmd-invariant-missing-thread"), - threadId: asThreadId("thread-missing"), + commandId: CommandId.makeUnsafe("cmd-invariant-missing-thread"), + threadId: ThreadId.makeUnsafe("thread-missing"), message: { messageId: asMessageId("msg-missing"), role: "user", @@ -390,7 +388,7 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-duplicate-create"), + commandId: CommandId.makeUnsafe("cmd-project-duplicate-create"), projectId: asProjectId("project-duplicate"), title: "Duplicate Project", workspaceRoot: "/tmp/project-duplicate", @@ -402,8 +400,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-duplicate-1"), - threadId: asThreadId("thread-duplicate"), + commandId: CommandId.makeUnsafe("cmd-thread-duplicate-1"), + threadId: ThreadId.makeUnsafe("thread-duplicate"), projectId: asProjectId("project-duplicate"), title: "duplicate", model: "gpt-5-codex", @@ -417,8 +415,8 @@ describe("OrchestrationEngine", () => { system.run( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-duplicate-2"), - threadId: asThreadId("thread-duplicate"), + commandId: CommandId.makeUnsafe("cmd-thread-duplicate-2"), + threadId: ThreadId.makeUnsafe("thread-duplicate"), projectId: asProjectId("project-duplicate"), title: "duplicate", model: "gpt-5-codex", diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 5985a2219c8347ac16aba60f2aa14f5f71b58149..99b531db04b11df219046ac0061eefc42b4eec19 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -5,7 +5,6 @@ import { execFileSync } from "node:child_process"; import type { ProviderRuntimeEvent, ProviderSession } from "@t3tools/contracts"; import { - CheckpointRef, CommandId, EventId, MessageId, @@ -36,19 +35,13 @@ import { ProviderService, type ProviderServiceShape, } from "../../provider/Services/ProviderService.ts"; +import { checkpointRefForThreadTurn } from "../../checkpointing/Refs.ts"; -const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); -const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); -function checkpointRefForTurn(threadId: ThreadId, turnCount: number): CheckpointRef { - const encodedThreadId = Buffer.from(threadId, "utf8").toString("base64url"); - return CheckpointRef.makeUnsafe(`refs/t3/checkpoints/${encodedThreadId}/turn/${turnCount}`); -} - function createProviderServiceHarness(cwd: string, hasSession = true) { const now = new Date().toISOString(); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); @@ -109,7 +102,7 @@ async function waitForThread( activities: ReadonlyArray<{ kind: string }>; }> => { const readModel = await Effect.runPromise(engine.getReadModel()); - const thread = readModel.threads.find((entry) => entry.id === asThreadId("thread-1")); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); if (thread && predicate(thread)) { return thread; } @@ -250,7 +243,7 @@ describe("CheckpointReactor", () => { await Effect.runPromise( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-create"), + commandId: CommandId.makeUnsafe("cmd-project-create"), projectId: asProjectId("project-1"), title: "Test Project", workspaceRoot: options?.projectWorkspaceRoot ?? cwd, @@ -261,8 +254,8 @@ describe("CheckpointReactor", () => { await Effect.runPromise( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-create"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-thread-create"), + threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", @@ -276,21 +269,21 @@ describe("CheckpointReactor", () => { await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, - checkpointRef: checkpointRefForTurn(asThreadId("thread-1"), 0), + checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), }), ); fs.writeFileSync(path.join(cwd, "README.md"), "v2\n", "utf8"); await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, - checkpointRef: checkpointRefForTurn(asThreadId("thread-1"), 1), + checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1), }), ); fs.writeFileSync(path.join(cwd, "README.md"), "v3\n", "utf8"); await runtime.runPromise( checkpointStore.captureCheckpoint({ cwd, - checkpointRef: checkpointRefForTurn(asThreadId("thread-1"), 2), + checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 2), }), ); } @@ -309,10 +302,10 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", - commandId: asCommandId("cmd-session-set-capture"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-set-capture"), + threadId: ThreadId.makeUnsafe("thread-1"), session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "ready", providerName: "codex", providerSessionId: asSessionId("sess-1"), @@ -334,7 +327,7 @@ describe("CheckpointReactor", () => { threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), turnId: asTurnId("turn-1"), }); - await waitForGitRefExists(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 0)); + await waitForGitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0)); fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8"); harness.provider.emit({ @@ -354,13 +347,25 @@ describe("CheckpointReactor", () => { (entry) => entry.latestTurnId === "turn-1" && entry.checkpoints.length === 1, ); expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); - expect(gitRefExists(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 0))).toBe(true); - expect(gitRefExists(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 1))).toBe(true); + expect(gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0))).toBe( + true, + ); + expect(gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1))).toBe( + true, + ); expect( - gitShowFileAtRef(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 0), "README.md"), + gitShowFileAtRef( + harness.cwd, + checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), + "README.md", + ), ).toBe("v1\n"); expect( - gitShowFileAtRef(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 1), "README.md"), + gitShowFileAtRef( + harness.cwd, + checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1), + "README.md", + ), ).toBe("v2\n"); }); @@ -374,8 +379,8 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", - commandId: asCommandId("cmd-turn-start-for-baseline"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-turn-start-for-baseline"), + threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: MessageId.makeUnsafe("message-user-1"), role: "user", @@ -386,9 +391,13 @@ describe("CheckpointReactor", () => { }), ); - await waitForGitRefExists(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 0)); + await waitForGitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0)); expect( - gitShowFileAtRef(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 0), "README.md"), + gitShowFileAtRef( + harness.cwd, + checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), + "README.md", + ), ).toBe("v1\n"); }); @@ -403,10 +412,10 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", - commandId: asCommandId("cmd-session-set-missing-provider-cwd"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-set-missing-provider-cwd"), + threadId: ThreadId.makeUnsafe("thread-1"), session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "running", providerName: "codex", providerSessionId: asSessionId("sess-missing"), @@ -432,9 +441,15 @@ describe("CheckpointReactor", () => { }); await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); - expect(gitRefExists(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 1))).toBe(true); + expect(gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1))).toBe( + true, + ); expect( - gitShowFileAtRef(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 1), "README.md"), + gitShowFileAtRef( + harness.cwd, + checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1), + "README.md", + ), ).toBe("v2\n"); }); @@ -445,10 +460,10 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", - commandId: asCommandId("cmd-session-set-checkpoint-captured"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-set-checkpoint-captured"), + threadId: ThreadId.makeUnsafe("thread-1"), session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "ready", providerName: "codex", providerSessionId: asSessionId("sess-1"), @@ -488,10 +503,10 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", - commandId: asCommandId("cmd-session-set"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-set"), + threadId: ThreadId.makeUnsafe("thread-1"), session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "ready", providerName: "codex", providerSessionId: asSessionId("sess-1"), @@ -507,11 +522,11 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.diff.complete", - commandId: asCommandId("cmd-diff-1"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-diff-1"), + threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), completedAt: createdAt, - checkpointRef: checkpointRefForTurn(asThreadId("thread-1"), 1), + checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 1), status: "ready", files: [], checkpointTurnCount: 1, @@ -521,11 +536,11 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.diff.complete", - commandId: asCommandId("cmd-diff-2"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-diff-2"), + threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-2"), completedAt: createdAt, - checkpointRef: checkpointRefForTurn(asThreadId("thread-1"), 2), + checkpointRef: checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 2), status: "ready", files: [], checkpointTurnCount: 2, @@ -536,8 +551,8 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.checkpoint.revert", - commandId: asCommandId("cmd-revert-request"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-revert-request"), + threadId: ThreadId.makeUnsafe("thread-1"), turnCount: 1, createdAt, }), @@ -555,7 +570,9 @@ describe("CheckpointReactor", () => { numTurns: 1, }); expect(fs.readFileSync(path.join(harness.cwd, "README.md"), "utf8")).toBe("v2\n"); - expect(gitRefExists(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 2))).toBe(false); + expect(gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 2))).toBe( + false, + ); }); it("appends an error activity when revert is requested without an active session", async () => { @@ -565,8 +582,8 @@ describe("CheckpointReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.checkpoint.revert", - commandId: asCommandId("cmd-revert-no-session"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-revert-no-session"), + threadId: ThreadId.makeUnsafe("thread-1"), turnCount: 1, createdAt, }), diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 2be629d7c53927c44e4ccf176d7ba40fe4146b88..234bb6f95cbf5a8da67346a79869e8c1d459eb6d 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -1,17 +1,18 @@ import { - CheckpointRef, CommandId, EventId, MessageId, ProviderSessionId, ThreadId, TurnId, + type ProjectId, type OrchestrationEvent, type ProviderRuntimeEvent, } from "@t3tools/contracts"; import { Effect, Layer, Option, Queue, Stream } from "effect"; import { parseTurnDiffFilesFromUnifiedDiff } from "../../checkpointing/Diffs.ts"; +import { checkpointRefForThreadTurn } from "../../checkpointing/Refs.ts"; import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { CheckpointReactor, type CheckpointReactorShape } from "../Services/CheckpointReactor.ts"; @@ -19,8 +20,6 @@ import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { CheckpointStoreError } from "../../checkpointing/Errors.ts"; import { OrchestrationDispatchError } from "../Errors.ts"; -const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints"; - type ReactorInput = | { readonly source: "runtime"; @@ -31,11 +30,6 @@ type ReactorInput = readonly event: OrchestrationEvent; }; -function checkpointRefForThreadTurn(threadId: ThreadId, turnCount: number): CheckpointRef { - const encodedThreadId = Buffer.from(threadId, "utf8").toString("base64url"); - return CheckpointRef.makeUnsafe(`${CHECKPOINT_REFS_PREFIX}/${encodedThreadId}/turn/${turnCount}`); -} - function toTurnId(value: string | undefined): TurnId | null { const normalized = value?.trim(); return normalized && normalized.length > 0 ? TurnId.makeUnsafe(normalized) : null; @@ -48,11 +42,11 @@ function trimToNonEmpty(value: string | null | undefined): string | undefined { function resolveThreadWorkspaceCwd(input: { readonly thread: { - readonly projectId: string; + readonly projectId: ProjectId; readonly worktreePath: string | null; }; readonly projects: ReadonlyArray<{ - readonly id: string; + readonly id: ProjectId; readonly workspaceRoot: string; }>; }): string | undefined { diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 5017c177f5cce0bfb38661ee19b5d023119c4883..52734f063fb41b7d1f5ff312a5222cfd5b1a00c7 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -5,13 +5,11 @@ import type { ThreadId, } from "@t3tools/contracts"; import { OrchestrationCommand } from "@t3tools/contracts"; -import { Deferred, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect"; +import { Deferred, Effect, Layer, Option, PubSub, Queue, Stream } from "effect"; import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepository } from "../../persistence/Services/OrchestrationCommandReceipts.ts"; import { - toOrchestrationCommandDecodeError, - toOrchestrationJsonParseError, OrchestrationCommandInvariantError, OrchestrationCommandPreviouslyRejectedError, type OrchestrationDispatchError, @@ -29,8 +27,6 @@ interface CommandEnvelope { result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>; } -const decodeUnknownCommand = Schema.decodeUnknownEffect(OrchestrationCommand); - function commandToAggregateRef(command: OrchestrationCommand): { readonly aggregateKind: "project" | "thread"; readonly aggregateId: ProjectId | ThreadId; @@ -175,30 +171,12 @@ const makeOrchestrationEngine = Effect.gen(function* () { return yield* Deferred.await(result); }); - const dispatchUnknownCommand: OrchestrationEngineShape["dispatchUnknownCommand"] = (command) => - Effect.gen(function* () { - const payload = - typeof command === "string" - ? yield* Effect.try({ - try: () => JSON.parse(command) as unknown, - catch: toOrchestrationJsonParseError, - }) - : command; - - const decoded = yield* decodeUnknownCommand(payload).pipe( - Effect.mapError(toOrchestrationCommandDecodeError), - ); - - return yield* dispatch(decoded); - }); - const streamDomainEvents: OrchestrationEngineShape["streamDomainEvents"] = Stream.fromPubSub(eventPubSub); return { getReadModel, readEvents, - dispatchUnknownCommand, dispatch, streamDomainEvents, } satisfies OrchestrationEngineShape; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 3e55887e20bdd87a26bf3b0e1031a71d14b0108a..f19b2891cce1d94214dd0bce15f8641cc18f1147 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -18,7 +18,6 @@ import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQu import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); -const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); @@ -225,7 +224,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { ]); assert.deepEqual(snapshot.threads, [ { - id: asThreadId("thread-1"), + id: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread 1", model: "gpt-5-codex", @@ -269,7 +268,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { }, ], session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "running", providerName: "codex", providerSessionId: asProviderSessionId("provider-session-1"), diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 0b65741a3f2cf15c8906ad09591426a80058e945..f028349f58752b8cab4321c8e078419db6af12cc 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -26,9 +26,7 @@ import { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; -const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); -const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asApprovalRequestId = (value: string): ApprovalRequestId => @@ -126,7 +124,7 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-project-create"), + commandId: CommandId.makeUnsafe("cmd-project-create"), projectId: asProjectId("project-1"), title: "Provider Project", workspaceRoot: "/tmp/provider-project", @@ -137,8 +135,8 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-create"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-thread-create"), + threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", @@ -165,8 +163,8 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", - commandId: asCommandId("cmd-turn-start-1"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-turn-start-1"), + threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("user-message-1"), role: "user", @@ -185,7 +183,7 @@ describe("ProviderCommandReactor", () => { }); const readModel = await Effect.runPromise(harness.engine.getReadModel()); - const thread = readModel.threads.find((entry) => entry.id === asThreadId("thread-1")); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); expect(thread?.session?.providerSessionId).toBe("sess-1"); }); @@ -196,10 +194,10 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", - commandId: asCommandId("cmd-session-set"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-set"), + threadId: ThreadId.makeUnsafe("thread-1"), session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "running", providerName: "codex", providerSessionId: asSessionId("sess-1"), @@ -215,8 +213,8 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.interrupt", - commandId: asCommandId("cmd-turn-interrupt"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-turn-interrupt"), + threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), createdAt: now, }), @@ -235,10 +233,10 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", - commandId: asCommandId("cmd-session-set-for-approval"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-set-for-approval"), + threadId: ThreadId.makeUnsafe("thread-1"), session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "running", providerName: "codex", providerSessionId: asSessionId("sess-1"), @@ -254,8 +252,8 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.approval.respond", - commandId: asCommandId("cmd-approval-respond"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-approval-respond"), + threadId: ThreadId.makeUnsafe("thread-1"), requestId: asApprovalRequestId("approval-request-1"), decision: "accept", createdAt: now, @@ -277,10 +275,10 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", - commandId: asCommandId("cmd-session-set-for-stop"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-set-for-stop"), + threadId: ThreadId.makeUnsafe("thread-1"), session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "ready", providerName: "codex", providerSessionId: asSessionId("sess-1"), @@ -296,15 +294,15 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.stop", - commandId: asCommandId("cmd-session-stop"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-stop"), + threadId: ThreadId.makeUnsafe("thread-1"), createdAt: now, }), ); await waitFor(() => harness.stopSession.mock.calls.length === 1); const readModel = await Effect.runPromise(harness.engine.getReadModel()); - const thread = readModel.threads.find((entry) => entry.id === asThreadId("thread-1")); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); expect(thread?.session).not.toBeNull(); expect(thread?.session?.status).toBe("stopped"); expect(thread?.session?.providerSessionId).toBeNull(); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 9d2f3bf80eb0c816fe80c9819e5e6a79cc94b8da..d39fe290be189e7b484950ee658db567f8ca0841 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -28,9 +28,7 @@ import { } from "../Services/OrchestrationEngine.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; -const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); -const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value); @@ -78,7 +76,7 @@ async function waitForThread( activities: ReadonlyArray<{ kind: string }>; }> => { const readModel = await Effect.runPromise(engine.getReadModel()); - const thread = readModel.threads.find((entry) => entry.id === asThreadId("thread-1")); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); if (thread && predicate(thread)) { return thread; } @@ -132,7 +130,7 @@ describe("ProviderRuntimeIngestion", () => { await Effect.runPromise( engine.dispatch({ type: "project.create", - commandId: asCommandId("cmd-provider-project-create"), + commandId: CommandId.makeUnsafe("cmd-provider-project-create"), projectId: asProjectId("project-1"), title: "Provider Project", workspaceRoot: "/tmp/provider-project", @@ -143,8 +141,8 @@ describe("ProviderRuntimeIngestion", () => { await Effect.runPromise( engine.dispatch({ type: "thread.create", - commandId: asCommandId("cmd-thread-create"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-thread-create"), + threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", @@ -156,10 +154,10 @@ describe("ProviderRuntimeIngestion", () => { await Effect.runPromise( engine.dispatch({ type: "thread.session.set", - commandId: asCommandId("cmd-session-seed"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-session-seed"), + threadId: ThreadId.makeUnsafe("thread-1"), session: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), status: "ready", providerName: "codex", providerSessionId: asSessionId("sess-1"), diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts index 2287f2ae7f1686f81301c27e905f019b2c170072..ed2bf6a35fe20ceafcdeef50a54f2a877beb9f19 100644 --- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts @@ -18,7 +18,7 @@ import type { import { ServiceMap } from "effect"; import type { Effect, Stream } from "effect"; -import type { OrchestrationDispatchError, OrchestrationEngineError } from "../Errors.ts"; +import type { OrchestrationDispatchError } from "../Errors.ts"; import type { OrchestrationEventStoreError } from "../../persistence/Errors.ts"; export interface OrchestrationEngineShape { @@ -39,18 +39,6 @@ export interface OrchestrationEngineShape { fromSequenceExclusive: number, ) => Stream.Stream; - /** - * Validate and dispatch an unknown command payload. - * - * Accepts either parsed values or raw JSON strings. - * - * @param command - Unknown command payload. - * @returns Effect containing the sequence of the persisted event. - */ - readonly dispatchUnknownCommand: ( - command: unknown, - ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError, never>; - /** * Dispatch a validated orchestration command. * diff --git a/apps/server/src/orchestration/decider.projectScripts.test.ts b/apps/server/src/orchestration/decider.projectScripts.test.ts index c56d427fb126e7553723c986d775d23938d03f48..78fbb858a0b236b69e341b9e51055f497a0df2e3 100644 --- a/apps/server/src/orchestration/decider.projectScripts.test.ts +++ b/apps/server/src/orchestration/decider.projectScripts.test.ts @@ -5,10 +5,8 @@ import { Effect } from "effect"; import { decideOrchestrationCommand } from "./decider.ts"; import { createEmptyReadModel, projectEvent } from "./projector.ts"; -const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); -const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); describe("decider project scripts", () => { @@ -20,7 +18,7 @@ describe("decider project scripts", () => { decideOrchestrationCommand({ command: { type: "project.create", - commandId: asCommandId("cmd-project-create-scripts"), + commandId: CommandId.makeUnsafe("cmd-project-create-scripts"), projectId: asProjectId("project-scripts"), title: "Scripts", workspaceRoot: "/tmp/scripts", @@ -46,9 +44,9 @@ describe("decider project scripts", () => { aggregateId: asProjectId("project-scripts"), type: "project.created", occurredAt: now, - commandId: asCommandId("cmd-project-create-scripts"), + commandId: CommandId.makeUnsafe("cmd-project-create-scripts"), causationEventId: null, - correlationId: asCommandId("cmd-project-create-scripts"), + correlationId: CommandId.makeUnsafe("cmd-project-create-scripts"), metadata: {}, payload: { projectId: asProjectId("project-scripts"), @@ -76,7 +74,7 @@ describe("decider project scripts", () => { decideOrchestrationCommand({ command: { type: "project.meta.update", - commandId: asCommandId("cmd-project-update-scripts"), + commandId: CommandId.makeUnsafe("cmd-project-update-scripts"), projectId: asProjectId("project-scripts"), scripts: Array.from(scripts), }, @@ -100,9 +98,9 @@ describe("decider project scripts", () => { aggregateId: asProjectId("project-1"), type: "project.created", occurredAt: now, - commandId: asCommandId("cmd-project-create"), + commandId: CommandId.makeUnsafe("cmd-project-create"), causationEventId: null, - correlationId: asCommandId("cmd-project-create"), + correlationId: CommandId.makeUnsafe("cmd-project-create"), metadata: {}, payload: { projectId: asProjectId("project-1"), @@ -120,15 +118,15 @@ describe("decider project scripts", () => { sequence: 2, eventId: asEventId("evt-thread-create"), aggregateKind: "thread", - aggregateId: asThreadId("thread-1"), + aggregateId: ThreadId.makeUnsafe("thread-1"), type: "thread.created", occurredAt: now, - commandId: asCommandId("cmd-thread-create"), + commandId: CommandId.makeUnsafe("cmd-thread-create"), causationEventId: null, - correlationId: asCommandId("cmd-thread-create"), + correlationId: CommandId.makeUnsafe("cmd-thread-create"), metadata: {}, payload: { - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", @@ -144,8 +142,8 @@ describe("decider project scripts", () => { decideOrchestrationCommand({ command: { type: "thread.turn.start", - commandId: asCommandId("cmd-turn-start"), - threadId: asThreadId("thread-1"), + commandId: CommandId.makeUnsafe("cmd-turn-start"), + threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("message-user-1"), role: "user", diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 87f33fad57fc19c21311a84dadac73f512a2a9f9..874bdea1d6ff2547c2d731f69c9b25d56812e261 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -23,7 +23,6 @@ import { CodexAdapter } from "../Services/CodexAdapter.ts"; import { makeCodexAdapterLive } from "./CodexAdapter.ts"; const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); -const asThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value); @@ -36,7 +35,7 @@ class FakeCodexManager extends CodexAppServerManager { sessionId: asSessionId("sess-1"), provider: "codex", status: "ready", - threadId: asThreadId("thread-1"), + threadId: ProviderThreadId.makeUnsafe("thread-1"), cwd: input.cwd, createdAt: now, updatedAt: now, @@ -46,7 +45,7 @@ class FakeCodexManager extends CodexAppServerManager { public sendTurnImpl = vi.fn( async (_input: ProviderSendTurnInput): Promise => ({ - threadId: asThreadId("thread-1"), + threadId: ProviderThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), }), ); @@ -56,12 +55,12 @@ class FakeCodexManager extends CodexAppServerManager { ); public readThreadImpl = vi.fn(async (_sessionId: ProviderSessionId) => ({ - threadId: asThreadId("thread-1"), + threadId: ProviderThreadId.makeUnsafe("thread-1"), turns: [], })); public rollbackThreadImpl = vi.fn(async (_sessionId: ProviderSessionId, _numTurns: number) => ({ - threadId: asThreadId("thread-1"), + threadId: ProviderThreadId.makeUnsafe("thread-1"), turns: [], })); @@ -194,7 +193,7 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { sessionId: asSessionId("sess-1"), createdAt: new Date().toISOString(), method: "item/completed", - threadId: asThreadId("thread-1"), + threadId: ProviderThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), itemId: asItemId("msg_1"), payload: { diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 5d47828922c64e2fa3bee2e1ef761fe1c7e3d553..ff8cdd0ea9c1d733aa331a2c04ff51fa48374b92 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -21,7 +21,7 @@ import { import { it, assert, vi } from "@effect/vitest"; import { assertFailure } from "@effect/vitest/utils"; -import { Effect, Fiber, Layer, Option, PubSub, Ref, Schema, Stream } from "effect"; +import { Effect, Fiber, Layer, Option, PubSub, Ref, Stream } from "effect"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { @@ -46,7 +46,6 @@ import { } from "../../persistence/Layers/Sqlite.ts"; const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); -const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); @@ -568,7 +567,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), status: "completed", }; @@ -616,7 +615,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), toolKind: "command", title: "Command run", @@ -628,7 +627,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), delta: "hello", }, @@ -638,7 +637,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - threadId: asThreadId("thread-1"), + threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), status: "completed", }, diff --git a/apps/server/src/serverLayers.ts b/apps/server/src/serverLayers.ts index 314da061a0f9447e6420cecef6d148e5b6083019..38bea2269587730c939eb58352ed5eb828d271dc 100644 --- a/apps/server/src/serverLayers.ts +++ b/apps/server/src/serverLayers.ts @@ -4,8 +4,10 @@ import { NodeServices } from "@effect/platform-node"; import { Effect, Layer } from "effect"; import * as SqlClient from "effect/unstable/sql/SqlClient"; +import { CheckpointDiffQueryLive } from "./checkpointing/Layers/CheckpointDiffQuery"; import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore"; import { ServerConfig } from "./config"; +import { CheckpointDiffQuery } from "./checkpointing/Services/CheckpointDiffQuery"; import { CheckpointStore } from "./checkpointing/Services/CheckpointStore"; import { OrchestrationCommandReceiptRepositoryLive } from "./persistence/Layers/OrchestrationCommandReceipts"; import { OrchestrationEventStoreLive } from "./persistence/Layers/OrchestrationEventStore"; @@ -55,6 +57,7 @@ export function makeServerRuntimeServicesLayer(): Layer.Layer< | OrchestrationEngineService | ProjectionSnapshotQuery | CheckpointStore + | CheckpointDiffQuery | OrchestrationReactor | TerminalManager | Keybindings, @@ -67,11 +70,16 @@ export function makeServerRuntimeServicesLayer(): Layer.Layer< Layer.provide(OrchestrationCommandReceiptRepositoryLive), ); const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); + const checkpointDiffQueryLayer = CheckpointDiffQueryLive.pipe( + Layer.provideMerge(OrchestrationProjectionSnapshotQueryLive), + Layer.provideMerge(checkpointStoreLayer), + ); const runtimeServicesLayer = Layer.mergeAll( orchestrationLayer, OrchestrationProjectionSnapshotQueryLive, checkpointStoreLayer, + checkpointDiffQueryLayer, ); const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(runtimeServicesLayer), diff --git a/apps/server/src/terminalManager.test.ts b/apps/server/src/terminalManager.test.ts index 861215d922005fb483cce65bf1a0932eaa7bcf62..b0e4bd1c05818cc0d15b2869b1fbce02257cbe1e 100644 --- a/apps/server/src/terminalManager.test.ts +++ b/apps/server/src/terminalManager.test.ts @@ -11,6 +11,7 @@ import { afterEach, describe, expect, it } from "vitest"; import type { PtyAdapterShape, PtyExitEvent, PtyProcess, PtySpawnInput } from "./ptyAdapter"; import { TerminalManagerRuntime } from "./terminalManager"; +import { Encoding } from "effect"; class FakePtyProcess implements PtyProcess { readonly writes: string[] = []; @@ -107,15 +108,15 @@ function openInput(overrides: Partial = {}): TerminalOpenInpu } function historyLogName(threadId: string): string { - return `terminal_${Buffer.from(threadId, "utf8").toString("base64url")}.log`; + return `terminal_${Encoding.encodeBase64Url(threadId)}.log`; } function multiTerminalHistoryLogName(threadId: string, terminalId: string): string { - const threadPart = `terminal_${Buffer.from(threadId, "utf8").toString("base64url")}`; + const threadPart = `terminal_${Encoding.encodeBase64Url(threadId)}`; if (terminalId === DEFAULT_TERMINAL_ID) { return `${threadPart}.log`; } - return `${threadPart}_${Buffer.from(terminalId, "utf8").toString("base64url")}.log`; + return `${threadPart}_${Encoding.encodeBase64Url(terminalId)}.log`; } function historyLogPath(logsDir: string, threadId = "thread-1"): string { diff --git a/apps/server/src/terminalManager.ts b/apps/server/src/terminalManager.ts index a79722cc7cc347c16aeeb0ca67ddea609ade57de..0a531435e061a888913d32a9fd1a931c0e3738f8 100644 --- a/apps/server/src/terminalManager.ts +++ b/apps/server/src/terminalManager.ts @@ -13,7 +13,7 @@ import { type TerminalSessionSnapshot, type TerminalSessionStatus, } from "@t3tools/contracts"; -import { Effect, Layer, Schema, ServiceMap } from "effect"; +import { Effect, Encoding, Layer, Schema, ServiceMap } from "effect"; import { createLogger } from "./logger"; import { PtyAdapter, PtyAdapterShape, type PtyExitEvent, type PtyProcess } from "./ptyAdapter"; @@ -269,11 +269,11 @@ function legacySafeThreadId(threadId: string): string { } function toSafeThreadId(threadId: string): string { - return `terminal_${Buffer.from(threadId, "utf8").toString("base64url")}`; + return `terminal_${Encoding.encodeBase64Url(threadId)}`; } function toSafeTerminalId(terminalId: string): string { - return Buffer.from(terminalId, "utf8").toString("base64url"); + return Encoding.encodeBase64Url(terminalId); } function toSessionKey(threadId: string, terminalId: string): string { diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index 43e2284470e4bea6a8b12c9a3b22a7e2709caa1b..ca9bf2959906c62676a6563c7b854f14733f60cd 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -175,7 +175,7 @@ class MockTerminalManager implements TerminalManagerShape { this.sessions.delete(this.key(input.threadId, input.terminalId)); return; } - for (const key of [...this.sessions.keys()]) { + for (const key of this.sessions.keys()) { if (key.startsWith(`${input.threadId}\u0000`)) { this.sessions.delete(key); } @@ -837,6 +837,26 @@ describe("WebSocket Server", () => { expect(response.error?.message).toContain("Thread 'thread-missing' not found."); }); + it("returns error when requesting turn diff with an inverted range", async () => { + server = await createTestServer({ cwd: "/test" }); + const addr = server.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-any", + fromTurnCount: 2, + toTurnCount: 1, + }); + expect(response.result).toBeUndefined(); + expect(response.error?.message).toContain( + "fromTurnCount must be less than or equal to toTurnCount", + ); + }); + it("returns error when requesting full thread diff for unknown thread", async () => { server = await createTestServer({ cwd: "/test" }); const addr = server.address(); diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index c9b3f1795b25bfe1414d12452932da6f0465d992..f86a6c0ad164fedd16be75f916ebe349ae1d324b 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -4,19 +4,19 @@ import path from "node:path"; import type { Duplex } from "node:stream"; import { - CheckpointRef, + CommandId, ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS, + ProjectId, TerminalEvent, WS_CHANNELS, WS_METHODS, WebSocketRequest, - WebSocketResponse, WsPush, WsResponse, } from "@t3tools/contracts"; import { NodeHttpServer } from "@effect/platform-node"; -import { Cause, Effect, Exit, Layer, Schema, Scope, ServiceMap, Stream } from "effect"; +import { Cause, Effect, Exit, Layer, Schema, Scope, ServiceMap, Stream, Struct } from "effect"; import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger"; @@ -37,7 +37,7 @@ import { OrchestrationEngineService } from "./orchestration/Services/Orchestrati import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery"; import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor"; import { ProviderService } from "./provider/Services/ProviderService"; -import { CheckpointStore } from "./checkpointing/Services/CheckpointStore"; +import { CheckpointDiffQuery } from "./checkpointing/Services/CheckpointDiffQuery"; import { clamp } from "effect/Number"; import { Open } from "./open"; import { ServerConfig } from "./config"; @@ -133,22 +133,14 @@ function websocketRawToString(raw: unknown): string | null { return null; } -function stripRequestTag(body: T): Omit { - const { _tag: _ignoredTag, ...rest } = body; - 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}`); +function stripRequestTag(body: T) { + return Struct.omit(body, ["_tag"]); } export type ServerCoreRuntimeServices = | OrchestrationEngineService | ProjectionSnapshotQuery - | CheckpointStore + | CheckpointDiffQuery | OrchestrationReactor | ProviderService; @@ -301,7 +293,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E const orchestrationEngine = yield* OrchestrationEngineService; const projectionReadModelQuery = yield* ProjectionSnapshotQuery; - const liveCheckpointStore = yield* CheckpointStore; + const checkpointDiffQuery = yield* CheckpointDiffQuery; const liveProviderService = yield* ProviderService; const orchestrationReactor = yield* OrchestrationReactor; const { openInEditor } = yield* Open; @@ -338,10 +330,10 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E if (!existing) { const createdAt = new Date().toISOString(); const projectName = path.basename(cwd) || "project"; - yield* orchestrationEngine.dispatchUnknownCommand({ + yield* orchestrationEngine.dispatch({ type: "project.create", - commandId: crypto.randomUUID(), - projectId: crypto.randomUUID(), + commandId: CommandId.makeUnsafe(crypto.randomUUID()), + projectId: ProjectId.makeUnsafe(crypto.randomUUID()), title: projectName, workspaceRoot: cwd, defaultModel: "gpt-5-codex", @@ -351,110 +343,6 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E } const routeRequest = Effect.fn(function* (request: WebSocketRequest) { - const computeTurnDiff = Effect.fn(function* (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 = yield* 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] = yield* Effect.all( - [ - liveCheckpointStore.hasCheckpointRef({ - cwd: workspaceCwd, - checkpointRef: fromCheckpointRef, - }), - liveCheckpointStore.hasCheckpointRef({ - cwd: workspaceCwd, - checkpointRef: toCheckpointRef, - }), - ], - { concurrency: "unbounded" }, - ); - 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 = yield* 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: return yield* projectionReadModelQuery.getSnapshot(); @@ -466,16 +354,12 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E case ORCHESTRATION_WS_METHODS.getTurnDiff: { const body = stripRequestTag(request.body); - return yield* computeTurnDiff(body); + return yield* checkpointDiffQuery.getTurnDiff(body); } case ORCHESTRATION_WS_METHODS.getFullThreadDiff: { const body = stripRequestTag(request.body); - return yield* computeTurnDiff({ - threadId: body.threadId, - fromTurnCount: 0, - toTurnCount: body.toTurnCount, - }); + return yield* checkpointDiffQuery.getFullThreadDiff(body); } case ORCHESTRATION_WS_METHODS.replayEvents: { @@ -599,46 +483,38 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E const messageText = websocketRawToString(raw); if (messageText === null) { - const errorResponse = yield* encodeResponse( - new WebSocketResponse({ - id: "unknown", - error: { message: "Invalid request format: Failed to read message" }, - }), - ); + const errorResponse = yield* encodeResponse({ + id: "unknown", + error: { message: "Invalid request format: Failed to read message" }, + }); ws.send(errorResponse); return; } const request = Schema.decodeExit(Schema.fromJsonString(WebSocketRequest))(messageText); if (request._tag === "Failure") { - const errorResponse = yield* encodeResponse( - new WebSocketResponse({ - id: "unknown", - error: { message: `Invalid request format: ${Cause.pretty(request.cause)}` }, - }), - ); + const errorResponse = yield* encodeResponse({ + id: "unknown", + error: { message: `Invalid request format: ${Cause.pretty(request.cause)}` }, + }); ws.send(errorResponse); return; } const result = yield* Effect.exit(routeRequest(request.value)); if (result._tag === "Failure") { - const errorResponse = yield* encodeResponse( - new WebSocketResponse({ - id: request.value.id, - error: { message: Cause.pretty(result.cause) }, - }), - ); - ws.send(errorResponse); - return; - } - - const response = yield* encodeResponse( - new WebSocketResponse({ + const errorResponse = yield* encodeResponse({ id: request.value.id, - result: result.value, - }), - ); + error: { message: Cause.pretty(result.cause) }, + }); + ws.send(errorResponse); + return; + } + + const response = yield* encodeResponse({ + id: request.value.id, + result: result.value, + }); ws.send(response); }); diff --git a/apps/web/src/components/BranchToolbar.logic.test.ts b/apps/web/src/components/BranchToolbar.logic.test.ts index dd2a25d06999f166a6d838a349376b000b2c5246..9b72890ce5dfe7611d6b41fb8d30a1ad8a165ca2 100644 --- a/apps/web/src/components/BranchToolbar.logic.test.ts +++ b/apps/web/src/components/BranchToolbar.logic.test.ts @@ -1,4 +1,4 @@ -import type { GitBranch } from "@t3tools/contracts"; +import { ThreadId, type GitBranch } from "@t3tools/contracts"; import { describe, expect, it } from "vitest"; import { deriveSyncedLocalBranch } from "./BranchToolbar.logic"; @@ -19,8 +19,9 @@ const branches: GitBranch[] = [ describe("deriveSyncedLocalBranch", () => { it("syncs to git current branch in local mode", () => { + const threadId = ThreadId.makeUnsafe("thread-1"); const result = deriveSyncedLocalBranch({ - activeThreadId: "thread-1", + activeThreadId: threadId, activeWorktreePath: null, envMode: "local", activeThreadBranch: "feature/demo", @@ -31,8 +32,9 @@ describe("deriveSyncedLocalBranch", () => { }); it("does not sync when creating a new worktree", () => { + const threadId = ThreadId.makeUnsafe("thread-1"); const result = deriveSyncedLocalBranch({ - activeThreadId: "thread-1", + activeThreadId: threadId, activeWorktreePath: null, envMode: "worktree", activeThreadBranch: "feature/demo", @@ -43,8 +45,9 @@ describe("deriveSyncedLocalBranch", () => { }); it("does not sync when thread already targets a worktree path", () => { + const threadId = ThreadId.makeUnsafe("thread-1"); const result = deriveSyncedLocalBranch({ - activeThreadId: "thread-1", + activeThreadId: threadId, activeWorktreePath: "/tmp/repo/worktrees/feature-demo", envMode: "local", activeThreadBranch: "feature/demo", diff --git a/apps/web/src/components/BranchToolbar.logic.ts b/apps/web/src/components/BranchToolbar.logic.ts index ab83f2ab93149018c1dc204300e9f2af71f903bc..a3185d0514328b2b2894ff65fcc6a89ee9f02b79 100644 --- a/apps/web/src/components/BranchToolbar.logic.ts +++ b/apps/web/src/components/BranchToolbar.logic.ts @@ -1,7 +1,7 @@ -import type { GitBranch } from "@t3tools/contracts"; +import type { GitBranch, ThreadId } from "@t3tools/contracts"; interface DeriveSyncedLocalBranchInput { - activeThreadId: string | undefined; + activeThreadId: ThreadId | undefined; activeWorktreePath: string | null; envMode: "local" | "worktree"; activeThreadBranch: string | null; diff --git a/apps/web/src/components/BranchToolbar.tsx b/apps/web/src/components/BranchToolbar.tsx index 5306ef1cc9b5bf251dc13b2dad0bc26ad9896e31..374902ea6e58b8f9f9766e962365967bbc96dde6 100644 --- a/apps/web/src/components/BranchToolbar.tsx +++ b/apps/web/src/components/BranchToolbar.tsx @@ -1,14 +1,14 @@ -import type { GitBranch } from "@t3tools/contracts"; +import type { GitBranch, ThreadId } from "@t3tools/contracts"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { useEffect, useState } from "react"; -import { useNativeApi } from "../hooks/useNativeApi"; -import { asThreadId, newCommandId } from "../lib/orchestrationIds"; +import { newCommandId } from "../lib/orchestrationIds"; import { gitBranchesQueryOptions, gitCheckoutMutationOptions, gitCreateBranchAndCheckoutMutationOptions, } from "../lib/gitReactQuery"; +import { readNativeApi } from "../nativeApi"; import { useStore } from "../store"; import { deriveSyncedLocalBranch } from "./BranchToolbar.logic"; import { Button } from "./ui/button"; @@ -24,7 +24,7 @@ import { import { ChevronDownIcon } from "lucide-react"; interface BranchToolbarProps { - threadId: string; + threadId: ThreadId; envMode: "local" | "worktree"; onEnvModeChange: (mode: "local" | "worktree") => void; envLocked: boolean; @@ -39,7 +39,6 @@ export default function BranchToolbar({ onComposerFocusRequest, }: BranchToolbarProps) { const { state, dispatch } = useStore(); - const api = useNativeApi(); const queryClient = useQueryClient(); const [isBranchMenuOpen, setIsBranchMenuOpen] = useState(false); @@ -54,7 +53,7 @@ export default function BranchToolbar({ // ── Queries ─────────────────────────────────────────────────────────── - const branchesQuery = useQuery(gitBranchesQueryOptions(api, branchCwd)); + const branchesQuery = useQuery(gitBranchesQueryOptions(branchCwd)); const branches = branchesQuery.data?.branches ?? []; const branchNames = branches.map((branch) => branch.name); @@ -72,11 +71,11 @@ export default function BranchToolbar({ // ── Mutations ───────────────────────────────────────────────────────── const checkoutMutation = useMutation( - gitCheckoutMutationOptions({ api, cwd: branchCwd, queryClient }), + gitCheckoutMutationOptions({ cwd: branchCwd, queryClient }), ); const createBranchMutation = useMutation( - gitCreateBranchAndCheckoutMutationOptions({ api, cwd: branchCwd, queryClient }), + gitCreateBranchAndCheckoutMutationOptions({ cwd: branchCwd, queryClient }), ); // ── Effects ─────────────────────────────────────────────────────────── @@ -92,12 +91,13 @@ export default function BranchToolbar({ queryBranches, }); if (!activeThreadId || !syncedBranch) return; + const api = readNativeApi(); if (api) { void api.orchestration.dispatchCommand({ type: "thread.meta.update", commandId: newCommandId(), - threadId: asThreadId(activeThreadId), + threadId: activeThreadId, branch: syncedBranch, worktreePath: null, }); @@ -115,7 +115,6 @@ export default function BranchToolbar({ queryBranches, envMode, dispatch, - api, ]); useEffect(() => { @@ -132,6 +131,7 @@ export default function BranchToolbar({ const setThreadBranch = (branch: string | null, worktreePath: string | null) => { if (!activeThreadId) return; + const api = readNativeApi(); // If the effective cwd is about to change, stop the running session so the // next message creates a new one with the correct cwd. const sessionId = activeThread?.session?.sessionId; @@ -140,7 +140,7 @@ export default function BranchToolbar({ .dispatchCommand({ type: "thread.session.stop", commandId: newCommandId(), - threadId: asThreadId(activeThreadId), + threadId: activeThreadId, createdAt: new Date().toISOString(), }) .catch(() => undefined); @@ -149,7 +149,7 @@ export default function BranchToolbar({ void api.orchestration.dispatchCommand({ type: "thread.meta.update", commandId: newCommandId(), - threadId: asThreadId(activeThreadId), + threadId: activeThreadId, branch, worktreePath, }); @@ -158,6 +158,7 @@ export default function BranchToolbar({ }; const selectBranch = (branch: GitBranch) => { + const api = readNativeApi(); if (!api || !activeThreadId || !branchCwd) return; // For new worktree mode, selecting a branch picks the base branch. @@ -198,6 +199,7 @@ export default function BranchToolbar({ const createBranch = (rawName: string) => { const name = rawName.trim(); + const api = readNativeApi(); if (!api || !activeThreadId || !branchCwd || !name || createBranchMutation.isPending) return; createBranchMutation.mutate(name, { onSuccess: () => { diff --git a/apps/web/src/components/ChatMarkdown.tsx b/apps/web/src/components/ChatMarkdown.tsx index 91972d787bbc5fd0c98948aacc0c241fb3358071..2676da0d21bd44570fadc02f37382a329412357b 100644 --- a/apps/web/src/components/ChatMarkdown.tsx +++ b/apps/web/src/components/ChatMarkdown.tsx @@ -3,7 +3,7 @@ import type { Components } from "react-markdown"; import ReactMarkdown from "react-markdown"; import rehypeHighlight from "rehype-highlight"; import remarkGfm from "remark-gfm"; -import { useNativeApi } from "../hooks/useNativeApi"; +import { readNativeApi } from "../nativeApi"; import { resolveMarkdownFileLinkTarget } from "../markdown-links"; import { preferredTerminalEditor } from "../terminal-links"; @@ -13,12 +13,11 @@ interface ChatMarkdownProps { } function ChatMarkdown({ text, cwd }: ChatMarkdownProps) { - const api = useNativeApi(); const markdownComponents = useMemo( () => ({ a({ node: _node, href, ...props }) { const targetPath = resolveMarkdownFileLinkTarget(href, cwd); - if (!targetPath || !api) { + if (!targetPath) { return ; } @@ -29,13 +28,18 @@ function ChatMarkdown({ text, cwd }: ChatMarkdownProps) { onClick={(event) => { event.preventDefault(); event.stopPropagation(); - void api.shell.openInEditor(targetPath, preferredTerminalEditor()); + const api = readNativeApi(); + if (api) { + void api.shell.openInEditor(targetPath, preferredTerminalEditor()); + } else { + console.warn("Native API not found. Unable to open file in editor."); + } }} /> ); }, }), - [api, cwd], + [cwd], ); return ( diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index e1ba97830181e28199661b9eb895088886b91a9c..42c5ad5e53afb175b16133116c934fbb855f76ac 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1,15 +1,25 @@ import { + type ApprovalRequestId, + DEFAULT_MODEL, + DEFAULT_REASONING, EDITORS, type EditorId, type KeybindingCommand, - type NativeApi, + type MessageId, + MODEL_OPTIONS, + type ProjectId, type ProjectEntry, type ProjectScript, ModelSlug, PROVIDER_SEND_TURN_MAX_ATTACHMENTS, PROVIDER_SEND_TURN_MAX_IMAGE_BYTES, + REASONING_OPTIONS, + type ReasoningEffort, type ResolvedKeybindingsConfig, type ProviderApprovalDecision, + type ThreadId, + type TurnId, + resolveModelSlug, } from "@t3tools/contracts"; import { memo, useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from "react"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; @@ -27,14 +37,6 @@ import { detectComposerTrigger, replaceTextRange, } from "../composer-logic"; -import { - DEFAULT_MODEL, - DEFAULT_REASONING, - MODEL_OPTIONS, - REASONING_OPTIONS, - ReasoningEffort, - resolveModelSlug, -} from "../model-logic"; import { derivePendingApprovals, derivePhase, @@ -66,7 +68,6 @@ import { } from "../keybindings"; import ChatMarkdown from "./ChatMarkdown"; import ThreadTerminalDrawer from "./ThreadTerminalDrawer"; -import { useNativeApi } from "../hooks/useNativeApi"; import { Alert, AlertAction, AlertDescription, AlertTitle } from "./ui/alert"; import { BotIcon, @@ -101,13 +102,8 @@ import { } from "~/projectScripts"; import { Toggle } from "./ui/toggle"; import { SidebarTrigger } from "./ui/sidebar"; -import { - asApprovalRequestId, - asProjectId, - asThreadId, - newCommandId, - newMessageId, -} from "~/lib/orchestrationIds"; +import { newCommandId, newMessageId } from "~/lib/orchestrationIds"; +import { readNativeApi } from "~/nativeApi"; function formatMessageMeta(createdAt: string, duration: string | null): string { if (!duration) return formatTimestamp(createdAt); @@ -326,7 +322,7 @@ const ComposerCommandMenu = memo(function ComposerCommandMenu(props: { }); interface ChatViewProps { - threadId: string; + threadId: ThreadId; } export default function ChatView({ threadId }: ChatViewProps) { @@ -336,12 +332,9 @@ export default function ChatView({ threadId }: ChatViewProps) { strict: false, select: (params) => parseDiffRouteSearch(params), }); - const api = useNativeApi(); const { resolvedTheme } = useTheme(); const queryClient = useQueryClient(); - const createWorktreeMutation = useMutation( - gitCreateWorktreeMutationOptions({ api, queryClient }), - ); + const createWorktreeMutation = useMutation(gitCreateWorktreeMutationOptions({ queryClient })); const [prompt, setPrompt] = useState(""); const promptRef = useRef(prompt); const [composerImages, setComposerImages] = useState([]); @@ -353,7 +346,7 @@ export default function ChatView({ threadId }: ChatViewProps) { const [selectedEffort, setSelectedEffort] = useState(DEFAULT_REASONING); const [envMode, setEnvMode] = useState<"local" | "worktree">("local"); const [isSwitchingRuntimeMode, setIsSwitchingRuntimeMode] = useState(false); - const [respondingRequestIds, setRespondingRequestIds] = useState([]); + const [respondingRequestIds, setRespondingRequestIds] = useState([]); const [expandedWorkGroups, setExpandedWorkGroups] = useState>({}); const [nowTick, setNowTick] = useState(() => Date.now()); const [terminalFocusRequestId, setTerminalFocusRequestId] = useState(0); @@ -429,7 +422,7 @@ export default function ChatView({ threadId }: ChatViewProps) { const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = useTurnDiffSummaries(activeThread); const turnDiffSummaryByAssistantMessageId = useMemo(() => { - const byMessageId = new Map(); + const byMessageId = new Map(); for (const summary of turnDiffSummaries) { if (!summary.assistantMessageId) continue; byMessageId.set(summary.assistantMessageId, summary); @@ -437,7 +430,7 @@ export default function ChatView({ threadId }: ChatViewProps) { return byMessageId; }, [turnDiffSummaries]); const revertTurnCountByUserMessageId = useMemo(() => { - const byUserMessageId = new Map(); + 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") { @@ -536,14 +529,13 @@ export default function ChatView({ threadId }: ChatViewProps) { (debouncerState) => ({ isPending: debouncerState.isPending }), ); const effectivePathQuery = pathTriggerQuery.length > 0 ? debouncedPathQuery : ""; - const branchesQuery = useQuery(gitBranchesQueryOptions(api, gitCwd)); + const branchesQuery = useQuery(gitBranchesQueryOptions(gitCwd)); const keybindingsQuery = useQuery({ - ...serverConfigQueryOptions(api), + ...serverConfigQueryOptions(), select: (config) => config.keybindings, }); const workspaceEntriesQuery = useQuery( projectSearchEntriesQueryOptions({ - api, cwd: gitCwd, query: effectivePathQuery, enabled: isPathTrigger, @@ -689,6 +681,7 @@ export default function ChatView({ threadId }: ChatViewProps) { ); const closeTerminal = useCallback( (terminalId: string) => { + const api = readNativeApi(); if (!activeThreadId || !api) return; const isFinalTerminal = (activeThread?.terminalIds.length ?? 0) <= 1; const fallbackExitWrite = () => @@ -714,7 +707,7 @@ export default function ChatView({ threadId }: ChatViewProps) { }); setTerminalFocusRequestId((value) => value + 1); }, - [activeThread?.terminalIds.length, activeThreadId, api, dispatch], + [activeThread?.terminalIds.length, activeThreadId, dispatch], ); const runProjectScript = useCallback( async ( @@ -727,6 +720,7 @@ export default function ChatView({ threadId }: ChatViewProps) { rememberAsLastInvoked?: boolean; }, ) => { + const api = readNativeApi(); if (!api || !activeThreadId || !activeProject || !activeThread) return; if (options?.rememberAsLastInvoked !== false) { setLastInvokedScriptByProjectId((current) => { @@ -799,23 +793,24 @@ export default function ChatView({ threadId }: ChatViewProps) { }); } }, - [activeProject, activeThread, activeThreadId, api, dispatch, gitCwd], + [activeProject, activeThread, activeThreadId, dispatch, gitCwd], ); const persistProjectScripts = useCallback( async (input: { - projectId: string; + projectId: ProjectId; projectCwd: string; previousScripts: ProjectScript[]; nextScripts: ProjectScript[]; keybinding?: string | null; keybindingCommand: KeybindingCommand; }) => { + const api = readNativeApi(); if (!api) return; await api.orchestration.dispatchCommand({ type: "project.meta.update", commandId: newCommandId(), - projectId: asProjectId(input.projectId), + projectId: input.projectId, scripts: input.nextScripts, }); @@ -836,7 +831,7 @@ export default function ChatView({ threadId }: ChatViewProps) { ); } }, - [api, queryClient], + [queryClient], ); const saveProjectScript = useCallback( async (input: NewProjectScriptInput) => { @@ -911,6 +906,7 @@ export default function ChatView({ threadId }: ChatViewProps) { if (mode === state.runtimeMode) return; dispatch({ type: "SET_RUNTIME_MODE", mode }); scheduleComposerFocus(); + const api = readNativeApi(); if (!api) return; const runningThreadIds = state.threads @@ -927,7 +923,7 @@ export default function ChatView({ threadId }: ChatViewProps) { .dispatchCommand({ type: "thread.session.stop", commandId: newCommandId(), - threadId: asThreadId(threadId), + threadId, createdAt: new Date().toISOString(), }) .catch(() => undefined), @@ -1182,7 +1178,7 @@ export default function ChatView({ threadId }: ChatViewProps) { ]); const setThreadError = useCallback( - (threadId: string | null, error: string | null) => { + (threadId: ThreadId | null, error: string | null) => { if (!threadId) return; dispatch({ type: "SET_ERROR", @@ -1298,9 +1294,9 @@ export default function ChatView({ threadId }: ChatViewProps) { const onRevertToTurnCount = useCallback( async (turnCount: number) => { - if (!api || !activeThread || isRevertingCheckpoint) { - return; - } + const api = readNativeApi(); + if (!api || !activeThread || isRevertingCheckpoint) return; + if (phase === "running" || isSending || isConnecting) { setThreadError(activeThread.id, "Interrupt the current turn before reverting checkpoints."); return; @@ -1322,7 +1318,7 @@ export default function ChatView({ threadId }: ChatViewProps) { await api.orchestration.dispatchCommand({ type: "thread.checkpoint.revert", commandId: newCommandId(), - threadId: asThreadId(activeThread.id), + threadId: activeThread.id, turnCount, createdAt: new Date().toISOString(), }); @@ -1335,11 +1331,12 @@ export default function ChatView({ threadId }: ChatViewProps) { setIsRevertingCheckpoint(false); } }, - [activeThread, api, isConnecting, isRevertingCheckpoint, isSending, phase, setThreadError], + [activeThread, isConnecting, isRevertingCheckpoint, isSending, phase, setThreadError], ); const onSend = async (e: React.SubmitEvent | React.KeyboardEvent) => { e.preventDefault(); + const api = readNativeApi(); if (!api || !activeThread || isSending || isConnecting) return; const trimmed = prompt.trim(); if (!trimmed && composerImages.length === 0) return; @@ -1363,7 +1360,7 @@ export default function ChatView({ threadId }: ChatViewProps) { await api.orchestration.dispatchCommand({ type: "thread.meta.update", commandId: newCommandId(), - threadId: asThreadId(activeThread.id), + threadId: activeThread.id, branch: result.worktree.branch, worktreePath: result.worktree.path, }); @@ -1397,7 +1394,7 @@ export default function ChatView({ threadId }: ChatViewProps) { await api.orchestration.dispatchCommand({ type: "thread.meta.update", commandId: newCommandId(), - threadId: asThreadId(activeThread.id), + threadId: activeThread.id, title, }); } @@ -1434,7 +1431,7 @@ export default function ChatView({ threadId }: ChatViewProps) { await api.orchestration.dispatchCommand({ type: "thread.turn.start", commandId: newCommandId(), - threadId: asThreadId(activeThread.id), + threadId: activeThread.id, message: { messageId: newMessageId(), role: "user", @@ -1456,17 +1453,19 @@ export default function ChatView({ threadId }: ChatViewProps) { }; const onInterrupt = async () => { + const api = readNativeApi(); if (!api || !activeThread) return; await api.orchestration.dispatchCommand({ type: "thread.turn.interrupt", commandId: newCommandId(), - threadId: asThreadId(activeThread.id), + threadId: activeThread.id, createdAt: new Date().toISOString(), }); }; const onRespondToApproval = useCallback( - async (requestId: string, decision: ProviderApprovalDecision) => { + async (requestId: ApprovalRequestId, decision: ProviderApprovalDecision) => { + const api = readNativeApi(); if (!api || !activeThreadId) return; setRespondingRequestIds((existing) => @@ -1476,8 +1475,8 @@ export default function ChatView({ threadId }: ChatViewProps) { await api.orchestration.dispatchCommand({ type: "thread.approval.respond", commandId: newCommandId(), - threadId: asThreadId(activeThreadId), - requestId: asApprovalRequestId(requestId), + threadId: activeThreadId, + requestId, decision, createdAt: new Date().toISOString(), }); @@ -1491,23 +1490,24 @@ export default function ChatView({ threadId }: ChatViewProps) { setRespondingRequestIds((existing) => existing.filter((id) => id !== requestId)); } }, - [activeThreadId, api, dispatch], + [activeThreadId, dispatch], ); const onModelSelect = useCallback( (model: ModelSlug) => { + const api = readNativeApi(); if (!activeThread) return; if (api) { void api.orchestration.dispatchCommand({ type: "thread.meta.update", commandId: newCommandId(), - threadId: asThreadId(activeThread.id), + threadId: activeThread.id, model: resolveModelSlug(model), }); } scheduleComposerFocus(); }, - [activeThread, api, scheduleComposerFocus], + [activeThread, scheduleComposerFocus], ); const onEffortSelect = useCallback( (effort: ReasoningEffort) => { @@ -1629,7 +1629,7 @@ export default function ChatView({ threadId }: ChatViewProps) { }); }, [diffOpen, navigate, threadId]); const onOpenTurnDiff = useCallback( - (turnId: string, filePath?: string) => { + (turnId: TurnId, filePath?: string) => { void navigate({ to: "/$threadId", params: { threadId }, @@ -1644,7 +1644,7 @@ export default function ChatView({ threadId }: ChatViewProps) { [navigate, threadId], ); const onRevertUserMessage = useCallback( - (messageId: string) => { + (messageId: MessageId) => { const targetTurnCount = revertTurnCountByUserMessageId.get(messageId); if (typeof targetTurnCount !== "number") { return; @@ -1698,7 +1698,6 @@ export default function ChatView({ threadId }: ChatViewProps) { activeProject ? (lastInvokedScriptByProjectId[activeProject.id] ?? null) : null } keybindings={keybindings} - api={api} gitCwd={gitCwd} diffOpen={diffOpen} onRunProjectScript={(script) => { @@ -1971,35 +1970,39 @@ export default function ChatView({ threadId }: ChatViewProps) { /> )} - {activeThread.terminalOpen && api && activeProject && ( - - dispatch({ - type: "SET_THREAD_TERMINAL_HEIGHT", - threadId: activeThread.id, - height, - }) - } - /> - )} + {(() => { + if (!activeThread.terminalOpen || !activeProject) { + return null; + } + return ( + + dispatch({ + type: "SET_THREAD_TERMINAL_HEIGHT", + threadId: activeThread.id, + height, + }) + } + /> + ); + })()} {expandedImage && (
void; @@ -2062,7 +2064,6 @@ const ChatHeader = memo(function ChatHeader({ activeProjectScripts, preferredScriptId, keybindings, - api, gitCwd, diffOpen, onRunProjectScript, @@ -2100,9 +2101,7 @@ const ChatHeader = memo(function ChatHeader({ {activeProjectName && ( )} - {activeProjectName && ( - - )} + {activeProjectName && } Promise; + respondingRequestIds: ApprovalRequestId[]; + onRespondToApproval: ( + requestId: ApprovalRequestId, + decision: ProviderApprovalDecision, + ) => Promise; } const PendingApprovalsPanel = memo(function PendingApprovalsPanel({ @@ -2211,13 +2213,13 @@ interface MessagesTimelineProps { timelineEntries: ReturnType; completionDividerBeforeEntryId: string | null; completionSummary: string | null; - turnDiffSummaryByAssistantMessageId: Map; + turnDiffSummaryByAssistantMessageId: Map; nowIso: string; expandedWorkGroups: Record; onToggleWorkGroup: (groupId: string) => void; - onOpenTurnDiff: (turnId: string, filePath?: string) => void; - revertTurnCountByUserMessageId: Map; - onRevertUserMessage: (messageId: string) => void; + onOpenTurnDiff: (turnId: TurnId, filePath?: string) => void; + revertTurnCountByUserMessageId: Map; + onRevertUserMessage: (messageId: MessageId) => void; isRevertingCheckpoint: boolean; onImageExpand: (image: ExpandedImagePreview) => void; markdownCwd: string | undefined; @@ -2667,7 +2669,7 @@ const OpenInPicker = memo(function OpenInPicker({ activeThreadId, }: { keybindings: ResolvedKeybindingsConfig; - activeThreadId: string | null; + activeThreadId: ThreadId | null; }) { const [lastEditor, setLastEditor] = useState(() => { const stored = localStorage.getItem(LAST_EDITOR_KEY); @@ -2692,13 +2694,13 @@ const OpenInPicker = memo(function OpenInPicker({ ] satisfies { label: string; Icon: Icon; value: EditorId }[]; const primaryOption = options.find(({ value }) => value === lastEditor); - const api = useNativeApi(); const { state } = useStore(); const activeThread = state.threads.find((t) => t.id === activeThreadId); const activeProject = state.projects.find((p) => p.id === activeThread?.projectId); const openInEditor = useCallback( (editorId: EditorId | null) => { + const api = readNativeApi(); if (!api || !activeProject) return; const editor = editorId ?? lastEditor; const cwd = activeThread?.worktreePath ?? activeProject.cwd; @@ -2706,7 +2708,7 @@ const OpenInPicker = memo(function OpenInPicker({ localStorage.setItem(LAST_EDITOR_KEY, editor); setLastEditor(editor); }, - [api, activeProject, activeThread, lastEditor, setLastEditor], + [activeProject, activeThread, lastEditor, setLastEditor], ); const openFavoriteEditorShortcutLabel = useMemo( @@ -2716,6 +2718,7 @@ const OpenInPicker = memo(function OpenInPicker({ useEffect(() => { const handler = (e: globalThis.KeyboardEvent) => { + const api = readNativeApi(); if (!isOpenFavoriteEditorShortcut(e, keybindings)) return; if (!api || !activeProject) return; @@ -2725,7 +2728,7 @@ const OpenInPicker = memo(function OpenInPicker({ }; window.addEventListener("keydown", handler); return () => window.removeEventListener("keydown", handler); - }, [api, activeProject, activeThread, keybindings, lastEditor]); + }, [activeProject, activeThread, keybindings, lastEditor]); return ( diff --git a/apps/web/src/components/DiffPanel.tsx b/apps/web/src/components/DiffPanel.tsx index cc5bd4a6e1310e93f3081abe3f1bcf416e30f951..7abe27fe873827101a637fb9cddf172a63891f9e 100644 --- a/apps/web/src/components/DiffPanel.tsx +++ b/apps/web/src/components/DiffPanel.tsx @@ -2,13 +2,13 @@ import { parsePatchFiles } from "@pierre/diffs"; import { FileDiff, type FileDiffMetadata, Virtualizer } from "@pierre/diffs/react"; import { useQuery } from "@tanstack/react-query"; import { useNavigate, useParams, useSearch } from "@tanstack/react-router"; +import { ThreadId, type TurnId } from "@t3tools/contracts"; import { ChevronLeftIcon, ChevronRightIcon, Columns2Icon, Rows3Icon } from "lucide-react"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { checkpointDiffQueryOptions } from "~/lib/providerReactQuery"; import { cn } from "~/lib/utils"; import { parseDiffRouteSearch, stripDiffSearchParams } from "../diffRouteSearch"; import { isElectron } from "../env"; -import { useNativeApi } from "../hooks/useNativeApi"; import { useTheme } from "../hooks/useTheme"; import { buildPatchCacheKey } from "../lib/diffRendering"; import { useTurnDiffSummaries } from "../hooks/useTurnDiffSummaries"; @@ -81,7 +81,6 @@ interface DiffPanelProps { export { DiffWorkerPoolProvider } from "./DiffWorkerPoolProvider"; export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { - const api = useNativeApi(); const navigate = useNavigate(); const { resolvedTheme } = useTheme(); const { state } = useStore(); @@ -90,13 +89,11 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { const turnStripRef = useRef(null); const [canScrollTurnStripLeft, setCanScrollTurnStripLeft] = useState(false); const [canScrollTurnStripRight, setCanScrollTurnStripRight] = useState(false); - const params = useParams({ strict: false }); - const rawSearch = useSearch({ strict: false }); - const routeThreadId = typeof params.threadId === "string" ? params.threadId : null; - const diffSearch = useMemo( - () => parseDiffRouteSearch(rawSearch as Record), - [rawSearch], - ); + const routeThreadId = useParams({ + strict: false, + select: (params) => (params.threadId ? ThreadId.makeUnsafe(params.threadId) : null), + }); + const diffSearch = useSearch({ strict: false, select: (search) => parseDiffRouteSearch(search) }); const activeThreadId = routeThreadId; const activeThread = state.threads.find((thread) => thread.id === activeThreadId); const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = @@ -169,7 +166,7 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { return `conversation:${orderedTurnDiffSummaries.map((summary) => summary.turnId).join(",")}`; }, [orderedTurnDiffSummaries, selectedTurn]); const activeCheckpointDiffQuery = useQuery( - checkpointDiffQueryOptions(api, { + checkpointDiffQueryOptions({ threadId: activeThreadId, fromTurnCount: activeCheckpointRange?.fromTurnCount ?? null, toTurnCount: activeCheckpointRange?.toTurnCount ?? null, @@ -216,7 +213,7 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { target?.scrollIntoView({ block: "nearest" }); }, [selectedFilePath, renderableFiles]); - const selectTurn = (turnId: string) => { + const selectTurn = (turnId: TurnId) => { if (!activeThread) return; void navigate({ to: "/$threadId", diff --git a/apps/web/src/components/GitActionsControl.tsx b/apps/web/src/components/GitActionsControl.tsx index af88bc8001a9be67dd0be6ea748e41b6d1263485..42bebc7271bf32f9f3a4b3d76559a7ccac7d9d93 100644 --- a/apps/web/src/components/GitActionsControl.tsx +++ b/apps/web/src/components/GitActionsControl.tsx @@ -1,4 +1,4 @@ -import { type GitStatusResult, type GitStackedAction, type NativeApi } from "@t3tools/contracts"; +import { type GitStatusResult, type GitStackedAction, type ThreadId } from "@t3tools/contracts"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { useCallback, useEffect, useMemo, useState } from "react"; import { ChevronDownIcon, CloudUploadIcon, GitCommitIcon, InfoIcon } from "lucide-react"; @@ -39,11 +39,11 @@ import { invalidateGitQueries, } from "~/lib/gitReactQuery"; import { preferredTerminalEditor, resolvePathLinkTarget } from "~/terminal-links"; +import { readNativeApi } from "~/nativeApi"; interface GitActionsControlProps { - api: NativeApi | undefined; gitCwd: string | null; - activeThreadId: string | null; + activeThreadId: ThreadId | null; } function getMenuActionDisabledReason( @@ -136,7 +136,7 @@ function GitQuickActionIcon({ quickAction }: { quickAction: GitQuickAction }) { return ; } -export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitActionsControlProps) { +export default function GitActionsControl({ gitCwd, activeThreadId }: GitActionsControlProps) { const threadToastData = useMemo( () => (activeThreadId ? { threadId: activeThreadId } : undefined), [activeThreadId], @@ -145,11 +145,9 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc const [activeDialogAction, setActiveDialogAction] = useState(null); const [dialogCommitMessage, setDialogCommitMessage] = useState(""); - const { data: gitStatus = null, error: gitStatusError } = useQuery( - gitStatusQueryOptions(api, gitCwd), - ); + const { data: gitStatus = null, error: gitStatusError } = useQuery(gitStatusQueryOptions(gitCwd)); - const { data: branchList = null } = useQuery(gitBranchesQueryOptions(api, gitCwd)); + const { data: branchList = null } = useQuery(gitBranchesQueryOptions(gitCwd)); // Default to true while loading so we don't flash init controls. const isRepo = branchList?.isRepo ?? true; const currentBranch = branchList?.branches.find((branch) => branch.current)?.name ?? null; @@ -163,12 +161,12 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc const gitStatusForActions = isGitStatusOutOfSync ? null : gitStatus; - const initMutation = useMutation(gitInitMutationOptions({ api, cwd: gitCwd, queryClient })); + const initMutation = useMutation(gitInitMutationOptions({ cwd: gitCwd, queryClient })); const runImmediateGitActionMutation = useMutation( - gitRunStackedActionMutationOptions({ api, cwd: gitCwd, queryClient }), + gitRunStackedActionMutationOptions({ cwd: gitCwd, queryClient }), ); - const pullMutation = useMutation(gitPullMutationOptions({ api, cwd: gitCwd, queryClient })); + const pullMutation = useMutation(gitPullMutationOptions({ cwd: gitCwd, queryClient })); const isGitActionRunning = runImmediateGitActionMutation.isPending || pullMutation.isPending; const isDefaultBranch = useMemo(() => { @@ -192,6 +190,7 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc const maybeConfirmPushToDefaultBranch = useCallback( async (action: GitStackedAction): Promise => { + const api = readNativeApi(); if (!api) return false; if ( !requiresDefaultBranchConfirmation(action, isDefaultBranch) || @@ -203,10 +202,11 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc `You're about to push to the default branch "${gitStatusForActions.branch}". Continue?`, ); }, - [api, gitStatusForActions?.branch, isDefaultBranch], + [gitStatusForActions?.branch, isDefaultBranch], ); const openExistingPr = useCallback(async () => { + const api = readNativeApi(); if (!api) { toastManager.add({ type: "error", @@ -232,8 +232,7 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc data: threadToastData, }); }); - }, [api, gitStatusForActions?.openPr?.url, threadToastData]); - + }, [gitStatusForActions?.openPr?.url, threadToastData]); const runGitActionWithToast = useCallback( async ({ action, @@ -329,8 +328,10 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc actionProps: { children: "Open PR", onClick: () => { + const api = readNativeApi(); + if (!api) return; closeResultToast(); - void api?.shell.openExternal(prUrl); + void api.shell.openExternal(prUrl); }, }, } @@ -356,8 +357,8 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc }); } }, + [ - api, gitStatusForActions?.branch, gitStatusForActions?.hasWorkingTreeChanges, gitStatusForActions?.openPr?.url, @@ -447,6 +448,7 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc const openChangedFileInEditor = useCallback( (filePath: string) => { + const api = readNativeApi(); if (!api || !gitCwd) { toastManager.add({ type: "error", @@ -465,7 +467,7 @@ export default function GitActionsControl({ api, gitCwd, activeThreadId }: GitAc }); }); }, - [api, gitCwd, threadToastData], + [gitCwd, threadToastData], ); if (!gitCwd) return null; diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 7bed62ff2096ab81c7f220200d476503f23a0979..3aef2b7f4b602a9a1a08858faba1631792fcae7c 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -1,25 +1,23 @@ import { ChevronRightIcon, TerminalIcon } from "lucide-react"; import { useCallback, useEffect, useMemo, useState } from "react"; -import type { ResolvedKeybindingsConfig } from "@t3tools/contracts"; +import { + DEFAULT_MODEL, + ProjectId, + ThreadId, + type ResolvedKeybindingsConfig, +} from "@t3tools/contracts"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { useNavigate, useParams } from "@tanstack/react-router"; import { useAppSettings } from "../appSettings"; import { isElectron } from "../env"; import { APP_STAGE_LABEL } from "../branding"; -import { DEFAULT_MODEL } from "../model-logic"; -import { - asProjectId, - asThreadId, - newCommandId, - newProjectId, - newThreadId, -} from "../lib/orchestrationIds"; +import { newCommandId, newProjectId, newThreadId } from "../lib/orchestrationIds"; import { useStore } from "../store"; import { isChatNewLocalShortcut, isChatNewShortcut } from "../keybindings"; import { type Thread } from "../types"; -import { useNativeApi } from "../hooks/useNativeApi"; import { gitRemoveWorktreeMutationOptions } from "../lib/gitReactQuery"; import { serverConfigQueryOptions } from "../lib/serverReactQuery"; +import { readNativeApi } from "../nativeApi"; import { toastManager } from "./ui/toast"; import { Collapsible, CollapsibleContent, CollapsibleTrigger } from "./ui/collapsible"; import { @@ -37,6 +35,7 @@ import { SidebarTrigger, } from "./ui/sidebar"; import { formatWorktreePathForDisplay, getOrphanedWorktreePathForThread } from "../worktreeCleanup"; +import { isNonEmpty as isNonEmptyString } from "effect/String"; const EMPTY_KEYBINDINGS: ResolvedKeybindingsConfig = []; @@ -143,33 +142,33 @@ function T3Wordmark() { export default function Sidebar() { const { state, dispatch } = useStore(); - const api = useNativeApi(); const navigate = useNavigate(); const { settings: appSettings } = useAppSettings(); - const params = useParams({ strict: false }); - const routeThreadId = typeof params.threadId === "string" ? params.threadId : null; + const routeThreadId = useParams({ + strict: false, + select: (params) => (params.threadId ? ThreadId.makeUnsafe(params.threadId) : null), + }); const { data: keybindings = EMPTY_KEYBINDINGS } = useQuery({ - ...serverConfigQueryOptions(api), + ...serverConfigQueryOptions(), select: (config) => config.keybindings, }); const queryClient = useQueryClient(); - const removeWorktreeMutation = useMutation( - gitRemoveWorktreeMutationOptions({ api, queryClient }), - ); + const removeWorktreeMutation = useMutation(gitRemoveWorktreeMutationOptions({ queryClient })); const [addingProject, setAddingProject] = useState(false); const [newCwd, setNewCwd] = useState(""); const [isPickingFolder, setIsPickingFolder] = useState(false); const [isAddingProject, setIsAddingProject] = useState(false); - const pendingApprovalByThreadId = useMemo(() => new Map(), []); + const pendingApprovalByThreadId = useMemo(() => new Map(), []); const handleNewThread = useCallback( ( - projectId: string, + projectId: ProjectId, options?: { branch?: string | null; worktreePath?: string | null; }, ): Promise => { + const api = readNativeApi(); if (!api) return Promise.resolve(); const threadId = newThreadId(); const createdAt = new Date().toISOString(); @@ -180,7 +179,7 @@ export default function Sidebar() { type: "thread.create", commandId: newCommandId(), threadId, - projectId: asProjectId(projectId), + projectId, title: "New thread", model, branch: options?.branch ?? null, @@ -202,11 +201,11 @@ export default function Sidebar() { }); })(); }, - [api, dispatch, navigate, state.projects], + [dispatch, navigate, state.projects], ); const focusMostRecentThreadForProject = useCallback( - (projectId: string) => { + (projectId: ProjectId) => { const latestThread = state.threads .filter((thread) => thread.projectId === projectId) .toSorted((a, b) => { @@ -228,6 +227,7 @@ export default function Sidebar() { async (rawCwd: string) => { const cwd = rawCwd.trim(); if (!cwd || isAddingProject) return; + const api = readNativeApi(); setIsAddingProject(true); try { @@ -240,7 +240,7 @@ export default function Sidebar() { const projectId = newProjectId(); const createdAt = new Date().toISOString(); - const title = cwd.split(/[/\\]/).filter(Boolean).at(-1) ?? cwd; + const title = cwd.split(/[/\\]/).findLast(isNonEmptyString) ?? cwd; await api.orchestration.dispatchCommand({ type: "project.create", commandId: newCommandId(), @@ -257,14 +257,7 @@ export default function Sidebar() { setAddingProject(false); } }, - [ - api, - focusMostRecentThreadForProject, - handleNewThread, - isAddingProject, - state.projects, - state.threads, - ], + [focusMostRecentThreadForProject, handleNewThread, isAddingProject, state.projects], ); const handleAddProject = () => { @@ -272,6 +265,7 @@ export default function Sidebar() { }; const handlePickFolder = async () => { + const api = readNativeApi(); if (!api || isPickingFolder) return; setIsPickingFolder(true); try { @@ -284,7 +278,8 @@ export default function Sidebar() { }; const handleThreadContextMenu = useCallback( - async (threadId: string, position: { x: number; y: number }) => { + async (threadId: ThreadId, position: { x: number; y: number }) => { + const api = readNativeApi(); if (!api) return; const clicked = await api.contextMenu.show([{ id: "delete", label: "Delete" }], position); if (clicked !== "delete") return; @@ -324,7 +319,7 @@ export default function Sidebar() { .dispatchCommand({ type: "thread.session.stop", commandId: newCommandId(), - threadId: asThreadId(threadId), + threadId, createdAt: new Date().toISOString(), }) .catch(() => undefined); @@ -344,7 +339,7 @@ export default function Sidebar() { await api.orchestration.dispatchCommand({ type: "thread.delete", commandId: newCommandId(), - threadId: asThreadId(threadId), + threadId, }); if (shouldNavigateToFallback) { if (fallbackThreadId) { @@ -384,9 +379,7 @@ export default function Sidebar() { } }, [ - api, appSettings.confirmThreadDelete, - dispatch, navigate, removeWorktreeMutation, routeThreadId, @@ -396,7 +389,8 @@ export default function Sidebar() { ); const handleProjectContextMenu = useCallback( - async (projectId: string, position: { x: number; y: number }) => { + async (projectId: ProjectId, position: { x: number; y: number }) => { + const api = readNativeApi(); if (!api) return; const clicked = await api.contextMenu.show([{ id: "delete", label: "Delete" }], position); if (clicked !== "delete") return; @@ -423,7 +417,7 @@ export default function Sidebar() { await api.orchestration.dispatchCommand({ type: "project.delete", commandId: newCommandId(), - projectId: asProjectId(projectId), + projectId, }); } catch (error) { const message = error instanceof Error ? error.message : "Unknown error deleting project."; @@ -435,7 +429,7 @@ export default function Sidebar() { }); } }, - [api, state.projects, state.threads], + [state.projects, state.threads], ); useEffect(() => { @@ -687,7 +681,7 @@ export default function Sidebar() { if (event.key === "Escape") setAddingProject(false); }} /> - {isElectron && api && ( + {isElectron && (