diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 20f1ccf94aab7..f6dba8d68c709 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -919,6 +919,7 @@ const registerHtmlPreview = Effect.fn("McpHttpServer.registerHtmlPreview")(funct const HANDOFF_TOOLS: ReadonlySet = new Set([ "delegate_task", "create_threads", + "t3_team_grow", "t3_thread_launch", "t3_thread_send", "t3_thread_send_attachments", @@ -955,6 +956,7 @@ const resultThreadIds = (content: unknown): ReadonlyArray => { if (typeof content !== "object" || content === null) return []; const record = content as Readonly>; if (Array.isArray(record.threads)) return record.threads.flatMap(resultThreadIds); + if (Array.isArray(record.seats)) return record.seats.flatMap(resultThreadIds); const key = resultThreadKeys.find((candidate) => typeof record[candidate] === "string"); return key === undefined ? [] : [record[key] as string]; }; diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index f6567dba53f2c..7bb3df3857361 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -4590,6 +4590,202 @@ describe("OrchestratorMcpService thread stop", () => { }), }); + it.effect("shrinks one team seat without stopping the caller, and grow restores it", () => + Effect.gen(function* () { + const parentThreadId = ThreadId.make("thread:mcp-team-parent"); + const seatThreadId = ThreadId.make("thread:mcp-team-seat"); + const otherProjectSeatId = ThreadId.make("thread:mcp-team-other-project"); + const parentRunId = RunId.make("run:mcp-team-parent"); + const teamProjectId = ProjectId.make("project:mcp-team"); + const codexInstanceId = ProviderInstanceId.make("codex"); + const dispatched = yield* Ref.make>([]); + const stoppedDelegates = yield* Ref.make>([]); + const archived = yield* Ref.make(false); + const parentProjection = { + thread: { + id: parentThreadId, + projectId: teamProjectId, + title: "Lead", + modelSelection: { instanceId: codexInstanceId, model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + archivedAt: null, + deletedAt: null, + }, + runs: [ + { + id: parentRunId, + ordinal: 1, + status: "running", + providerInstanceId: codexInstanceId, + rootNodeId: NodeId.make("node:mcp-team-parent-root"), + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const seatShell = (isArchived: boolean) => ({ + ...liveThreadShell(seatThreadId, { activeRunId: RunId.make("run:mcp-team-seat") }), + projectId: teamProjectId, + title: "Seat", + archivedAt: isArchived ? DateTime.makeUnsafe("2026-10-03T10:00:00Z") : null, + }); + const seatProjection = (isArchived: boolean) => { + const shell = seatShell(isArchived); + return { + ...idleThreadProjection(shell), + runs: [ + { + id: RunId.make("run:mcp-team-seat"), + ordinal: 1, + status: "running", + providerInstanceId: codexInstanceId, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + }; + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: (threadId) => + Effect.gen(function* () { + if (threadId === parentThreadId) return parentProjection; + const isArchived = yield* Ref.get(archived); + return seatProjection(isArchived); + }), + getThreadShell: (threadId) => + Effect.gen(function* () { + if (threadId === parentThreadId) { + return liveThreadShell(parentThreadId, { activeRunId: parentRunId }); + } + if (threadId === otherProjectSeatId) { + return { + ...liveThreadShell(otherProjectSeatId), + projectId: ProjectId.make("project:other"), + }; + } + return seatShell(yield* Ref.get(archived)); + }), + getProjectThreadRecords: ({ threadId }) => + Effect.gen(function* () { + const isArchived = yield* Ref.get(archived); + if (threadId === otherProjectSeatId) { + const foreign = seatProjection(false); + return { + ...foreign, + thread: { + ...foreign.thread, + id: otherProjectSeatId, + projectId: ProjectId.make("project:other"), + }, + } as unknown as OrchestrationV2ThreadProjection; + } + return seatProjection(isArchived); + }), + dispatch: (command) => + Ref.update(dispatched, (commands) => [...commands, command]).pipe( + Effect.andThen( + command.type === "thread.archive" + ? Ref.set(archived, true) + : command.type === "thread.unarchive" + ? Ref.set(archived, false) + : Effect.void, + ), + Effect.as({ storedEvents: [] } as never), + ), + stopDelegatedTasks: ({ threadId }) => + Ref.update(stoppedDelegates, (ids) => [...ids, threadId]), + sendToThread: () => + Effect.succeed({ + threadId: seatThreadId, + runId: RunId.make("run:mcp-team-seat"), + delivery: "started", + } as never), + }), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:mcp-team"), + requestNamespace: "provider-session:mcp-team", + thread: { + threadId: parentThreadId, + providerSessionId: "provider-session:mcp-team", + providerInstanceId: codexInstanceId, + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const selfShrink = yield* service + .shrinkTeam(scope, { threadIds: [parentThreadId], clientRequestId: "shrink-self" }) + .pipe(Effect.flip); + assert.equal(selfShrink.code, "invalid_request"); + + const foreign = yield* service + .shrinkTeam(scope, { + threadIds: [otherProjectSeatId], + clientRequestId: "shrink-foreign", + }) + .pipe(Effect.flip); + assert.equal(foreign.code, "invalid_request"); + + const shrunk = yield* service.shrinkTeam(scope, { + threadIds: [seatThreadId], + reason: "capacity", + clientRequestId: "shrink-seat", + }); + assert.equal(shrunk.seats[0]?.threadId, seatThreadId); + assert.equal(shrunk.seats[0]?.status, "shrunk"); + assert.equal(shrunk.seats[0]?.title, "Seat"); + assert.isTrue(shrunk.seats[0]?.link?.includes(String(seatThreadId)) === true); + assert.equal(yield* Ref.get(archived), true); + + const commands = yield* Ref.get(dispatched); + assert.deepEqual( + commands.flatMap((command) => + command.type === "thread.stop" || command.type === "thread.archive" + ? [{ type: command.type, threadId: command.threadId }] + : [], + ), + [ + { type: "thread.stop", threadId: seatThreadId }, + { type: "thread.archive", threadId: seatThreadId }, + ], + ); + assert.deepEqual(yield* Ref.get(stoppedDelegates), [seatThreadId]); + assert.isFalse( + commands.some((command) => "threadId" in command && command.threadId === parentThreadId), + ); + + const again = yield* service.shrinkTeam(scope, { + threadIds: [seatThreadId], + clientRequestId: "shrink-again", + }); + assert.equal(again.seats[0]?.status, "already_archived"); + + const restored = yield* service.growTeam(scope, { + seats: [{ threadId: seatThreadId, prompt: "Welcome back." }], + clientRequestId: "grow-restore", + }); + assert.equal(restored.seats[0]?.threadId, seatThreadId); + assert.equal(yield* Ref.get(archived), false); + assert.isTrue( + (yield* Ref.get(dispatched)).some((command) => command.type === "thread.unarchive"), + ); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + }), + ); + it.effect.each([ { name: "a running turn", diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 406ac408cf1be..0f0775e1b5df3 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -20,6 +20,11 @@ import { type OrchestratorMcpCreateThreadsInput, type OrchestratorMcpCreateThreadsResult, type OrchestratorMcpCreatedThread, + type OrchestratorMcpTeamGrowInput, + type OrchestratorMcpTeamGrowResult, + type OrchestratorMcpTeamShrinkInput, + type OrchestratorMcpTeamShrinkResult, + type OrchestratorMcpTeamShrunkSeat, type OrchestratorMcpDelegateTaskInput, type OrchestratorMcpDelegateTaskResult, type OrchestratorMcpDeleteScheduledTaskInput, @@ -189,6 +194,16 @@ export interface OrchestratorMcpServiceShape { scope: McpInvocationScope, input: OrchestratorMcpCreateThreadsInput, ) => Effect.Effect; + /** Add seats to the running project team without stopping anyone else. */ + readonly growTeam: ( + scope: McpInvocationScope, + input: OrchestratorMcpTeamGrowInput, + ) => Effect.Effect; + /** Remove seats without stopping the rest of the team. */ + readonly shrinkTeam: ( + scope: McpInvocationScope, + input: OrchestratorMcpTeamShrinkInput, + ) => Effect.Effect; readonly scheduleTask: ( scope: McpInvocationScope, input: OrchestratorMcpScheduleTaskInput, @@ -2142,6 +2157,194 @@ const make = Effect.gen(function* () { return task; }); + const createThreads: OrchestratorMcpServiceShape["createThreads"] = Effect.fn( + "OrchestratorMcpService.createThreads", + )(function* (callerScope, input) { + const { scope, parent } = yield* loadThreadCaller(callerScope, "create_threads"); + const parentRun = ThreadManagementService.latestActiveRun(parent); + if ( + parentRun === undefined || + parentRun.rootNodeId === null || + parentRun.providerInstanceId !== scope.thread.providerInstanceId + ) { + return yield* failure( + "parent_not_active", + "Thread creation requires an active run owned by this MCP provider session.", + ); + } + const parentNodeId = parentRun.rootNodeId; + const providers = yield* loadProviders; + const key = yield* requestKey(input.clientRequestId); + const created = yield* Effect.forEach( + input.threads, + (request, index) => + Effect.gen(function* () { + const commandId = stableCommandId({ + scope, + requestKey: key, + operation: "create-thread", + index, + }); + const threadId = stableThreadId({ + scope, + requestKey: key, + index, + }); + const isRetryOfAcceptedCreation = yield* threadManagement.hasReceipt(commandId); + const committedThread = isRetryOfAcceptedCreation + ? yield* threadManagement + .getThreadShell(threadId) + .pipe(Effect.mapError(threadManagementFailure)) + : null; + const target = yield* resolveTargetRechecking({ + parent, + target: + committedThread === null + ? request.target + : { + ...request.target, + providerInstanceId: committedThread.modelSelection.instanceId, + }, + providers, + skipProjectEnablementFilter: isRetryOfAcceptedCreation, + }); + const runtimeMode = yield* resolveRuntimeMode( + parent.thread.runtimeMode, + request.runtimeMode, + target, + ); + const interactionMode = yield* resolveInteractionMode( + parent.thread.interactionMode, + request.interactionMode, + ); + yield* assertTargetEnabledForProject(parent, commandId, target); + const title = threadTitle({ + parentTitle: parent.thread.title, + prompt: request.prompt, + title: request.title, + index, + }); + const initialMessage = + request.prompt === undefined + ? undefined + : { + senderThreadId: scope.thread.threadId, + messageId: stableMessageId({ + scope, + requestKey: key, + index, + }), + text: request.prompt, + attachments: [], + }; + const created = yield* threadManagement + .dispatch({ + type: "thread.create", + createdBy: "agent", + creationSource: "mcp", + commandId, + threadId, + projectId: parent.thread.projectId, + title, + modelSelection: target.modelSelection, + runtimeMode, + interactionMode, + branch: parent.thread.branch, + worktreePath: parent.thread.worktreePath, + ...(parent.thread.linkOrigin === undefined + ? {} + : { linkOrigin: parent.thread.linkOrigin }), + ...(initialMessage === undefined ? {} : { initialMessage }), + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to create thread ${index + 1}: ${userFacingDispatchErrorMessage(error) ?? errorMessage(error)}`, + ), + ), + ); + // A receipt can appear after target resolution; use the committed thread's selection. + const resolvedModelSelection = (yield* threadManagement + .getThreadRecords(threadId, []) + .pipe(Effect.mapError(threadManagementFailure))).thread.modelSelection; + // A replayed creation that carried no prompt has no run yet. + if ( + initialMessage !== undefined && + !created.storedEvents.some(({ event }) => event.type === "run.created") + ) { + yield* threadManagement + .dispatch({ + ...initialMessage, + type: "message.dispatch", + createdBy: "agent", + creationSource: "mcp", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: "dispatch-thread", + index, + }), + threadId, + modelSelection: resolvedModelSelection, + dispatchMode: { type: "start_immediately" }, + }) + .pipe( + Effect.mapError((error) => + error._tag === "ThreadManagementProviderInstanceDisabledError" + ? threadManagementFailure(error) + : failure( + "orchestration_error", + `Unable to start thread ${index + 1}: ${userFacingDispatchErrorMessage(error) ?? errorMessage(error)}`, + ), + ), + ); + } + const projection = yield* loadThreadRecords(threadId, ["runs"]); + const run = projection.runs.at(-1); + yield* threadManagement + .dispatch({ + type: "thread.created.record", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: "record-created-thread", + index, + }), + parentThreadId: scope.thread.threadId, + parentRunId: parentRun.id, + parentNodeId, + targetThreadId: threadId, + targetRunId: run?.id ?? null, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to record thread ${index + 1} in the parent timeline: ${userFacingDispatchErrorMessage(error) ?? errorMessage(error)}`, + ), + ), + ); + return { + threadId, + link: formatThreadLink( + { environmentId: scope.environmentId, threadId }, + projection.thread.title, + ), + runId: run?.id ?? null, + status: run?.status ?? "idle", + title: projection.thread.title, + createdBy: projection.thread.createdBy, + creationSource: projection.thread.creationSource, + providerInstanceId: resolvedModelSelection.instanceId, + model: resolvedModelSelection.model, + } satisfies OrchestratorMcpCreatedThread; + }), + { concurrency: 1 }, + ); + return { threads: created }; + }); + return OrchestratorMcpService.of({ scheduleTask: Effect.fn("OrchestratorMcpService.scheduleTask")(function* (scope, input) { const { parent, limits } = yield* loadCaller(scope, ["runs"]); @@ -2931,193 +3134,243 @@ const make = Effect.gen(function* () { status: "cancel_requested", }; }), - createThreads: Effect.fn("OrchestratorMcpService.createThreads")( - function* (callerScope, input) { - const { scope, parent } = yield* loadThreadCaller(callerScope, "create_threads"); - const parentRun = ThreadManagementService.latestActiveRun(parent); + createThreads, + growTeam: Effect.fn("OrchestratorMcpService.growTeam")(function* (callerScope, input) { + const { scope, parent } = yield* loadThreadCaller(callerScope, "t3_team_grow"); + const parentRun = ThreadManagementService.latestActiveRun(parent); + if ( + parentRun === undefined || + parentRun.rootNodeId === null || + parentRun.providerInstanceId !== scope.thread.providerInstanceId + ) { + return yield* failure( + "parent_not_active", + "Growing the team requires an active run owned by this MCP provider session.", + ); + } + const key = yield* requestKey(input.clientRequestId); + const seats: Array = []; + for (const [index, seat] of input.seats.entries()) { + if (seat.threadId === undefined) { + const created = yield* createThreads(callerScope, { + threads: [ + { + ...(seat.prompt === undefined ? {} : { prompt: seat.prompt }), + ...(seat.title === undefined ? {} : { title: seat.title }), + ...(seat.target === undefined ? {} : { target: seat.target }), + ...(seat.runtimeMode === undefined ? {} : { runtimeMode: seat.runtimeMode }), + ...(seat.interactionMode === undefined + ? {} + : { interactionMode: seat.interactionMode }), + }, + ], + clientRequestId: `${key}:grow:${index}`, + }); + seats.push(created.threads[0]!); + continue; + } + + const threadId = seat.threadId; + if (threadId === parent.thread.id) { + return yield* failure( + "invalid_request", + "Cannot restore the calling thread as a new team seat.", + ); + } if ( - parentRun === undefined || - parentRun.rootNodeId === null || - parentRun.providerInstanceId !== scope.thread.providerInstanceId + seat.target !== undefined || + seat.runtimeMode !== undefined || + seat.interactionMode !== undefined || + seat.title !== undefined ) { return yield* failure( - "parent_not_active", - "Thread creation requires an active run owned by this MCP provider session.", + "invalid_request", + "Restoring a seat ignores title, target, and modes; pass only threadId and optional prompt.", ); } - const parentNodeId = parentRun.rootNodeId; - const providers = yield* loadProviders; - const key = yield* requestKey(input.clientRequestId); - const created = yield* Effect.forEach( - input.threads, - (request, index) => - Effect.gen(function* () { - const commandId = stableCommandId({ + const target = yield* loadTargetThread(threadId); + if (target.thread.projectId !== parent.thread.projectId) { + return yield* failure( + "invalid_request", + `Thread ${threadId} is not in this project team.`, + ); + } + if (target.thread.archivedAt !== null) { + yield* threadManagement + .dispatch({ + type: "thread.unarchive", + commandId: stableCommandId({ scope, requestKey: key, - operation: "create-thread", + operation: "team-grow-unarchive", index, - }); - const threadId = stableThreadId({ + }), + threadId, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to restore seat ${threadId}: ${errorMessage(error)}`, + ), + ), + ); + } + if (seat.prompt !== undefined) { + yield* threadManagement + .sendToThread({ + projectId: target.thread.projectId, + commandId: stableCommandId({ scope, requestKey: key, + operation: "team-grow-send", index, - }); - const isRetryOfAcceptedCreation = yield* threadManagement.hasReceipt(commandId); - const committedThread = isRetryOfAcceptedCreation - ? yield* threadManagement - .getThreadShell(threadId) - .pipe(Effect.mapError(threadManagementFailure)) - : null; - const target = yield* resolveTargetRechecking({ - parent, - target: - committedThread === null - ? request.target - : { - ...request.target, - providerInstanceId: committedThread.modelSelection.instanceId, - }, - providers, - skipProjectEnablementFilter: isRetryOfAcceptedCreation, - }); - const runtimeMode = yield* resolveRuntimeMode( - parent.thread.runtimeMode, - request.runtimeMode, - target, - ); - const interactionMode = yield* resolveInteractionMode( - parent.thread.interactionMode, - request.interactionMode, - ); - yield* assertTargetEnabledForProject(parent, commandId, target); - const title = threadTitle({ - parentTitle: parent.thread.title, - prompt: request.prompt, - title: request.title, + }), + threadId, + senderThreadId: scope.thread.threadId, + messageId: stableMessageId({ + scope, + requestKey: key, index, - }); - const initialMessage = - request.prompt === undefined - ? undefined - : { - senderThreadId: scope.thread.threadId, - messageId: stableMessageId({ - scope, - requestKey: key, - index, - }), - text: request.prompt, - attachments: [], - }; - const created = yield* threadManagement - .dispatch({ - type: "thread.create", - createdBy: "agent", - creationSource: "mcp", - commandId, - threadId, - projectId: parent.thread.projectId, - title, - modelSelection: target.modelSelection, - runtimeMode, - interactionMode, - branch: parent.thread.branch, - worktreePath: parent.thread.worktreePath, - ...(parent.thread.linkOrigin === undefined - ? {} - : { linkOrigin: parent.thread.linkOrigin }), - ...(initialMessage === undefined ? {} : { initialMessage }), - }) - .pipe( - Effect.mapError((error) => - failure( - "orchestration_error", - `Unable to create thread ${index + 1}: ${userFacingDispatchErrorMessage(error) ?? errorMessage(error)}`, - ), - ), - ); - // A receipt can appear after target resolution; use the committed thread's selection. - const resolvedModelSelection = (yield* threadManagement - .getThreadRecords(threadId, []) - .pipe(Effect.mapError(threadManagementFailure))).thread.modelSelection; - // A replayed creation that carried no prompt has no run yet. - if ( - initialMessage !== undefined && - !created.storedEvents.some(({ event }) => event.type === "run.created") - ) { - yield* threadManagement - .dispatch({ - ...initialMessage, - type: "message.dispatch", - createdBy: "agent", - creationSource: "mcp", - commandId: stableCommandId({ - scope, - requestKey: key, - operation: "dispatch-thread", - index, - }), - threadId, - modelSelection: resolvedModelSelection, - dispatchMode: { type: "start_immediately" }, - }) - .pipe( - Effect.mapError((error) => - error._tag === "ThreadManagementProviderInstanceDisabledError" - ? threadManagementFailure(error) - : failure( - "orchestration_error", - `Unable to start thread ${index + 1}: ${userFacingDispatchErrorMessage(error) ?? errorMessage(error)}`, - ), - ), - ); - } - const projection = yield* loadThreadRecords(threadId, ["runs"]); - const run = projection.runs.at(-1); - yield* threadManagement - .dispatch({ - type: "thread.created.record", - commandId: stableCommandId({ - scope, - requestKey: key, - operation: "record-created-thread", - index, - }), - parentThreadId: scope.thread.threadId, - parentRunId: parentRun.id, - parentNodeId, - targetThreadId: threadId, - targetRunId: run?.id ?? null, - }) - .pipe( - Effect.mapError((error) => - failure( + }), + text: seat.prompt, + attachments: [], + mode: "auto", + createdBy: "agent", + creationSource: "mcp", + }) + .pipe( + Effect.mapError((error) => + isThreadManagementError(error) + ? threadManagementFailure(error) + : failure( "orchestration_error", - `Unable to record thread ${index + 1} in the parent timeline: ${userFacingDispatchErrorMessage(error) ?? errorMessage(error)}`, + `Unable to message restored seat ${threadId}: ${errorMessage(error)}`, ), - ), - ); - return { - threadId, - link: formatThreadLink( - { environmentId: scope.environmentId, threadId }, - projection.thread.title, - ), - runId: run?.id ?? null, - status: run?.status ?? "idle", - title: projection.thread.title, - createdBy: projection.thread.createdBy, - creationSource: projection.thread.creationSource, - providerInstanceId: resolvedModelSelection.instanceId, - model: resolvedModelSelection.model, - } satisfies OrchestratorMcpCreatedThread; - }), - { concurrency: 1 }, + ), + ); + } + const projection = yield* loadThreadRecords(threadId, ["runs"]); + const run = projection.runs.at(-1); + seats.push({ + threadId, + link: formatThreadLink( + { environmentId: scope.environmentId, threadId }, + projection.thread.title, + ), + runId: run?.id ?? null, + status: run?.status ?? "idle", + title: projection.thread.title, + createdBy: projection.thread.createdBy, + creationSource: projection.thread.creationSource, + providerInstanceId: projection.thread.modelSelection.instanceId, + model: projection.thread.modelSelection.model, + }); + } + return { seats }; + }), + shrinkTeam: Effect.fn("OrchestratorMcpService.shrinkTeam")(function* (callerScope, input) { + const { scope, parent } = yield* loadThreadCaller(callerScope, "t3_team_shrink"); + const parentRun = ThreadManagementService.latestActiveRun(parent); + if ( + parentRun === undefined || + parentRun.providerInstanceId !== scope.thread.providerInstanceId + ) { + return yield* failure( + "parent_not_active", + "Shrinking the team requires an active run owned by this MCP provider session.", ); - return { threads: created }; - }, - ), + } + const key = yield* requestKey(input.clientRequestId); + const seats: Array = []; + for (const [index, threadId] of input.threadIds.entries()) { + if (threadId === parent.thread.id) { + return yield* failure( + "invalid_request", + "Cannot shrink the calling thread; shrink another seat on the team.", + ); + } + const target = yield* loadTargetThread(threadId); + if (target.thread.projectId !== parent.thread.projectId) { + return yield* failure( + "invalid_request", + `Thread ${threadId} is not in this project team.`, + ); + } + const link = formatThreadLink( + { environmentId: scope.environmentId, threadId }, + target.thread.title, + ); + if (target.thread.archivedAt !== null) { + seats.push({ + threadId, + link, + title: target.thread.title, + status: "already_archived", + }); + continue; + } + const stopCommandId = stableCommandId({ + scope, + requestKey: key, + operation: "team-shrink-stop", + index, + }); + const stopFailed = (error: unknown) => + Effect.logWarning("orchestrator-mcp.team-shrink-stop-failed", { + threadId, + error, + }).pipe( + Effect.andThen( + Effect.fail( + failure("orchestration_error", `Unable to stop seat ${threadId} before shrinking.`), + ), + ), + ); + // Stop only this seat and its own delegates; siblings and the caller keep running. + yield* threadManagement + .dispatch({ + type: "thread.stop", + commandId: stopCommandId, + threadId, + ...(input.reason === undefined ? {} : { reason: input.reason }), + }) + .pipe(Effect.catch(stopFailed)); + yield* threadManagement + .stopDelegatedTasks({ + threadId, + commandId: stopCommandId, + reason: input.reason, + }) + .pipe(Effect.catch(stopFailed)); + yield* threadManagement + .dispatch({ + type: "thread.archive", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: "team-shrink-archive", + index, + }), + threadId, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to archive seat ${threadId}: ${errorMessage(error)}`, + ), + ), + ); + seats.push({ + threadId, + link, + title: target.thread.title, + status: "shrunk", + }); + } + return { seats }; + }), listThreads: Effect.fn("OrchestratorMcpService.listThreads")(function* (scope, input) { const { parent } = yield* loadCaller(scope, []); const projectId = input.allProjects diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts index dfa1ad58545ee..1f87297a1b3fd 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -150,6 +150,20 @@ const handlers = { return yield* service.createThreads(scope, input); }), ), + t3_team_grow: McpToolAccess.actsAsCaller((input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + return yield* service.growTeam(scope, input); + }), + ), + t3_team_shrink: McpToolAccess.actsAsCaller((input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + return yield* service.shrinkTeam(scope, input); + }), + ), t3_thread_list: McpToolAccess.reads((input) => Effect.gen(function* () { const scope = yield* McpInvocationContext.McpInvocationContext; diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 3a84c6a8e9274..301f24d461187 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -9,6 +9,10 @@ import { OrchestratorMcpEnvironmentUnlinkResult, OrchestratorMcpCreateThreadsInput, OrchestratorMcpCreateThreadsResult, + OrchestratorMcpTeamGrowInput, + OrchestratorMcpTeamGrowResult, + OrchestratorMcpTeamShrinkInput, + OrchestratorMcpTeamShrinkResult, OrchestratorMcpDelegateTaskInput, OrchestratorMcpDelegateTaskResult, OrchestratorMcpDeleteScheduledTaskInput, @@ -229,6 +233,32 @@ const CreateThreadsTool = Tool.make("create_threads", { .annotate(Tool.Destructive, true) .annotate(Tool.OpenWorld, true); +const TeamGrowTool = Tool.make("t3_team_grow", { + description: + "Add seats to a running project team without stopping anyone else (OpenRig `rig grow`). New seat: omit `threadId`. Restore after shrink: pass the archived seat's `threadId` (optional `prompt`). In your reply, name each seat as its `link` verbatim.", + parameters: OrchestratorMcpTeamGrowInput, + success: OrchestratorMcpTeamGrowResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Grow a running team") + .annotate(Tool.Destructive, true) + .annotate(Tool.OpenWorld, true); + +const TeamShrinkTool = Tool.make("t3_team_shrink", { + description: + "Remove seats from a running project team without stopping the rest (OpenRig `rig shrink`). Stops and archives each listed seat and its own delegates only. Reverse with `t3_team_grow` and that seat's `threadId`. Cannot shrink the calling thread.", + parameters: OrchestratorMcpTeamShrinkInput, + success: OrchestratorMcpTeamShrinkResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Shrink a running team") + .annotate(Tool.Destructive, true) + .annotate(Tool.Idempotent, true); + const ThreadListTool = Tool.make("t3_thread_list", { description: "List threads newest first, in one project or with `allProjects`. Page with `cursor`.", @@ -368,6 +398,8 @@ export const OrchestratorToolkit = Toolkit.make( DeleteScheduledTaskTool, RequestSecretTool, CreateThreadsTool, + TeamGrowTool, + TeamShrinkTool, ThreadListTool, ThreadReadTool, ThreadUpdateTool, diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 7ceed7057c195..4f746ea98c3e0 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5930, "codexAdditionalContextBytes": 7324, - "mcpToolListJsonBytes": 49791 + "mcpToolListJsonBytes": 51655 } diff --git a/apps/server/src/telemetry/ProviderDimensions.ts b/apps/server/src/telemetry/ProviderDimensions.ts index 8352ed0d15cc0..f584519084a2b 100644 --- a/apps/server/src/telemetry/ProviderDimensions.ts +++ b/apps/server/src/telemetry/ProviderDimensions.ts @@ -150,6 +150,18 @@ export function handoffSettings(tool: string, args: unknown, result: unknown): F targetChosen: threads.some((thread) => asFields(thread).target !== undefined), }; } + case "t3_team_grow": { + const seats = Array.isArray(input.seats) ? input.seats : []; + return { + batchSize: seats.length, + restoreCount: seats.filter((seat) => asFields(seat).threadId !== undefined).length, + targetChosen: seats.some((seat) => asFields(seat).target !== undefined), + }; + } + case "t3_team_shrink": { + const threadIds = Array.isArray(input.threadIds) ? input.threadIds : []; + return { batchSize: threadIds.length }; + } case "t3_thread_launch": return defined({ targetChosen: input.modelSelection !== undefined, diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 19163466972bb..18f6d5d5617c5 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -315,6 +315,64 @@ export const OrchestratorMcpCreateThreadsResult = Schema.Struct({ }); export type OrchestratorMcpCreateThreadsResult = typeof OrchestratorMcpCreateThreadsResult.Type; +/** + * Add a seat to a running project team without stopping anyone else (OpenRig `rig grow`). + * Omit `threadId` to create; pass an archived seat's id to restore after shrink. + */ +export const OrchestratorMcpTeamGrowSeat = Schema.Struct({ + threadId: Schema.optional( + ThreadId.annotate({ + description: + "Archived seat to restore (reverse of `t3_team_shrink`). Omit to create a new seat.", + }), + ), + prompt: Schema.optional(OrchestratorMcpPrompt), + title: Schema.optional(OrchestratorMcpTitle), + target: Schema.optional(OrchestratorMcpTarget), + runtimeMode: Schema.optional(OrchestratorMcpRuntimeMode), + interactionMode: Schema.optional(OrchestratorMcpInteractionMode), +}); +export type OrchestratorMcpTeamGrowSeat = typeof OrchestratorMcpTeamGrowSeat.Type; + +export const OrchestratorMcpTeamGrowInput = Schema.Struct({ + seats: Schema.Array(OrchestratorMcpTeamGrowSeat).check( + Schema.isMinLength(1), + Schema.isMaxLength(20), + ), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), +}); +export type OrchestratorMcpTeamGrowInput = typeof OrchestratorMcpTeamGrowInput.Type; + +export const OrchestratorMcpTeamGrowResult = Schema.Struct({ + seats: Schema.Array(OrchestratorMcpCreatedThread), +}); +export type OrchestratorMcpTeamGrowResult = typeof OrchestratorMcpTeamGrowResult.Type; + +/** Remove seats from a running team without stopping the rest (OpenRig `rig shrink`). */ +export const OrchestratorMcpTeamShrinkInput = Schema.Struct({ + threadIds: Schema.Array(ThreadId).check(Schema.isMinLength(1), Schema.isMaxLength(20)), + reason: Schema.optional( + TrimmedNonEmptyString.check(Schema.isMaxLength(2_000)).annotate({ + description: "Why this seat is leaving the team.", + }), + ), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), +}); +export type OrchestratorMcpTeamShrinkInput = typeof OrchestratorMcpTeamShrinkInput.Type; + +export const OrchestratorMcpTeamShrunkSeat = Schema.Struct({ + threadId: ThreadId, + link: Schema.optional(Schema.String), + title: Schema.String, + status: Schema.Literals(["shrunk", "already_archived"]), +}); +export type OrchestratorMcpTeamShrunkSeat = typeof OrchestratorMcpTeamShrunkSeat.Type; + +export const OrchestratorMcpTeamShrinkResult = Schema.Struct({ + seats: Schema.Array(OrchestratorMcpTeamShrunkSeat), +}); +export type OrchestratorMcpTeamShrinkResult = typeof OrchestratorMcpTeamShrinkResult.Type; + export const OrchestratorMcpThreadStatus = Schema.Union([ Schema.Literal("idle"), OrchestrationV2RunStatus, diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index a813585b27a6e..10c7379423885 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -166,6 +166,8 @@ const T3_MCP_TOOLS: Readonly> = { ), request_secret: tool(["Ask for", "Asking for", "Asked for", "a secret"], "secret-request"), create_threads: tool(["Create", "Creating", "Created", "T3 threads"], "thread-create"), + t3_team_grow: tool(["Grow", "Growing", "Grew", "the project team"], "thread-create"), + t3_team_shrink: tool(["Shrink", "Shrinking", "Shrunk", "the project team"], "thread-organize"), t3_thread_start: tool(["Start", "Starting", "Started", "a T3 thread"], "thread-create"), t3_thread_list: tool(["List", "Listing", "Listed", "T3 threads"], "thread-list"), t3_thread_read: tool(["Read", "Reading", "Read", "a T3 thread"], "thread-read"),