perf(server): finalizing a subagent reads only that child's records

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

@@ -29,7 +29,9 @@ import * as Effect from "effect/Effect";
 import * as Deferred from "effect/Deferred";
 import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
 import * as Stream from "effect/Stream";
+import * as Statement from "effect/sql/Statement";
 import * as Tracer from "effect/Tracer";
 import * as TestClock from "effect/testing/TestClock";
 
@@ -3706,6 +3708,108 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))(
   },
 );
 
+// The run-update listener mirrors a follow-up's status onto its task outside any test fiber, so
+// its statements are recorded from the layer down, for one parent thread at a time.
+let recordReadsOf: ThreadId | undefined;
+const recordedReads: Array<string> = [];
+const layerRecordingReads = Layer.succeed(Statement.CurrentTransformer, (statement) => {
+  const [query, params] = statement.compile();
+  if (recordReadsOf !== undefined && params.includes(recordReadsOf)) recordedReads.push(query);
+  return Effect.succeed(statement);
+});
+
+it.layer(
+  Layer.merge(layerTest, ProviderContinuationRequests.layer).pipe(
+    Layer.provide(layerRecordingReads),
+  ),
+)("subagent follow-up status", (it) => {
+  it.effect("mirrors a follow-up's status reading only its task's node and card", () =>
+    Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const eventSink = yield* EventSink.EventSinkV2;
+      const now = yield* DateTime.now;
+      const parentThreadId = ThreadId.make("thread:narrow-status-parent");
+      const projectId = ProjectId.make("project:narrow-status");
+      const parentRunId = RunId.make("run:narrow-status-parent");
+      const rootNodeId = NodeId.make("node:narrow-status-parent");
+      yield* seedParentWithTerminalTask({
+        threadId: parentThreadId,
+        projectId,
+        runId: parentRunId,
+        rootNodeId,
+        taskId: NodeId.make("node:narrow-status-seed"),
+        deliveryState: "acknowledged",
+        now,
+      });
+      const child = yield* seedRestartCancelledChild({
+        parentThreadId,
+        projectId,
+        parentRunId,
+        rootNodeId,
+        name: "narrow-status-child",
+        completionWake: "always",
+        continuationPending: false,
+        runStatus: "completed",
+        now,
+      });
+      yield* orchestrator.recoverDelegatedTasks;
+      for (const name of ["active", "queued"]) {
+        yield* orchestrator.dispatch({
+          type: "message.dispatch",
+          createdBy: "agent",
+          creationSource: "mcp",
+          commandId: CommandId.make(`command:narrow-status:${name}`),
+          threadId: child.childThreadId,
+          senderThreadId: parentThreadId,
+          messageId: MessageId.make(`message:narrow-status:${name}`),
+          text: name,
+          attachments: [],
+          dispatchMode: { type: "queue_after_active" },
+        });
+      }
+      const queued = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find(
+        (run) => run.ordinal === 3,
+      )!;
+      assert.equal(queued.status, "queued");
+      const afterSequence = yield* eventSink.latestSequence();
+      recordReadsOf = parentThreadId;
+      yield* orchestrator.dispatch({
+        type: "thread.stop",
+        threadId: child.childThreadId,
+        commandId: CommandId.make("command:narrow-status:stop"),
+      });
+      yield* eventSink.stream({ afterSequence, eventType: "subagent.updated" }).pipe(
+        Stream.filter(
+          (stored) =>
+            stored.event.type === "subagent.updated" &&
+            stored.event.payload.id === queued.delegatedTaskId &&
+            stored.event.payload.status === "idle",
+        ),
+        Stream.take(1),
+        Stream.runHead,
+      );
+      recordReadsOf = undefined;
+      const paused = yield* orchestrator.getThreadProjection(parentThreadId);
+      assert.equal(paused.nodes.find((row) => row.id === queued.delegatedTaskId)?.status, "idle");
+      assert.equal(
+        paused.turnItems.find(
+          (row) => row.type === "subagent" && row.subagentId === queued.delegatedTaskId,
+        )?.status,
+        "idle",
+      );
+      const nodeReads = recordedReads.filter((query) =>
+        /SELECT payload_json\s+FROM orchestration_v2_projection_(nodes|turn_items)\b/.test(query),
+      );
+      assert.isNotEmpty(nodeReads);
+      assert.deepEqual(
+        nodeReads.filter((query) => !/AND (node\.)?node_id IN/.test(query)),
+        [],
+        "a parent node or card read is not narrowed",
+      );
+    }),
+  );
+});
+
 it.layer(layerTest)("delegated tasks across a server restart", (it) => {
   it.effect.each([
     { name: "releases", key: "", pendingBackgroundTasks: [], released: true },
@@ -3842,6 +3946,234 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => {
       }),
   );
 
+  // Finalizing a child reads, from its parent, only that child's task, node, card and the transfers
+  // it sent, and the message and other tasks of a delivery it joins by id.
+  it.effect("finalizing a delegated child reads only that child's records from its parent", () =>
+    Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const sink = yield* EventSink.EventSinkV2;
+      const now = yield* DateTime.now;
+      const threadId = ThreadId.make("thread:narrow-finalize-parent");
+      const projectId = ProjectId.make("project:narrow-finalize");
+      const runId = RunId.make("run:narrow-finalize-parent");
+      const rootNodeId = NodeId.make("node:narrow-finalize-parent");
+      const siblingTaskId = NodeId.make("node:narrow-finalize-seed");
+      const messageId = MessageId.make(`message:delegated-delivery:${threadId}`);
+      yield* seedParentWithTerminalTask({
+        threadId,
+        projectId,
+        runId,
+        rootNodeId,
+        taskId: siblingTaskId,
+        deliveryState: "claimed",
+        completionWake: "always",
+        deliveryTaskIds: [siblingTaskId],
+        now,
+      });
+      yield* orchestrator.dispatch({
+        type: "message.dispatch",
+        commandId: CommandId.make("command:narrow-finalize-delivery"),
+        threadId,
+        messageId,
+        text: `Delegated task ${siblingTaskId} reached a terminal state.`,
+        attachments: [],
+        dispatchMode: { type: "queue_after_active" },
+        createdBy: "agent",
+        creationSource: "server",
+        delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [siblingTaskId] },
+      });
+      const child = yield* seedRestartCancelledChild({
+        parentThreadId: threadId,
+        projectId,
+        parentRunId: runId,
+        rootNodeId,
+        name: "narrow-finalize-child",
+        completionWake: "always",
+        continuationPending: false,
+        runStatus: "completed",
+        now,
+      });
+      const cards = [
+        { taskId: siblingTaskId, status: "completed" as const, ordinal: 100 },
+        { taskId: child.taskId, status: "running" as const, ordinal: 101 },
+      ];
+      yield* sink.write({
+        commandId: CommandId.make("command:narrow-finalize-cards"),
+        events: cards.flatMap(({ taskId, status, ordinal }) => [
+          {
+            id: EventId.make(`event:narrow-finalize-node:${taskId}`),
+            type: "node.updated" as const,
+            threadId,
+            runId,
+            nodeId: taskId,
+            occurredAt: now,
+            payload: {
+              id: taskId,
+              threadId,
+              runId,
+              parentNodeId: rootNodeId,
+              rootNodeId,
+              kind: "subagent" as const,
+              status,
+              countsForRun: false,
+              providerThreadId: null,
+              providerTurnId: null,
+              nativeItemRef: null,
+              runtimeRequestId: null,
+              checkpointScopeId: null,
+              startedAt: now,
+              completedAt: status === "completed" ? now : null,
+            },
+          },
+          {
+            id: EventId.make(`event:narrow-finalize-card:${taskId}`),
+            type: "turn-item.updated" as const,
+            threadId,
+            runId,
+            nodeId: taskId,
+            occurredAt: now,
+            payload: {
+              id: TurnItemId.make(`turn-item:narrow-finalize:${taskId}`),
+              type: "subagent" as const,
+              threadId,
+              runId,
+              nodeId: taskId,
+              providerThreadId: null,
+              providerTurnId: null,
+              nativeItemRef: null,
+              parentItemId: null,
+              ordinal,
+              status,
+              title: null,
+              startedAt: now,
+              completedAt: status === "completed" ? now : null,
+              updatedAt: now,
+              subagentId: taskId,
+              origin: "app_owned" as const,
+              driver,
+              providerInstanceId: modelSelection.instanceId,
+              childThreadId: taskId === child.taskId ? child.childThreadId : null,
+              prompt: `Run ${taskId}.`,
+              result: status === "completed" ? "child finished" : null,
+            },
+          },
+        ]),
+      });
+      const parentReads: Array<string> = [];
+      const transferReadParams: Array<ReadonlyArray<unknown>> = [];
+      // The parent runs read while the child's result is recorded, with their bound ids.
+      const resultRunReads: Array<readonly [string, ReadonlyArray<unknown>]> = [];
+      const recordsResult = (span: Tracer.AnySpan | undefined): boolean =>
+        span?._tag === "Span" &&
+        (span.name === "Orchestrator.finalizeAppOwnedSubagentTask" ||
+          recordsResult(Option.getOrUndefined(span.parent)));
+      const observe: Statement.Transformer = (statement, _sql, _fiber, span) => {
+        const [query, params] = statement.compile();
+        if (
+          params.includes(threadId) &&
+          query.includes("FROM orchestration_v2_projection_runs") &&
+          recordsResult(span)
+        ) {
+          resultRunReads.push([query, params]);
+        }
+        if (
+          params.includes(threadId) &&
+          (/FROM orchestration_v2_projection_(subagents|context_transfers)\b/.test(query) ||
+            /SELECT payload_json\s+FROM orchestration_v2_projection_(nodes|turn_items|messages)\b/.test(
+              query,
+            ))
+        ) {
+          parentReads.push(query);
+          if (query.includes("FROM orchestration_v2_projection_context_transfers")) {
+            transferReadParams.push(params);
+          }
+        }
+        return Effect.succeed(statement);
+      };
+      yield* orchestrator.recoverDelegatedTasks.pipe(
+        Effect.provideService(Statement.CurrentTransformer, observe),
+      );
+      const projection = yield* orchestrator.getThreadProjection(threadId);
+      const task = projection.subagents.find((candidate) => candidate.id === child.taskId);
+      assert.equal(task?.status, "completed");
+      // Every finalize read names its rows: tasks by child thread or id, transfers by source.
+      const subagentReads = parentReads.filter((query) =>
+        query.includes("FROM orchestration_v2_projection_subagents"),
+      );
+      const transferReads = parentReads.filter((query) =>
+        query.includes("FROM orchestration_v2_projection_context_transfers"),
+      );
+      assert.isNotEmpty(subagentReads);
+      assert.isNotEmpty(transferReads);
+      assert.deepEqual(
+        subagentReads.filter((query) => !/AND (child_thread_id|subagent_id) (=|IN)/.test(query)),
+        [],
+        "a parent subagent read is not narrowed",
+      );
+      assert.deepEqual(
+        transferReads.filter((query) => !/AND source_thread_id (=|IN)/.test(query)),
+        [],
+        "a parent context-transfer read is not narrowed",
+      );
+      // Only the task's run and the run its delivery message started, each read by id.
+      const deliveryRunId = projection.runs.find((run) => run.userMessageId === messageId)?.id;
+      assert.isDefined(deliveryRunId);
+      assert.isNotEmpty(resultRunReads);
+      assert.deepEqual(
+        resultRunReads.filter(([query]) => !/AND run_id IN/.test(query)),
+        [],
+        "a parent runs read is not narrowed",
+      );
+      assert.deepEqual(
+        [
+          ...new Set(
+            resultRunReads.flatMap(([, params]) =>
+              params
+                .filter((param) => param !== threadId)
+                .flatMap((param) => JSON.parse(String(param)) as Array<string>),
+            ),
+          ),
+        ].toSorted(),
+        [runId, deliveryRunId].toSorted(),
+      );
+      // An existing result transfer is found only when the read asks for this child's.
+      assert.deepEqual(
+        transferReadParams.filter((params) => !params.includes(child.childThreadId)),
+        [],
+        "a parent context-transfer read names another thread",
+      );
+      // Nodes and cards by the task's node, messages by id.
+      for (const [table, narrowed] of [
+        ["nodes", /AND node\.node_id IN/],
+        ["turn_items", /AND node_id IN/],
+        ["messages", /AND message_id IN/],
+      ] as const) {
+        const reads = parentReads.filter((query) =>
+          query.includes(`FROM orchestration_v2_projection_${table}`),
+        );
+        assert.isNotEmpty(reads, table);
+        assert.deepEqual(
+          reads.filter((query) => !narrowed.test(query)),
+          [],
+          `a parent ${table} read is not narrowed`,
+        );
+      }
+      // The child's node and card still end with its task.
+      assert.equal(projection.nodes.find((row) => row.id === child.taskId)?.status, "completed");
+      assert.equal(
+        projection.turnItems.find(
+          (row) => row.type === "subagent" && row.subagentId === child.taskId,
+        )?.status,
+        "completed",
+      );
+      // The delivery the child joins still names the sibling it was opened for.
+      const message = projection.messages.find((candidate) => candidate.id === messageId);
+      assert.deepEqual(message?.delegatedCompletion?.taskIds, [siblingTaskId, child.taskId]);
+      assert.include(message?.text ?? "", `Delegated task ${siblingTaskId} [completed]`);
+      assert.include(message?.text ?? "", `Delegated task ${child.taskId}`);
+    }),
+  );
+
   it.effect("holds a restart-cancelled child for its continuation's result", () =>
     Effect.gen(function* () {
       const orchestrator = yield* Orchestrator.OrchestratorV2;