fix(chat): requeued steers no longer look delivered

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

@@ -1973,13 +1973,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     const projection = yield* projectionStore.getThreadRecords(
       threadId,
       ["runs", "nodes", "attempts", "providerThreads", "turnItems", "messages"],
-      { turnItemTypes: [], messageRoles: ["user"] },
+      { turnItemTypes: ["user_message"], messageRoles: ["user"] },
     );
     const queuedRun =
       queuedRunId === undefined
         ? nextQueuedRun(projection)
         : projection.runs.find((run) => run.id === queuedRunId && run.status === "queued");
     if (queuedRun === undefined) return;
+    // A steer that fell back to the queue keeps its user item, hidden while the
+    // run is queued. Republish it before the failure so clients whose bounded
+    // snapshot omitted it can show the prompt beside the error.
+    const retainedUserItems = projection.turnItems.filter(
+      (item) => item.type === "user_message" && item.runId === queuedRun.id,
+    );
     const now = yield* DateTime.now;
     const rootNode = projection.nodes.find((node) => node.id === queuedRun.rootNodeId);
     const attempt = projection.attempts.find((entry) => entry.id === queuedRun.activeAttemptId);
@@ -2005,6 +2011,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       class: handoffUnsupported ? "validation_error" : "unknown",
     });
     yield* writeSystemEvents([
+      ...retainedUserItems.map((item) => ({
+        type: "turn-item.updated" as const,
+        threadId,
+        runId: queuedRun.id,
+        ...(rootNode === undefined ? {} : { nodeId: rootNode.id }),
+        providerInstanceId: queuedRun.providerInstanceId,
+        occurredAt: now,
+        payload: { ...item, status: "failed" as const, completedAt: now, updatedAt: now },
+      })),
       ...(attempt !== undefined && rootNode !== undefined
         ? [
             {

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

@@ -5018,26 +5018,36 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           })}
         )
       `;
-      const cancelledSuffixSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
-        rowLimit: sqlPageLimit,
-      });
-      const cancelledSuffixPage = buildBoundedThreadProjection({
-        policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
-        projection: cancelledSuffixSnapshot.projection,
-        snapshotSequence: cancelledSuffixSnapshot.snapshotSequence,
-      });
-      assert.strictEqual(cancelledSuffixPage.latestLocalTurnOrdinal, 1_202);
-      assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 75);
-      assert.isFalse(
-        cancelledSuffixPage.projection.visibleTurnItems.some((row) =>
-          String(row.sourceItemId).includes(":cancelled:"),
-        ),
-      );
-      assert.isTrue(
-        cancelledSuffixPage.projection.visibleTurnItems.some(
-          (row) => row.sourceItemId === runlessQueuedId,
-        ),
-      );
+      // A steer that fell back to the queue keeps its queued-turn item while its
+      // run waits. End on cancelled for the checks below.
+      for (const status of ["queued", "cancelled"] as const) {
+        yield* sql`
+          UPDATE orchestration_v2_projection_runs
+          SET status = ${status}, payload_json = json_set(payload_json, '$.status', ${status})
+          WHERE run_id = ${cancelledRunId}
+        `;
+        const cancelledSuffixSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
+          rowLimit: sqlPageLimit,
+        });
+        const cancelledSuffixPage = buildBoundedThreadProjection({
+          policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
+          projection: cancelledSuffixSnapshot.projection,
+          snapshotSequence: cancelledSuffixSnapshot.snapshotSequence,
+        });
+        assert.strictEqual(cancelledSuffixPage.latestLocalTurnOrdinal, 1_202);
+        assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 75);
+        assert.isFalse(
+          cancelledSuffixPage.projection.visibleTurnItems.some((row) =>
+            String(row.sourceItemId).includes(":cancelled:"),
+          ),
+          status,
+        );
+        assert.isTrue(
+          cancelledSuffixPage.projection.visibleTurnItems.some(
+            (row) => row.sourceItemId === runlessQueuedId,
+          ),
+        );
+      }
 
       // A steer cancels the queued run before it starts while the steered run keeps writing, so
       // the queued message leaves the window but its run is still the shell's latest run.
@@ -5830,6 +5840,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
             : [
                 "rolled back",
                 "cancelled queued turn",
+                "held queued turn",
                 "replaced attempt",
                 "stopped",
                 "stopped",
@@ -5860,9 +5871,11 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
                 ? "rolled_back"
                 : kind === "cancelled queued turn"
                   ? "cancelled"
-                  : kind === "short" && chance(0.5)
-                    ? "interrupted"
-                    : "completed";
+                  : kind === "held queued turn"
+                    ? "queued"
+                    : kind === "short" && chance(0.5)
+                      ? "interrupted"
+                      : "completed";
           runs.push({
             id: runId,
             threadId,
@@ -5941,7 +5954,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
               ? run % 10 !== 0
               : kind !== "late wake" && (agentDriven ? run > 1 : chance(0.2));
           if (wakes) notification();
-          else userMessage(kind === "cancelled queued turn" ? "queued_turn" : "turn_start");
+          else
+            userMessage(
+              kind === "cancelled queued turn" || kind === "held queued turn"
+                ? "queued_turn"
+                : "turn_start",
+            );
           // A stopped run never folds, so it is long enough for page ends to fall inside it.
           const work =
             kind === "late wake"
@@ -6573,7 +6591,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
       assert.isNotEmpty(countQueries);
       for (const countQuery of countQueries) {
         for (const detail of yield* planOf(countQuery)) {
-          assert.notInclude(detail, "orchestration_v2_projection_turn_items");
+          if (detail.includes("orchestration_v2_projection_turn_items")) {
+            // Only held input is read from the sparse index; transcript history stays unread.
+            assert.include(
+              detail,
+              "SEARCH held USING INDEX orchestration_v2_projection_turn_items_queued_input_idx (run_id=?)",
+            );
+          }
         }
       }
     }),
@@ -10416,6 +10440,236 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
     }),
   );
 
+  it.effect("keeps undelivered queued input out of forks taken after a later run", () =>
+    Effect.gen(function* () {
+      const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
+      const now = yield* DateTime.now;
+      const projectId = ProjectId.make("project:projection-fork-held-queue");
+
+      for (const heldStatus of ["queued", "cancelled"] as const) {
+        const prefix = `projection-fork-held-queue:${heldStatus}`;
+        const sourceThreadId = ThreadId.make(`thread:${prefix}:source`);
+        const targetThreadId = ThreadId.make(`thread:${prefix}:target`);
+        const heldRunId = RunId.make(`run:${prefix}:held`);
+        const heldAttemptId = RunAttemptId.make(`attempt:${prefix}:held`);
+        const answeredRunId = RunId.make(`run:${prefix}:answered`);
+
+        for (const [threadId, forkedFrom] of [
+          [sourceThreadId, null],
+          [targetThreadId, { type: "run", threadId: sourceThreadId, runId: answeredRunId }],
+        ] as const) {
+          yield* projectionStore.apply({
+            id: EventId.make(`event:${threadId}:created`),
+            type: "thread.created",
+            threadId,
+            occurredAt: now,
+            payload: {
+              createdBy: "user",
+              creationSource: "web",
+              id: threadId,
+              projectId,
+              title: threadId,
+              providerInstanceId,
+              modelSelection,
+              runtimeMode: "full-access",
+              interactionMode: "default",
+              branch: null,
+              worktreePath: null,
+              activeProviderThreadId: null,
+              lineage: {
+                parentThreadId: forkedFrom === null ? null : sourceThreadId,
+                relationshipToParent: forkedFrom === null ? null : "fork",
+                rootThreadId: sourceThreadId,
+              },
+              forkedFrom,
+              createdAt: now,
+              updatedAt: now,
+              archivedAt: null,
+              settledOverride: null,
+              settledAt: null,
+              lastVisitedAt: null,
+              deletedAt: null,
+            },
+          });
+        }
+
+        const applyRun = (
+          runId: RunId,
+          ordinal: number,
+          status: typeof heldStatus | "completed" | "rolled_back",
+        ) =>
+          projectionStore.apply({
+            id: EventId.make(`event:${runId}:${status}`),
+            type: "run.updated",
+            threadId: sourceThreadId,
+            runId,
+            nodeId: NodeId.make(`node:${runId}`),
+            driver,
+            occurredAt: now,
+            payload: {
+              id: runId,
+              threadId: sourceThreadId,
+              ordinal,
+              providerInstanceId,
+              modelSelection,
+              providerThreadId: null,
+              userMessageId: MessageId.make(`message:${runId}`),
+              rootNodeId: NodeId.make(`node:${runId}`),
+              activeAttemptId: runId === heldRunId ? heldAttemptId : null,
+              status,
+              requestedAt: now,
+              startedAt: runId === heldRunId ? null : now,
+              completedAt: status === "queued" ? null : now,
+              checkpointId: null,
+              contextHandoffId: null,
+            },
+          });
+        const applyUserItem = (input: {
+          readonly threadId: ThreadId;
+          readonly runId: RunId | null;
+          readonly ordinal: number;
+          readonly text: string;
+          readonly inputIntent: "turn_start" | "queued_turn";
+        }) =>
+          projectionStore.apply({
+            id: EventId.make(`event:${input.threadId}:item:${input.ordinal}`),
+            type: "turn-item.updated",
+            threadId: input.threadId,
+            ...(input.runId === null ? {} : { runId: input.runId }),
+            occurredAt: now,
+            payload: {
+              createdBy: "user",
+              creationSource: "web",
+              id: TurnItemId.make(`turn-item:${input.threadId}:${input.ordinal}`),
+              threadId: input.threadId,
+              runId: input.runId,
+              nodeId: null,
+              providerThreadId: null,
+              providerTurnId: null,
+              nativeItemRef: null,
+              parentItemId: null,
+              ordinal: input.ordinal,
+              status: "completed",
+              title: null,
+              startedAt: now,
+              completedAt: now,
+              updatedAt: now,
+              type: "user_message",
+              messageId: MessageId.make(`message:${input.threadId}:${input.ordinal}`),
+              inputIntent: input.inputIntent,
+              text: input.text,
+              attachments: [],
+            },
+          });
+
+        yield* applyRun(heldRunId, 1, heldStatus);
+        yield* projectionStore.apply({
+          id: EventId.make(`event:${heldAttemptId}:created`),
+          type: "run-attempt.created",
+          threadId: sourceThreadId,
+          runId: heldRunId,
+          nodeId: NodeId.make(`node:${heldRunId}`),
+          driver,
+          occurredAt: now,
+          payload: {
+            id: heldAttemptId,
+            runId: heldRunId,
+            attemptOrdinal: 1,
+            rootNodeId: NodeId.make(`node:${heldRunId}`),
+            providerInstanceId,
+            providerThreadId: ProviderThreadId.make(`provider-thread:${prefix}`),
+            providerTurnId: null,
+            reason: "initial",
+            status: heldStatus === "queued" ? "pending" : "cancelled",
+            startedAt: null,
+            completedAt: heldStatus === "queued" ? null : now,
+          },
+        });
+        yield* applyRun(answeredRunId, 2, "completed");
+        yield* applyUserItem({
+          threadId: sourceThreadId,
+          runId: heldRunId,
+          ordinal: 100,
+          text: "held",
+          inputIntent: "queued_turn",
+        });
+        yield* applyUserItem({
+          threadId: sourceThreadId,
+          runId: answeredRunId,
+          ordinal: 200,
+          text: "answered",
+          inputIntent: "turn_start",
+        });
+        yield* applyUserItem({
+          threadId: sourceThreadId,
+          runId: answeredRunId,
+          ordinal: 201,
+          text: "delivered steer",
+          inputIntent: "queued_turn",
+        });
+        // Enough local turns for a turn-limited window to count inherited rows in SQL.
+        for (let index = 0; index < THREAD_HISTORY_PAGE_POLICY.maxUserTurns; index += 1) {
+          yield* applyUserItem({
+            threadId: targetThreadId,
+            runId: null,
+            ordinal: index,
+            text: `local ${index}`,
+            inputIntent: "turn_start",
+          });
+        }
+
+        const assertForkHidesHeldInput = Effect.fn("assertForkHidesHeldInput")(function* () {
+          const target = yield* projectionStore.getThreadProjection(targetThreadId);
+          assert.deepEqual(
+            target.visibleTurnItems
+              .slice(0, 3)
+              .map((row) => (row.item.type === "user_message" ? row.item.text : row.item.type)),
+            ["answered", "delivered steer", "fork"],
+          );
+          const page = yield* projectionStore.getTimelinePage(targetThreadId, {
+            view: "activity",
+            limit: target.visibleTurnItems.length + 1,
+          });
+          assert.deepEqual(page.items, target.visibleTurnItems);
+          const bounded = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
+            rowLimit: target.visibleTurnItems.length + 1,
+          });
+          assert.deepEqual(bounded.projection.visibleTurnItems, target.visibleTurnItems);
+          const shellSnapshot = yield* projectionStore.getShellSnapshot();
+          for (const projection of [
+            yield* projectionStore.getThreadProjection(sourceThreadId),
+            target,
+          ]) {
+            for (const shell of [
+              yield* projectionStore.getThreadShell(projection.thread.id),
+              shellSnapshot.threads.find((thread) => thread.id === projection.thread.id),
+            ]) {
+              assert.equal(shell?.visibleItemCount, projection.visibleTurnItems.length);
+            }
+          }
+          const filledLocalPage = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
+            rowLimit: target.visibleTurnItems.length,
+            userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns,
+          });
+          assert.deepEqual(
+            filledLocalPage.projection.visibleTurnItems.filter((row) => row.visibility === "local"),
+            target.visibleTurnItems.filter((row) => row.visibility === "local"),
+          );
+        });
+
+        yield* assertForkHidesHeldInput();
+        // Forks keep rolled-back source history, but still not undelivered input.
+        yield* applyRun(answeredRunId, 2, "rolled_back");
+        yield* assertForkHidesHeldInput();
+        // Rollback also takes cancelled runs, and overwrites the status that hid their input.
+        if (heldStatus === "cancelled") {
+          yield* applyRun(heldRunId, 1, "rolled_back");
+          yield* assertForkHidesHeldInput();
+        }
+      }
+    }),
+  );
+
   it.effect("hides source turns rewound before the fork from a fork's visible items", () =>
     Effect.gen(function* () {
       const projectionStore = yield* ProjectionStore.ProjectionStoreV2;