diff --git a/extensions/memory-core/src/dreaming.test.ts b/extensions/memory-core/src/dreaming.test.ts index 54cfaf11dcb6..fd61accf4a76 100644 --- a/extensions/memory-core/src/dreaming.test.ts +++ b/extensions/memory-core/src/dreaming.test.ts @@ -937,6 +937,70 @@ describe("gateway startup reconciliation", () => { } }); + it("reconciles disabled->enabled config changes without waiting for another agent turn", async () => { + vi.useFakeTimers(); + clearInternalHooks(); + const logger = createLogger(); + const harness = createCronHarness(); + const onMock = vi.fn(); + const api: DreamingPluginApiTestDouble = { + config: { + plugins: { + entries: { + "memory-core": { + config: { + dreaming: { + enabled: false, + frequency: "0 2 * * *", + timezone: "UTC", + }, + }, + }, + }, + }, + }, + pluginConfig: {}, + logger, + runtime: {}, + on: onMock, + }; + + try { + registerShortTermPromotionDreamingForTest(api); + await triggerGatewayStart(onMock, { + config: api.config, + getCron: () => harness.cron, + }); + + expect(harness.addCalls).toHaveLength(0); + + api.config = { + plugins: { + entries: { + "memory-core": { + config: { + dreaming: { + enabled: true, + frequency: "30 6 * * *", + timezone: "America/New_York", + }, + }, + }, + }, + }, + } as OpenClawConfig; + + await vi.advanceTimersByTimeAsync(constants.RUNTIME_CRON_RECONCILE_INTERVAL_MS); + + expect(harness.addCalls).toHaveLength(1); + expectCronSchedule(requireAddCall(harness, 0).schedule, "30 6 * * *", "America/New_York"); + } finally { + await triggerGatewayStop(onMock).catch(() => undefined); + vi.useRealTimers(); + clearInternalHooks(); + } + }); + it("reconciles cadence/timezone updates against the active cron service after startup", async () => { clearInternalHooks(); const logger = createLogger(); diff --git a/extensions/memory-core/src/dreaming.ts b/extensions/memory-core/src/dreaming.ts index a6eb6114e599..3a3ddee6e7df 100644 --- a/extensions/memory-core/src/dreaming.ts +++ b/extensions/memory-core/src/dreaming.ts @@ -695,6 +695,7 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void let lastRuntimeConfigKey: string | null = null; let lastRuntimeCronRef: CronServiceLike | null = null; let startupCronRetryTimer: ReturnType | null = null; + let runtimeCronReconcileTimer: ReturnType | null = null; let startupCronRetryAttempts = 0; let disposed = false; @@ -723,6 +724,10 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void const disposeStartupCronRetry = (): void => { disposed = true; clearStartupCronRetry(); + if (runtimeCronReconcileTimer) { + clearInterval(runtimeCronReconcileTimer); + runtimeCronReconcileTimer = null; + } gatewayContext = null; resolveStartupCron = null; }; @@ -856,6 +861,18 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void }, STARTUP_CRON_RETRY_DELAY_MS); }; + const startRuntimeCronReconcileTimer = (): void => { + if (runtimeCronReconcileTimer) { + return; + } + runtimeCronReconcileTimer = setInterval(() => { + void reconcileManagedDreamingCron({ reason: "runtime" }).catch((err) => { + api.logger.error(`memory-core: dreaming cron reconcile failed: ${formatErrorMessage(err)}`); + }); + }, RUNTIME_CRON_RECONCILE_INTERVAL_MS); + runtimeCronReconcileTimer.unref?.(); + }; + api.on("gateway_start", async (_event, ctx) => { disposed = false; // Store the gateway context for runtime cron resolution retries. @@ -866,6 +883,7 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void startupConfig: ctx.config, startupCron: () => resolveCronServiceFromGatewayContext(ctx), }); + startRuntimeCronReconcileTimer(); scheduleStartupCronRetry(); } catch (err) { api.logger.error( @@ -933,6 +951,7 @@ export const testing = { DEFAULT_DREAMING_MIN_RECALL_COUNT: DEFAULT_MEMORY_DREAMING_MIN_RECALL_COUNT, DEFAULT_DREAMING_MIN_UNIQUE_QUERIES: DEFAULT_MEMORY_DREAMING_MIN_UNIQUE_QUERIES, DEFAULT_DREAMING_RECENCY_HALF_LIFE_DAYS: DEFAULT_MEMORY_DREAMING_RECENCY_HALF_LIFE_DAYS, + RUNTIME_CRON_RECONCILE_INTERVAL_MS, STARTUP_CRON_RETRY_DELAY_MS, STARTUP_CRON_RETRY_MAX_ATTEMPTS, }, diff --git a/extensions/qa-lab/src/providers/mock-openai/server.test.ts b/extensions/qa-lab/src/providers/mock-openai/server.test.ts index 911ff86c830c..de08b44818fb 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.test.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.test.ts @@ -3034,6 +3034,101 @@ describe("qa mock openai server", () => { expect(requireRecord(await debug.json(), "debug request").imageInputCount).toBe(1); }); + it("answers image prompts when media context is the latest text part", async () => { + const server = await startQaMockOpenAiServer({ + host: "127.0.0.1", + port: 0, + }); + cleanups.push(async () => { + await server.stop(); + }); + + const response = await fetch(`${server.baseUrl}/v1/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + stream: false, + model: "mock-openai/gpt-5.5", + input: [ + { + role: "user", + content: [ + { type: "input_text", text: "Image understanding check: what do you see?" }, + { + type: "input_image", + source: { + type: "base64", + mime_type: "image/png", + data: QA_IMAGE_PNG_BASE64, + }, + }, + { + type: "input_text", + text: "[media attached: media://inbound/red-top-blue-bottom.png (image/png)]", + }, + ], + }, + ], + }), + }); + expect(response.status).toBe(200); + const payload = (await response.json()) as { + output?: Array<{ content?: Array<{ text?: string }> }>; + }; + const text = payload.output?.[0]?.content?.[0]?.text ?? ""; + expect(text.toLowerCase()).toContain("red"); + expect(text.toLowerCase()).toContain("blue"); + }); + + it("lets image prompts beat stale exact marker directives from chat history", async () => { + const server = await startMockServer(); + + const response = await fetch(`${server.baseUrl}/v1/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + stream: false, + model: "mock-openai/gpt-5.5", + input: [ + makeUserInput("Control UI bridge check. Marker exact marker: `ui bridge armed`"), + { + role: "assistant", + content: [{ type: "output_text", text: "ui bridge armed" }], + }, + { + role: "user", + content: [ + { + type: "input_text", + text: "Image understanding check: describe the top and bottom colors.", + }, + { + type: "input_image", + source: { + type: "base64", + mime_type: "image/png", + data: QA_IMAGE_PNG_BASE64, + }, + }, + { + type: "input_text", + text: "[media attached: media://inbound/red-top-blue-bottom.png (image/png)]", + }, + ], + }, + ], + }), + }); + expect(response.status).toBe(200); + const payload = (await response.json()) as { + output?: Array<{ content?: Array<{ text?: string }> }>; + }; + const text = payload.output?.[0]?.content?.[0]?.text ?? ""; + expect(text.toLowerCase()).toContain("red"); + expect(text.toLowerCase()).toContain("blue"); + expect(text).not.toBe("ui bridge armed"); + }); + it("handles deeply nested image input shapes without recursive traversal failure", async () => { const server = await startQaMockOpenAiServer({ host: "127.0.0.1", diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index b1d753ce363e..13625ca9d9ea 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.ts @@ -1036,6 +1036,12 @@ function buildAssistantText( if (isHeartbeatPrompt(prompt)) { return "HEARTBEAT_OK"; } + if (/roundtrip image inspection check/i.test(allInputText) && imageInputCount > 0) { + return "Protocol note: the generated attachment shows the same QA lighthouse scene from the previous step."; + } + if (/image understanding check/i.test(allInputText) && imageInputCount > 0) { + return "Protocol note: the attached image is split horizontally, with red on top and blue on the bottom."; + } if (/\bmarker\b/i.test(allInputText) && exactReplyDirective) { return exactReplyDirective; } @@ -1118,12 +1124,6 @@ function buildAssistantText( "- Re-open the copied file for final verification.", ].join("\n"); } - if (/roundtrip image inspection check/i.test(prompt) && imageInputCount > 0) { - return "Protocol note: the generated attachment shows the same QA lighthouse scene from the previous step."; - } - if (/image understanding check/i.test(prompt) && imageInputCount > 0) { - return "Protocol note: the attached image is split horizontally, with red on top and blue on the bottom."; - } if ( /interrupted by a gateway reload/i.test(prompt) && /subagent recovery worker/i.test(allInputText) @@ -1565,6 +1565,7 @@ async function buildResponsesPayload( extractExactReplyDirective(prompt) ?? extractExactReplyDirective(allInputText); const exactMarkerDirective = extractExactMarkerDirective(prompt) ?? extractExactMarkerDirective(allInputText); + const imageInputCount = countImageInputs(input); const firstExactMarkerDirective = extractLabeledMarkerDirective( allInputText, "first exact marker", @@ -1651,6 +1652,16 @@ async function buildResponsesPayload( if (/fanout worker beta/i.test(prompt)) { return buildAssistantEvents("BETA-OK"); } + if (/roundtrip image inspection check/i.test(allInputText) && imageInputCount > 0) { + return buildAssistantEvents( + "Protocol note: the generated attachment shows the same QA lighthouse scene from the previous step.", + ); + } + if (/image understanding check/i.test(allInputText) && imageInputCount > 0) { + return buildAssistantEvents( + "Protocol note: the attached image is split horizontally, with red on top and blue on the bottom.", + ); + } if (QA_REASONING_ONLY_RECOVERY_PROMPT_RE.test(allInputText)) { if (!scenarioToolOutput) { return buildToolCallEventsWithArgs("read", { path: "QA_KICKOFF_TASK.md" }); diff --git a/qa/scenarios/agents/subagent-completion-direct-fallback.md b/qa/scenarios/agents/subagent-completion-direct-fallback.md index 4b62d365a184..5c30bded1d9a 100644 --- a/qa/scenarios/agents/subagent-completion-direct-fallback.md +++ b/qa/scenarios/agents/subagent-completion-direct-fallback.md @@ -50,26 +50,38 @@ steps: - set: sessionKey value: expr: "`agent:qa:subagent-direct-fallback:${randomUUID().slice(0, 8)}`" - - call: runAgentPrompt - args: - - ref: env - - sessionKey: - ref: sessionKey - message: - expr: config.prompt - timeoutMs: - expr: liveTurnTimeoutMs(env, 90000) - - call: waitForCondition - saveAs: outbound - args: - - lambda: - expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound' && String(message.text ?? '').includes(config.expectedMarker)).at(-1)" - - expr: liveTurnTimeoutMs(env, 60000) - - expr: "env.providerMode === 'mock-openai' ? 100 : 250" - - assert: - expr: "String(outbound.text ?? '').trim().includes(config.expectedMarker)" - message: - expr: "`fallback completion marker missing from outbound QA DM: ${recentOutboundSummary(state)}`" + - try: + actions: + - call: runAgentPrompt + args: + - ref: env + - sessionKey: + ref: sessionKey + message: + expr: config.prompt + timeoutMs: + expr: liveTurnTimeoutMs(env, 90000) + - call: waitForCondition + saveAs: outbound + args: + - lambda: + expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound' && String(message.text ?? '').includes(config.expectedMarker)).at(-1)" + - expr: liveTurnTimeoutMs(env, 180000) + - expr: "env.providerMode === 'mock-openai' ? 100 : 250" + - assert: + expr: "String(outbound.text ?? '').trim().includes(config.expectedMarker)" + message: + expr: "`fallback completion marker missing from outbound QA DM: ${recentOutboundSummary(state)}`" + catchAs: fallbackError + catch: + - set: fallbackDebugRequests + value: + expr: "env.mock ? [...(await fetchJson(`${env.mock.baseUrl}/debug/requests`))].slice(-20).map((request) => ({ plannedToolName: request.plannedToolName ?? null, plannedToolArgs: request.plannedToolArgs ?? null, prompt: String(request.prompt ?? '').slice(0, 280), allInputText: String(request.allInputText ?? '').slice(0, 280), toolOutput: request.toolOutput ? String(request.toolOutput).slice(0, 280) : null })) : []" + - set: fallbackTasks + value: + expr: "(await runQaCli(env, ['tasks', 'list', '--json', '--runtime', 'subagent'], { timeoutMs: liveTurnTimeoutMs(env, 60000), json: true }).catch((error) => ({ error: String(error?.message ?? error) })))" + - throw: + expr: "`subagent fallback marker missing: ${fallbackError?.message ?? fallbackError}; outbound=${recentOutboundSummary(state, 8)} tasks=${JSON.stringify(fallbackTasks)} requests=${JSON.stringify(fallbackDebugRequests)}`" - if: expr: "Boolean(env.mock)" then: @@ -89,7 +101,7 @@ steps: args: - lambda: expr: "(async () => { const payload = await runQaCli(env, ['tasks', 'list', '--json', '--runtime', 'subagent'], { timeoutMs: liveTurnTimeoutMs(env, 60000), json: true }); return (payload.tasks ?? []).find((task) => task.label === config.expectedLabel && task.deliveryStatus === 'delivered' && task.status === 'succeeded') ?? null; })()" - - expr: liveTurnTimeoutMs(env, 30000) + - expr: liveTurnTimeoutMs(env, 60000) - 250 - assert: expr: "deliveredTask.deliveryStatus === 'delivered'" diff --git a/qa/scenarios/media/image-generation-roundtrip.md b/qa/scenarios/media/image-generation-roundtrip.md index a3ba5ba6a04f..238591da122f 100644 --- a/qa/scenarios/media/image-generation-roundtrip.md +++ b/qa/scenarios/media/image-generation-roundtrip.md @@ -94,7 +94,8 @@ steps: expr: "!env.mock || Boolean((await fetchJson(`${env.mock.baseUrl}/debug/requests`)).find((request) => request.plannedToolName === 'image_generate' && String(request.prompt ?? '').includes(config.generatePromptSnippet)))" message: expected image_generate call before roundtrip inspection - assert: - expr: "!env.mock || (((await fetchJson(`${env.mock.baseUrl}/debug/requests`)).find((request) => String(request.prompt ?? '').includes(config.inspectPrompt))?.imageInputCount ?? 0) >= 1)" - message: expected generated artifact to be reattached on follow-up turn + expr: "!env.mock || (await fetchJson(`${env.mock.baseUrl}/debug/requests`)).some((request) => String(request.prompt ?? '').includes(config.inspectPrompt) && (request.imageInputCount ?? 0) >= 1)" + message: + expr: "`expected generated artifact to be reattached on follow-up turn; recentRequests=${JSON.stringify((await fetchJson(`${env.mock.baseUrl}/debug/requests`)).slice(-12).map((request) => ({ prompt: String(request.prompt ?? '').slice(0, 240), imageInputCount: request.imageInputCount, allInputText: String(request.allInputText ?? '').slice(0, 240) })))}`" detailsExpr: "`MEDIA:${mediaPath}\\n${outbound.text}`" ``` diff --git a/qa/scenarios/memory/memory-dreaming-sweep.md b/qa/scenarios/memory/memory-dreaming-sweep.md index c6bb27addc90..8aa282299494 100644 --- a/qa/scenarios/memory/memory-dreaming-sweep.md +++ b/qa/scenarios/memory/memory-dreaming-sweep.md @@ -88,7 +88,7 @@ steps: - lambda: async: true expr: "(() => readDoctorMemoryStatus(env).then((payload) => payload.dreaming?.phases?.deep?.managedCronPresent === true ? payload : undefined))()" - - 30000 + - expr: liveTurnTimeoutMs(env, 90000) - 500 - call: listCronJobs saveAs: jobs @@ -108,6 +108,12 @@ steps: expr: "managed.id" catchAs: enableError catch: + - set: enableFailureStatus + value: + expr: "(await readDoctorMemoryStatus(env).catch((error) => ({ error: String(error?.message ?? error) })))" + - set: enableFailureJobs + value: + expr: "(await listCronJobs(env).catch((error) => [{ error: String(error?.message ?? error) }]))" - call: patchConfig args: - env: @@ -127,7 +133,7 @@ steps: - ref: env - 60000 - throw: - expr: enableError + expr: "`managed dreaming cron missing: ${enableError?.message ?? enableError}; status=${JSON.stringify(enableFailureStatus)} jobs=${JSON.stringify(enableFailureJobs)}`" detailsExpr: "JSON.stringify({ enabled: status.dreaming?.enabled ?? false, managedCronPresent: status.dreaming?.phases?.deep?.managedCronPresent ?? false, nextRunAtMs: status.dreaming?.phases?.deep?.nextRunAtMs ?? null })" - name: runs the sweep after repeated recall signals and writes promotion artifacts diff --git a/qa/scenarios/ui/control-ui-qa-channel-image-roundtrip.md b/qa/scenarios/ui/control-ui-qa-channel-image-roundtrip.md index a18d6ae3ff88..17b735752d9f 100644 --- a/qa/scenarios/ui/control-ui-qa-channel-image-roundtrip.md +++ b/qa/scenarios/ui/control-ui-qa-channel-image-roundtrip.md @@ -227,16 +227,25 @@ steps: altText: red on top blue on bottom contentBase64: expr: imageUnderstandingValidPngBase64 - - call: waitForOutboundMessage - saveAs: imageOutbound - args: - - ref: state - - lambda: - params: [candidate] - expr: "candidate.conversation.id === config.conversationId && config.requiredColorGroups.every((group) => group.some((color) => normalizeLowercaseStringOrEmpty(candidate.text).includes(color)))" - - expr: liveTurnTimeoutMs(env, 45000) - - sinceIndex: - ref: secondOutboundStartIndex + - try: + actions: + - call: waitForOutboundMessage + saveAs: imageOutbound + args: + - ref: state + - lambda: + params: [candidate] + expr: "candidate.conversation.id === config.conversationId && config.requiredColorGroups.every((group) => group.some((color) => normalizeLowercaseStringOrEmpty(candidate.text).includes(color)))" + - expr: liveTurnTimeoutMs(env, 90000) + - sinceIndex: + ref: secondOutboundStartIndex + catchAs: imageWaitError + catch: + - set: imageDebugRequests + value: + expr: "env.mock ? [...(await fetchJson(`${env.mock.baseUrl}/debug/requests`))].slice(-16).map((request) => ({ plannedToolName: request.plannedToolName ?? null, prompt: String(request.prompt ?? '').slice(0, 260), allInputText: String(request.allInputText ?? '').slice(0, 260), imageInputCount: request.imageInputCount ?? null })) : []" + - throw: + expr: "`qa-channel image reply missing: ${imageWaitError?.message ?? imageWaitError}; outbound=${recentOutboundSummary(state, 8)} requests=${JSON.stringify(imageDebugRequests)}`" - set: missingColorGroup value: expr: "config.requiredColorGroups.find((group) => !group.some((color) => normalizeLowercaseStringOrEmpty(imageOutbound.text).includes(color)))" @@ -272,7 +281,7 @@ steps: - lambda: async: true expr: "await (async () => { const snapshot = await webSnapshot({ pageId: uiImagePageId, maxChars: 12000, timeoutMs: liveTurnTimeoutMs(env, 30000) }); const text = normalizeLowercaseStringOrEmpty(snapshot.text); const hasPrompt = text.includes(config.imagePromptNeedle); const hasColors = config.requiredColorGroups.every((group) => group.some((color) => text.includes(color))); return hasPrompt && hasColors ? snapshot : undefined; })()" - - expr: liveTurnTimeoutMs(env, 45000) + - expr: liveTurnTimeoutMs(env, 90000) - 500 catch: - call: webSnapshot diff --git a/src/agents/subagent-announce-delivery.test.ts b/src/agents/subagent-announce-delivery.test.ts index aae0947e90cd..ce4f025f34d4 100644 --- a/src/agents/subagent-announce-delivery.test.ts +++ b/src/agents/subagent-announce-delivery.test.ts @@ -937,6 +937,90 @@ describe("deliverSubagentAnnouncement completion delivery", () => { expect(queueEmbeddedPiMessageWithOutcome).not.toHaveBeenCalled(); }); + it("directly delivers direct-message subagent text when the announce agent returns no visible output", async () => { + const callGateway = createGatewayMock({ + result: { + payloads: [], + }, + }); + const sendMessage = createSendMessageMock(); + + const result = await deliverDiscordDirectMessageCompletion({ + callGateway, + sendMessage, + internalEvents: [ + { + type: "task_completion", + source: "subagent", + childSessionKey: "agent:worker:subagent:child", + childSessionId: "child-session-id", + announceType: "subagent task", + taskLabel: "direct completion smoke", + status: "ok", + statusLabel: "completed successfully", + result: "child completion output", + replyInstruction: "Summarize the result.", + }, + ], + }); + + expectRecordFields(result, { + delivered: true, + path: "direct", + }); + expect(sendMessage).toHaveBeenCalledWith( + expect.objectContaining({ + channel: "discord", + accountId: "acct-1", + to: "dm:U123", + content: "child completion output", + idempotencyKey: "announce-dm-fallback-empty:text-direct", + }), + ); + }); + + it("directly delivers direct-message subagent text when the announce agent returns incomplete", async () => { + const callGateway = vi.fn(async () => { + throw new Error( + "FailoverError: mock-openai/gpt-5.5 ended with an incomplete terminal response: code=incomplete_result", + ); + }) as unknown as typeof runtimeCallGateway; + const sendMessage = createSendMessageMock(); + + const result = await deliverDiscordDirectMessageCompletion({ + callGateway, + sendMessage, + internalEvents: [ + { + type: "task_completion", + source: "subagent", + childSessionKey: "agent:worker:subagent:child", + childSessionId: "child-session-id", + announceType: "subagent task", + taskLabel: "direct completion smoke", + status: "ok", + statusLabel: "completed successfully", + result: "child completion output", + replyInstruction: "Summarize the result.", + }, + ], + }); + + expectRecordFields(result, { + delivered: true, + path: "direct", + }); + expect(sendMessage).toHaveBeenCalledWith( + expect.objectContaining({ + channel: "discord", + accountId: "acct-1", + to: "dm:U123", + content: "child completion output", + idempotencyKey: "announce-dm-fallback-empty:text-direct", + }), + ); + }); + it("uses in-process agent dispatch for dormant completion requesters", async () => { const callGateway = createGatewayMock(); const dispatchGatewayMethodInProcess = createInProcessGatewayMock({ diff --git a/src/agents/subagent-announce-delivery.ts b/src/agents/subagent-announce-delivery.ts index a229660a01ab..1fb47f5a5a07 100644 --- a/src/agents/subagent-announce-delivery.ts +++ b/src/agents/subagent-announce-delivery.ts @@ -282,6 +282,11 @@ function isPermanentAnnounceDeliveryError(error: unknown): boolean { ); } +function isIncompleteAnnounceAgentResultError(error: unknown): boolean { + const message = summarizeDeliveryError(error); + return /(?:incomplete terminal response|code=incomplete_result)\b/i.test(message); +} + async function waitForAnnounceRetryDelay(ms: number, signal?: AbortSignal): Promise { if (ms <= 0) { return; @@ -709,6 +714,89 @@ async function deliverGeneratedMediaCompletionDirect(params: { } } +function isDirectMessageDeliveryTarget(target: { to?: string; threadId?: string }): boolean { + if (target.threadId) { + return false; + } + const normalizedTo = normalizeOptionalLowercaseString(target.to); + return Boolean( + normalizedTo && + (normalizedTo.startsWith("dm:") || + normalizedTo.startsWith("direct:") || + normalizedTo.includes(":dm:") || + normalizedTo.includes(":direct:")), + ); +} + +function resolveTextCompletionDirectFallback(events: readonly AgentInternalEvent[] | undefined) { + for (let index = (events?.length ?? 0) - 1; index >= 0; index -= 1) { + const event = events?.[index]; + if (event?.type !== "task_completion" || event.source !== "subagent") { + continue; + } + const result = typeof event.result === "string" ? event.result.trim() : ""; + if (result) { + return result; + } + } + return undefined; +} + +async function deliverTextCompletionDirect(params: { + cfg: OpenClawConfig; + requesterSessionKey: string; + directIdempotencyKey: string; + deliveryTarget: { + deliver: boolean; + channel?: string; + to?: string; + accountId?: string; + threadId?: string; + }; + internalEvents?: readonly AgentInternalEvent[]; +}): Promise { + const content = resolveTextCompletionDirectFallback(params.internalEvents); + if ( + !content || + !params.deliveryTarget.deliver || + !params.deliveryTarget.channel || + !params.deliveryTarget.to || + !isDirectMessageDeliveryTarget(params.deliveryTarget) + ) { + return undefined; + } + const agentId = resolveAgentIdFromSessionKey(params.requesterSessionKey); + const idempotencyKey = `${params.directIdempotencyKey}:text-direct`; + try { + await subagentAnnounceDeliveryDeps.sendMessage({ + cfg: params.cfg, + channel: params.deliveryTarget.channel, + to: params.deliveryTarget.to, + accountId: params.deliveryTarget.accountId, + threadId: params.deliveryTarget.threadId, + requesterSessionKey: params.requesterSessionKey, + agentId, + content, + idempotencyKey, + mirror: { + sessionKey: params.requesterSessionKey, + agentId, + idempotencyKey, + }, + }); + return { + delivered: true, + path: "direct", + }; + } catch (err) { + return { + delivered: false, + path: "direct", + error: `text completion direct delivery failed: ${summarizeDeliveryError(err)}`, + }; + } +} + function resolveGeneratedMediaDirectFallbackUrls(params: { expectedMediaUrls: readonly string[]; announceResponse?: unknown; @@ -979,6 +1067,23 @@ async function sendSubagentAnnounceDirectly(params: { if (isPermanentAnnounceDeliveryError(err)) { throw err; } + if ( + params.expectsCompletionMessage && + shouldDeliverAgentFinal && + isSubagentCompletion && + isIncompleteAnnounceAgentResultError(err) + ) { + const textDelivery = await deliverTextCompletionDirect({ + cfg, + requesterSessionKey: canonicalRequesterSessionKey, + directIdempotencyKey: params.directIdempotencyKey, + deliveryTarget, + internalEvents: params.internalEvents, + }); + if (textDelivery) { + return textDelivery; + } + } // The requester-agent handoff is the delivery contract for background // completions. A failed handoff should retry/fail visibly instead // of sending the child result directly to the external channel. @@ -1034,6 +1139,25 @@ async function sendSubagentAnnounceDirectly(params: { error: directDeliveryFailure, }; } + if ( + params.expectsCompletionMessage && + shouldDeliverAgentFinal && + isSubagentCompletion && + !hasVisibleGatewayAgentPayload(directAnnounceResponse) && + !hasGatewayAgentMessagingToolDeliveryEvidence(directAnnounceResponse) && + !hasIntentionalSilentGatewayAgentPayload(directAnnounceResponse) + ) { + const textDelivery = await deliverTextCompletionDirect({ + cfg, + requesterSessionKey: canonicalRequesterSessionKey, + directIdempotencyKey: params.directIdempotencyKey, + deliveryTarget, + internalEvents: params.internalEvents, + }); + if (textDelivery) { + return textDelivery; + } + } if ( params.expectsCompletionMessage && requiresMessageToolDelivery &&