From a21df05a7dd1c01931059bc2edbb5daa853d281b Mon Sep 17 00:00:00 2001 From: wSedlacek Date: Wed, 11 Nov 2026 14:13:08 -0800 Subject: [PATCH 1/3] feat(orchestration): durable specialist ownership on threads Bind OpenRig-style named specialist roles to listed project threads so agents can resume domain-owned context by stable key instead of spawning turn-scoped subagents. --- .../features/threads/thread-list-v2-items.tsx | 6 +- apps/server/src/mcp/McpHttpServer.ts | 1 + apps/server/src/mcp/OrchestratorMcpService.ts | 308 ++++++++++++++++++ .../src/mcp/toolkits/orchestrator/handlers.ts | 7 + .../src/mcp/toolkits/orchestrator/tools.ts | 17 + .../src/orchestration-v2/Orchestrator.ts | 43 +++ .../src/orchestration-v2/ProjectionStore.ts | 9 + .../ThreadForkService.test.ts | 2 + .../src/orchestration-v2/ThreadForkService.ts | 7 +- .../src/orchestration-v2/runtimeLayer.test.ts | 40 +++ .../testkit/OrchestratorScenario.ts | 1 + .../src/provider/agentContextBudget.json | 2 +- apps/server/src/relay/AgentAwarenessRelay.ts | 1 + .../components/sidebar/SidebarThreadRow.tsx | 13 +- .../client-runtime/src/operations/commands.ts | 16 + .../src/state/orchestrationV2Projection.ts | 3 +- .../src/state/threadCommands.ts | 7 + packages/contracts/src/index.ts | 1 + packages/contracts/src/orchestrationV2.ts | 17 + packages/contracts/src/orchestratorMcp.ts | 59 ++++ .../contracts/src/threadSpecialist.test.ts | 43 +++ packages/contracts/src/threadSpecialist.ts | 31 ++ 22 files changed, 629 insertions(+), 5 deletions(-) create mode 100644 packages/contracts/src/threadSpecialist.test.ts create mode 100644 packages/contracts/src/threadSpecialist.ts diff --git a/apps/mobile/src/features/threads/thread-list-v2-items.tsx b/apps/mobile/src/features/threads/thread-list-v2-items.tsx index d724d8056e95..b6453a565cf1 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -776,7 +776,9 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { : workingLabel) ?? (isUnread ? DONE_STATUS_LABEL : undefined); const recede = shouldRecedeThreadRow({ status, selected }); const isRegeneratingTitle = thread.titleRegeneration != null; - const rowAccessibilityLabel = `${thread.title}${providerNameLabel ? `, ${providerNameLabel}` : ""}${props.hasQueuedMessages ? ", messages queued to send" : ""}${isRegeneratingTitle ? ", regenerating title" : ""}`; + const specialistPrefix = + thread.source.specialist === undefined ? "" : `@${thread.source.specialist.key} `; + const rowAccessibilityLabel = `${specialistPrefix}${thread.title}${providerNameLabel ? `, ${providerNameLabel}` : ""}${props.hasQueuedMessages ? ", messages queued to send" : ""}${isRegeneratingTitle ? ", regenerating title" : ""}`; const timeLabel = props.timeLabel; const visitThread = useAtomCommand(threadEnvironment.visit, { reportFailure: false }); @@ -1191,6 +1193,7 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { : undefined } > + {specialistPrefix} {thread.title} @@ -1499,6 +1502,7 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { )} numberOfLines={1} > + {specialistPrefix} {thread.title} {providerNameLabel ? ( diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index ccd739e93a6f..b76fcf6b1e58 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -882,6 +882,7 @@ const registerHtmlPreview = Effect.fn("McpHttpServer.registerHtmlPreview")(funct const HANDOFF_TOOLS: ReadonlySet = new Set([ "delegate_task", "create_threads", + "find_or_open_specialist", "t3_thread_launch", "t3_thread_send", "t3_thread_send_attachments", diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 71427548f86e..6bc913b6a2a0 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -20,7 +20,10 @@ import { type OrchestratorMcpCreateThreadsInput, type OrchestratorMcpCreateThreadsResult, type OrchestratorMcpCreatedThread, + type OrchestratorMcpFindOrOpenSpecialistInput, + type OrchestratorMcpFindOrOpenSpecialistResult, type OrchestratorMcpDelegateTaskInput, + type ThreadSpecialist, type OrchestratorMcpDelegateTaskResult, type OrchestratorMcpDeleteScheduledTaskInput, type OrchestratorMcpRequestSecretInput, @@ -62,6 +65,7 @@ import { type ServerProvider, type ServerProviderModel, ThreadId, + isListedThreadLineage, isOrchestrationV2WorkActive, latestProviderTurnForAttempt, resolveProjectProviderInstanceEnabled, @@ -176,6 +180,10 @@ export interface OrchestratorMcpServiceShape { scope: McpInvocationScope, input: OrchestratorMcpCreateThreadsInput, ) => Effect.Effect; + readonly findOrOpenSpecialist: ( + scope: McpInvocationScope, + input: OrchestratorMcpFindOrOpenSpecialistInput, + ) => Effect.Effect; readonly scheduleTask: ( scope: McpInvocationScope, input: OrchestratorMcpScheduleTaskInput, @@ -838,6 +846,7 @@ function listItemFromShell( ...threadSnooze(shell, nowMs), parentThreadId: shell.lineage.parentThreadId, relationshipToParent: shell.lineage.relationshipToParent, + ...(shell.specialist === undefined ? {} : { specialist: shell.specialist }), itemCount: shell.visibleItemCount, createdAt: DateTime.formatIso(shell.createdAt), updatedAt: DateTime.formatIso(shell.updatedAt), @@ -889,6 +898,9 @@ function threadDetail( archived: projection.thread.archivedAt !== null, ...threadSettlement(projection.thread), ...threadSnooze(shell, nowMs), + ...(projection.thread.specialist === undefined + ? {} + : { specialist: projection.thread.specialist }), createdAt: DateTime.formatIso(projection.thread.createdAt), updatedAt: DateTime.formatIso(projection.thread.updatedAt), }; @@ -2951,6 +2963,298 @@ const make = Effect.gen(function* () { return { threads: created }; }, ), + findOrOpenSpecialist: Effect.fn("OrchestratorMcpService.findOrOpenSpecialist")( + function* (callerScope, input) { + const { parent: listParent } = yield* loadCaller(callerScope, []); + const projectId = yield* resolveProjectTarget( + listParent, + input.projectId, + "Pass projectId: this MCP client is not running inside a T3 thread.", + ); + const label = input.label ?? input.key; + const specialist: ThreadSpecialist = { + key: input.key, + label, + ...(input.instructions === undefined ? {} : { instructions: input.instructions }), + }; + const snapshot = yield* threadManagement + .getShellSnapshot({ projectId }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to look up specialists: ${errorMessage(error)}`, + ), + ), + ); + const existing = [...snapshot.threads, ...snapshot.archivedThreads] + .filter( + (thread) => + thread.deletedAt === null && + thread.specialist?.key === input.key && + isListedThreadLineage(thread.lineage), + ) + .toSorted( + (left, right) => + (left.archivedAt === null ? 0 : 1) - (right.archivedAt === null ? 0 : 1) || + DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt), + )[0]; + + const key = yield* requestKey(input.clientRequestId); + + if (existing !== undefined) { + const needsRebind = + existing.specialist?.label !== specialist.label || + existing.specialist?.instructions !== specialist.instructions; + if (needsRebind) { + yield* threadManagement + .dispatch({ + type: "thread.specialist.set", + commandId: stableCommandId({ + scope: callerScope, + requestKey: key, + operation: "rebind-specialist", + }), + threadId: existing.id, + specialist, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to update specialist binding: ${errorMessage(error)}`, + ), + ), + ); + } + let runId: RunId | null = existing.latestRunId; + let status: OrchestratorMcpFindOrOpenSpecialistResult["status"] = + existing.activityRunStatus ?? existing.status; + if (input.prompt !== undefined) { + const sent = yield* threadManagement + .sendToThread({ + projectId, + commandId: stableCommandId({ + scope: callerScope, + requestKey: key, + operation: "specialist-send", + }), + threadId: existing.id, + ...(listParent === undefined ? {} : { senderThreadId: listParent.thread.id }), + messageId: stableOperationMessageId({ + scope: callerScope, + requestKey: key, + operation: "specialist-send", + }), + text: input.prompt, + attachments: [], + mode: "auto", + createdBy: "agent", + creationSource: "mcp", + }) + .pipe( + Effect.mapError((error) => + isThreadManagementError(error) + ? threadManagementFailure(error) + : failure( + "orchestration_error", + `Unable to message specialist: ${userFacingDispatchErrorMessage(error) ?? errorMessage(error)}`, + ), + ), + ); + runId = sent.run.id; + status = sent.run.status; + } + return { + threadId: existing.id, + link: threadLinkFor(callerScope, existing.id, existing.title), + created: false, + specialist, + runId, + status, + title: existing.title, + } satisfies OrchestratorMcpFindOrOpenSpecialistResult; + } + + const { scope, parent } = yield* loadThreadCaller(callerScope, "find_or_open_specialist"); + yield* assertLiveCallerForOtherProject(scope, parent, projectId); + const parentRun = ThreadManagementService.latestActiveRun(parent); + if ( + parentRun === undefined || + parentRun.rootNodeId === null || + parentRun.providerInstanceId !== scope.thread.providerInstanceId + ) { + return yield* failure( + "parent_not_active", + "Creating a specialist requires an active run owned by this MCP provider session.", + ); + } + if (parent.thread.projectId !== projectId) { + return yield* failure( + "target_required", + "Creating a specialist in another project is not supported yet; open or create it from a thread in that project.", + ); + } + const providers = yield* loadProviders; + const commandId = stableCommandId({ + scope, + requestKey: key, + operation: "create-specialist", + }); + const threadId = stableThreadId({ scope, requestKey: key, index: 0 }); + 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 + ? input.target + : { + ...input.target, + providerInstanceId: committedThread.modelSelection.instanceId, + }, + providers, + skipProjectEnablementFilter: isRetryOfAcceptedCreation, + }); + const runtimeMode = yield* resolveRuntimeMode( + parent.thread.runtimeMode, + input.runtimeMode, + target, + ); + const interactionMode = yield* resolveInteractionMode( + parent.thread.interactionMode, + input.interactionMode, + ); + yield* assertTargetEnabledForProject(parent, commandId, target); + const title = threadTitle({ + parentTitle: parent.thread.title, + prompt: input.prompt, + title: input.title ?? label, + index: 0, + }); + if (committedThread === null) { + yield* threadManagement + .dispatch({ + type: "thread.create", + createdBy: "agent", + creationSource: "mcp", + commandId, + threadId, + projectId, + title, + modelSelection: target.modelSelection, + runtimeMode, + interactionMode, + branch: parent.thread.branch, + worktreePath: parent.thread.worktreePath, + ...(parent.thread.linkOrigin === undefined + ? {} + : { linkOrigin: parent.thread.linkOrigin }), + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to create specialist thread: ${errorMessage(error)}`, + ), + ), + ); + } + yield* threadManagement + .dispatch({ + type: "thread.specialist.set", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: "bind-specialist", + }), + threadId, + specialist, + }) + .pipe( + Effect.mapError((error) => + failure("orchestration_error", `Unable to bind specialist: ${errorMessage(error)}`), + ), + ); + const resolvedModelSelection = (yield* threadManagement + .getThreadRecords(threadId, []) + .pipe(Effect.mapError(threadManagementFailure))).thread.modelSelection; + if (input.prompt !== undefined) { + yield* threadManagement + .dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "mcp", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: "dispatch-specialist", + }), + threadId, + senderThreadId: scope.thread.threadId, + messageId: stableMessageId({ + scope, + requestKey: key, + index: 0, + }), + text: input.prompt, + attachments: [], + modelSelection: resolvedModelSelection, + dispatchMode: { type: "start_immediately" }, + }) + .pipe( + Effect.mapError((error) => + error._tag === "ThreadManagementProviderInstanceDisabledError" + ? threadManagementFailure(error) + : failure( + "orchestration_error", + `Unable to start specialist: ${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-specialist", + }), + parentThreadId: scope.thread.threadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId, + targetThreadId: threadId, + targetRunId: run?.id ?? null, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to record specialist in the parent timeline: ${errorMessage(error)}`, + ), + ), + ); + return { + threadId, + link: formatThreadLink( + { environmentId: scope.environmentId, threadId }, + projection.thread.title, + ), + created: true, + specialist, + runId: run?.id ?? null, + status: run?.status ?? "idle", + title: projection.thread.title, + } satisfies OrchestratorMcpFindOrOpenSpecialistResult; + }, + ), listThreads: Effect.fn("OrchestratorMcpService.listThreads")(function* (scope, input) { const { parent } = yield* loadCaller(scope, []); const projectId = input.allProjects @@ -2991,6 +3295,10 @@ const make = Effect.gen(function* () { .filter( (thread) => titleContains === undefined || thread.title.toLocaleLowerCase().includes(titleContains), + ) + .filter( + (thread) => + input.specialistKey === undefined || thread.specialist?.key === input.specialistKey, ); const cursor = input.cursor ?? 0; const limit = input.limit ?? DEFAULT_THREAD_LIST_LIMIT; diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts index 3db297bc0d30..c0fc1d727182 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -145,6 +145,13 @@ const handlers = { return yield* service.createThreads(scope, input); }), ), + find_or_open_specialist: McpToolAccess.actsAsCaller((input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + return yield* service.findOrOpenSpecialist(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 2ba8a5f2dd5f..851682b19e3b 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -9,6 +9,8 @@ import { OrchestratorMcpEnvironmentUnlinkResult, OrchestratorMcpCreateThreadsInput, OrchestratorMcpCreateThreadsResult, + OrchestratorMcpFindOrOpenSpecialistInput, + OrchestratorMcpFindOrOpenSpecialistResult, OrchestratorMcpDelegateTaskInput, OrchestratorMcpDelegateTaskResult, OrchestratorMcpDeleteScheduledTaskInput, @@ -222,6 +224,20 @@ const CreateThreadsTool = Tool.make("create_threads", { .annotate(Tool.Destructive, true) .annotate(Tool.OpenWorld, true); +const FindOrOpenSpecialistTool = Tool.make("find_or_open_specialist", { + description: + "Resume or create a durable specialist thread by stable `key` (e.g. `dev-ui`) in a project. Keeps context across days; not a turn-scoped `delegate_task`. Creating needs an active run. In your reply, name it as its `link` verbatim.", + parameters: OrchestratorMcpFindOrOpenSpecialistInput, + success: OrchestratorMcpFindOrOpenSpecialistResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Find or open a durable specialist") + .annotate(Tool.Destructive, true) + .annotate(Tool.OpenWorld, 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`.", @@ -361,6 +377,7 @@ export const OrchestratorToolkit = Toolkit.make( DeleteScheduledTaskTool, RequestSecretTool, CreateThreadsTool, + FindOrOpenSpecialistTool, ThreadListTool, ThreadReadTool, ThreadUpdateTool, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 7af71e48d67e..13e2f2b72d1b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -612,6 +612,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "thread.pin.reorder": case "thread.active.reorder": case "thread.tag.set": + case "thread.specialist.set": case "thread.visit": case "thread.mark-unread": case "thread.metadata.update": @@ -3630,6 +3631,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | "thread.unpin" | "thread.pin.reorder" | "thread.tag.set" + | "thread.specialist.set" | "thread.active.reorder" | "thread.mark-unread" | "thread.metadata.update" @@ -3665,6 +3667,34 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} is deleted.`, }); } + if (command.type === "thread.specialist.set") { + if (thread.lineage.relationshipToParent === "subagent" || thread.lineage.sideChat === true) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} is a subagent or side chat and cannot own a specialist.`, + }); + } + const specialist = command.specialist; + if (specialist !== null) { + const snapshot = yield* projectionStore + .getShellSnapshot({ projectId: thread.projectId }) + .pipe(mapDispatchError(command)); + const conflict = [...snapshot.threads, ...snapshot.archivedThreads].find( + (candidate) => + candidate.id !== thread.id && + candidate.deletedAt === null && + candidate.specialist?.key === specialist.key, + ); + if (conflict !== undefined) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Specialist key "${specialist.key}" is already bound to thread ${conflict.id}.`, + }); + } + } + } if ( command.type === "thread.pull-request.watch" && command.watching && @@ -4210,6 +4240,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : current.filter((tag) => tag !== command.tag), }; } + case "thread.specialist.set": { + if (command.specialist === null) { + const { specialist: _cleared, ...rest } = thread; + return rest; + } + return { + ...thread, + specialist: command.specialist, + }; + } case "thread.mark-unread": { // Retire the wake before rewinding the visit, preserving its settlement timer. const snoozeExpired = @@ -4589,6 +4629,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return "thread.active-reordered" as const; case "thread.tag.set": return "thread.tags-updated" as const; + case "thread.specialist.set": + return "thread.specialist-updated" as const; case "thread.mark-unread": return "thread.marked-unread" as const; case "thread.metadata.update": @@ -13855,6 +13897,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.pin.reorder": case "thread.active.reorder": case "thread.tag.set": + case "thread.specialist.set": case "thread.mark-unread": case "thread.metadata.update": case "thread.pull-request.link": diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9e2721035064..a36e870f5ddd 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -876,6 +876,7 @@ export function applyToProjection( // Reordering and PR refreshes preserve the orchestrator's activity timestamp. case "thread.active-reordered": case "thread.tags-updated": + case "thread.specialist-updated": case "thread.pull-request-synced": return { ...base, @@ -1718,6 +1719,9 @@ export function threadShellFromProjection( autoSettleDisabledAt: projection.thread.autoSettleDisabledAt ?? null, pinOrderKey: projection.thread.pinOrderKey ?? null, ...(projection.thread.tags?.length ? { tags: projection.thread.tags } : {}), + ...(projection.thread.specialist === undefined + ? {} + : { specialist: projection.thread.specialist }), lastVisitedAt: projection.thread.lastVisitedAt, opens: projection.thread.opens ?? null, titleRegeneration: projection.thread.titleRegeneration ?? null, @@ -2005,6 +2009,9 @@ function shellFromState(input: { autoSettleDisabledAt: input.state.thread.autoSettleDisabledAt ?? null, pinOrderKey: input.state.thread.pinOrderKey ?? null, ...(input.state.thread.tags?.length ? { tags: input.state.thread.tags } : {}), + ...(input.state.thread.specialist === undefined + ? {} + : { specialist: input.state.thread.specialist }), lastVisitedAt: input.state.thread.lastVisitedAt, opens: input.state.thread.opens ?? null, titleRegeneration: input.state.thread.titleRegeneration ?? null, @@ -2088,6 +2095,7 @@ export const layer: Layer.Layer = case "thread.pin-reordered": case "thread.active-reordered": case "thread.tags-updated": + case "thread.specialist-updated": case "thread.visited": case "thread.marked-unread": case "thread.metadata-updated": @@ -2905,6 +2913,7 @@ export const layer: Layer.Layer = event.type !== "thread.pin-reordered" && event.type !== "thread.active-reordered" && event.type !== "thread.tags-updated" && + event.type !== "thread.specialist-updated" && event.type !== "thread.visited" && event.type !== "thread.marked-unread" && event.type !== "thread.metadata-updated" && diff --git a/apps/server/src/orchestration-v2/ThreadForkService.test.ts b/apps/server/src/orchestration-v2/ThreadForkService.test.ts index f9f6ff25b2f0..24ca3dc683ed 100644 --- a/apps/server/src/orchestration-v2/ThreadForkService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadForkService.test.ts @@ -152,6 +152,7 @@ it.effect("resets a fork's pin, snooze, settle and delegation state but keeps it historyOrigin: "v1_import", linkOrigin, tags: ["private"], + specialist: { key: "dev-ui", label: "Dev UI", instructions: "Owns the web client." }, }; const sourceRun = makeSourceRun("completed"); const result = yield* planFork(sourceRun, { @@ -167,6 +168,7 @@ it.effect("resets a fork's pin, snooze, settle and delegation state but keeps it assert.isNull(result.targetThread.activeOrderKey); assert.isNull(result.targetThread.autoSettleDisabledAt); assert.isUndefined(result.targetThread.delegatedFrom); + assert.isUndefined(result.targetThread.specialist); assert.equal(result.targetThread.historyOrigin, "v1_import"); assert.deepEqual(result.targetThread.linkOrigin, linkOrigin); assert.deepEqual(result.targetThread.tags, ["private"]); diff --git a/apps/server/src/orchestration-v2/ThreadForkService.ts b/apps/server/src/orchestration-v2/ThreadForkService.ts index 1dc6d1eaf4a5..7f28e4a18648 100644 --- a/apps/server/src/orchestration-v2/ThreadForkService.ts +++ b/apps/server/src/orchestration-v2/ThreadForkService.ts @@ -100,7 +100,12 @@ export const layer: Layer.Layer = Layer.succeed( }); } // The remote parent delegated the source thread, not its forks. - const { delegatedFrom: _delegatedFrom, ...inherited } = input.sourceProjection.thread; + // Specialist ownership is project-unique; forks must not inherit the key. + const { + delegatedFrom: _delegatedFrom, + specialist: _specialist, + ...inherited + } = input.sourceProjection.thread; const targetThread: OrchestrationV2AppThread = { ...inherited, // PR context follows the conversation; watching remains the source thread's responsibility. diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 9cce2d21a85f..d0249df56917 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -2903,6 +2903,46 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { DateTime.toEpochMillis(untaggedShell.updatedAt), ); + yield* orchestrator.dispatch({ + type: "thread.specialist.set", + commandId: CommandId.make("runtime-layer-lifecycle-specialist-bind"), + threadId, + specialist: { key: "dev-ui", label: "Dev UI", instructions: "Owns the web client." }, + }); + assert.deepEqual(shellOf(yield* orchestrator.getShellSnapshot()).specialist, { + key: "dev-ui", + label: "Dev UI", + instructions: "Owns the web client.", + }); + const specialistShell = shellOf(yield* orchestrator.getShellSnapshot()); + assert.equal( + DateTime.toEpochMillis(specialistShell.updatedAt), + DateTime.toEpochMillis(untaggedShell.updatedAt), + ); + const otherThreadId = ThreadId.make("runtime-layer-lifecycle-specialist-other"); + yield* orchestrator.dispatch({ + ...create, + commandId: CommandId.make("runtime-layer-lifecycle-specialist-other-create"), + threadId: otherThreadId, + title: "Other lifecycle thread", + }); + const specialistConflict = yield* orchestrator + .dispatch({ + type: "thread.specialist.set", + commandId: CommandId.make("runtime-layer-lifecycle-specialist-conflict"), + threadId: otherThreadId, + specialist: { key: "dev-ui", label: "Dev UI" }, + }) + .pipe(Effect.flip); + assert.instanceOf(specialistConflict, Orchestrator.OrchestratorDispatchError); + yield* orchestrator.dispatch({ + type: "thread.specialist.set", + commandId: CommandId.make("runtime-layer-lifecycle-specialist-clear"), + threadId, + specialist: null, + }); + assert.isUndefined(shellOf(yield* orchestrator.getShellSnapshot()).specialist); + const preAutoProjection = yield* orchestrator.getThreadProjection(threadId); const staleAutoSettle = yield* orchestrator .dispatch({ diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts index b413c0aefa25..a2f227d6b204 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts @@ -152,6 +152,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray + {thread.source.specialist !== undefined ? ( + + @{thread.source.specialist.key} + + ) : null} {thread.title} ); - const accessibleTitle = isRenaming ? null : {thread.title}; + const accessibleTitle = isRenaming ? null : ( + + {thread.source.specialist !== undefined + ? `${thread.source.specialist.label} (@${thread.source.specialist.key}): ${thread.title}` + : thread.title} + + ); const prBadgeShape = supportsMultiplePullRequests ? resolveThreadPullRequestBadge(thread.pullRequests) diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index f4ece3888ec9..3e9ba83bd64a 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -25,6 +25,7 @@ import { type RuntimeRequestId, type ThreadId, type ThreadEnvMode, + type ThreadSpecialist, type UploadChatAttachment, type GuestMessageId, } from "@t3tools/contracts"; @@ -109,6 +110,10 @@ export interface ReorderActiveThreadInput extends ThreadCommandInput { readonly orderKey: string; } +export interface SetThreadSpecialistInput extends ThreadCommandInput { + readonly specialist: ThreadSpecialist | null; +} + export interface SnoozeThreadInput extends ThreadCommandInput { readonly snoozedUntil: string; } @@ -546,6 +551,17 @@ export const reorderActiveThread = Effect.fn("EnvironmentCommands.reorderActiveT }); }); +export const setThreadSpecialist = Effect.fn("EnvironmentCommands.setThreadSpecialist")(function* ( + input: SetThreadSpecialistInput, +) { + return yield* dispatch({ + type: "thread.specialist.set", + commandId: yield* allocateCommandId(input), + threadId: input.threadId, + specialist: input.specialist, + }); +}); + export const unpinThread = Effect.fn("EnvironmentCommands.unpinThread")(function* ( input: UnpinThreadInput, ) { diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index d9368ef4058a..ee2bb5445dc4 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -293,9 +293,10 @@ export function applyOrchestrationV2ProjectionEvent( case "thread.provider-switched": case "thread.guest-share-updated": return { ...base, thread: event.payload }; - // Visited tracking and tags are not activity: skip the updatedAt bump. + // Visited tracking, tags, and specialist bindings are not activity: skip the updatedAt bump. case "thread.visited": case "thread.tags-updated": + case "thread.specialist-updated": case "thread.marked-unread": return { ...projection, thread: event.payload }; case "run.created": diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 3761d0fa6e71..43c7aca5d40a 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -42,6 +42,7 @@ import { revertThreadCheckpoint, setThreadInteractionMode, setThreadRuntimeMode, + setThreadSpecialist, pinThread, reorderPinnedThread, reorderActiveThread, @@ -204,6 +205,12 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + setSpecialist: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:set-specialist", + execute: setThreadSpecialist, + scheduler, + concurrency, + }), reorderActive: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:reorder-active", execute: reorderActiveThread, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 8dd47df33a21..9952145f61e2 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -48,6 +48,7 @@ export * from "./threadMetadataMcp.ts"; export * from "./threadPullRequest.ts"; export * from "./threadSearch.ts"; export * from "./threadTags.ts"; +export * from "./threadSpecialist.ts"; export * from "./threadTitle.ts"; export * from "./t3ProjectFile.ts"; export * from "./editor.ts"; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 915bb37a001a..09e4846331f1 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -45,6 +45,7 @@ import { AuthMcpClientAccess } from "./auth.ts"; import { ChatAttachment } from "./chatAttachment.ts"; import { ThreadGuestShare } from "./guestShare.ts"; import { StoredThreadTags, ThreadTag } from "./threadTags.ts"; +import { ThreadSpecialist } from "./threadSpecialist.ts"; import { OrchestrationSearchThreadsInput, OrchestrationSearchThreadsResult, @@ -495,6 +496,11 @@ export const OrchestrationV2AppThread = Schema.Struct({ activeOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), /** Absent on threads written before tags existed; read as empty. */ tags: Schema.optional(StoredThreadTags), + /** + * Durable specialist ownership for this listed thread. Absent when unbound. + * Keys are unique per project among non-deleted threads. + */ + specialist: Schema.optional(ThreadSpecialist), lastVisitedAt: Schema.NullOr(Schema.DateTimeUtc).pipe( Schema.withDecodingDefault(Effect.succeed(null)), ), @@ -1930,6 +1936,7 @@ export const OrchestrationV2DomainEvent = Schema.Union([ "thread.pin-reordered", "thread.active-reordered", "thread.tags-updated", + "thread.specialist-updated", "thread.visited", "thread.marked-unread", "thread.metadata-updated", @@ -2186,6 +2193,8 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ activeOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), /** Absent on threads written before tags existed; read as empty. */ tags: Schema.optional(StoredThreadTags), + /** Absent when the thread is not a durable specialist. */ + specialist: Schema.optional(ThreadSpecialist), lastVisitedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), opens: Schema.optional(LenientThreadOpens), titleRegeneration: Schema.optional( @@ -2825,6 +2834,7 @@ export const OrchestrationV2DomainEventJson = Schema.Union([ "thread.pin-reordered", "thread.active-reordered", "thread.tags-updated", + "thread.specialist-updated", "thread.visited", "thread.marked-unread", "thread.pull-request-synced", @@ -3074,6 +3084,13 @@ export const OrchestrationV2Command = Schema.Union([ tag: ThreadTag, tagged: Schema.Boolean, }), + Schema.Struct({ + type: Schema.Literal("thread.specialist.set"), + commandId: CommandId, + threadId: ThreadId, + /** Bind a specialist, or `null` to clear. */ + specialist: Schema.NullOr(ThreadSpecialist), + }), Schema.Struct({ type: Schema.Literal("thread.pin.reorder"), commandId: CommandId, diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index d1bb9961a0c6..6d3f49585c9e 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -35,6 +35,12 @@ import { import { secretRequestPageUrl } from "./secretRequest.ts"; import { hasOnlyPairedSurrogates } from "./threadMetadataMcp.ts"; import { ThreadTitleRegeneration } from "./threadTitle.ts"; +import { + ThreadSpecialist, + ThreadSpecialistInstructions, + ThreadSpecialistKey, + ThreadSpecialistLabel, +} from "./threadSpecialist.ts"; import { OrchestrationV2Actor, OrchestrationV2CreationSource, @@ -335,6 +341,52 @@ export const OrchestratorMcpEnvironmentTarget = Schema.optional( }), ); +/** Resume or create a project-scoped durable specialist thread (OpenRig-style named role). */ +export const OrchestratorMcpFindOrOpenSpecialistInput = Schema.Struct({ + projectId: OrchestratorMcpProjectTarget, + key: ThreadSpecialistKey.annotate({ + description: "Stable slug, e.g. `dev-ui`. Unique per project among non-deleted threads.", + }), + label: Schema.optional( + ThreadSpecialistLabel.annotate({ + description: "Display name. Default: the key. Used when creating or rebinding.", + }), + ), + instructions: Schema.optional( + ThreadSpecialistInstructions.annotate({ + description: "Domain brief stored on the binding. Applied when creating or rebinding.", + }), + ), + title: Schema.optional( + OrchestratorMcpTitle.annotate({ + description: "Title for a newly created specialist thread. Default: the label.", + }), + ), + prompt: Schema.optional( + OrchestratorMcpPrompt.annotate({ + description: "Optional first message when creating, or follow-up when resuming.", + }), + ), + target: Schema.optional(OrchestratorMcpTarget), + runtimeMode: Schema.optional(OrchestratorMcpRuntimeMode), + interactionMode: Schema.optional(OrchestratorMcpInteractionMode), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), +}); +export type OrchestratorMcpFindOrOpenSpecialistInput = + typeof OrchestratorMcpFindOrOpenSpecialistInput.Type; + +export const OrchestratorMcpFindOrOpenSpecialistResult = Schema.Struct({ + threadId: ThreadId, + link: Schema.optional(Schema.String), + created: Schema.Boolean, + specialist: ThreadSpecialist, + runId: Schema.NullOr(RunId), + status: OrchestratorMcpThreadStatus, + title: Schema.String, +}); +export type OrchestratorMcpFindOrOpenSpecialistResult = + typeof OrchestratorMcpFindOrOpenSpecialistResult.Type; + export const OrchestratorMcpThreadListInput = Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget, projectId: OrchestratorMcpProjectTarget, @@ -343,6 +395,11 @@ export const OrchestratorMcpThreadListInput = Schema.Struct({ Schema.Array(OrchestratorMcpThreadStatus).check(Schema.isMaxLength(10)), ), titleContains: Schema.optional(TrimmedNonEmptyString.check(Schema.isMaxLength(256))), + specialistKey: Schema.optional( + ThreadSpecialistKey.annotate({ + description: "Only threads bound to this durable specialist key.", + }), + ), settled: Schema.optional(Schema.Boolean), snoozed: Schema.optional(Schema.Boolean), includeSubagents: Schema.optional(Schema.Boolean), @@ -381,6 +438,7 @@ export const OrchestratorMcpThreadListItem = Schema.Struct({ snoozedUntil: Schema.NullOr(IsoDateTime), parentThreadId: Schema.NullOr(ThreadId), relationshipToParent: Schema.NullOr(Schema.Literals(["fork", "subagent"])), + specialist: Schema.optional(ThreadSpecialist), itemCount: NonNegativeInt, createdAt: IsoDateTime, updatedAt: IsoDateTime, @@ -459,6 +517,7 @@ export const OrchestratorMcpThreadDetail = Schema.Struct({ settledAt: Schema.NullOr(IsoDateTime), snoozed: Schema.Boolean, snoozedUntil: Schema.NullOr(IsoDateTime), + specialist: Schema.optional(ThreadSpecialist), createdAt: IsoDateTime, updatedAt: IsoDateTime, }); diff --git a/packages/contracts/src/threadSpecialist.test.ts b/packages/contracts/src/threadSpecialist.test.ts new file mode 100644 index 000000000000..d9296ef1d964 --- /dev/null +++ b/packages/contracts/src/threadSpecialist.test.ts @@ -0,0 +1,43 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as Schema from "effect/Schema"; + +import { ThreadSpecialist, ThreadSpecialistKey } from "./threadSpecialist.ts"; + +const decodeKey = Schema.decodeUnknownSync(ThreadSpecialistKey); +const decodeSpecialist = Schema.decodeUnknownSync(ThreadSpecialist); + +describe("ThreadSpecialistKey", () => { + it("accepts lowercase slugs", () => { + assert.equal(decodeKey("dev-ui"), "dev-ui"); + assert.equal(decodeKey("factory"), "factory"); + assert.equal(decodeKey("a"), "a"); + }); + + it("rejects uppercase, spaces, and leading hyphens", () => { + assert.throws(() => decodeKey("Dev-UI")); + assert.throws(() => decodeKey("dev ui")); + assert.throws(() => decodeKey("-dev")); + assert.throws(() => decodeKey("1dev")); + }); +}); + +describe("ThreadSpecialist", () => { + it("requires key and label", () => { + assert.deepEqual( + decodeSpecialist({ + key: "dev-ui", + label: "Dev UI", + instructions: "Owns the web client.", + }), + { + key: "dev-ui", + label: "Dev UI", + instructions: "Owns the web client.", + }, + ); + assert.deepEqual(decodeSpecialist({ key: "dev-ui", label: "Dev UI" }), { + key: "dev-ui", + label: "Dev UI", + }); + }); +}); diff --git a/packages/contracts/src/threadSpecialist.ts b/packages/contracts/src/threadSpecialist.ts new file mode 100644 index 000000000000..a08333f480d4 --- /dev/null +++ b/packages/contracts/src/threadSpecialist.ts @@ -0,0 +1,31 @@ +import * as Schema from "effect/Schema"; + +import { TrimmedNonEmptyString } from "./baseSchemas.ts"; + +/** + * Stable project-scoped handle for a durable specialist thread, like OpenRig's + * `dev-ui@factory`. Lowercase slug so agents can address it without fuzzy title match. + */ +export const ThreadSpecialistKey = TrimmedNonEmptyString.check( + Schema.isMaxLength(64), + Schema.makeFilter((value) => + /^[a-z][a-z0-9-]{0,62}$/.test(value) + ? true + : "A specialist key must be a lowercase slug (letters, digits, hyphens).", + ), +); +export type ThreadSpecialistKey = typeof ThreadSpecialistKey.Type; + +export const ThreadSpecialistLabel = TrimmedNonEmptyString.check(Schema.isMaxLength(128)); +export type ThreadSpecialistLabel = typeof ThreadSpecialistLabel.Type; + +export const ThreadSpecialistInstructions = TrimmedNonEmptyString.check(Schema.isMaxLength(8_000)); +export type ThreadSpecialistInstructions = typeof ThreadSpecialistInstructions.Type; + +export const ThreadSpecialist = Schema.Struct({ + key: ThreadSpecialistKey, + label: ThreadSpecialistLabel, + /** Domain brief kept with the binding; prompt injection lands in a later slice. */ + instructions: Schema.optional(ThreadSpecialistInstructions), +}); +export type ThreadSpecialist = typeof ThreadSpecialist.Type; From 834101ccb5b756cb93b59788d1ae0fc2afdb2ec6 Mon Sep 17 00:00:00 2001 From: wSedlacek Date: Wed, 11 Nov 2026 14:36:18 -0800 Subject: [PATCH 2/3] fix(mcp): harden find_or_open_specialist resume path Preserve omitted specialist fields on resume, reuse create-path identities for accepted retries, enforce caller/mode guards before mutations, and unarchive archived specialists when opening them. Co-authored-by: Cursor --- apps/server/src/mcp/OrchestratorMcpService.ts | 114 ++++++++++++------ 1 file changed, 75 insertions(+), 39 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 6bc913b6a2a0..64d14a2c2121 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2965,18 +2965,22 @@ const make = Effect.gen(function* () { ), findOrOpenSpecialist: Effect.fn("OrchestratorMcpService.findOrOpenSpecialist")( function* (callerScope, input) { - const { parent: listParent } = yield* loadCaller(callerScope, []); + const { parent: listParent, limits } = yield* loadCaller(callerScope, ["runs"]); const projectId = yield* resolveProjectTarget( listParent, input.projectId, "Pass projectId: this MCP client is not running inside a T3 thread.", ); - const label = input.label ?? input.key; - const specialist: ThreadSpecialist = { - key: input.key, - label, - ...(input.instructions === undefined ? {} : { instructions: input.instructions }), - }; + const key = yield* requestKey(input.clientRequestId); + // Same identity as the create path so a retry cannot fall into resume and re-send. + const createCommandId = stableCommandId({ + scope: callerScope, + requestKey: key, + operation: "create-specialist", + }); + const createThreadId = stableThreadId({ scope: callerScope, requestKey: key, index: 0 }); + const isCreateRetry = yield* threadManagement.hasReceipt(createCommandId); + const snapshot = yield* threadManagement .getShellSnapshot({ projectId }) .pipe( @@ -3000,12 +3004,46 @@ const make = Effect.gen(function* () { DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt), )[0]; - const key = yield* requestKey(input.clientRequestId); - - if (existing !== undefined) { + if (existing !== undefined && !(isCreateRetry && existing.id === createThreadId)) { + const prior = existing.specialist; + const specialist: ThreadSpecialist = { + key: input.key, + label: input.label ?? prior?.label ?? input.key, + ...(input.instructions !== undefined + ? { instructions: input.instructions } + : prior?.instructions !== undefined + ? { instructions: prior.instructions } + : {}), + }; const needsRebind = - existing.specialist?.label !== specialist.label || - existing.specialist?.instructions !== specialist.instructions; + prior?.label !== specialist.label || prior?.instructions !== specialist.instructions; + const prompt = input.prompt; + if (needsRebind || prompt !== undefined || existing.archivedAt !== null) { + const target = yield* loadTargetThread(existing.id); + yield* assertLiveCallerForOtherThread(callerScope, listParent, target); + yield* resolveRuntimeMode(limits.runtimeMode, target.thread.runtimeMode); + yield* resolveInteractionMode(limits.interactionMode, target.thread.interactionMode); + } + if (existing.archivedAt !== null) { + yield* threadManagement + .dispatch({ + type: "thread.unarchive", + commandId: stableCommandId({ + scope: callerScope, + requestKey: key, + operation: "unarchive-specialist", + }), + threadId: existing.id, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to reopen archived specialist: ${errorMessage(error)}`, + ), + ), + ); + } if (needsRebind) { yield* threadManagement .dispatch({ @@ -3030,7 +3068,7 @@ const make = Effect.gen(function* () { let runId: RunId | null = existing.latestRunId; let status: OrchestratorMcpFindOrOpenSpecialistResult["status"] = existing.activityRunStatus ?? existing.status; - if (input.prompt !== undefined) { + if (prompt !== undefined) { const sent = yield* threadManagement .sendToThread({ projectId, @@ -3046,7 +3084,7 @@ const make = Effect.gen(function* () { requestKey: key, operation: "specialist-send", }), - text: input.prompt, + text: prompt, attachments: [], mode: "auto", createdBy: "agent", @@ -3076,6 +3114,11 @@ const make = Effect.gen(function* () { } satisfies OrchestratorMcpFindOrOpenSpecialistResult; } + const specialist: ThreadSpecialist = { + key: input.key, + label: input.label ?? input.key, + ...(input.instructions === undefined ? {} : { instructions: input.instructions }), + }; const { scope, parent } = yield* loadThreadCaller(callerScope, "find_or_open_specialist"); yield* assertLiveCallerForOtherProject(scope, parent, projectId); const parentRun = ThreadManagementService.latestActiveRun(parent); @@ -3096,16 +3139,9 @@ const make = Effect.gen(function* () { ); } const providers = yield* loadProviders; - const commandId = stableCommandId({ - scope, - requestKey: key, - operation: "create-specialist", - }); - const threadId = stableThreadId({ scope, requestKey: key, index: 0 }); - const isRetryOfAcceptedCreation = yield* threadManagement.hasReceipt(commandId); - const committedThread = isRetryOfAcceptedCreation + const committedThread = isCreateRetry ? yield* threadManagement - .getThreadShell(threadId) + .getThreadShell(createThreadId) .pipe(Effect.mapError(threadManagementFailure)) : null; const target = yield* resolveTargetRechecking({ @@ -3118,7 +3154,7 @@ const make = Effect.gen(function* () { providerInstanceId: committedThread.modelSelection.instanceId, }, providers, - skipProjectEnablementFilter: isRetryOfAcceptedCreation, + skipProjectEnablementFilter: isCreateRetry, }); const runtimeMode = yield* resolveRuntimeMode( parent.thread.runtimeMode, @@ -3129,11 +3165,11 @@ const make = Effect.gen(function* () { parent.thread.interactionMode, input.interactionMode, ); - yield* assertTargetEnabledForProject(parent, commandId, target); + yield* assertTargetEnabledForProject(parent, createCommandId, target); const title = threadTitle({ parentTitle: parent.thread.title, prompt: input.prompt, - title: input.title ?? label, + title: input.title ?? specialist.label, index: 0, }); if (committedThread === null) { @@ -3142,8 +3178,8 @@ const make = Effect.gen(function* () { type: "thread.create", createdBy: "agent", creationSource: "mcp", - commandId, - threadId, + commandId: createCommandId, + threadId: createThreadId, projectId, title, modelSelection: target.modelSelection, @@ -3168,11 +3204,11 @@ const make = Effect.gen(function* () { .dispatch({ type: "thread.specialist.set", commandId: stableCommandId({ - scope, + scope: callerScope, requestKey: key, operation: "bind-specialist", }), - threadId, + threadId: createThreadId, specialist, }) .pipe( @@ -3181,7 +3217,7 @@ const make = Effect.gen(function* () { ), ); const resolvedModelSelection = (yield* threadManagement - .getThreadRecords(threadId, []) + .getThreadRecords(createThreadId, []) .pipe(Effect.mapError(threadManagementFailure))).thread.modelSelection; if (input.prompt !== undefined) { yield* threadManagement @@ -3190,14 +3226,14 @@ const make = Effect.gen(function* () { createdBy: "agent", creationSource: "mcp", commandId: stableCommandId({ - scope, + scope: callerScope, requestKey: key, operation: "dispatch-specialist", }), - threadId, + threadId: createThreadId, senderThreadId: scope.thread.threadId, messageId: stableMessageId({ - scope, + scope: callerScope, requestKey: key, index: 0, }), @@ -3217,20 +3253,20 @@ const make = Effect.gen(function* () { ), ); } - const projection = yield* loadThreadRecords(threadId, ["runs"]); + const projection = yield* loadThreadRecords(createThreadId, ["runs"]); const run = projection.runs.at(-1); yield* threadManagement .dispatch({ type: "thread.created.record", commandId: stableCommandId({ - scope, + scope: callerScope, requestKey: key, operation: "record-specialist", }), parentThreadId: scope.thread.threadId, parentRunId: parentRun.id, parentNodeId: parentRun.rootNodeId, - targetThreadId: threadId, + targetThreadId: createThreadId, targetRunId: run?.id ?? null, }) .pipe( @@ -3242,9 +3278,9 @@ const make = Effect.gen(function* () { ), ); return { - threadId, + threadId: createThreadId, link: formatThreadLink( - { environmentId: scope.environmentId, threadId }, + { environmentId: scope.environmentId, threadId: createThreadId }, projection.thread.title, ), created: true, From b2046832d1439a7dd9e098ad3006f52667ef27df Mon Sep 17 00:00:00 2001 From: wSedlacek Date: Wed, 11 Nov 2026 15:58:13 -0800 Subject: [PATCH 3/3] feat(web): human specialist bind and retire lifecycle Thread Details can bind, edit, clear, stop-archive, and restore specialists; MCP retire_specialist absorbs grow/shrink stop-archive without a team manager. Co-authored-by: Cursor --- apps/server/src/mcp/OrchestratorMcpService.ts | 124 ++++++++ .../src/mcp/toolkits/orchestrator/handlers.ts | 7 + .../src/mcp/toolkits/orchestrator/tools.ts | 18 +- .../src/provider/agentContextBudget.json | 2 +- .../components/chat/ThreadDetailsPanel.tsx | 8 + .../components/chat/ThreadSpecialistPanel.tsx | 276 ++++++++++++++++++ .../src/components/chat/threadDetailsItems.ts | 1 + packages/contracts/src/orchestratorMcp.ts | 28 ++ 8 files changed, 462 insertions(+), 2 deletions(-) create mode 100644 apps/web/src/components/chat/ThreadSpecialistPanel.tsx diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 64d14a2c2121..7fccaaba681c 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -22,6 +22,8 @@ import { type OrchestratorMcpCreatedThread, type OrchestratorMcpFindOrOpenSpecialistInput, type OrchestratorMcpFindOrOpenSpecialistResult, + type OrchestratorMcpRetireSpecialistInput, + type OrchestratorMcpRetireSpecialistResult, type OrchestratorMcpDelegateTaskInput, type ThreadSpecialist, type OrchestratorMcpDelegateTaskResult, @@ -184,6 +186,10 @@ export interface OrchestratorMcpServiceShape { scope: McpInvocationScope, input: OrchestratorMcpFindOrOpenSpecialistInput, ) => Effect.Effect; + readonly retireSpecialist: ( + scope: McpInvocationScope, + input: OrchestratorMcpRetireSpecialistInput, + ) => Effect.Effect; readonly scheduleTask: ( scope: McpInvocationScope, input: OrchestratorMcpScheduleTaskInput, @@ -3291,6 +3297,124 @@ const make = Effect.gen(function* () { } satisfies OrchestratorMcpFindOrOpenSpecialistResult; }, ), + retireSpecialist: Effect.fn("OrchestratorMcpService.retireSpecialist")( + function* (callerScope, input) { + const { parent: listParent } = yield* loadCaller(callerScope, ["runs"]); + const projectId = yield* resolveProjectTarget( + listParent, + input.projectId, + "Pass projectId: this MCP client is not running inside a T3 thread.", + ); + const key = yield* requestKey(input.clientRequestId); + const snapshot = yield* threadManagement + .getShellSnapshot({ projectId }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to look up specialists: ${errorMessage(error)}`, + ), + ), + ); + const existing = [...snapshot.threads, ...snapshot.archivedThreads] + .filter( + (thread) => + thread.deletedAt === null && + thread.specialist?.key === input.key && + isListedThreadLineage(thread.lineage), + ) + .toSorted( + (left, right) => + (left.archivedAt === null ? 0 : 1) - (right.archivedAt === null ? 0 : 1) || + DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt), + )[0]; + if (existing === undefined || existing.specialist === undefined) { + return yield* failure( + "invalid_request", + `No durable specialist with key '${input.key}' in this project.`, + ); + } + const specialist = existing.specialist; + const link = threadLinkFor(callerScope, existing.id, existing.title); + if (existing.archivedAt !== null) { + return { + threadId: existing.id, + link, + title: existing.title, + specialist, + status: "already_archived", + } satisfies OrchestratorMcpRetireSpecialistResult; + } + if (listParent !== undefined && existing.id === listParent.thread.id) { + return yield* failure( + "invalid_request", + "Cannot retire the calling thread; retire another specialist seat.", + ); + } + const target = yield* loadTargetThread(existing.id); + yield* assertLiveCallerForOtherThread(callerScope, listParent, target); + const stopCommandId = stableCommandId({ + scope: callerScope, + requestKey: key, + operation: "retire-specialist-stop", + }); + const stopFailed = (error: unknown) => + Effect.logWarning("orchestrator-mcp.retire-specialist-stop-failed", { + threadId: existing.id, + error, + }).pipe( + Effect.andThen( + Effect.fail( + failure( + "orchestration_error", + `Unable to stop specialist ${existing.id} before retiring.`, + ), + ), + ), + ); + // Stop only this seat and its own delegates; siblings and the caller keep running. + yield* threadManagement + .dispatch({ + type: "thread.stop", + commandId: stopCommandId, + threadId: existing.id, + ...(input.reason === undefined ? {} : { reason: input.reason }), + }) + .pipe(Effect.catch(stopFailed)); + yield* threadManagement + .stopDelegatedTasks({ + threadId: existing.id, + commandId: stopCommandId, + reason: input.reason, + }) + .pipe(Effect.catch(stopFailed)); + yield* threadManagement + .dispatch({ + type: "thread.archive", + commandId: stableCommandId({ + scope: callerScope, + requestKey: key, + operation: "retire-specialist-archive", + }), + threadId: existing.id, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to archive specialist ${existing.id}: ${errorMessage(error)}`, + ), + ), + ); + return { + threadId: existing.id, + link, + title: existing.title, + specialist, + status: "retired", + } satisfies OrchestratorMcpRetireSpecialistResult; + }, + ), 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 c0fc1d727182..60dd75e29733 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -152,6 +152,13 @@ const handlers = { return yield* service.findOrOpenSpecialist(scope, input); }), ), + retire_specialist: McpToolAccess.actsAsCaller((input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + return yield* service.retireSpecialist(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 851682b19e3b..4fbada781795 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -11,6 +11,8 @@ import { OrchestratorMcpCreateThreadsResult, OrchestratorMcpFindOrOpenSpecialistInput, OrchestratorMcpFindOrOpenSpecialistResult, + OrchestratorMcpRetireSpecialistInput, + OrchestratorMcpRetireSpecialistResult, OrchestratorMcpDelegateTaskInput, OrchestratorMcpDelegateTaskResult, OrchestratorMcpDeleteScheduledTaskInput, @@ -226,7 +228,7 @@ const CreateThreadsTool = Tool.make("create_threads", { const FindOrOpenSpecialistTool = Tool.make("find_or_open_specialist", { description: - "Resume or create a durable specialist thread by stable `key` (e.g. `dev-ui`) in a project. Keeps context across days; not a turn-scoped `delegate_task`. Creating needs an active run. In your reply, name it as its `link` verbatim.", + "Add or restore a durable specialist thread by stable `key` (e.g. `dev-ui`) in a project: resume an existing binding, unarchive a retired seat, or create one. Keeps context across days; not a turn-scoped `delegate_task`. Creating needs an active run. Pair with `retire_specialist` to stop-archive. In your reply, name it as its `link` verbatim.", parameters: OrchestratorMcpFindOrOpenSpecialistInput, success: OrchestratorMcpFindOrOpenSpecialistResult, failure: OrchestratorMcpFailure, @@ -238,6 +240,19 @@ const FindOrOpenSpecialistTool = Tool.make("find_or_open_specialist", { .annotate(Tool.OpenWorld, true) .annotate(Tool.Idempotent, true); +const RetireSpecialistTool = Tool.make("retire_specialist", { + description: + "Stop and archive a durable specialist by `key` without clearing its binding (siblings keep running). Reverse with `find_or_open_specialist`. Not a team manager; membership stays the thread binding.", + parameters: OrchestratorMcpRetireSpecialistInput, + success: OrchestratorMcpRetireSpecialistResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Retire a durable specialist") + .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`.", @@ -378,6 +393,7 @@ export const OrchestratorToolkit = Toolkit.make( RequestSecretTool, CreateThreadsTool, FindOrOpenSpecialistTool, + RetireSpecialistTool, ThreadListTool, ThreadReadTool, ThreadUpdateTool, diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 562ab4fb4645..fde7c0f78378 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5597, "codexAdditionalContextBytes": 7327, - "mcpToolListJsonBytes": 49508 + "mcpToolListJsonBytes": 50048 } diff --git a/apps/web/src/components/chat/ThreadDetailsPanel.tsx b/apps/web/src/components/chat/ThreadDetailsPanel.tsx index bb1546fd4b82..c5f3d847b447 100644 --- a/apps/web/src/components/chat/ThreadDetailsPanel.tsx +++ b/apps/web/src/components/chat/ThreadDetailsPanel.tsx @@ -23,6 +23,7 @@ import { ThreadDetailsSection } from "./ThreadDetailsSection"; import { ThreadAutomationsPanel } from "./ThreadAutomationsPanel"; import { ThreadRelationshipsPanel } from "./ThreadRelationshipsControl"; import { ThreadServersPanel } from "./ThreadServersPanel"; +import { ThreadSpecialistPanel } from "./ThreadSpecialistPanel"; import { ThreadDetailsContext, ThreadDetailsUsage, @@ -244,6 +245,13 @@ export function ThreadDetailsPanel(props: ThreadDetailsPanelProps) { ) : null} + {density === "full" && !props.draftId && visible.specialist ? ( + + ) : null} + {density === "full" && !props.draftId && visible.automations ? ( ; +} + +function ThreadSpecialistEditor(props: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly shell: EnvironmentThreadShell; +}) { + const threadRef = scopeThreadRef(props.environmentId, props.threadId); + const canOperate = readEnvironmentScope(props.environmentId, AuthOrchestrationOperateScope); + const setSpecialist = useOrchestrationCommand(threadEnvironment.setSpecialist, { + label: "thread specialist bind", + }); + const interruptTurn = useOrchestrationCommand(threadEnvironment.interruptTurn, { + label: "thread specialist stop", + reportFailure: false, + }); + const { archiveThread, unarchiveThread } = useThreadActions(); + const bound = props.shell.source.specialist; + const [fields, setFields] = useState(() => specialistFields(bound)); + const [busy, setBusy] = useState(false); + + const archived = props.shell.archivedAt !== null; + const dirty = + fields.key !== (bound?.key ?? "") || + fields.label !== (bound?.label ?? "") || + fields.instructions !== (bound?.instructions ?? ""); + const keyValid = KEY_PATTERN.test(fields.key.trim()); + const labelValue = fields.label.trim() || fields.key.trim(); + const canSave = + canOperate && !busy && keyValid && labelValue.length > 0 && (dirty || bound === undefined); + + const reportFailure = (title: string, error: unknown) => { + toastManager.add( + stackedThreadToast({ + type: "error", + title, + description: error instanceof Error ? error.message : String(error), + }), + ); + }; + + const save = async () => { + if (!canSave) return; + setBusy(true); + const specialist: ThreadSpecialist = { + key: fields.key.trim(), + label: labelValue, + ...(fields.instructions.trim() === "" ? {} : { instructions: fields.instructions.trim() }), + }; + const result = await setSpecialist({ + environmentId: props.environmentId, + input: { threadId: props.threadId, specialist }, + }); + setBusy(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + reportFailure("Could not bind specialist", squashAtomCommandFailure(result)); + } + }; + + const clear = async () => { + if (!canOperate || busy || bound === undefined) return; + setBusy(true); + const result = await setSpecialist({ + environmentId: props.environmentId, + input: { threadId: props.threadId, specialist: null }, + }); + setBusy(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + reportFailure("Could not clear specialist", squashAtomCommandFailure(result)); + } + }; + + const stopAndArchive = async () => { + if (!canOperate || busy || bound === undefined || archived) return; + setBusy(true); + if (!threadRuntimeCanArchive(props.shell.runtime)) { + const stopped = await interruptTurn({ + environmentId: props.environmentId, + input: { threadId: props.threadId }, + }); + if (stopped._tag === "Failure" && !isAtomCommandInterrupted(stopped)) { + setBusy(false); + reportFailure("Could not stop specialist", squashAtomCommandFailure(stopped)); + return; + } + } + const archivedResult = await archiveThread(threadRef); + setBusy(false); + if (archivedResult._tag === "Failure" && !isAtomCommandInterrupted(archivedResult)) { + reportFailure("Could not archive specialist", squashAtomCommandFailure(archivedResult)); + } + }; + + const restore = async () => { + if (!canOperate || busy || bound === undefined || !archived) return; + setBusy(true); + const result = await unarchiveThread(threadRef); + setBusy(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + reportFailure("Could not restore specialist", squashAtomCommandFailure(result)); + } + }; + + return ( + +
+
+ + + {bound === undefined + ? "Bind a durable role to this thread. Agents resume it by key." + : archived + ? `Archived @${bound.key}. Restore to reopen, or clear the binding.` + : `@${bound.key} · ${bound.label}`} + +
+ + + + + +