fix(orchestration-v2): show nested agents in parent fleet

Author: T3 Code Test

apps/server/src/orchestration-v2/SubagentTreeProjection.test.ts

@@ -0,0 +1,85 @@
+import { describe, expect, it } from "vite-plus/test";
+import {
+  EventId,
+  NodeId,
+  ProviderDriverKind,
+  ProviderInstanceId,
+  ThreadId,
+  type ApplicationStoredEvent,
+  type OrchestrationV2Subagent,
+} from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
+
+import { routeSubagentTreeEvent } from "./SubagentTreeProjection.ts";
+
+const now = DateTime.makeUnsafe("2026-08-18T00:00:00.000Z");
+
+function subagent(input: {
+  readonly id: string;
+  readonly threadId: ThreadId;
+  readonly childThreadId: ThreadId;
+}): OrchestrationV2Subagent {
+  return {
+    id: NodeId.make(input.id),
+    threadId: input.threadId,
+    runId: null,
+    parentNodeId: NodeId.make("parent"),
+    origin: "provider_native",
+    createdBy: "agent",
+    driver: ProviderDriverKind.make("codex"),
+    providerInstanceId: ProviderInstanceId.make("codex"),
+    providerThreadId: null,
+    childThreadId: input.childThreadId,
+    nativeTaskRef: null,
+    prompt: "",
+    title: input.id,
+    model: "gpt-5.6-sol",
+    kind: "subagent",
+    role: { name: "general-purpose", source: "app_default" },
+    status: "running",
+    result: null,
+    usage: null,
+    currentActivationId: null,
+    activationCount: 1,
+    workflow: null,
+    workflowMembership: null,
+    recentActivity: [],
+    startedAt: now,
+    completedAt: null,
+    updatedAt: now,
+  };
+}
+
+describe("subagent tree projection", () => {
+  it("routes nested lifecycle rows to the root fleet and discovers deeper children", () => {
+    const rootThreadId = ThreadId.make("root-thread");
+    const childThreadId = ThreadId.make("child-thread");
+    const grandchildThreadId = ThreadId.make("grandchild-thread");
+    const nested = subagent({
+      id: "nested-agent",
+      threadId: childThreadId,
+      childThreadId: grandchildThreadId,
+    });
+
+    const stored = {
+      sequence: 2,
+      commandId: null,
+      event: {
+        id: EventId.make("nested-update"),
+        type: "subagent.updated",
+        threadId: childThreadId,
+        nodeId: nested.id,
+        occurredAt: now,
+        payload: nested,
+      },
+    } satisfies ApplicationStoredEvent;
+    const [next, routed] = routeSubagentTreeEvent(
+      { rootThreadId, threadIds: new Set([rootThreadId, childThreadId]) },
+      stored,
+    );
+
+    expect(routed[0]?.event.threadId).toBe(rootThreadId);
+    expect(routed[0]?.event.type).toBe("subagent.updated");
+    expect(next.threadIds.has(grandchildThreadId)).toBe(true);
+  });
+});

apps/server/src/orchestration-v2/SubagentTreeProjection.ts

@@ -0,0 +1,93 @@
+import type {
+  ApplicationStoredEvent,
+  OrchestrationV2StoredEvent,
+  OrchestrationV2ThreadProjection,
+  ThreadId,
+} from "@t3tools/contracts";
+
+function upsertById<T extends { readonly id: string }>(
+  target: Map<string, T>,
+  values: ReadonlyArray<T>,
+) {
+  for (const value of values) target.set(value.id, value);
+}
+
+/** Add descendant agent rows to the root projection without copying their transcripts. */
+export function mergeSubagentTreeProjection(
+  root: OrchestrationV2ThreadProjection,
+  descendants: ReadonlyArray<OrchestrationV2ThreadProjection>,
+): OrchestrationV2ThreadProjection {
+  if (descendants.length === 0) return root;
+
+  const subagents = new Map(root.subagents.map((subagent) => [subagent.id, subagent]));
+  const activations = new Map(
+    root.subagentActivations.map((activation) => [activation.id, activation]),
+  );
+  for (const projection of descendants) {
+    upsertById(subagents, projection.subagents);
+    upsertById(activations, projection.subagentActivations);
+  }
+  return {
+    ...root,
+    subagents: Array.from(subagents.values()),
+    subagentActivations: Array.from(activations.values()),
+  };
+}
+
+export interface SubagentTreeStreamState {
+  readonly rootThreadId: ThreadId;
+  readonly threadIds: ReadonlySet<ThreadId>;
+}
+
+function withChildThread(
+  state: SubagentTreeStreamState,
+  childThreadId: ThreadId | null,
+): SubagentTreeStreamState {
+  if (childThreadId === null || state.threadIds.has(childThreadId)) return state;
+  return { ...state, threadIds: new Set([...state.threadIds, childThreadId]) };
+}
+
+/**
+ * Keep the root event stream complete while folding descendant subagent lifecycle
+ * events into the same client-side fleet projection.
+ */
+export function routeSubagentTreeEvent(
+  state: SubagentTreeStreamState,
+  stored: ApplicationStoredEvent,
+): readonly [SubagentTreeStreamState, ReadonlyArray<OrchestrationV2StoredEvent>] {
+  if (!("event" in stored)) return [state, []];
+
+  const event = stored.event;
+  if (
+    event.type === "thread.created" &&
+    event.payload.lineage.relationshipToParent === "subagent" &&
+    event.payload.lineage.parentThreadId !== null &&
+    state.threadIds.has(event.payload.lineage.parentThreadId)
+  ) {
+    return [withChildThread(state, event.payload.id), []];
+  }
+  if (event.threadId === state.rootThreadId) {
+    return [
+      event.type === "subagent.updated"
+        ? withChildThread(state, event.payload.childThreadId)
+        : state,
+      [stored],
+    ];
+  }
+  if (!state.threadIds.has(event.threadId)) return [state, []];
+  if (event.type !== "subagent.updated" && event.type !== "subagent-activation.updated") {
+    return [state, []];
+  }
+
+  const next =
+    event.type === "subagent.updated" ? withChildThread(state, event.payload.childThreadId) : state;
+  return [
+    next,
+    [
+      {
+        ...stored,
+        event: { ...event, threadId: state.rootThreadId },
+      },
+    ],
+  ];
+}

apps/server/src/ws.ts

@@ -32,6 +32,7 @@ import {
   OrchestrationV2ThreadLaunchError,
   type OrchestrationProjectShell,
   type OrchestrationV2ShellSnapshot,
+  type OrchestrationV2ThreadProjection,
   type ProjectEntriesFailure,
   type ProjectFileFailure,
   type ProjectFileOperation,
@@ -95,6 +96,10 @@ import {
   projectDomainEventForWire,
   projectThreadProjectionForWire,
 } from "./orchestration-v2/WireProjection.ts";
+import {
+  mergeSubagentTreeProjection,
+  routeSubagentTreeEvent,
+} from "./orchestration-v2/SubagentTreeProjection.ts";
 import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
 import * as OrchestrationEventStore from "./persistence/Services/OrchestrationEventStore.ts";
 import { userFacingDispatchErrorMessage } from "./orchestration-v2/UserFacingErrors.ts";
@@ -624,27 +629,61 @@ const makeWsRpcLayer = (
             ),
           );
 
-          const eventStreamFrom = (afterSequence: number) =>
-            threadManagement
-              .streamStoredEventsFrom({
-                threadId: input.threadId,
-                afterSequence,
-              })
-              .pipe(
-                Stream.map((stored) => ({
-                  kind: "event" as const,
-                  sequence: stored.sequence,
-                  event: projectDomainEventForWire(stored.event),
-                })),
-                Stream.mapError(
-                  (cause) =>
-                    new OrchestrationV2GetThreadProjectionError({
-                      threadId: input.threadId,
-                      message: `Failed while streaming orchestration V2 thread ${input.threadId}`,
-                      cause,
-                    }),
+          const loadSubagentTree = Effect.fn("ws.orchestrationV2.loadSubagentTree")(function* (
+            root: OrchestrationV2ThreadProjection,
+          ) {
+            const threadIds = new Set<ThreadId>([root.thread.id]);
+            const descendants: OrchestrationV2ThreadProjection[] = [];
+            let frontier = root.subagents.flatMap((subagent) =>
+              subagent.childThreadId === null ? [] : [subagent.childThreadId],
+            );
+
+            while (frontier.length > 0) {
+              const nextIds = [...new Set(frontier)].filter((threadId) => !threadIds.has(threadId));
+              if (nextIds.length === 0) break;
+              for (const threadId of nextIds) threadIds.add(threadId);
+              const projections = yield* Effect.forEach(
+                nextIds,
+                (threadId) => threadManagement.getThreadProjection(threadId),
+                { concurrency: 8 },
+              );
+              descendants.push(...projections);
+              frontier = projections.flatMap((projection) =>
+                projection.subagents.flatMap((subagent) =>
+                  subagent.childThreadId === null ? [] : [subagent.childThreadId],
                 ),
               );
+            }
+
+            return {
+              threadIds,
+              projection: mergeSubagentTreeProjection(root, descendants),
+            };
+          });
+
+          const eventStreamFrom = (
+            afterSequence: number,
+            initialThreadIds: ReadonlySet<ThreadId>,
+          ) =>
+            applicationEvents.streamApplicationEvents({ afterSequence }).pipe(
+              Stream.mapAccum(
+                () => ({ rootThreadId: input.threadId, threadIds: initialThreadIds }),
+                routeSubagentTreeEvent,
+              ),
+              Stream.map((stored) => ({
+                kind: "event" as const,
+                sequence: stored.sequence,
+                event: projectDomainEventForWire(stored.event),
+              })),
+              Stream.mapError(
+                (cause) =>
+                  new OrchestrationV2GetThreadProjectionError({
+                    threadId: input.threadId,
+                    message: `Failed while streaming orchestration V2 thread ${input.threadId}`,
+                    cause,
+                  }),
+              ),
+            );
 
           const loadReplayThrough = (afterSequence: number, throughSequence: number) =>
             applicationEvents
@@ -679,6 +718,16 @@ const makeWsRpcLayer = (
 
           const snapshotThenLive = Effect.fn("ws.orchestrationV2.threadSnapshotThenLive")(
             function* () {
+              const snapshotSequence = yield* applicationEvents.latestApplicationSequence.pipe(
+                Effect.mapError(
+                  (cause) =>
+                    new OrchestrationV2GetThreadProjectionError({
+                      threadId: input.threadId,
+                      message: `Failed to prepare orchestration V2 thread ${input.threadId} snapshot`,
+                      cause,
+                    }),
+                ),
+              );
               const snapshot = yield* threadManagement.getThreadSnapshot(input.threadId).pipe(
                 Effect.mapError(
                   (cause) =>
@@ -689,8 +738,17 @@ const makeWsRpcLayer = (
                     }),
                 ),
               );
-              const { snapshotSequence } = snapshot;
-              const projection = projectThreadProjectionForWire(snapshot.projection);
+              const tree = yield* loadSubagentTree(snapshot.projection).pipe(
+                Effect.mapError(
+                  (cause) =>
+                    new OrchestrationV2GetThreadProjectionError({
+                      threadId: input.threadId,
+                      message: `Failed to load nested agents for orchestration V2 thread ${input.threadId}`,
+                      cause,
+                    }),
+                ),
+              );
+              const projection = projectThreadProjectionForWire(tree.projection);
               return Stream.concat(
                 Stream.concat(
                   Stream.make({
@@ -700,7 +758,7 @@ const makeWsRpcLayer = (
                   }),
                   completionMarker,
                 ),
-                eventStreamFrom(snapshotSequence),
+                eventStreamFrom(snapshotSequence, tree.threadIds),
               );
             },
           );
@@ -713,7 +771,7 @@ const makeWsRpcLayer = (
           // published during the replay window is lost; overlapping events are
           // deduped by sequence on the client.
           if (input.afterSequence !== undefined) {
-            const highWater = yield* applicationEvents.latestAgentSequence(input.threadId).pipe(
+            const highWater = yield* applicationEvents.latestApplicationSequence.pipe(
               Effect.mapError(
                 (cause) =>
                   new OrchestrationV2GetThreadProjectionError({
@@ -723,6 +781,41 @@ const makeWsRpcLayer = (
                   }),
               ),
             );
+            const currentProjection = yield* threadManagement
+              .getThreadProjection(input.threadId)
+              .pipe(
+                Effect.mapError(
+                  (cause) =>
+                    new OrchestrationV2GetThreadProjectionError({
+                      threadId: input.threadId,
+                      message: `Failed to inspect nested agents for orchestration V2 thread ${input.threadId}`,
+                      cause,
+                    }),
+                ),
+              );
+            const currentTree = yield* loadSubagentTree(currentProjection).pipe(
+              Effect.mapError(
+                (cause) =>
+                  new OrchestrationV2GetThreadProjectionError({
+                    threadId: input.threadId,
+                    message: `Failed to inspect nested agents for orchestration V2 thread ${input.threadId}`,
+                    cause,
+                  }),
+              ),
+            );
+            if (currentTree.threadIds.size > 1) {
+              return Stream.concat(
+                Stream.concat(
+                  Stream.make({
+                    kind: "snapshot" as const,
+                    snapshotSequence: highWater,
+                    projection: projectThreadProjectionForWire(currentTree.projection),
+                  }),
+                  completionMarker,
+                ),
+                eventStreamFrom(highWater, currentTree.threadIds),
+              );
+            }
             const replay = yield* loadReplayThrough(input.afterSequence, highWater);
             const plan = decideThreadResume({
               afterSequence: input.afterSequence,
@@ -735,7 +828,7 @@ const makeWsRpcLayer = (
             }
             return Stream.concat(
               Stream.concat(Stream.fromIterable(replay), completionMarker),
-              eventStreamFrom(highWater),
+              eventStreamFrom(highWater, currentTree.threadIds),
             );
           }