# Files changed · PR \#16654 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/16654/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [feat\(server\): t3_thread_launch takes clientRequestId](https://git.cafe/pingdotgg/t3code/pull/16654.md) Version: 24 Head commit: 071e6616495613844a1ff062446e41b4fc586d22 Base commit: d868b6296c8a24bbc150ebb4d96d264316206099 Comparison base: d868b6296c8a24bbc150ebb4d96d264316206099 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/mcp/toolkits/project/handlers.test.ts Status: modified; +131; −3 ```diff @@ -6,6 +6,7 @@ type Project as ProjectRecord, type ModelSelection, type RuntimeMode, + type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -43,6 +44,10 @@ instanceId: ProviderInstanceId.make("claudeAgent"), model: "claude-opus", }; + // The project's own default model, which a test can clear between launches. + const projectDefaults: { model: ModelSelection | null } = { + model: input.overrideModel === undefined ? modelSelection : null, + }; const layerDependencies = Layer.mergeAll( ServerSettings.layerTest({ ...(input.defaultRuntimeMode === undefined @@ -70,7 +75,22 @@ issuedAt: 0, capabilities: new Set(["orchestration" as const]), }), + Layer.mock(ThreadManagement.ThreadManagementService)({ + // A thread exists once a launch started it. + getThreadShell: (threadId) => { + const launch = input.launched.find((candidate) => candidate.threadId === threadId); - Layer.mock(ThreadManagement.ThreadManagementService)({}), + return Effect.succeed( + launch === undefined + ? null + : ({ + id: launch.threadId, + projectId: launch.projectId, + modelSelection: launch.modelSelection, + deletedAt: null, + } as unknown as OrchestrationV2ThreadShell), + ); + }, + }), Layer.mock(ThreadLaunch.ThreadLaunchService)({ checkWorktreeBase: input.checkWorktreeBase ?? (() => Effect.void), launch: (launch) => { @@ -96,7 +116,7 @@ ? Option.some({ id, workspaceRoot: input.workspaceRoot ?? "/projects/client-target", - defaultModelSelection: input.overrideModel === undefined ? modelSelection : null, + defaultModelSelection: projectDefaults.model, } as unknown as ProjectRecord) : Option.none(), ), @@ -113,7 +133,7 @@ ), Layer.provideMerge(NodeServices.layer), ); - return { projectId, modelSelection, dependencies: layerDependencies }; + return { projectId, modelSelection, projectDefaults, dependencies: layerDependencies }; }; it.effect( @@ -197,6 +217,71 @@ }), ); +it.effect("a launch retried with the same clientRequestId replays the first one", () => + Effect.gen(function* () { + const launched: Array = []; + const { projectId, dependencies } = clientLaunchHarness({ + runtimeModeCeiling: "auto", + launched, + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe(Layer.provide(dependencies)), + ), + ); + const launch = (clientRequestId: string | undefined) => + toolkit + .handle("t3_thread_launch", { + title: "Fix", + projectId, + message: "Fix the bug", + ...(clientRequestId === undefined ? {} : { clientRequestId }), + }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); + + yield* launch("fix-1"); + yield* launch("fix-1"); + yield* launch("fix-2"); + yield* launch(undefined); + yield* launch(undefined); + const ids = launched.map((input) => [input.commandId, input.threadId]); + // The same key sends the same command and thread ids, which the launch + // service replays instead of starting a second thread. + expect(ids[0]).toEqual(ids[1]); + expect(ids[0]?.[0]).toContain("client%3Asession-1"); + expect(ids[2]).not.toEqual(ids[0]); + // Without a key every launch is new. + expect(ids[3]).not.toEqual(ids[4]); + }), +); + +it.effect("a retried launch replays after the project loses its default model", () => + Effect.gen(function* () { + const launched: Array = []; + const { projectId, modelSelection, projectDefaults, dependencies } = clientLaunchHarness({ + runtimeModeCeiling: "auto", + launched, + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe(Layer.provide(dependencies)), + ), + ); + const launch = toolkit + .handle("t3_thread_launch", { title: "Fix", projectId, clientRequestId: "fix-1" }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); + + yield* launch; + projectDefaults.model = null; + const retried = (yield* launch).at(-1)?.result; + expect(retried).toMatchObject({ projectId, modelSelection }); + expect(launched.map((input) => input.threadId)).toEqual([ + launched[0]?.threadId, + launched[0]?.threadId, + ]); + }), +); + it.effect("a launch binds only an existing checkout that is one of the project's worktrees", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -299,3 +384,46 @@ expect(launched).toHaveLength(1); }), ); + +it.effect("a retried worktree launch replays after its base ref is gone", () => + Effect.gen(function* () { + const launched: Array = []; + const base = { exists: true }; + const { projectId, dependencies } = clientLaunchHarness({ + runtimeModeCeiling: "auto", + launched, + checkWorktreeBase: ({ projectId, workspaceStrategy }) => + base.exists || workspaceStrategy.type !== "worktree" + ? Effect.void + : Effect.fail( + new ThreadLaunch.ThreadLaunchBaseRefError({ + projectId, + projectTitle: "Client target", + workspaceRoot: "/projects/client-target", + baseRef: workspaceStrategy.baseRef, + }), + ), + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe(Layer.provide(dependencies)), + ), + ); + const launch = toolkit + .handle("t3_thread_launch", { + title: "Fix", + projectId, + clientRequestId: "fix-worktree", + workspaceStrategy: { type: "worktree", baseRef: "feature", startFromOrigin: false }, + }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); + + yield* launch; + base.exists = false; + expect((yield* launch).at(-1)?.result).toMatchObject({ projectId }); + expect(launched.map((input) => input.threadId)).toEqual([ + launched[0]?.threadId, + launched[0]?.threadId, + ]); + }), +); ``` ## apps/server/src/mcp/toolkits/project/handlers.ts Status: modified; +31; −6 ```diff @@ -1,4 +1,10 @@ +import { + CommandId, + MessageId, -import { MessageId, ThreadId, OrchestratorMcpFailure, ProjectId } from "@t3tools/contracts"; + ThreadId, + OrchestratorMcpFailure, + ProjectId, +} from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; @@ -59,8 +65,16 @@ (input, { runtimeMode, interactionMode }) => Effect.gen(function* () { const context = yield* readCaller(); + const { caller, scope } = context; + // A retry with the same key replays the first launch: the ids derive + // from the caller and the key, and the launch service replays a + // command it already accepted. + const commandId = + input.clientRequestId === undefined + ? yield* newCommandId() - const { caller } = context; + : CommandId.make( + `mcp:launch:${encodeURIComponent(scope.requestNamespace)}:${encodeURIComponent(input.clientRequestId)}`, + ); - const commandId = yield* newCommandId(); const threadId = ThreadId.make(commandId); const messageId = MessageId.make(commandId); const attachments = input.attachments ?? []; @@ -96,7 +110,17 @@ Effect.mapError(unavailable), Effect.map(Option.getOrUndefined), ); + // A retry of an accepted launch replays it: it keeps the model it started + // with and skips the checks only a new launch needs, which the project + // may no longer pass, such as a base branch deleted since. + const launched = - if (input.workspaceStrategy?.type === "existing_worktree") { + input.clientRequestId === undefined + ? null + : yield* context.threads.getThreadShell(threadId).pipe(Effect.mapError(unavailable)); + const launchedModel = + launched === null || launched.deletedAt !== null ? undefined : launched.modelSelection; + const replaying = launchedModel !== undefined; + if (!replaying && input.workspaceStrategy?.type === "existing_worktree") { const project = yield* readProject; if (project === undefined) return yield* new OrchestratorMcpFailure({ @@ -108,7 +132,7 @@ // A launch is accepted before its worktree is provisioned, so a base // ref that cannot resolve is refused while the agent can still act on it. const strategy = input.workspaceStrategy; + if (!replaying && strategy?.type === "worktree") - if (strategy?.type === "worktree") yield* ThreadLaunch.ThreadLaunchService.pipe( Effect.flatMap((launches) => launches.checkWorktreeBase({ projectId, workspaceStrategy: strategy }), @@ -129,7 +153,7 @@ // project's effective defaults, as the composer does. Its mode stays // within the ceiling it was approved with. const projectDefaults = - caller === undefined + caller === undefined && !replaying ? yield* ServerSettings.ServerSettingsService.pipe( Effect.flatMap((settings) => settings.getSettings), Effect.mapError(unavailable), @@ -143,6 +167,7 @@ : undefined; const modelSelection = input.modelSelection ?? + launchedModel ?? caller?.modelSelection ?? projectDefaults?.defaultModelSelection ?? undefined; ``` ## apps/server/src/mcp/toolkits/project/tools.ts Status: modified; +3; −1 ```diff @@ -13,6 +13,7 @@ ProjectCreatePayload, ProjectUpdatePayload, ProjectId, + OrchestratorMcpClientRequestId, OrchestratorMcpFailure, SourceControlCloneRepositoryInput, SourceControlCloneRepositoryResult, @@ -104,7 +105,7 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", { ...shared, description: - 'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. To fetch and start from the base branch\'s remote, set startFromOrigin:true. Omitted workspaceStrategy means the project root, NOT the caller\'s worktree. Omit projectId/modelSelection/modes to inherit those settings from the calling thread; a caller outside a T3 thread must pass projectId and gets the project\'s default model. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Each call creates a new launch with no retry key; retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. After errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with.', + 'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. To fetch and start from the base branch\'s remote, set startFromOrigin:true. Omitted workspaceStrategy means the project root, NOT the caller\'s worktree. Omit projectId/modelSelection/modes to inherit those settings from the calling thread; a caller outside a T3 thread must pass projectId and gets the project\'s default model. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Pass clientRequestId to retry safely: the same key returns the thread the first call created instead of launching another. Retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. Without a key, after errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with.', parameters: Schema.Struct({ projectId: Schema.optional(ProjectId), scratch: Schema.optional( @@ -130,6 +131,7 @@ }), ), attachments: Schema.optional(Schema.Array(McpAttachmentInput).check(Schema.isMaxLength(8))), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), }), success: Schema.Struct({ threadId: ThreadId, ``` ## apps/server/src/orchestration-v2/ThreadLaunchService.test.ts Status: modified; +34; −0 ```diff @@ -1423,6 +1423,40 @@ }).pipe(Effect.provide(harness.layer)); }); +it.effect("an agent's launch never starts work above the modes it asks for", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + // The first launch, at full access, creates the thread without a message. + const created = yield* launches.launch( + launchInput({ command: "command:launch:narrowed-retry", thread: "thread:narrowed" }), + ); + // The same key comes back from a narrower agent, now carrying a message, + // as a retry or a concurrent launch that lost the create would. + const failed = yield* launches + .launch({ + ...launchInput({ + command: "command:launch:narrowed-retry", + thread: "thread:narrowed", + message: "Run this", + }), + runtimeMode: "approval-required", + interactionMode: "plan", + createdBy: "agent", + creationSource: "mcp", + }) + .pipe(Effect.flip); + if (failed._tag !== "ThreadLaunchError") { + assert.fail(`expected ThreadLaunchError, got ${failed._tag}`); + } + assert.equal(failed.operation, "dispatch-message"); + const projection = yield* threads.getThreadProjection(created.threadId); + assert.equal(projection.runs.length, 0); + assert.equal(projection.messages.length, 0); + }).pipe(Effect.provide(harness.layer)); +}); + it.effect("rejects a server-allocated launch replay with a mismatching thread id", () => { const harness = makeHarness(); return Effect.gen(function* () { ``` ## apps/server/src/orchestration-v2/ThreadLaunchService.ts Status: modified; +15; −2 ```diff @@ -49,6 +49,7 @@ import type * as Orchestrator from "./Orchestrator.ts"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid"; +import { DispatchModeLimit } from "./DispatchModeLimit.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; export type ThreadLaunchWorkspaceStrategy = @@ -940,7 +941,6 @@ if (project.id !== input.projectId) { return yield* mapError(input, "resolve-project", threadId)("Project identity changed."); } - let runId: RunId | null = null; let messageWasAlreadyAccepted = false; if (input.initialMessage !== undefined) { @@ -973,7 +973,20 @@ createdBy: input.createdBy, creationSource: input.creationSource, }) + .pipe( + // An agent's launch may land on a thread a replay, or a + // concurrent launch with the same key, created with broader modes + // than it asks for. The orchestrator refuses the message under + // the thread's lock. The user's own clients launch into their + // own drafts, whatever modes those hold. + input.createdBy === "agent" - .pipe(Effect.mapError(mapError(input, "dispatch-message", threadId))); + ? Effect.provideService(DispatchModeLimit, { + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + }) + : (effect) => effect, + Effect.mapError(mapError(input, "dispatch-message", threadId)), + ); const runCreated = dispatched.storedEvents.find( (stored) => stored.event.type === "run.created", ); ``` ## packages/contracts/src/orchestratorMcp.ts Status: modified; +1; −1 ```diff @@ -46,7 +46,7 @@ const OrchestratorMcpTitle = TrimmedNonEmptyString.check(Schema.isMaxLength(512)).annotate({ description: "Optional concise display title.", }); -const OrchestratorMcpClientRequestId = TrimmedNonEmptyString.check( +export const OrchestratorMcpClientRequestId = TrimmedNonEmptyString.check( Schema.isMaxLength(256), ).annotate({ description: "Stable idempotency key to reuse when retrying this mutation." }); ``` ## packages/provider-core/src/server/orchestrationInstructions.ts Status: modified; +1; −1 ```diff @@ -28,7 +28,7 @@ \`t3_thread_launch\` is the single-thread launch tool. Use \`create_threads\` only for a batch of threads intentionally sharing the caller's checkout: it always inherits the caller's project, branch, and worktree and has no workspace override. Asking an agent to run \`git worktree add\` or \`cd\` in its prompt does not update T3's thread binding. Select the workspace in the launch call instead. \`t3_worktree_handoff\` moves the calling thread, not another thread, and cannot move a thread already attached to a worktree. -\`t3_thread_launch\` has no idempotency key. Retain its returned threadId and inspect it with \`t3_thread_read\` / \`t3_thread_wait\`; preparation can still be running after acceptance. If a launch fails or its response is lost, inspect \`t3_thread_list\` before retrying, since a thread may already exist.`, +Pass a \`clientRequestId\` to \`t3_thread_launch\` and reuse it when retrying: the same key returns the thread the first call created. Retain its returned threadId and inspect it with \`t3_thread_read\` / \`t3_thread_wait\`; preparation can still be running after acceptance.`, `Tool names may include a harness-normalized MCP prefix, such as \`mcp__t3_code__delegate_task\`; the semantics are the same. Some harnesses attach optional MCP servers lazily: if an initial tool-catalog scan does not show T3 tools, do not conclude that cross-provider delegation is unavailable. Make one bounded direct attempt using the known T3 tool name on the next tool step. In Codex code mode, for example, call \`tools.mcp__t3_code__orchestrator_capabilities({})\` before reporting that the capability is absent. Keep polling/wait loops bounded, do not duplicate active work, and use stable \`clientRequestId\` values when retrying tools that accept them. ACP fallback: some ACP agents accept the injected MCP server but fail to expose its tools. When the T3 tools are absent and \`T3_ACP_MCP_NODE\` is present, call the same tools through the terminal: \`ELECTRON_RUN_AS_NODE=1 "$T3_ACP_MCP_NODE" \${T3_ACP_MCP_ENTRYPOINT:+"$T3_ACP_MCP_ENTRYPOINT"} acp-mcp-call orchestrator_capabilities '{}'\` (\`T3_ACP_MCP_ENTRYPOINT\` is unset when T3 runs as a standalone executable). Delegate with \`acp-mcp-call delegate_task '{"task":"...","target":{"providerInstanceId":"...","model":"..."},"mode":"async","clientRequestId":"..."}'\`. This is the supported T3 transport fallback, not an ordinary shell-based substitute for delegation. ```