fix(server,web,mobile): agent-launched chats keep their launcher as parent

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,

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);
 

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,

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<ThreadLaunch.ThreadLaunchInput["delegatedFrom"] | null>;
+  /** 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<LaunchedWith | null>;
   /** The first message the thread there was launched with. */
@@ -137,6 +142,7 @@ const makeBoxThread = Effect.gen(function* () {
       null,
     ),
     delegatedFrom: yield* Ref.make<ThreadLaunch.ThreadLaunchInput["delegatedFrom"] | null>(null),
+    parent: yield* Ref.make<BoxThread["parent"] extends Ref.Ref<infer A> ? A : never>(null),
     launchedWith: yield* Ref.make<LaunchedWith | null>(null),
     message: yield* Ref.make<string | null>(null),
     status: yield* Ref.make<OrchestrationV2Run["status"]>("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);

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* () {

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))),

apps/server/src/threadTask/ThreadTaskStore.ts

@@ -121,6 +121,8 @@ export class ThreadTaskStore extends Context.Service<
     ) => Effect.Effect<void, ThreadTaskStoreError>;
     /** Writes the task under the next cursor and returns it as stored. */
     readonly put: (task: ThreadTask) => Effect.Effect<ThreadTask, ThreadTaskStoreError>;
+    /** Drops the worker's task; its owner's delta reads it as a departure. */
+    readonly remove: (workerThreadId: ThreadId) => Effect.Effect<void, ThreadTaskStoreError>;
   }
 >()("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)) }),
+      ),
+    ),
   });
 });
 

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: [] };