diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 3e10e09906796..4f4c204278d0c 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1967,6 +1967,7 @@ export function makeClaudeQueryOptions(input: { readonly supportedDialogKinds?: ClaudeQueryOptions["supportedDialogKinds"]; readonly allowDangerouslySkipPermissions?: boolean; readonly appendSystemPromptFileText?: string; + readonly hooks?: ClaudeQueryOptions["hooks"]; // Flag settings, because a user's settings.json env beats the process env in Claude Code. readonly toolSearchOverride?: "true"; }): ClaudeAgentSdkQueryOptions { @@ -2104,9 +2105,16 @@ export function makeClaudeQueryOptions(input: { ...launchAppendSystemPrompt, ].join("\n\n"), }, - ...(deviceInstructions - ? { hooks: { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } } - : {}), + ...(input.hooks === undefined && !deviceInstructions + ? {} + : { + hooks: { + ...input.hooks, + ...(deviceInstructions + ? { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } + : {}), + }, + }), ...(Object.keys(extraArgs).length === 0 ? {} : { extraArgs }), }; const additionalDirectories = [ @@ -5890,6 +5898,38 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return Queue.offer(events, event).pipe(Effect.asVoid); }; + // EnterWorktree/ExitWorktree move the live process without a T3 handoff. + let liveSession = session; + const moveSession = Effect.fn("ClaudeAdapterV2.moveSession")(function* ( + cwd: string, + cwdChangeSource: string | undefined, + ) { + if (cwd === liveSession.cwd && cwdChangeSource === liveSession.cwdChangeSource) return; + const { cwdChangeSource: _previousSource, ...rest } = liveSession; + liveSession = { + ...rest, + cwd, + ...(cwdChangeSource === undefined ? {} : { cwdChangeSource }), + updatedAt: yield* DateTime.now, + }; + yield* emitProviderEvent({ + type: "provider_session.updated", + driver: CLAUDE_PROVIDER, + providerSession: liveSession, + }); + }); + const onWorktreeTool: HookCallback = (hookInput) => + runPromise( + Effect.gen(function* () { + // Subagent worktrees do not move the root session. + if (hookInput.agent_id !== undefined || hookInput.cwd.length === 0) return {}; + const turn = yield* Ref.get(activeTurn); + const source = turn?.input.runtimePolicy.cwd ?? undefined; + if (source !== undefined) yield* moveSession(hookInput.cwd, source); + return {}; + }), + ); + const persistAttachmentBudget = Effect.fnUntraced(function* (nativeThreadId: string) { const providerThread = (yield* Ref.get(lastProviderThreadByNativeThread)).get( nativeThreadId, @@ -12083,6 +12123,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( supportedDialogKinds: ["resume_return"], ...(appendSystemPromptFileText === undefined ? {} : { appendSystemPromptFileText }), ...(toolSearchOverride === undefined ? {} : { toolSearchOverride }), + hooks: { + PostToolUse: [{ matcher: "^(EnterWorktree|ExitWorktree)$", hooks: [onWorktreeTool] }], + }, }); const resolvedTurnCwd = yield* fileSystem .realPath(turnInput.runtimePolicy.cwd ?? ".") @@ -12109,6 +12152,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( : Effect.void, ), ); + // A new process starts at its requested cwd, not where the previous one moved. + yield* moveSession(queryOptions.cwd ?? process.cwd(), undefined); // Mark only successful opens; retries must not resume sessions that failed to create. yield* Ref.update(openedNativeThreads, (current) => { if (current.has(nativeThreadId)) { @@ -13012,7 +13057,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( instanceId: adapterOptions.instanceId, driver: CLAUDE_PROVIDER, providerSessionId: input.providerSessionId, - providerSession: session, + get providerSession() { + return liveSession; + }, getModelContextWindow: (selection) => claudeContextWindow(selection) ?? undefined, events: Stream.fromEffectRepeat(Queue.take(events)), backgroundTasksEndWithProcess: true, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c186c40f9e73a..e869a963f4f05 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3948,6 +3948,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} branch changed before the metadata update could be applied.`, }); } + if (command.type === "thread.metadata.update" && command.expectedIdle === true) { + const { runs } = yield* projectionStore + .getThreadRecords(command.threadId, ["runs"]) + .pipe(mapDispatchError(command)); + if (runs.some(isActiveRun)) + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} started a run before the metadata update could be applied.`, + }); + } if (command.type === "thread.metadata.update" && command.expectedEmpty === true) { const records = yield* projectionStore .getThreadRecords(command.threadId, ["runs"]) diff --git a/apps/server/src/orchestration-v2/WorktreeBranchSync.test.ts b/apps/server/src/orchestration-v2/WorktreeBranchSync.test.ts index bb95778860e5e..aaac658ba3e69 100644 --- a/apps/server/src/orchestration-v2/WorktreeBranchSync.test.ts +++ b/apps/server/src/orchestration-v2/WorktreeBranchSync.test.ts @@ -1,6 +1,9 @@ import { assert, it } from "@effect/vitest"; import { + ProjectId, ThreadId, + type OrchestrationV2DomainEvent, + type OrchestrationV2ProviderSession, type OrchestrationV2ThreadShell, type OrchestrationV2ServerCommand, } from "@t3tools/contracts"; @@ -11,6 +14,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; +import * as ProjectService from "../project/ProjectService.ts"; import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -84,7 +88,9 @@ it.effect.each([false, true])( Layer.mock(GitWorkflowService.GitWorkflowService)({ currentBranch: () => Effect.succeed("union"), }), + Layer.mock(ProjectService.ProjectService)({}), Layer.mock(ThreadManagement.ThreadManagementService)({ + streamDomainEvents: Stream.empty, getShellSnapshot: () => Effect.succeed(snapshot), dispatch: (command) => Effect.gen(function* () { @@ -109,3 +115,115 @@ it.effect.each([false, true])( ); }, ); + +const sessionMoveCases = [ + { + name: "EnterWorktree from the project root", + thread: { worktreePath: null, branch: "main", activeRunId: null }, + session: { cwd: "/repo/.claude/worktrees/w", cwdChangeSource: "/repo" }, + expected: { worktreePath: "/repo/.claude/worktrees/w", branch: "feature" }, + }, + { + name: "ExitWorktree back to the project root", + thread: { worktreePath: "/repo/.claude/worktrees/w", branch: "feature", activeRunId: null }, + session: { cwd: "/repo", cwdChangeSource: "/repo/.claude/worktrees/w" }, + expected: { worktreePath: null, branch: "main" }, + }, + { + name: "a move observed before the thread was re-pointed", + thread: { worktreePath: "/other", branch: "other", activeRunId: null }, + session: { cwd: "/repo/.claude/worktrees/w", cwdChangeSource: "/repo" }, + expected: null, + }, + { + name: "a move while a run is active", + thread: { worktreePath: null, branch: "main", activeRunId: "run-2" }, + session: { cwd: "/repo/.claude/worktrees/w", cwdChangeSource: "/repo" }, + expected: null, + }, +] as const; + +it.effect.each(sessionMoveCases)("follows a provider session's own worktree move: $name", (row) => { + const drained = Deferred.makeUnsafe(); + const commands: OrchestrationV2ServerCommand[] = []; + const threadId = ThreadId.make("thread"); + const session = { + status: "ready", + capabilities: { sessions: { supportsMultipleProviderThreadsPerSession: false } }, + ...row.session, + } as unknown as OrchestrationV2ProviderSession; + const events = [ + { type: "run.updated", threadId, payload: { status: "completed" } }, + ] as unknown as OrchestrationV2DomainEvent[]; + return Effect.gen(function* () { + yield* Layer.build(WorktreeBranchSync.layer); + yield* Deferred.await(drained); + const updates = commands.map((command) => + command.type === "thread.metadata.update" + ? { + worktreePath: command.worktreePath, + branch: command.branch, + expectedWorktreePath: command.expectedWorktreePath, + expectedBranch: command.expectedBranch, + expectedIdle: command.expectedIdle, + } + : command.type, + ); + assert.deepEqual( + updates, + row.expected === null + ? [] + : [ + { + ...row.expected, + expectedWorktreePath: row.thread.worktreePath, + expectedBranch: row.thread.branch, + expectedIdle: true, + }, + ], + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + FileSystem.layerNoop({ realPath: (path) => Effect.succeed(path) }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getThreadShell: () => + Effect.succeed({ + id: threadId, + projectId: ProjectId.make("project"), + deletedAt: null, + ...row.thread, + } as unknown as OrchestrationV2ThreadShell), + getThreadProviderContext: () => + Effect.succeed({ + providerSessions: [ + session, + { ...session, status: "stopped", cwd: "/elsewhere" }, + { ...session, cwdChangeSource: undefined, cwd: "/repo/x" }, + ], + } as never), + }), + Layer.mock(ProjectService.ProjectService)({ + getById: () => Effect.succeedSome({ workspaceRoot: "/repo" } as never), + }), + Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({ + streamBranchChanges: Stream.empty, + }), + Layer.mock(GitWorkflowService.GitWorkflowService)({ + currentBranch: (cwd) => Effect.succeed(cwd === "/repo" ? "main" : "feature"), + }), + Layer.mock(ThreadManagement.ThreadManagementService)({ + streamDomainEvents: Stream.fromIterable(events).pipe( + Stream.ensuring(Deferred.succeed(drained, undefined)), + ), + dispatch: (command) => + Effect.sync(() => { + commands.push(command); + return { sequence: commands.length, storedEvents: [] }; + }), + }), + ), + ), + Effect.scoped, + ); +}); diff --git a/apps/server/src/orchestration-v2/WorktreeBranchSync.ts b/apps/server/src/orchestration-v2/WorktreeBranchSync.ts index d821e192dca9f..017910f61d3ae 100644 --- a/apps/server/src/orchestration-v2/WorktreeBranchSync.ts +++ b/apps/server/src/orchestration-v2/WorktreeBranchSync.ts @@ -1,11 +1,13 @@ -import { CommandId } from "@t3tools/contracts"; +import { CommandId, type OrchestrationV2DomainEvent, type ThreadId } from "@t3tools/contracts"; import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; +import * as ProjectService from "../project/ProjectService.ts"; import { forkParked } from "../serverActivation.ts"; import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; @@ -67,10 +69,82 @@ const make = Effect.gen(function* () { }); }); +// A provider that moved itself (Claude's EnterWorktree/ExitWorktree) reports where it went on +// its session. Adopting that location detaches the session, so wait until the thread is idle. +const makeFollowSessionMove = Effect.gen(function* () { + const threads = yield* ThreadManagement.ThreadManagementService; + const git = yield* GitWorkflowService.GitWorkflowService; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const projects = yield* ProjectService.ProjectService; + const fs = yield* FileSystem.FileSystem; + return Effect.fn("WorktreeBranchSync.followSessionMove")(function* (threadId: ThreadId) { + const thread = yield* projections.getThreadShell(threadId); + if (thread === null || thread.deletedAt !== null || thread.activeRunId !== null) return; + const { providerSessions } = yield* projections.getThreadProviderContext(threadId); + const moved = providerSessions.filter( + (session) => + session.cwdChangeSource !== undefined && + session.status !== "stopped" && + session.status !== "error" && + !session.capabilities.sessions.supportsMultipleProviderThreadsPerSession, + ); + const session = moved.length === 1 ? moved[0] : undefined; + if (session === undefined) return; + const project = yield* projects.getById(thread.projectId); + if (Option.isNone(project)) return; + const { workspaceRoot } = project.value; + // A move observed from another checkout is stale once the thread was re-pointed. + if (session.cwdChangeSource !== (thread.worktreePath ?? workspaceRoot)) return; + const resolve = (path: string) => fs.realPath(path).pipe(Effect.orElseSucceed(() => path)); + const location = yield* resolve(session.cwd); + if (location === (yield* resolve(session.cwdChangeSource))) return; + const worktreePath = location === (yield* resolve(workspaceRoot)) ? null : session.cwd; + yield* threads + .dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(yield* randomUuidV4), + threadId, + worktreePath, + branch: yield* git.currentBranch(location), + expectedBranch: thread.branch, + expectedWorktreePath: thread.worktreePath, + // The reads above race a follow-up turn, which this update's detach would end. + expectedIdle: true, + }) + .pipe( + Effect.catchTags({ + OrchestratorDispatchError: (cause) => + Effect.logWarning("Worktree move update was rejected", { threadId, cause }), + }), + ); + }); +}); + +const isSessionMoveTrigger = (event: OrchestrationV2DomainEvent) => + (event.type === "provider-session.updated" && event.payload.cwdChangeSource !== undefined) || + (event.type === "run.updated" && + ["completed", "interrupted", "failed", "cancelled"].includes(event.payload.status)); + export const layer = Layer.effectDiscard( Effect.gen(function* () { const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const threads = yield* ThreadManagement.ThreadManagementService; const reconcile = yield* make; + const followSessionMove = yield* makeFollowSessionMove; + yield* forkParked( + Stream.runForEach(threads.streamDomainEvents, (event) => + isSessionMoveTrigger(event) + ? followSessionMove(event.threadId).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to follow a provider worktree move", { + threadId: event.threadId, + cause, + }), + ), + ) + : Effect.void, + ), + ); yield* forkParked( Stream.runForEach(broadcaster.streamBranchChanges, (input) => reconcile(input).pipe( diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 1557e411af93f..641e6abb6e5eb 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -5894,6 +5894,16 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }) .pipe(Effect.flip); assert.instanceOf(nonEmptyClaim, Orchestrator.OrchestratorDispatchError); + const busyMove = yield* orchestrator + .dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make("runtime-layer-active-settle-idle-move"), + threadId, + worktreePath: "/tmp/moved-while-running", + expectedIdle: true, + }) + .pipe(Effect.flip); + assert.instanceOf(busyMove, Orchestrator.OrchestratorDispatchError); const error = yield* orchestrator .dispatch({ @@ -5907,6 +5917,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { assert.equal(error.cause, `Thread ${threadId} is still running. Stop it before settling.`); const projection = yield* orchestrator.getThreadProjection(threadId); assert.equal(projection.runs[0]?.status, "starting"); + assert.equal(projection.thread.worktreePath, "/tmp/runtime-layer-active-settle"); assert.isNull(projection.thread.settledOverride); assert.isNull(projection.thread.settledAt); assert.isNotNull(projection.thread.unsettledAt); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index d83f0603d7f7c..3bc927936dcdc 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -385,7 +385,9 @@ export const layer = Layer.mergeAll( export const layerProduction = Layer.mergeAll( WorktreeBranchSync.layer.pipe( - Layer.provide(Layer.merge(layerThreadManagementProvided, ProjectionStore.layer)), + Layer.provide( + Layer.mergeAll(layerThreadManagementProvided, ProjectionStore.layer, layerProjectService), + ), ), layer.pipe(Layer.provide(layerProjectService)), layerProjectService, diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 1fb918956933f..7aef43f119de0 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -961,6 +961,8 @@ export const OrchestrationV2ProviderSession = Schema.Struct({ providerInstanceId: ProviderInstanceId, status: Schema.Literals(["starting", "ready", "running", "waiting", "stopped", "error"]), cwd: TrimmedNonEmptyString, + // The thread checkout a native move started from; set only after the provider moved itself. + cwdChangeSource: Schema.optional(TrimmedNonEmptyString), model: Schema.NullOr(TrimmedNonEmptyString), capabilities: OrchestrationV2ProviderCapabilities, extensionStatuses: Schema.optional( @@ -3148,6 +3150,8 @@ export const OrchestrationV2Command = Schema.Union([ expectedBranch: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), /** Reject unless no message or run has landed on this thread. */ expectedEmpty: Schema.optional(Schema.Boolean), + /** Reject while a run is active, so a worktree change cannot detach its session. */ + expectedIdle: Schema.optional(Schema.Boolean), limitRecovery: Schema.optional(Schema.NullOr(OrchestrationV2LimitRecoveryUpdate)), /** Link (object) or unlink (null) a pull request; absent leaves it unchanged. */ linkedPullRequest: Schema.optional(Schema.NullOr(ThreadLinkedPullRequest)),