diff --git a/src/agents/embedded-agent-runner/run/attempt.async-tasks.test.ts b/src/agents/embedded-agent-runner/run/attempt.async-tasks.test.ts index 51f2542e210a..92364a210b43 100644 --- a/src/agents/embedded-agent-runner/run/attempt.async-tasks.test.ts +++ b/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. diff --git a/src/agents/embedded-agent-runner/run/attempt.async-tasks.ts b/src/agents/embedded-agent-runner/run/attempt.async-tasks.ts index e2efe71bce9d..1a0e305f5d16 100644 --- a/src/agents/embedded-agent-runner/run/attempt.async-tasks.ts +++ b/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 diff --git a/src/agents/embedded-agent-runner/run/attempt.ts b/src/agents/embedded-agent-runner/run/attempt.ts index 72b5f112cde8..57380da18703 100644 --- a/src/agents/embedded-agent-runner/run/attempt.ts +++ b/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 = () => diff --git a/src/agents/subagent-announce-delivery.test.ts b/src/agents/subagent-announce-delivery.test.ts index 5be010c0c8d7..b571ad2ece05 100644 --- a/src/agents/subagent-announce-delivery.test.ts +++ b/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; }) { @@ -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", }), ); }); diff --git a/src/agents/subagent-announce-delivery.ts b/src/agents/subagent-announce-delivery.ts index 2e1ddc2158b8..04603f0d8ac3 100644 --- a/src/agents/subagent-announce-delivery.ts +++ b/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) {