From 8c5b64bcbaf5800de93165bbb14c3ff35ae3bd61 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 17:48:16 +0000 Subject: [PATCH 1/2] fix(server): a thread follows Claude's EnterWorktree and ExitWorktree The server never recorded a Claude session moving itself with EnterWorktree or ExitWorktree, so the thread kept its creation-time branch and empty worktreePath. The sidebar, details panel, diff base, PR badge and file previews all kept pointing at the original checkout. The Claude adapter now reports the move on its provider session from a root-agent PostToolUse hook, along with the checkout the turn started in. WorktreeBranchSync applies it to the thread once the thread is idle. It uses the existing guarded thread.metadata.update, ignores a move observed before the thread was re-pointed, and clears worktreePath when the agent returns to the project root. Fixes pingdotgg/t3code#11078 AI-assisted; verified with the tests and tools listed above. --- .../Adapters/ClaudeAdapterV2.ts | 55 ++++++++- .../WorktreeBranchSync.test.ts | 116 ++++++++++++++++++ .../orchestration-v2/WorktreeBranchSync.ts | 74 ++++++++++- .../src/orchestration-v2/runtimeLayer.ts | 4 +- packages/contracts/src/orchestrationV2.ts | 2 + 5 files changed, 245 insertions(+), 6 deletions(-) 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/WorktreeBranchSync.test.ts b/apps/server/src/orchestration-v2/WorktreeBranchSync.test.ts index bb95778860e5e..f0017c604a1d1 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,113 @@ 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, + } + : command.type, + ); + assert.deepEqual( + updates, + row.expected === null + ? [] + : [ + { + ...row.expected, + expectedWorktreePath: row.thread.worktreePath, + expectedBranch: row.thread.branch, + }, + ], + ); + }).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..5247cd15abd4c 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,80 @@ 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, + }) + .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.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..9f7fa53e18152 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( From 4f9d8ceb93b99afe6973961afe344dd78d526060 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 18:05:58 +0000 Subject: [PATCH 2/2] fix(server): a worktree move never detaches a turn that just started Following a Claude worktree move reads the idle thread, then the project, filesystem and Git before it dispatches. A follow-up turn could start on the existing session in that gap. The orchestrator then accepted the worktree change and detached the session, ending the new turn. thread.metadata.update gains expectedIdle. With it set, the orchestrator rejects the update while a run is active, under the same serialization as the other expected guards. WorktreeBranchSync sets it. The update is retried when that run ends. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/orchestration-v2/Orchestrator.ts | 11 +++++++++++ .../src/orchestration-v2/WorktreeBranchSync.test.ts | 2 ++ .../server/src/orchestration-v2/WorktreeBranchSync.ts | 2 ++ apps/server/src/orchestration-v2/runtimeLayer.test.ts | 11 +++++++++++ packages/contracts/src/orchestrationV2.ts | 2 ++ 5 files changed, 28 insertions(+) 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 f0017c604a1d1..aaac658ba3e69 100644 --- a/apps/server/src/orchestration-v2/WorktreeBranchSync.test.ts +++ b/apps/server/src/orchestration-v2/WorktreeBranchSync.test.ts @@ -165,6 +165,7 @@ it.effect.each(sessionMoveCases)("follows a provider session's own worktree move branch: command.branch, expectedWorktreePath: command.expectedWorktreePath, expectedBranch: command.expectedBranch, + expectedIdle: command.expectedIdle, } : command.type, ); @@ -177,6 +178,7 @@ it.effect.each(sessionMoveCases)("follows a provider session's own worktree move ...row.expected, expectedWorktreePath: row.thread.worktreePath, expectedBranch: row.thread.branch, + expectedIdle: true, }, ], ); diff --git a/apps/server/src/orchestration-v2/WorktreeBranchSync.ts b/apps/server/src/orchestration-v2/WorktreeBranchSync.ts index 5247cd15abd4c..017910f61d3ae 100644 --- a/apps/server/src/orchestration-v2/WorktreeBranchSync.ts +++ b/apps/server/src/orchestration-v2/WorktreeBranchSync.ts @@ -108,6 +108,8 @@ const makeFollowSessionMove = Effect.gen(function* () { 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({ 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/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 9f7fa53e18152..7aef43f119de0 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3150,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)),