fix(announce): durable in-process queue fallback for direct-pending handoffs

bradfreels · bradfreels · commit 006d0a33c32e · 2026-06-04T03:19:02.000Z
When subagent direct-announce dispatch returns a non-terminal status
("accepted"/"started"/"in_flight") and the requester parent is not
streaming (yielded, queued, compaction-prep, or tool-result truncated),
the previous behaviour returned {delivered:false, error:"completion
agent handoff is still pending"}, surfacing as the dominant
`delivery_failed` audit-finding signature.

Route those cases into the requester's in-process durable system-event
inbox instead, keyed by the announce idempotency key. The inbox is
idempotent on (text, contextKey, deliveryContext), bounded at 20 events
per session, and drained on the parent's next turn-start (the same path
already used by jobs, hooks, restart sentinels, ACP child spawns,
monitor events, and the group/channel completion path in task-registry).

The `SubagentDeliveryPath` union gains a new `"durable_queue""` literal
so consumers can route on it. The harness-task-runtime durability check
(`isDurableAgentHarnessCompletionDelivery`) is extended to recognise the
new path, preventing Codex native-subagent monitor retry from churning
duplicates against an already-committed inbox enqueue.

If `enqueueSystemEvent` itself throws, the original give-up payload
(structured `reason: "completion_handoff_pending"` + error string) is
returned as a safety net so existing retry/give-up bookkeeping remains
the floor.

Tests:
- 5 unit tests in subagent-announce-delivery.test.ts covering the new
  branch, idempotency on duplicate fires, no-fallback-on-active-stream,
  no-fallback-with-media (preserve existing path), and result-envelope
  shape.
- 1 integration test in embedded-agent-runner/sessions-yield.orchestration
  reproducing the full failure class end-to-end: parent yield -&gt;
  isEmbeddedAgentRunActive(false) -&gt; announce with stubbed callGateway
  returning `accepted` -&gt; assert path:"durable_queue" -&gt; drain returns
  the trigger.
- 1 unit test in agent-harness-task-runtime.test.ts asserting
  durable_queue is treated as durable and that delivered:false still
  fails fast.

Targeted suite (5 files): 150/150 pass.
Full unit lane: 10071 pass; the 5 pre-existing failures on main
(`status.summary.runtime`, `subagents-tool` config-isolation issues)
reproduce with no patch applied and are not regressions from this
change.
diff --git a/src/agents/embedded-agent-runner/sessions-yield.orchestration.test.ts b/src/agents/embedded-agent-runner/sessions-yield.orchestration.test.ts
@@ -3,7 +3,17 @@
  * with no pending tool calls, so the parent session is idle when subagent
  * results arrive.
  */
-import { beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  drainSystemEvents,
+  peekSystemEvents,
+  resetSystemEventsForTest,
+} from "../../infra/system-events.js";
+import {
+  deliverSubagentAnnouncement,
+  testing as subagentAnnounceTesting,
+} from "../subagent-announce-delivery.js";
+import { callGateway as runtimeCallGateway } from "../subagent-announce-delivery.runtime.js";
 import { makeAttemptResult } from "./run.overflow-compaction.fixture.js";
 import {
   loadRunOverflowCompactionHarness,
@@ -126,4 +136,117 @@ describe("sessions_yield orchestration", () => {
     expect(result.meta.stopReason).toBeUndefined();
     expect(result.meta.pendingToolCalls).toBeUndefined();
   });
+
+  // ──────────────────────────────────────────────────────────────────────
+  // Integration repro: parent yield → child completion → durable inbox →
+  // next-turn drain. End-to-end coverage of the failure class that
+  // produced today's `delivery_failed` audit findings: a yielded parent
+  // session has no active embedded run, so the gateway agent-call dispatch
+  // returns a non-terminal `accepted/started/in_flight` status. The patch
+  // routes through the durable in-process system-event inbox keyed by the
+  // announce idempotency key; the parent drains the inbox at next
+  // turn-start. This test verifies the loop closes — the trigger message
+  // is in the inbox after the announce, and `drainSystemEvents` returns it
+  // for the next prompt build.
+  // ──────────────────────────────────────────────────────────────────────
+  it("end-to-end: yielded parent + child completion via direct-pending → durable inbox → next-turn drain", async () => {
+    // Use a canonical sessionKey (starts with 'agent:') so
+    // resolveRequesterStoreKey leaves it untouched.
+    const parentSessionKey = "agent:main:session:yield-integration-parent-key";
+    const parentSessionId = "yield-integration-parent-session";
+
+    // Stage 1: parent yields. After this attempt resolves with
+    // yieldDetected=true, the parent is removed from ACTIVE_EMBEDDED_RUNS
+    // and isStreaming() returns false — exactly the `not_streaming` state
+    // that makes embedded steer fail and triggers the patched fallback.
+    mockedRunEmbeddedAttempt.mockResolvedValueOnce(
+      makeAttemptResult({
+        promptError: null,
+        sessionIdUsed: parentSessionId,
+        yieldDetected: true,
+      }),
+    );
+
+    const yieldResult = await runEmbeddedAgent({
+      ...overflowBaseRunParams,
+      sessionId: parentSessionId,
+      sessionKey: parentSessionKey,
+      runId: "run-yield-integration",
+    });
+
+    expect(yieldResult.meta.stopReason).toBe("end_turn");
+    expect(yieldResult.meta.pendingToolCalls).toBeUndefined();
+    expect(isEmbeddedAgentRunActive(parentSessionId)).toBe(false);
+
+    // Sanity: a steer attempt now would fail with no_active_run — same
+    // condition that drives the announce flow's direct-dispatch branch.
+    const steerProbe = queueEmbeddedAgentMessageWithOutcome(parentSessionId, "would-be-steer");
+    expect(steerProbe.queued).toBe(false);
+
+    // Stage 2: child completion fires via deliverSubagentAnnouncement.
+    // Stub callGateway to mimic the `not_streaming` path: the gateway
+    // accepts the run but returns a non-terminal status, which is the
+    // exact bug-class signature in today's audit drops (4 of 4 had error
+    // "completion agent handoff is still pending").
+    resetSystemEventsForTest();
+    const callGatewayStub = vi.fn(async () => ({
+      runId: "agent:main:run:integration-pending",
+      status: "accepted" as const,
+      acceptedAt: Date.now(),
+    })) as unknown as typeof runtimeCallGateway;
+    subagentAnnounceTesting.setDepsForTest({
+      callGateway: callGatewayStub,
+      getRequesterSessionActivity: () => ({
+        sessionId: parentSessionId,
+        isActive: false, // mirrors yielded/non-streaming reality
+      }),
+    });
+
+    const triggerText = "child cipher:integration-test done: APPROVE — durable-queue integration";
+    const announceOrigin = {
+      channel: "discord",
+      to: "channel:integration-test",
+      accountId: "acct-integration",
+    } as const;
+
+    const announceResult = await deliverSubagentAnnouncement({
+      requesterSessionKey: parentSessionKey,
+      targetRequesterSessionKey: parentSessionKey,
+      triggerMessage: triggerText,
+      steerMessage: triggerText,
+      requesterOrigin: announceOrigin,
+      requesterSessionOrigin: announceOrigin,
+      completionDirectOrigin: announceOrigin,
+      directOrigin: announceOrigin,
+      requesterIsSubagent: false,
+      expectsCompletionMessage: true,
+      bestEffortDeliver: true,
+      directIdempotencyKey: "integration-durable-queue",
+      sourceTool: "sessions_spawn",
+    });
+
+    // Stage 3: announce was redirected to durable_queue (not failed).
+    expect(announceResult.delivered).toBe(true);
+    expect(announceResult.path).toBe("durable_queue");
+    expect(announceResult.error).toBeUndefined();
+
+    // The trigger message is in the parent's durable inbox before drain.
+    const queuedBeforeDrain = peekSystemEvents(parentSessionKey);
+    expect(queuedBeforeDrain).toEqual([triggerText]);
+
+    // Stage 4: simulate next-turn drain. drainSystemEvents is the
+    // primitive that the auto-reply pipeline calls when building the
+    // next prompt for an idle parent.
+    const drained = drainSystemEvents(parentSessionKey);
+    expect(drained).toEqual([triggerText]);
+
+    // Inbox is empty after drain; the parent's next prompt would carry
+    // the child completion as a System: line.
+    expect(peekSystemEvents(parentSessionKey)).toEqual([]);
+
+    // Cleanup: leaving stubbed deps in place would leak into any
+    // subsequent suite that imports subagent-announce-delivery.
+    subagentAnnounceTesting.setDepsForTest();
+    resetSystemEventsForTest();
+  }, 20_000);
 });
diff --git a/src/agents/subagent-announce-delivery.test.ts b/src/agents/subagent-announce-delivery.test.ts
@@ -1,9 +1,14 @@
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { OutboundDeliveryError } from "../infra/outbound/deliver-types.js";
 import {
   testing as sessionBindingServiceTesting,
   registerSessionBindingAdapter,
 } from "../infra/outbound/session-binding-service.js";
+import {
+  drainSystemEvents,
+  peekSystemEvents,
+  resetSystemEventsForTest,
+} from "../infra/system-events.js";
 import { setActivePluginRegistry } from "../plugins/runtime.js";
 import { createChannelTestPluginBase, createTestRegistry } from "../test-utils/channel-plugins.js";
 import type {
@@ -4490,3 +4495,153 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
     });
   });
 });
+
+describe("durable-queue fallback when direct handoff is pending", () => {
+  beforeEach(() => {
+    resetSystemEventsForTest();
+  });
+
+  afterEach(() => {
+    resetSystemEventsForTest();
+  });
+
+  // The canonical requester store key (resolveRequesterStoreKey) leaves keys
+  // that already start with "agent:" untouched. Using a canonical key in the
+  // test avoids the "agent:<inferred>:<raw>" remap and keeps the assertions
+  // clean.
+  const parentSessionKey = "agent:main:session:durable-queue-test-parent";
+
+  function pendingHandoffArgs(
+    overrides: Partial<Parameters<typeof deliverSubagentAnnouncement>[0]> = {},
+  ): Parameters<typeof deliverSubagentAnnouncement>[0] {
+    const callGateway = vi.fn(async () => ({
+      runId: "agent:main:run:announce-pending",
+      status: "accepted" as const,
+      acceptedAt: 4_000,
+    })) as unknown as typeof runtimeCallGateway;
+    testing.setDepsForTest({
+      callGateway,
+      getRequesterSessionActivity: () => ({
+        sessionId: "durable-queue-test-parent-session",
+        isActive: false,
+      }),
+      queueEmbeddedAgentMessageWithOutcome: createQueueOutcomeMock(false),
+    });
+
+    return {
+      requesterSessionKey: parentSessionKey,
+      targetRequesterSessionKey: parentSessionKey,
+      triggerMessage: "child cipher:abc done: APPROVE",
+      steerMessage: "child cipher:abc done: APPROVE",
+      requesterOrigin: slackThreadOrigin,
+      requesterSessionOrigin: slackThreadOrigin,
+      completionDirectOrigin: slackThreadOrigin,
+      directOrigin: slackThreadOrigin,
+      requesterIsSubagent: false,
+      expectsCompletionMessage: true,
+      bestEffortDeliver: true,
+      directIdempotencyKey: "durable-queue-test-idem",
+      sourceTool: "sessions_spawn",
+      ...overrides,
+    } as Parameters<typeof deliverSubagentAnnouncement>[0];
+  }
+
+  it("enqueues into the durable inbox when direct handoff is pending and parent is non-streaming", async () => {
+    const result = await deliverSubagentAnnouncement(pendingHandoffArgs());
+
+    expect(result.delivered).toBe(true);
+    expect(result.path).toBe("durable_queue");
+    expect(result.error).toBeUndefined();
+    expect(peekSystemEvents(parentSessionKey)).toEqual(["child cipher:abc done: APPROVE"]);
+  });
+
+  it("is idempotent on duplicate fallback fires for the same idempotency key", async () => {
+    const args = pendingHandoffArgs({
+      directIdempotencyKey: "durable-queue-test-dup",
+    });
+    const r1 = await deliverSubagentAnnouncement(args);
+    const r2 = await deliverSubagentAnnouncement(args);
+
+    expect(r1.path).toBe("durable_queue");
+    expect(r2.path).toBe("durable_queue");
+    // Inbox dedupes on (text, contextKey, deliveryContext); duplicate fires
+    // collapse to one entry.
+    expect(peekSystemEvents(parentSessionKey)).toEqual(["child cipher:abc done: APPROVE"]);
+  });
+
+  it("does not invoke durable fallback when parent is active-streaming (steered path wins)", async () => {
+    const callGateway = vi.fn(async () => ({
+      runId: "agent:main:run:announce-pending",
+      status: "accepted" as const,
+      acceptedAt: 4_000,
+    })) as unknown as typeof runtimeCallGateway;
+    testing.setDepsForTest({
+      callGateway,
+      getRequesterSessionActivity: () => ({
+        sessionId: "durable-queue-test-parent-session",
+        isActive: true,
+      }),
+      queueEmbeddedAgentMessageWithOutcome: createQueueOutcomeMock(true),
+    });
+
+    const result = await deliverSubagentAnnouncement({
+      requesterSessionKey: parentSessionKey,
+      targetRequesterSessionKey: parentSessionKey,
+      triggerMessage: "child active",
+      steerMessage: "child active",
+      requesterOrigin: slackThreadOrigin,
+      requesterIsSubagent: false,
+      expectsCompletionMessage: true,
+      directIdempotencyKey: "durable-queue-test-active",
+    });
+
+    expect(result.path).toBe("steered");
+    expect(peekSystemEvents(parentSessionKey)).toEqual([]);
+  });
+
+  it("does NOT redirect to durable_queue when expectedMediaUrls is non-empty (existing direct/media path preserved)", async () => {
+    const args = pendingHandoffArgs({
+      directIdempotencyKey: "durable-queue-test-media",
+      internalEvents: [
+        {
+          type: "task_completion",
+          source: "image_generation",
+          childSessionKey: "image_generate:task-test",
+          childSessionId: "task-test",
+          announceType: "image generation task",
+          taskLabel: "test image",
+          status: "ok",
+          statusLabel: "completed successfully",
+          result: "Generated.\nMEDIA:/tmp/generated-test.png",
+          mediaUrls: ["/tmp/generated-test.png"],
+          replyInstruction: "Deliver the generated image.",
+        },
+      ],
+    });
+    const result = await deliverSubagentAnnouncement(args);
+
+    // Either the existing direct/media path returns delivered:true
+    // (downstream of the patched branch) or the existing pending give-up
+    // returns delivered:false; the contract this test enforces is simply
+    // that the durable_queue branch is NOT taken when media is in play.
+    expect(result.path).not.toBe("durable_queue");
+    expect(peekSystemEvents(parentSessionKey)).toEqual([]);
+  });
+
+  it("records the result envelope shape consumers can route on", async () => {
+    const result = await deliverSubagentAnnouncement(
+      pendingHandoffArgs({
+        directIdempotencyKey: "durable-queue-test-shape",
+      }),
+    );
+
+    expect(result).toEqual({
+      delivered: true,
+      path: "durable_queue",
+      phases: expect.any(Array),
+    });
+    // Drain confirms the next-turn prompt would receive the trigger.
+    expect(drainSystemEvents(parentSessionKey)).toEqual(["child cipher:abc done: APPROVE"]);
+    expect(peekSystemEvents(parentSessionKey)).toEqual([]);
+  });
+});
diff --git a/src/agents/subagent-announce-delivery.ts b/src/agents/subagent-announce-delivery.ts
@@ -13,6 +13,7 @@ import type { OpenClawConfig } from "../config/types.openclaw.js";
 import { isOutboundDeliveryError } from "../infra/outbound/deliver-types.js";
 import type { ConversationRef } from "../infra/outbound/session-binding-service.js";
 import { sourceDeliveryTargetsMatch } from "../infra/outbound/source-delivery-plan.js";
+import { enqueueSystemEvent } from "../infra/system-events.js";
 import { stringifyRouteThreadId } from "../plugin-sdk/channel-route.js";
 import { normalizeAccountId } from "../routing/session-key.js";
 import { defaultRuntime } from "../runtime.js";
@@ -1474,12 +1475,46 @@ async function sendSubagentAnnounceDirectly(params: {
         expectedMediaUrls.length === 0 &&
         !requiresMessageToolDelivery
       ) {
-        return {
-          delivered: false,
-          path: "direct",
-          reason: "completion_handoff_pending",
-          error: "completion agent handoff is still pending",
-        };
+        // The gateway accepted the agent run but it has not yet reached a
+        // streaming/terminal state. This commonly happens when the
+        // requester parent is registered in ACTIVE_EMBEDDED_RUNS but its
+        // handle.isStreaming() is false (yielded, queued behind work,
+        // compaction-prep, or tool-result truncation). Returning
+        // {delivered:false} here is what produces today's `delivery_failed`
+        // audit findings: the requester is stuck waiting for an event that
+        // already arrived in the gateway but cannot deliver back.
+        //
+        // Instead, enqueue the trigger message into the requester's
+        // in-process durable system-event inbox keyed by the announce's
+        // direct idempotency key (which uniquely identifies this child
+        // announce). The inbox is drained on the requester's next
+        // turn-start (the same path used by jobs, hooks, restart
+        // sentinels, ACP child spawns, monitor events, and the
+        // group/channel completion path in task-registry). Idempotent on
+        // (text, contextKey, deliveryContext); MAX_EVENTS=20 per session.
+        try {
+          const durableContextKey = `subagent-announce:${params.directIdempotencyKey}`;
+          const durableDeliveryContext =
+            requesterSessionOrigin ?? completionExternalFallbackOrigin ?? directOrigin;
+          enqueueSystemEvent(params.triggerMessage, {
+            sessionKey: canonicalRequesterSessionKey,
+            contextKey: durableContextKey,
+            deliveryContext: durableDeliveryContext,
+          });
+          return {
+            delivered: true,
+            path: "durable_queue",
+          };
+        } catch {
+          // Fall through to the original give-up shape so existing
+          // retry/give-up bookkeeping stays as the safety net.
+          return {
+            delivered: false,
+            path: "direct",
+            reason: "completion_handoff_pending",
+            error: "completion agent handoff is still pending",
+          };
+        }
       }
       return {
         delivered: true,
diff --git a/src/agents/subagent-announce-dispatch.ts b/src/agents/subagent-announce-dispatch.ts
@@ -1,4 +1,17 @@
-type SubagentDeliveryPath = "steered" | "direct" | "none";
+type SubagentDeliveryPath =
+  | "steered"
+  | "direct"
+  | "none"
+  /**
+   * The trigger message was committed to the requester's in-process
+   * durable system-event inbox because the direct agent-call dispatch
+   * returned a non-terminal pending status (e.g., gateway accepted but
+   * parent is yielded/queued/compaction-prep). The parent drains the
+   * inbox at next turn-start. Treated as a delivered path; downstream
+   * consumers that previously short-circuited on `direct`/`steered`
+   * should treat `durable_queue` as durable as well.
+   */
+  | "durable_queue";
 export type SubagentAnnounceDeliveryFailureReason =
   | "completion_handoff_pending"
   | "generated_media_missing"
diff --git a/src/plugin-sdk/agent-harness-task-runtime.test.ts b/src/plugin-sdk/agent-harness-task-runtime.test.ts
diff --git a/src/plugin-sdk/agent-harness-task-runtime.ts b/src/plugin-sdk/agent-harness-task-runtime.ts
