diff --git a/apps/mobile/src/features/threads/ThreadHeader.tsx b/apps/mobile/src/features/threads/ThreadHeader.tsx index 0388bfb299299..62458855aea3f 100644 --- a/apps/mobile/src/features/threads/ThreadHeader.tsx +++ b/apps/mobile/src/features/threads/ThreadHeader.tsx @@ -21,6 +21,8 @@ export function ThreadHeader(props: { readonly hasThreadCwd: boolean; readonly inspectorMode: ThreadInspectorMode | null; readonly onOpenFilesInspector: () => void; + /** Present only on a delegated child the user may pop out. */ + readonly onPopOut?: () => void; }) { const goBack = useWorkspaceBackNavigation(); const { toggleAuxiliaryPane, startNewTask } = useAdaptiveWorkspaceLayout(); @@ -40,6 +42,16 @@ export function ThreadHeader(props: { }, ] : []; + if (props.onPopOut) { + actions.push({ + accessibilityLabel: "Pop out to top level", + icon: "arrow.up.right.circle", + subtitle: "Make it an ordinary thread", + overflowOnly: true, + separateBackground: false, + onPress: props.onPopOut, + }); + } const menus = [gitMenu, terminalMenu].filter((menu): menu is AppBarMenu => menu !== null); return ( diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 51e1e7182ae5a..9bda468b09c96 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -90,7 +90,10 @@ import { useSelectedThreadGitState } from "../../state/use-selected-thread-git-s import { useSelectedThreadRequests } from "../../state/use-selected-thread-requests"; import { useSelectedThreadWorktree } from "../../state/use-selected-thread-worktree"; import { useThreadComposerState } from "../../state/use-thread-composer-state"; -import { resolveMergeBackTargetThreadId } from "@t3tools/client-runtime/state/thread-relationships"; +import { + resolveMergeBackTargetThreadId, + threadPopOutAvailability, +} from "@t3tools/client-runtime/state/thread-relationships"; import { resolveLatestMergeBackRun } from "@t3tools/client-runtime/state/thread-workflows"; import { threadEnvironment } from "../../state/threads"; import { EMPTY_ENVIRONMENT_THREAD_STATE } from "@t3tools/client-runtime/state/threads"; @@ -447,6 +450,44 @@ function ThreadRouteContent( const navigation = useNavigation(); const selectDetail = useSelectWorkspaceDetail(); const openPullRequest = useOpenPullRequest(); + const popOut = useAtomCommand(threadEnvironment.popOut, { reportFailure: false }); + const popOutBusyRef = useRef(false); + const popOutAvailability = + selectedThread && canOperateThread + ? threadPopOutAvailability(selectedThread.source) + : "unsupported"; + const handlePopOut = () => { + if (!selectedThread || popOutBusyRef.current) return; + if (popOutAvailability === "blocked") { + Dialog.alert("Can't pop out yet", "Wait for it to finish or stop it first."); + return; + } + const environmentId = selectedThread.environmentId; + const threadId = selectedThread.id; + Dialog.confirm({ + title: "Pop out to top level?", + message: + "It becomes an ordinary thread in your list and stops reporting to the thread that started it.", + confirmText: "Pop out", + onConfirm: () => { + if (popOutBusyRef.current) return; + popOutBusyRef.current = true; + void popOut({ environmentId, input: { threadId } }) + .then((result) => { + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + Dialog.alert( + "Could not pop out thread", + error instanceof Error ? error.message : "Try again.", + ); + } + }) + .finally(() => { + popOutBusyRef.current = false; + }); + }, + }); + }; const mergeBack = useAtomCommand(threadEnvironment.mergeBack, "merge thread back"); const mergeBackTargetThreadId = resolveMergeBackTargetThreadId(selectedThreadDetail); const mergeBackRun = @@ -983,6 +1024,7 @@ function ThreadRouteContent( inspectorMode={inspectorMode} onOpenFilesInspector={handleOpenFilesInspector} onReturnToThread={props.onReturnToThread} + onPopOut={popOutAvailability === "unsupported" ? undefined : handlePopOut} /> ) } diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 3f8c8420e8630..e5d65c8e95b50 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2751,6 +2751,22 @@ const make = Effect.gen(function* () { } return yield* remoteDelegation.cancel(scope, remoteTask, input); } + // Pop-out needs every task in the child to have ended, and it leaves this thread's control. + const childShell = yield* threadManagement + .getThreadShell(childThreadId) + .pipe(Effect.mapError(threadManagementFailure)); + if (childShell?.lineage.detachedFrom?.threadId === scope.thread.threadId) { + if (!isTerminalTaskStatus(current.status)) { + return yield* failure( + "invalid_request", + `Thread ${childThreadId} was popped out to top level and no longer belongs to this task.`, + ); + } + return { + taskId: input.taskId, + status: current.status, + } satisfies OrchestratorMcpTaskCancelResult; + } const key = yield* requestKey(input.clientRequestId); const parentProjection = yield* loadThreadRecords(scope.thread.threadId, []); // Cancelling stops the child and every task under it, each a write to a diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 35a18f5f0b460..bd29856af842f 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -605,6 +605,60 @@ it.effect("rejects an unknown organize target instead of archiving the calling t ); }); +it.effect("allows an agent to pop out only its own child", () => { + const dispatched: Array = []; + return Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const otherThreadId = ThreadId.make("mcp-core-child"); + for (const target of [undefined, threadId, otherThreadId, ThreadId.make("unrelated")]) { + const result = yield* server + .callTool({ + name: "t3_thread_organize", + arguments: { action: "pop_out", ...(target === undefined ? {} : { threadId: target }) }, + }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + ); + if (target === otherThreadId) expect(result.isError).not.toBe(true); + else + expect(declaredFailure(result)).toMatchObject({ + code: "invalid_request", + message: expect.stringContaining("cannot pop itself out"), + }); + } + expect(dispatched).toMatchObject([{ type: "thread.detach", threadId: otherThreadId }]); + }).pipe( + Effect.provide( + layerThreadToolkit.pipe( + Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (id) => Effect.succeed(McpToolAccessTestkit.liveThreadShell(id)), + getProjectThreadRecords: ({ threadId: id }) => { + const shell = McpToolAccessTestkit.liveThreadShell(id); + return Effect.succeed( + McpToolAccessTestkit.idleThreadProjection({ + ...shell, + lineage: { + ...shell.lineage, + parentThreadId: id === ThreadId.make("mcp-core-child") ? threadId : null, + }, + }), + ); + }, + dispatch: (command) => { + dispatched.push(command); + return Effect.succeed({ sequence: 1 } as never); + }, + }), + ), + ), + ), + ); +}); + it.effect("keeps unexpected handler defects private through the production registration", () => Effect.gen(function* () { const server = yield* McpServer.McpServer; diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index 88bb91bf62b77..783296b87e790 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -344,7 +344,7 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { ), t3_thread_organize: writesThread((input) => Effect.gen(function* () { - const { threads, projection, caller } = yield* readThread(input.threadId); + const { threads, projection, caller, scope } = yield* readThread(input.threadId); const common = { commandId: yield* newCommandId(), threadId: projection.thread.id }; if (input.action === "settle") { return yield* threads @@ -369,6 +369,21 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { case "mark_unread": command = { ...common, type: "thread.mark-unread" }; break; + case "pop_out": + // Otherwise a child could leave its parent's Stop on its own. + if ( + scope.client?.senderThread !== undefined || + (caller !== undefined && + (caller.id === projection.thread.id || + caller.id !== projection.thread.lineage.parentThreadId)) + ) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "A thread cannot pop itself out; its parent or the user can.", + }); + } + command = { ...common, type: "thread.detach" }; + break; default: command = { ...common, type: `thread.${input.action}` }; } diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index bea03c83c8465..734e5acf5e836 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -32,7 +32,7 @@ import * as McpInvocationContext from "../../McpInvocationContext.ts"; const ThreadOrganizeTool = Tool.make("t3_thread_organize", { description: - "Pin, snooze, settle, archive, or mark a thread unread. Omit `threadId` for this thread. `snooze` needs `snoozedUntil`. Settling this thread applies at turn end. A failed turn or queued message leaves it active.", + "Pin, snooze, settle, archive, mark unread, or pop out a finished child. Omit `threadId` for this thread. `snooze` needs `snoozedUntil`. Settling this thread waits for a clean turn end and empty queue.", parameters: Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId), @@ -46,6 +46,7 @@ const ThreadOrganizeTool = Tool.make("t3_thread_organize", { "archive", "unarchive", "mark_unread", + "pop_out", ]), snoozedUntil: Schema.optional(IsoDateTime), }), diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index e26487fdf2667..361c1e3c45610 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -20,6 +20,7 @@ import { ProviderSessionId, ProviderThreadId, RunId, + isListedThreadLineage, RuntimeRequestId, ThreadId, TurnItemId, @@ -1354,6 +1355,14 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ); }); +const refusalOf = (dispatch: Effect.Effect) => + dispatch.pipe( + Effect.flip, + Effect.map((error) => + error._tag === "OrchestratorDispatchError" ? String(error.cause) : error._tag, + ), + ); + const reconcileCommandId = (name: string) => CommandId.make(`command:runtime-reconcile:${name}`); const parentProviderThreadId = (threadId: ThreadId) => @@ -2434,6 +2443,184 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }), ); + it.effect("pops a reported child out in place, and its later turns stay its own", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make("thread:pop-out-parent"); + const projectId = ProjectId.make("project:pop-out"); + const parentRunId = RunId.make("run:pop-out-parent"); + const rootNodeId = NodeId.make("node:pop-out-parent"); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make("node:pop-out-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: "pop-out-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + const popOut = (name: string) => + orchestrator.dispatch({ + type: "thread.detach", + commandId: CommandId.make(`command:pop-out:${name}`), + threadId: child.childThreadId, + }); + assert.include( + yield* refusalOf(popOut("unreported")), + "has a delegated task whose result has not reached its parent", + ); + yield* orchestrator.recoverDelegatedTasks; + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make("command:pop-out:acknowledge"), + parentThreadId, + taskId: child.taskId, + observedByRunId: parentRunId, + }); + yield* popOut("reported"); + + const popped = yield* orchestrator.getThreadProjection(child.childThreadId); + assert.deepStrictEqual(popped.thread.lineage, { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: parentThreadId, + detachedFrom: { threadId: parentThreadId }, + }); + assert.deepStrictEqual(yield* orchestrator.listAppOwnedChildThreadIds(parentThreadId), []); + const shell = (yield* orchestrator.getShellSnapshot()).threads.find( + (row) => row.id === child.childThreadId, + ); + assert.isTrue(shell !== undefined && isListedThreadLineage(shell.lineage)); + + // The old parent's agent can still message it, but that starts no task of its own. + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:pop-out:followup"), + threadId: child.childThreadId, + senderThreadId: parentThreadId, + messageId: MessageId.make("message:pop-out:followup"), + text: "Keep going on your own.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + const followup = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( + (run) => run.ordinal === 2, + ); + assert.isDefined(followup); + assert.isUndefined(followup?.delegatedTaskId); + const tasks = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.filter( + (task) => task.childThreadId === child.childThreadId, + ); + assert.deepStrictEqual( + tasks.map((task) => [task.id, task.status, task.completionDelivery?.state]), + [[child.taskId, "completed", "acknowledged"]], + ); + }), + ); + + it.effect( + "refuses to pop out a working child, a provider's subagent, or a top-level thread", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make("thread:pop-out-refused-parent"); + const projectId = ProjectId.make("project:pop-out-refused"); + const parentRunId = RunId.make("run:pop-out-refused-parent"); + const rootNodeId = NodeId.make("node:pop-out-refused-parent"); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make("node:pop-out-refused-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: "pop-out-refused-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make("command:pop-out-refused:acknowledge"), + parentThreadId, + taskId: child.taskId, + observedByRunId: parentRunId, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:pop-out-refused:followup"), + threadId: child.childThreadId, + senderThreadId: parentThreadId, + messageId: MessageId.make("message:pop-out-refused:followup"), + text: "One more thing.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + const providerChildThreadId = ThreadId.make("thread:pop-out-refused-native"); + const created = (yield* orchestrator.getThreadProjection(child.childThreadId)).thread; + yield* sink.write({ + events: [ + { + id: EventId.make("event:pop-out-refused-native"), + type: "thread.created", + threadId: providerChildThreadId, + occurredAt: now, + payload: { + ...created, + id: providerChildThreadId, + creationSource: "provider", + forkedFrom: null, + }, + }, + ], + }); + const refusals = []; + for (const threadId of [child.childThreadId, providerChildThreadId, parentThreadId]) { + refusals.push( + yield* refusalOf( + orchestrator.dispatch({ + type: "thread.detach", + commandId: CommandId.make(`command:pop-out-refused:${threadId}`), + threadId, + }), + ), + ); + } + assert.deepStrictEqual(refusals, [ + `Thread ${child.childThreadId} is still working. Wait for it to finish or stop it first.`, + `Thread ${providerChildThreadId} is run by its provider's own subagent tool and cannot be popped out.`, + `Thread ${parentThreadId} is not a delegated child thread.`, + ]); + }), + ); + it.effect("checkpoint retry releases a pending sibling after its wake offer failed", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 61929131e9d72..069a57da71eaf 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -28,6 +28,8 @@ import { type GuestMessageId, isCompactionCommandMessage, isOrchestrationV2WorkActive, + isDelegatedChildThread, + isDelegatedTaskReported, isProviderNativeSubagentThread, MessageId, type NodeId, @@ -387,6 +389,10 @@ export interface OrchestratorV2DispatchResult { } export interface OrchestratorV2Shape { + readonly withDelegatedTasksStopLock: ( + threadId: ThreadId, + walk: (dispatch: OrchestratorV2Shape["dispatch"]) => Effect.Effect, + ) => Effect.Effect; readonly resumeQueuedRuns: Effect.Effect; readonly recoverDelegatedTasks: Effect.Effect; readonly recoverDelegatedTask: ( @@ -633,6 +639,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "thread.snooze": case "thread.unsnooze": case "thread.auto-settle.set": + case "thread.detach": case "thread.pin": case "thread.unpin": case "thread.pin.reorder": @@ -5488,6 +5495,95 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); + // Only finished, reported work may leave: an in-flight task's result would have nowhere to go. + const dispatchThreadDetach = Effect.fn("orchestrationV2.dispatch.threadDetach")(function* ( + command: Extract, + events: Ref.Ref>, + ) { + const refuse = (cause: string) => + new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} ${cause}`, + }); + const child = yield* loadProjectionForCommand(command, [ + "runs", + "runtimeRequests", + "providerThreads", + "providerSessions", + "subagents", + ]); + const thread = child.thread; + if (thread.deletedAt !== null) return yield* refuse("is deleted."); + if (thread.archivedAt !== null) return yield* refuse("is archived."); + if (isProviderNativeSubagentThread(thread)) { + return yield* refuse("is run by its provider's own subagent tool and cannot be popped out."); + } + const parentThreadId = thread.lineage.parentThreadId; + if (!isDelegatedChildThread(thread) || parentThreadId === null) { + return yield* refuse("is not a delegated child thread."); + } + if ( + threadHasUnfinishedWork(child) || + child.providerThreads.some((thread) => (thread.pendingBackgroundTasks?.length ?? 0) > 0) || + child.subagents.some((task) => isOrchestrationV2WorkActive(task.status)) + ) { + return yield* refuse("is still working. Wait for it to finish or stop it first."); + } + for (const session of child.providerSessions) { + if (yield* sessionHasBackgroundWork(session, child.providerThreads)) { + return yield* refuse("is still working. Wait for it to finish or stop it first."); + } + } + const tasks = (yield* projectionStore + .getThreadRecords(parentThreadId, ["subagents"]) + .pipe(mapDispatchError(command))).subagents.filter( + (task) => task.origin === "app_owned" && task.childThreadId === thread.id, + ); + const spawningRunIds = tasks.flatMap((task) => + task.completionDelivery === undefined && task.runId !== null ? [task.runId] : [], + ); + const liveSpawningRunIds = new Set( + spawningRunIds.length === 0 + ? [] + : (yield* projectionStore + .getThreadRecords(parentThreadId, ["runs"], { runIds: spawningRunIds }) + .pipe(mapDispatchError(command))).runs + .filter(isLiveRun) + .map((run) => run.id), + ); + if ( + tasks.some( + (task) => + !isDelegatedTaskReported(task, task.runId !== null && liveSpawningRunIds.has(task.runId)), + ) + ) { + return yield* refuse( + "has a delegated task whose result has not reached its parent. Wait for it to finish or stop it first.", + ); + } + const now = yield* DateTime.now; + yield* emit( + events, + command, + )({ + type: "thread.detached", + threadId: thread.id, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { + ...thread, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: thread.lineage.rootThreadId, + detachedFrom: { threadId: parentThreadId }, + }, + updatedAt: now, + }, + }); + }); + const dispatchProviderSessionDetach = Effect.fn("orchestrationV2.dispatch.providerSessionDetach")( function* ( command: Extract, @@ -13694,6 +13790,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); if (projection.thread.deletedAt !== null) return undefined; + if ( + command.expectedParentThreadId !== undefined && + projection.thread.lineage.parentThreadId !== command.expectedParentThreadId + ) + return undefined; // Queued after the interrupt, since a thread's effects run in order. const stopDelegatedTasks = command.stopDelegatedTasks === false || @@ -14019,8 +14120,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); /** - * Thread lineage and fork origin are immutable, so this - * is safe to read without holding either thread's dispatch lock. + * Safe to read without holding either thread's dispatch lock: fork origin is immutable and + * lineage only ever leaves a parent (pop-out), so a stale read names the parent the child had, + * and work under that parent's lock re-reads the child's lineage. */ const appOwnedSubagentParentThreadId = Effect.fn("Orchestrator.appOwnedSubagentParentThreadId")( function* (childThreadId: ThreadId) { @@ -15070,6 +15172,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.guest-message.dismiss": yield* dispatchGuestShare(command, events, effects); break; + case "thread.detach": + yield* dispatchThreadDetach(command, events); + break; case "provider-session.detach": yield* dispatchProviderSessionDetach(command, events, effects); break; @@ -15683,8 +15788,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? withSubagentParentLock(message.threadId, dispatch) : dispatch, ); - // Answering a message-mode question can also start a follow-up. - return yield* command.type === "runtime-request.respond" + // Answering a message-mode question can also start a follow-up, and a pop-out must not + // race the parent writing a follow-up task for the child. + return yield* command.type === "runtime-request.respond" || command.type === "thread.detach" ? withSubagentParentLock(command.threadId, dispatch) : dispatch; }); @@ -16077,6 +16183,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio { messageRoles: ["user"] }, ) .pipe(mapDispatchError(command)); + if (child.thread.lineage.parentThreadId !== input.parentThreadId) return; const followUpRun = child.runs .filter((run) => run.delegatedTaskId === input.taskId) .toSorted((a, b) => b.ordinal - a.ordinal)[0]; @@ -16133,6 +16240,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio new OrchestratorProjectionError({ threadId: ThreadId.make("thread:shell"), cause }); return OrchestratorV2.of({ + withDelegatedTasksStopLock: Effect.fn("Orchestrator.withDelegatedTasksStopLock")( + (threadId, walk) => + threadDispatch.withLock( + threadId, + walk((command) => + commandThreadId(command) === threadId + ? dispatchWithReceiptEffect(command) + : dispatchWithReceipt(command), + ), + ), + ), resumeQueuedRuns, recoverDelegatedTasks, recoverDelegatedTask, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index a003ade1ae8f4..3e141deaa67d5 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -97,7 +97,7 @@ import type * as Statement from "effect/sql/Statement"; import { MCP_APP_OUTPUT_KEY } from "@t3tools/shared/mcpApp"; import { threadHtmlRenderAttachmentIds } from "../attachmentStore.ts"; -import { readableSearchText } from "./messageSearchIndex.ts"; +import { indexThreadMessages, readableSearchText } from "./messageSearchIndex.ts"; import { isFoldableWorkItem, isThreadHistoryUserTurn, @@ -929,6 +929,7 @@ export function applyToProjection( case "thread.unsettled": case "thread.snoozed": case "thread.unsnoozed": + case "thread.detached": case "thread.auto-settle-set": case "thread.pinned": case "thread.unpinned": @@ -2317,6 +2318,7 @@ export const layer: Layer.Layer = case "thread.unsettled": case "thread.snoozed": case "thread.unsnoozed": + case "thread.detached": case "thread.auto-settle-set": case "thread.pinned": case "thread.unpinned": @@ -2380,6 +2382,7 @@ export const layer: Layer.Layer = deleted_at = excluded.deleted_at, payload_json = excluded.payload_json `; + if (event.type === "thread.detached") yield* indexThreadMessages(sql, event.threadId); break; } case "run.created": @@ -3134,6 +3137,7 @@ export const layer: Layer.Layer = event.type !== "thread.unsettled" && event.type !== "thread.snoozed" && event.type !== "thread.unsnoozed" && + event.type !== "thread.detached" && event.type !== "thread.auto-settle-set" && event.type !== "thread.pinned" && event.type !== "thread.unpinned" && @@ -6249,8 +6253,13 @@ export const layer: Layer.Layer = threadId, ) => sql<{ readonly child_thread_id: string }>`SELECT child_thread_id - FROM orchestration_v2_projection_subagents + FROM orchestration_v2_projection_subagents AS task WHERE thread_id = ${threadId} AND origin = 'app_owned' AND child_thread_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_threads AS child + WHERE child.thread_id = task.child_thread_id + AND json_extract(child.payload_json, '$.lineage.detachedFrom.threadId') = ${threadId} + ) ORDER BY COALESCE(started_at, ''), subagent_id ASC`.pipe( Effect.map((rows) => [...new Set(rows.map((row) => ThreadId.make(row.child_thread_id)))]), Effect.mapError(controlReadError(threadId)), diff --git a/apps/server/src/orchestration-v2/ShellStream.test.ts b/apps/server/src/orchestration-v2/ShellStream.test.ts index fbc610235dd01..8735412407d6d 100644 --- a/apps/server/src/orchestration-v2/ShellStream.test.ts +++ b/apps/server/src/orchestration-v2/ShellStream.test.ts @@ -86,6 +86,16 @@ describe("subagent shell events", () => { ).toBe(true); expect(subagents.has("new-child")).toBe(true); }); + + it("publishes a popped-out child's transcript once it leaves its parent", () => { + const subagents = new Set(["child"]); + expect(shouldPublishShellApplicationEvent(stored("child", "thread.detached"), subagents)).toBe( + true, + ); + expect( + shouldPublishShellApplicationEvent(stored("child", "turn-item.updated"), subagents), + ).toBe(true); + }); }); it.effect("coalesces a slow shell subscriber by identity and location in sequence order", () => diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 5ee1005b16830..01f9ecb037e06 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -544,6 +544,10 @@ export function shouldPublishShellApplicationEvent( } return true; } + if (event.type === "thread.detached") { + subagentThreadIds.delete(event.threadId); + return true; + } if (event.type === "thread.deleted") { // Replay and the live tail overlap; keep known IDs so late events stay filtered. return true; diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 611d9a3244f24..96a6a36020f59 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1611,7 +1611,16 @@ const make = Effect.gen(function* () { const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = Effect.fn( "ThreadManagementService.stopDelegatedTasks", )(function* (input) { + if (input.remoteDispatch === undefined) { + return yield* orchestrator.withDelegatedTasksStopLock(input.threadId, (remoteDispatch) => + stopDelegatedTasks({ ...input, remoteDispatch }), + ); + } const { subagents } = yield* orchestrator.getThreadRecords(input.threadId, ["subagents"]); + // A child popped out to top level no longer answers to this parent's Stop. + const attachedChildThreadIds = new Set( + yield* orchestrator.listAppOwnedChildThreadIds(input.threadId), + ); const failures: Array< Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError > = []; @@ -1655,17 +1664,26 @@ const make = Effect.gen(function* () { } const threadId = task.childThreadId; // Follow-up tasks reuse a child thread; walk its subtree only once. - if (stoppedChildThreadIds.has(threadId)) continue; + if (stoppedChildThreadIds.has(threadId) || !attachedChildThreadIds.has(threadId)) continue; stoppedChildThreadIds.add(threadId); yield* dispatch({ type: "thread.stop", commandId: CommandId.make(`${input.commandId}:stop:${threadId}`), threadId, stopDelegatedTasks: false, + expectedParentThreadId: input.threadId, ...(input.reason === undefined ? {} : { reason: input.reason }), }).pipe( Effect.andThen( - stopDelegatedTasks({ threadId, commandId: input.commandId, reason: input.reason }), + Effect.gen(function* () { + const child = yield* orchestrator.getThreadRecords(threadId, []); + if (child.thread.lineage.parentThreadId !== input.threadId) return; + yield* stopDelegatedTasks({ + threadId, + commandId: input.commandId, + reason: input.reason, + }); + }), ), Effect.catch((error) => Effect.logWarning("Unable to stop a delegated task", { diff --git a/apps/server/src/orchestration-v2/ThreadSearch.test.ts b/apps/server/src/orchestration-v2/ThreadSearch.test.ts index ad4812cd23d35..27257f9e3e056 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.test.ts @@ -516,6 +516,28 @@ it.layer(layerTest)("ThreadSearch", (it) => { assert.deepEqual((yield* search.search({ query: "ab" })).matches, []); assert.lengthOf((yield* search.search({ query: "renam be" })).matches, 1); assert.lengthOf((yield* search.search({ query: "renam ab" })).matches, 0); + + // Popping the subagent out lists it, along with what it wrote before. + const created = thread(subagent, project, { subagentOf: parent }); + if (created.type !== "thread.created") throw new Error("expected thread.created"); + yield* projections.apply({ + ...created, + id: EventId.make(`detached:${subagent}`), + type: "thread.detached", + payload: { + ...created.payload, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: parent, + detachedFrom: { threadId: parent }, + }, + }, + }); + assert.deepEqual( + (yield* search.search({ query: "search servi" })).matches.map((match) => match.threadId), + [parent, subagent], + ); }), ); diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index fc52191d32b29..10f635bf0c60a 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -611,6 +611,48 @@ it.effect.each(["run.interrupt", "thread.stop"] as const)( }).pipe(Effect.provide(layerTest)), ); +it.effect("a parent's Stop no longer reaches a child popped out of it", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threads = yield* ThreadManagementService.ThreadManagementService; + const parentThreadId = ThreadId.make("thread:stop-popped-parent"); + yield* createWatchingThread(parentThreadId, 1); + yield* send(parentThreadId, "first", "start_immediately"); + const childThreadId = yield* delegate(parentThreadId, "popped child"); + const child = (yield* orchestrator.getThreadProjection(childThreadId)).thread; + const before = yield* threadState(childThreadId); + yield* projections.apply({ + id: EventId.make("event:stop-popped:detached"), + type: "thread.detached", + threadId: childThreadId, + occurredAt: yield* DateTime.now, + payload: { + ...child, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: child.lineage.rootThreadId, + detachedFrom: { threadId: parentThreadId }, + }, + }, + }); + + yield* orchestrator.dispatch({ + type: "thread.stop", + threadId: childThreadId, + commandId: CommandId.make("stop-popped-stale-parent"), + expectedParentThreadId: parentThreadId, + }); + assert.deepEqual(yield* threadState(childThreadId), before); + yield* threads.stopDelegatedTasks({ + threadId: parentThreadId, + commandId: CommandId.make("stop-popped-parent"), + }); + assert.deepEqual(yield* threadState(childThreadId), before); + }).pipe(Effect.provide(layerTest)), +); + it.effect("a repeated Stop that changes nothing still stops the delegated tasks", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/messageSearchIndex.ts b/apps/server/src/orchestration-v2/messageSearchIndex.ts index 4545b2a35ece6..e70c1fa15345a 100644 --- a/apps/server/src/orchestration-v2/messageSearchIndex.ts +++ b/apps/server/src/orchestration-v2/messageSearchIndex.ts @@ -151,6 +151,31 @@ export const recreateMessageSearchTriggers = Effect.fn("MessageSearchIndex.recre }, ); +/** Indexes a thread's messages that `indexedMessageCondition` skipped while it was a subagent. */ +export const indexThreadMessages = Effect.fn("MessageSearchIndex.indexThread")(function* ( + sql: SqlClient.SqlClient, + threadId: string, +) { + const index = yield* sql` + SELECT 1 FROM sqlite_master WHERE name = 'orchestration_v2_message_search' + `; + if (index.length === 0) return; + const text = yield* indexedText(sql); + yield* sql` + DELETE FROM orchestration_v2_message_search WHERE rowid IN ( + SELECT rowid FROM orchestration_v2_projection_messages WHERE thread_id = ${threadId} + ) + `; + yield* sql` + INSERT INTO orchestration_v2_message_search (rowid, text) + SELECT messages.rowid, ${text("messages")} + FROM orchestration_v2_projection_messages AS messages + WHERE messages.thread_id = ${threadId} + AND ${indexedMessageCondition(sql)} + AND json_extract(messages.payload_json, '$.text') IS NOT NULL + `; +}); + /** * Refills the search index from the messages table. The index is keyed by * message rowid, so anything that may renumber or copy rows (VACUUM, a dev diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts index 0d428e1a6e1e0..f33008cb91a2e 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts @@ -150,6 +150,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray ({ feature: "thread.archive" }), "thread.snooze": () => ({ feature: "thread.snooze" }), "thread.pin": () => ({ feature: "thread.pin" }), + "thread.detach": () => ({ feature: "thread.detach" }), "thread.settle": () => ({ feature: "thread.settle" }), "thread.pull-request.link": () => ({ feature: "thread.pull_request.link" }), "thread.pull-request.watch": (command) => diff --git a/apps/web/src/components/chat/ThreadRelationshipsControl.browser.test.tsx b/apps/web/src/components/chat/ThreadRelationshipsControl.browser.test.tsx index 1fc292f7ff1ae..1d1ee6f504f95 100644 --- a/apps/web/src/components/chat/ThreadRelationshipsControl.browser.test.tsx +++ b/apps/web/src/components/chat/ThreadRelationshipsControl.browser.test.tsx @@ -45,6 +45,9 @@ vi.mock(import("../../state/environments"), () => ({ useEnvironment: (() => null) as never, useEnvironmentProviders: (() => new Map()) as never, })); +vi.mock(import("../../hooks/useThreadActions"), () => ({ + usePopOutThread: (() => async () => AsyncResult.success(undefined)) as never, +})); vi.mock(import("../../lib/archivedThreadsState"), () => ({ useArchivedThreadSnapshots: (() => ({ snapshots: [{ environmentId: "environment", snapshot: { threads: state.archivedThreads } }], diff --git a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx index 95c6d8f0966dd..455b265f80136 100644 --- a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx +++ b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx @@ -7,6 +7,10 @@ import { SectionHeaderStatus } from "../ui/collapsible-section-header"; import { SubagentTooltipContent } from "./SubagentTooltipContent"; import { PullRequestGlyph } from "../pullRequest/pullRequestIcons"; import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, +} from "@t3tools/client-runtime/state/runtime"; import { deriveWorkflowGroups, projectedSubagentsToRuntime, @@ -21,6 +25,7 @@ import { threadRelationshipRowStatus, orderWebThreadLineageRows, resolveMergeBackTargetThreadId, + threadPopOutAvailability, type ThreadRelationshipEdge, type ThreadRelationshipWalkRow, } from "@t3tools/client-runtime/state/thread-relationships"; @@ -53,6 +58,7 @@ import { MessagesSquareIcon, MoreHorizontalIcon, PlusIcon, + SquareArrowOutUpRightIcon, SquareIcon, UnplugIcon, } from "lucide-react"; @@ -82,6 +88,8 @@ import { AgentElapsed } from "./AgentElapsed"; import { ThreadRelationshipIcon, threadRelationshipStatusLabel } from "./ThreadRelationshipIcon"; import { ThreadLineageWorkflowRow } from "./ThreadLineageWorkflowRow"; import { openSideChatTab } from "../../lib/sideChatPane"; +import { usePopOutThread } from "../../hooks/useThreadActions"; +import { POP_OUT_BLOCKED_HINT } from "../threadActionMenu.logic"; import { Menu, MenuItem, MenuPopup, MenuTrigger } from "../ui/menu"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; @@ -204,6 +212,9 @@ function relationshipLabel(edge: ThreadRelationshipEdge, currentThreadId: Thread if (edge.kind === "subagent") { return edge.sourceThreadId === currentThreadId ? "Subagent" : "Parent agent"; } + if (edge.kind === "popped_out") { + return edge.sourceThreadId === currentThreadId ? "Popped out" : "Former parent"; + } return edge.sourceThreadId === currentThreadId ? "Fork" : "Parent thread"; } @@ -432,6 +443,20 @@ export function ThreadRelationshipsPanel(props: { const providersByEnvironment = useEnvironmentProviders(); const settings = useAtomValue(serverEnvironment.settingsValueAtom(props.environmentId)); const relationshipShells = useThreadRelationshipShells(ref); + const family = useEnvironmentQuery( + props.showLineage && + projection?.subagents.some( + (task) => + task.origin === "app_owned" && + task.childThreadId !== null && + !relationshipShells.some((thread) => thread.id === task.childThreadId), + ) + ? orchestrationEnvironment.threadFamily({ + environmentId: props.environmentId, + input: { threadId: props.threadId }, + }) + : null, + ); const hiddenKeys = useHiddenThreadKeys(); const hidingAnywhere = useAnyEnvironmentHiding(); const allProjects = useProjects(); @@ -468,6 +493,8 @@ export function ThreadRelationshipsPanel(props: { const mergeErrorToastId = useRef(null); const detachErrorToastId = useRef(null); const [stoppingId, setStoppingId] = useState(null); + const popOutThread = usePopOutThread(); + const [poppingOutId, setPoppingOutId] = useState(null); // Derive graphs after hooks to preserve their compiler cache. const providers = providersByEnvironment.get(props.environmentId); const derivedProviderEntries = deriveProviderInstanceEntries(providers ?? []); @@ -509,6 +536,7 @@ export function ThreadRelationshipsPanel(props: { ); const shells: ReadonlyArray = [ ...relationshipShells, + ...(family.data?.threads ?? []), ...(archivedShells ?? []), ]; const graph = deriveThreadRelationshipGraph({ threads: shells, projection }); @@ -699,6 +727,21 @@ export function ThreadRelationshipsPanel(props: { } }; + const popOut = async (childThreadId: ThreadId) => { + if (poppingOutId !== null) return; + setPoppingOutId(childThreadId); + const result = await popOutThread(scopeThreadRef(props.environmentId, childThreadId)); + setPoppingOutId(null); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + toastManager.add({ + type: "error", + title: "Could not pop out thread", + description: error instanceof Error ? error.message : undefined, + }); + } + }; + const stopWorkflow = async (subagentId: string) => { if (stoppingId !== null) return; setStoppingId(subagentId); @@ -732,9 +775,11 @@ export function ThreadRelationshipsPanel(props: { ? CornerLeftUpIcon : isSubagent ? BotIcon - : node?.thread?.lineage.sideChat === true - ? MessagesSquareIcon - : GitForkIcon; + : edge.kind === "popped_out" + ? SquareArrowOutUpRightIcon + : node?.thread?.lineage.sideChat === true + ? MessagesSquareIcon + : GitForkIcon; const relationship = node?.thread?.lineage.sideChat === true && !isParent ? "Side chat" @@ -755,6 +800,17 @@ export function ThreadRelationshipsPanel(props: { ["pending", "running", "waiting"].includes(agent.nativeStatus))) && agent.startedAt && ["pending", "running", "waiting"].includes(agent.status); + const popOutAvailability = + agent?.origin === "app_owned" && node?.thread != null + ? threadPopOutAvailability( + node.thread, + projected.filter( + (task) => task.origin === "app_owned" && task.childThreadId === threadId, + ), + projection?.runs, + ) + : "unsupported"; + const canPopOut = popOutAvailability !== "unsupported"; const threadTitle = relationshipThreadTitle({ title: workflowGroup?.workflow.workflowName ?? @@ -777,7 +833,7 @@ export function ThreadRelationshipsPanel(props: { workflowGroup?.workflow.status === "running" && agent?.origin === "provider_native" && providerDriver === "claudeAgent"; - const trailingVisibilityClass = (workflowGroup ? canStopWorkflow : canStop) + const trailingVisibilityClass = (workflowGroup ? canStopWorkflow : canStop || canPopOut) ? "group-hover:opacity-0 group-focus-within:opacity-0 pointer-coarse:opacity-0 [@media(hover:none)]:opacity-0" : ""; const project = projects.find((project) => project.id === node?.thread?.projectId); @@ -937,30 +993,63 @@ export function ThreadRelationshipsPanel(props: { ) : ( relationshipLink )} - {canStop && agent ? ( -
- - void stopSubagent(threadId, agent)} - /> - } - > - {stoppingId === threadId ? ( - - ) : ( - - )} - - Stop subagent - + {(canStop && agent) || canPopOut ? ( +
+ {canPopOut ? ( + + { + if (popOutAvailability !== "blocked") void popOut(threadId); + }} + /> + } + > + {poppingOutId === threadId ? ( + + ) : ( + + )} + + + {popOutAvailability === "blocked" + ? `Pop out to top level. ${POP_OUT_BLOCKED_HINT}.` + : "Pop out to top level"} + + + ) : null} + {canStop && agent ? ( + + void stopSubagent(threadId, agent)} + /> + } + > + {stoppingId === threadId ? ( + + ) : ( + + )} + + Stop subagent + + ) : null}
) : null} diff --git a/apps/web/src/components/threadActionMenu.logic.test.ts b/apps/web/src/components/threadActionMenu.logic.test.ts index bd9dfdcdcc4b6..ddfab783f73e1 100644 --- a/apps/web/src/components/threadActionMenu.logic.test.ts +++ b/apps/web/src/components/threadActionMenu.logic.test.ts @@ -74,4 +74,17 @@ describe("buildThreadActionMenuItems", () => { expect(items.find((item) => item.id === "archive")?.disabled).toBe(true); expect(items.find((item) => item.id === "turn-into-project")?.disabled).toBe(true); }); + + it("offers pop-out only for delegated children, disabled with a hint until it can run", () => { + const popOut = (state: Partial) => + buildThreadActionMenuItems({ ...baseState, ...state }).find((item) => item.id === "pop-out"); + expect(popOut({})).toBeUndefined(); + expect(popOut({ popOut: "unsupported" })).toBeUndefined(); + expect(popOut({ popOut: "available" })?.disabled).toBeUndefined(); + expect(popOut({ popOut: "blocked" })).toMatchObject({ + disabled: true, + detail: "Wait for it to finish or stop it first", + }); + expect(popOut({ popOut: "available", canOperate: false })?.disabled).toBe(true); + }); }); diff --git a/apps/web/src/components/threadActionMenu.logic.ts b/apps/web/src/components/threadActionMenu.logic.ts index 1e3eddd0b9389..b8fd00bb9d85c 100644 --- a/apps/web/src/components/threadActionMenu.logic.ts +++ b/apps/web/src/components/threadActionMenu.logic.ts @@ -1,5 +1,8 @@ import type { ContextMenuItem } from "@t3tools/contracts"; import type { SnoozePreset } from "@t3tools/client-runtime/state/thread-settled"; +import type { ThreadPopOutAvailability } from "@t3tools/client-runtime/state/thread-relationships"; + +export const POP_OUT_BLOCKED_HINT = "Wait for it to finish or stop it first"; export type ThreadActionMenuId = | "find" @@ -20,6 +23,7 @@ export type ThreadActionMenuId = | "rename" | "regenerate-title" | "mark-unread" + | "pop-out" | "share-guest" | "copy" | "copy-path" @@ -95,6 +99,7 @@ export interface ThreadActionMenuState { readonly canShareWithGuest?: boolean; /** A No project thread with its own folder, on a server that can promote it. */ readonly canTurnIntoProject?: boolean; + readonly popOut?: ThreadPopOutAvailability; } /** Local navigation, read markers, and copying remain available to read-only clients. */ @@ -179,6 +184,16 @@ export function buildThreadActionMenuItems( ] : []), { id: "mark-unread", label: "Mark unread", icon: "mail-open" }, + ...(state.popOut === "available" || state.popOut === "blocked" + ? [ + { + id: "pop-out" as const, + label: "Pop out to top level", + icon: "square-arrow-out-up-right", + ...(state.popOut === "blocked" ? { disabled: true, detail: POP_OUT_BLOCKED_HINT } : {}), + }, + ] + : []), ...(state.canShareWithGuest ? [{ id: "share-guest" as const, label: "Share with guest", icon: "user-plus" }] : []), diff --git a/apps/web/src/contextMenuFallback.ts b/apps/web/src/contextMenuFallback.ts index 22c0c575b4c84..953fd81ebeb11 100644 --- a/apps/web/src/contextMenuFallback.ts +++ b/apps/web/src/contextMenuFallback.ts @@ -109,6 +109,11 @@ const ICON_PATHS: Record ({ granted: new Set(), effects: [] as string[], completed: deferred(), + child: false, + delivery: "pending" as "pending" | "acknowledged", show: vi.fn< ( items: ReadonlyArray>, @@ -58,11 +60,34 @@ vi.mock("../state/entities", () => ({ worktreePath: null, runtime: null, latestRun: null, + source: { + creationSource: "server", + status: "completed", + pendingRuntimeRequest: null, + lineage: { + parentThreadId: state.child ? "parent" : null, + relationshipToParent: state.child ? "subagent" : null, + }, + }, }), useProjects: () => [{ id: "project", environmentId: "secondary" }], })); vi.mock("../state/environments", () => ({ usePrimaryEnvironmentId: () => "primary" })); vi.mock("../state/threads", () => ({ threadEnvironment: { updateMetadata: "metadata" } })); +vi.mock("../state/orchestration", () => ({ orchestrationEnvironment: { popOutParent: "parent" } })); +vi.mock("../state/use-atom-query-runner", () => ({ + useAtomQueryRunner: () => async () => + AsyncResult.success({ + subagents: [ + { + childThreadId: "thread", + status: "completed", + completionDelivery: { state: state.delivery }, + }, + ], + runs: [], + }), +})); vi.mock("../state/use-atom-command", () => ({ useAtomCommand: () => async () => { recordEffect("metadata"); @@ -120,6 +145,7 @@ vi.mock("./useSettings", () => ({ })); vi.mock("./useThreadActions", () => ({ canTurnThreadIntoProject: () => false, + usePopOutThread: () => async () => AsyncResult.success(undefined), useThreadActions: () => Object.fromEntries( [ @@ -162,10 +188,25 @@ beforeEach(() => { state.granted = new Set(["primary"]); state.effects = []; state.completed = deferred(); + state.child = false; + state.delivery = "pending"; state.show.mockReset().mockResolvedValue(null); }); describe("thread menu permissions", () => { + it.each(["pending", "acknowledged"] as const)( + "pop-out waits for %s completion delivery", + async (delivery) => { + state.granted.add("secondary"); + state.child = true; + state.delivery = delivery; + createMenu().openMenu(position); + await vi.waitFor(() => expect(state.show).toHaveBeenCalledOnce()); + const item = state.show.mock.calls[0]![0].find((item) => item.id === "pop-out"); + expect(item).toBeDefined(); + expect(item?.disabled === true).toBe(delivery === "pending"); + }, + ); it.each(["rename", "regenerate-title", "delete", "pin", "settle", "archive"] as const)( "%s rechecks after the native menu closes", async (action) => { diff --git a/apps/web/src/hooks/useThreadActionMenu.ts b/apps/web/src/hooks/useThreadActionMenu.ts index f96baf908b3d9..2ef40336f5312 100644 --- a/apps/web/src/hooks/useThreadActionMenu.ts +++ b/apps/web/src/hooks/useThreadActionMenu.ts @@ -8,11 +8,7 @@ import { squashAtomCommandFailure, } from "@t3tools/client-runtime/state/runtime"; import { canSnooze, effectiveSnoozed } from "@t3tools/client-runtime/state/thread-settled"; -import { - AuthOrchestrationOperateScope, - type ScopedThreadRef, - type ThreadId, -} from "@t3tools/contracts"; +import { AuthOrchestrationOperateScope, type ScopedThreadRef, ThreadId } from "@t3tools/contracts"; import { useRouter } from "@tanstack/react-router"; import { pickSnoozePreset, resolveSnoozePresets } from "../components/Sidebar.snooze"; @@ -44,10 +40,13 @@ import { } from "@t3tools/client-runtime/state/project-grouping"; import { buildPhysicalToLogicalProjectKeyMap } from "../sidebarProjectGrouping"; import { threadRuntimeCanArchive } from "@t3tools/client-runtime/state/models"; +import { threadPopOutAvailability } from "@t3tools/client-runtime/state/thread-relationships"; +import { orchestrationEnvironment } from "../state/orchestration"; +import { useAtomQueryRunner } from "../state/use-atom-query-runner"; import { useCopyToClipboard } from "./useCopyToClipboard"; import { useNewThreadHandler } from "./useHandleNewThread"; import { useClientSettings } from "./useSettings"; -import { canTurnThreadIntoProject, useThreadActions } from "./useThreadActions"; +import { canTurnThreadIntoProject, usePopOutThread, useThreadActions } from "./useThreadActions"; function failureToast(title: string, error: unknown) { toastManager.add( @@ -89,6 +88,8 @@ export function useThreadActionMenu(input: { markThreadUnread, turnThreadIntoProject, } = useThreadActions(); + const popOutThread = usePopOutThread(); + const loadPopOutParent = useAtomQueryRunner(orchestrationEnvironment.popOutParent); const regenerateThreadTitle = useOrchestrationCommand(threadEnvironment.regenerateTitle, { reportFailure: false, }); @@ -128,6 +129,21 @@ export function useThreadActionMenu(input: { }; const isRegeneratingTitle = thread.titleRegeneration != null; const snoozePresets = resolveSnoozePresets(now, timestampFormat, threadRef.environmentId); + let popOut = threadPopOutAvailability(thread.source); + if (popOut === "available" && thread.source.lineage.parentThreadId !== null) { + const parent = await loadPopOutParent({ + environmentId: threadRef.environmentId, + input: { threadId: ThreadId.make(thread.source.lineage.parentThreadId) }, + }); + popOut = + parent._tag === "Success" + ? threadPopOutAvailability( + thread.source, + parent.value.subagents.filter((task) => task.childThreadId === threadRef.threadId), + parent.value.runs, + ) + : "blocked"; + } const items = buildThreadActionMenuItems({ canOperate: readEnvironmentScope(threadRef.environmentId, AuthOrchestrationOperateScope), canFind: onFind !== undefined, @@ -144,6 +160,7 @@ export function useThreadActionMenu(input: { snoozePresets, canShareWithGuest, canTurnIntoProject: canTurnThreadIntoProject(threadRef), + popOut, }); const clicked = await settlePromise(() => api.contextMenu.show(items, position)); if (clicked._tag === "Failure" || clicked.value === null) return; @@ -256,6 +273,9 @@ export function useThreadActionMenu(input: { case "mark-unread": markThreadUnread(threadRef); return; + case "pop-out": + await reportFailure("Failed to pop out thread", () => popOutThread(threadRef)); + return; case "share-guest": openGuestShareDialog(threadRef); return; diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index 9c5ae260923cd..a401b1b021d2d 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -201,6 +201,17 @@ export async function navigateAfterThreadDeletion(navigate: () => Promise) } } +export function usePopOutThread() { + const popOutThreadMutation = useOrchestrationCommand(threadEnvironment.popOut, { + reportFailure: false, + }); + return (target: ScopedThreadRef) => + popOutThreadMutation({ + environmentId: target.environmentId, + input: { threadId: target.threadId }, + }); +} + function useMarkThreadUnread() { const markThreadUnreadMutation = useAtomCommand(threadEnvironment.markUnread, { reportFailure: false, diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 4c02f1c79baee..75d3462383dcf 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -276,6 +276,13 @@ fill the screen; close the full-screen view to return to the parent and inspecto Stop on a thread also stops the subagents it delegated to. Before archiving the parent, wait for its delegated tasks to finish or cancel them. +To keep working in a delegated thread on its own, choose **Pop out to top level** +from its thread menu or its row in the parent's **Lineage**; in the mobile app it's +in the thread's header, under the overflow menu on Android. It keeps its history, +model, and folder, gets its own sidebar row, and no longer reports to the parent or +stops with it. It is available once the delegated task has finished and its result +reached the parent; stop it first if you can't wait. + Subagent threads started by the agent can't take messages; message the parent thread instead. When such a subagent needs an approval or an answer, the parent thread asks for it. diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 50717d0cafd41..81bce637d0367 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -681,6 +681,17 @@ export const markThreadUnread = Effect.fn("EnvironmentCommands.markThreadUnread" }); }); +export type PopOutThreadInput = ThreadCommandInput; +export const popOutThread = Effect.fn("EnvironmentCommands.popOutThread")(function* ( + input: PopOutThreadInput, +) { + return yield* dispatch({ + type: "thread.detach", + commandId: yield* allocateCommandId(input), + threadId: input.threadId, + }); +}); + export const updateThreadMetadata = Effect.fn("EnvironmentCommands.updateThreadMetadata")( function* (input: UpdateThreadMetadataInput) { const commandId = yield* allocateCommandId(input); diff --git a/packages/client-runtime/src/state/orchestration.ts b/packages/client-runtime/src/state/orchestration.ts index f22ac79383d66..130f2ad107387 100644 --- a/packages/client-runtime/src/state/orchestration.ts +++ b/packages/client-runtime/src/state/orchestration.ts @@ -253,6 +253,12 @@ export function createOrchestrationEnvironmentAtoms( staleTimeMs: 0, idleTtlMs: 0, }), + popOutParent: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:orchestration:pop-out-parent", + tag: ORCHESTRATION_V2_WS_METHODS.getThreadProjection, + staleTimeMs: 0, + idleTtlMs: 0, + }), // Read right before deleting a worktree, so never served from cache. worktreeThreadIds: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:orchestration:worktree-thread-ids", diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index 37da408a1c35b..14f77c200d068 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -307,6 +307,7 @@ export function applyOrchestrationV2ProjectionEvent( case "thread.unsettled": case "thread.snoozed": case "thread.unsnoozed": + case "thread.detached": case "thread.auto-settle-set": case "thread.pinned": case "thread.unpinned": diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 0e9de03f5c066..1204caa79eb48 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -35,6 +35,7 @@ import { forkSideChat, forkThreadFromRun, markThreadUnread, + popOutThread, mergeThreadBack, promoteQueuedRun, resumeThreadQueue, @@ -262,6 +263,12 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + popOut: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:pop-out", + execute: popOutThread, + scheduler, + concurrency, + }), updateMetadata: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:update-metadata", execute: updateThreadMetadata, @@ -424,6 +431,7 @@ export function createThreadEnvironmentAtoms( }; const commands = { ...sent, + popOut: rowAction(sent.popOut), delete: rowAction(sent.delete), archive: rowAction(sent.archive), unarchive: rowAction(sent.unarchive), diff --git a/packages/client-runtime/src/state/threadRelationships.test.ts b/packages/client-runtime/src/state/threadRelationships.test.ts index 90dc68c21dd82..a88eae25e25ad 100644 --- a/packages/client-runtime/src/state/threadRelationships.test.ts +++ b/packages/client-runtime/src/state/threadRelationships.test.ts @@ -7,6 +7,7 @@ import { immediateThreadRelationships, orderWebThreadLineageRows, resolveMergeBackTargetThreadId, + threadPopOutAvailability, threadRelationshipRowStatus, } from "./threadRelationships.ts"; @@ -207,6 +208,88 @@ describe("thread relationships", () => { ); }); + it("shows a popped-out child to its former parent instead of as its subagent", () => { + const parent = ThreadId.make("thread-parent"); + const child = ThreadId.make("thread-popped"); + const graph = deriveThreadRelationshipGraph({ + threads: [ + { + id: parent, + status: "completed", + forkedFrom: null, + lineage: { rootThreadId: parent, parentThreadId: null, relationshipToParent: null }, + }, + { + id: child, + status: "running", + forkedFrom: { type: "node", nodeId: "task" }, + lineage: { + rootThreadId: parent, + parentThreadId: null, + relationshipToParent: null, + detachedFrom: { threadId: parent }, + }, + }, + ] as never, + projection: { + thread: { id: parent }, + subagents: [{ childThreadId: child, status: "completed" }], + contextTransfers: [{ sourceThreadId: parent, targetThreadId: child, status: "consumed" }], + } as never, + }); + const edgeTo = (from: ThreadId) => + immediateThreadRelationships(graph, from).map((row) => [row.edge.kind, row.edge.status]); + expect(edgeTo(parent)).toEqual([["popped_out", "running"]]); + expect(edgeTo(child)).toEqual([["popped_out", "running"]]); + expect(graph.edges.some((edge) => edge.kind === "subagent")).toBe(false); + }); + + it.each([ + ["an idle delegated child whose tasks reported", {}, [], "available"], + ["a running child", { status: "running" }, [], "blocked"], + ["a child waiting on a question", { pendingRuntimeRequest: {} }, [], "blocked"], + [ + "a child whose result has not reached its parent", + {}, + [{ status: "completed", completionDelivery: { state: "pending" } }], + "blocked", + ], + ["a provider's own subagent", { creationSource: "provider" }, [], "unsupported"], + [ + "a terminal task whose spawning run is still live", + {}, + [{ status: "completed", runId: "parent-run" }], + "blocked", + ], + [ + "a top-level thread", + { lineage: { parentThreadId: null, relationshipToParent: null } }, + [], + "unsupported", + ], + ])("pop-out for %s", (_name, overrides, tasks, expected) => { + const thread = { + creationSource: "server", + lineage: { + parentThreadId: "parent", + relationshipToParent: "subagent", + rootThreadId: "parent", + }, + activityRunStatus: null, + status: "idle", + pendingRuntimeRequest: null, + pendingBackgroundTasks: [], + ...overrides, + }; + expect( + threadPopOutAvailability( + thread as never, + tasks as never, + [{ id: "parent-run", status: "running" }] as never, + ), + ).toBe(expected); + }); + it("keeps the live shell when an archived snapshot contains the same thread id", () => { const parent = ThreadId.make("thread-parent"); const staleParent = ThreadId.make("thread-stale-parent"); diff --git a/packages/client-runtime/src/state/threadRelationships.ts b/packages/client-runtime/src/state/threadRelationships.ts index 26ff658a5191b..a546ee6476ef9 100644 --- a/packages/client-runtime/src/state/threadRelationships.ts +++ b/packages/client-runtime/src/state/threadRelationships.ts @@ -1,5 +1,10 @@ import { + isDelegatedChildThread, + isDelegatedTaskReported, + isLiveSubagentShell, isOrchestrationV2WorkActive, + type OrchestrationV2Subagent, + type OrchestrationV2Run, type OrchestrationV2ThreadShell, type ThreadId, } from "@t3tools/contracts"; @@ -7,7 +12,7 @@ import * as DateTime from "effect/DateTime"; import type { ThreadStructure } from "./models.ts"; -export type ThreadRelationshipKind = "parent" | "fork" | "subagent" | "transfer"; +export type ThreadRelationshipKind = "parent" | "fork" | "subagent" | "popped_out" | "transfer"; export interface ThreadRelationshipNode { readonly threadId: ThreadId; @@ -87,6 +92,16 @@ export function deriveThreadRelationshipGraph(input: { ? [...threads, { ...selectedThread, activityRunStatus: null, status: null }] : threads; for (const thread of lineageThreads) { + const detachedFrom = thread.lineage.detachedFrom?.threadId; + if (detachedFrom !== undefined) { + addEdge({ + sourceThreadId: detachedFrom, + targetThreadId: thread.id, + kind: "popped_out", + status: thread.activityRunStatus ?? thread.status, + }); + continue; + } const parentThreadId = thread.forkedFrom?.type === "run" ? thread.forkedFrom.threadId @@ -105,6 +120,7 @@ export function deriveThreadRelationshipGraph(input: { for (const subagent of input.projection.subagents) { if (subagent.childThreadId === null) continue; const child = threadsById.get(subagent.childThreadId); + if (child?.lineage.detachedFrom?.threadId === ownerThreadId) continue; addEdge({ sourceThreadId: ownerThreadId, targetThreadId: subagent.childThreadId, @@ -206,3 +222,41 @@ export function orderWebThreadLineageRows(input: { return left.threadId < right.threadId ? -1 : left.threadId > right.threadId ? 1 : 0; }); } + +export type ThreadPopOutAvailability = "unsupported" | "blocked" | "available"; + +/** + * Whether a delegated child can be popped out to top level: once it is idle and every task it was + * given has reported to its parent. The server re-checks against the parent's records. + */ +export function threadPopOutAvailability( + thread: Pick< + OrchestrationV2ThreadShell, + | "lineage" + | "creationSource" + | "activityRunStatus" + | "status" + | "pendingRuntimeRequest" + | "pendingBackgroundTasks" + >, + tasks: ReadonlyArray< + Pick + > = [], + runs: ReadonlyArray> = [], +): ThreadPopOutAvailability { + if (!isDelegatedChildThread(thread)) return "unsupported"; + return isLiveSubagentShell(thread) || + tasks.some( + (task) => + !isDelegatedTaskReported( + task, + runs.some( + (run) => + run.id === task.runId && + (run.status === "preparing" || run.status === "starting" || run.status === "running"), + ), + ), + ) + ? "blocked" + : "available"; +} diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 44e9aa8290e9a..9eeb805257863 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -169,6 +169,8 @@ export const OrchestrationV2AppThreadLineage = Schema.Struct({ relationshipToParent: Schema.NullOr(Schema.Literals(["fork", "subagent"])), rootThreadId: ThreadId, sideChat: Schema.optional(Schema.Boolean), + /** The parent a delegated child was popped out of; it keeps the parent's `rootThreadId`. */ + detachedFrom: Schema.optional(Schema.Struct({ threadId: ThreadId })), }); export type OrchestrationV2AppThreadLineage = typeof OrchestrationV2AppThreadLineage.Type; @@ -572,6 +574,38 @@ export function isProviderNativeSubagentThread( return thread.lineage.relationshipToParent === "subagent" && thread.creationSource === "provider"; } +/** A T3 delegate_task child still under its parent, the kind that can be popped out to top level. */ +export function isDelegatedChildThread( + thread: Pick, +): boolean { + return ( + thread.lineage.relationshipToParent === "subagent" && + thread.lineage.parentThreadId !== null && + thread.creationSource !== "provider" + ); +} + +/** + * Whether a delegated task ended and its result reached the parent. A task with no delivery + * record leaves its result to the turn that spawned it, so it counts once that turn has ended. + */ +export function isDelegatedTaskReported( + task: Pick, + spawningRunLive: boolean, +): boolean { + if ( + task.status !== "completed" && + task.status !== "failed" && + task.status !== "cancelled" && + task.status !== "interrupted" + ) + return false; + const delivery = task.completionDelivery?.state; + return delivery === undefined + ? !spawningRunLive + : delivery === "acknowledged" || delivery === "delivered" || delivery === "disposed"; +} + export const IDLE_COMPACTION_TEXT = "/compact"; interface CompactionMessageFields { @@ -1991,6 +2025,7 @@ export const OrchestrationV2DomainEvent = Schema.Union([ "thread.unsettled", "thread.snoozed", "thread.unsnoozed", + "thread.detached", "thread.pinned", "thread.auto-settle-set", "thread.unpinned", @@ -2935,6 +2970,7 @@ export const OrchestrationV2DomainEventJson = Schema.Union([ "thread.unsettled", "thread.snoozed", "thread.unsnoozed", + "thread.detached", "thread.pinned", "thread.auto-settle-set", "thread.unpinned", @@ -3082,6 +3118,7 @@ export const OrchestrationV2Command = Schema.Union([ * do under their caller's dispatch limit. Otherwise Stop also stops every task under the thread. */ stopDelegatedTasks: Schema.optional(Schema.Boolean), + expectedParentThreadId: Schema.optional(ThreadId), }), Schema.Struct({ type: Schema.Literal("thread.create"), @@ -3195,6 +3232,11 @@ export const OrchestrationV2Command = Schema.Union([ threadId: ThreadId, enabled: Schema.Boolean, }), + Schema.Struct({ + type: Schema.Literal("thread.detach"), + commandId: CommandId, + threadId: ThreadId, + }), Schema.Struct({ type: Schema.Literal("thread.pin"), commandId: CommandId,