diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index da8fc58c39ea43699a88aee8424a808036d10e17..caa4d1f8d4a6e6da8463c9ceca149b7186fcaa54 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -220,6 +220,8 @@ describe("CheckpointReactor", () => { async function createHarness(options?: { readonly hasSession?: boolean; readonly seedFilesystemCheckpoints?: boolean; + readonly projectWorkspaceRoot?: string; + readonly threadWorktreePath?: string | null; }) { const cwd = createGitRepository(); tempDirs.push(cwd); @@ -252,7 +254,7 @@ describe("CheckpointReactor", () => { commandId: asCommandId("cmd-project-create"), projectId: asProjectId("project-1"), title: "Test Project", - workspaceRoot: "/tmp/test-project", + workspaceRoot: options?.projectWorkspaceRoot ?? cwd, defaultModel: "gpt-5-codex", createdAt, }), @@ -266,7 +268,7 @@ describe("CheckpointReactor", () => { title: "Thread", model: "gpt-5-codex", branch: null, - worktreePath: cwd, + worktreePath: options?.threadWorktreePath ?? cwd, createdAt, }), ); @@ -363,10 +365,11 @@ describe("CheckpointReactor", () => { ).toBe("v2\n"); }); - it("captures pre-turn baseline from thread.turn.start domain flow before runtime turn events", async () => { + it("captures pre-turn baseline from project workspace root when thread worktree is unset", async () => { const harness = await createHarness({ hasSession: false, seedFilesystemCheckpoints: false, + threadWorktreePath: null, }); await Effect.runPromise( @@ -390,6 +393,52 @@ describe("CheckpointReactor", () => { ).toBe("v1\n"); }); + it("captures turn completion checkpoint from project workspace root when provider session cwd is unavailable", async () => { + const harness = await createHarness({ + hasSession: false, + seedFilesystemCheckpoints: false, + threadWorktreePath: null, + }); + const createdAt = new Date().toISOString(); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: asCommandId("cmd-session-set-missing-provider-cwd"), + threadId: asThreadId("thread-1"), + session: { + threadId: asThreadId("thread-1"), + status: "running", + providerName: "codex", + providerSessionId: asSessionId("sess-missing"), + providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-missing"), + activeTurnId: asTurnId("turn-missing-cwd"), + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }), + ); + + fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8"); + harness.provider.emit({ + type: "turn.completed", + eventId: EventId.makeUnsafe("evt-turn-completed-missing-provider-cwd"), + provider: "codex", + sessionId: asSessionId("sess-missing"), + createdAt: new Date().toISOString(), + threadId: ProviderThreadId.makeUnsafe("provider-thread-missing"), + turnId: asTurnId("turn-missing-cwd"), + status: "completed", + }); + + await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); + expect(gitRefExists(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 1))).toBe(true); + expect( + gitShowFileAtRef(harness.cwd, checkpointRefForTurn(asThreadId("thread-1"), 1), "README.md"), + ).toBe("v2\n"); + }); + it("maps runtime checkpoint.captured into thread.turn.diff.complete when missing in read model", async () => { const harness = await createHarness(); const createdAt = new Date().toISOString(); diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 974670dd3ec64056223f535919636aa2c4b75b3e..eb8b5d71342e8e55d0935dc20688ff4a32c7a7be 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -46,6 +46,27 @@ function trimToNonEmpty(value: string | null | undefined): string | undefined { return normalized && normalized.length > 0 ? normalized : undefined; } +function resolveThreadWorkspaceCwd(input: { + readonly thread: { + readonly projectId: string; + readonly worktreePath: string | null; + }; + readonly projects: ReadonlyArray<{ + readonly id: string; + readonly workspaceRoot: string; + }>; +}): string | undefined { + const worktreeCwd = trimToNonEmpty(input.thread.worktreePath); + if (worktreeCwd) { + return worktreeCwd; + } + + const workspaceRoot = input.projects.find( + (project) => project.id === input.thread.projectId, + )?.workspaceRoot; + return trimToNonEmpty(workspaceRoot); +} + function checkpointStatusFromRuntime(status: string | undefined): "ready" | "missing" | "error" { switch (status) { case "failed": @@ -133,6 +154,7 @@ const make = Effect.gen(function* () { const readModel = yield* orchestrationEngine.getReadModel(); const thread = readModel.threads.find((entry) => entry.id === threadId); const projectedSessionId = thread?.session?.providerSessionId ?? null; + const projectedProviderThreadId = trimToNonEmpty(thread?.session?.providerThreadId); const sessions = yield* providerService.listSessions(); @@ -154,8 +176,17 @@ const make = Effect.gen(function* () { } } - const matchedSession = sessions.find((session) => session.threadId?.trim() === threadId.trim()); - return findSessionWithCwd(matchedSession); + if (projectedProviderThreadId) { + const matchedSession = sessions.find( + (session) => session.threadId?.trim() === projectedProviderThreadId, + ); + const fromProviderThread = findSessionWithCwd(matchedSession); + if (Option.isSome(fromProviderThread)) { + return fromProviderThread; + } + } + + return Option.none(); }); const captureCheckpointFromTurnCompletion = Effect.fnUntraced(function* ( @@ -179,7 +210,16 @@ const make = Effect.gen(function* () { } const sessionRuntime = yield* resolveSessionRuntimeForThread(thread.id); - if (Option.isNone(sessionRuntime)) { + const checkpointCwd = + Option.match(sessionRuntime, { + onNone: () => undefined, + onSome: (runtime) => runtime.cwd, + }) ?? + resolveThreadWorkspaceCwd({ + thread, + projects: readModel.projects, + }); + if (!checkpointCwd) { yield* Effect.logWarning("checkpoint capture skipped: no active provider session cwd", { threadId: thread.id, turnId, @@ -197,7 +237,7 @@ const make = Effect.gen(function* () { const targetCheckpointRef = checkpointRefForThreadTurn(thread.id, nextTurnCount); const fromCheckpointExists = yield* checkpointStore.hasCheckpointRef({ - cwd: sessionRuntime.value.cwd, + cwd: checkpointCwd, checkpointRef: fromCheckpointRef, }); if (!fromCheckpointExists) { @@ -209,13 +249,13 @@ const make = Effect.gen(function* () { } yield* checkpointStore.captureCheckpoint({ - cwd: sessionRuntime.value.cwd, + cwd: checkpointCwd, checkpointRef: targetCheckpointRef, }); const files = yield* checkpointStore .diffCheckpoints({ - cwd: sessionRuntime.value.cwd, + cwd: checkpointCwd, fromCheckpointRef, toCheckpointRef: targetCheckpointRef, fallbackFromToHead: false, @@ -296,9 +336,12 @@ const make = Effect.gen(function* () { return; } - const checkpointCwdFromThread = trimToNonEmpty(thread.worktreePath); + const checkpointCwdFromThreadOrProject = resolveThreadWorkspaceCwd({ + thread, + projects: readModel.projects, + }); const checkpointCwd = - checkpointCwdFromThread ?? + checkpointCwdFromThreadOrProject ?? Option.match(yield* resolveSessionRuntimeForThread(thread.id), { onNone: () => undefined, onSome: (runtime) => runtime.cwd, @@ -349,9 +392,12 @@ const make = Effect.gen(function* () { return; } - const checkpointCwdFromThread = trimToNonEmpty(thread.worktreePath); + const checkpointCwdFromThreadOrProject = resolveThreadWorkspaceCwd({ + thread, + projects: readModel.projects, + }); const checkpointCwd = - checkpointCwdFromThread ?? + checkpointCwdFromThreadOrProject ?? Option.match(yield* resolveSessionRuntimeForThread(threadId), { onNone: () => undefined, onSome: (runtime) => runtime.cwd, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 95265cfe5a575dc6b667a31d0940f78db285be94..d8c58df8531d3b40ef1f371a816e1fcd89d70488 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -181,6 +181,10 @@ describe("ProviderCommandReactor", () => { await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); + expect(harness.startSession.mock.calls[0]?.[0]).toMatchObject({ + cwd: "/tmp/provider-project", + model: "gpt-5-codex", + }); const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === asThreadId("thread-1")); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 0dbaa514e53e35d2be4554e1dbd42b83eb1f9e50..ae10b413798e2ff03a49b5c87369f0edbad269c3 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -5,7 +5,6 @@ import { type OrchestrationEvent, type ProviderKind, type OrchestrationSession, - type ProviderSessionId, type ThreadId, type TurnId, } from "@t3tools/contracts"; @@ -34,6 +33,27 @@ function trimToNonEmptyOrUndefined(value: string | undefined): string | undefine return normalized && normalized.length > 0 ? normalized : undefined; } +function resolveEffectiveThreadCwd(input: { + readonly thread: { + readonly projectId: string; + readonly worktreePath: string | null; + }; + readonly projects: ReadonlyArray<{ + readonly id: string; + readonly workspaceRoot: string; + }>; +}): string | undefined { + const worktreeCwd = trimToNonEmptyOrUndefined(input.thread.worktreePath ?? undefined); + if (worktreeCwd) { + return worktreeCwd; + } + + const workspaceRoot = input.projects.find( + (project) => project.id === input.thread.projectId, + )?.workspaceRoot; + return trimToNonEmptyOrUndefined(workspaceRoot); +} + function mapProviderSessionStatusToOrchestrationStatus( status: "connecting" | "ready" | "running" | "error" | "closed", ): OrchestrationSession["status"] { @@ -115,7 +135,8 @@ const make = Effect.gen(function* () { threadId: ThreadId, createdAt: string, ) { - const thread = yield* resolveThread(threadId); + const readModel = yield* orchestrationEngine.getReadModel(); + const thread = readModel.threads.find((entry) => entry.id === threadId); if (!thread) { return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`)); } @@ -129,12 +150,14 @@ const make = Effect.gen(function* () { thread.session?.providerName === "codex" || thread.session?.providerName === "claudeCode" ? thread.session.providerName : undefined; + const effectiveCwd = resolveEffectiveThreadCwd({ + thread, + projects: readModel.projects, + }); const startedSession = yield* providerService.startSession({ ...(preferredProvider ? { provider: preferredProvider } : {}), - ...(trimToNonEmptyOrUndefined(thread.worktreePath ?? undefined) - ? { cwd: trimToNonEmptyOrUndefined(thread.worktreePath ?? undefined) } - : {}), + ...(effectiveCwd ? { cwd: effectiveCwd } : {}), ...(trimToNonEmptyOrUndefined(thread.model) ? { model: thread.model } : {}), });