diff --git a/apps/desktop/package.json b/apps/desktop/package.json index fed658d7a326927cd5b1f46a5b3b9ec78fdfa486..d64017af4e6b91ef1baf1ced33b235481137d3f1 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -2,7 +2,6 @@ "name": "@t3tools/desktop", "version": "0.0.0", "private": true, - "productName": "T3 Code (Alpha)", "main": "dist-electron/main.js", "scripts": { "dev": "bun run --parallel dev:bundle dev:electron", @@ -18,11 +17,13 @@ "electron": "40.6.0" }, "devDependencies": { + "@t3tools/contracts": "workspace:*", "@types/node": "^22.10.2", "electronmon": "^2.0.2", "tsdown": "^0.20.3", "typescript": "^5.7.3", "vitest": "^4.0.0", "wait-on": "^8.0.2" - } + }, + "productName": "T3 Code (Alpha)" } diff --git a/apps/desktop/src/preload.ts b/apps/desktop/src/preload.ts index 464832b33d307420860bfd9d90a6977d9b9d7247..b7ee8454cccbf36488dfb39194e0620a8f11400c 100644 --- a/apps/desktop/src/preload.ts +++ b/apps/desktop/src/preload.ts @@ -1,4 +1,5 @@ import { contextBridge, ipcRenderer } from "electron"; +import type { DesktopBridge } from "@t3tools/contracts"; const PICK_FOLDER_CHANNEL = "desktop:pick-folder"; const CONFIRM_CHANNEL = "desktop:confirm"; @@ -9,12 +10,11 @@ const wsUrl = process.env.T3CODE_DESKTOP_WS_URL ?? null; contextBridge.exposeInMainWorld("desktopBridge", { getWsUrl: () => wsUrl, - pickFolder: () => ipcRenderer.invoke(PICK_FOLDER_CHANNEL) as Promise, - confirm: (message: string) => ipcRenderer.invoke(CONFIRM_CHANNEL, message) as Promise, - showContextMenu: (items: { id: string; label: string }[]) => - ipcRenderer.invoke(CONTEXT_MENU_CHANNEL, items) as Promise, - openExternal: (url: string) => ipcRenderer.invoke(OPEN_EXTERNAL_CHANNEL, url) as Promise, - onMenuAction: (listener: (action: string) => void) => { + pickFolder: () => ipcRenderer.invoke(PICK_FOLDER_CHANNEL), + confirm: (message) => ipcRenderer.invoke(CONFIRM_CHANNEL, message), + showContextMenu: (items) => ipcRenderer.invoke(CONTEXT_MENU_CHANNEL, items), + openExternal: (url: string) => ipcRenderer.invoke(OPEN_EXTERNAL_CHANNEL, url), + onMenuAction: (listener) => { const wrappedListener = (_event: Electron.IpcRendererEvent, action: unknown) => { if (typeof action !== "string") return; listener(action); @@ -25,4 +25,4 @@ contextBridge.exposeInMainWorld("desktopBridge", { ipcRenderer.removeListener(MENU_ACTION_CHANNEL, wrappedListener); }; }, -}); +} satisfies DesktopBridge); diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts new file mode 100644 index 0000000000000000000000000000000000000000..a65a0d1d73d9adcc6a315ef9a837a5d83f45ef5c --- /dev/null +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -0,0 +1,402 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { execFileSync } from "node:child_process"; + +import { NodeServices } from "@effect/platform-node"; +import { + ApprovalRequestId, + type OrchestrationEvent, + type OrchestrationThread, + type ThreadId, +} from "@t3tools/contracts"; +import { Effect, Exit, Layer, ManagedRuntime, Option, Schedule, Scope, Stream } from "effect"; + +import { CheckpointStoreLive } from "../src/checkpointing/Layers/CheckpointStore.ts"; +import { CheckpointDiffBlobRepositoryLive } from "../src/persistence/Layers/CheckpointDiffBlobs.ts"; +import { OrchestrationCommandReceiptRepositoryLive } from "../src/persistence/Layers/OrchestrationCommandReceipts.ts"; +import { OrchestrationEventStoreLive } from "../src/persistence/Layers/OrchestrationEventStore.ts"; +import { ProjectionCheckpointRepositoryLive } from "../src/persistence/Layers/ProjectionCheckpoints.ts"; +import { ProjectionPendingApprovalRepositoryLive } from "../src/persistence/Layers/ProjectionPendingApprovals.ts"; +import { ProviderSessionRuntimeRepositoryLive } from "../src/persistence/Layers/ProviderSessionRuntime.ts"; +import { makeSqlitePersistenceLive } from "../src/persistence/Layers/Sqlite.ts"; +import { CheckpointDiffBlobRepository } from "../src/persistence/Services/CheckpointDiffBlobs.ts"; +import { ProjectionCheckpointRepository } from "../src/persistence/Services/ProjectionCheckpoints.ts"; +import { ProjectionPendingApprovalRepository } from "../src/persistence/Services/ProjectionPendingApprovals.ts"; +import { ProviderUnsupportedError } from "../src/provider/Errors.ts"; +import { ProviderAdapterRegistry } from "../src/provider/Services/ProviderAdapterRegistry.ts"; +import { ProviderSessionDirectoryLive } from "../src/provider/Layers/ProviderSessionDirectory.ts"; +import { makeProviderServiceLive } from "../src/provider/Layers/ProviderService.ts"; +import { ProviderService } from "../src/provider/Services/ProviderService.ts"; +import { CheckpointReactorLive } from "../src/orchestration/Layers/CheckpointReactor.ts"; +import { OrchestrationEngineLive } from "../src/orchestration/Layers/OrchestrationEngine.ts"; +import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts"; +import { OrchestrationProjectionSnapshotQueryLive } from "../src/orchestration/Layers/ProjectionSnapshotQuery.ts"; +import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts"; +import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts"; +import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts"; +import { + OrchestrationEngineService, + type OrchestrationEngineShape, +} from "../src/orchestration/Services/OrchestrationEngine.ts"; +import { OrchestrationReactor } from "../src/orchestration/Services/OrchestrationReactor.ts"; +import { ProjectionSnapshotQuery } from "../src/orchestration/Services/ProjectionSnapshotQuery.ts"; + +import { + makeTestProviderAdapterHarness, + type TestProviderAdapterHarness, +} from "./TestProviderAdapter.integration.ts"; + +function runGit(cwd: string, args: ReadonlyArray) { + return execFileSync("git", args, { + cwd, + stdio: ["ignore", "pipe", "pipe"], + encoding: "utf8", + }); +} + +function initializeGitWorkspace(cwd: string) { + runGit(cwd, ["init", "--initial-branch=main"]); + runGit(cwd, ["config", "user.email", "test@example.com"]); + runGit(cwd, ["config", "user.name", "Test User"]); + fs.writeFileSync(path.join(cwd, "README.md"), "v1\n", "utf8"); + runGit(cwd, ["add", "."]); + runGit(cwd, ["commit", "-m", "Initial"]); +} + +export function checkpointRefForTurn(threadId: string, turnCount: number): string { + const encodedThreadId = Buffer.from(threadId, "utf8").toString("base64url"); + return `refs/t3/checkpoints/${encodedThreadId}/turn/${turnCount}`; +} + +export function gitRefExists(cwd: string, ref: string): boolean { + try { + runGit(cwd, ["show-ref", "--verify", "--quiet", ref]); + return true; + } catch { + return false; + } +} + +export function gitShowFileAtRef(cwd: string, ref: string, filePath: string): string { + return runGit(cwd, ["show", `${ref}:${filePath}`]); +} + +function waitFor( + read: Effect.Effect, + predicate: (value: A) => boolean, + description: string, + timeoutMs?: number, +): Effect.Effect; +function waitFor( + read: Effect.Effect, + predicate: (value: A) => value is B, + description: string, + timeoutMs?: number, +): Effect.Effect; +function waitFor( + read: Effect.Effect, + predicate: (value: A) => boolean, + description: string, + timeoutMs = 3000, +): Effect.Effect { + const RETRY_SIGNAL = "wait_for_retry"; + const retryIntervalMs = 10; + const maxRetries = Math.max(0, Math.floor(timeoutMs / retryIntervalMs)); + const retrySchedule = Schedule.spaced(`${retryIntervalMs} millis`); + + return read.pipe( + Effect.filterOrFail(predicate, () => RETRY_SIGNAL), + Effect.retry({ + schedule: retrySchedule, + times: maxRetries, + while: (error) => error === RETRY_SIGNAL, + }), + Effect.mapError((error) => + error === RETRY_SIGNAL ? new Error(`Timed out waiting for ${description}.`) : error, + ), + Effect.orDie, + ); +} + +export interface OrchestrationIntegrationHarness { + readonly rootDir: string; + readonly workspaceDir: string; + readonly dbPath: string; + readonly adapterHarness: TestProviderAdapterHarness; + readonly engine: OrchestrationEngineShape; + readonly snapshotQuery: ProjectionSnapshotQuery["Service"]; + readonly providerService: ProviderService["Service"]; + readonly diffBlobRepository: CheckpointDiffBlobRepository["Service"]; + readonly checkpointRepository: ProjectionCheckpointRepository["Service"]; + readonly pendingApprovalRepository: ProjectionPendingApprovalRepository["Service"]; + readonly waitForThread: ( + threadId: string, + predicate: (thread: OrchestrationThread) => boolean, + timeoutMs?: number, + ) => Effect.Effect; + readonly waitForDomainEvent: ( + predicate: (event: OrchestrationEvent) => boolean, + timeoutMs?: number, + ) => Effect.Effect, never>; + readonly waitForPendingApproval: ( + requestId: string, + predicate: (row: { + readonly status: "pending" | "resolved"; + readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null; + readonly resolvedAt: string | null; + }) => boolean, + timeoutMs?: number, + ) => Effect.Effect< + { + readonly status: "pending" | "resolved"; + readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null; + readonly resolvedAt: string | null; + }, + never + >; + readonly waitForDiffBlob: ( + input: { + readonly threadId: ThreadId; + readonly fromTurnCount: number; + readonly toTurnCount: number; + }, + timeoutMs?: number, + ) => Effect.Effect< + { + readonly threadId: ThreadId; + readonly fromTurnCount: number; + readonly toTurnCount: number; + readonly diff: string; + readonly createdAt: string; + }, + never + >; + readonly dispose: Effect.Effect; +} + +export const makeOrchestrationIntegrationHarness = Effect.gen(function* () { + const sleep = (ms: number) => + Effect.promise(() => new Promise((resolve) => setTimeout(resolve, ms))); + const adapterHarness = yield* makeTestProviderAdapterHarness; + + const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-orchestration-integration-")); + const workspaceDir = path.join(rootDir, "workspace"); + const stateDir = path.join(rootDir, "state"); + const dbPath = path.join(stateDir, "orchestration.sqlite"); + fs.mkdirSync(workspaceDir, { recursive: true }); + fs.mkdirSync(stateDir, { recursive: true }); + initializeGitWorkspace(workspaceDir); + + const registry: typeof ProviderAdapterRegistry.Service = { + getByProvider: (provider) => + provider === "codex" + ? Effect.succeed(adapterHarness.adapter) + : Effect.fail(new ProviderUnsupportedError({ provider })), + listProviders: () => Effect.succeed(["codex"]), + }; + + const persistenceLayer = makeSqlitePersistenceLive(dbPath); + const orchestrationLayer = OrchestrationEngineLive.pipe( + Layer.provide(OrchestrationProjectionPipelineLive), + Layer.provide(OrchestrationEventStoreLive), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + ); + const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRuntimeRepositoryLive), + ); + const providerLayer = makeProviderServiceLive().pipe( + Layer.provide(providerSessionDirectoryLayer), + Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), + ); + const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); + + const runtimeServicesLayer = Layer.mergeAll( + orchestrationLayer, + OrchestrationProjectionSnapshotQueryLive, + CheckpointDiffBlobRepositoryLive, + ProjectionCheckpointRepositoryLive, + ProjectionPendingApprovalRepositoryLive, + checkpointStoreLayer, + providerLayer, + ); + const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe( + Layer.provideMerge(runtimeServicesLayer), + ); + const providerCommandReactorLayer = ProviderCommandReactorLive.pipe( + Layer.provideMerge(runtimeServicesLayer), + ); + const checkpointReactorLayer = CheckpointReactorLive.pipe( + Layer.provideMerge(runtimeServicesLayer), + ); + const orchestrationReactorLayer = OrchestrationReactorLive.pipe( + Layer.provideMerge(runtimeIngestionLayer), + Layer.provideMerge(providerCommandReactorLayer), + Layer.provideMerge(checkpointReactorLayer), + ); + const layer = orchestrationReactorLayer.pipe( + Layer.provide(persistenceLayer), + Layer.provideMerge(NodeServices.layer), + ); + + const runtime = ManagedRuntime.make(layer); + const engine = yield* Effect.promise(() => + runtime.runPromise(Effect.service(OrchestrationEngineService)), + ); + const reactor = yield* Effect.promise(() => + runtime.runPromise(Effect.service(OrchestrationReactor)), + ); + const snapshotQuery = yield* Effect.promise(() => + runtime.runPromise(Effect.service(ProjectionSnapshotQuery)), + ); + const providerService = yield* Effect.promise(() => + runtime.runPromise(Effect.service(ProviderService)), + ); + const diffBlobRepository = yield* Effect.promise(() => + runtime.runPromise(Effect.service(CheckpointDiffBlobRepository)), + ); + const checkpointRepository = yield* Effect.promise(() => + runtime.runPromise(Effect.service(ProjectionCheckpointRepository)), + ); + const pendingApprovalRepository = yield* Effect.promise(() => + runtime.runPromise(Effect.service(ProjectionPendingApprovalRepository)), + ); + + const scope = yield* Effect.promise(() => Effect.runPromise(Scope.make("sequential"))); + yield* Effect.promise(() => runtime.runPromise(reactor.start.pipe(Scope.provide(scope)))); + yield* sleep(10); + + const waitForThread: OrchestrationIntegrationHarness["waitForThread"] = ( + threadId, + predicate, + timeoutMs, + ) => + waitFor( + snapshotQuery + .getSnapshot() + .pipe( + Effect.map( + (snapshot) => snapshot.threads.find((thread) => thread.id === threadId) ?? null, + ), + ), + (thread): thread is OrchestrationThread => thread !== null && predicate(thread), + `projected thread '${threadId}'`, + timeoutMs, + ) as Effect.Effect; + + const waitForDomainEvent: OrchestrationIntegrationHarness["waitForDomainEvent"] = ( + predicate, + timeoutMs, + ) => + waitFor( + Stream.runCollect(engine.readEvents(0)).pipe( + Effect.map((chunk): ReadonlyArray => Array.from(chunk)), + ), + (events) => events.some(predicate), + "domain event", + timeoutMs, + ); + + const waitForPendingApproval: OrchestrationIntegrationHarness["waitForPendingApproval"] = ( + requestId, + predicate, + timeoutMs, + ) => + waitFor( + pendingApprovalRepository + .getByRequestId({ requestId: ApprovalRequestId.makeUnsafe(requestId) }) + .pipe( + Effect.map((row) => + Option.match(row, { + onNone: () => null, + onSome: (value) => ({ + status: value.status, + decision: value.decision, + resolvedAt: value.resolvedAt, + }), + }), + ), + ), + ( + row, + ): row is { + readonly status: "pending" | "resolved"; + readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null; + readonly resolvedAt: string | null; + } => row !== null && predicate(row), + `pending approval '${requestId}'`, + timeoutMs, + ) as Effect.Effect< + { + readonly status: "pending" | "resolved"; + readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null; + readonly resolvedAt: string | null; + }, + never + >; + + const waitForDiffBlob: OrchestrationIntegrationHarness["waitForDiffBlob"] = (input, timeoutMs) => + waitFor( + diffBlobRepository.get(input).pipe( + Effect.map((blob) => + Option.match(blob, { + onNone: () => null, + onSome: (value) => value, + }), + ), + ), + ( + blob, + ): blob is { + readonly threadId: ThreadId; + readonly fromTurnCount: number; + readonly toTurnCount: number; + readonly diff: string; + readonly createdAt: string; + } => blob !== null, + `checkpoint diff blob ${input.threadId}:${input.fromTurnCount}->${input.toTurnCount}`, + timeoutMs, + ) as Effect.Effect< + { + readonly threadId: ThreadId; + readonly fromTurnCount: number; + readonly toTurnCount: number; + readonly diff: string; + readonly createdAt: string; + }, + never + >; + + let disposed = false; + const dispose = Effect.gen(function* () { + if (disposed) { + return; + } + disposed = true; + + yield* Effect.promise(() => runtime.runPromise(providerService.stopAll())); + yield* Effect.promise(() => Effect.runPromise(Scope.close(scope, Exit.void))); + yield* Effect.promise(() => runtime.dispose()); + yield* Effect.sync(() => { + fs.rmSync(rootDir, { recursive: true, force: true }); + }); + }); + + return { + rootDir, + workspaceDir, + dbPath, + adapterHarness, + engine, + snapshotQuery, + providerService, + diffBlobRepository, + checkpointRepository, + pendingApprovalRepository, + waitForThread, + waitForDomainEvent, + waitForPendingApproval, + waitForDiffBlob, + dispose, + } satisfies OrchestrationIntegrationHarness; +}); diff --git a/apps/server/integration/TestProviderAdapter.integration.ts b/apps/server/integration/TestProviderAdapter.integration.ts index c5e6aa232580f29bb1bb358b6a126da8012115e5..4b18819f3702bc7c6fc52de7beb0d22d057e7548 100644 --- a/apps/server/integration/TestProviderAdapter.integration.ts +++ b/apps/server/integration/TestProviderAdapter.integration.ts @@ -1,8 +1,14 @@ import { randomUUID } from "node:crypto"; -import type { +import { + ApprovalRequestId, + EventId, + ProviderApprovalDecision, ProviderRuntimeEvent, ProviderSession, + ProviderSessionId, + ProviderThreadId, + ProviderTurnId, ProviderTurnStartResult, } from "@t3tools/contracts"; import { Effect, Queue, Stream } from "effect"; @@ -40,7 +46,15 @@ export interface TestProviderAdapterHarness { sessionId: string, response: TestTurnResponse, ) => Effect.Effect; + readonly queueTurnResponseForNextSession: ( + response: TestTurnResponse, + ) => Effect.Effect; readonly getRollbackCalls: (sessionId: string) => ReadonlyArray; + readonly getApprovalResponses: (sessionId: string) => ReadonlyArray<{ + readonly sessionId: ProviderSessionId; + readonly requestId: ApprovalRequestId; + readonly decision: ProviderApprovalDecision; + }>; } const PROVIDER = "codex" as const; @@ -64,6 +78,15 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { const runtimeEvents = yield* Queue.unbounded(); let sessionCount = 0; const sessions = new Map(); + const queuedResponsesForNextSession: TestTurnResponse[] = []; + const approvalResponsesBySession = new Map< + string, + Array<{ + readonly sessionId: ProviderSessionId; + readonly requestId: ApprovalRequestId; + readonly decision: ProviderApprovalDecision; + }> + >(); const emit = (event: ProviderRuntimeEvent) => Queue.offer(runtimeEvents, event); @@ -78,8 +101,8 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { } sessionCount += 1; - const sessionId = `test-session-${sessionCount}`; - const threadId = `test-thread-${sessionCount}`; + const sessionId = ProviderSessionId.makeUnsafe(`test-session-${sessionCount}`); + const threadId = ProviderThreadId.makeUnsafe(`test-thread-${sessionCount}`); const createdAt = nowIso(); const session: ProviderSession = { @@ -99,7 +122,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { turns: [], }, turnCount: 0, - queuedResponses: [], + queuedResponses: queuedResponsesForNextSession.splice(0), rollbackCalls: [], }); @@ -115,7 +138,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { state.turnCount += 1; const turnCount = state.turnCount; - const turnId = `turn-${turnCount}`; + const turnId = ProviderTurnId.makeUnsafe(`turn-${turnCount}`); const response = state.queuedResponses.shift(); if (!response) { @@ -182,7 +205,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { if (deferredTurnCompletedEvents.length === 0) { yield* emit({ type: "turn.completed", - eventId: randomUUID(), + eventId: EventId.makeUnsafe(randomUUID()), provider: PROVIDER, sessionId: input.sessionId, createdAt: nowIso(), @@ -209,9 +232,20 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { const respondToRequest: ProviderAdapterShape["respondToRequest"] = ( sessionId, - _requestId, - _decision, - ) => (sessions.has(sessionId) ? Effect.void : missingSessionEffect(sessionId)); + requestId, + decision, + ) => + sessions.has(sessionId) + ? Effect.sync(() => { + const existing = approvalResponsesBySession.get(sessionId) ?? []; + existing.push({ + sessionId, + requestId, + decision, + }); + approvalResponsesBySession.set(sessionId, existing); + }) + : missingSessionEffect(sessionId); const stopSession: ProviderAdapterShape["stopSession"] = (sessionId) => Effect.sync(() => { @@ -295,6 +329,13 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { ), ); + const queueTurnResponseForNextSession = ( + response: TestTurnResponse, + ): Effect.Effect => + Effect.sync(() => { + queuedResponsesForNextSession.push(response); + }); + const getRollbackCalls = (sessionId: string): ReadonlyArray => { const state = sessions.get(sessionId); if (!state) { @@ -303,9 +344,25 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { return [...state.rollbackCalls]; }; + const getApprovalResponses = ( + sessionId: string, + ): ReadonlyArray<{ + readonly sessionId: ProviderSessionId; + readonly requestId: ApprovalRequestId; + readonly decision: ProviderApprovalDecision; + }> => { + const responses = approvalResponsesBySession.get(sessionId); + if (!responses) { + return []; + } + return [...responses]; + }; + return { adapter, queueTurnResponse, + queueTurnResponseForNextSession, getRollbackCalls, + getApprovalResponses, } satisfies TestProviderAdapterHarness; }); diff --git a/apps/server/integration/fixtures/providerRuntime.ts b/apps/server/integration/fixtures/providerRuntime.ts index 6ea6587da70c32003ef1246547907b6634e650df..e33e1336e840d8d85054d2757d9212c95ce98a2d 100644 --- a/apps/server/integration/fixtures/providerRuntime.ts +++ b/apps/server/integration/fixtures/providerRuntime.ts @@ -1,16 +1,24 @@ -import type { ProviderRuntimeEvent } from "@t3tools/contracts"; +import { + ApprovalRequestId, + EventId, + ProviderSessionId, + ProviderThreadId, + ProviderTurnId, + type ProviderRuntimeEvent, +} from "@t3tools/contracts"; const PROVIDER = "codex" as const; -const SESSION_ID = "fixture-session"; -const THREAD_ID = "fixture-thread"; -const TURN_ID = "fixture-turn"; +const SESSION_ID = ProviderSessionId.makeUnsafe("fixture-session"); +const THREAD_ID = ProviderThreadId.makeUnsafe("fixture-thread"); +const TURN_ID = ProviderTurnId.makeUnsafe("fixture-turn"); +const REQUEST_ID = ApprovalRequestId.makeUnsafe("req-1"); function baseEvent(eventId: string, createdAt: string): Pick< ProviderRuntimeEvent, "eventId" | "provider" | "sessionId" | "createdAt" > { return { - eventId, + eventId: EventId.makeUnsafe(eventId), provider: PROVIDER, sessionId: SESSION_ID, createdAt, @@ -100,7 +108,7 @@ export const codexTurnApprovalFixture = [ ...baseEvent("evt-22", "2026-02-23T00:02:00.100Z"), threadId: THREAD_ID, turnId: TURN_ID, - requestId: "req-1", + requestId: REQUEST_ID, requestKind: "command", detail: "Please approve command", }, @@ -109,7 +117,7 @@ export const codexTurnApprovalFixture = [ ...baseEvent("evt-23", "2026-02-23T00:02:00.200Z"), threadId: THREAD_ID, turnId: TURN_ID, - requestId: "req-1", + requestId: REQUEST_ID, requestKind: "command", decision: "accept", }, diff --git a/apps/server/integration/orchestrationEngine.integration.test.ts b/apps/server/integration/orchestrationEngine.integration.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..434457c9c661b6f18de0369d7b925b8e07cf9281 --- /dev/null +++ b/apps/server/integration/orchestrationEngine.integration.test.ts @@ -0,0 +1,640 @@ +import fs from "node:fs"; +import path from "node:path"; + +import { + ApprovalRequestId, + CommandId, + EventId, + MessageId, + ProjectId, + ProviderSessionId, + ProviderThreadId, + ProviderTurnId, + ThreadId, +} from "@t3tools/contracts"; +import { assert, it } from "@effect/vitest"; +import { Effect, Option } from "effect"; + +import type { TestTurnResponse } from "./TestProviderAdapter.integration.ts"; +import { + checkpointRefForTurn, + gitRefExists, + gitShowFileAtRef, + makeOrchestrationIntegrationHarness, + type OrchestrationIntegrationHarness, +} from "./OrchestrationEngineHarness.integration.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); +const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); +const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); +const asApprovalRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); + +const PROJECT_ID = asProjectId("project-1"); +const THREAD_ID = asThreadId("thread-1"); +const FIXTURE_SESSION_ID = asProviderSessionId("fixture-session"); +const FIXTURE_THREAD_ID = asProviderThreadId("fixture-thread"); +const FIXTURE_TURN_ID = asProviderTurnId("fixture-turn"); +const APPROVAL_REQUEST_ID = asApprovalRequestId("req-approval-1"); + +function nowIso() { + return new Date().toISOString(); +} + +const sleep = (ms: number) => Effect.promise(() => new Promise((resolve) => setTimeout(resolve, ms))); + +function waitForSync( + read: () => A, + predicate: (value: A) => boolean, + description: string, + timeoutMs = 3000, +): Effect.Effect { + return Effect.gen(function* () { + const deadline = Date.now() + timeoutMs; + + while (true) { + const value = read(); + if (predicate(value)) { + return value; + } + if (Date.now() >= deadline) { + return yield* Effect.die(new Error(`Timed out waiting for ${description}.`)); + } + yield* sleep(10); + } + }); +} + +function runtimeBase(eventId: string, createdAt: string) { + return { + eventId: asEventId(eventId), + provider: "codex" as const, + sessionId: FIXTURE_SESSION_ID, + createdAt, + }; +} + +function withHarness( + use: (harness: OrchestrationIntegrationHarness) => Effect.Effect, +) { + return Effect.acquireUseRelease( + makeOrchestrationIntegrationHarness, + use, + (harness) => harness.dispose, + ); +} + +const seedProjectAndThread = (harness: OrchestrationIntegrationHarness) => + Effect.gen(function* () { + const createdAt = nowIso(); + + yield* harness.engine.dispatch({ + type: "project.create", + commandId: asCommandId("cmd-project-create"), + projectId: PROJECT_ID, + title: "Integration Project", + workspaceRoot: harness.workspaceDir, + defaultModel: "gpt-5-codex", + createdAt, + }); + + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: asCommandId("cmd-thread-create"), + threadId: THREAD_ID, + projectId: PROJECT_ID, + title: "Integration Thread", + model: "gpt-5-codex", + branch: null, + worktreePath: harness.workspaceDir, + createdAt, + }); + }); + +const startTurn = (input: { + readonly harness: OrchestrationIntegrationHarness; + readonly commandId: string; + readonly messageId: string; + readonly text: string; +}) => + input.harness.engine.dispatch({ + type: "thread.turn.start", + commandId: asCommandId(input.commandId), + threadId: THREAD_ID, + message: { + messageId: asMessageId(input.messageId), + role: "user", + text: input.text, + attachments: [], + }, + createdAt: nowIso(), + }); + +it.live("runs a single turn end-to-end and persists checkpoint state in sqlite + git", () => + withHarness((harness) => + Effect.gen(function* () { + yield* seedProjectAndThread(harness); + + const turnResponse: TestTurnResponse = { + events: [ + { + type: "turn.started", + ...runtimeBase("evt-single-1", "2026-02-24T10:00:00.000Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + }, + { + type: "message.delta", + ...runtimeBase("evt-single-2", "2026-02-24T10:00:00.100Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + delta: "Single turn response.\n", + }, + { + type: "turn.completed", + ...runtimeBase("evt-single-3", "2026-02-24T10:00:00.200Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + status: "completed", + }, + ], + }; + + yield* harness.adapterHarness.queueTurnResponseForNextSession(turnResponse); + yield* startTurn({ + harness, + commandId: "cmd-turn-start-single", + messageId: "msg-user-single", + text: "Say hello", + }); + + const thread = yield* harness.waitForThread( + THREAD_ID, + (entry) => + entry.session?.status === "ready" && + entry.messages.some((message) => message.role === "assistant" && message.streaming === false) && + entry.checkpoints.length === 1, + ); + assert.equal(thread.checkpoints[0]?.status, "ready"); + assert.equal(thread.checkpoints[0]?.checkpointTurnCount, 1); + + const checkpointRows = yield* harness.checkpointRepository.listByThreadId({ + threadId: THREAD_ID, + }); + assert.equal(checkpointRows.length, 1); + assert.equal(checkpointRows[0]?.checkpointTurnCount, 1); + assert.equal(checkpointRows[0]?.status, "ready"); + assert.deepEqual(checkpointRows[0]?.files, []); + + yield* harness.waitForDomainEvent((event) => event.type === "thread.turn-diff-completed"); + + const ref0 = checkpointRefForTurn(THREAD_ID, 0); + const ref1 = checkpointRefForTurn(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"); + assert.equal(gitShowFileAtRef(harness.workspaceDir, ref1, "README.md"), "v1\n"); + }), + ), +); + +it.live("runs multi-turn file edits and persists checkpoint diffs + blobs", () => + withHarness((harness) => + Effect.gen(function* () { + yield* seedProjectAndThread(harness); + + yield* harness.adapterHarness.queueTurnResponseForNextSession({ + events: [ + { + type: "turn.started", + ...runtimeBase("evt-multi-1", "2026-02-24T10:01:00.000Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + }, + { + type: "tool.started", + ...runtimeBase("evt-multi-2", "2026-02-24T10:01:00.100Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + toolKind: "command", + title: "Edit file", + detail: "README.md", + }, + { + type: "tool.completed", + ...runtimeBase("evt-multi-3", "2026-02-24T10:01:00.200Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + toolKind: "command", + title: "Edit file", + detail: "README.md", + }, + { + type: "message.delta", + ...runtimeBase("evt-multi-4", "2026-02-24T10:01:00.300Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + delta: "Updated README to v2.\n", + }, + { + type: "turn.completed", + ...runtimeBase("evt-multi-5", "2026-02-24T10:01:00.400Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + status: "completed", + }, + ], + mutateWorkspace: ({ cwd }) => + Effect.sync(() => { + fs.writeFileSync(path.join(cwd, "README.md"), "v2\n", "utf8"); + }), + }); + + yield* startTurn({ + harness, + commandId: "cmd-turn-start-multi-1", + messageId: "msg-user-multi-1", + text: "Make first edit", + }); + + const firstTurnThread = yield* harness.waitForThread( + THREAD_ID, + (entry) => entry.checkpoints.length === 1 && entry.session?.providerSessionId !== null, + ); + const sessionId = firstTurnThread.session?.providerSessionId; + assert.equal(sessionId !== null, true); + if (!sessionId) { + return; + } + + yield* harness.adapterHarness.queueTurnResponse(sessionId, { + events: [ + { + type: "turn.started", + ...runtimeBase("evt-multi-6", "2026-02-24T10:02:00.000Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + }, + { + type: "message.delta", + ...runtimeBase("evt-multi-7", "2026-02-24T10:02:00.100Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + delta: "Updated README to v3.\n", + }, + { + type: "turn.completed", + ...runtimeBase("evt-multi-8", "2026-02-24T10:02:00.200Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + status: "completed", + }, + ], + mutateWorkspace: ({ cwd }) => + Effect.sync(() => { + fs.writeFileSync(path.join(cwd, "README.md"), "v3\n", "utf8"); + }), + }); + + yield* startTurn({ + harness, + commandId: "cmd-turn-start-multi-2", + messageId: "msg-user-multi-2", + text: "Make second edit", + }); + + const secondTurnThread = yield* harness.waitForThread( + THREAD_ID, + (entry) => + entry.latestTurnId === "turn-2" && + entry.checkpoints.length === 2 && + entry.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 2), + ); + const secondCheckpoint = secondTurnThread.checkpoints.find( + (checkpoint) => checkpoint.checkpointTurnCount === 2, + ); + assert.equal(secondCheckpoint?.files.some((file) => file.path === "README.md"), true); + + const checkpointRows = yield* harness.checkpointRepository.listByThreadId({ + threadId: THREAD_ID, + }); + assert.deepEqual( + checkpointRows.map((row) => row.checkpointTurnCount), + [1, 2], + ); + + const diffBlob = yield* harness.waitForDiffBlob({ + threadId: THREAD_ID, + fromTurnCount: 1, + toTurnCount: 2, + }); + assert.equal(diffBlob.diff.includes("README.md"), true); + + assert.equal( + gitShowFileAtRef(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 1), "README.md"), + "v2\n", + ); + assert.equal( + gitShowFileAtRef(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 2), "README.md"), + "v3\n", + ); + }), + ), +); + +it.live("tracks approval requests and resolves pending approvals on user response", () => + withHarness((harness) => + Effect.gen(function* () { + yield* seedProjectAndThread(harness); + + yield* harness.adapterHarness.queueTurnResponseForNextSession({ + events: [ + { + type: "turn.started", + ...runtimeBase("evt-approval-1", "2026-02-24T10:03:00.000Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + }, + { + type: "approval.requested", + ...runtimeBase("evt-approval-2", "2026-02-24T10:03:00.100Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + requestId: APPROVAL_REQUEST_ID, + requestKind: "command", + detail: "Approve command execution", + }, + { + type: "turn.completed", + ...runtimeBase("evt-approval-3", "2026-02-24T10:03:00.200Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + status: "completed", + }, + ], + }); + + yield* startTurn({ + harness, + commandId: "cmd-turn-start-approval", + messageId: "msg-user-approval", + text: "Run command needing approval", + }); + + const thread = yield* harness.waitForThread( + THREAD_ID, + (entry) => entry.activities.some((activity) => activity.kind === "approval.requested"), + ); + assert.equal( + thread.activities.some((activity) => activity.kind === "approval.requested"), + true, + ); + + const pendingRow = yield* harness.waitForPendingApproval( + "req-approval-1", + (row) => row.status === "pending" && row.decision === null, + ); + assert.equal(pendingRow.status, "pending"); + + yield* harness.engine.dispatch({ + type: "thread.approval.respond", + commandId: asCommandId("cmd-approval-respond"), + threadId: THREAD_ID, + requestId: APPROVAL_REQUEST_ID, + decision: "accept", + createdAt: nowIso(), + }); + + const resolvedRow = yield* harness.waitForPendingApproval( + "req-approval-1", + (row) => row.status === "resolved" && row.decision === "accept", + ); + assert.equal(resolvedRow.status, "resolved"); + assert.equal(resolvedRow.decision, "accept"); + + const providerSessionId = thread.session?.providerSessionId ?? "test-session-1"; + const approvalResponses = yield* waitForSync( + () => harness.adapterHarness.getApprovalResponses(providerSessionId), + (responses) => responses.length === 1, + "provider approval response", + ); + assert.equal(approvalResponses.length, 1); + assert.equal(approvalResponses[0]?.requestId, "req-approval-1"); + assert.equal(approvalResponses[0]?.decision, "accept"); + }), + ), +); + +it.live("records failed turn runtime state and checkpoint status as error", () => + withHarness((harness) => + Effect.gen(function* () { + yield* seedProjectAndThread(harness); + + yield* harness.adapterHarness.queueTurnResponseForNextSession({ + events: [ + { + type: "turn.started", + ...runtimeBase("evt-failure-1", "2026-02-24T10:04:00.000Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + }, + { + type: "message.delta", + ...runtimeBase("evt-failure-2", "2026-02-24T10:04:00.100Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + delta: "Partial output before failure.\n", + }, + { + type: "runtime.error", + ...runtimeBase("evt-failure-3", "2026-02-24T10:04:00.200Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + message: "Sandbox command failed.", + }, + { + type: "turn.completed", + ...runtimeBase("evt-failure-4", "2026-02-24T10:04:00.300Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + status: "failed", + errorMessage: "Sandbox command failed.", + }, + ], + }); + + yield* startTurn({ + harness, + commandId: "cmd-turn-start-failure", + messageId: "msg-user-failure", + text: "Run risky command", + }); + + const thread = yield* harness.waitForThread( + THREAD_ID, + (entry) => + entry.session?.status === "error" && + entry.session?.lastError === "Sandbox command failed." && + entry.activities.some((activity) => activity.kind === "runtime.error") && + entry.checkpoints.length === 1, + ); + assert.equal(thread.session?.status, "error"); + assert.equal(thread.checkpoints[0]?.status, "error"); + + const checkpointRow = yield* harness.checkpointRepository.getByThreadAndTurnCount({ + threadId: THREAD_ID, + checkpointTurnCount: 1, + }); + assert.equal(Option.isSome(checkpointRow), true); + if (Option.isSome(checkpointRow)) { + assert.equal(checkpointRow.value.status, "error"); + } + assert.equal(gitRefExists(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 1)), true); + }), + ), +); + +it.live("reverts to an earlier checkpoint and trims checkpoint projections + git refs", () => + withHarness((harness) => + Effect.gen(function* () { + yield* seedProjectAndThread(harness); + + yield* harness.adapterHarness.queueTurnResponseForNextSession({ + events: [ + { + type: "turn.started", + ...runtimeBase("evt-revert-1", "2026-02-24T10:05:00.000Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + }, + { + type: "turn.completed", + ...runtimeBase("evt-revert-2", "2026-02-24T10:05:00.100Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + status: "completed", + }, + ], + mutateWorkspace: ({ cwd }) => + Effect.sync(() => { + fs.writeFileSync(path.join(cwd, "README.md"), "v2\n", "utf8"); + }), + }); + yield* startTurn({ + harness, + commandId: "cmd-turn-start-revert-1", + messageId: "msg-user-revert-1", + text: "First edit", + }); + + const firstTurnThread = yield* harness.waitForThread( + THREAD_ID, + (entry) => entry.session?.providerSessionId !== null && entry.checkpoints.length === 1, + ); + const sessionId = firstTurnThread.session?.providerSessionId; + assert.equal(sessionId !== null, true); + if (!sessionId) { + return; + } + + yield* harness.adapterHarness.queueTurnResponse(sessionId, { + events: [ + { + type: "turn.started", + ...runtimeBase("evt-revert-3", "2026-02-24T10:05:01.000Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + }, + { + type: "turn.completed", + ...runtimeBase("evt-revert-4", "2026-02-24T10:05:01.100Z"), + threadId: FIXTURE_THREAD_ID, + turnId: FIXTURE_TURN_ID, + status: "completed", + }, + ], + mutateWorkspace: ({ cwd }) => + Effect.sync(() => { + fs.writeFileSync(path.join(cwd, "README.md"), "v3\n", "utf8"); + }), + }); + yield* startTurn({ + harness, + commandId: "cmd-turn-start-revert-2", + messageId: "msg-user-revert-2", + text: "Second edit", + }); + + yield* harness.waitForThread( + THREAD_ID, + (entry) => entry.latestTurnId === "turn-2" && entry.checkpoints.length === 2, + 8000, + ); + + yield* harness.engine.dispatch({ + type: "thread.checkpoint.revert", + commandId: asCommandId("cmd-checkpoint-revert"), + threadId: THREAD_ID, + turnCount: 1, + createdAt: nowIso(), + }); + + yield* harness.waitForDomainEvent((event) => event.type === "thread.reverted"); + const revertedThread = yield* harness.waitForThread( + THREAD_ID, + (entry) => + entry.checkpoints.length === 1 && entry.checkpoints[0]?.checkpointTurnCount === 1, + ); + assert.equal(revertedThread.checkpoints[0]?.checkpointTurnCount, 1); + assert.equal(fs.readFileSync(path.join(harness.workspaceDir, "README.md"), "utf8"), "v2\n"); + assert.equal(gitRefExists(harness.workspaceDir, checkpointRefForTurn(THREAD_ID, 2)), false); + assert.deepEqual(harness.adapterHarness.getRollbackCalls(sessionId), [1]); + + const checkpointRows = yield* harness.checkpointRepository.listByThreadId({ + threadId: THREAD_ID, + }); + assert.equal(checkpointRows.length, 1); + }), + ), +); + +it.live("appends checkpoint.revert.failed activity when revert is requested without an active session", () => + withHarness((harness) => + Effect.gen(function* () { + yield* seedProjectAndThread(harness); + + yield* harness.engine.dispatch({ + type: "thread.checkpoint.revert", + commandId: asCommandId("cmd-checkpoint-revert-no-session"), + threadId: THREAD_ID, + turnCount: 0, + createdAt: nowIso(), + }); + + const thread = yield* harness.waitForThread( + THREAD_ID, + (entry) => + entry.activities.some( + (activity) => + activity.kind === "checkpoint.revert.failed" && + typeof activity.payload === "object" && + activity.payload !== null, + ), + ); + const failureActivity = thread.activities.find( + (activity) => activity.kind === "checkpoint.revert.failed", + ); + assert.equal(failureActivity !== undefined, true); + assert.equal( + String((failureActivity?.payload as { readonly detail?: string } | undefined)?.detail).includes( + "No active provider session", + ), + true, + ); + }), + ), +); diff --git a/apps/server/integration/providerService.integration.test.ts b/apps/server/integration/providerService.integration.test.ts index f630e0619afea5bd0850be37ec30f20ceb682496..e3337094e93ae67574834da0b84c2d41e8dbaf68 100644 --- a/apps/server/integration/providerService.integration.test.ts +++ b/apps/server/integration/providerService.integration.test.ts @@ -1,4 +1,4 @@ -import type { ProviderRuntimeEvent } from "@t3tools/contracts"; +import type { ProviderRuntimeEvent, ProviderSessionId } from "@t3tools/contracts"; import { NodeServices } from "@effect/platform-node"; import { it, assert } from "@effect/vitest"; import { Effect, FileSystem, Layer, Path, Queue, Stream } from "effect"; @@ -92,7 +92,7 @@ const collectEventsDuring = ( const runTurn = (input: { readonly provider: ProviderServiceShape; readonly harness: TestProviderAdapterHarness; - readonly sessionId: string; + readonly sessionId: ProviderSessionId; readonly userText: string; readonly response: TestTurnResponse; }) => diff --git a/apps/server/src/checkpointing/Layers/CheckpointStore.ts b/apps/server/src/checkpointing/Layers/CheckpointStore.ts index 7e4615d30950a0031e744c9239cc9e023955bf9d..dd13373151faa3623c9a60a229b71301d0f68c0f 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointStore.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointStore.ts @@ -132,14 +132,12 @@ const makeCheckpointStore = Effect.gen(function* () { }); const treeOid = writeTreeResult.stdout.trim(); if (treeOid.length === 0) { - return yield* Effect.fail( - new GitCommandError({ - operation, - command: "git write-tree", - cwd: normalizedCwd, - detail: "git write-tree returned an empty tree oid.", - }), - ); + return yield* new GitCommandError({ + operation, + command: "git write-tree", + cwd: normalizedCwd, + detail: "git write-tree returned an empty tree oid.", + }); } const message = `t3 checkpoint ref=${checkpointRef}`; @@ -151,14 +149,12 @@ const makeCheckpointStore = Effect.gen(function* () { }); const commitOid = commitTreeResult.stdout.trim(); if (commitOid.length === 0) { - return yield* Effect.fail( - new GitCommandError({ - operation, - command: "git commit-tree", - cwd: normalizedCwd, - detail: "git commit-tree returned an empty commit oid.", - }), - ); + return yield* new GitCommandError({ + operation, + command: "git commit-tree", + cwd: normalizedCwd, + detail: "git commit-tree returned an empty commit oid.", + }); } yield* git.execute({ @@ -183,11 +179,9 @@ const makeCheckpointStore = Effect.gen(function* () { }); const hasCheckpointRef: CheckpointStoreShape["hasCheckpointRef"] = (input) => - Effect.gen(function* () { - return yield* resolveCheckpointCommit(input.cwd, input.checkpointRef).pipe( - Effect.map((commit) => commit !== null), - ); - }); + resolveCheckpointCommit(input.cwd, input.checkpointRef).pipe( + Effect.map((commit) => commit !== null), + ); const restoreCheckpoint: CheckpointStoreShape["restoreCheckpoint"] = (input) => Effect.gen(function* () { @@ -241,14 +235,12 @@ const makeCheckpointStore = Effect.gen(function* () { } if (!fromCommitOid || !toCommitOid) { - return yield* Effect.fail( - new GitCommandError({ - operation, - command: "git diff", - cwd: input.cwd, - detail: "Checkpoint ref is unavailable for diff operation.", - }), - ); + return yield* new GitCommandError({ + operation, + command: "git diff", + cwd: input.cwd, + detail: "Checkpoint ref is unavailable for diff operation.", + }); } const result = yield* git.execute({ diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index 74a4dc9f1804ad2a58efdd8b5c0a534086916f55..6142627784829c59bd54c6d918b927041138e937 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -103,12 +103,10 @@ const makeProviderSessionDirectory = Effect.gen(function* () { const existingRuntime = Option.getOrUndefined(existing); const resolvedThreadId = binding.threadId ?? existingRuntime?.threadId; if (!resolvedThreadId) { - return yield* Effect.fail( - new ProviderValidationError({ - operation: "ProviderSessionDirectory.upsert", - issue: "threadId must be a non-empty string.", - }), - ); + return yield* new ProviderValidationError({ + operation: "ProviderSessionDirectory.upsert", + issue: "threadId must be a non-empty string.", + }); } const now = new Date().toISOString(); diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index 834e408943611433145a64c13be302a159468a6d..38b19e621989ac917a0957af456d15386553e63a 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -189,7 +189,13 @@ function waitForMessage(ws: WebSocket): Promise { async function sendRequest(ws: WebSocket, method: string, params?: unknown): Promise { const id = crypto.randomUUID(); - const message = JSON.stringify({ id, method, ...(params !== undefined ? { params } : {}) }); + const body = + method === ORCHESTRATION_WS_METHODS.dispatchCommand + ? { _tag: method, command: params } + : params && typeof params === "object" && !Array.isArray(params) + ? { _tag: method, ...(params as Record) } + : { _tag: method }; + const message = JSON.stringify({ id, body }); ws.send(message); // Wait for response with matching id @@ -198,6 +204,9 @@ async function sendRequest(ws: WebSocket, method: string, params?: unknown): Pro if (parsed.id === id) { return parsed as WsResponse; } + if (parsed.id === "unknown") { + return parsed as WsResponse; + } } } @@ -658,18 +667,27 @@ describe("WebSocket Server", () => { const response = await sendRequest(ws, "nonexistent.method"); expect(response.error).toBeDefined(); - expect(response.error!.message).toContain("Unknown method"); + expect(response.error!.message).toContain("Invalid request format"); }); it("keeps orchestration domain push behavior for provider runtime events", async () => { const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); + const sessionId = "sess-test"; const emitRuntimeEvent = (event: ProviderRuntimeEvent) => { Effect.runSync(PubSub.publish(runtimeEventPubSub, event)); }; const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; const providerService: ProviderServiceShape = { - startSession: () => unsupported(), - sendTurn: () => unsupported(), + startSession: () => + Effect.succeed({ + sessionId, + provider: "codex", + status: "ready", + threadId: "provider-thread-1", + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }), + sendTurn: () => Effect.succeed({ threadId: "provider-thread-1", turnId: "provider-turn-1" }), interruptTurn: () => unsupported(), respondToRequest: () => unsupported(), stopSession: () => unsupported(), @@ -716,29 +734,30 @@ describe("WebSocket Server", () => { }); expect(createThreadResponse.error).toBeUndefined(); - const setSessionResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.dispatchCommand, { - type: "thread.session", - commandId: "cmd-ws-runtime-session-set", + const startTurnResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.dispatchCommand, { + type: "thread.turn.start", + commandId: "cmd-ws-runtime-turn-start", threadId: "thread-1", - session: { - sessionId: "sess-test", - provider: "codex", - status: "ready", - threadId: "thread-1", - activeTurnId: null, - createdAt, - updatedAt: createdAt, - lastError: null, + message: { + messageId: "msg-ws-runtime-1", + role: "user", + text: "hello", + attachments: [], }, createdAt, }); - expect(setSessionResponse.error).toBeUndefined(); + expect(startTurnResponse.error).toBeUndefined(); + + await waitForPush(ws, ORCHESTRATION_WS_CHANNELS.domainEvent, (push) => { + const event = push.data as { type?: string }; + return event.type === "thread.session-set"; + }); emitRuntimeEvent({ type: "message.delta", eventId: "evt-ws-runtime-message-delta", provider: "codex", - sessionId: "sess-test", + sessionId, createdAt: new Date().toISOString(), turnId: "turn-1", itemId: "item-1", @@ -746,13 +765,16 @@ describe("WebSocket Server", () => { }); const domainPush = await waitForPush(ws, ORCHESTRATION_WS_CHANNELS.domainEvent, (push) => { - const event = push.data as { type?: string; payload?: { id?: string; text?: string } }; - return event.type === "message.sent" && event.payload?.id === "assistant:item-1"; + const event = push.data as { type?: string; payload?: { messageId?: string; text?: string } }; + return event.type === "thread.message-sent" && event.payload?.messageId === "assistant:item-1"; }); - const domainEvent = domainPush.data as { type: string; payload: { id: string; text: string } }; - expect(domainEvent.type).toBe("message.sent"); - expect(domainEvent.payload.id).toBe("assistant:item-1"); + const domainEvent = domainPush.data as { + type: string; + payload: { messageId: string; text: string }; + }; + expect(domainEvent.type).toBe("thread.message-sent"); + expect(domainEvent.payload.messageId).toBe("assistant:item-1"); expect(domainEvent.payload.text).toBe("hello from runtime"); }); @@ -895,7 +917,7 @@ describe("WebSocket Server", () => { expect(response!.error!.message).toContain("Invalid request format"); }); - it("returns unknown method for removed projects CRUD methods", async () => { + it("returns errors for removed projects CRUD methods", async () => { server = createTestServer({ cwd: "/test" }); await server.start(); const addr = server.httpServer.address(); @@ -907,19 +929,19 @@ describe("WebSocket Server", () => { const listResponse = await sendRequest(ws, WS_METHODS.projectsList); expect(listResponse.result).toBeUndefined(); - expect(listResponse.error?.message).toContain("Unknown method"); + expect(listResponse.error?.message).toContain("Invalid request format"); const addResponse = await sendRequest(ws, WS_METHODS.projectsAdd, { cwd: "/tmp/project-a", }); expect(addResponse.result).toBeUndefined(); - expect(addResponse.error?.message).toContain("Unknown method"); + expect(addResponse.error?.message).toContain("Invalid request format"); const removeResponse = await sendRequest(ws, WS_METHODS.projectsRemove, { id: "project-a", }); expect(removeResponse.result).toBeUndefined(); - expect(removeResponse.error?.message).toContain("Unknown method"); + expect(removeResponse.error?.message).toContain("Invalid request format"); }); it("supports projects.searchEntries", async () => { @@ -989,7 +1011,7 @@ describe("WebSocket Server", () => { expect(pullResponse.error?.message).not.toContain("Unknown method"); }); - it("returns unknown method for git.status", async () => { + it("returns error for git.status", async () => { const gitManager = { status: vi.fn().mockResolvedValue({ branch: "feature/test", diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index 3399b30578cef0c2d71e652b6753b77498dc6588..0fbfa82d37366ebd5e527f4c04fff865a86ba8c6 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -132,6 +132,45 @@ function parseBooleanEnv(value: string | undefined): boolean | undefined { return undefined; } +function websocketRawToString(raw: unknown): string | null { + if (typeof raw === "string") { + return raw; + } + if (raw instanceof Uint8Array) { + return Buffer.from(raw).toString("utf8"); + } + if (raw instanceof ArrayBuffer) { + return Buffer.from(new Uint8Array(raw)).toString("utf8"); + } + if (Array.isArray(raw)) { + const chunks: string[] = []; + for (const chunk of raw) { + if (typeof chunk === "string") { + chunks.push(chunk); + continue; + } + if (chunk instanceof Uint8Array) { + chunks.push(Buffer.from(chunk).toString("utf8")); + continue; + } + if (chunk instanceof ArrayBuffer) { + chunks.push(Buffer.from(new Uint8Array(chunk)).toString("utf8")); + continue; + } + return null; + } + return chunks.join(""); + } + return null; +} + +function stripRequestTag( + body: T, +): Omit { + const { _tag: _ignoredTag, ...rest } = body; + return rest; +} + type EffectRuntime = ManagedRuntime.ManagedRuntime< | OrchestrationEngineService | ProjectionSnapshotQuery @@ -360,9 +399,19 @@ export function createServer(options: ServerOptions) { }); async function handleMessage(ws: WebSocket, raw: unknown) { + const messageText = websocketRawToString(raw); + if (messageText === null) { + const errorResponse: WsResponse = { + id: "unknown", + error: { message: "Invalid request format" }, + }; + ws.send(JSON.stringify(errorResponse)); + return; + } + let request: WebSocketRequest; try { - request = Schema.decodeUnknownSync(Schema.fromJsonString(WebSocketRequest))(raw); + request = Schema.decodeUnknownSync(Schema.fromJsonString(WebSocketRequest))(messageText); } catch { const errorResponse: WsResponse = { id: "unknown", @@ -406,17 +455,18 @@ export function createServer(options: ServerOptions) { } case ORCHESTRATION_WS_METHODS.getTurnDiff: { - const blob = await effectRuntime.runPromise(diffBlobRepository.get(request.body)); + const body = stripRequestTag(request.body); + const blob = await effectRuntime.runPromise(diffBlobRepository.get(body)); if (Option.isNone(blob)) { throw new Error( - `Turn diff not found for thread '${request.body.threadId}' from ${request.body.fromTurnCount} to ${request.body.toTurnCount}.`, + `Turn diff not found for thread '${body.threadId}' from ${body.fromTurnCount} to ${body.toTurnCount}.`, ); } return { - threadId: request.body.threadId, - fromTurnCount: request.body.fromTurnCount, - toTurnCount: request.body.toTurnCount, + threadId: body.threadId, + fromTurnCount: body.fromTurnCount, + toTurnCount: body.toTurnCount, diff: blob.value?.diff, }; } @@ -434,7 +484,7 @@ export function createServer(options: ServerOptions) { ); case WS_METHODS.projectsSearchEntries: - return searchWorkspaceEntries(request.body); + return searchWorkspaceEntries(stripRequestTag(request.body)); case WS_METHODS.shellOpenInEditor: { const body = request.body; @@ -500,48 +550,48 @@ export function createServer(options: ServerOptions) { // } case WS_METHODS.gitPull: - return pullGitBranch(request.body); + return pullGitBranch(stripRequestTag(request.body)); case WS_METHODS.gitRunStackedAction: - return gitManager.runStackedAction(request.body); + return gitManager.runStackedAction(stripRequestTag(request.body)); case WS_METHODS.gitListBranches: - return listGitBranches(request.body); + return listGitBranches(stripRequestTag(request.body)); case WS_METHODS.gitCreateWorktree: - return createGitWorktree(request.body); + return createGitWorktree(stripRequestTag(request.body)); case WS_METHODS.gitRemoveWorktree: - return removeGitWorktree(request.body); + return removeGitWorktree(stripRequestTag(request.body)); case WS_METHODS.gitCreateBranch: - return createGitBranch(request.body); + return createGitBranch(stripRequestTag(request.body)); case WS_METHODS.gitCheckout: - return checkoutGitBranch(request.body); + return checkoutGitBranch(stripRequestTag(request.body)); case WS_METHODS.gitInit: - return initGitRepo(request.body); + return initGitRepo(stripRequestTag(request.body)); case WS_METHODS.terminalOpen: - return terminalManager.open(request.body); + return terminalManager.open(stripRequestTag(request.body)); case WS_METHODS.terminalWrite: - await terminalManager.write(request.body); + await terminalManager.write(stripRequestTag(request.body)); return undefined; case WS_METHODS.terminalResize: - await terminalManager.resize(request.body); + await terminalManager.resize(stripRequestTag(request.body)); return undefined; case WS_METHODS.terminalClear: - await terminalManager.clear(request.body); + await terminalManager.clear(stripRequestTag(request.body)); return undefined; case WS_METHODS.terminalRestart: - return terminalManager.restart(request.body); + return terminalManager.restart(stripRequestTag(request.body)); case WS_METHODS.terminalClose: - await terminalManager.close(request.body); + await terminalManager.close(stripRequestTag(request.body)); return undefined; case WS_METHODS.serverGetConfig: @@ -551,7 +601,7 @@ export function createServer(options: ServerOptions) { }; case WS_METHODS.serverUpsertKeybinding: - keybindingsConfig = upsertKeybindingRule(logger, request.body); + keybindingsConfig = upsertKeybindingRule(logger, stripRequestTag(request.body)); return { keybindings: keybindingsConfig, }; diff --git a/apps/server/tsconfig.json b/apps/server/tsconfig.json index f136dcf9dee28ba74c6340a9b3f9189b33c007b6..23300dbb118c26b6ee0b591cbc7e337918d81b92 100644 --- a/apps/server/tsconfig.json +++ b/apps/server/tsconfig.json @@ -12,6 +12,6 @@ } ] }, - "include": ["src", "tsdown.config.ts", "scripts"], - "exclude": ["src/**/*.test.ts", "integration"] + "include": ["src", "tsdown.config.ts", "scripts", "integration"], + "exclude": ["src/**/*.test.ts"] } diff --git a/apps/server/tsdown.config.ts b/apps/server/tsdown.config.ts index 1a67bab626281e5ad6623586828a15525b10dbaa..ebc765577236aeaa44eb9bddaf6882860ae364e3 100644 --- a/apps/server/tsdown.config.ts +++ b/apps/server/tsdown.config.ts @@ -6,7 +6,7 @@ export default defineConfig({ outDir: "dist", sourcemap: true, clean: true, - noExternal: ["@t3tools/contracts", "zod"], + noExternal: ["@t3tools/contracts"], inlineOnly: false, banner: { js: "#!/usr/bin/env node\n", diff --git a/apps/web/package.json b/apps/web/package.json index e3aa497bd76a29854840f65e5c9474cb3fbcce92..e825c6f26b338ccba06d53bcf674be5986b55ad4 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -21,6 +21,7 @@ "@xterm/addon-fit": "^0.11.0", "@xterm/xterm": "^6.0.0", "class-variance-authority": "^0.7.1", + "effect": "4.0.0-beta.10", "highlight.js": "^11.11.1", "lucide-react": "^0.564.0", "react": "^19.0.0", @@ -28,8 +29,7 @@ "react-markdown": "^10.1.0", "rehype-highlight": "^7.0.2", "remark-gfm": "^4.0.1", - "tailwind-merge": "^3.4.0", - "zod": "^3.24.1" + "tailwind-merge": "^3.4.0" }, "devDependencies": { "@tailwindcss/vite": "^4.0.0", diff --git a/apps/web/src/appSettings.ts b/apps/web/src/appSettings.ts index cd3da7224c7855b1c2cd446925bb5bcbe663e479..046ca14383a25bf024ff0365890fcab35bb28465 100644 --- a/apps/web/src/appSettings.ts +++ b/apps/web/src/appSettings.ts @@ -1,17 +1,20 @@ import { useCallback, useSyncExternalStore } from "react"; -import { z } from "zod"; +import { Option, Schema } from "effect"; const APP_SETTINGS_STORAGE_KEY = "t3code:app-settings:v1"; -const appSettingsSchema = z.object({ - codexBinaryPath: z.string().trim().max(4096).default(""), - codexHomePath: z.string().trim().max(4096).default(""), - confirmThreadDelete: z.boolean().default(true), +const AppSettingsSchema = Schema.Struct({ + codexBinaryPath: Schema.NonEmptyString.check(Schema.isMaxLength(4096)).pipe( + Schema.withConstructorDefault(() => Option.some("")), + ), + codexHomePath: Schema.NonEmptyString.check(Schema.isMaxLength(4096)).pipe( + Schema.withConstructorDefault(() => Option.some("")), + ), + confirmThreadDelete: Schema.Boolean.pipe(Schema.withConstructorDefault(() => Option.some(true))), }); +export type AppSettings = typeof AppSettingsSchema.Type; -export type AppSettings = z.infer; - -const DEFAULT_APP_SETTINGS: AppSettings = appSettingsSchema.parse({}); +const DEFAULT_APP_SETTINGS = AppSettingsSchema.makeUnsafe({}); let listeners: Array<() => void> = []; let cachedRawSettings: string | null | undefined; @@ -29,8 +32,7 @@ function parsePersistedSettings(value: string | null): AppSettings { } try { - const parsed = JSON.parse(value) as unknown; - return appSettingsSchema.parse(parsed); + return Schema.decodeSync(Schema.fromJsonString(AppSettingsSchema))(value); } catch { return DEFAULT_APP_SETTINGS; } @@ -84,10 +86,14 @@ function subscribe(listener: () => void): () => void { } export function useAppSettings() { - const settings = useSyncExternalStore(subscribe, getAppSettingsSnapshot, () => DEFAULT_APP_SETTINGS); + const settings = useSyncExternalStore( + subscribe, + getAppSettingsSnapshot, + () => DEFAULT_APP_SETTINGS, + ); const updateSettings = useCallback((patch: Partial) => { - const next = appSettingsSchema.parse({ + const next = Schema.decodeSync(AppSettingsSchema)({ ...getAppSettingsSnapshot(), ...patch, }); diff --git a/apps/web/src/components/BranchToolbar.logic.ts b/apps/web/src/components/BranchToolbar.logic.ts index 03d04bc3cf39c65a66edd6480242c201abc9ca19..ab83f2ab93149018c1dc204300e9f2af71f903bc 100644 --- a/apps/web/src/components/BranchToolbar.logic.ts +++ b/apps/web/src/components/BranchToolbar.logic.ts @@ -5,7 +5,7 @@ interface DeriveSyncedLocalBranchInput { activeWorktreePath: string | null; envMode: "local" | "worktree"; activeThreadBranch: string | null; - queryBranches: GitBranch[] | undefined; + queryBranches: ReadonlyArray | undefined; } export function deriveSyncedLocalBranch({ diff --git a/apps/web/src/components/BranchToolbar.tsx b/apps/web/src/components/BranchToolbar.tsx index 6fe19dfd95464492a9a272604cae71be6e6219e2..60415140eb818e28e9e8d3c4fb88f98b0a174070 100644 --- a/apps/web/src/components/BranchToolbar.tsx +++ b/apps/web/src/components/BranchToolbar.tsx @@ -3,6 +3,7 @@ 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 { gitBranchesQueryOptions, gitCheckoutMutationOptions, @@ -95,11 +96,10 @@ export default function BranchToolbar({ if (api) { void api.orchestration.dispatchCommand({ type: "thread.meta.update", - commandId: crypto.randomUUID(), - threadId: activeThreadId, + commandId: newCommandId(), + threadId: asThreadId(activeThreadId), branch: syncedBranch, worktreePath: null, - createdAt: new Date().toISOString(), }); } dispatch({ @@ -127,17 +127,23 @@ export default function BranchToolbar({ // 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; - if (sessionId && worktreePath !== activeWorktreePath) { - void api?.providers.stopSession({ sessionId }).catch(() => undefined); + if (sessionId && worktreePath !== activeWorktreePath && api) { + void api.orchestration + .dispatchCommand({ + type: "thread.session.stop", + commandId: newCommandId(), + threadId: asThreadId(activeThreadId), + createdAt: new Date().toISOString(), + }) + .catch(() => undefined); } if (api) { void api.orchestration.dispatchCommand({ type: "thread.meta.update", - commandId: crypto.randomUUID(), - threadId: activeThreadId, + commandId: newCommandId(), + threadId: asThreadId(activeThreadId), branch, worktreePath, - createdAt: new Date().toISOString(), }); } dispatch({ type: "SET_THREAD_BRANCH", threadId: activeThreadId, branch, worktreePath }); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index c308809c55710c4e9ba6e6328228e7f481259c8e..13653399fac8899d977627ba48e16afc8cd4d1e2 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -5,12 +5,10 @@ import { type ProjectEntry, type ProjectScript, ModelSlug, - PROVIDER_SEND_TURN_MAX_INPUT_CHARS, PROVIDER_SEND_TURN_MAX_ATTACHMENTS, PROVIDER_SEND_TURN_MAX_IMAGE_BYTES, type ResolvedKeybindingsConfig, type ProviderApprovalDecision, - type ProviderSendTurnAttachmentInput, } from "@t3tools/contracts"; import { type ClipboardEvent, @@ -36,8 +34,6 @@ import { serverConfigQueryOptions, serverQueryKeys } from "~/lib/serverReactQuer import { isElectron } from "../env"; import { parseDiffRouteSearch, stripDiffSearchParams } from "../diffRouteSearch"; -import { useAppSettings } from "../appSettings"; -import { buildBootstrapInput } from "../historyBootstrap"; import { type ComposerTriggerKind, detectComposerTrigger, @@ -116,6 +112,13 @@ import { setupProjectScript, } from "~/projectScripts"; import { Toggle } from "./ui/toggle"; +import { + asApprovalRequestId, + asProjectId, + asThreadId, + newCommandId, + newMessageId, +} from "~/lib/orchestrationIds"; function formatMessageMeta(createdAt: string, duration: string | null): string { if (!duration) return formatTimestamp(createdAt); @@ -165,14 +168,6 @@ function workToneClass(tone: "thinking" | "tool" | "info" | "error"): string { return "text-muted-foreground/40"; } -type SessionContinuityState = "resumed" | "new" | "fallback_new"; - -interface EnsuredSessionInfo { - sessionId: string; - resolvedThreadId: string | null; - continuityState: SessionContinuityState; -} - interface ComposerImageAttachment extends Omit { previewUrl: string; file: File; @@ -351,7 +346,6 @@ export default function ChatView({ threadId }: ChatViewProps) { const rawSearch = useSearch({ strict: false }); const api = useNativeApi(); const { resolvedTheme } = useTheme(); - const { settings: appSettings } = useAppSettings(); const queryClient = useQueryClient(); const createWorktreeMutation = useMutation( gitCreateWorktreeMutationOptions({ api, queryClient }), @@ -383,8 +377,6 @@ export default function ChatView({ threadId }: ChatViewProps) { const composerImagesRef = useRef([]); const dragDepthRef = useRef(0); const terminalOpenByThreadRef = useRef>({}); - const checkpointHydrationSessionRequestRef = useRef(new Set()); - const checkpointTurnCountSyncFingerprintRef = useRef(new Map()); const activeThread = state.threads.find((t) => t.id === threadId); const diffSearch = useMemo( @@ -393,9 +385,6 @@ export default function ChatView({ threadId }: ChatViewProps) { ); const diffOpen = diffSearch.diff === "1"; const activeThreadId = activeThread?.id ?? null; - const activeSessionId = activeThread?.session?.sessionId ?? null; - const activeThreadRuntimeId = - activeThread?.codexThreadId ?? activeThread?.session?.threadId ?? null; const activeProject = state.projects.find((p) => p.id === activeThread?.projectId); const selectedModel = resolveModelSlug( activeThread?.model ?? activeProject?.model ?? DEFAULT_MODEL, @@ -521,11 +510,6 @@ export default function ChatView({ threadId }: ChatViewProps) { completionSummary, timelineEntries, ]); - const runtimeApprovalPolicy = state.runtimeMode === "full-access" ? "never" : "on-request"; - const runtimeSandboxMode = - state.runtimeMode === "full-access" ? "danger-full-access" : "workspace-write"; - const codexBinaryPath = appSettings.codexBinaryPath.trim(); - const codexHomePath = appSettings.codexHomePath.trim(); const gitCwd = activeThread?.worktreePath ?? activeProject?.cwd ?? null; const composerTrigger = useMemo( () => detectComposerTrigger(prompt, composerCursor), @@ -820,17 +804,16 @@ export default function ChatView({ threadId }: ChatViewProps) { scripts: input.nextScripts, }); - if (!isElectron || !api) return; - - let scriptsPersisted = false; + if (!api) return; try { - const updated = await api.projects.updateScripts({ - id: input.projectId, + await api.orchestration.dispatchCommand({ + type: "project.meta.update", + commandId: newCommandId(), + projectId: asProjectId(input.projectId), scripts: input.nextScripts, }); - scriptsPersisted = true; - if (input.keybinding) { + if (isElectron && input.keybinding) { const keybindingUpdate = await api.server.upsertKeybinding({ key: input.keybinding, command: input.keybindingCommand, @@ -843,24 +826,10 @@ export default function ChatView({ threadId }: ChatViewProps) { : { cwd: input.projectCwd, keybindings: keybindingUpdate.keybindings, - }, + }, ); } - - dispatch({ - type: "SET_PROJECT_SCRIPTS", - projectId: updated.project.id, - scripts: updated.project.scripts, - }); } catch (error) { - if (scriptsPersisted) { - await api.projects - .updateScripts({ - id: input.projectId, - scripts: input.previousScripts, - }) - .catch(() => undefined); - } dispatch({ type: "SET_PROJECT_SCRIPTS", projectId: input.projectId, @@ -946,17 +915,25 @@ export default function ChatView({ threadId }: ChatViewProps) { scheduleComposerFocus(); if (!api) return; - const sessionIds = state.threads - .map((t) => t.session) - .filter((s): s is NonNullable => s !== null && s.status !== "closed") - .map((s) => s.sessionId); + const runningThreadIds = state.threads + .filter((thread) => thread.session !== null && thread.session.status !== "closed") + .map((thread) => thread.id); - if (sessionIds.length === 0) return; + if (runningThreadIds.length === 0) return; setIsSwitchingRuntimeMode(true); try { await Promise.all( - sessionIds.map((id) => api.providers.stopSession({ sessionId: id }).catch(() => undefined)), + runningThreadIds.map((threadId) => + api.orchestration + .dispatchCommand({ + type: "thread.session.stop", + commandId: newCommandId(), + threadId: asThreadId(threadId), + createdAt: new Date().toISOString(), + }) + .catch(() => undefined), + ), ); } finally { setIsSwitchingRuntimeMode(false); @@ -1321,167 +1298,6 @@ export default function ChatView({ threadId }: ChatViewProps) { focusComposer(); }; - const syncCheckpointTurnCounts = useCallback( - async (input: { - threadId: string; - sessionId: string; - threadRuntimeId: string | null; - }): Promise => { - if (!api || turnDiffSummaries.length === 0) { - return; - } - - const turnIds = turnDiffSummaries.map((summary) => summary.turnId); - if (turnIds.length === 0) { - return; - } - - const syncKey = `${input.threadId}:${input.sessionId}:${input.threadRuntimeId ?? "none"}`; - const turnFingerprint = turnIds.join(","); - if (checkpointTurnCountSyncFingerprintRef.current.get(syncKey) === turnFingerprint) { - return; - } - checkpointTurnCountSyncFingerprintRef.current.set(syncKey, turnFingerprint); - - try { - await api.providers.listCheckpoints({ - sessionId: input.sessionId, - }); - } catch { - checkpointTurnCountSyncFingerprintRef.current.delete(syncKey); - } - }, - [api, turnDiffSummaries], - ); - - const ensureSession = useCallback( - async (cwdOverride?: string): Promise => { - if (!api || !activeThread || !activeProject) return null; - if (activeThread.session && activeThread.session.status !== "closed") { - const sessionThreadId = activeThread.session.threadId ?? null; - const continuityState: SessionContinuityState = - activeThread.codexThreadId === null - ? "new" - : sessionThreadId === activeThread.codexThreadId - ? "resumed" - : "fallback_new"; - await syncCheckpointTurnCounts({ - threadId: activeThread.id, - sessionId: activeThread.session.sessionId, - threadRuntimeId: sessionThreadId, - }); - return { - sessionId: activeThread.session.sessionId, - resolvedThreadId: sessionThreadId, - continuityState, - } satisfies EnsuredSessionInfo; - } - - const priorCodexThreadId = activeThread.codexThreadId; - setIsConnecting(true); - try { - const session = await api.providers.startSession({ - provider: "codex", - cwd: cwdOverride ?? activeThread.worktreePath ?? activeProject.cwd, - model: selectedModel || undefined, - resumeThreadId: priorCodexThreadId ?? undefined, - ...(codexBinaryPath.length > 0 ? { codexBinaryPath } : {}), - ...(codexHomePath.length > 0 ? { codexHomePath } : {}), - approvalPolicy: runtimeApprovalPolicy, - sandboxMode: runtimeSandboxMode, - }); - await api.orchestration.dispatchCommand({ - type: "thread.session", - commandId: crypto.randomUUID(), - threadId: activeThread.id, - session: { - sessionId: session.sessionId, - provider: session.provider, - status: session.status, - threadId: session.threadId ?? activeThread.id, - activeTurnId: session.activeTurnId ?? null, - createdAt: session.createdAt, - updatedAt: session.updatedAt, - lastError: session.lastError ?? null, - }, - createdAt: new Date().toISOString(), - }); - const resolvedThreadId = session.threadId ?? null; - const continuityState: SessionContinuityState = - priorCodexThreadId === null - ? "new" - : resolvedThreadId === priorCodexThreadId - ? "resumed" - : "fallback_new"; - await syncCheckpointTurnCounts({ - threadId: activeThread.id, - sessionId: session.sessionId, - threadRuntimeId: resolvedThreadId, - }); - return { - sessionId: session.sessionId, - resolvedThreadId, - continuityState, - }; - } catch (err) { - dispatch({ - type: "SET_ERROR", - threadId: activeThread.id, - error: err instanceof Error ? err.message : "Failed to connect.", - }); - return null; - } finally { - setIsConnecting(false); - } - }, - [ - activeProject, - activeThread, - api, - codexBinaryPath, - codexHomePath, - runtimeApprovalPolicy, - runtimeSandboxMode, - selectedModel, - syncCheckpointTurnCounts, - ], - ); - - useEffect(() => { - if (!api || !activeThreadId || activeSessionId) return; - if (phase === "running" || isSending || isConnecting || isRevertingCheckpoint) return; - const hasThreadIdentity = Boolean(activeThreadRuntimeId); - if (!hasThreadIdentity) return; - if (turnDiffSummaries.length === 0) return; - - const requestKey = `${activeThreadId}:${activeThreadRuntimeId ?? "none"}`; - if (checkpointHydrationSessionRequestRef.current.has(requestKey)) { - return; - } - checkpointHydrationSessionRequestRef.current.add(requestKey); - setThreadError(activeThreadId, null); - void ensureSession().then( - () => { - checkpointHydrationSessionRequestRef.current.delete(requestKey); - }, - () => { - // Keep the key on failure so this effect does not re-enter in a tight loop. - }, - ); - }, [ - activeThreadId, - activeThreadRuntimeId, - activeSessionId, - api, - ensureSession, - isConnecting, - isRevertingCheckpoint, - isSending, - phase, - setThreadError, - turnDiffSummaries, - ]); - const onRevertToTurnCount = useCallback( async (turnCount: number) => { if (!api || !activeThread || isRevertingCheckpoint) { @@ -1505,13 +1321,12 @@ export default function ChatView({ threadId }: ChatViewProps) { setIsRevertingCheckpoint(true); setThreadError(activeThread.id, null); try { - const sessionInfo = await ensureSession(); - if (!sessionInfo) { - return; - } - await api.providers.revertToCheckpoint({ - sessionId: sessionInfo.sessionId, + await api.orchestration.dispatchCommand({ + type: "thread.checkpoint.revert", + commandId: newCommandId(), + threadId: asThreadId(activeThread.id), turnCount, + createdAt: new Date().toISOString(), }); } catch (err) { setThreadError( @@ -1525,8 +1340,6 @@ export default function ChatView({ threadId }: ChatViewProps) { [ activeThread, api, - dispatch, - ensureSession, isConnecting, isRevertingCheckpoint, isSending, @@ -1544,7 +1357,6 @@ export default function ChatView({ threadId }: ChatViewProps) { const composerImagesSnapshot = [...composerImages]; // On first message: lock in branch + create worktree if needed. - let sessionCwd: string | undefined; if ( activeThread.messages.length === 0 && activeThread.branch && @@ -1558,17 +1370,13 @@ export default function ChatView({ threadId }: ChatViewProps) { branch: activeThread.branch, newBranch, }); - sessionCwd = result.worktree.path; - if (api) { - await api.orchestration.dispatchCommand({ - type: "thread.meta.update", - commandId: crypto.randomUUID(), - threadId: activeThread.id, - branch: result.worktree.branch, - worktreePath: result.worktree.path, - createdAt: new Date().toISOString(), - }); - } + await api.orchestration.dispatchCommand({ + type: "thread.meta.update", + commandId: newCommandId(), + threadId: asThreadId(activeThread.id), + branch: result.worktree.branch, + worktreePath: result.worktree.path, + }); const setupScript = setupProjectScript(activeProject.scripts); if (setupScript) { void runProjectScript(setupScript, { @@ -1598,51 +1406,31 @@ export default function ChatView({ threadId }: ChatViewProps) { if (api) { await api.orchestration.dispatchCommand({ type: "thread.meta.update", - commandId: crypto.randomUUID(), - threadId: activeThread.id, + commandId: newCommandId(), + threadId: asThreadId(activeThread.id), title, - createdAt: new Date().toISOString(), }); } } setThreadError(activeThread.id, null); - const messageAttachments: ChatImageAttachment[] = composerImagesSnapshot.map((image) => ({ - type: "image", - id: image.id, - name: image.name, - mimeType: image.mimeType, - sizeBytes: image.sizeBytes, - previewUrl: image.previewUrl, - })); - const userMessageId = crypto.randomUUID(); - if (api) { - await api.orchestration.dispatchCommand({ - type: "message.send", - commandId: crypto.randomUUID(), - threadId: activeThread.id, - messageId: userMessageId, - role: "user", - text: trimmed, - streaming: false, - createdAt: new Date().toISOString(), - }); - } - const previousMessages = activeThread.messages; promptRef.current = ""; setPrompt(""); setComposerImages([]); setComposerCursor(0); setComposerHighlightedItemId(null); - const sessionInfo = await ensureSession(sessionCwd); - if (!sessionInfo) return; - setIsSending(true); try { const turnAttachments = await Promise.all( composerImagesSnapshot.map( - async (image): Promise => ({ + async (image): Promise<{ + type: "image"; + name: string; + mimeType: string; + sizeBytes: number; + dataUrl: string; + }> => ({ type: "image", name: image.name, mimeType: image.mimeType, @@ -1651,23 +1439,19 @@ export default function ChatView({ threadId }: ChatViewProps) { }), ), ); - const shouldBootstrap = - previousMessages.length > 0 && - (sessionInfo.continuityState === "new" || sessionInfo.continuityState === "fallback_new"); - const latestPromptForBootstrap = trimmed || IMAGE_ONLY_BOOTSTRAP_PROMPT; - const input = shouldBootstrap - ? buildBootstrapInput( - previousMessages, - latestPromptForBootstrap, - PROVIDER_SEND_TURN_MAX_INPUT_CHARS, - ).text - : trimmed || undefined; - await api.providers.sendTurn({ - sessionId: sessionInfo.sessionId, - ...(input ? { input } : {}), - ...(turnAttachments.length > 0 ? { attachments: turnAttachments } : {}), + await api.orchestration.dispatchCommand({ + type: "thread.turn.start", + commandId: newCommandId(), + threadId: asThreadId(activeThread.id), + message: { + messageId: newMessageId(), + role: "user", + text: trimmed || IMAGE_ONLY_BOOTSTRAP_PROMPT, + attachments: turnAttachments, + }, model: selectedModel || undefined, effort: selectedEffort || undefined, + createdAt: new Date().toISOString(), }); } catch (err) { setThreadError( @@ -1680,25 +1464,30 @@ export default function ChatView({ threadId }: ChatViewProps) { }; const onInterrupt = async () => { - if (!api || !activeThread?.session) return; - await api.providers.interruptTurn({ - sessionId: activeThread.session.sessionId, - turnId: activeThread.session.activeTurnId, + if (!api || !activeThread) return; + await api.orchestration.dispatchCommand({ + type: "thread.turn.interrupt", + commandId: newCommandId(), + threadId: asThreadId(activeThread.id), + createdAt: new Date().toISOString(), }); }; const onRespondToApproval = useCallback( async (requestId: string, decision: ProviderApprovalDecision) => { - if (!api || !activeSessionId || !activeThreadId) return; + if (!api || !activeThreadId) return; setRespondingRequestIds((existing) => existing.includes(requestId) ? existing : [...existing, requestId], ); try { - await api.providers.respondToRequest({ - sessionId: activeSessionId, - requestId, + await api.orchestration.dispatchCommand({ + type: "thread.approval.respond", + commandId: newCommandId(), + threadId: asThreadId(activeThreadId), + requestId: asApprovalRequestId(requestId), decision, + createdAt: new Date().toISOString(), }); } catch (err) { dispatch({ @@ -1710,7 +1499,7 @@ export default function ChatView({ threadId }: ChatViewProps) { setRespondingRequestIds((existing) => existing.filter((id) => id !== requestId)); } }, - [activeSessionId, activeThreadId, api, dispatch], + [activeThreadId, api, dispatch], ); const onModelSelect = useCallback( @@ -1719,10 +1508,9 @@ export default function ChatView({ threadId }: ChatViewProps) { if (api) { void api.orchestration.dispatchCommand({ type: "thread.meta.update", - commandId: crypto.randomUUID(), - threadId: activeThread.id, + commandId: newCommandId(), + threadId: asThreadId(activeThread.id), model: resolveModelSlug(model), - createdAt: new Date().toISOString(), }); } scheduleComposerFocus(); diff --git a/apps/web/src/components/DiffPanel.tsx b/apps/web/src/components/DiffPanel.tsx index 3a25f678132fe6b74841025dd1b7c8a59952bafb..0650081587a804cf15f0da4e77e1aec013093479 100644 --- a/apps/web/src/components/DiffPanel.tsx +++ b/apps/web/src/components/DiffPanel.tsx @@ -99,9 +99,6 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { ); const activeThreadId = routeThreadId; const activeThread = state.threads.find((thread) => thread.id === activeThreadId); - const activeThreadRuntimeId = - activeThread?.codexThreadId ?? activeThread?.session?.threadId ?? null; - const activeSessionId = activeThread?.session?.sessionId ?? null; const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = useTurnDiffSummaries(activeThread); const orderedTurnDiffSummaries = useMemo( @@ -173,8 +170,7 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { }, [orderedTurnDiffSummaries, selectedTurn]); const activeCheckpointDiffQuery = useQuery( checkpointDiffQueryOptions(api, { - sessionId: activeSessionId, - threadRuntimeId: activeThreadRuntimeId, + threadId: activeThreadId, fromTurnCount: activeCheckpointRange?.fromTurnCount ?? null, toTurnCount: activeCheckpointRange?.toTurnCount ?? null, cacheScope: selectedTurn ? `turn:${selectedTurn.turnId}` : conversationCacheScope, diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 7371b19bc411b60333feab25f40951adda40361a..7f45829815b1943b2413dee18dec36bf7b295650 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -7,6 +7,7 @@ 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 { useStore } from "../store"; import { isChatNewLocalShortcut, isChatNewShortcut } from "../keybindings"; import { type Thread } from "../types"; @@ -133,14 +134,14 @@ export default function Sidebar() { }, ) => { if (!api) return; - const threadId = crypto.randomUUID(); + const threadId = newThreadId(); const createdAt = new Date().toISOString(); const model = state.projects.find((project) => project.id === projectId)?.model ?? DEFAULT_MODEL; void api.orchestration.dispatchCommand({ type: "thread.create", - commandId: crypto.randomUUID(), + commandId: newCommandId(), threadId, - projectId, + projectId: asProjectId(projectId), title: "New thread", model, branch: options?.branch ?? null, @@ -152,7 +153,7 @@ export default function Sidebar() { params: { threadId }, }); }, - [api, dispatch, navigate, state.projects], + [api, navigate, state.projects], ); const focusMostRecentThreadForProject = useCallback( @@ -188,16 +189,19 @@ export default function Sidebar() { return; } - const result = await api.projects.add({ cwd }); - const projects = await api.projects.list(); - dispatch({ type: "SYNC_PROJECTS", projects }); - - const hasThread = state.threads.some((thread) => thread.projectId === result.project.id); - if (hasThread) { - focusMostRecentThreadForProject(result.project.id); - return; - } - handleNewThread(result.project.id); + const projectId = newProjectId(); + const createdAt = new Date().toISOString(); + const title = cwd.split(/[/\\]/).filter(Boolean).at(-1) ?? cwd; + await api.orchestration.dispatchCommand({ + type: "project.create", + commandId: newCommandId(), + projectId, + title, + workspaceRoot: cwd, + defaultModel: DEFAULT_MODEL, + createdAt, + }); + handleNewThread(projectId); } finally { setIsAddingProject(false); setNewCwd(""); @@ -206,7 +210,6 @@ export default function Sidebar() { }, [ api, - dispatch, focusMostRecentThreadForProject, handleNewThread, isAddingProject, @@ -267,15 +270,15 @@ export default function Sidebar() { ].join("\n"), )); - // Stop active session if running - if (thread.session?.sessionId) { - try { - await api.providers.stopSession({ - sessionId: thread.session.sessionId, - }); - } catch { - // Session may already be stopped - } + if (thread.session && thread.session.status !== "closed") { + await api.orchestration + .dispatchCommand({ + type: "thread.session.stop", + commandId: newCommandId(), + threadId: asThreadId(threadId), + createdAt: new Date().toISOString(), + }) + .catch(() => undefined); } try { @@ -291,9 +294,8 @@ export default function Sidebar() { const fallbackThreadId = state.threads.find((entry) => entry.id !== threadId)?.id ?? null; await api.orchestration.dispatchCommand({ type: "thread.delete", - commandId: crypto.randomUUID(), - threadId, - createdAt: new Date().toISOString(), + commandId: newCommandId(), + threadId: asThreadId(threadId), }); if (shouldNavigateToFallback) { if (fallbackThreadId) { @@ -369,9 +371,11 @@ export default function Sidebar() { if (!confirmed) return; try { - await api.projects.remove({ id: projectId }); - const projects = await api.projects.list(); - dispatch({ type: "SYNC_PROJECTS", projects }); + await api.orchestration.dispatchCommand({ + type: "project.delete", + commandId: newCommandId(), + projectId: asProjectId(projectId), + }); } catch (error) { const message = error instanceof Error ? error.message : "Unknown error deleting project."; @@ -383,7 +387,7 @@ export default function Sidebar() { }); } }, - [api, dispatch, state.projects, state.threads], + [api, state.projects, state.threads], ); useEffect(() => { diff --git a/apps/web/src/lib/gitReactQuery.ts b/apps/web/src/lib/gitReactQuery.ts index fc0c7160a9b436f38d12da6918cee49674de7036..9a75c9520b4e4599254d30422f9a165405b3b246 100644 --- a/apps/web/src/lib/gitReactQuery.ts +++ b/apps/web/src/lib/gitReactQuery.ts @@ -152,15 +152,17 @@ export function gitCreateWorktreeMutationOptions(input: { cwd, branch, newBranch, + path, }: { cwd: string; branch: string; newBranch: string; + path?: string | null; }) => { if (!input.api) { throw new Error("Git worktree creation is unavailable."); } - return input.api.git.createWorktree({ cwd, branch, newBranch }); + return input.api.git.createWorktree({ cwd, branch, newBranch, path: path ?? null }); }, onSettled: async () => { await invalidateGitQueries(input.queryClient); @@ -177,7 +179,7 @@ export function gitRemoveWorktreeMutationOptions(input: { if (!input.api) { throw new Error("Git worktree removal is unavailable."); } - return input.api.git.removeWorktree({ cwd, path, force }); + return input.api.git.removeWorktree({ cwd, path, force: force ?? null }); }, onSettled: async () => { await invalidateGitQueries(input.queryClient); diff --git a/apps/web/src/lib/orchestrationIds.ts b/apps/web/src/lib/orchestrationIds.ts new file mode 100644 index 0000000000000000000000000000000000000000..110cc39831867e990f4471b5baeea4b25d0d59de --- /dev/null +++ b/apps/web/src/lib/orchestrationIds.ts @@ -0,0 +1,16 @@ +import { ApprovalRequestId, CommandId, MessageId, ProjectId, ThreadId } from "@t3tools/contracts"; + +export const newCommandId = (): CommandId => CommandId.makeUnsafe(crypto.randomUUID()); + +export const newProjectId = (): ProjectId => ProjectId.makeUnsafe(crypto.randomUUID()); + +export const newThreadId = (): ThreadId => ThreadId.makeUnsafe(crypto.randomUUID()); + +export const newMessageId = (): MessageId => MessageId.makeUnsafe(crypto.randomUUID()); + +export const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); + +export const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); + +export const asApprovalRequestId = (value: string): ApprovalRequestId => + ApprovalRequestId.makeUnsafe(value); diff --git a/apps/web/src/lib/providerReactQuery.test.ts b/apps/web/src/lib/providerReactQuery.test.ts index 8c424a7df191f6a32c00b2e14a76ee98de6cf13e..91edd765677fc6cd67fa111037bb7b4dea082ee7 100644 --- a/apps/web/src/lib/providerReactQuery.test.ts +++ b/apps/web/src/lib/providerReactQuery.test.ts @@ -6,8 +6,7 @@ import { checkpointDiffQueryOptions, providerQueryKeys } from "./providerReactQu describe("providerQueryKeys.checkpointDiff", () => { it("includes cacheScope so reused turn counts do not collide", () => { const baseInput = { - sessionId: "session-id", - threadRuntimeId: "thread-id", + threadId: "thread-id", fromTurnCount: 1, toTurnCount: 2, } as const; @@ -28,16 +27,15 @@ describe("providerQueryKeys.checkpointDiff", () => { describe("checkpointDiffQueryOptions", () => { it("forwards checkpoint range to the provider API", async () => { - const getCheckpointDiff = vi.fn().mockResolvedValue({ diff: "patch" }); + const getTurnDiff = vi.fn().mockResolvedValue({ diff: "patch" }); const api = { - providers: { - getCheckpointDiff, + orchestration: { + getTurnDiff, }, } as unknown as NativeApi; const options = checkpointDiffQueryOptions(api, { - sessionId: "session-id", - threadRuntimeId: "thread-id", + threadId: "thread-id", fromTurnCount: 3, toTurnCount: 4, cacheScope: "turn:abc", @@ -46,8 +44,8 @@ describe("checkpointDiffQueryOptions", () => { const queryClient = new QueryClient(); await queryClient.fetchQuery(options); - expect(getCheckpointDiff).toHaveBeenCalledWith({ - sessionId: "session-id", + expect(getTurnDiff).toHaveBeenCalledWith({ + threadId: "thread-id", fromTurnCount: 3, toTurnCount: 4, }); @@ -55,8 +53,7 @@ describe("checkpointDiffQueryOptions", () => { it("retries checkpoint-not-ready errors longer than generic failures", () => { const options = checkpointDiffQueryOptions(undefined, { - sessionId: "session-id", - threadRuntimeId: "thread-id", + threadId: "thread-id", fromTurnCount: 1, toTurnCount: 2, cacheScope: "turn:abc", @@ -86,8 +83,7 @@ describe("checkpointDiffQueryOptions", () => { it("backs off longer for checkpoint-not-ready errors", () => { const options = checkpointDiffQueryOptions(undefined, { - sessionId: "session-id", - threadRuntimeId: "thread-id", + threadId: "thread-id", fromTurnCount: 1, toTurnCount: 2, cacheScope: "turn:abc", diff --git a/apps/web/src/lib/providerReactQuery.ts b/apps/web/src/lib/providerReactQuery.ts index 14529432b819e06f215d29ccc644f18a538ddb3e..bb69cc9e56a4517b6057d2a1a773c5b08df92445 100644 --- a/apps/web/src/lib/providerReactQuery.ts +++ b/apps/web/src/lib/providerReactQuery.ts @@ -1,9 +1,9 @@ import type { NativeApi } from "@t3tools/contracts"; import { queryOptions } from "@tanstack/react-query"; +import { asThreadId } from "./orchestrationIds"; export interface CheckpointDiffQueryInput { - sessionId: string | null; - threadRuntimeId: string | null; + threadId: string | null; fromTurnCount: number | null; toTurnCount: number | null; cacheScope?: string | null; @@ -15,8 +15,7 @@ export const providerQueryKeys = { [ "providers", "checkpointDiff", - input.sessionId, - input.threadRuntimeId, + input.threadId, input.fromTurnCount, input.toTurnCount, input.cacheScope ?? null, @@ -58,20 +57,20 @@ export function checkpointDiffQueryOptions( return queryOptions({ queryKey: providerQueryKeys.checkpointDiff(input), queryFn: async () => { - if (!api || !input.sessionId || !hasValidRange) { + if (!api || !input.threadId || !hasValidRange) { throw new Error("Checkpoint diff is unavailable."); } const { fromTurnCount, toTurnCount } = input; if (typeof fromTurnCount !== "number" || typeof toTurnCount !== "number") { throw new Error("Checkpoint diff range is invalid."); } - return api.providers.getCheckpointDiff({ - sessionId: input.sessionId, + return api.orchestration.getTurnDiff({ + threadId: asThreadId(input.threadId), fromTurnCount, toTurnCount, }); }, - enabled: !!api && !!input.sessionId && hasValidRange, + enabled: !!api && !!input.threadId && hasValidRange, staleTime: Infinity, retry: (failureCount, error) => { if (isCheckpointTemporarilyUnavailable(error)) { diff --git a/apps/web/src/routes/__root.tsx b/apps/web/src/routes/__root.tsx index 041c57fdfecaed6edafca3963b269bd1e7f66d32..eb9d06cbc38d7f8f3d17e72902db049776685c73 100644 --- a/apps/web/src/routes/__root.tsx +++ b/apps/web/src/routes/__root.tsx @@ -13,6 +13,7 @@ import { isElectron } from "../env"; import { useNativeApi } from "../hooks/useNativeApi"; import { DEFAULT_MODEL } from "../model-logic"; import { useStore } from "../store"; +import { newCommandId, newProjectId, newThreadId } from "../lib/orchestrationIds"; import { onServerWelcome } from "../wsNativeApi"; export const Route = createRootRouteWithContext<{ @@ -134,7 +135,7 @@ function EventRouter() { const flushSnapshotSync = async (): Promise => { const snapshot = await api.orchestration.getSnapshot(); if (disposed) return; - latestSequence = Math.max(latestSequence, snapshot.sequence); + latestSequence = Math.max(latestSequence, snapshot.snapshotSequence); dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot }); if (pending) { pending = false; @@ -158,12 +159,7 @@ function EventRouter() { } }; - void Promise.all([api.projects.list(), syncSnapshot()]) - .then(([projects]) => { - if (disposed) return; - dispatch({ type: "SYNC_PROJECTS", projects }); - }) - .catch(() => undefined); + void syncSnapshot().catch(() => undefined); const unsubDomainEvent = api.orchestration.onDomainEvent((event) => { if (event.sequence <= latestSequence) { @@ -209,32 +205,34 @@ function AutoProjectBootstrap() { bootstrappedRef.current = true; const now = new Date().toISOString(); - void api.projects - .add({ cwd: payload.cwd }) - .then(async (result) => { - const projects = await api.projects.list(); - dispatch({ type: "SYNC_PROJECTS", projects }); - - const hasThread = state.threads.some((thread) => thread.projectId === result.project.id); - if (hasThread) { - return; - } - - return api.orchestration.dispatchCommand({ + const projectId = newProjectId(); + const threadId = newThreadId(); + void api.orchestration + .dispatchCommand({ + type: "project.create", + commandId: newCommandId(), + projectId, + title: payload.projectName, + workspaceRoot: payload.cwd, + defaultModel: DEFAULT_MODEL, + createdAt: now, + }) + .then(() => + api.orchestration.dispatchCommand({ type: "thread.create", - commandId: crypto.randomUUID(), - threadId: crypto.randomUUID(), - projectId: result.project.id, + commandId: newCommandId(), + threadId, + projectId, title: "New thread", model: DEFAULT_MODEL, branch: null, worktreePath: null, createdAt: now, - }); - }) + }), + ) .catch(() => undefined); }); - }, [api, state.projects, state.threads, dispatch]); + }, [api, state.projects, dispatch]); return null; } diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index 92787a544483b2e58544209c048cbe7e18384f7b..3f4b646c9e32c4fe268a4b4c75f3e8a1644067af 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -1,4 +1,4 @@ -import type { OrchestrationThreadActivity } from "@t3tools/contracts"; +import { EventId, TurnId, type OrchestrationThreadActivity } from "@t3tools/contracts"; import { describe, expect, it } from "vitest"; import { @@ -7,18 +7,24 @@ import { deriveWorkLogEntries, } from "./session-logic"; -function makeActivity( - overrides: Partial, -): OrchestrationThreadActivity { +function makeActivity(overrides: { + id?: string; + createdAt?: string; + kind?: string; + summary?: string; + tone?: OrchestrationThreadActivity["tone"]; + payload?: Record; + turnId?: string; +}): OrchestrationThreadActivity { + const payload = overrides.payload ?? {}; return { - id: overrides.id ?? crypto.randomUUID(), + id: EventId.makeUnsafe(overrides.id ?? crypto.randomUUID()), createdAt: overrides.createdAt ?? "2026-02-23T00:00:00.000Z", - label: overrides.label ?? "Tool call", + kind: overrides.kind ?? "tool.started", + summary: overrides.summary ?? "Tool call", tone: overrides.tone ?? "tool", - ...(overrides.detail !== undefined ? { detail: overrides.detail } : {}), - ...(overrides.turnId !== undefined ? { turnId: overrides.turnId } : {}), - ...(overrides.requestId !== undefined ? { requestId: overrides.requestId } : {}), - ...(overrides.requestKind !== undefined ? { requestKind: overrides.requestKind } : {}), + payload, + turnId: overrides.turnId ? TurnId.makeUnsafe(overrides.turnId) : null, }; } @@ -28,26 +34,30 @@ describe("derivePendingApprovals", () => { makeActivity({ id: "approval-open", createdAt: "2026-02-23T00:00:01.000Z", - label: "Command approval requested", - detail: "bun run lint", - tone: "tool", - requestId: "req-1", - requestKind: "command", + kind: "approval.requested", + summary: "Command approval requested", + tone: "approval", + payload: { + requestId: "req-1", + requestKind: "command", + detail: "bun run lint", + }, }), makeActivity({ id: "approval-close", createdAt: "2026-02-23T00:00:02.000Z", - label: "Approval resolved", + kind: "approval.resolved", + summary: "Approval resolved", tone: "info", - requestId: "req-2", + payload: { requestId: "req-2" }, }), makeActivity({ id: "approval-closed-request", createdAt: "2026-02-23T00:00:01.500Z", - label: "File-change approval requested", - tone: "tool", - requestId: "req-2", - requestKind: "file-change", + kind: "approval.requested", + summary: "File-change approval requested", + tone: "approval", + payload: { requestId: "req-2", requestKind: "file-change" }, }), ]; @@ -68,12 +78,14 @@ describe("deriveWorkLogEntries", () => { makeActivity({ id: "tool-complete", createdAt: "2026-02-23T00:00:03.000Z", - label: "Tool call complete", + summary: "Tool call complete", + kind: "tool.completed", }), makeActivity({ id: "tool-start", createdAt: "2026-02-23T00:00:02.000Z", - label: "Tool call", + summary: "Tool call", + kind: "tool.started", }), ]; @@ -83,9 +95,14 @@ describe("deriveWorkLogEntries", () => { it("filters by turn id when provided", () => { const activities: OrchestrationThreadActivity[] = [ - makeActivity({ id: "turn-1", turnId: "turn-1", label: "Tool call" }), - makeActivity({ id: "turn-2", turnId: "turn-2", label: "Tool call complete" }), - makeActivity({ id: "no-turn", label: "Checkpoint captured", tone: "info" }), + makeActivity({ id: "turn-1", turnId: "turn-1", summary: "Tool call", kind: "tool.started" }), + makeActivity({ + id: "turn-2", + turnId: "turn-2", + summary: "Tool call complete", + kind: "tool.completed", + }), + makeActivity({ id: "no-turn", summary: "Checkpoint captured", tone: "info" }), ]; const entries = deriveWorkLogEntries(activities, "turn-2"); @@ -97,14 +114,15 @@ describe("deriveWorkLogEntries", () => { makeActivity({ id: "checkpoint", createdAt: "2026-02-23T00:00:01.000Z", - label: "Checkpoint captured", + summary: "Checkpoint captured", tone: "info", }), makeActivity({ id: "tool-complete", createdAt: "2026-02-23T00:00:02.000Z", - label: "Command run complete", + summary: "Command run complete", tone: "tool", + kind: "tool.completed", }), ]; diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 0055f975cfab1642f4cefaad56965618ede0871a..23f405fe92ff4ae9d31e7541dda38b8c4fb589fd 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -1,10 +1,15 @@ import type { OrchestrationThreadActivity, ProviderKind, - ProviderSession, } from "@t3tools/contracts"; -import type { ChatMessage, SessionPhase, TurnDiffFileChange, TurnDiffSummary } from "./types"; +import type { + ChatMessage, + SessionPhase, + ThreadSession, + TurnDiffFileChange, + TurnDiffSummary, +} from "./types"; export const PROVIDER_OPTIONS: Array<{ value: ProviderKind; @@ -83,18 +88,29 @@ export function derivePendingApprovals( ); for (const activity of ordered) { - if (activity.requestId && activity.requestKind) { - openByRequestId.set(activity.requestId, { - requestId: activity.requestId, - requestKind: activity.requestKind, + const payload = + activity.payload && typeof activity.payload === "object" + ? (activity.payload as Record) + : null; + const requestId = payload && typeof payload.requestId === "string" ? payload.requestId : null; + const requestKind = + payload && (payload.requestKind === "command" || payload.requestKind === "file-change") + ? payload.requestKind + : null; + const detail = payload && typeof payload.detail === "string" ? payload.detail : undefined; + + if (activity.kind === "approval.requested" && requestId && requestKind) { + openByRequestId.set(requestId, { + requestId, + requestKind, createdAt: activity.createdAt, - ...(activity.detail ? { detail: activity.detail } : {}), + ...(detail ? { detail } : {}), }); continue; } - if (activity.requestId && activity.label === "Approval resolved") { - openByRequestId.delete(activity.requestId); + if (activity.kind === "approval.resolved" && requestId) { + openByRequestId.delete(requestId); } } @@ -112,16 +128,20 @@ export function deriveWorkLogEntries( ); return ordered .filter((activity) => (latestTurnId ? activity.turnId === latestTurnId : true)) - .filter((activity) => activity.label !== "Checkpoint captured") + .filter((activity) => activity.summary !== "Checkpoint captured") .map((activity) => { + const payload = + activity.payload && typeof activity.payload === "object" + ? (activity.payload as Record) + : null; const entry: WorkLogEntry = { id: activity.id, createdAt: activity.createdAt, - label: activity.label, - tone: activity.tone, + label: activity.summary, + tone: activity.tone === "approval" ? "info" : activity.tone, }; - if (activity.detail) { - entry.detail = activity.detail; + if (payload && typeof payload.detail === "string" && payload.detail.length > 0) { + entry.detail = payload.detail; } return entry; }); @@ -207,7 +227,7 @@ export function deriveTurnDiffSummaries(_events: ReadonlyArray (t.id === threadId ? updater(t) : t)); } -function mapProjects( - incoming: ProjectListResult, +function cloneProjectScripts(scripts: ReadonlyArray): ProjectScript[] { + return scripts.map((script) => ({ ...script })); +} + +function mapProjectsFromReadModel( + incoming: OrchestrationReadModel["projects"], previous: Project[], ): Project[] { return incoming.map((project) => { const existing = previous.find((entry) => entry.id === project.id) ?? - previous.find((entry) => entry.cwd === project.cwd); + previous.find((entry) => entry.cwd === project.workspaceRoot); return { id: project.id, - name: project.name, - cwd: project.cwd, - model: existing?.model ?? DEFAULT_MODEL, + name: project.title, + cwd: project.workspaceRoot, + model: existing?.model ?? resolveModelSlug(project.defaultModel ?? DEFAULT_MODEL), expanded: existing?.expanded ?? (persistedExpandedProjectCwds.size > 0 - ? persistedExpandedProjectCwds.has(project.cwd) + ? persistedExpandedProjectCwds.has(project.workspaceRoot) : true), - scripts: normalizeProjectScripts(project.scripts), + scripts: cloneProjectScripts(project.scripts), }; }); } +function toLegacySessionStatus( + status: OrchestrationSessionStatus, +): "connecting" | "ready" | "running" | "error" | "closed" { + switch (status) { + case "starting": + return "connecting"; + case "running": + return "running"; + case "error": + return "error"; + case "ready": + case "interrupted": + return "ready"; + case "idle": + case "stopped": + return "closed"; + } +} + +function toLegacyProvider(providerName: string | null): "codex" | "claudeCode" { + return providerName === "claudeCode" ? "claudeCode" : "codex"; +} + function normalizeTerminalIds(terminalIds: string[]): string[] { const ids = terminalIds.map((id) => id.trim()).filter((id) => id.length > 0); @@ -354,12 +380,32 @@ function closeThreadTerminal(thread: Thread, terminalId: string): Thread { export function reducer(state: AppState, action: Action): AppState { switch (action.type) { case "SYNC_SERVER_READ_MODEL": { + const projects = mapProjectsFromReadModel( + action.readModel.projects.filter((project) => project.deletedAt === null), + state.projects, + ); const existingThreadById = new Map(state.threads.map((thread) => [thread.id, thread] as const)); - const threads = action.readModel.threads.map((thread) => { + const threads = action.readModel.threads + .filter((thread) => thread.deletedAt === null) + .map((thread) => { const existing = existingThreadById.get(thread.id); + const latestTurnStartedAt = + thread.latestTurnId === null + ? undefined + : thread.messages.find( + (message) => message.turnId === thread.latestTurnId && message.role === "user", + )?.createdAt; + const latestTurnCompletedAt = thread.checkpoints.find( + (checkpoint) => checkpoint.turnId === thread.latestTurnId, + )?.completedAt; + const latestTurnDurationMs = + latestTurnStartedAt && latestTurnCompletedAt + ? Math.max(0, Date.parse(latestTurnCompletedAt) - Date.parse(latestTurnStartedAt)) + : undefined; + return normalizeThreadTerminals({ id: thread.id, - codexThreadId: thread.session?.threadId ?? null, + codexThreadId: thread.session?.providerThreadId ?? null, projectId: thread.projectId, title: thread.title, model: resolveModelSlug(thread.model), @@ -374,52 +420,72 @@ export function reducer(state: AppState, action: Action): AppState { activeTerminalGroupId: existing?.activeTerminalGroupId ?? `group-${DEFAULT_THREAD_TERMINAL_ID}`, session: thread.session ? { - sessionId: thread.session.sessionId, - provider: thread.session.provider, - status: thread.session.status, - threadId: thread.session.threadId, + sessionId: thread.session.providerSessionId ?? `thread:${thread.id}`, + provider: toLegacyProvider(thread.session.providerName), + status: toLegacySessionStatus(thread.session.status), + orchestrationStatus: thread.session.status, + threadId: thread.session.providerThreadId, activeTurnId: thread.session.activeTurnId ?? undefined, - createdAt: thread.session.createdAt, + createdAt: thread.session.updatedAt, updatedAt: thread.session.updatedAt, ...(thread.session.lastError ? { lastError: thread.session.lastError } : {}), } : null, - messages: thread.messages.map((message) => ({ ...message })), - error: thread.error, + messages: thread.messages.map((message) => { + const attachments = message.attachments?.map((attachment, index) => ({ + type: "image" as const, + id: `${message.id}:${index}`, + name: attachment.name, + mimeType: attachment.mimeType, + sizeBytes: attachment.sizeBytes, + previewUrl: attachment.dataUrl, + })); + const normalizedMessage: ChatMessage = { + id: message.id, + role: message.role, + text: message.text, + createdAt: message.createdAt, + streaming: message.streaming, + ...(message.streaming ? {} : { completedAt: message.updatedAt }), + ...(attachments && attachments.length > 0 ? { attachments } : {}), + }; + return normalizedMessage; + }), + error: thread.session?.lastError ?? null, createdAt: thread.createdAt, latestTurnId: thread.latestTurnId ?? undefined, - latestTurnStartedAt: thread.latestTurnStartedAt ?? undefined, - latestTurnCompletedAt: thread.latestTurnCompletedAt ?? undefined, - latestTurnDurationMs: thread.latestTurnDurationMs ?? undefined, + latestTurnStartedAt, + latestTurnCompletedAt, + latestTurnDurationMs, lastVisitedAt: existing?.lastVisitedAt ?? thread.updatedAt, branch: thread.branch, worktreePath: thread.worktreePath, - turnDiffSummaries: thread.turnDiffSummaries.map((summary) => ({ - ...summary, - files: summary.files.map((file) => ({ ...file })), + turnDiffSummaries: thread.checkpoints.map((checkpoint) => ({ + turnId: checkpoint.turnId, + completedAt: checkpoint.completedAt, + status: checkpoint.status, + assistantMessageId: checkpoint.assistantMessageId ?? undefined, + checkpointTurnCount: checkpoint.checkpointTurnCount, + checkpointRef: checkpoint.checkpointRef, + files: checkpoint.files.map((file) => ({ ...file })), })), activities: thread.activities.map((activity) => ({ ...activity })), }); }); return { ...state, + projects, threads, threadsHydrated: true, }; } - case "SYNC_PROJECTS": - return { - ...state, - projects: mapProjects(action.projects, state.projects), - }; - case "SET_PROJECT_SCRIPTS": return { ...state, projects: state.projects.map((project) => project.id === action.projectId - ? { ...project, scripts: normalizeProjectScripts(action.scripts) } + ? { ...project, scripts: cloneProjectScripts(action.scripts) } : project, ), }; diff --git a/apps/web/src/types.ts b/apps/web/src/types.ts index 44c9dd27cb43f71daf5893fc4cc475961bc32b37..11acd3f0ad8b02ae6be265382c71977e71412220 100644 --- a/apps/web/src/types.ts +++ b/apps/web/src/types.ts @@ -1,7 +1,7 @@ import type { + OrchestrationSessionStatus, OrchestrationThreadActivity, ProjectScript as ContractProjectScript, - ProviderSession, } from "@t3tools/contracts"; export type SessionPhase = "disconnected" | "connecting" | "ready" | "running"; @@ -30,7 +30,7 @@ export type ChatAttachment = ChatImageAttachment; export interface ChatMessage { id: string; - role: "user" | "assistant"; + role: "user" | "assistant" | "system"; text: string; attachments?: ChatAttachment[]; createdAt: string; @@ -50,6 +50,7 @@ export interface TurnDiffSummary { completedAt: string; status?: string | undefined; files: TurnDiffFileChange[]; + checkpointRef?: string | undefined; assistantMessageId?: string | undefined; checkpointTurnCount?: number | undefined; } @@ -76,7 +77,7 @@ export interface Thread { activeTerminalId: string; terminalGroups: ThreadTerminalGroup[]; activeTerminalGroupId: string; - session: ProviderSession | null; + session: ThreadSession | null; messages: ChatMessage[]; error: string | null; createdAt: string; @@ -90,3 +91,15 @@ export interface Thread { turnDiffSummaries: TurnDiffSummary[]; activities: OrchestrationThreadActivity[]; } + +export interface ThreadSession { + sessionId: string; + provider: "codex" | "claudeCode"; + status: SessionPhase | "error" | "closed"; + threadId: string | null; + activeTurnId?: string | undefined; + createdAt: string; + updatedAt: string; + lastError?: string; + orchestrationStatus: OrchestrationSessionStatus; +} diff --git a/apps/web/src/vite-env.d.ts b/apps/web/src/vite-env.d.ts index 93076c718b369f48f230c54b31ed5d9ff033f994..a7773ab2727b739db8b90f1709f4a67fb6d82739 100644 --- a/apps/web/src/vite-env.d.ts +++ b/apps/web/src/vite-env.d.ts @@ -1,15 +1,6 @@ /// -import type { NativeApi } from "@t3tools/contracts"; - -interface DesktopBridge { - getWsUrl: () => string | null; - pickFolder: () => Promise; - confirm: (message: string) => Promise; - showContextMenu: (items: readonly { id: string; label: string }[]) => Promise; - openExternal: (url: string) => Promise; - onMenuAction: (listener: (action: string) => void) => () => void; -} +import type { NativeApi, DesktopBridge } from "@t3tools/contracts"; declare global { interface Window { diff --git a/apps/web/src/wsNativeApi.ts b/apps/web/src/wsNativeApi.ts index d40439c888b5b63876f4949bddd1682bc1ccc00a..88a11e1424077be293de3a10361f0709ba77da50 100644 --- a/apps/web/src/wsNativeApi.ts +++ b/apps/web/src/wsNativeApi.ts @@ -56,12 +56,6 @@ export function createWsNativeApi(): NativeApi { }); const api: NativeApi = { - todos: { - list: async () => [], - add: async () => [], - toggle: async () => [], - remove: async () => [], - }, dialogs: { pickFolder: async () => { if (!window.desktopBridge) return null; @@ -84,30 +78,8 @@ export function createWsNativeApi(): NativeApi { onEvent: (callback) => transport.subscribe(WS_CHANNELS.terminalEvent, callback as (data: unknown) => void), }, - agent: { - spawn: async () => "", - kill: async () => {}, - write: async () => {}, - onOutput: () => () => {}, - onExit: () => () => {}, - }, - providers: { - startSession: (input) => transport.request(WS_METHODS.providersStartSession, input), - sendTurn: (input) => transport.request(WS_METHODS.providersSendTurn, input), - interruptTurn: (input) => transport.request(WS_METHODS.providersInterruptTurn, input), - respondToRequest: (input) => transport.request(WS_METHODS.providersRespondToRequest, input), - stopSession: (input) => transport.request(WS_METHODS.providersStopSession, input), - listCheckpoints: (input) => transport.request(WS_METHODS.providersListCheckpoints, input), - getCheckpointDiff: (input) => transport.request(WS_METHODS.providersGetCheckpointDiff, input), - revertToCheckpoint: (input) => - transport.request(WS_METHODS.providersRevertToCheckpoint, input), - }, projects: { - list: () => transport.request(WS_METHODS.projectsList), - add: (input) => transport.request(WS_METHODS.projectsAdd, input), - remove: (input) => transport.request(WS_METHODS.projectsRemove, input), searchEntries: (input) => transport.request(WS_METHODS.projectsSearchEntries, input), - updateScripts: (input) => transport.request(WS_METHODS.projectsUpdateScripts, input), }, shell: { openInEditor: (cwd, editor) => @@ -157,6 +129,7 @@ export function createWsNativeApi(): NativeApi { getSnapshot: () => transport.request(ORCHESTRATION_WS_METHODS.getSnapshot), dispatchCommand: (command) => transport.request(ORCHESTRATION_WS_METHODS.dispatchCommand, command), + getTurnDiff: (input) => transport.request(ORCHESTRATION_WS_METHODS.getTurnDiff, input), replayEvents: (fromSequenceExclusive) => transport.request(ORCHESTRATION_WS_METHODS.replayEvents, { fromSequenceExclusive }), onDomainEvent: (callback) => diff --git a/apps/web/src/wsTransport.ts b/apps/web/src/wsTransport.ts index e1cee484ab91dd2cc8e8309f368dc37fddb5c220..de05dca98fceb72b7949cedd6a7a1ad4d7ecae10 100644 --- a/apps/web/src/wsTransport.ts +++ b/apps/web/src/wsTransport.ts @@ -1,4 +1,4 @@ -import type { WsPush, WsRequest, WsResponse } from "@t3tools/contracts"; +import type { WsPush } from "@t3tools/contracts"; type PushListener = (data: unknown) => void; @@ -11,6 +11,14 @@ interface PendingRequest { const REQUEST_TIMEOUT_MS = 60_000; const RECONNECT_DELAYS_MS = [500, 1_000, 2_000, 4_000, 8_000]; +interface WsRequestEnvelope { + id: string; + body: { + _tag: string; + [key: string]: unknown; + }; +} + export class WsTransport { private ws: WebSocket | null = null; private nextId = 1; @@ -41,7 +49,11 @@ export class WsTransport { throw new Error("Request method is required"); } const id = String(this.nextId++); - const message: WsRequest = { id, method, ...(params !== undefined ? { params } : {}) }; + const body = + params !== undefined + ? { _tag: method, ...(params as Record) } + : { _tag: method }; + const message: WsRequestEnvelope = { id, body }; return new Promise((resolve, reject) => { const timeout = setTimeout(() => { @@ -144,22 +156,26 @@ export class WsTransport { // Response to a request if (typeof message.id === "string") { - const response = message as unknown as WsResponse; - const pending = this.pending.get(response.id); + const responseId = message.id; + const pending = this.pending.get(responseId); if (!pending) return; clearTimeout(pending.timeout); - this.pending.delete(response.id); + this.pending.delete(responseId); - if (response.error) { - pending.reject(new Error(response.error.message)); + if ( + message.error && + typeof message.error === "object" && + typeof (message.error as { message?: unknown }).message === "string" + ) { + pending.reject(new Error((message.error as { message: string }).message)); } else { - pending.resolve(response.result); + pending.resolve(message.result); } } } - private send(message: WsRequest) { + private send(message: WsRequestEnvelope) { if (this.ws?.readyState === WebSocket.OPEN) { this.ws.send(JSON.stringify(message)); return; diff --git a/bun.lock b/bun.lock index 6fc8349814a234c37a0ca23bd6b1b53bbb28cdad..6e316d6e4ee10c659038f01aef0c4c14424fa60b 100644 --- a/bun.lock +++ b/bun.lock @@ -19,6 +19,7 @@ "electron": "40.6.0", }, "devDependencies": { + "@t3tools/contracts": "workspace:*", "@types/node": "^22.10.2", "electronmon": "^2.0.2", "tsdown": "^0.20.3", @@ -68,6 +69,7 @@ "@xterm/addon-fit": "^0.11.0", "@xterm/xterm": "^6.0.0", "class-variance-authority": "^0.7.1", + "effect": "4.0.0-beta.10", "highlight.js": "^11.11.1", "lucide-react": "^0.564.0", "react": "^19.0.0", @@ -76,7 +78,6 @@ "rehype-highlight": "^7.0.2", "remark-gfm": "^4.0.1", "tailwind-merge": "^3.4.0", - "zod": "^3.24.1", }, "devDependencies": { "@tailwindcss/vite": "^4.0.0", @@ -95,7 +96,6 @@ "version": "0.0.0", "dependencies": { "effect": "4.0.0-beta.10", - "zod": "^3.24.1", }, "devDependencies": { "@effect/language-service": "0.75.1", diff --git a/packages/contracts/src/git.ts b/packages/contracts/src/git.ts index 769480c6e9c77d281ebe4544628c5555ad6d5054..503d928a065272fa011ffba95665244344a8beaf 100644 --- a/packages/contracts/src/git.ts +++ b/packages/contracts/src/git.ts @@ -5,6 +5,7 @@ const TrimmedNonEmptyString = Schema.Trimmed.check(Schema.isNonEmpty()); // Domain Types export const GitStackedAction = Schema.Literals(["commit", "commit_push", "commit_push_pr"]); +export type GitStackedAction = typeof GitStackedAction.Type; export const GitCommitStepStatus = Schema.Literals(["created", "skipped_no_changes"]); export const GitPushStepStatus = Schema.Literals([ "pushed", diff --git a/packages/contracts/src/ipc.ts b/packages/contracts/src/ipc.ts index 85661ed8fb2c52745e091dcbd591fa4bb133fd00..154b7adab4feac1806f982974076b1fd5ce90e8e 100644 --- a/packages/contracts/src/ipc.ts +++ b/packages/contracts/src/ipc.ts @@ -35,6 +35,18 @@ import type { } from "./orchestration"; import { EditorId } from "./editor"; +export interface DesktopBridge { + getWsUrl: () => string | null; + pickFolder: () => Promise; + confirm: (message: string) => Promise; + showContextMenu: ( + items: readonly { id: T; label: string }[], + position?: { x: number; y: number }, + ) => Promise; + openExternal: (url: string) => Promise; + onMenuAction: (listener: (action: string) => void) => () => void; +} + export interface NativeApi { dialogs: { pickFolder: () => Promise; diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 531a20e2304cb661686cc92741a99f0388b29c55..b8012890a999fc1585d985abedde414c3da9eab7 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -105,6 +105,7 @@ export const ProjectScriptIcon = Schema.Literals([ "build", "debug", ]); +export type ProjectScriptIcon = typeof ProjectScriptIcon.Type; export const ProjectScript = Schema.Struct({ id: Schema.String, diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index e2d28c759f4d3a3533034e76ae2129a66be87c88..ad491ce11230fa26d49232eb6753f29be6f8ba3f 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -5,17 +5,12 @@ export const ServerConfig = Schema.Struct({ cwd: Schema.NonEmptyString, keybindings: ResolvedKeybindingsConfig, }); -export type ServerConfig = Schema.Codec.Encoded; +export type ServerConfig = typeof ServerConfig.Type; export const ServerUpsertKeybindingInput = KeybindingRule; -export type ServerUpsertKeybindingInput = Schema.Codec.Encoded< - typeof ServerUpsertKeybindingInput ->; +export type ServerUpsertKeybindingInput = typeof ServerUpsertKeybindingInput.Type; export const ServerUpsertKeybindingResult = Schema.Struct({ keybindings: ResolvedKeybindingsConfig, }); -export type ServerUpsertKeybindingResult = Schema.Codec.Encoded< - typeof ServerUpsertKeybindingResult ->; - +export type ServerUpsertKeybindingResult = typeof ServerUpsertKeybindingResult.Type;