fix(cron): preserve yielded media completions (#92146)

* fix(agents): preserve sessions_yield media pauses

* fix(agents): resume cron media completions

* fix(agents): restore cron media wait predicate import

---------

Co-authored-by: WhatsSkiLL <284122573+IWhatsskill@users.noreply.github.com>

src/agents/embedded-agent-runner/run/attempt.async-tasks.test.ts

@@ -7,6 +7,7 @@ import {
 import { resetTaskRegistryForTests, type TaskRecord } from "../../../tasks/runtime-internal.js";
 import {
   requiresCompletionRequiredAsyncTaskWait,
+  shouldWaitForCompletionRequiredAsyncTasks,
   waitForCompletionRequiredAsyncTasks,
   type AsyncStartedToolMeta,
 } from "./attempt.async-tasks.js";
@@ -97,6 +98,46 @@ describe("waitForCompletionRequiredAsyncTasks", () => {
     ).toBe(true);
   });
 
+  it("skips media task waiting after sessions_yield pauses the attempt", () => {
+    resetTaskRegistryForTests();
+    const sessionKey = "agent:main:cron:daily-media:run:run-123";
+    createRunningTaskRun({
+      runtime: "cli",
+      taskKind: "image_generation",
+      sourceId: "image_generate:openai",
+      requesterSessionKey: sessionKey,
+      ownerKey: sessionKey,
+      scopeKind: "session",
+      runId: "tool:image_generate:run-123",
+      task: "daily image",
+      deliveryStatus: "not_applicable",
+      notifyPolicy: "silent",
+      startedAt: 1,
+      lastEventAt: 1,
+    });
+
+    expect(
+      shouldWaitForCompletionRequiredAsyncTasks({
+        sessionKey,
+        toolMetas: [
+          {
+            toolName: "image_generate",
+            asyncStarted: true,
+            asyncTaskRunId: "tool:image_generate:run-123",
+          },
+        ],
+        yieldDetected: true,
+      }),
+    ).toBe(false);
+    expect(
+      shouldWaitForCompletionRequiredAsyncTasks({
+        sessionKey,
+        toolMetas: [],
+        yieldDetected: false,
+      }),
+    ).toBe(true);
+  });
+
   it("waits for active cron media tasks from the task registry", async () => {
     // Cron media tools may start tasks before metadata is flushed, so the
     // registry is also consulted by session key.

src/agents/embedded-agent-runner/run/attempt.async-tasks.ts

@@ -160,6 +160,23 @@ export function requiresCompletionRequiredAsyncTaskWait(params: {
   );
 }
 
+/** Returns whether the current attempt should synchronously wait for media tasks. */
+export function shouldWaitForCompletionRequiredAsyncTasks(params: {
+  sessionKey: string | undefined;
+  toolMetas: readonly AsyncStartedToolMeta[];
+  yieldDetected?: boolean;
+}): boolean {
+  if (params.yieldDetected === true) {
+    // sessions_yield pauses the turn so the completion event can wake it later;
+    // waiting here would reuse the internal abort signal and turn the pause into AbortError.
+    return false;
+  }
+  return requiresCompletionRequiredAsyncTaskWait({
+    sessionKey: params.sessionKey,
+    toolMetas: params.toolMetas,
+  });
+}
+
 /**
  * Polls completion-required async tasks until they reach terminal state, time
  * out at the run deadline, or abort. Newly discovered task run ids are folded

src/agents/embedded-agent-runner/run/attempt.ts

@@ -316,6 +316,7 @@ import {
 } from "./attempt-trajectory-status.js";
 import {
   requiresCompletionRequiredAsyncTaskWait,
+  shouldWaitForCompletionRequiredAsyncTasks,
   waitForCompletionRequiredAsyncTasks,
   type AsyncStartedToolMeta,
   type CompletionRequiredAsyncTaskWaitResult,
@@ -4571,9 +4572,10 @@ export async function runEmbeddedAttempt(
         await sessionLockController.releaseForPrompt();
 
         if (
-          requiresCompletionRequiredAsyncTaskWait({
+          shouldWaitForCompletionRequiredAsyncTasks({
             sessionKey: params.sessionKey,
             toolMetas,
+            yieldDetected: yieldAborted,
           })
         ) {
           const getAsyncStartedToolMetas = () =>

src/agents/subagent-announce-delivery.test.ts

@@ -345,6 +345,8 @@ async function deliverSlackChannelAnnouncement(params: {
   queueEmbeddedAgentMessageWithOutcome?: QueueEmbeddedAgentMessageWithOutcome;
   sendMessage?: typeof runtimeSendMessage;
   internalEvents?: AgentInternalEvent[];
+  sourceSessionKey?: string;
+  sourceChannel?: string;
   sourceTool?: string;
   runtimeConfig?: Record<string, unknown>;
 }) {
@@ -381,6 +383,8 @@ async function deliverSlackChannelAnnouncement(params: {
     bestEffortDeliver: true,
     directIdempotencyKey: params.directIdempotencyKey,
     internalEvents: params.internalEvents,
+    sourceSessionKey: params.sourceSessionKey,
+    sourceChannel: params.sourceChannel,
     sourceTool: params.sourceTool,
   });
 }
@@ -4015,8 +4019,21 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
     expect(sendMessage).not.toHaveBeenCalled();
   });
 
-  it("directly delivers stale isolated cron run media completions", async () => {
-    const callGateway = createGatewayMock();
+  it("runs inactive isolated cron media completions through the requester agent first", async () => {
+    const callGateway = createGatewayMock({
+      result: {
+        payloads: [{ text: "queued the generated image confirmation" }],
+        messagingToolSentTargets: [
+          {
+            tool: "sessions_send",
+            provider: "slack",
+            to: "channel:C123",
+            text: "The daily media workflow continued after the image callback.",
+            mediaUrls: ["/tmp/generated-daily.png"],
+          },
+        ],
+      },
+    });
     const sendMessage = createSendMessageMock();
     const queueEmbeddedAgentMessageWithOutcome = createQueueOutcomeMock(true);
     const result = await deliverSlackChannelAnnouncement({
@@ -4044,6 +4061,8 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
           replyInstruction: "Deliver the generated image through the requester run.",
         },
       ],
+      sourceSessionKey: "image_generate:task-123",
+      sourceChannel: "internal",
     });
 
     expectRecordFields(result, {
@@ -4051,7 +4070,71 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
       path: "direct",
     });
     expect(queueEmbeddedAgentMessageWithOutcome).not.toHaveBeenCalled();
-    expect(callGateway).not.toHaveBeenCalled();
+    expect(callGateway).toHaveBeenCalledTimes(1);
+    const params = expectGatewayAgentParams(callGateway, {
+      sessionKey: "agent:main:cron:daily-media:run:run-123",
+      deliver: true,
+      channel: "slack",
+      accountId: "acct-1",
+      to: "channel:C123",
+      idempotencyKey: "announce-stale-cron-media",
+    });
+    expectRecordFields(params.inputProvenance, {
+      kind: "inter_session",
+      sourceSessionKey: "image_generate:task-123",
+      sourceChannel: "internal",
+      sourceTool: "image_generate",
+    });
+    expect(sendMessage).not.toHaveBeenCalled();
+  });
+
+  it("directly delivers inactive isolated cron media only after requester-agent fallback misses media", async () => {
+    const callGateway = createGatewayMock();
+    const sendMessage = createSendMessageMock();
+    const queueEmbeddedAgentMessageWithOutcome = createQueueOutcomeMock(true);
+    const result = await deliverSlackChannelAnnouncement({
+      callGateway,
+      sendMessage,
+      queueEmbeddedAgentMessageWithOutcome,
+      sessionId: "stale-cron-run-session",
+      isActive: false,
+      requesterSessionKey: "agent:main:cron:daily-media:run:run-123",
+      expectsCompletionMessage: true,
+      directIdempotencyKey: "announce-stale-cron-media-fallback",
+      sourceTool: "image_generate",
+      internalEvents: [
+        {
+          type: "task_completion",
+          source: "image_generation",
+          childSessionKey: "image_generate:task-123",
+          childSessionId: "task-123",
+          announceType: "image generation task",
+          taskLabel: "daily media",
+          status: "ok",
+          statusLabel: "completed successfully",
+          result: "Generated 1 image.\nMEDIA:/tmp/generated-daily.png",
+          mediaUrls: ["/tmp/generated-daily.png"],
+          replyInstruction: "Deliver the generated image through the requester run.",
+        },
+      ],
+      sourceSessionKey: "image_generate:task-123",
+      sourceChannel: "internal",
+    });
+
+    expectRecordFields(result, {
+      delivered: true,
+      path: "direct",
+    });
+    expect(queueEmbeddedAgentMessageWithOutcome).not.toHaveBeenCalled();
+    expect(callGateway).toHaveBeenCalledTimes(1);
+    expectGatewayAgentParams(callGateway, {
+      sessionKey: "agent:main:cron:daily-media:run:run-123",
+      deliver: true,
+      channel: "slack",
+      accountId: "acct-1",
+      to: "channel:C123",
+      idempotencyKey: "announce-stale-cron-media-fallback",
+    });
     expect(sendMessage).toHaveBeenCalledWith(
       expect.objectContaining({
         channel: "slack",
@@ -4059,7 +4142,7 @@ describe("deliverSubagentAnnouncement completion delivery", () => {
         to: "channel:C123",
         content: "The generated image is ready.",
         mediaUrls: ["/tmp/generated-daily.png"],
-        idempotencyKey: "announce-stale-cron-media:generated-media-direct",
+        idempotencyKey: "announce-stale-cron-media-fallback:generated-media-direct",
       }),
     );
   });

src/agents/subagent-announce-delivery.ts

@@ -1387,7 +1387,8 @@ async function sendSubagentAnnounceDirectly(params: {
     if (
       params.expectsCompletionMessage &&
       isCronRunSessionKey(canonicalRequesterSessionKey) &&
-      !resolveRequesterSessionActivity(canonicalRequesterSessionKey).isActive
+      !resolveRequesterSessionActivity(canonicalRequesterSessionKey).isActive &&
+      !agentMediatedCompletion
     ) {
       const generatedMediaDelivery = await tryGeneratedMediaDirectDelivery();
       if (generatedMediaDelivery) {