perf(server): new prompts in a compacted thread read only what follows it

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

@@ -3167,6 +3167,59 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => {
     }),
   );
 
+  it.effect("settles a child without reading all of the parent's nodes", () =>
+    Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const now = yield* DateTime.now;
+      const threadId = ThreadId.make("thread:bounded-finalize-parent");
+      const projectId = ProjectId.make("project:bounded-finalize");
+      const runId = RunId.make("run:bounded-finalize-parent");
+      const rootNodeId = NodeId.make("node:bounded-finalize-parent");
+      yield* seedParentWithTerminalTask({
+        threadId,
+        projectId,
+        runId,
+        rootNodeId,
+        taskId: NodeId.make("node:bounded-finalize-seed"),
+        deliveryState: "acknowledged",
+        now,
+      });
+      const child = yield* seedRestartCancelledChild({
+        parentThreadId: threadId,
+        projectId,
+        parentRunId: runId,
+        rootNodeId,
+        name: "bounded-finalize-child",
+        completionWake: "always",
+        continuationPending: false,
+        runStatus: "completed",
+        now,
+      });
+      const statements: string[] = [];
+      const tracer = Tracer.make({
+        span(options) {
+          const span = new Tracer.NativeSpan(options);
+          const end = span.end.bind(span);
+          span.end = (time, exit) => {
+            end(time, exit);
+            const query = span.attributes.get("db.query.text");
+            if (typeof query === "string") statements.push(query);
+          };
+          return span;
+        },
+      });
+      yield* orchestrator.recoverDelegatedTasks.pipe(Effect.withTracer(tracer));
+      const settled = yield* orchestrator.getThreadProjection(threadId);
+      assert.equal(settled.subagents.find((task) => task.id === child.taskId)?.status, "completed");
+      // A dispatcher parent holds thousands of nodes; settling one child names its own.
+      const nodeReads = statements.filter((query) =>
+        query.includes("FROM orchestration_v2_projection_nodes"),
+      );
+      assert.isNotEmpty(nodeReads);
+      assert.isTrue(nodeReads.every((query) => query.includes("node_id IN")));
+    }),
+  );
+
   it.effect("holds a restart-cancelled child for its continuation's result", () =>
     Effect.gen(function* () {
       const orchestrator = yield* Orchestrator.OrchestratorV2;

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

@@ -1298,7 +1298,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   });
 
   // Runs in every command transaction; avoid reading thousands of finished leaf nodes.
-  const readCommandProjection = (threadId: ThreadId) =>
+  const readCommandProjection = (
+    threadId: ThreadId,
+    recent?: { keepRunIds: ReadonlyArray<RunId> },
+  ) =>
     projectionStore
       .getThreadRecords(
         threadId,
@@ -1323,6 +1326,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
           // Commands here read only undelivered or active tasks; a dispatcher thread keeps
           // thousands of delivered ones, each carrying its full prompt.
           omitSettledSubagents: true,
+          ...(recent === undefined ? {} : { sinceLastCompaction: recent }),
         },
       )
       .pipe(
@@ -1411,9 +1415,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     Effect.fnUntraced(function* (
       threadId: ThreadId,
       events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
+      recent?: { keepRunIds: ReadonlyArray<RunId> },
     ) {
       const pending = (yield* Ref.get(events)).filter((event) => event.threadId === threadId);
-      const stored = yield* Effect.option(readCommandProjection(threadId));
+      const stored = yield* Effect.option(readCommandProjection(threadId, recent));
       let projection: OrchestrationV2ThreadProjection;
       if (Option.isSome(stored)) {
         projection = stored.value;
@@ -5845,7 +5850,31 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
     effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
   ) {
-    let projection = yield* getProjectionWithPendingEvents(command.threadId, events);
+    // A compacted thread needs only its recent runs, plus any older run the command names.
+    let projection = yield* getProjectionWithPendingEvents(command.threadId, events, {
+      keepRunIds: [
+        command.manualContinuationOfRunId,
+        command.restartContinuationOfRunId,
+        command.usageLimitContinuationOfRunId,
+        command.delegatedCompletion?.parentRunId,
+        ...(command.dispatchMode.type === "steer_active" ||
+        command.dispatchMode.type === "restart_active"
+          ? [command.dispatchMode.targetRunId]
+          : []),
+      ].flatMap((runId) => (runId === undefined ? [] : [runId])),
+    });
+    const activeProviderInstanceId = projection.providerThreads.find(
+      (candidate) => candidate.id === projection.thread.activeProviderThreadId,
+    )?.providerInstanceId;
+    // A fresh session or another provider is handed the whole history, so it reads all of it.
+    if (
+      command.freshContext === true ||
+      (activeProviderInstanceId !== undefined &&
+        activeProviderInstanceId !==
+          (command.modelSelection ?? projection.thread.modelSelection).instanceId)
+    ) {
+      projection = yield* getProjectionWithPendingEvents(command.threadId, events);
+    }
     if (command.manualContinuationOfRunId !== undefined) {
       const source = projection.runs.find((run) => run.id === command.manualContinuationOfRunId);
       const limited = latestRootProviderFailure(source ?? null, projection.turnItems);
@@ -6018,16 +6047,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       });
       projection = applyToProjection(projection, event);
     }
-    const userMessages = projection.messages.filter((message) => message.role === "user");
-    const onlyMaintenanceHistory =
-      userMessages.length > 0 && userMessages.every(isNativeMaintenanceCommand);
     if (
       !isNativeMaintenanceCommand(command) &&
       ((command.titleSeed !== undefined &&
         (yield* projectionStore
           .getMessageCount(command.threadId)
           .pipe(mapDispatchError(command))) === 0) ||
-        (command.createdBy === "user" && onlyMaintenanceHistory))
+        (command.createdBy === "user" &&
+          (yield* projectionStore
+            .hasOnlyMaintenanceUserMessages(command.threadId)
+            .pipe(mapDispatchError(command)))))
     ) {
       const now = yield* DateTime.now;
       const thread: OrchestrationV2AppThread = {
@@ -13008,21 +13037,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         turnItems: childResult.turnItems,
       };
       const parentThreadId = childControls.thread.lineage.parentThreadId;
-      const parentProjection = yield* projectionStore.getThreadRecords(
+      // A dispatcher parent holds thousands of tasks, runs and prompts, so read only this task's
+      // records and the delivery cohort the completion plan joins.
+      const taskRecords = yield* projectionStore.getThreadRecords(
         parentThreadId,
-        [
-          "runs",
-          "messages",
-          "subagents",
-          "providerThreads",
-          "providerTurns",
-          "nodes",
-          "turnItems",
-          "contextTransfers",
-        ],
-        { turnItemTypes: ["subagent"], messageRoles: ["user"] },
+        ["subagents", "nodes", "turnItems"],
+        {
+          subagentIds: [options.task.id],
+          nodeIds: [options.task.id],
+          turnItemNodeIds: [options.task.id],
+          turnItemTypes: ["subagent"],
+        },
       );
-      const task = parentProjection.subagents.find(
+      const task = taskRecords.subagents.find(
         (candidate) =>
           candidate.id === options.task.id &&
           candidate.origin === "app_owned" &&
@@ -13031,7 +13058,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       if (task === undefined) {
         return;
       }
-      const existingResultTransfer = parentProjection.contextTransfers.find(
+      // The result transfer runs from the child, so the child's side lists it too.
+      const { contextTransfers } = yield* projectionStore.getThreadRecords(childThreadId, [
+        "contextTransfers",
+      ]);
+      const existingResultTransfer = contextTransfers.find(
         (transfer) =>
           transfer.type === "subagent_result" &&
           transfer.sourceThreadId === childThreadId &&
@@ -13045,12 +13076,37 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
 
       const now = yield* DateTime.now;
       const result = subagentResultForRun(childProjection, { ...childRun, status: outcome });
-      const parentRun =
-        task.runId === null
+      const taskRunRecords = yield* projectionStore.getThreadRecords(parentThreadId, ["runs"], {
+        runIds: task.runId === null ? [] : [task.runId],
+      });
+      const parentRun = taskRunRecords.runs.find((candidate) => candidate.id === task.runId);
+      const delivery = parentRun?.delegatedCompletion?.delivery;
+      const deliveryRecords =
+        delivery == null
+          ? undefined
+          : yield* projectionStore.getThreadRecords(parentThreadId, ["messages", "subagents"], {
+              messageIds: [delivery.messageId],
+              subagentIds: [...new Set([...delivery.taskIds, task.id])],
+            });
+      const deliveryRunId = deliveryRecords?.messages[0]?.runId ?? null;
+      const deliveryRun =
+        deliveryRunId === null || deliveryRunId === parentRun?.id
           ? undefined
-          : parentProjection.runs.find((candidate) => candidate.id === task.runId);
-      const parentNode = parentProjection.nodes.find((candidate) => candidate.id === task.id);
-      const parentTurnItem = parentProjection.turnItems.find(
+          : (yield* projectionStore.getThreadRecords(parentThreadId, ["runs"], {
+              runIds: [deliveryRunId],
+            })).runs[0];
+      const parentProjection = {
+        thread: taskRecords.thread,
+        runs: [
+          ...(parentRun === undefined ? [] : [parentRun]),
+          ...(deliveryRun ? [deliveryRun] : []),
+        ],
+        messages: deliveryRecords?.messages ?? [],
+        subagents: deliveryRecords?.subagents ?? [task],
+        providerTurns: [],
+      };
+      const parentNode = taskRecords.nodes.find((candidate) => candidate.id === task.id);
+      const parentTurnItem = taskRecords.turnItems.find(
         (candidate) => candidate.type === "subagent" && candidate.subagentId === task.id,
       );
       const updatedTask: OrchestrationV2Subagent = {
@@ -13183,8 +13239,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     )
       return;
     const progress = delegatedTaskProgress(childControls);
-    const parent = yield* projectionStore.getThreadRecords(parentThreadId, [
-      "subagents",
+    // A dispatcher parent holds thousands of tasks and transfers; this child owns a few of each.
+    const parent = yield* projectionStore.getThreadRecords(parentThreadId, ["subagents"], {
+      subagentChildThreadId: childThreadId,
+    });
+    const { contextTransfers } = yield* projectionStore.getThreadRecords(childThreadId, [
       "contextTransfers",
     ]);
     for (const task of parent.subagents) {
@@ -13194,9 +13253,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         (task.id === childControls.thread.forkedFrom.nodeId &&
           progress.state !== "result_available") ||
         (task.result !== null &&
-          parent.contextTransfers.some(
+          contextTransfers.some(
             (transfer) =>
-              transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId,
+              transfer.type === "subagent_result" &&
+              transfer.sourceThreadId === childThreadId &&
+              transfer.targetThreadId === parentThreadId,
           ))
       )
         continue;