fix(server): drop a finished child's queued question wake

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

@@ -4,6 +4,8 @@ import {
   type OrchestrationV2DomainEvent,
   type OrchestrationV2RuntimeRequest,
   type OrchestrationV2Subagent,
+  type RuntimeRequestId,
+  type ThreadId,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
@@ -58,6 +60,37 @@ const make = Effect.gen(function* () {
     });
   });
 
+  const cancelQueuedWakes = Effect.fnUntraced(function* (
+    parentThreadId: ThreadId,
+    requestIds: ReadonlyArray<RuntimeRequestId>,
+  ) {
+    const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]);
+    for (const requestId of requestIds) {
+      const key = `${parentThreadId}:${requestId}`;
+      const messageId = MessageId.make(`message:delegated-task-question:${key}`);
+      const queued = parent.runs.find(
+        (run) => run.status === "queued" && run.userMessageId === messageId,
+      );
+      if (queued === undefined) continue;
+      yield* engine
+        .dispatch({
+          type: "queued-run.cancel",
+          commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`),
+          threadId: parentThreadId,
+          runId: queued.id,
+        })
+        .pipe(
+          Effect.catchCause((cause) =>
+            Effect.logWarning("cancelling a delegated task question wake failed", {
+              threadId: parentThreadId,
+              requestId,
+              cause,
+            }),
+          ),
+        );
+    }
+  });
+
   // Runs for every domain event; most events return before any work is needed.
   const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = Effect.fnUntraced(
     function* (event) {
@@ -100,28 +133,55 @@ const make = Effect.gen(function* () {
         }
         return;
       }
+      if (event.type === "subagent.updated") {
+        const task = event.payload;
+        if (
+          task.origin !== "app_owned" ||
+          task.childThreadId === null ||
+          !["completed", "failed", "cancelled", "interrupted"].includes(task.status)
+        ) {
+          return;
+        }
+        // A settled task can no longer use an answer to its still-pending questions.
+        const { runtimeRequests } = yield* projections.getThreadRecords(task.childThreadId, [
+          "runtimeRequests",
+        ]);
+        const questions = runtimeRequests.filter((request) => request.kind === "user_input");
+        if (questions.length === 0) return;
+        const child = yield* projections.getThreadRecords(task.childThreadId, ["runs", "nodes"]);
+        const taskRunIds = new Set(
+          child.runs
+            .filter(
+              (run) =>
+                run.delegatedTaskId === task.id ||
+                (run.delegatedTaskId === undefined &&
+                  child.thread.forkedFrom?.type === "node" &&
+                  child.thread.forkedFrom.nodeId === task.id),
+            )
+            .map((run) => run.id),
+        );
+        const taskNodeIds = new Set(
+          child.nodes
+            .filter((node) => node.runId !== null && taskRunIds.has(node.runId))
+            .map((node) => node.id),
+        );
+        yield* cancelQueuedWakes(
+          task.threadId,
+          questions
+            .filter((request) => taskNodeIds.has(request.nodeId))
+            .map((request) => request.id),
+        );
+        return;
+      }
       if (event.type !== "runtime-request.updated" || event.payload.kind !== "user_input") {
         return;
       }
       const child = yield* projections.getThread(event.threadId);
       const parentThreadId = child.lineage.parentThreadId;
       if (parentThreadId === null) return;
       const request = yield* projections.getRuntimeRequest(event.threadId, event.payload.id);
-      const key = `${parentThreadId}:${event.payload.id}`;
-      const messageId = MessageId.make(`message:delegated-task-question:${key}`);
       if (event.payload.status !== "pending" || request?.status !== "pending") {
-        const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]);
-        const queued = parent.runs.find(
-          (run) => run.status === "queued" && run.userMessageId === messageId,
-        );
-        if (queued !== undefined) {
-          yield* engine.dispatch({
-            type: "queued-run.cancel",
-            commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`),
-            threadId: parentThreadId,
-            runId: queued.id,
-          });
-        }
+        yield* cancelQueuedWakes(parentThreadId, [event.payload.id]);
         return;
       }
       const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]);

apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts

@@ -1574,6 +1574,234 @@ it.effect("wakes the parent once when its delegated task asks a question", () =>
   }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))),
 );
 
+it.effect.each([false, true])(
+  "drops a stale question wake (missed terminal event: %s)",
+  (missedEvent) =>
+    Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const projections = yield* ProjectionStore.ProjectionStoreV2;
+      const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor;
+      const eventSink = yield* EventSink.EventSinkV2;
+      const parentThreadId = ThreadId.make("thread:stale-question-parent");
+      yield* orchestrator.dispatch({
+        type: "thread.create",
+        commandId: CommandId.make("create-stale-question-parent"),
+        threadId: parentThreadId,
+        projectId: ProjectId.make("project:stale-question"),
+        title: "Parent",
+        modelSelection,
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        branch: null,
+        worktreePath: null,
+        createdBy: "user",
+        creationSource: "web",
+      });
+      yield* orchestrator.dispatch({
+        type: "message.dispatch",
+        commandId: CommandId.make("start-stale-question-parent"),
+        threadId: parentThreadId,
+        messageId: MessageId.make("message:stale-question-parent"),
+        text: "Delegate the proof",
+        attachments: [],
+        dispatchMode: { type: "start_immediately" },
+        createdBy: "user",
+        creationSource: "web",
+      });
+      const parentRun = (yield* projections.getThreadProjection(parentThreadId)).runs[0]!;
+      yield* orchestrator.dispatch({
+        type: "delegated_task.request",
+        commandId: CommandId.make("delegate-stale-question-task"),
+        parentThreadId,
+        parentRunId: parentRun.id,
+        parentNodeId: parentRun.rootNodeId!,
+        task: "Capture the proof",
+        completionWake: "settled_only",
+        modelSelection,
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        createdBy: "agent",
+        creationSource: "mcp",
+      });
+      const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!;
+      const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"]))
+        .runs[0]!;
+      const now = yield* DateTime.now;
+      const question = {
+        id: EventId.make("event:stale-question"),
+        type: "runtime-request.updated",
+        threadId: task.childThreadId!,
+        runId: childRun.id,
+        occurredAt: now,
+        payload: {
+          id: RuntimeRequestId.make("request:stale-question"),
+          nodeId: childRun.rootNodeId!,
+          providerTurnId: null,
+          nativeRequestRef: null,
+          kind: "user_input",
+          status: "pending",
+          responseCapability: { type: "message" },
+          createdAt: now,
+          resolvedAt: null,
+        },
+      } satisfies OrchestrationV2DomainEvent;
+      yield* eventSink.write({ events: [question] });
+      yield* reactor.handle(question);
+      const wake = (yield* projections.getThreadRecords(parentThreadId, [
+        "messages",
+      ])).messages.find((message) => message.notification !== undefined)!;
+      const wakeRun = () =>
+        projections
+          .getThreadRecords(parentThreadId, ["runs"])
+          .pipe(Effect.map(({ runs }) => runs.find((run) => run.userMessageId === wake.id)));
+      assert.equal((yield* wakeRun())?.status, "queued");
+
+      // The child finishes without ever resolving its message-capability question.
+      const childCompleted = {
+        id: EventId.make("event:stale-question-child-completed"),
+        type: "run.updated",
+        threadId: task.childThreadId!,
+        runId: childRun.id,
+        occurredAt: now,
+        payload: { ...childRun, status: "completed", completedAt: now },
+      } satisfies OrchestrationV2DomainEvent;
+      const taskCompleted = {
+        id: EventId.make("event:stale-question-task-completed"),
+        type: "subagent.updated",
+        threadId: parentThreadId,
+        occurredAt: now,
+        payload: { ...task, status: "completed", result: "Proof captured.", completedAt: now },
+      } satisfies OrchestrationV2DomainEvent;
+      yield* eventSink.write({
+        commandId: CommandId.make("command:runtime-reconcile:stale-question"),
+        events: [childCompleted, taskCompleted],
+      });
+      yield* reactor.handle(childCompleted);
+      if (missedEvent) {
+        // An acknowledged result drops the task from the parent's command projection.
+        yield* orchestrator.dispatch({
+          type: "delegated_task.completion-delivery.acknowledge",
+          commandId: CommandId.make("ack-stale-question-task"),
+          parentThreadId,
+          taskId: task.id,
+          observedByRunId: parentRun.id,
+        });
+        // Persisted terminal state can predate the reactor subscription after restart.
+        yield* projections.apply({
+          id: EventId.make("event:stale-question-parent-finished"),
+          type: "run.updated",
+          threadId: parentThreadId,
+          occurredAt: now,
+          payload: { ...parentRun, status: "completed", completedAt: now },
+        });
+        yield* orchestrator.dispatch({
+          type: "queue.resume",
+          commandId: CommandId.make("resume-stale-question"),
+          threadId: parentThreadId,
+        });
+      } else {
+        yield* reactor.handle(taskCompleted);
+      }
+      assert.equal((yield* wakeRun())?.status, "cancelled");
+      if (missedEvent) {
+        const before = (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.length;
+        const delayedQuestion = {
+          ...question,
+          id: EventId.make("event:delayed-question"),
+          payload: { ...question.payload, id: RuntimeRequestId.make("request:delayed-question") },
+        };
+        yield* eventSink.write({ events: [delayedQuestion] });
+        yield* reactor.handle(delayedQuestion);
+        // A stale delivery is accepted without starting work, not rejected.
+        yield* orchestrator.dispatch({
+          type: "message.dispatch",
+          commandId: CommandId.make("stale-question-redelivery"),
+          threadId: parentThreadId,
+          messageId: MessageId.make("message:stale-question-redelivery"),
+          text: "Answer the delegated question",
+          notification: {
+            source: {
+              kind: "delegated_task",
+              taskIds: [task.id],
+              childThreadId: task.childThreadId!,
+            },
+            outcome: "updated",
+            summary: "Delegated task has a question",
+          },
+          attachments: [],
+          dispatchMode: { type: "queue_after_active" },
+          createdBy: "agent",
+          creationSource: "server",
+        });
+        assert.equal(
+          (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.length,
+          before,
+        );
+        return;
+      }
+
+      yield* orchestrator.dispatch({
+        type: "message.dispatch",
+        commandId: CommandId.make("stale-question-followup"),
+        threadId: task.childThreadId!,
+        messageId: MessageId.make("message:stale-question-followup"),
+        text: "Capture another proof",
+        attachments: [],
+        dispatchMode: { type: "start_immediately" },
+        createdBy: "user",
+        creationSource: "web",
+      });
+      const followup = (yield* projections.getThreadRecords(task.childThreadId!, [
+        "runs",
+      ])).runs.find((run) => run.userMessageId === "message:stale-question-followup")!;
+      const followupQuestion = {
+        ...question,
+        id: EventId.make("event:stale-question-followup"),
+        runId: followup.id,
+        payload: {
+          ...question.payload,
+          id: RuntimeRequestId.make("request:stale-question-followup"),
+          nodeId: followup.rootNodeId!,
+        },
+      };
+      yield* eventSink.write({ events: [followupQuestion] });
+      yield* reactor.handle(followupQuestion);
+      const followupWake = (yield* projections.getThreadRecords(parentThreadId, [
+        "runs",
+      ])).runs.find(
+        (run) =>
+          run.userMessageId ===
+          `message:delegated-task-question:${parentThreadId}:${followupQuestion.payload.id}`,
+      )!;
+      assert.equal(followupWake.status, "queued");
+      yield* reactor.handle(taskCompleted);
+      assert.equal(
+        (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find(
+          (run) => run.id === followupWake.id,
+        )?.status,
+        "queued",
+      );
+      yield* projections.apply({
+        id: EventId.make("event:followup-question-parent-finished"),
+        type: "run.updated",
+        threadId: parentThreadId,
+        occurredAt: now,
+        payload: { ...parentRun, status: "completed", completedAt: now },
+      });
+      yield* orchestrator.dispatch({
+        type: "queue.resume",
+        commandId: CommandId.make("resume-followup-question"),
+        threadId: parentThreadId,
+      });
+      assert.equal(
+        (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find(
+          (run) => run.id === followupWake.id,
+        )?.status,
+        "starting",
+      );
+    }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))),
+);
+
 it.effect.each([false, true])("rejects thread.create for an existing id (deleted: %s)", (deleted) =>
   Effect.gen(function* () {
     const orchestrator = yield* Orchestrator.OrchestratorV2;