fix(server): retry delegated terminal result delivery

apps/server/src/mcp/OrchestratorMcpService.test.ts

@@ -3,6 +3,7 @@ import { assert, describe, it } from "@effect/vitest";
 import {
   CommandId,
   EnvironmentId,
+  type OrchestrationV2StoredEvent,
   NodeId,
   type OrchestrationV2ThreadShell,
   type ScheduledTask,
@@ -16,9 +17,13 @@ import {
   type ServerProvider,
 } from "@t3tools/contracts";
 import * as DateTime from "effect/DateTime";
+import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as Ref from "effect/Ref";
+import * as Queue from "effect/Queue";
+import * as Stream from "effect/Stream";
 
 import {
   DispatchModeLimit,
@@ -49,7 +54,7 @@ const layerOrchestratorService = OrchestratorMcpService.layer.pipe(
 );
 
 describe("OrchestratorMcpService", () => {
-  it.effect("retries terminal acknowledgement with a fresh command id", () =>
+  it.effect("defers acknowledgement until publication and retries with a fresh command id", () =>
     Effect.gen(function* () {
       const parentThreadId = ThreadId.make("thread:mcp-ack-parent");
       const childThreadId = ThreadId.make("thread:mcp-ack-child");
@@ -94,24 +99,26 @@ describe("OrchestratorMcpService", () => {
         ],
         contextTransfers: [],
         messages: [],
+        turnItems: [],
         subagents: [],
         providerThreads: [],
         runtimeRequests: [],
       } as unknown as OrchestrationV2ThreadProjection;
       let hasNestedWork = true;
+      let resultPublished = true;
       const layerDependencies = Layer.mergeAll(
         NodeServices.layer,
         Layer.mock(ThreadManagementService.ThreadManagementService)({
           getThreadRecords: (threadId) =>
             Effect.succeed(
               threadId === parentThreadId
-                ? hasNestedWork
+                ? hasNestedWork || !resultPublished
                   ? {
                       ...parentProjection,
                       subagents: parentProjection.subagents.map((task) => ({
                         ...task,
                         result: null,
-                        status: "running" as const,
+                        status: hasNestedWork ? ("running" as const) : ("completed" as const),
                       })),
                     }
                   : parentProjection
@@ -124,14 +131,15 @@ describe("OrchestratorMcpService", () => {
                     }
                   : childProjection,
             ),
+          delegatedTaskResultPending: () => Effect.succeed(false),
           dispatch: (command) =>
             Ref.update(acknowledgementCommandIds, (commandIds) => [
               ...commandIds,
               String(command.commandId),
             ]).pipe(
               Effect.andThen(Ref.updateAndGet(acknowledgementAttempts, (count) => count + 1)),
               Effect.flatMap((attempt) =>
-                attempt === 1
+                attempt === 1 && resultPublished
                   ? Effect.fail(new Error("simulated acknowledgement failure") as never)
                   : Effect.succeed({} as never),
               ),
@@ -167,6 +175,13 @@ describe("OrchestratorMcpService", () => {
         assert.isNull(pending.summary);
         assert.equal(yield* Ref.get(acknowledgementAttempts), 0);
         hasNestedWork = false;
+        resultPublished = false;
+        const unpublished = yield* service.taskStatus(scope, taskId);
+        assert.equal(unpublished.status, "completed");
+        assert.equal(unpublished.workState, "result_available");
+        assert.isNull(unpublished.resultContextTransferId);
+        assert.equal(yield* Ref.get(acknowledgementAttempts), 0);
+        resultPublished = true;
         const error = yield* service.taskStatus(scope, taskId).pipe(Effect.flip);
         assert.equal(error.code, "orchestration_error");
 
@@ -190,7 +205,7 @@ describe("OrchestratorMcpService", () => {
       const dispatched = yield* Ref.make(0);
       let awaitsRestart = true;
       let readFails = true;
-      const parentProjection = {
+      let parentProjection = {
         thread: { id: parentThreadId },
         runs: [],
         contextTransfers: [],
@@ -273,6 +288,18 @@ describe("OrchestratorMcpService", () => {
         awaitsRestart = false;
         const settled = yield* service.taskStatus(scope, taskId);
         assert.equal(settled.status, "cancelled");
+        assert.equal(yield* Ref.get(dispatched), 0);
+        parentProjection = {
+          ...parentProjection,
+          subagents: parentProjection.subagents.map((task) => ({
+            ...task,
+            status: "cancelled" as const,
+            result: "Child cancelled after restart.",
+          })),
+        };
+        const published = yield* service.taskStatus(scope, taskId);
+        assert.equal(published.status, "cancelled");
+        assert.equal(published.summary, "Child cancelled after restart.");
         assert.equal(yield* Ref.get(dispatched), 1);
       }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
     }),
@@ -878,6 +905,112 @@ describe("OrchestratorMcpService provider resolution", () => {
     turnItems: [],
   } as unknown as OrchestrationV2ThreadProjection;
 
+  it.effect("waits for publication before returning and acknowledging a terminal task", () =>
+    Effect.gen(function* () {
+      const watching = yield* Deferred.make<void>();
+      const updates = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
+      const acknowledgements = yield* Ref.make(0);
+      let published = false;
+      const task = {
+        id: taskId,
+        threadId: parentThreadId,
+        childThreadId,
+        origin: "app_owned",
+        driver: "codex",
+        providerInstanceId: codexInstanceId,
+        status: "completed",
+        model: "gpt-5.4",
+        result: null,
+        completionDelivery: { state: "pending" },
+      };
+      const layerDependencies = Layer.mergeAll(
+        NodeServices.layer,
+        Layer.mock(ThreadManagementService.ThreadManagementService)({
+          hasReceipt: () => Effect.succeed(false),
+          assertProviderInstanceEnabled: () => Effect.void,
+          getThreadEventSequence: () => Effect.succeed(0),
+          getThreadRecords: (threadId) =>
+            Effect.sync(() =>
+              threadId === parentThreadId
+                ? parentProjection([{ ...task, result: published ? "terminal result" : null }])
+                : ({
+                    ...childProjection,
+                    runs: [
+                      {
+                        id: RunId.make("run:mcp-wait-child"),
+                        ordinal: 1,
+                        status: "completed" as const,
+                        startedAt: DateTime.makeUnsafe("2026-10-03T10:00:00Z"),
+                      },
+                    ],
+                  } as unknown as OrchestrationV2ThreadProjection),
+            ),
+          delegatedTaskResultPending: () => Effect.succeed(false),
+          streamStoredEventsFrom: (
+            options: Parameters<
+              ThreadManagementService.ThreadManagementService["Service"]["streamStoredEventsFrom"]
+            >[0],
+          ) =>
+            options?.threadId === parentThreadId && options.eventType === "subagent.updated"
+              ? Stream.fromEffect(Deferred.succeed(watching, undefined)).pipe(
+                  Stream.drain,
+                  Stream.concat(Stream.fromQueue(updates)),
+                )
+              : Stream.never,
+          dispatch: (command) =>
+            (command.type === "delegated_task.completion-delivery.acknowledge"
+              ? Ref.update(acknowledgements, (count) => count + 1)
+              : Effect.void
+            ).pipe(
+              Effect.as({
+                sequence: 1,
+                storedEvents: [{ event: { type: "subagent.updated", payload: task } }],
+              } as never),
+            ),
+        }),
+        providerRegistryLayer([
+          providerSnapshot({
+            instanceId: codexInstanceId,
+            driver: ProviderDriverKind.make("codex"),
+            model: "gpt-5.4",
+          }),
+        ]),
+        adapterRegistryLayer([codexInstanceId]),
+        Layer.mock(ProjectService.ProjectService)({}),
+        Layer.mock(SecretRequests.SecretRequests)({}),
+        Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
+        ServerSettings.layerTest(),
+      );
+      yield* Effect.gen(function* () {
+        const service = yield* OrchestratorMcpService.OrchestratorMcpService;
+        const delegate = service.delegateTask(scope, {
+          task: "Summarize the diff.",
+          mode: "wait",
+          clientRequestId: "wait-for-publication",
+        });
+        const waiter = yield* delegate.pipe(Effect.forkChild);
+        yield* Effect.raceFirst(
+          Deferred.await(watching),
+          Fiber.join(waiter).pipe(Effect.andThen(Effect.die("Wait returned before publication"))),
+        );
+        assert.equal(yield* Ref.get(acknowledgements), 0);
+        published = true;
+        yield* Queue.offer(updates, {
+          event: { type: "subagent.updated", threadId: parentThreadId, payload: task },
+        } as OrchestrationV2StoredEvent);
+        const result = yield* Fiber.join(waiter);
+        assert.equal(result.status, "completed");
+        assert.equal(result.summary, "terminal result");
+        assert.isFalse(result.waitTimedOut);
+        assert.equal(yield* Ref.get(acknowledgements), 1);
+        // Legacy published rows need not have a result transfer.
+        assert.isNull(result.resultContextTransferId);
+        assert.equal((yield* delegate).summary, "terminal result");
+        assert.equal(yield* Ref.get(acknowledgements), 2);
+      }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
+    }),
+  );
+
   it.effect(
     "advertises orchestration capability from registered adapters rather than a driver allowlist",
     () =>

apps/server/src/mcp/OrchestratorMcpService.ts

@@ -1339,13 +1339,16 @@ const make = Effect.gen(function* () {
       ? crypto.randomUUIDv4().pipe(Effect.orDie)
       : Effect.succeed(clientRequestId);
 
-  const readTask = (
+  const readTaskObservation = (
     scope: McpThreadInvocationScope,
     taskId: NodeId,
     waitTimedOut = false,
     acknowledgeTerminal = false,
     acknowledgementOperation = "task-status-acknowledge",
-  ): Effect.Effect<OrchestratorMcpDelegateTaskResult, OrchestratorMcpFailure> =>
+  ): Effect.Effect<
+    { readonly response: OrchestratorMcpDelegateTaskResult; readonly published: boolean },
+    OrchestratorMcpFailure
+  > =>
     Effect.gen(function* () {
       yield* requireCapability(scope);
       const parentProjection = yield* loadProjection(scope.thread.threadId);
@@ -1487,6 +1490,7 @@ const make = Effect.gen(function* () {
       if (
         acknowledgeTerminal &&
         isTerminalTaskStatus(status) &&
+        task.result !== null &&
         task.completionDelivery?.state !== "acknowledged" &&
         task.completionDelivery?.state !== "disposed"
       ) {
@@ -1516,9 +1520,12 @@ const make = Effect.gen(function* () {
             ),
           );
       }
-      return response;
+      return { response, published: task.result !== null };
     });
 
+  const readTask = (...args: Parameters<typeof readTaskObservation>) =>
+    readTaskObservation(...args).pipe(Effect.map(({ response }) => response));
+
   const waitForTask = (scope: McpThreadInvocationScope, taskId: NodeId, timeoutMs: number) =>
     Effect.gen(function* () {
       const streamError = (error: unknown) =>
@@ -1531,24 +1538,32 @@ const make = Effect.gen(function* () {
       const afterSequence = yield* threadManagement
         .getThreadEventSequence(scope.thread.threadId)
         .pipe(Effect.mapError(streamError));
-      const initial = yield* readTask(scope, taskId, false, true);
-      if (initial.hasPendingQuestion || isTerminalTaskStatus(initial.status)) {
-        return Option.some(initial);
+      // Status can derive a terminal result before publication. A blocking
+      // wait owns delivery until it can acknowledge the published task row.
+      const settled = ({
+        response,
+        published,
+      }: Effect.Success<ReturnType<typeof readTaskObservation>>) =>
+        response.hasPendingQuestion || (isTerminalTaskStatus(response.status) && published);
+      const initial = yield* readTaskObservation(scope, taskId, false, true);
+      if (settled(initial)) {
+        return Option.some(initial.response);
       }
       // One stream per event type, so transcript events never fill a buffer.
       return yield* Stream.mergeAll(
         TASK_WAKE_EVENTS.map(({ thread, eventType }) =>
           threadManagement.streamStoredEventsFrom({
-            threadId: thread === "parent" ? scope.thread.threadId : initial.childThreadId,
+            threadId: thread === "parent" ? scope.thread.threadId : initial.response.childThreadId,
             afterSequence,
             eventType,
           }),
         ),
         { concurrency: "unbounded" },
       ).pipe(
         Stream.mapError(streamError),
-        Stream.mapEffect(() => readTask(scope, taskId, false, true)),
-        Stream.filter((result) => result.hasPendingQuestion || isTerminalTaskStatus(result.status)),
+        Stream.mapEffect(() => readTaskObservation(scope, taskId, false, true)),
+        Stream.filter(settled),
+        Stream.map(({ response }) => response),
         Stream.runHead,
       );
     }).pipe(Effect.timeoutOption(Duration.millis(timeoutMs)), Effect.map(Option.flatten));