fix: recover stale subagent completion announces

galiniliev · web-flow · commit 04eac15f43d5 · 2026-05-18T23:09:20.000-07:00
Recover stale subagent completion delivery by retrying unsupported transcript-wait wakes without transcript waiting and forcing the existing message-tool handoff when the requester run is stale and direct completion is invisible.\n\nAdds regression coverage for the stale wake sequence and records the maintainer changelog entry.\n\nFixes #83699.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -6,6 +6,7 @@ Docs: https://docs.openclaw.ai
 
 ### Fixes
 
+- Agents/subagents: recover stale completion announces by retrying unsupported transcript-wait wakes without transcript waiting and forcing a message-tool handoff when the requester run is already stale. Fixes #83699. (#83700) Thanks @galiniliev.
 - Agents/subagents: skip stale embedded-run wake probes for dormant completion requesters, so late subagent completions go straight to requester-agent/direct handoff instead of producing `reason=no_active_run` queue noise. (#82964) Thanks @galiniliev.
 - CLI: retry config snapshot reads after a transient failure so one rejected read no longer poisons later commands in the same process. (#83931) Thanks @honor2030.
 - WhatsApp: clarify inbound group diagnostics so observed but unregistered groups point to `channels.whatsapp.groups` without changing routing or sender authorization. (#83846) Thanks @neeravmakwana.
diff --git a/src/agents/subagent-announce-delivery.test.ts b/src/agents/subagent-announce-delivery.test.ts
@@ -5,6 +5,7 @@ import {
 } from "../infra/outbound/session-binding-service.js";
 import type { AgentInternalEvent } from "./internal-events.js";
 import type {
+  EmbeddedPiQueueFailureReason,
   EmbeddedPiQueueMessageOptions,
   EmbeddedPiQueueMessageOutcome,
 } from "./pi-embedded-runner/runs.js";
@@ -36,6 +37,17 @@ function createGatewayMock(response: Record<string, unknown> = {}) {
   return vi.fn(async () => response) as unknown as typeof runtimeCallGateway;
 }
 
+function createGatewaySequenceMock(
+  responses: Record<string, unknown>[],
+): ReturnType<typeof vi.fn> & typeof runtimeCallGateway {
+  let index = 0;
+  return vi.fn(async () => {
+    const response = responses[Math.min(index, responses.length - 1)] ?? {};
+    index += 1;
+    return response;
+  }) as unknown as ReturnType<typeof vi.fn> & typeof runtimeCallGateway;
+}
+
 function createInProcessGatewayMock(response: Record<string, unknown> = {}) {
   return vi.fn(async () => response) as unknown as typeof runtimeDispatchGatewayMethodInProcess;
 }
@@ -78,6 +90,29 @@ function createQueueOutcomeMock(
   );
 }
 
+function createQueueOutcomeSequenceMock(
+  queuedOutcomes: (boolean | EmbeddedPiQueueFailureReason)[],
+): ReturnType<typeof vi.fn<QueueEmbeddedPiMessageWithOutcome>> {
+  let index = 0;
+  return vi.fn((sessionId: string) => {
+    const outcome = queuedOutcomes[Math.min(index, queuedOutcomes.length - 1)] ?? false;
+    index += 1;
+    return outcome === true
+      ? {
+          queued: true,
+          sessionId,
+          target: "embedded_run",
+          gatewayHealth: "live",
+        }
+      : {
+          queued: false,
+          sessionId,
+          reason: typeof outcome === "string" ? outcome : "not_streaming",
+          gatewayHealth: "live",
+        };
+  });
+}
+
 const longChildCompletionOutput = [
   "34/34 tests pass, clean build. Now docker repro:",
   "Root cause: the requester's announce delivery accepted a prefix-only assistant payload as delivered.",
@@ -1170,6 +1205,104 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
     expect(sendMessage).not.toHaveBeenCalled();
   });
 
+  it("forces message-tool thread completions after transcript-wait wake falls stale", async () => {
+    const callGateway = createGatewaySequenceMock([
+      {
+        result: {
+          payloads: [],
+        },
+      },
+      {
+        result: {
+          payloads: [],
+          messagingToolSentTargets: [
+            {
+              tool: "message",
+              provider: "slack",
+              accountId: "acct-1",
+              to: "channel:C123",
+              threadId: "171.222",
+              text: "The background task completed.",
+            },
+          ],
+        },
+      },
+    ]);
+    const sendMessage = createSendMessageMock();
+    const queueEmbeddedPiMessageWithOutcome = createQueueOutcomeSequenceMock([
+      "transcript_commit_wait_unsupported",
+      "no_active_run",
+    ]);
+    const result = await deliverSlackThreadAnnouncement({
+      callGateway,
+      sendMessage,
+      queueEmbeddedPiMessageWithOutcome,
+      sessionId: "requester-session-4",
+      isActive: true,
+      expectsCompletionMessage: true,
+      directIdempotencyKey: "announce-thread-fallback-empty",
+      internalEvents: [
+        {
+          type: "task_completion",
+          source: "subagent",
+          childSessionKey: "agent:worker:subagent:child",
+          childSessionId: "child-session-id",
+          announceType: "subagent task",
+          taskLabel: "thread completion smoke",
+          status: "ok",
+          statusLabel: "completed successfully",
+          result: "child completion output",
+          replyInstruction: "Summarize the result.",
+        },
+      ],
+    });
+
+    expectRecordFields(result, {
+      delivered: true,
+      path: "direct",
+    });
+    expect(callGateway).toHaveBeenCalledTimes(2);
+    expectGatewayAgentParams(callGateway, {
+      deliver: true,
+      channel: "slack",
+      accountId: "acct-1",
+      to: "channel:C123",
+      threadId: "171.222",
+    });
+    expectRecordFields(mockCallArg(callGateway, 1).params, {
+      deliver: false,
+      channel: "slack",
+      accountId: "acct-1",
+      to: "channel:C123",
+      threadId: "171.222",
+      sourceReplyDeliveryMode: "message_tool_only",
+      idempotencyKey: "announce-thread-fallback-empty:message-tool",
+    });
+    expect(queueEmbeddedPiMessageWithOutcome).toHaveBeenCalledTimes(2);
+    expect(queueEmbeddedPiMessageWithOutcome).toHaveBeenNthCalledWith(
+      1,
+      "requester-session-4",
+      "child done",
+      {
+        debounceMs: 500,
+        deliveryTimeoutMs: 120_000,
+        steeringMode: "all",
+        waitForTranscriptCommit: true,
+      },
+    );
+    expect(queueEmbeddedPiMessageWithOutcome).toHaveBeenNthCalledWith(
+      2,
+      "requester-session-4",
+      "child done",
+      {
+        debounceMs: 500,
+        deliveryTimeoutMs: 120_000,
+        steeringMode: "all",
+      },
+    );
+    expect(sendMessage).not.toHaveBeenCalled();
+  });
+
   it("keeps concise requester rewrites primary even when child output is long", async () => {
     const callGateway = createGatewayMock({
       result: {
@@ -1434,18 +1567,23 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
     expect(callGateway).toHaveBeenCalledTimes(1);
   });
 
-  it("reports failure when announce-agent returns no visible output", async () => {
+  it("reports message-tool failure when stale thread completion remains invisible", async () => {
     const callGateway = createGatewayMock({
       result: {
         payloads: [],
       },
     });
     const sendMessage = createSendMessageMock();
+    const queueEmbeddedPiMessageWithOutcome = createQueueOutcomeSequenceMock([
+      "transcript_commit_wait_unsupported",
+      "no_active_run",
+    ]);
     const result = await deliverSlackThreadAnnouncement({
       callGateway,
       sendMessage,
+      queueEmbeddedPiMessageWithOutcome,
       sessionId: "requester-session-4",
-      isActive: false,
+      isActive: true,
       expectsCompletionMessage: true,
       directIdempotencyKey: "announce-thread-fallback-empty",
       internalEvents: [
@@ -1467,9 +1605,9 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
     expectRecordFields(result, {
       delivered: false,
       path: "direct",
-      error: "completion agent did not produce a visible reply",
+      error: "completion agent did not deliver through the message tool",
     });
-    expect(callGateway).toHaveBeenCalledTimes(1);
+    expect(callGateway).toHaveBeenCalledTimes(2);
     expect(sendMessage).not.toHaveBeenCalled();
   });
 
@@ -2141,18 +2279,23 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
     expect(sendMessage).not.toHaveBeenCalled();
   });
 
-  it("reports channel completion failure when announce-agent returns no visible output", async () => {
+  it("reports channel message-tool failure when stale completion remains invisible", async () => {
     const callGateway = createGatewayMock({
       result: {
         payloads: [],
       },
     });
     const sendMessage = createSendMessageMock();
+    const queueEmbeddedPiMessageWithOutcome = createQueueOutcomeSequenceMock([
+      "transcript_commit_wait_unsupported",
+      "no_active_run",
+    ]);
     const result = await deliverSlackChannelAnnouncement({
       callGateway,
       sendMessage,
+      queueEmbeddedPiMessageWithOutcome,
       sessionId: "requester-session-channel",
-      isActive: false,
+      isActive: true,
       expectsCompletionMessage: true,
       directIdempotencyKey: "announce-channel-fallback-empty",
       internalEvents: [
@@ -2174,9 +2317,9 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
     expectRecordFields(result, {
       delivered: false,
       path: "direct",
-      error: "completion agent did not produce a visible reply",
+      error: "completion agent did not deliver through the message tool",
     });
-    expect(callGateway).toHaveBeenCalledTimes(1);
+    expect(callGateway).toHaveBeenCalledTimes(2);
     expect(sendMessage).not.toHaveBeenCalled();
   });
 
diff --git a/src/agents/subagent-announce-delivery.ts b/src/agents/subagent-announce-delivery.ts
@@ -29,7 +29,10 @@ import {
   hasVisibleAgentPayload,
 } from "./pi-embedded-runner/delivery-evidence.js";
 import type { EmbeddedPiQueueMessageOptions } from "./pi-embedded-runner/run-state.js";
-import type { EmbeddedPiQueueMessageOutcome } from "./pi-embedded-runner/runs.js";
+import type {
+  EmbeddedPiQueueFailureReason,
+  EmbeddedPiQueueMessageOutcome,
+} from "./pi-embedded-runner/runs.js";
 import {
   callGateway,
   createBoundDeliveryRouter,
@@ -680,6 +683,8 @@ async function sendSubagentAnnounceDirectly(params: {
       ? "message_tool_only"
       : undefined;
     const shouldDeliverAgentFinal = deliveryTarget.deliver && !requiresMessageToolDelivery;
+    let completionWakeFailureReason: EmbeddedPiQueueFailureReason | undefined;
+    let completionWakeRetriedWithoutTranscriptWait = false;
     const requesterActivity = resolveRequesterSessionActivity(canonicalRequesterSessionKey);
     const requesterQueueSettings = resolveQueueSettings({
       cfg,
@@ -696,21 +701,32 @@ async function sendSubagentAnnounceDirectly(params: {
       requesterActivity.sessionId &&
       requesterActivity.isActive
     ) {
-      const wakeOutcome = await resolveQueueEmbeddedPiMessageOutcome(
+      const wakeOptions: EmbeddedPiQueueMessageOptions = {
+        deliveryTimeoutMs: announceTimeoutMs,
+        steeringMode: "all",
+        ...(completionSourceReplyDeliveryMode
+          ? { sourceReplyDeliveryMode: completionSourceReplyDeliveryMode }
+          : {}),
+        ...(requesterQueueSettings.debounceMs !== undefined
+          ? { debounceMs: requesterQueueSettings.debounceMs }
+          : {}),
+        waitForTranscriptCommit: true,
+      };
+      let wakeOutcome = await resolveQueueEmbeddedPiMessageOutcome(
         requesterActivity.sessionId,
         params.triggerMessage,
-        {
-          deliveryTimeoutMs: announceTimeoutMs,
-          steeringMode: "all",
-          ...(completionSourceReplyDeliveryMode
-            ? { sourceReplyDeliveryMode: completionSourceReplyDeliveryMode }
-            : {}),
-          ...(requesterQueueSettings.debounceMs !== undefined
-            ? { debounceMs: requesterQueueSettings.debounceMs }
-            : {}),
-          waitForTranscriptCommit: true,
-        },
+        wakeOptions,
       );
+      if (!wakeOutcome.queued && wakeOutcome.reason === "transcript_commit_wait_unsupported") {
+        const bestEffortWakeOptions = { ...wakeOptions };
+        delete bestEffortWakeOptions.waitForTranscriptCommit;
+        completionWakeRetriedWithoutTranscriptWait = true;
+        wakeOutcome = await resolveQueueEmbeddedPiMessageOutcome(
+          requesterActivity.sessionId,
+          params.triggerMessage,
+          bestEffortWakeOptions,
+        );
+      }
       if (wakeOutcome.queued) {
         return {
           delivered: true,
@@ -719,6 +735,7 @@ async function sendSubagentAnnounceDirectly(params: {
           path: "steered",
         };
       }
+      completionWakeFailureReason = wakeOutcome.reason;
       defaultRuntime.log(
         `[warn] Active requester session could not be woken for subagent completion; falling back to requester-agent handoff: ${formatQueueWakeFailureError(
           "active requester session could not be woken",
@@ -834,6 +851,41 @@ async function sendSubagentAnnounceDirectly(params: {
       shouldDeliverAgentFinal &&
       !hasVisibleGatewayAgentPayload(directAnnounceResponse)
     ) {
+      if (
+        completionWakeRetriedWithoutTranscriptWait &&
+        (completionWakeFailureReason === "no_active_run" ||
+          completionWakeFailureReason === "transcript_commit_wait_unsupported")
+      ) {
+        const forcedMessageToolResponse = await runAnnounceDeliveryWithRetry({
+          operation: "completion message-tool announce agent call",
+          signal: params.signal,
+          run: async () =>
+            await runAnnounceAgentCall({
+              agentParams: {
+                ...directAgentParams,
+                deliver: false,
+                sourceReplyDeliveryMode: "message_tool_only",
+                idempotencyKey: `${params.directIdempotencyKey}:message-tool`,
+              },
+              expectFinal: true,
+              timeoutMs: announceTimeoutMs,
+            }),
+        });
+        if (
+          isGatewayAgentRunPending(forcedMessageToolResponse) ||
+          hasGatewayAgentMessagingToolDelivery(forcedMessageToolResponse)
+        ) {
+          return {
+            delivered: true,
+            path: "direct",
+          };
+        }
+        return {
+          delivered: false,
+          path: "direct",
+          error: "completion agent did not deliver through the message tool",
+        };
+      }
       return {
         delivered: false,
         path: "direct",
