From 2a789b776c245e605b9be077a52897622e5555d5 Mon Sep 17 00:00:00 2001 From: giri256 <80974392+giri256@users.noreply.github.com> Date: Fri, 6 Nov 2026 20:00:33 +0000 Subject: [PATCH 1/6] feat(server): agents can move their thread to another folder A thread could not change folders after it started, so work that belonged in another folder meant a new thread and a hand-written summary. t3_thread_move takes a folder, uses the project rooted there or registers one, and moves the calling thread when its turn ends. A server-only thread.project.move command changes the project and detaches the idle provider session, so the next turn resumes the same conversation in the new folder. Waits and reads that race with the move follow the thread to its new project. Model: Claude Opus 5.5, harness: Claude Code in T3 Code. --- apps/server/src/mcp/OrchestratorMcpService.ts | 69 +++---- .../server/src/mcp/WorktreeMcpService.test.ts | 122 +++++++++++- apps/server/src/mcp/WorktreeMcpService.ts | 128 +++++++++++- apps/server/src/mcp/threadAccess.test.ts | 34 +++- apps/server/src/mcp/threadAccess.ts | 58 ++++-- .../src/mcp/toolkits/orchestrator/tools.ts | 2 +- .../src/mcp/toolkits/worktree/handlers.ts | 7 + .../server/src/mcp/toolkits/worktree/tools.ts | 18 ++ .../src/orchestration-v2/Orchestrator.ts | 52 ++++- .../ThreadManagementService.test.ts | 188 ++++++++++++++++++ .../ThreadManagementService.ts | 152 +++++++++++--- .../src/orchestration-v2/runtimeLayer.test.ts | 71 +++++++ .../src/provider/agentContextBudget.json | 2 +- .../src/relay/AgentAwarenessRelay.test.ts | 1 + docs/user/thread-sidebar.md | 6 + packages/client-runtime/src/t3ToolSummary.ts | 3 + .../src/work-log/presentation.ts | 1 + packages/contracts/src/orchestrationV2.ts | 7 + packages/contracts/src/worktreeMcp.ts | 24 ++- packages/shared/src/t3McpToolPresentation.ts | 2 + 20 files changed, 858 insertions(+), 89 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 9d15408483a3e..fb95d47aaa24a 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -124,7 +124,7 @@ import { runtimeModeRank, taskPrompt, } from "./delegatedTaskModes.ts"; -import { isLiveCallerShell, threadNotFound } from "./threadAccess.ts"; +import { isLiveCallerShell, readInCurrentProject, threadNotFound } from "./threadAccess.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as PeerForwarding from "../peer/PeerForwarding.ts"; import * as RemoteDelegation from "../peer/RemoteDelegation.ts"; @@ -1120,22 +1120,20 @@ const make = Effect.gen(function* () { projectId: OrchestrationV2ThreadProjection["thread"]["projectId"], threadId: ThreadId, ) => - threadManagement - .getProjectThreadRecords( - { projectId, threadId }, - [ - "runs", - "messages", - "subagents", - "providerThreads", - "providerTurns", - "runtimeRequests", - "contextTransfers", - "turnItems", - ], - { turnItemTypes: [], messageRoles: ["user"] }, - ) - .pipe(Effect.mapError(threadManagementFailure)); + threadManagement.getProjectThreadRecords( + { projectId, threadId }, + [ + "runs", + "messages", + "subagents", + "providerThreads", + "providerTurns", + "runtimeRequests", + "contextTransfers", + "turnItems", + ], + { turnItemTypes: [], messageRoles: ["user"] }, + ); const loadCaller = Effect.fn("OrchestratorMcpService.loadCaller")(function* ( scope: McpInvocationScope, @@ -1194,11 +1192,11 @@ const make = Effect.gen(function* () { return yield* threadNotFound(threadManagement, threadId, `Thread ${threadId} was not found.`); } // Native wait owns its history reads; keep legacy import and project validation here. - return yield* includeHistory - ? loadProjectThread(shell.projectId, threadId) - : threadManagement - .getProjectThreadRecords({ projectId: shell.projectId, threadId }, []) - .pipe(Effect.mapError(threadManagementFailure)); + return yield* readInCurrentProject(threadManagement, threadId, shell.projectId, (projectId) => + includeHistory + ? loadProjectThread(projectId, threadId) + : threadManagement.getProjectThreadRecords({ projectId, threadId }, []), + ).pipe(Effect.mapError(threadManagementFailure)); }); const loadScopedThread = Effect.fn("OrchestratorMcpService.loadScopedThread")(function* ( @@ -1275,19 +1273,22 @@ const make = Effect.gen(function* () { `Thread ${threadId} is no longer available.`, ); } - const input = { projectId: shell.projectId, threadId }; // Transfers are needed only to acknowledge a caller's child-task result. - const records = - parent === undefined - ? threadManagement - .getProjectThreadRecords(input, ["runs", "runtimeRequests"]) - .pipe(Effect.map((target) => ({ ...target, contextTransfers: [] }))) - : threadManagement.getProjectThreadRecords(input, [ - "runs", - "runtimeRequests", - "contextTransfers", - ]); - const target = yield* records.pipe(Effect.mapError(threadManagementFailure)); + const target = yield* readInCurrentProject( + threadManagement, + threadId, + shell.projectId, + (projectId) => + parent === undefined + ? threadManagement + .getProjectThreadRecords({ projectId, threadId }, ["runs", "runtimeRequests"]) + .pipe(Effect.map((target) => ({ ...target, contextTransfers: [] }))) + : threadManagement.getProjectThreadRecords({ projectId, threadId }, [ + "runs", + "runtimeRequests", + "contextTransfers", + ]), + ).pipe(Effect.mapError(threadManagementFailure)); return { parent, target, shell } as const; }); diff --git a/apps/server/src/mcp/WorktreeMcpService.test.ts b/apps/server/src/mcp/WorktreeMcpService.test.ts index 90cf3c7f8401a..d14b2656fbca7 100644 --- a/apps/server/src/mcp/WorktreeMcpService.test.ts +++ b/apps/server/src/mcp/WorktreeMcpService.test.ts @@ -19,6 +19,7 @@ import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Path from "effect/Path"; import * as PlatformError from "effect/PlatformError"; import * as Schema from "effect/Schema"; @@ -117,6 +118,8 @@ interface HarnessOptions { readonly deleteLocalBranchFails?: boolean; readonly createWorktreeGate?: Effect.Effect; readonly existsFails?: boolean; + readonly subagentStatus?: string; + readonly destinationProjectExists?: boolean; } const makeHarness = (options: HarnessOptions = {}) => { @@ -176,7 +179,11 @@ const makeHarness = (options: HarnessOptions = {}) => { return Effect.succeed(makeProjection({ ...thread, archivedAt: "2026-01-02T00:00:00.000Z" })); } return id === threadId && thread !== null - ? Effect.succeed(makeProjection(thread)) + ? Effect.succeed({ + ...makeProjection(thread), + subagents: + options.subagentStatus === undefined ? [] : [{ status: options.subagentStatus }], + } as unknown as OrchestrationV2ThreadProjection) : Effect.fail(new OrchestratorProjectionError({ threadId: id })); }); const sendToThread = vi.fn((_: unknown) => { @@ -204,6 +211,17 @@ const makeHarness = (options: HarnessOptions = {}) => { : Option.none(), ), ); + const bootstrap = vi.fn((input: { readonly projectId: ProjectId; readonly title: string }) => + Effect.succeed({ + project: { + ...project, + id: options.destinationProjectExists ? ProjectId.make("project-existing") : input.projectId, + title: input.title, + }, + created: options.destinationProjectExists !== true, + }), + ); + const moveThreadToProject = vi.fn((_: unknown) => Effect.succeed({ movesWhenTurnEnds: true })); const removeWorktree = vi.fn((_: unknown) => options.removeWorktreeFails ? (Effect.fail("simulated worktree removal failure") as never) @@ -299,9 +317,11 @@ const makeHarness = (options: HarnessOptions = {}) => { dispatch, getThreadRecords, sendToThread, + moveThreadToProject: moveThreadToProject as never, } satisfies Partial), Layer.mock(ProjectService.ProjectService)({ getById, + bootstrap: bootstrap as never, } satisfies Partial), ServerSettings.layerTest({ newWorktreesStartFromOrigin: options.newWorktreesStartFromOrigin ?? false, @@ -362,6 +382,8 @@ const makeHarness = (options: HarnessOptions = {}) => { deleteLocalBranch, localStatus, runForThread, + bootstrap, + moveThreadToProject, }; }; @@ -397,6 +419,104 @@ const runStatus = (harness: ReturnType) => const decodeHandoffInput = Schema.decodeUnknownEffect(WorktreeMcpHandoffInput); +describe("t3_thread_move", () => { + const moveCapabilities = new Set(["worktree", "orchestration"] as const); + const makeFolder = Effect.fn("makeFolder")(function* () { + const fileSystem = yield* FileSystem.FileSystem; + return yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-thread-move-" }); + }); + const runMove = ( + harness: ReturnType, + input: Parameters[1], + ) => + WorktreeMcpService.WorktreeMcpService.pipe( + Effect.flatMap((service) => service.moveToFolder(harness.scope, input)), + Effect.provide(harness.layer), + ); + + it.effect("registers the folder as a project and defers the move and its continuation", () => + Effect.scoped( + Effect.gen(function* () { + const folder = yield* makeFolder(); + const harness = makeHarness({ capabilities: moveCapabilities }); + const result = yield* runMove(harness, { path: folder, continuationPrompt: "Carry on." }); + + expect(result).toMatchObject({ + projectCreated: true, + workspaceRoot: folder, + movesWhenTurnEnds: true, + }); + expect(harness.moveThreadToProject).toHaveBeenCalledWith( + expect.objectContaining({ + threadId, + projectId: result.projectId, + continuation: expect.objectContaining({ text: "Carry on." }), + }), + ); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("reuses the project already rooted at the folder", () => + Effect.scoped( + Effect.gen(function* () { + const folder = yield* makeFolder(); + const harness = makeHarness({ + capabilities: moveCapabilities, + destinationProjectExists: true, + }); + const result = yield* runMove(harness, { path: folder }); + + expect(result.projectCreated).toBe(false); + expect(result.projectId).toBe(ProjectId.make("project-existing")); + expect(harness.moveThreadToProject).toHaveBeenCalledWith( + expect.not.objectContaining({ continuation: expect.anything() }), + ); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("refuses a move it cannot honor before changing anything", () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const folder = yield* makeFolder(); + const file = path.join(folder, "file.txt"); + yield* fileSystem.writeFileString(file, ""); + const cases = [ + { input: "relative/folder", code: "invalid_request" }, + { input: path.join(folder, "missing"), code: "invalid_request" }, + { input: file, code: "invalid_request" }, + { + input: folder, + code: "invalid_request", + options: { thread: { worktreePath: folder } }, + }, + { input: folder, code: "invalid_request", options: { subagentStatus: "running" } }, + { input: folder, code: "invalid_request", options: { thread: { archivedAt: "now" } } }, + { + input: folder, + code: "capability_denied", + options: { capabilities: new Set(["worktree"] as const) }, + }, + ] satisfies ReadonlyArray<{ + readonly input: string; + readonly code: string; + readonly options?: HarnessOptions; + }>; + for (const { input, code, options } of cases) { + const harness = makeHarness({ capabilities: moveCapabilities, ...options }); + const exit = yield* Effect.exit(runMove(harness, { path: input })); + expectTypedFailure(exit, { code }); + expect(harness.bootstrap).not.toHaveBeenCalled(); + expect(harness.moveThreadToProject).not.toHaveBeenCalled(); + } + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); +}); + describe("t3_worktree_handoff", () => { it.effect("reports a continuation failure without failing the handoff", () => { const harness = makeHarness({ continuation: "fails" }); diff --git a/apps/server/src/mcp/WorktreeMcpService.ts b/apps/server/src/mcp/WorktreeMcpService.ts index 0ee45bb7f8bb3..dcade742af439 100644 --- a/apps/server/src/mcp/WorktreeMcpService.ts +++ b/apps/server/src/mcp/WorktreeMcpService.ts @@ -2,7 +2,9 @@ import { CommandId, type GitCommandError, MessageId, - type ProjectId, + ProjectId, + type ThreadMoveMcpInput, + type ThreadMoveMcpResult, WorktreeMcpFailure, type WorktreeMcpContinuationStatus, type WorktreeMcpHandoffInput, @@ -20,6 +22,7 @@ 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 Path from "effect/Path"; import type * as PlatformError from "effect/PlatformError"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; @@ -37,6 +40,11 @@ export class WorktreeMcpService extends Context.Service< scope: McpInvocationScope, input: WorktreeMcpHandoffInput, ) => Effect.Effect; + /** Moves the calling thread to the project rooted at a folder, registering one if needed. */ + readonly moveToFolder: ( + scope: McpInvocationScope, + input: ThreadMoveMcpInput, + ) => Effect.Effect; readonly status: ( scope: McpInvocationScope, ) => Effect.Effect; @@ -61,6 +69,9 @@ function errorMessage(error: unknown): string { return String(error); } +const isUnfinishedSubagent = ({ status }: { readonly status: string }) => + status === "pending" || status === "running" || status === "waiting"; + const asOperationFailed = (prefix: string) => Effect.mapError((error: unknown) => failure("operation_failed", `${prefix}: ${errorMessage(error)}`), @@ -69,6 +80,7 @@ const asOperationFailed = (prefix: string) => const make = Effect.gen(function* () { const random = yield* SecureRandom.SecureRandom; const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const threadManagement = yield* ThreadManagementService.ThreadManagementService; const projects = yield* ProjectService.ProjectService; const serverSettings = yield* ServerSettings.ServerSettingsService; @@ -426,6 +438,117 @@ const make = Effect.gen(function* () { }, ); + const moveToFolder: WorktreeMcpService["Service"]["moveToFolder"] = Effect.fn( + "WorktreeMcpService.moveToFolder", + )(function* (callerScope, input) { + yield* requireCapability(callerScope); + // Registering the destination project is an orchestration write. + if (!callerScope.capabilities.has("orchestration")) { + return yield* failure( + "capability_denied", + "This MCP credential cannot register the destination project.", + ); + } + const scope = yield* requireThreadScope(callerScope); + if (scope.thread.subagent) { + return yield* failure("invalid_request", "A subagent stays with its parent thread."); + } + if (!path.isAbsolute(input.path)) { + return yield* failure("invalid_request", `'${input.path}' is not an absolute path.`); + } + const workspaceRoot = path.resolve(input.path); + const destination = yield* fileSystem.realPath(workspaceRoot).pipe( + Effect.flatMap((resolved) => + fileSystem.stat(resolved).pipe(Effect.map((info) => ({ resolved, info }))), + ), + Effect.mapError(() => + failure("invalid_request", `Folder '${workspaceRoot}' does not exist.`), + ), + ); + if (destination.info.type !== "Directory") { + return yield* failure("invalid_request", `'${workspaceRoot}' is not a folder.`); + } + + const records = yield* threadManagement + .getThreadRecords(scope.thread.threadId, ["subagents"]) + .pipe(asOperationFailed(`Unable to read thread ${scope.thread.threadId}`)); + const { thread } = records; + if (thread.deletedAt !== null) { + return yield* failure("thread_not_found", `Thread '${scope.thread.threadId}' was not found.`); + } + if (thread.archivedAt !== null) { + return yield* failure("invalid_request", `Thread '${thread.id}' is archived.`); + } + if (records.subagents.some(isUnfinishedSubagent)) { + return yield* failure( + "invalid_request", + "This thread still has running subagents. Wait for them to finish, then move it.", + ); + } + const current = yield* loadProject(scope, thread.projectId); + const currentFolder = yield* fileSystem + .realPath(thread.worktreePath ?? current.workspaceRoot) + .pipe(Effect.orElseSucceed(() => thread.worktreePath ?? current.workspaceRoot)); + if (currentFolder === destination.resolved) { + return yield* failure("invalid_request", `This thread already works in '${workspaceRoot}'.`); + } + + const uuid = yield* random.uuidV4().pipe(Effect.orDie); + const part = (kind: string, operation: string) => + [kind, "mcp", encodeURIComponent(scope.thread.providerSessionId), operation, uuid].join(":"); + const { projectId, projectCreated } = yield* projects + .bootstrap({ + commandId: CommandId.make(part("command", "thread-move-project")), + projectId: ProjectId.make(uuid), + title: path.basename(workspaceRoot) || workspaceRoot, + workspaceRoot, + }) + .pipe( + Effect.map(({ project, created }) => ({ projectId: project.id, projectCreated: created })), + // bootstrap reads before locking; racing creates return the winner in a conflict. + Effect.catchTags({ + ProjectConflictError: (conflict) => + Effect.succeed({ + projectId: conflict.conflictingProjectId, + projectCreated: false, + }), + }), + asOperationFailed(`Unable to register a project for '${workspaceRoot}'`), + ); + const { movesWhenTurnEnds } = yield* threadManagement + .moveThreadToProject({ + threadId: thread.id, + projectId, + commandId: CommandId.make(part("command", "thread-move")), + ...(input.continuationPrompt === undefined + ? {} + : { + continuation: { + commandId: CommandId.make(part("command", "thread-move-continuation")), + messageId: MessageId.make(part("message", "thread-move-continuation")), + text: input.continuationPrompt, + }, + }), + }) + .pipe(asOperationFailed("Unable to move the thread")); + + return { + projectId, + workspaceRoot, + projectCreated, + movesWhenTurnEnds, + note: `The thread moves to ${workspaceRoot} ${ + movesWhenTurnEnds + ? "when this turn ends, so finish now and do not start more work here. A failed or stopped turn drops the move" + : "now" + }. ${ + input.continuationPrompt === undefined + ? "The conversation continues there when the thread receives its next message." + : "The continuation prompt then starts the next turn there." + } Files and any worktree in the previous folder stay where they are.`, + }; + }); + const status: WorktreeMcpService["Service"]["status"] = Effect.fn("WorktreeMcpService.status")( function* (callerScope) { yield* requireCapability(callerScope); @@ -457,7 +580,7 @@ const make = Effect.gen(function* () { return yield* gitWorkflow.listRefs({ ...refs, cwd }); }); - return WorktreeMcpService.of({ handoff, status, listRefs }); + return WorktreeMcpService.of({ handoff, moveToFolder, status, listRefs }); }); export const layer: Layer.Layer< @@ -465,6 +588,7 @@ export const layer: Layer.Layer< never, | SecureRandom.SecureRandom | FileSystem.FileSystem + | Path.Path | ThreadManagementService.ThreadManagementService | ProjectService.ProjectService | ServerSettings.ServerSettingsService diff --git a/apps/server/src/mcp/threadAccess.test.ts b/apps/server/src/mcp/threadAccess.test.ts index aa107711f334e..75a0f4f9425a9 100644 --- a/apps/server/src/mcp/threadAccess.test.ts +++ b/apps/server/src/mcp/threadAccess.test.ts @@ -1,12 +1,12 @@ import { expect, it } from "@effect/vitest"; -import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { EnvironmentId, ProjectId, ThreadId } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import * as McpInvocationContext from "./McpInvocationContext.ts"; import { liveThreadShell } from "./McpToolAccess.testkit.ts"; -import { readThread } from "./threadAccess.ts"; +import { readInCurrentProject, readThread } from "./threadAccess.ts"; const bareId = ThreadId.make("5d0b4a3e-8c1f-4a52-9d37-2f6c7e1b9a04"); const client: McpInvocationContext.McpInvocationScope = { @@ -38,3 +38,33 @@ it.effect("points a bare uuid at the mcp: thread it belongs to", () => expect(error.message).toBe(`The thread was not found. Did you mean \`mcp:${bareId}\`?`); }), ); + +it.effect.each([ + { name: "retries in the project the thread moved to", currentProject: "target", reads: 2 }, + { name: "keeps the miss when the thread did not move", currentProject: "source", reads: 1 }, +])("readInCurrentProject $name", ({ currentProject, reads }) => + Effect.gen(function* () { + const attempted: Array = []; + const threads = { + getThreadShell: (threadId: ThreadId) => + Effect.succeed({ + ...liveThreadShell(threadId), + projectId: ProjectId.make(currentProject), + }), + }; + const exit = yield* readInCurrentProject( + threads as never, + bareId, + ProjectId.make("source"), + (projectId) => { + attempted.push(projectId); + return projectId === ProjectId.make("target") + ? Effect.succeed("records") + : Effect.fail({ _tag: "ThreadManagementThreadNotFoundError" as const }); + }, + ).pipe(Effect.exit); + + expect(attempted).toHaveLength(reads); + expect(exit._tag).toBe(currentProject === "target" ? "Success" : "Failure"); + }), +); diff --git a/apps/server/src/mcp/threadAccess.ts b/apps/server/src/mcp/threadAccess.ts index 4bf90f0182fc4..6c1c1d7b54bd6 100644 --- a/apps/server/src/mcp/threadAccess.ts +++ b/apps/server/src/mcp/threadAccess.ts @@ -22,6 +22,32 @@ import * as McpInvocationContext from "./McpInvocationContext.ts"; * bare uuid. The id is never rewritten, so permission checks keyed on it still * hold; the miss only names a live thread the caller could read by its full id. */ +/** + * Reads thread records scoped to the project its shell just reported. A thread + * can move to another project between the two reads, so a miss re-reads the + * project once before it counts. + */ +export const readInCurrentProject = ( + threads: Pick, + threadId: ThreadId, + projectId: ProjectId, + read: (projectId: ProjectId) => Effect.Effect, +) => + read(projectId).pipe( + Effect.catchIf( + (error) => error._tag === "ThreadManagementThreadNotFoundError", + (error) => + threads.getThreadShell(threadId).pipe( + Effect.orElseSucceed(() => null), + Effect.flatMap((shell) => + shell === null || shell.projectId === projectId + ? Effect.fail(error) + : read(shell.projectId), + ), + ), + ), + ); + export const threadNotFound = Effect.fn("OrchestratorMcpService.threadNotFound")(function* ( threads: Pick, threadId: ThreadId, @@ -188,20 +214,24 @@ export const readThread = Effect.fn("mcp.readThread")(function* < if (shell === null || shell.deletedAt !== null) { return yield* threadNotFound(context.threads, targetId, "The thread was not found."); } - const projection = yield* context.threads - .getProjectThreadRecords({ projectId: shell.projectId, threadId: targetId }, fields, { - turnItemTypes: ["user_input_request"], - }) - .pipe( - Effect.mapError((error) => - error._tag === "ThreadManagementThreadNotFoundError" - ? new OrchestratorMcpFailure({ - code: "thread_not_found", - message: "The thread was not found.", - }) - : unavailable(), - ), - ); + const projection = yield* readInCurrentProject( + context.threads, + targetId, + shell.projectId, + (projectId) => + context.threads.getProjectThreadRecords({ projectId, threadId: targetId }, fields, { + turnItemTypes: ["user_input_request"], + }), + ).pipe( + Effect.mapError((error) => + error._tag === "ThreadManagementThreadNotFoundError" + ? new OrchestratorMcpFailure({ + code: "thread_not_found", + message: "The thread was not found.", + }) + : unavailable(), + ), + ); return { ...context, projection }; }); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index f0affbf6dec8c..543a912969244 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -253,7 +253,7 @@ const ThreadReadTool = Tool.make("t3_thread_read", { const ThreadUpdateTool = Tool.make("t3_thread_update", { description: - "Rename a thread (`action:'rename'` with `title`) or regenerate its title. Omit `threadId` for this thread. Workspace and branch cannot change here.", + "Rename a thread (`action:'rename'` with `title`) or regenerate its title. Omit `threadId` for this thread. Workspace: `t3_thread_move` or `t3_worktree_handoff`.", // Keeps the input's action/field check, which rebuilding the struct would drop. parameters: ThreadMetadataMcpUpdateInput.mapFields( Struct.assign({ environmentId: OrchestratorMcpEnvironmentTarget }), diff --git a/apps/server/src/mcp/toolkits/worktree/handlers.ts b/apps/server/src/mcp/toolkits/worktree/handlers.ts index 2e78dd2a4f996..b18fb55a02bfd 100644 --- a/apps/server/src/mcp/toolkits/worktree/handlers.ts +++ b/apps/server/src/mcp/toolkits/worktree/handlers.ts @@ -46,6 +46,13 @@ const handlers = { return yield* service.handoff(scope, input); }), ), + t3_thread_move: McpToolAccess.actsAsCaller((input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const service = yield* WorktreeMcpService.WorktreeMcpService; + return yield* service.moveToFolder(scope, input); + }), + ), t3_worktree_status: McpToolAccess.readsAsCaller(() => Effect.gen(function* () { const scope = yield* McpInvocationContext.McpInvocationContext; diff --git a/apps/server/src/mcp/toolkits/worktree/tools.ts b/apps/server/src/mcp/toolkits/worktree/tools.ts index 18d6840d72291..5cfe6df78a820 100644 --- a/apps/server/src/mcp/toolkits/worktree/tools.ts +++ b/apps/server/src/mcp/toolkits/worktree/tools.ts @@ -4,6 +4,8 @@ import { OrchestratorMcpFailure, VcsListRefsInput, VcsListRefsResult, + ThreadMoveMcpInput, + ThreadMoveMcpResult, WorktreeMcpHandoffInput, WorktreeMcpHandoffResult, WorktreeMcpStatusResult, @@ -40,6 +42,21 @@ const WorktreeHandoffTool = Tool.make("t3_worktree_handoff", { .annotate(Tool.Idempotent, false) .annotate(Tool.OpenWorld, true); +const ThreadMoveTool = Tool.make("t3_thread_move", { + description: + "Move THIS thread to another folder on this machine, keeping its conversation. Use it when the user asks to move or continue this thread in another folder or project. A separate thread: `t3_thread_launch`. Uses the project rooted at `path`, registering one if none exists. Files stay where they are; the branch and worktree are cleared. Call last in your turn: the move happens when the turn ends. `continuationPrompt` resumes work there next turn. Without it, the thread waits.", + parameters: ThreadMoveMcpInput, + success: ThreadMoveMcpResult, + failure: WorktreeToolFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Move thread to a folder") + .annotate(Tool.Readonly, false) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, false) + .annotate(Tool.OpenWorld, false); + const WorktreeStatusTool = Tool.make("t3_worktree_status", { description: "Report this thread's worktree: attached or not, path, branch, project root, and the `startFromOrigin` default. Check before `t3_worktree_handoff`.", @@ -84,6 +101,7 @@ const WorktreeListTool = Tool.make("t3_worktree_list", { .annotate(Tool.Destructive, false); export const WorktreeToolkit = Toolkit.make( WorktreeHandoffTool, + ThreadMoveTool, WorktreeStatusTool, WorktreeListTool, ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bcec0a13cf20a..72afdd3e3fe7e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -581,6 +581,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "thread.pull-request.sync": case "thread.title.regeneration.complete": case "thread.project.adopt-folder": + case "thread.project.move": case "thread.runtime-mode.set": case "thread.interaction-mode.set": case "thread.model-selection.set": @@ -3245,6 +3246,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | "thread.active.reorder" | "thread.mark-unread" | "thread.metadata.update" + | "thread.project.move" | "thread.pull-request.link" | "thread.pull-request.unlink" | "thread.pull-request-link.sync" @@ -3277,6 +3279,32 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} is deleted.`, }); } + if (command.type === "thread.project.move") { + if (isProviderNativeSubagentThread(thread)) { + return yield* new OrchestratorSubagentThreadReadOnlyError({ + commandId: command.commandId, + threadId: command.threadId, + }); + } + const records = yield* projectionStore + .getThreadRecords(command.threadId, ["runs", "runtimeRequests"], { turnItemTypes: [] }) + .pipe(mapDispatchError(command)); + if (threadHasUnfinishedWork(records)) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} still has work in progress.`, + }); + } + const target = yield* projects.get(command.projectId).pipe(mapDispatchError(command)); + if (Option.isNone(target) || target.value.deletedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} cannot move to missing project ${command.projectId}.`, + }); + } + } if ( command.type === "thread.pull-request.watch" && command.watching && @@ -3544,6 +3572,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.type === "provider.switch" || command.type === "thread.archive" || command.type === "thread.settle" || + command.type === "thread.project.move" || (command.type === "thread.metadata.update" && command.worktreePath !== undefined && command.worktreePath !== thread.worktreePath); @@ -3822,6 +3851,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : {}), }; } + case "thread.project.move": + return { + ...thread, + projectId: command.projectId, + branch: null, + worktreePath: null, + updatedAt: now, + }; case "thread.metadata.update": { const previousRecovery = thread.limitRecovery?.runId === command.limitRecovery?.runId && @@ -4185,6 +4222,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.mark-unread": return "thread.marked-unread" as const; case "thread.metadata.update": + case "thread.project.move": case "thread.title.regeneration.complete": return "thread.metadata-updated" as const; case "thread.pull-request.link": @@ -4329,9 +4367,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const detachSessionIds = new Set( command.type === "thread.archive" || command.type === "thread.settle" ? (providerContext?.providerSessions ?? []).map((session) => session.id) - : command.type === "thread.metadata.update" && - command.worktreePath !== undefined && - command.worktreePath !== thread.worktreePath + : command.type === "thread.project.move" || + (command.type === "thread.metadata.update" && + command.worktreePath !== undefined && + command.worktreePath !== thread.worktreePath) ? (providerContext?.providerSessions ?? []).map((session) => session.id) : command.type === "thread.runtime-mode.set" ? (providerContext?.providerSessions ?? []) @@ -4369,7 +4408,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? "Thread archived." : command.type === "thread.settle" ? "Thread settled." - : command.type === "thread.metadata.update" + : command.type === "thread.metadata.update" || + command.type === "thread.project.move" ? "Workspace changed." : command.type === "thread.runtime-mode.set" ? "Runtime mode changed." @@ -4388,7 +4428,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? "Thread archived." : command.type === "thread.settle" ? "Thread settled." - : command.type === "thread.metadata.update" + : command.type === "thread.metadata.update" || + command.type === "thread.project.move" ? "Workspace changed." : command.type === "thread.runtime-mode.set" ? "Runtime mode changed." @@ -13479,6 +13520,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.tag.set": case "thread.mark-unread": case "thread.metadata.update": + case "thread.project.move": case "thread.pull-request.link": case "thread.pull-request.unlink": case "thread.pull-request-link.sync": diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 742616f27bd3e..342f41719c040 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -703,6 +703,194 @@ it.effect.each([ }), ); +it.effect.each([ + { status: "completed", moves: true }, + { status: "failed", moves: false }, +] as const)("moveThreadToProject handles a $status turn", ({ status, moves }) => + Effect.gen(function* () { + const sourceProjectId = ProjectId.make("project:thread-management:move-source"); + const targetProjectId = ProjectId.make("project:thread-management:move-target"); + const threadId = ThreadId.make("thread:thread-management:move"); + const runId = RunId.make("run:thread-management:move"); + const queuedRunId = RunId.make("run:thread-management:move-queued"); + const events = yield* Queue.unbounded(); + const subscribed = yield* Deferred.make(); + const continued = yield* Deferred.make(); + const refused = yield* Deferred.make(); + const runs: Array<{ + id: RunId; + ordinal: number; + status: OrchestrationV2Run["status"]; + userMessageId: MessageId; + }> = [{ id: runId, ordinal: 1, status: "running", userMessageId: MessageId.make("original") }]; + let projectId = sourceProjectId; + let sequence = 0; + let queuedRunBusy = true; + const terminalRead = yield* Deferred.make(); + const moveCommands: Array = []; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.sync(() => sequence), + getThreadShell: () => + Effect.succeed({ id: threadId, projectId, activeRunId: runId } as never), + restartContinuationPending: () => Effect.succeed(false), + streamStoredEventsFrom: (input) => + input?.eventType === "run.updated" + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(events)), + ) + : Stream.never, + getThreadRecords: () => + Effect.gen(function* () { + if (runs[0]!.status !== "running") yield* Deferred.succeed(terminalRead, undefined); + return { + thread: { + id: threadId, + projectId, + archivedAt: null, + deletedAt: null, + modelSelection: { + instanceId: ProviderInstanceId.make("claudeAgent"), + model: "test", + }, + }, + runs, + messages: [], + turnItems: [], + providerTurns: [], + } as unknown as OrchestrationV2ThreadProjection; + }), + dispatch: (command) => + Effect.gen(function* () { + if (command.type === "thread.project.move") { + moveCommands.push(command.commandId); + // A message queued during the turn runs before the move can apply. + if (queuedRunBusy) { + yield* Deferred.succeed(refused, undefined); + return yield* new Orchestrator.OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Thread still has work in progress.", + }); + } + projectId = command.projectId; + } + if (command.type === "message.dispatch") { + expect(projectId).toBe(targetProjectId); + expect(command.text).toBe("Carry on."); + yield* Deferred.succeed(continued, undefined); + } + return { sequence: 1, storedEvents: [] } as never; + }), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + + yield* Effect.gen(function* () { + const service = yield* ThreadManagementService.ThreadManagementService; + const result = yield* service.moveThreadToProject({ + threadId, + projectId: targetProjectId, + commandId: CommandId.make("command:thread-management:move"), + continuation: { + commandId: CommandId.make("command:thread-management:move-continuation"), + messageId: MessageId.make("message:thread-management:move-continuation"), + text: "Carry on.", + }, + }); + expect(result).toEqual({ movesWhenTurnEnds: true }); + yield* Deferred.await(subscribed); + expect(moveCommands).toEqual([]); + + runs[0]!.status = status; + sequence = 1; + yield* Queue.offer(events, { + sequence, + event: { type: "run.updated", threadId, payload: { id: runId } }, + } as unknown as OrchestrationV2StoredEvent); + yield* Deferred.await(terminalRead); + if (!moves) { + expect(moveCommands).toEqual([]); + return; + } + + yield* Deferred.await(refused); + expect(projectId).toBe(sourceProjectId); + queuedRunBusy = false; + sequence = 2; + yield* Queue.offer(events, { + sequence, + event: { type: "run.updated", threadId, payload: { id: queuedRunId } }, + } as unknown as OrchestrationV2StoredEvent); + yield* Deferred.await(continued); + expect(new Set(moveCommands).size).toBe(moveCommands.length); + }).pipe(Effect.provide(layerTest)); + }), +); + +it.effect("waitForThread follows a thread that moves to another project", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project:thread-management:wait-source"); + const threadId = ThreadId.make("thread:thread-management:wait-move"); + const runId = RunId.make("run:thread-management:wait-move"); + const events = yield* Queue.unbounded(); + const run: { id: RunId; ordinal: number; status: OrchestrationV2Run["status"] } = { + id: runId, + ordinal: 1, + status: "running", + }; + let currentProjectId = projectId; + const subscribed = yield* Deferred.make(); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide( + ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + restartContinuationPending: () => Effect.succeed(false), + streamStoredEventsFrom: (input) => + input?.eventType === "run.updated" + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(events)), + ) + : Stream.never, + getThreadRecords: () => + Effect.sync( + () => + ({ + thread: { id: threadId, projectId: currentProjectId, deletedAt: null }, + runs: [run], + }) as unknown as OrchestrationV2ThreadProjection, + ), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ), + ), + ); + const fiber = yield* service + .waitForThread({ projectId, threadId, runId, timeoutMs: 60_000 }) + .pipe(Effect.forkChild); + yield* Deferred.await(subscribed); + run.status = "completed"; + currentProjectId = ProjectId.make("project:thread-management:wait-target"); + yield* Queue.offer(events, { + sequence: 1, + event: { type: "run.updated", threadId, payload: { id: runId } }, + } as unknown as OrchestrationV2StoredEvent); + const result = yield* Fiber.join(fiber); + expect(result).toMatchObject({ timedOut: false, run: { status: "completed" } }); + }), +); + it.effect.each([ { name: "kills the thread's own agent and ends its turn", diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 46ea454310554..ff79949f564c3 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -453,6 +453,20 @@ export interface ThreadManagementServiceShape { readonly threadId: ThreadId; readonly runId: RunId; }) => Effect.Effect; + /** Moves the thread once it is idle. Dropped if the turn in progress does not complete. */ + readonly moveThreadToProject: (input: { + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly commandId: CommandId; + readonly continuation?: { + readonly commandId: CommandId; + readonly messageId: MessageId; + readonly text: string; + }; + }) => Effect.Effect< + { readonly movesWhenTurnEnds: boolean }, + Orchestrator.OrchestratorV2Error | ThreadManagementFailure + >; readonly settleThread: (input: { readonly threadId: ThreadId; readonly commandId: CommandId; @@ -757,31 +771,32 @@ const make = Effect.gen(function* () { Effect.andThen(orchestrator.dispatch(command)), ); - const getProjectThreadRecords: ThreadManagementServiceShape["getProjectThreadRecords"] = ( - input, - fields, - filter, - ) => - ensureProjectionTranscript(input.threadId).pipe( - Effect.andThen(orchestrator.getThreadRecords(input.threadId, fields, filter)), - Effect.mapError( - (cause) => - new ThreadManagementProjectionLoadError({ - projectId: input.projectId, - threadId: input.threadId, - cause, - }), - ), - Effect.filterOrFail( - (projection) => - projection.thread.projectId === input.projectId && projection.thread.deletedAt === null, - (_projection) => - new ThreadManagementThreadNotFoundError({ - projectId: input.projectId, - threadId: input.threadId, - }), - ), - ); + const readThreadRecords = + (anyProject: boolean): ThreadManagementServiceShape["getProjectThreadRecords"] => + (input, fields, filter) => + ensureProjectionTranscript(input.threadId).pipe( + Effect.andThen(orchestrator.getThreadRecords(input.threadId, fields, filter)), + Effect.mapError( + (cause) => + new ThreadManagementProjectionLoadError({ + projectId: input.projectId, + threadId: input.threadId, + cause, + }), + ), + Effect.filterOrFail( + (projection) => + (anyProject || projection.thread.projectId === input.projectId) && + projection.thread.deletedAt === null, + (_projection) => + new ThreadManagementThreadNotFoundError({ + projectId: input.projectId, + threadId: input.threadId, + }), + ), + ); + const getProjectThreadRecords = readThreadRecords(false); + const readThreadRecordsInAnyProject = readThreadRecords(true); const listProjectThreads: ThreadManagementServiceShape["listProjectThreads"] = (input) => orchestrator @@ -948,7 +963,8 @@ const make = Effect.gen(function* () { }); const untilDelegatedWorkSettled = input.untilDelegatedWorkSettled === true; const delegatedWorkResult = Effect.gen(function* () { - const records = yield* getProjectThreadRecords( + // The thread may move to another project once its turn ends; the wait follows it. + const records = yield* readThreadRecordsInAnyProject( input, ["runs", "messages", "subagents", "providerThreads"], { messageRoles: ["user"], monitorNotificationsOnly: true }, @@ -1023,7 +1039,7 @@ const make = Effect.gen(function* () { : Stream.empty, ).pipe( Stream.mapEffect(() => - getProjectThreadRecords(input, ["runs"]).pipe( + readThreadRecordsInAnyProject(input, ["runs"]).pipe( Effect.flatMap((current) => evaluate(current.runs)), ), ), @@ -1036,7 +1052,9 @@ const make = Effect.gen(function* () { // The run may have reached a terminal status while the timeout was // winning the race; the final projection read decides what actually // happened, so only a still-active run counts as timed out. - return settledResult(yield* evaluate((yield* getProjectThreadRecords(input, ["runs"])).runs)); + return settledResult( + yield* evaluate((yield* readThreadRecordsInAnyProject(input, ["runs"])).runs), + ); }); const settleAfterRun: ThreadManagementServiceShape["settleAfterRun"] = Effect.fn( @@ -1114,6 +1132,83 @@ const make = Effect.gen(function* () { ); }); + const moveThreadToProject: ThreadManagementServiceShape["moveThreadToProject"] = Effect.fn( + "orchestrationV2.threadManagement.moveThreadToProject", + )(function* (input) { + const apply = Effect.gen(function* () { + const afterSequence = yield* orchestrator.getThreadEventSequence(input.threadId); + const tryMove = Effect.gen(function* () { + const sequence = yield* orchestrator.getThreadEventSequence(input.threadId); + return yield* dispatch({ + type: "thread.project.move", + commandId: CommandId.make(`${input.commandId}:${sequence}`), + threadId: input.threadId, + projectId: input.projectId, + }).pipe( + Effect.as(true), + // A queued message may run first; the next run update tries again. + Effect.catchTags({ + OrchestratorDispatchError: () => Effect.succeed(false), + OrchestratorCommandPreviouslyRejectedError: () => Effect.succeed(false), + }), + ); + }); + const moved = yield* Stream.fromEffect(tryMove).pipe( + Stream.concat( + orchestrator + .streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "run.updated", + }) + .pipe(Stream.mapEffect(() => tryMove)), + ), + Stream.filter((done) => done), + Stream.runHead, + Effect.timeoutOption(Duration.millis(SETTLE_AFTER_RUN_WAIT_MS)), + ); + if (Option.isNone(moved) || Option.isNone(moved.value) || input.continuation === undefined) { + return; + } + yield* sendToThread({ + projectId: input.projectId, + commandId: input.continuation.commandId, + threadId: input.threadId, + messageId: input.continuation.messageId, + text: input.continuation.text, + attachments: [], + mode: "queue", + createdBy: "agent", + creationSource: "mcp", + }); + }); + const shell = yield* orchestrator.getThreadShell(input.threadId); + if (shell == null || shell.activeRunId === null) { + yield* apply; + return { movesWhenTurnEnds: false }; + } + const { projectId, activeRunId } = shell; + yield* waitForThread({ + projectId, + threadId: input.threadId, + runId: activeRunId, + timeoutMs: SETTLE_AFTER_RUN_WAIT_MS, + }).pipe( + Effect.flatMap(({ run }) => + run?.status === "completed" + ? apply + : Effect.logInfo("Dropped a thread move because its turn did not complete", { + threadId: input.threadId, + runId: activeRunId, + status: run?.status ?? null, + }), + ), + Effect.ignoreCause({ log: true }), + Effect.forkIn(layerScope), + ); + return { movesWhenTurnEnds: true }; + }); + const settleThread: ThreadManagementServiceShape["settleThread"] = Effect.fn( "orchestrationV2.threadManagement.settleThread", )(function* (input) { @@ -1616,6 +1711,7 @@ const make = Effect.gen(function* () { hasReceipt, waitForThread, settleAfterRun, + moveThreadToProject, settleThread, interruptThread, stopDelegatedTasks, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index b90f491dfa94e..25c05489f6659 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -7511,6 +7511,77 @@ it.layer(layerSharedApplicationDataPlaneTest)("snooze projection", (it) => { assert.equal(stayed?.projectId, scratchProjectId); }), ); + + it.effect("moves an idle thread to another project and refuses a busy one", () => + Effect.gen(function* () { + const projects = yield* ProjectService.ProjectService; + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sourceProjectId = ProjectId.make("runtime-layer-move-source"); + const targetProjectId = ProjectId.make("runtime-layer-move-target"); + for (const projectId of [sourceProjectId, targetProjectId]) { + yield* projects.create({ + commandId: CommandId.make(`${projectId}-create`), + projectId, + title: projectId, + workspaceRoot: `/tmp/${projectId}`, + }); + } + const createThread = (threadId: ThreadId) => + orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}-create`), + threadId, + projectId: sourceProjectId, + title: "Move me", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: "feature/old", + worktreePath: "/tmp/runtime-layer-move-worktree", + }); + const move = (threadId: ThreadId, projectId: ProjectId, suffix: string) => + orchestrator.dispatch({ + type: "thread.project.move", + commandId: CommandId.make(`${threadId}-move-${suffix}`), + threadId, + projectId, + }); + const idleThreadId = ThreadId.make("runtime-layer-move-idle"); + const busyThreadId = ThreadId.make("runtime-layer-move-busy"); + yield* createThread(idleThreadId); + yield* createThread(busyThreadId); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-move-busy-message"), + threadId: busyThreadId, + messageId: MessageId.make("runtime-layer-move-busy-message"), + text: "Keep this run active.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + + const missing = yield* move(idleThreadId, ProjectId.make("missing"), "missing").pipe( + Effect.flip, + ); + assert.equal(missing.cause, `Thread ${idleThreadId} cannot move to missing project missing.`); + const busy = yield* move(busyThreadId, targetProjectId, "busy").pipe(Effect.flip); + assert.equal(busy.cause, `Thread ${busyThreadId} still has work in progress.`); + + yield* move(idleThreadId, targetProjectId, "target"); + const shell = yield* orchestrator.getShellSnapshot(); + const moved = shell.threads.find((thread) => thread.id === idleThreadId); + assert.equal(moved?.projectId, targetProjectId); + assert.isNull(moved?.branch); + assert.isNull(moved?.worktreePath); + const stayed = shell.threads.find((thread) => thread.id === busyThreadId); + assert.equal(stayed?.projectId, sourceProjectId); + }), + ); }); it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => { diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 0d9f26d44c40a..ac8a567fd2b12 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5366, "codexAdditionalContextBytes": 7096, - "mcpToolListJsonBytes": 48119 + "mcpToolListJsonBytes": 48783 } diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index e76e2a6b64b05..dcae1769c58c6 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -225,6 +225,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( isProviderInstanceEnabledForProject: unused, hasReceipt: unused, waitForThread: unused, + moveThreadToProject: unused, settleAfterRun: unused, settleThread: unused, interruptThread: unused, diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 25095b70f1c3d..82af595efc908 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -32,6 +32,12 @@ list. It starts on your current machine; before sending, pick another machine from the machine menu to move it there. To move a draft into a project, pick the project in the heading. +To move a thread that has already started, ask its agent to move to the folder +you name. The conversation carries over and the thread lands in that folder's +project, registering one if there is none. The move happens when the turn ends; +a failed or stopped turn drops it. Files stay where they are, and the thread's +branch and worktree are cleared. + Each thread without a project works in its own folder under `~/.t3/scratch` (the `scratch` folder of your T3 data directory), named after its date, the first words of its first message, and a short id, like diff --git a/packages/client-runtime/src/t3ToolSummary.ts b/packages/client-runtime/src/t3ToolSummary.ts index 1d57c1f0c91b7..6192430f7f134 100644 --- a/packages/client-runtime/src/t3ToolSummary.ts +++ b/packages/client-runtime/src/t3ToolSummary.ts @@ -314,6 +314,9 @@ export function summarizeT3ToolCalls( case "secret-request": label = phrase("Asked for", "ask for", quantity(selected.length, "secret")); break; + case "thread-move": + label = phrase("Moved", "move", quantity(selected.length, "thread")); + break; case "worktree-handoff": label = phrase( "Handed off to", diff --git a/packages/client-runtime/src/work-log/presentation.ts b/packages/client-runtime/src/work-log/presentation.ts index fd0dca831e06d..a12d84831bf1d 100644 --- a/packages/client-runtime/src/work-log/presentation.ts +++ b/packages/client-runtime/src/work-log/presentation.ts @@ -538,6 +538,7 @@ function summaryActionPriority(action: ToolGroupAction | T3McpToolSummaryAction) case "queue-steer": case "question-respond": case "worktree-handoff": + case "thread-move": case "project-create": case "project-update": case "project-delete": diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 461ec5e4b3f3e..68f97b75dee29 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3421,6 +3421,13 @@ const OrchestrationV2InternalCommand = Schema.Union([ threadId: ThreadId, projectId: ProjectId, }), + /** Moves an idle thread to another project's root. Its provider session restarts there on the next turn. */ + Schema.Struct({ + type: Schema.Literal("thread.project.move"), + commandId: CommandId, + threadId: ThreadId, + projectId: ProjectId, + }), Schema.Struct({ type: Schema.Literal("thread.pull-request-watch.sync"), commandId: CommandId, diff --git a/packages/contracts/src/worktreeMcp.ts b/packages/contracts/src/worktreeMcp.ts index 87f45dea27598..9f3dbaed4edd0 100644 --- a/packages/contracts/src/worktreeMcp.ts +++ b/packages/contracts/src/worktreeMcp.ts @@ -1,6 +1,6 @@ import * as Schema from "effect/Schema"; -import { TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { ProjectId, TrimmedNonEmptyString } from "./baseSchemas.ts"; /** Uses the managed worktree folder; handoff detaches the provider, ending this turn and resuming there next run. */ export const WorktreeMcpHandoffInput = Schema.Struct({ @@ -74,6 +74,28 @@ export const WorktreeMcpHandoffResult = Schema.Struct({ }); export type WorktreeMcpHandoffResult = typeof WorktreeMcpHandoffResult.Type; +/** Keeps the conversation; the move waits for this turn to end, and the next turn resumes in the folder. */ +export const ThreadMoveMcpInput = Schema.Struct({ + path: TrimmedNonEmptyString.annotate({ + description: "Absolute path of an existing folder on this machine.", + }), + continuationPrompt: Schema.optional( + TrimmedNonEmptyString.check(Schema.isMaxLength(120_000)).annotate({ + description: "Work to resume in the folder next turn. Omitted: thread waits.", + }), + ), +}); +export type ThreadMoveMcpInput = typeof ThreadMoveMcpInput.Type; + +export const ThreadMoveMcpResult = Schema.Struct({ + projectId: ProjectId, + workspaceRoot: TrimmedNonEmptyString, + projectCreated: Schema.Boolean, + movesWhenTurnEnds: Schema.Boolean, + note: Schema.String, +}); +export type ThreadMoveMcpResult = typeof ThreadMoveMcpResult.Type; + export const WorktreeMcpStatusResult = Schema.Struct({ attached: Schema.Boolean, worktreePath: Schema.NullOr(TrimmedNonEmptyString), diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index 7a6c2fab91187..92836461dfe3f 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -42,6 +42,7 @@ export type T3McpToolSummaryAction = | "question-respond" | "secret-request" | "worktree-handoff" + | "thread-move" | "worktree-list" | "worktree-status" | "project-list" @@ -180,6 +181,7 @@ const T3_MCP_TOOLS: Readonly> = { ["Interrupt", "Interrupting", "Requested an interrupt of", "a T3 thread"], "thread-interrupt", ), + t3_thread_move: tool(["Move", "Moving", "Moved", "thread to a folder"], "thread-move"), t3_worktree_handoff: tool( ["Hand off", "Handing off", "Handed off", "thread to a git worktree"], "worktree-handoff", From 9f88858600b47205595207b354d1052249ca2ead Mon Sep 17 00:00:00 2001 From: giri256 <80974392+giri256@users.noreply.github.com> Date: Sat, 7 Nov 2026 02:51:58 +0000 Subject: [PATCH 2/6] fix(server): tighten thread moves after review Moving to a folder widens what the next turn can write, so it now needs the same full access as registering a project. A move also takes the thread's scheduled tasks along, refuses a destination that disables the thread's provider, and returns at once when an immediate move fails. Model: Claude Opus 5.5, harness: Claude Code in T3 Code. --- apps/server/src/mcp/McpToolAccess.test.ts | 8 ++ apps/server/src/mcp/McpToolAccess.ts | 13 ++ .../server/src/mcp/WorktreeMcpService.test.ts | 45 +++++- apps/server/src/mcp/WorktreeMcpService.ts | 18 ++- .../src/mcp/toolkits/orchestrator/tools.ts | 2 +- .../src/mcp/toolkits/worktree/handlers.ts | 2 +- .../server/src/mcp/toolkits/worktree/tools.ts | 4 +- .../ThreadManagementService.test.ts | 132 +++++++++++++++++- .../ThreadManagementService.ts | 103 +++++++++----- .../src/provider/agentContextBudget.json | 2 +- docs/user/thread-sidebar.md | 2 +- packages/contracts/src/worktreeMcp.ts | 4 +- 12 files changed, 286 insertions(+), 49 deletions(-) diff --git a/apps/server/src/mcp/McpToolAccess.test.ts b/apps/server/src/mcp/McpToolAccess.test.ts index 6d63287f4eb2a..4a30f06d96a34 100644 --- a/apps/server/src/mcp/McpToolAccess.test.ts +++ b/apps/server/src/mcp/McpToolAccess.test.ts @@ -115,6 +115,7 @@ const ProbeToolkit = Toolkit.make( Tool.make("reads", probe), Tool.make("reads_as_caller", probe), Tool.make("acts_as_caller", probe), + Tool.make("acts_as_full_access_caller", probe), Tool.make("writes", probe), Tool.make("writes_with_capability", probe), Tool.make("writes_threads", { @@ -135,6 +136,7 @@ const probeHandlers: McpToolAccess.Handlers = { reads: McpToolAccess.reads(() => ran), reads_as_caller: McpToolAccess.readsAsCaller(() => ran), acts_as_caller: McpToolAccess.actsAsCaller(() => ran), + acts_as_full_access_caller: McpToolAccess.actsAsFullAccessCaller(() => ran), writes: McpToolAccess.writes(() => ran), writes_with_capability: McpToolAccess.writesWithCapability(() => ran), writes_threads: McpToolAccess.writesThreads( @@ -181,6 +183,7 @@ export const refusedAtCompileTime = () => { reads: unchecked, reads_as_caller: unchecked, acts_as_caller: unchecked, + acts_as_full_access_caller: unchecked, writes: unchecked, writes_with_capability: unchecked, writes_threads: unchecked, @@ -263,6 +266,11 @@ it.effect.each([ ["acts_as_caller", archivedBackground, {}, "parent_not_active"], ["acts_as_caller", unknownOwnerBackground, {}, "parent_not_active"], ["acts_as_caller", fullAccessClient, {}, "thread_credential_required"], + ["acts_as_full_access_caller", fullAccess, {}, "ran"], + ["acts_as_full_access_caller", supervised, {}, "capability_denied"], + ["acts_as_full_access_caller", plan, {}, "capability_denied"], + ["acts_as_full_access_caller", ended, {}, "parent_not_active"], + ["acts_as_full_access_caller", fullAccessClient, {}, "thread_credential_required"], ["writes", supervised, {}, "ran"], ["writes", ended, {}, "parent_not_active"], diff --git a/apps/server/src/mcp/McpToolAccess.ts b/apps/server/src/mcp/McpToolAccess.ts index b82d9bb5ae59c..179526629d251 100644 --- a/apps/server/src/mcp/McpToolAccess.ts +++ b/apps/server/src/mcp/McpToolAccess.ts @@ -284,6 +284,19 @@ export const either = ( const fullAccessRequired = "Changing projects or environment settings needs a live full-access/default calling thread or a full-access client."; +/** The calling thread picks a folder its next turn can write, so this needs the bar for adding a project. */ +export const actsAsFullAccessCaller = (handle: (params: P) => Effect.Effect) => + declare( + (params: P) => + requireThreadCaller.pipe( + Effect.andThen(writingCaller), + Effect.tap((caller) => assertNotLinked(caller, "change projects or environment settings")), + Effect.tap((caller) => assertFullAccess(caller, fullAccessRequired)), + Effect.flatMap(() => handle(params)), + ), + undefined, + ); + /** Recheck after waits before writing; caller modes may have changed. */ export const writesEnvironment = ( handle: ( diff --git a/apps/server/src/mcp/WorktreeMcpService.test.ts b/apps/server/src/mcp/WorktreeMcpService.test.ts index d14b2656fbca7..91b604769e04c 100644 --- a/apps/server/src/mcp/WorktreeMcpService.test.ts +++ b/apps/server/src/mcp/WorktreeMcpService.test.ts @@ -10,6 +10,7 @@ import { ProviderInstanceId, ThreadId, WorktreeMcpHandoffInput, + ScheduledTaskError, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; @@ -33,6 +34,7 @@ import { import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts"; import type * as McpInvocationContext from "./McpInvocationContext.ts"; @@ -120,6 +122,8 @@ interface HarnessOptions { readonly existsFails?: boolean; readonly subagentStatus?: string; readonly destinationProjectExists?: boolean; + readonly moveContinuationFailure?: string; + readonly moveTasksFails?: boolean; } const makeHarness = (options: HarnessOptions = {}) => { @@ -221,7 +225,21 @@ const makeHarness = (options: HarnessOptions = {}) => { created: options.destinationProjectExists !== true, }), ); - const moveThreadToProject = vi.fn((_: unknown) => Effect.succeed({ movesWhenTurnEnds: true })); + const moveThreadToProject = vi.fn((input: { readonly afterMove: Effect.Effect }) => + input.afterMove.pipe( + Effect.as({ + movesWhenTurnEnds: options.moveContinuationFailure === undefined, + ...(options.moveContinuationFailure === undefined + ? {} + : { continuationFailure: options.moveContinuationFailure }), + }), + ), + ); + const moveThreadTasks = vi.fn((_: unknown) => + options.moveTasksFails + ? Effect.fail(new ScheduledTaskError({ message: "simulated" })) + : Effect.void, + ); const removeWorktree = vi.fn((_: unknown) => options.removeWorktreeFails ? (Effect.fail("simulated worktree removal failure") as never) @@ -323,6 +341,9 @@ const makeHarness = (options: HarnessOptions = {}) => { getById, bootstrap: bootstrap as never, } satisfies Partial), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({ + moveThreadTasks: moveThreadTasks as never, + }), ServerSettings.layerTest({ newWorktreesStartFromOrigin: options.newWorktreesStartFromOrigin ?? false, ...(options.projectNewWorktreesStartFromOrigin === undefined @@ -384,6 +405,7 @@ const makeHarness = (options: HarnessOptions = {}) => { runForThread, bootstrap, moveThreadToProject, + moveThreadTasks, }; }; @@ -453,6 +475,10 @@ describe("t3_thread_move", () => { continuation: expect.objectContaining({ text: "Carry on." }), }), ); + expect(harness.moveThreadTasks).toHaveBeenCalledWith({ + threadIds: [threadId], + projectId: result.projectId, + }); }), ).pipe(Effect.provide(NodeServices.layer)), ); @@ -476,6 +502,23 @@ describe("t3_thread_move", () => { ).pipe(Effect.provide(NodeServices.layer)), ); + it.effect("reports a moved thread whose tasks or continuation could not follow", () => + Effect.scoped( + Effect.gen(function* () { + const folder = yield* makeFolder(); + const harness = makeHarness({ + capabilities: moveCapabilities, + moveContinuationFailure: "provider disabled", + moveTasksFails: true, + }); + const result = yield* runMove(harness, { path: folder, continuationPrompt: "Carry on." }); + + expect(result.movesWhenTurnEnds).toBe(false); + expect(result.note).toContain("was not queued (provider disabled)"); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + it.effect("refuses a move it cannot honor before changing anything", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/mcp/WorktreeMcpService.ts b/apps/server/src/mcp/WorktreeMcpService.ts index dcade742af439..00ecf987fc345 100644 --- a/apps/server/src/mcp/WorktreeMcpService.ts +++ b/apps/server/src/mcp/WorktreeMcpService.ts @@ -29,6 +29,7 @@ import * as GitWorkflowService from "../git/GitWorkflowService.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts"; import type { McpInvocationScope, McpThreadInvocationScope } from "./McpInvocationContext.ts"; @@ -84,6 +85,7 @@ const make = Effect.gen(function* () { const threadManagement = yield* ThreadManagementService.ThreadManagementService; const projects = yield* ProjectService.ProjectService; const serverSettings = yield* ServerSettings.ServerSettingsService; + const scheduledTasks = yield* ScheduledTaskService.ScheduledTaskService; const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; const setupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; const vcsStatusBroadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; @@ -515,11 +517,14 @@ const make = Effect.gen(function* () { }), asOperationFailed(`Unable to register a project for '${workspaceRoot}'`), ); - const { movesWhenTurnEnds } = yield* threadManagement + const { movesWhenTurnEnds, continuationFailure } = yield* threadManagement .moveThreadToProject({ threadId: thread.id, projectId, commandId: CommandId.make(part("command", "thread-move")), + afterMove: scheduledTasks + .moveThreadTasks({ threadIds: [thread.id], projectId }) + .pipe(Effect.ignoreCause({ log: true })), ...(input.continuationPrompt === undefined ? {} : { @@ -530,7 +535,7 @@ const make = Effect.gen(function* () { }, }), }) - .pipe(asOperationFailed("Unable to move the thread")); + .pipe(asOperationFailed("The thread did not move")); return { projectId, @@ -542,9 +547,11 @@ const make = Effect.gen(function* () { ? "when this turn ends, so finish now and do not start more work here. A failed or stopped turn drops the move" : "now" }. ${ - input.continuationPrompt === undefined - ? "The conversation continues there when the thread receives its next message." - : "The continuation prompt then starts the next turn there." + continuationFailure !== undefined + ? `The continuation prompt was not queued (${continuationFailure}), so send it again.` + : input.continuationPrompt === undefined + ? "The conversation continues there when the thread receives its next message." + : "The continuation prompt then starts the next turn there." } Files and any worktree in the previous folder stay where they are.`, }; }); @@ -592,6 +599,7 @@ export const layer: Layer.Layer< | ThreadManagementService.ThreadManagementService | ProjectService.ProjectService | ServerSettings.ServerSettingsService + | ScheduledTaskService.ScheduledTaskService | GitWorkflowService.GitWorkflowService | ProjectSetupScriptRunner.ProjectSetupScriptRunner | VcsStatusBroadcaster.VcsStatusBroadcaster diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 543a912969244..6e10a959091cb 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -253,7 +253,7 @@ const ThreadReadTool = Tool.make("t3_thread_read", { const ThreadUpdateTool = Tool.make("t3_thread_update", { description: - "Rename a thread (`action:'rename'` with `title`) or regenerate its title. Omit `threadId` for this thread. Workspace: `t3_thread_move` or `t3_worktree_handoff`.", + "Rename a thread (`action:'rename'` with `title`) or regenerate its title. Omit `threadId` for this thread. Workspace: `t3_thread_move`.", // Keeps the input's action/field check, which rebuilding the struct would drop. parameters: ThreadMetadataMcpUpdateInput.mapFields( Struct.assign({ environmentId: OrchestratorMcpEnvironmentTarget }), diff --git a/apps/server/src/mcp/toolkits/worktree/handlers.ts b/apps/server/src/mcp/toolkits/worktree/handlers.ts index b18fb55a02bfd..a5b108ffe2d34 100644 --- a/apps/server/src/mcp/toolkits/worktree/handlers.ts +++ b/apps/server/src/mcp/toolkits/worktree/handlers.ts @@ -46,7 +46,7 @@ const handlers = { return yield* service.handoff(scope, input); }), ), - t3_thread_move: McpToolAccess.actsAsCaller((input) => + t3_thread_move: McpToolAccess.actsAsFullAccessCaller((input) => Effect.gen(function* () { const scope = yield* McpInvocationContext.McpInvocationContext; const service = yield* WorktreeMcpService.WorktreeMcpService; diff --git a/apps/server/src/mcp/toolkits/worktree/tools.ts b/apps/server/src/mcp/toolkits/worktree/tools.ts index 5cfe6df78a820..1a5487e1bb1f3 100644 --- a/apps/server/src/mcp/toolkits/worktree/tools.ts +++ b/apps/server/src/mcp/toolkits/worktree/tools.ts @@ -44,14 +44,14 @@ const WorktreeHandoffTool = Tool.make("t3_worktree_handoff", { const ThreadMoveTool = Tool.make("t3_thread_move", { description: - "Move THIS thread to another folder on this machine, keeping its conversation. Use it when the user asks to move or continue this thread in another folder or project. A separate thread: `t3_thread_launch`. Uses the project rooted at `path`, registering one if none exists. Files stay where they are; the branch and worktree are cleared. Call last in your turn: the move happens when the turn ends. `continuationPrompt` resumes work there next turn. Without it, the thread waits.", + "Move THIS thread to another folder when the user asks, keeping its conversation. A separate thread: `t3_thread_launch`. Uses or registers the project rooted at `path`. Files stay; branch and worktree are cleared. Call last: the move happens when the turn ends. Needs full access.", parameters: ThreadMoveMcpInput, success: ThreadMoveMcpResult, failure: WorktreeToolFailure, failureMode: "return", dependencies, }) - .annotate(Tool.Title, "Move thread to a folder") + .annotate(Tool.Title, "Move thread") .annotate(Tool.Readonly, false) .annotate(Tool.Destructive, false) .annotate(Tool.Idempotent, false) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 342f41719c040..14ce36d415844 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -728,13 +728,19 @@ it.effect.each([ let queuedRunBusy = true; const terminalRead = yield* Deferred.make(); const moveCommands: Array = []; + let afterMoveRuns = 0; const layerTest = ThreadManagementService.layer.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ getThreadEventSequence: () => Effect.sync(() => sequence), getThreadShell: () => - Effect.succeed({ id: threadId, projectId, activeRunId: runId } as never), + Effect.succeed({ + id: threadId, + projectId, + activeRunId: runId, + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + } as never), restartContinuationPending: () => Effect.succeed(false), streamStoredEventsFrom: (input) => input?.eventType === "run.updated" @@ -781,6 +787,7 @@ it.effect.each([ if (command.type === "message.dispatch") { expect(projectId).toBe(targetProjectId); expect(command.text).toBe("Carry on."); + expect(afterMoveRuns).toBe(1); yield* Deferred.succeed(continued, undefined); } return { sequence: 1, storedEvents: [] } as never; @@ -797,6 +804,9 @@ it.effect.each([ threadId, projectId: targetProjectId, commandId: CommandId.make("command:thread-management:move"), + afterMove: Effect.sync(() => { + afterMoveRuns += 1; + }), continuation: { commandId: CommandId.make("command:thread-management:move-continuation"), messageId: MessageId.make("message:thread-management:move-continuation"), @@ -816,6 +826,7 @@ it.effect.each([ yield* Deferred.await(terminalRead); if (!moves) { expect(moveCommands).toEqual([]); + expect(afterMoveRuns).toBe(0); return; } @@ -833,6 +844,125 @@ it.effect.each([ }), ); +it.effect.each([ + { name: "a refused move fails at once", moveFails: true, disabled: false, sendFails: false }, + { + name: "a disabled provider refuses the move", + moveFails: false, + disabled: true, + sendFails: false, + }, + { + name: "a failed continuation still reports the move", + moveFails: false, + disabled: false, + sendFails: true, + }, +] as const)("moveThreadToProject on an idle thread: $name", ({ moveFails, disabled, sendFails }) => + Effect.gen(function* () { + const targetProjectId = ProjectId.make("project:thread-management:idle-move-target"); + const threadId = ThreadId.make("thread:thread-management:idle-move"); + const instanceId = ProviderInstanceId.make("claudeAgent"); + let moves = 0; + let afterMoveRuns = 0; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + getThreadShell: () => + Effect.succeed({ + id: threadId, + projectId: ProjectId.make("project:thread-management:idle-move-source"), + activeRunId: null, + providerInstanceId: instanceId, + } as never), + // A retry stream would wait here forever. + streamStoredEventsFrom: () => Stream.never, + getThreadRecords: () => + Effect.succeed({ + thread: { + id: threadId, + projectId: targetProjectId, + archivedAt: null, + deletedAt: null, + modelSelection: { instanceId, model: "test" }, + }, + runs: [], + messages: [], + turnItems: [], + providerTurns: [], + } as unknown as OrchestrationV2ThreadProjection), + dispatch: (command) => { + if (command.type === "thread.project.move") { + moves += 1; + if (moveFails) { + return Effect.fail( + new Orchestrator.OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Thread still has work in progress.", + }), + ); + } + } + if (command.type === "message.dispatch" && sendFails) { + return Effect.fail( + new Orchestrator.OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Message refused.", + }), + ); + } + return Effect.succeed({ sequence: 1, storedEvents: [] } as never); + }, + }), + ), + Layer.provide( + ServerSettings.layerTest( + disabled + ? { + projectSettingsOverrides: { + [targetProjectId]: { providerInstances: { [instanceId]: { enabled: false } } }, + }, + } + : {}, + ), + ), + Layer.provide(layerReceiptsTest), + ); + + const exit = yield* Effect.gen(function* () { + const service = yield* ThreadManagementService.ThreadManagementService; + return yield* service.moveThreadToProject({ + threadId, + projectId: targetProjectId, + commandId: CommandId.make("command:thread-management:idle-move"), + afterMove: Effect.sync(() => { + afterMoveRuns += 1; + }), + continuation: { + commandId: CommandId.make("command:thread-management:idle-move-continuation"), + messageId: MessageId.make("message:thread-management:idle-move-continuation"), + text: "Carry on.", + }, + }); + }).pipe(Effect.provide(layerTest), Effect.exit); + + expect(moves).toBe(disabled ? 0 : 1); + expect(afterMoveRuns).toBe(moveFails || disabled ? 0 : 1); + if (moveFails || disabled) { + expect(exit._tag).toBe("Failure"); + return; + } + expect(exit).toMatchObject({ + _tag: "Success", + value: { movesWhenTurnEnds: false, continuationFailure: expect.any(String) }, + }); + }), +); + it.effect("waitForThread follows a thread that moves to another project", () => Effect.gen(function* () { const projectId = ProjectId.make("project:thread-management:wait-source"); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index ff79949f564c3..d74b11d76f3d1 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -458,13 +458,15 @@ export interface ThreadManagementServiceShape { readonly threadId: ThreadId; readonly projectId: ProjectId; readonly commandId: CommandId; + /** Runs once the move lands, before the continuation. */ + readonly afterMove: Effect.Effect; readonly continuation?: { readonly commandId: CommandId; readonly messageId: MessageId; readonly text: string; }; }) => Effect.Effect< - { readonly movesWhenTurnEnds: boolean }, + { readonly movesWhenTurnEnds: boolean; readonly continuationFailure?: string }, Orchestrator.OrchestratorV2Error | ThreadManagementFailure >; readonly settleThread: (input: { @@ -1135,25 +1137,69 @@ const make = Effect.gen(function* () { const moveThreadToProject: ThreadManagementServiceShape["moveThreadToProject"] = Effect.fn( "orchestrationV2.threadManagement.moveThreadToProject", )(function* (input) { - const apply = Effect.gen(function* () { + const dispatchMove = Effect.gen(function* () { + const sequence = yield* orchestrator.getThreadEventSequence(input.threadId); + return yield* dispatch({ + type: "thread.project.move", + commandId: CommandId.make(`${input.commandId}:${sequence}`), + threadId: input.threadId, + projectId: input.projectId, + }); + }); + const continueThere = + input.continuation === undefined + ? Effect.void + : sendToThread({ + projectId: input.projectId, + commandId: input.continuation.commandId, + threadId: input.threadId, + messageId: input.continuation.messageId, + text: input.continuation.text, + attachments: [], + mode: "queue", + createdBy: "agent", + creationSource: "mcp", + }).pipe(Effect.asVoid); + const shell = yield* orchestrator.getThreadShell(input.threadId); + if (shell == null) return yield* dispatchMove.pipe(Effect.as({ movesWhenTurnEnds: false })); + const instanceId = shell.providerInstanceId; + // Without its provider the thread could not run again in the destination. + const providerEnabled = isProviderInstanceEnabledForProject({ + projectId: input.projectId, + instanceId, + }); + if (!(yield* providerEnabled)) { + return yield* new ThreadManagementProviderInstanceDisabledError({ + projectId: input.projectId, + instanceId, + }); + } + if (shell.activeRunId === null) { + yield* dispatchMove; + yield* input.afterMove; + return yield* continueThere.pipe( + Effect.as({ movesWhenTurnEnds: false }), + // The move landed, so this call must not report that it failed. + Effect.catch((error) => + Effect.succeed({ movesWhenTurnEnds: false, continuationFailure: error.message }), + ), + ); + } + const { projectId, activeRunId } = shell; + const moveWhenIdle = Effect.gen(function* () { const afterSequence = yield* orchestrator.getThreadEventSequence(input.threadId); const tryMove = Effect.gen(function* () { - const sequence = yield* orchestrator.getThreadEventSequence(input.threadId); - return yield* dispatch({ - type: "thread.project.move", - commandId: CommandId.make(`${input.commandId}:${sequence}`), - threadId: input.threadId, - projectId: input.projectId, - }).pipe( - Effect.as(true), + if (!(yield* providerEnabled)) return "dropped" as const; + return yield* dispatchMove.pipe( + Effect.as("moved" as const), // A queued message may run first; the next run update tries again. Effect.catchTags({ - OrchestratorDispatchError: () => Effect.succeed(false), - OrchestratorCommandPreviouslyRejectedError: () => Effect.succeed(false), + OrchestratorDispatchError: () => Effect.succeed("retry" as const), + OrchestratorCommandPreviouslyRejectedError: () => Effect.succeed("retry" as const), }), ); }); - const moved = yield* Stream.fromEffect(tryMove).pipe( + const outcome = yield* Stream.fromEffect(tryMove).pipe( Stream.concat( orchestrator .streamStoredEventsFrom({ @@ -1163,31 +1209,20 @@ const make = Effect.gen(function* () { }) .pipe(Stream.mapEffect(() => tryMove)), ), - Stream.filter((done) => done), + Stream.filter((result) => result !== "retry"), Stream.runHead, Effect.timeoutOption(Duration.millis(SETTLE_AFTER_RUN_WAIT_MS)), ); - if (Option.isNone(moved) || Option.isNone(moved.value) || input.continuation === undefined) { - return; + if (Option.isNone(outcome) || Option.isNone(outcome.value)) return; + if (outcome.value.value === "dropped") { + return yield* Effect.logWarning( + "Dropped a thread move because the destination disabled the thread's provider", + { threadId: input.threadId, projectId: input.projectId, instanceId }, + ); } - yield* sendToThread({ - projectId: input.projectId, - commandId: input.continuation.commandId, - threadId: input.threadId, - messageId: input.continuation.messageId, - text: input.continuation.text, - attachments: [], - mode: "queue", - createdBy: "agent", - creationSource: "mcp", - }); + yield* input.afterMove; + yield* continueThere; }); - const shell = yield* orchestrator.getThreadShell(input.threadId); - if (shell == null || shell.activeRunId === null) { - yield* apply; - return { movesWhenTurnEnds: false }; - } - const { projectId, activeRunId } = shell; yield* waitForThread({ projectId, threadId: input.threadId, @@ -1196,7 +1231,7 @@ const make = Effect.gen(function* () { }).pipe( Effect.flatMap(({ run }) => run?.status === "completed" - ? apply + ? moveWhenIdle : Effect.logInfo("Dropped a thread move because its turn did not complete", { threadId: input.threadId, runId: activeRunId, diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index ac8a567fd2b12..05d9ce99c83fe 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5366, "codexAdditionalContextBytes": 7096, - "mcpToolListJsonBytes": 48783 + "mcpToolListJsonBytes": 48560 } diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 82af595efc908..a549b811b6aff 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -36,7 +36,7 @@ To move a thread that has already started, ask its agent to move to the folder you name. The conversation carries over and the thread lands in that folder's project, registering one if there is none. The move happens when the turn ends; a failed or stopped turn drops it. Files stay where they are, and the thread's -branch and worktree are cleared. +branch and worktree are cleared. The agent needs full access to do this. Each thread without a project works in its own folder under `~/.t3/scratch` (the `scratch` folder of your T3 data directory), named after its date, the first words diff --git a/packages/contracts/src/worktreeMcp.ts b/packages/contracts/src/worktreeMcp.ts index 9f3dbaed4edd0..0d7fb3e94527d 100644 --- a/packages/contracts/src/worktreeMcp.ts +++ b/packages/contracts/src/worktreeMcp.ts @@ -77,11 +77,11 @@ export type WorktreeMcpHandoffResult = typeof WorktreeMcpHandoffResult.Type; /** Keeps the conversation; the move waits for this turn to end, and the next turn resumes in the folder. */ export const ThreadMoveMcpInput = Schema.Struct({ path: TrimmedNonEmptyString.annotate({ - description: "Absolute path of an existing folder on this machine.", + description: "Absolute path of an existing folder.", }), continuationPrompt: Schema.optional( TrimmedNonEmptyString.check(Schema.isMaxLength(120_000)).annotate({ - description: "Work to resume in the folder next turn. Omitted: thread waits.", + description: "Work to resume there. Omitted: thread waits.", }), ), }); From 5e7c4881e1eaca4e2ec579101769cc8a2895cea1 Mon Sep 17 00:00:00 2001 From: giri256 <80974392+giri256@users.noreply.github.com> Date: Sat, 7 Nov 2026 03:35:10 +0000 Subject: [PATCH 3/6] fix(server): allow one pending move per thread A second move in the same turn would race the first for the thread and could queue both continuations, so it is now refused until the first ends. A missing thread shell no longer skips the post-move steps, and a deferred move that times out now logs a warning. Model: Claude Opus 5.5, harness: Claude Code in T3 Code. --- .../ThreadManagementService.test.ts | 51 ++++++++++------- .../ThreadManagementService.ts | 55 ++++++++++++++----- 2 files changed, 71 insertions(+), 35 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 14ce36d415844..67c92408ac606 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -817,6 +817,18 @@ it.effect.each([ yield* Deferred.await(subscribed); expect(moveCommands).toEqual([]); + const second = yield* service + .moveThreadToProject({ + threadId, + projectId: ProjectId.make("project:thread-management:move-other"), + commandId: CommandId.make("command:thread-management:move-second"), + afterMove: Effect.void, + }) + .pipe(Effect.flip); + expect(second).toMatchObject({ + cause: `Thread ${threadId} already has a move pending.`, + }); + runs[0]!.status = status; sequence = 1; yield* Queue.offer(events, { @@ -845,21 +857,16 @@ it.effect.each([ ); it.effect.each([ - { name: "a refused move fails at once", moveFails: true, disabled: false, sendFails: false }, - { - name: "a disabled provider refuses the move", - moveFails: false, - disabled: true, - sendFails: false, - }, - { - name: "a failed continuation still reports the move", - moveFails: false, - disabled: false, - sendFails: true, - }, -] as const)("moveThreadToProject on an idle thread: $name", ({ moveFails, disabled, sendFails }) => + { name: "a refused move fails at once", moveFails: true }, + { name: "a disabled provider refuses the move", disabled: true }, + { name: "a failed continuation still reports the move", sendFails: true }, + { name: "a missing shell still runs the post-move steps", shellMissing: true, sendFails: true }, +] as const)("moveThreadToProject on an idle thread: $name", (row) => Effect.gen(function* () { + const moveFails = "moveFails" in row; + const disabled = "disabled" in row; + const sendFails = "sendFails" in row; + const shellMissing = "shellMissing" in row; const targetProjectId = ProjectId.make("project:thread-management:idle-move-target"); const threadId = ThreadId.make("thread:thread-management:idle-move"); const instanceId = ProviderInstanceId.make("claudeAgent"); @@ -871,12 +878,16 @@ it.effect.each([ Layer.mock(Orchestrator.OrchestratorV2)({ getThreadEventSequence: () => Effect.succeed(0), getThreadShell: () => - Effect.succeed({ - id: threadId, - projectId: ProjectId.make("project:thread-management:idle-move-source"), - activeRunId: null, - providerInstanceId: instanceId, - } as never), + Effect.succeed( + (shellMissing + ? null + : { + id: threadId, + projectId: ProjectId.make("project:thread-management:idle-move-source"), + activeRunId: null, + providerInstanceId: instanceId, + }) as never, + ), // A retry stream would wait here forever. streamStoredEventsFrom: () => Stream.never, getThreadRecords: () => diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index d74b11d76f3d1..4a32948b7dd18 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1134,6 +1134,9 @@ const make = Effect.gen(function* () { ); }); + // Held until the deferred move's fiber ends; two pending moves would race for the thread. + const threadsWithPendingMove = new Set(); + const moveThreadToProject: ThreadManagementServiceShape["moveThreadToProject"] = Effect.fn( "orchestrationV2.threadManagement.moveThreadToProject", )(function* (input) { @@ -1160,8 +1163,25 @@ const make = Effect.gen(function* () { createdBy: "agent", creationSource: "mcp", }).pipe(Effect.asVoid); + const movePending = new Orchestrator.OrchestratorDispatchError({ + commandId: input.commandId, + commandType: "thread.project.move", + cause: `Thread ${input.threadId} already has a move pending.`, + }); + const moveNow = Effect.gen(function* () { + if (threadsWithPendingMove.has(input.threadId)) return yield* movePending; + yield* dispatchMove; + yield* input.afterMove; + return yield* continueThere.pipe( + Effect.as({ movesWhenTurnEnds: false }), + // The move landed, so this call must not report that it failed. + Effect.catch((error) => + Effect.succeed({ movesWhenTurnEnds: false, continuationFailure: error.message }), + ), + ); + }); const shell = yield* orchestrator.getThreadShell(input.threadId); - if (shell == null) return yield* dispatchMove.pipe(Effect.as({ movesWhenTurnEnds: false })); + if (shell == null) return yield* moveNow; const instanceId = shell.providerInstanceId; // Without its provider the thread could not run again in the destination. const providerEnabled = isProviderInstanceEnabledForProject({ @@ -1174,17 +1194,7 @@ const make = Effect.gen(function* () { instanceId, }); } - if (shell.activeRunId === null) { - yield* dispatchMove; - yield* input.afterMove; - return yield* continueThere.pipe( - Effect.as({ movesWhenTurnEnds: false }), - // The move landed, so this call must not report that it failed. - Effect.catch((error) => - Effect.succeed({ movesWhenTurnEnds: false, continuationFailure: error.message }), - ), - ); - } + if (shell.activeRunId === null) return yield* moveNow; const { projectId, activeRunId } = shell; const moveWhenIdle = Effect.gen(function* () { const afterSequence = yield* orchestrator.getThreadEventSequence(input.threadId); @@ -1213,7 +1223,13 @@ const make = Effect.gen(function* () { Stream.runHead, Effect.timeoutOption(Duration.millis(SETTLE_AFTER_RUN_WAIT_MS)), ); - if (Option.isNone(outcome) || Option.isNone(outcome.value)) return; + if (Option.isNone(outcome) || Option.isNone(outcome.value)) { + return yield* Effect.logWarning("Dropped a thread move that never found the thread idle", { + threadId: input.threadId, + projectId: input.projectId, + instanceId, + }); + } if (outcome.value.value === "dropped") { return yield* Effect.logWarning( "Dropped a thread move because the destination disabled the thread's provider", @@ -1223,7 +1239,7 @@ const make = Effect.gen(function* () { yield* input.afterMove; yield* continueThere; }); - yield* waitForThread({ + const waitThenMove = waitForThread({ projectId, threadId: input.threadId, runId: activeRunId, @@ -1239,7 +1255,16 @@ const make = Effect.gen(function* () { }), ), Effect.ignoreCause({ log: true }), - Effect.forkIn(layerScope), + Effect.interruptible, + Effect.ensuring(Effect.sync(() => threadsWithPendingMove.delete(input.threadId))), + ); + // Uninterruptible so cancellation cannot hold the guard with no fiber to release it. + yield* Effect.uninterruptible( + Effect.suspend(() => { + if (threadsWithPendingMove.has(input.threadId)) return Effect.fail(movePending); + threadsWithPendingMove.add(input.threadId); + return waitThenMove.pipe(Effect.forkIn(layerScope)); + }), ); return { movesWhenTurnEnds: true }; }); From 4bb743f742de0cdd7bb8d52d3a232f2f7a636a45 Mon Sep 17 00:00:00 2001 From: giri256 <80974392+giri256@users.noreply.github.com> Date: Sat, 7 Nov 2026 04:33:35 +0000 Subject: [PATCH 4/6] fix(server): settle a thread that moved at the end of its turn A settle and a move requested in the same turn wake together. If the move landed first, the settle's project-scoped reads failed and the settle was lost, so it now follows the thread as the wait does. Model: Claude Opus 5.5, harness: Claude Code in T3 Code. --- .../ThreadManagementService.test.ts | 17 ++++++++++++++--- .../orchestration-v2/ThreadManagementService.ts | 5 +++-- 2 files changed, 17 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 67c92408ac606..320e03ce510b1 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -563,6 +563,7 @@ it.effect.each([ { status: "failed" as const, followUp: "none", settles: false }, { status: "interrupted" as const, followUp: "none", settles: false }, { status: "completed" as const, followUp: "notification", settles: true }, + { status: "completed" as const, followUp: "moved", settles: true }, { status: "completed" as const, followUp: "replayed-notification", settles: true }, { status: "completed" as const, followUp: "delegated", settles: true }, { status: "completed" as const, followUp: "failed-notification", settles: false }, @@ -575,11 +576,13 @@ it.effect.each([ const threadId = ThreadId.make("thread:thread-management:settle-after-run"); const runId = RunId.make("run:thread-management:settle-after-run"); const followUpId = RunId.make(`${runId}:follow-up`); + const hasFollowUp = followUp !== "none" && followUp !== "moved"; const subscribed = yield* Deferred.make(); const events = yield* Queue.unbounded(); const replayRejected = yield* Deferred.make(); let sequence = 0; let sequenceReads = 0; + let recordReads = 0; let rejectedCommandId: CommandId | undefined; const runs = [{ id: runId, ordinal: 1, status, userMessageId: MessageId.make("original") }]; const nextRun = { @@ -610,7 +613,15 @@ it.effect.each([ Effect.sync( () => ({ - thread: { id: threadId, projectId, deletedAt: null }, + thread: { + id: threadId, + // A deferred move can land right after the wait's entry check. + projectId: + followUp === "moved" && recordReads++ > 0 + ? ProjectId.make("project:thread-management:settle-moved") + : projectId, + deletedAt: null, + }, runs: raced ? [...runs, nextRun] : runs, messages: [ { id: MessageId.make("original"), runId, role: "user" as const }, @@ -644,7 +655,7 @@ it.effect.each([ ), dispatch: (command) => Effect.gen(function* () { - if (followUp !== "none" && !raced) { + if (hasFollowUp && !raced) { raced = true; rejectedCommandId = command.commandId; return yield* new Orchestrator.OrchestratorDispatchError({ @@ -675,7 +686,7 @@ it.effect.each([ const fiber = yield* service .settleAfterRun({ projectId, threadId, runId }) .pipe(Effect.forkChild); - if (followUp !== "none") { + if (hasFollowUp) { yield* Deferred.await(subscribed); expect(dispatched).toEqual([]); nextRun.status = diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 4a32948b7dd18..bb7d16c6b1c90 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1067,10 +1067,11 @@ const make = Effect.gen(function* () { const afterSequence = yield* orchestrator.getThreadEventSequence(input.threadId); const trySettle = Effect.gen(function* () { const sequence = yield* orchestrator.getThreadEventSequence(input.threadId); - const records = yield* getProjectThreadRecords(input, ["runs"]); + // A deferred move may land as this wakes; settling follows the thread like the wait does. + const records = yield* readThreadRecordsInAnyProject(input, ["runs"]); const following = records.runs.filter((candidate) => candidate.ordinal > run.ordinal); if (following.length > 0) { - const { messages } = yield* getProjectThreadRecords(input, ["messages"], { + const { messages } = yield* readThreadRecordsInAnyProject(input, ["messages"], { messageRunIds: following.map((candidate) => candidate.id), messageRoles: ["user"], }); From d67ab963e655ade6270c84a444e60c30ac3c1707 Mon Sep 17 00:00:00 2001 From: giri256 <80974392+giri256@users.noreply.github.com> Date: Sat, 7 Nov 2026 04:58:19 +0000 Subject: [PATCH 5/6] fix(server): let a settle start after its thread moved If the move landed before the settle's first read, the wait rejected the old project and the settle was lost. The settle was authorized when it was requested, so its first read now follows the thread. Public waits keep their project check. Model: Claude Opus 5.5, harness: Claude Code in T3 Code. --- .../ThreadManagementService.test.ts | 5 ++--- .../ThreadManagementService.ts | 18 +++++++++++++----- 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 320e03ce510b1..8ad8cce8025d2 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -582,7 +582,6 @@ it.effect.each([ const replayRejected = yield* Deferred.make(); let sequence = 0; let sequenceReads = 0; - let recordReads = 0; let rejectedCommandId: CommandId | undefined; const runs = [{ id: runId, ordinal: 1, status, userMessageId: MessageId.make("original") }]; const nextRun = { @@ -615,9 +614,9 @@ it.effect.each([ ({ thread: { id: threadId, - // A deferred move can land right after the wait's entry check. + // A deferred move can land before the settle's first read. projectId: - followUp === "moved" && recordReads++ > 0 + followUp === "moved" ? ProjectId.make("project:thread-management:settle-moved") : projectId, deletedAt: null, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index bb7d16c6b1c90..5be8368db885f 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -919,9 +919,10 @@ const make = Effect.gen(function* () { return { dispatch: dispatched, projection, message, run, turnItem, delivery }; }); - const waitForThread: ThreadManagementServiceShape["waitForThread"] = Effect.fn( - "ThreadManagementService.waitForThread", - )(function* (input) { + const waitForThreadReadingFirst = Effect.fn("ThreadManagementService.waitForThread")(function* ( + input: ThreadManagementWaitInput, + readFirst: ThreadManagementServiceShape["getProjectThreadRecords"], + ) { const loadError = (cause: unknown) => new ThreadManagementProjectionLoadError({ projectId: input.projectId, @@ -933,7 +934,7 @@ const make = Effect.gen(function* () { const afterSequence = yield* orchestrator .getThreadEventSequence(input.threadId) .pipe(Effect.mapError(loadError)); - const target = yield* getProjectThreadRecords(input, ["runs"]); + const target = yield* readFirst(input, ["runs"]); const selectedRun = input.runId === undefined ? latestRun(target) @@ -1059,10 +1060,17 @@ const make = Effect.gen(function* () { ); }); + const waitForThread: ThreadManagementServiceShape["waitForThread"] = (input) => + waitForThreadReadingFirst(input, getProjectThreadRecords); + const settleAfterRun: ThreadManagementServiceShape["settleAfterRun"] = Effect.fn( "orchestrationV2.threadManagement.settleAfterRun", )(function* (input) { - const { run } = yield* waitForThread({ ...input, timeoutMs: SETTLE_AFTER_RUN_WAIT_MS }); + // The settle was authorized when requested; a move can land before the first read. + const { run } = yield* waitForThreadReadingFirst( + { ...input, timeoutMs: SETTLE_AFTER_RUN_WAIT_MS }, + readThreadRecordsInAnyProject, + ); if (run?.status !== "completed") return; const afterSequence = yield* orchestrator.getThreadEventSequence(input.threadId); const trySettle = Effect.gen(function* () { From fe13e06a6a0cabbcceebf7c35d04fc0edd7ef869 Mon Sep 17 00:00:00 2001 From: giri256 <80974392+giri256@users.noreply.github.com> Date: Sat, 7 Nov 2026 09:33:34 +0000 Subject: [PATCH 6/6] fix(server): hold the thread during an immediate move Only deferred moves held the per-thread guard, so two immediate moves could both commit and leave scheduled tasks in a different project from the thread. Immediate moves now hold the guard too and release it on every exit. Model: Claude Opus 5.5, harness: Claude Code in T3 Code. --- .../ThreadManagementService.test.ts | 59 +++++++++++++++++++ .../ThreadManagementService.ts | 43 ++++++++------ 2 files changed, 84 insertions(+), 18 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 8ad8cce8025d2..97b0036e89ab2 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -984,6 +984,65 @@ it.effect.each([ }), ); +it.effect("moveThreadToProject holds the thread during an immediate move", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread:thread-management:concurrent-move"); + const entered = yield* Deferred.make(); + const gate = yield* Deferred.make(); + const moved: Array = []; + let moveCalls = 0; + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide( + ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + getThreadShell: () => + Effect.succeed({ + id: threadId, + projectId: ProjectId.make("project:thread-management:concurrent-source"), + activeRunId: null, + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + } as never), + dispatch: () => + Effect.gen(function* () { + moveCalls += 1; + if (moveCalls === 1) { + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(gate); + } + return { sequence: 1, storedEvents: [] } as never; + }), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ), + ), + ); + const move = (destination: string) => + service.moveThreadToProject({ + threadId, + projectId: ProjectId.make(destination), + commandId: CommandId.make(`command:thread-management:${destination}`), + afterMove: Effect.sync(() => { + moved.push(destination); + }), + }); + + const first = yield* move("first").pipe(Effect.forkChild); + yield* Deferred.await(entered); + const second = yield* move("second").pipe(Effect.flip); + expect(second).toMatchObject({ cause: `Thread ${threadId} already has a move pending.` }); + yield* Deferred.succeed(gate, undefined); + yield* Fiber.join(first); + + yield* move("third"); + expect(moved).toEqual(["first", "third"]); + }), +); + it.effect("waitForThread follows a thread that moves to another project", () => Effect.gen(function* () { const projectId = ProjectId.make("project:thread-management:wait-source"); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 5be8368db885f..980c9043a9b57 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1143,7 +1143,7 @@ const make = Effect.gen(function* () { ); }); - // Held until the deferred move's fiber ends; two pending moves would race for the thread. + // Held until a move finishes, including a deferred move's fiber; two moves would race for the thread. const threadsWithPendingMove = new Set(); const moveThreadToProject: ThreadManagementServiceShape["moveThreadToProject"] = Effect.fn( @@ -1177,18 +1177,29 @@ const make = Effect.gen(function* () { commandType: "thread.project.move", cause: `Thread ${input.threadId} already has a move pending.`, }); - const moveNow = Effect.gen(function* () { - if (threadsWithPendingMove.has(input.threadId)) return yield* movePending; - yield* dispatchMove; - yield* input.afterMove; - return yield* continueThere.pipe( - Effect.as({ movesWhenTurnEnds: false }), - // The move landed, so this call must not report that it failed. - Effect.catch((error) => - Effect.succeed({ movesWhenTurnEnds: false, continuationFailure: error.message }), - ), - ); + const reserveThread = Effect.suspend(() => { + if (threadsWithPendingMove.has(input.threadId)) return Effect.fail(movePending); + threadsWithPendingMove.add(input.threadId); + return Effect.void; }); + const releaseThread = Effect.sync(() => threadsWithPendingMove.delete(input.threadId)); + const moveNow = Effect.acquireUseRelease( + reserveThread, + () => + dispatchMove.pipe( + Effect.andThen(input.afterMove), + Effect.andThen( + continueThere.pipe( + Effect.as({ movesWhenTurnEnds: false }), + // The move landed, so this call must not report that it failed. + Effect.catch((error) => + Effect.succeed({ movesWhenTurnEnds: false, continuationFailure: error.message }), + ), + ), + ), + ), + () => releaseThread, + ); const shell = yield* orchestrator.getThreadShell(input.threadId); if (shell == null) return yield* moveNow; const instanceId = shell.providerInstanceId; @@ -1265,15 +1276,11 @@ const make = Effect.gen(function* () { ), Effect.ignoreCause({ log: true }), Effect.interruptible, - Effect.ensuring(Effect.sync(() => threadsWithPendingMove.delete(input.threadId))), + Effect.ensuring(releaseThread), ); // Uninterruptible so cancellation cannot hold the guard with no fiber to release it. yield* Effect.uninterruptible( - Effect.suspend(() => { - if (threadsWithPendingMove.has(input.threadId)) return Effect.fail(movePending); - threadsWithPendingMove.add(input.threadId); - return waitThenMove.pipe(Effect.forkIn(layerScope)); - }), + reserveThread.pipe(Effect.andThen(waitThenMove.pipe(Effect.forkIn(layerScope)))), ); return { movesWhenTurnEnds: true }; });