From 769a28c43b5933af646641a2ab530bbd79a19573 Mon Sep 17 00:00:00 2001 From: JJ Eaton <58495321+oldbettie@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:38:42 +0700 Subject: [PATCH] fix(server,web,mobile): agent-launched chats keep their launcher as parent Chats an agent launched sometimes showed top-level on the Agents board: a remote delegate_task or peer-link t3_thread_launch reached the other machine with no calling thread, a chat a delegate_task helper launched pointed at a parent that is never a card, and a detached child kept its task with the launcher. - A peer-forwarded t3_thread_launch always names its caller in delegatedFrom, and the receiving side nests the chat under it. - The board lists a chat launched by a subagent under the subagent's owner, on web, desktop and mobile. - A task follows its worker's parent: it moves with a reparent and goes with a detach, and restart reconciles tasks that drifted earlier. - Naming the calling chat as parent of a chat on another machine resolves to the caller's environment instead of failing. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/toolkits/project/handlers.ts | 13 +++- apps/server/src/peer/PeerForwarding.test.ts | 5 ++ apps/server/src/peer/PeerForwarding.ts | 10 +++ apps/server/src/peer/RemoteDelegation.test.ts | 17 +++++ .../src/threadTask/ThreadTaskService.test.ts | 50 ++++++++++++++ .../src/threadTask/ThreadTaskService.ts | 66 ++++++++++++++++++- apps/server/src/threadTask/ThreadTaskStore.ts | 42 ++++++++++-- .../client-runtime/src/state/agents.test.ts | 35 ++++++++++ packages/client-runtime/src/state/agents.ts | 39 ++++++++++- packages/mcp-gateway/src/tools.test.ts | 13 +++- packages/mcp-gateway/src/tools.ts | 12 ++-- 11 files changed, 281 insertions(+), 21 deletions(-) diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index 542cfe0cf63c7..7885a922f91c6 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -287,9 +287,16 @@ export const layer = McpToolAccess.toLayer( ...(profile === undefined ? {} : { profileSelection: profile.selection }), commandId, threadId, - ...(context.scope.thread === undefined - ? {} - : { parentThreadId: context.scope.thread.threadId }), + // A linked environment's launch reaches here as its link session, not as the + // calling thread, so the thread it stamps in delegatedFrom is the parent. + ...(context.scope.thread !== undefined + ? { parentThreadId: context.scope.thread.threadId } + : linkOrigin !== undefined && input.delegatedFrom !== undefined + ? { + parentThreadId: input.delegatedFrom.threadId, + parentEnvironmentId: input.delegatedFrom.environmentId, + } + : {}), projectId, title: input.title, modelSelection, diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index 258cd8d76a3b0..5bd69a42b79cd 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -573,6 +573,11 @@ it.effect("launches in a linked environment with the caller's modes, retry-safe" // The thread there runs with the caller's own modes here, below the // link's full access, and a retry asks for the same thread. expect(launched).toMatchObject({ runtimeMode: "full-access", interactionMode: "plan" }); + // The box sees only the link, so the launch names the caller to nest under. + expect(launched).toMatchObject({ + parentThreadId: "thread:laptop-plan", + parentEnvironmentId: laptop.environmentId, + }); expect(relaunched?.threadId).toBe(launched?.threadId); expect(retried.structuredContent).toEqual(first.structuredContent); diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts index 075b6ea627a5b..134307f16c158 100644 --- a/apps/server/src/peer/PeerForwarding.ts +++ b/apps/server/src/peer/PeerForwarding.ts @@ -148,6 +148,16 @@ const make = Effect.gen(function* () { if (typeof args.clientRequestId === "string") { args.clientRequestId = yield* scopedRequestId(scope.requestNamespace, args.clientRequestId); } + // The peer sees only the link session, so a launch names its calling thread to nest under. + if (toolName === "t3_thread_launch" && scope.thread !== undefined && !args.delegatedFrom) { + const threadId = scope.thread.threadId; + const shell = yield* threads.getThreadShell(threadId).pipe(Effect.orElseSucceed(() => null)); + args.delegatedFrom = { + environmentId: scope.environmentId, + threadId, + title: shell?.title ?? threadId, + }; + } return yield* peers.call({ environmentId, tool: toolName, diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index e2fdf6dacc2cf..1e5bea2c7ebe1 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -89,6 +89,11 @@ interface BoxThread { readonly linkOrigin: Ref.Ref<{ readonly sessionId: string; readonly label: string } | null>; /** The parent the launch names, so the thread there can link back to it. */ readonly delegatedFrom: Ref.Ref; + /** The parent the launch nests the thread under there. */ + readonly parent: Ref.Ref<{ + readonly parentThreadId: ThreadLaunch.ThreadLaunchInput["parentThreadId"] | undefined; + readonly parentEnvironmentId: ThreadLaunch.ThreadLaunchInput["parentEnvironmentId"] | undefined; + } | null>; /** The model and modes it was launched with. */ readonly launchedWith: Ref.Ref; /** The first message the thread there was launched with. */ @@ -137,6 +142,7 @@ const makeBoxThread = Effect.gen(function* () { null, ), delegatedFrom: yield* Ref.make(null), + parent: yield* Ref.make ? A : never>(null), launchedWith: yield* Ref.make(null), message: yield* Ref.make(null), status: yield* Ref.make("running"), @@ -425,6 +431,12 @@ const boxToolkitLayer = (thread: BoxThread) => { Ref.set(thread.launched, input.threadId!).pipe( Effect.andThen(Ref.set(thread.linkOrigin, input.linkOrigin ?? null)), Effect.andThen(Ref.set(thread.delegatedFrom, input.delegatedFrom ?? null)), + Effect.andThen( + Ref.set(thread.parent, { + parentThreadId: input.parentThreadId, + parentEnvironmentId: input.parentEnvironmentId, + }), + ), Effect.andThen(Ref.set(thread.message, input.initialMessage?.text ?? null)), Effect.andThen( Ref.set(thread.launchedWith, { @@ -821,6 +833,11 @@ it.effect("a task delegated to the box wakes the laptop's parent when it ends th threadId: parentThreadId, title: "Laptop parent", }); + // ...and nests it there under that parent, on the laptop. + expect(yield* Ref.get(thread.parent)).toEqual({ + parentThreadId, + parentEnvironmentId: laptop.environmentId, + }); // A retry with the same key launches nothing new and records nothing new. const retried = yield* delegateToBox(a); expect(retried.taskId).toBe(delegated.taskId); diff --git a/apps/server/src/threadTask/ThreadTaskService.test.ts b/apps/server/src/threadTask/ThreadTaskService.test.ts index 536cd8a53d216..6ce2b34f96766 100644 --- a/apps/server/src/threadTask/ThreadTaskService.test.ts +++ b/apps/server/src/threadTask/ThreadTaskService.test.ts @@ -715,6 +715,56 @@ it.effect("a read after a cursor returns only changed, settled and departed task ).pipe(Effect.provide(testLayer)), ); +it.effect("a task follows its worker's parent: moved, detached, or drifted before a restart", () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture; + const orchestrator = yield* OrchestratorV2; + const store = yield* ThreadTaskStore.ThreadTaskStore; + const tasks = yield* makeService; + yield* tasks.start(); + yield* f.create("captain"); + yield* f.create("stranger"); + yield* f.create("worker", "captain"); + yield* tasks.assign(owner, { threadId: worker.threadId, summary: "Work" }); + const assigned = Option.getOrThrow(yield* store.get(worker.threadId)); + const before = yield* tasks.read(owner, {}); + const moveTo = (parent: string | null, step: string) => + orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`move:${step}`), + threadId: worker.threadId, + parentThreadId: parent === null ? null : ThreadId.make(parent), + }); + const ownerOf = (id: ThreadId) => + store.get(id).pipe(Effect.map((task) => Option.getOrUndefined(task)?.ownerThreadId)); + + // Moved under another chat, the task goes with it and leaves the old owner's delta. + yield* moveTo("stranger", "1"); + yield* tasks.drain; + assert.strictEqual(yield* ownerOf(worker.threadId), stranger.threadId); + assert.deepEqual((yield* tasks.read(owner, { afterCursor: before.cursor })).removed, [ + worker.threadId, + ]); + + // Detached, it is nobody's task. + const moved = yield* tasks.read(stranger, {}); + yield* moveTo(null, "2"); + yield* tasks.drain; + assert.isUndefined(yield* ownerOf(worker.threadId)); + assert.deepEqual((yield* tasks.read(stranger, { afterCursor: moved.cursor })).removed, [ + worker.threadId, + ]); + + // A detached worker whose task an older server left with its launcher, as M1 was, + // loses the task on restart. + yield* store.put(assigned); + yield* tasks.recover; + assert.isUndefined(yield* ownerOf(worker.threadId)); + }), + ).pipe(Effect.provide(testLayer)), +); + it.effect("a captain settles itself after its turn once its children are accepted and quiet", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/threadTask/ThreadTaskService.ts b/apps/server/src/threadTask/ThreadTaskService.ts index 812226505bd88..df6f09d1dc185 100644 --- a/apps/server/src/threadTask/ThreadTaskService.ts +++ b/apps/server/src/threadTask/ThreadTaskService.ts @@ -927,6 +927,41 @@ export const make = Effect.gen(function* () { threadParentRelationship(shell) === "child" && shell.profileSnapshot?.profileId != null; + /** + * The parent link is canonical: a task's owner is always the worker's parent. A child + * moved under another chat here takes its task along; otherwise the task goes. Tasks owned + * on another environment are kept in step by their link instead. + */ + const followParent = Effect.fn("ThreadTaskService.followParent")(function* ( + workerThreadId: ThreadId, + ) { + yield* writes.withPermits(1)( + Effect.gen(function* () { + const task = yield* getTask(workerThreadId); + if (task === undefined || task.ownerEnvironmentId !== null) return; + const worker = yield* getShell(workerThreadId); + if (worker === undefined) return; + const parent = worker.parentEnvironmentId == null ? worker.parentThreadId : null; + if (parent === task.ownerThreadId) return; + if (parent != null && threadParentRelationship(worker) === "child") { + yield* writeAssignment({ + owner: parent, + actor: "server", + worker, + summary: task.summary, + taskId: undefined, + settleWhenAccepted: undefined, + clientRequestId: undefined, + existing: task, + }); + return; + } + yield* store.remove(workerThreadId).pipe(Effect.catch(unavailable)); + yield* PubSub.publish(changes, undefined); + }), + ); + }); + /** * The chat's task, creating it for a named child that has none yet, owned by * its parent. A parent on another environment adopts the task on its first @@ -1763,7 +1798,9 @@ export const make = Effect.gen(function* () { const handle = Effect.fnUntraced( function* (stored: OrchestrationV2StoredEvent) { const event = stored.event; - if (event.type === "run.updated" && TERMINAL_RUN_STATUSES.has(event.payload.status)) { + if (event.type === "thread.metadata-updated") { + yield* followParent(event.threadId); + } else if (event.type === "run.updated" && TERMINAL_RUN_STATUSES.has(event.payload.status)) { yield* evaluateSettlement(event.threadId); yield* reevaluatePendingSettlements; yield* reevaluateSettleRequests; @@ -1799,6 +1836,29 @@ export const make = Effect.gen(function* () { const links = yield* projections.getThreadParentLinks(); // Scan every thread, including owners whose child was deleted or reparented. yield* Effect.forEach(links, (link) => clearQueuedWakes(link.threadId), { discard: true }); + // Tasks whose worker moved while an older server ran follow its parent now. + const parentOf = new Map( + links.map((link) => [ + link.threadId, + link.parentEnvironmentId === null ? link.parentThreadId : null, + ]), + ); + for (const task of tasks) { + if ( + task.ownerEnvironmentId === null && + parentOf.has(task.workerThreadId) && + parentOf.get(task.workerThreadId) !== task.ownerThreadId + ) { + yield* followParent(task.workerThreadId).pipe( + Effect.catch((error) => + Effect.logWarning("thread task could not follow its parent", { + workerThreadId: task.workerThreadId, + error, + }), + ), + ); + } + } // Adopt named children launched before tasks existed for readable task state. for (const link of links) { if ( @@ -1833,7 +1893,9 @@ export const make = Effect.gen(function* () { yield* TxRef.set(seenSequence, after).pipe(Effect.tx); yield* forkParked( Stream.runForEach(eventSink.stream({ afterSequence: after }), (stored) => - (stored.event.type === "run.updated" || stored.event.type === "runtime-request.updated" + (stored.event.type === "run.updated" || + stored.event.type === "runtime-request.updated" || + stored.event.type === "thread.metadata-updated" ? worker.enqueue(stored) : Effect.void ).pipe(Effect.andThen(TxRef.set(seenSequence, stored.sequence).pipe(Effect.tx))), diff --git a/apps/server/src/threadTask/ThreadTaskStore.ts b/apps/server/src/threadTask/ThreadTaskStore.ts index 1c59932330a60..4ff4541364e27 100644 --- a/apps/server/src/threadTask/ThreadTaskStore.ts +++ b/apps/server/src/threadTask/ThreadTaskStore.ts @@ -121,6 +121,8 @@ export class ThreadTaskStore extends Context.Service< ) => Effect.Effect; /** Writes the task under the next cursor and returns it as stored. */ readonly put: (task: ThreadTask) => Effect.Effect; + /** Drops the worker's task; its owner's delta reads it as a departure. */ + readonly remove: (workerThreadId: ThreadId) => Effect.Effect; } >()("t3/threadTask/ThreadTaskStore") {} @@ -131,8 +133,12 @@ const make = Effect.gen(function* () { const decodeRows = (operation: string) => (rows: ReadonlyArray<{ readonly payload: string }>) => Effect.forEach(rows, (row) => decodeTask(row.payload)).pipe(Effect.mapError(fail(operation))); + // A removed task's departure can hold the latest cursor, so cursors never repeat. const latestCursor = sql<{ readonly cursor: number }>` - SELECT COALESCE(MAX(cursor), 0) AS cursor FROM thread_tasks + SELECT MAX( + (SELECT COALESCE(MAX(cursor), 0) FROM thread_tasks), + (SELECT COALESCE(MAX(cursor), 0) FROM thread_task_departures) + ) AS cursor `.pipe( Effect.map((rows) => Number(rows[0]?.cursor ?? 0)), Effect.mapError(fail("latest-cursor")), @@ -311,19 +317,41 @@ const make = Effect.gen(function* () { VALUES (${stored.workerThreadId}, ${previousOwner}, ${cursor}) ON CONFLICT (worker_thread_id, owner_thread_id) DO UPDATE SET cursor = excluded.cursor `.pipe(Effect.mapError(fail("put"))); - // Back with an earlier owner: its delta shows the task again, not a departure. - yield* sql` - DELETE FROM thread_task_departures - WHERE worker_thread_id = ${stored.workerThreadId} - AND owner_thread_id = ${stored.ownerThreadId} - `.pipe(Effect.mapError(fail("put"))); } + // Back with an earlier owner: its delta shows the task again, not a departure. + yield* sql` + DELETE FROM thread_task_departures + WHERE worker_thread_id = ${stored.workerThreadId} + AND owner_thread_id = ${stored.ownerThreadId} + `.pipe(Effect.mapError(fail("put"))); return stored; }).pipe( sql.withTransaction, Effect.catchTags({ SqlError: (cause) => Effect.fail(fail("put")(cause)) }), ), ), + remove: Effect.fn("ThreadTaskStore.remove")((workerThreadId) => + Effect.gen(function* () { + const previous = yield* sql<{ readonly owner: ThreadId }>` + SELECT owner_thread_id AS owner FROM thread_tasks + WHERE worker_thread_id = ${workerThreadId} + `.pipe(Effect.mapError(fail("remove"))); + const owner = previous[0]?.owner; + if (owner === undefined) return; + const cursor = (yield* latestCursor) + 1; + yield* sql` + DELETE FROM thread_tasks WHERE worker_thread_id = ${workerThreadId} + `.pipe(Effect.mapError(fail("remove"))); + yield* sql` + INSERT INTO thread_task_departures (worker_thread_id, owner_thread_id, cursor) + VALUES (${workerThreadId}, ${owner}, ${cursor}) + ON CONFLICT (worker_thread_id, owner_thread_id) DO UPDATE SET cursor = excluded.cursor + `.pipe(Effect.mapError(fail("remove"))); + }).pipe( + sql.withTransaction, + Effect.catchTags({ SqlError: (cause) => Effect.fail(fail("remove")(cause)) }), + ), + ), }); }); diff --git a/packages/client-runtime/src/state/agents.test.ts b/packages/client-runtime/src/state/agents.test.ts index 80924741ddb09..e211757467530 100644 --- a/packages/client-runtime/src/state/agents.test.ts +++ b/packages/client-runtime/src/state/agents.test.ts @@ -444,6 +444,41 @@ describe("subagents and Agent chats on the board", () => { expect(selectAgentWorkspaceThreads(sidebarThreads, null, "delegation").active).toEqual([]); }); + it("nests a chat a subagent launched in the card of the subagent's owner", () => { + // Captain's delegate_task helper (a subagent) calls t3_thread_launch: the + // chat's parent is the helper, which is never a card. + const helper = { ...delegated, parentRelationship: "subagent" as const }; + const nestedHelper = { + ...helper, + id: ThreadId.make("nested-helper"), + parentThreadId: helper.id, + lineage: { ...helper.lineage, parentThreadId: helper.id }, + }; + const launched = { + ...thread("doug-chat", "doug"), + parentThreadId: helper.id, + parentRelationship: "child" as const, + }; + const deeper = { + ...thread("deeper-chat", "doug"), + parentThreadId: nestedHelper.id, + parentRelationship: "child" as const, + }; + const sidebarThreads = selectAgentSidebarThreads([ + captain, + helper, + nestedHelper, + launched, + deeper, + ]); + const board = nestAgentRuns({ + lists: { pinned: [], ...selectAgentWorkspaceThreads(sidebarThreads, null, "") }, + all: sidebarThreads, + }); + expect(board.lists.active.map((run) => run.id)).toEqual([captain.id]); + expect(childIds(board, "local:captain-chat")).toEqual([deeper.id, launched.id].toSorted()); + }); + it("nests a child Agent chat inside its parent's card instead of its own card", () => { const sidebarThreads = selectAgentSidebarThreads(all); const lists = { pinned: [], active: sidebarThreads, settled: [] }; diff --git a/packages/client-runtime/src/state/agents.ts b/packages/client-runtime/src/state/agents.ts index 130911d0da743..a1f8d238e528a 100644 --- a/packages/client-runtime/src/state/agents.ts +++ b/packages/client-runtime/src/state/agents.ts @@ -259,9 +259,44 @@ function isAgentSubagentThread( return isSubagentThread(run); } -/** Every chat that is a card on the board or a row in the sidebar: everything but subagents. */ +/** + * Every chat that is a card on the board or a row in the sidebar: everything but subagents. + * A chat a subagent launched is listed under the nearest ancestor that is not one. + */ export function selectAgentSidebarThreads(threads: readonly EnvironmentThreadShell[]) { - return threads.filter((thread) => !isAgentSubagentThread(thread)); + const subagents = new Map( + threads.filter(isAgentSubagentThread).map((thread) => [threadKey(thread), thread]), + ); + if (subagents.size === 0) return threads; + // A subagent's owner: its recorded parent, or for older rows its lineage parent. + const ownerOf = (subagent: EnvironmentThreadShell) => ({ + environmentId: (subagent.parentEnvironmentId ?? subagent.environmentId) as EnvironmentId, + id: subagent.parentThreadId ?? subagent.lineage.parentThreadId, + }); + return threads.flatMap((thread) => { + if (subagents.has(threadKey(thread))) return []; + let parentKey = agentRunParentKey(thread); + let owner: ReturnType | undefined; + const seen = new Set(); + for ( + let subagent = subagents.get(parentKey ?? ""); + subagent !== undefined && !seen.has(parentKey!); + subagent = subagents.get(parentKey ?? "") + ) { + seen.add(parentKey!); + owner = ownerOf(subagent); + parentKey = owner.id === null ? null : threadKey({ ...owner, id: owner.id }); + } + if (owner === undefined) return [thread]; + return [ + { + ...thread, + parentThreadId: owner.id, + parentEnvironmentId: + owner.environmentId === thread.environmentId ? null : owner.environmentId, + }, + ]; + }); } type AgentRunList = "pinned" | "active" | "settled"; diff --git a/packages/mcp-gateway/src/tools.test.ts b/packages/mcp-gateway/src/tools.test.ts index ba82faf4163fc..780b38398436f 100644 --- a/packages/mcp-gateway/src/tools.test.ts +++ b/packages/mcp-gateway/src/tools.test.ts @@ -536,7 +536,10 @@ describe("gateway chat tools", () => { : [], }), getThread: async (environmentId: string, threadId: string) => { - if (environmentId !== "local" || !["other-parent", "archived-parent"].includes(threadId)) + if ( + environmentId !== "local" || + !["coordinator", "other-parent", "archived-parent"].includes(threadId) + ) throw new Error("Failed to load thread"); const thread = await base.getThread(environmentId, threadId); return threadId === "archived-parent" @@ -570,6 +573,11 @@ describe("gateway chat tools", () => { await create({ parentThreadId: "other-parent" }, "child-explicit"); await create({ parentThreadId: null }, "child-standalone"); await create({ environmentId: "remote", projectId: "remote-project" }, "child-remote"); + // Naming itself as the parent of a chat on another machine means the caller's environment. + await create( + { environmentId: "remote", projectId: "remote-project", parentThreadId: "coordinator" }, + "child-remote-self", + ); await callGatewayTool( context, "t3_create_and_start_thread", @@ -590,6 +598,7 @@ describe("gateway chat tools", () => { [undefined, undefined], // A chat on another machine records which environment its parent is on. ["coordinator", "local"], + ["coordinator", "local"], ["coordinator", undefined], ]); @@ -613,7 +622,7 @@ describe("gateway chat tools", () => { message: `parentThreadId ${input.parentThreadId} is not a chat on environment local; omit parentThreadId to nest under the calling chat.`, }); } - expect(parents).toHaveLength(5); + expect(parents).toHaveLength(6); }); it.each(["constructor", "toString", "__proto__"])( diff --git a/packages/mcp-gateway/src/tools.ts b/packages/mcp-gateway/src/tools.ts index 8c6faa64d4098..b3c6a22e2f317 100644 --- a/packages/mcp-gateway/src/tools.ts +++ b/packages/mcp-gateway/src/tools.ts @@ -2154,11 +2154,13 @@ export async function callGatewayTool( const parentEnvironmentId = parentThreadId === undefined ? undefined - : typeof input.parentThreadId === "string" - ? typeof input.parentEnvironmentId === "string" - ? input.parentEnvironmentId.trim() - : undefined - : invocation.caller?.environmentId; + : typeof input.parentThreadId === "string" && + typeof input.parentEnvironmentId === "string" + ? input.parentEnvironmentId.trim() + : // The calling chat is on the caller's environment, even when named explicitly. + parentThreadId === invocation.caller?.threadId + ? invocation.caller.environmentId + : undefined; // Checked here: the target environment cannot verify a parent on another environment. if (typeof input.parentThreadId === "string" && parentThreadId !== undefined) { const parentEnvironment = parentEnvironmentId ?? environmentId;