/**

* Subagent completion announcement coordinator.

*

* Captures child output, applies wait outcomes, routes announcements, and performs cleanup decisions.

*/

import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";

import {

isSilentReplyText,

SILENT_REPLY_TOKEN,

startsWithSilentToken,

stripLeadingSilentToken,

stripSilentToken,

} from "../auto-reply/tokens.js";

import { defaultRuntime } from "../runtime.js";

import { isCronSessionKey } from "../sessions/session-key-utils.js";

import { createLazyImportLoader } from "../shared/lazy-promise.js";

import { type DeliveryContext, normalizeDeliveryContext } from "../utils/delivery-context.js";

import { INTERNAL_MESSAGE_CHANNEL } from "../utils/message-channel.js";

import {

buildAnnounceIdFromChildRun,

buildAnnounceIdempotencyKey,

} from "./announce-idempotency.js";

import { formatAgentInternalEventsForPrompt, type AgentInternalEvent } from "./internal-events.js";

import {

deliverSubagentAnnouncement,

loadRequesterSessionEntry,

loadSessionEntryByKey,

runAnnounceDeliveryWithRetry,

resolveSubagentAnnounceTimeoutMs,

resolveSubagentCompletionOrigin,

} from "./subagent-announce-delivery.js";

import type { SubagentAnnounceDeliveryResult } from "./subagent-announce-dispatch.js";

import { resolveAnnounceOrigin } from "./subagent-announce-origin.js";

import {

applySubagentWaitOutcome,

buildChildCompletionFindings,

buildCompactAnnounceStatsLine,

dedupeLatestChildCompletionRows,

filterCurrentDirectChildCompletionRows,

readLatestSubagentOutputWithRetry,

readSubagentOutput,

type SubagentRunOutcome,

waitForSubagentRunOutcome,

} from "./subagent-announce-output.js";

import {

callGateway,

dispatchGatewayMethodInProcess,

isEmbeddedAgentRunActive,

getRuntimeConfig,

waitForEmbeddedAgentRunEnd,

} from "./subagent-announce.runtime.js";

import { getSubagentDepthFromSessionStore } from "./subagent-depth.js";

import { deleteSubagentSessionForCleanup } from "./subagent-session-cleanup.js";

import type { SpawnSubagentMode } from "./subagent-spawn.types.js";

import { isAnnounceSkip } from "./tools/sessions-send-tokens.js";

type SubagentAnnounceDeps = {

callGateway: typeof callGateway;

dispatchGatewayMethodInProcess: typeof dispatchGatewayMethodInProcess;

getRuntimeConfig: typeof getRuntimeConfig;

loadSubagentRegistryRuntime: typeof loadSubagentRegistryRuntime;

};

const defaultSubagentAnnounceDeps: SubagentAnnounceDeps = {

callGateway,

dispatchGatewayMethodInProcess,

getRuntimeConfig,

loadSubagentRegistryRuntime,

};

let subagentAnnounceDeps: SubagentAnnounceDeps = defaultSubagentAnnounceDeps;

const subagentRegistryRuntimeLoader = createLazyImportLoader(

() => import("./subagent-announce.registry.runtime.js"),

);

function loadSubagentRegistryRuntime() {

return subagentRegistryRuntimeLoader.load();

}

export { buildSubagentSystemPrompt } from "./subagent-system-prompt.js";

export { captureSubagentCompletionReply } from "./subagent-announce-output.js";

export type { SubagentRunOutcome } from "./subagent-announce-output.js";

export type SubagentAnnounceType = "subagent task" | "cron job";

function buildAnnounceReplyInstruction(params: {

requesterIsSubagent: boolean;

announceType: SubagentAnnounceType;

expectsCompletionMessage?: boolean;

}): string {

if (params.requesterIsSubagent) {

return `Convert this completion into a concise internal orchestration update for your parent agent in your own words. Keep this internal context private (don't mention system/log/stats/session details or announce type). If this result is duplicate or no update is needed, reply ONLY: ${SILENT_REPLY_TOKEN}.`;

}

if (params.expectsCompletionMessage) {

return `A completed ${params.announceType} is ready for parent review. Review/verify the result above before deciding whether the original task is done. If additional action is required, continue the task or record a follow-up; otherwise send a truthful user-facing update. Keep this internal context private (don't mention system/log/stats/session details or announce type). Reply ONLY: ${SILENT_REPLY_TOKEN} only when this exact result is already visible to the user in this same turn.`;

}

return `A completed ${params.announceType} is ready for parent review. Review/verify the result above before deciding whether the original task is done. If additional action is required, continue the task or record a follow-up; otherwise send a truthful user-facing update. Keep this internal context private (don't mention system/log/stats/session details or announce type), and do not copy the internal event text verbatim. Reply ONLY: ${SILENT_REPLY_TOKEN} if this exact result was already delivered to the user in this same turn.`;

}

function buildAnnounceSteerMessage(events: AgentInternalEvent[]): string {

return (

formatAgentInternalEventsForPrompt(events) ||

"A background task finished. Process the completion update now."

);

}

function hasUsableSessionEntry(entry: unknown): boolean {

if (!entry || typeof entry !== "object") {

return false;

}

const sessionId = (entry as { sessionId?: unknown }).sessionId;

return typeof sessionId !== "string" || sessionId.trim() !== "";

}

function buildDescendantWakeMessage(params: { findings: string; taskLabel: string }): string {

return [

"[Subagent Context] Your prior run ended while waiting for descendant subagent completions.",

"[Subagent Context] All pending descendants for that run have now settled.",

"[Subagent Context] Continue your workflow using these results. Spawn more subagents if needed, otherwise send your final answer.",

"",

`Task: ${params.taskLabel}`,

"",

params.findings,

].join("\n");

}

const WAKE_RUN_SUFFIX = ":wake";

function stripWakeRunSuffixes(runId: string): string {

let next = runId.trim();

while (next.endsWith(WAKE_RUN_SUFFIX)) {

next = next.slice(0, -WAKE_RUN_SUFFIX.length);

}

return next || runId.trim();

}

function isWakeContinuationRun(runId: string): boolean {

const trimmed = runId.trim();

if (!trimmed) {

return false;

}

return stripWakeRunSuffixes(trimmed) !== trimmed;

}

function stripAndClassifyReply(text: string): string | null {

let result = text;

let didStrip = false;

const hasLeadingSilentToken = startsWithSilentToken(result, SILENT_REPLY_TOKEN);

if (hasLeadingSilentToken) {

result = stripLeadingSilentToken(result, SILENT_REPLY_TOKEN);

didStrip = true;

}

if (hasLeadingSilentToken || result.toLowerCase().includes(SILENT_REPLY_TOKEN.toLowerCase())) {

result = stripSilentToken(result, SILENT_REPLY_TOKEN);

didStrip = true;

}

if (

didStrip &&

(!result.trim() || isSilentReplyText(result, SILENT_REPLY_TOKEN) || isAnnounceSkip(result))

) {

return null;

}

return result;

}

async function wakeSubagentRunAfterDescendants(params: {

runId: string;

childSessionKey: string;

taskLabel: string;

findings: string;

announceId: string;

signal?: AbortSignal;

}): Promise<boolean> {

if (params.signal?.aborted) {

return false;

}

const childEntry = loadSessionEntryByKey(params.childSessionKey);

if (!hasUsableSessionEntry(childEntry)) {

return false;

}

const cfg = subagentAnnounceDeps.getRuntimeConfig();

const announceTimeoutMs = resolveSubagentAnnounceTimeoutMs(cfg);

const wakeMessage = buildDescendantWakeMessage({

findings: params.findings,

taskLabel: params.taskLabel,

});

let wakeRunId;

try {

const wakeResponse = await runAnnounceDeliveryWithRetry<{ runId?: string }>({

operation: "descendant wake agent call",

signal: params.signal,

run: async () =>

await subagentAnnounceDeps.dispatchGatewayMethodInProcess(

"agent",

{

sessionKey: params.childSessionKey,

message: wakeMessage,

deliver: false,

inputProvenance: {

kind: "inter_session",

sourceSessionKey: params.childSessionKey,

sourceChannel: INTERNAL_MESSAGE_CHANNEL,

sourceTool: "subagent_announce",

},

idempotencyKey: buildAnnounceIdempotencyKey(`${params.announceId}:wake`),

},

{

timeoutMs: announceTimeoutMs,

},

),

});

wakeRunId = normalizeOptionalString(wakeResponse?.runId) ?? "";

} catch {

return false;

}

if (!wakeRunId) {

return false;

}

const { replaceSubagentRunAfterSteer } = await loadSubagentRegistryRuntime();

return replaceSubagentRunAfterSteer({

previousRunId: params.runId,

nextRunId: wakeRunId,

preserveFrozenResultFallback: true,

});

}

export async function runSubagentAnnounceFlow(params: {

childSessionKey: string;

childRunId: string;

requesterSessionKey: string;

requesterOrigin?: DeliveryContext;

requesterDisplayKey: string;

task: string;

timeoutMs: number;

cleanup: "delete" | "keep";

roundOneReply?: string;

/**

* Fallback text preserved from the pre-wake run when a wake continuation

* completes with NO_REPLY despite an earlier final summary already existing.

*/

fallbackReply?: string;

waitForCompletion?: boolean;

startedAt?: number;

endedAt?: number;

label?: string;

outcome?: SubagentRunOutcome;

announceType?: SubagentAnnounceType;

expectsCompletionMessage?: boolean;

spawnMode?: SpawnSubagentMode;

wakeOnDescendantSettle?: boolean;

signal?: AbortSignal;

bestEffortDeliver?: boolean;

onDeliveryResult?: (delivery: SubagentAnnounceDeliveryResult) => void;

}): Promise<boolean> {

let didAnnounce = false;

const expectsCompletionMessage = params.expectsCompletionMessage === true;

const announceType = params.announceType ?? "subagent task";

let shouldDeleteChildSession = params.cleanup === "delete";

try {

let targetRequesterSessionKey = params.requesterSessionKey;

let targetRequesterOrigin = normalizeDeliveryContext(params.requesterOrigin);

const childSessionId = (() => {

const entry = loadSessionEntryByKey(params.childSessionKey);

return typeof entry?.sessionId === "string" && entry.sessionId.trim()

? entry.sessionId.trim()

: undefined;

})();

const settleTimeoutMs = Math.min(Math.max(params.timeoutMs, 1), 120_000);

let reply = params.roundOneReply;

let outcome: SubagentRunOutcome | undefined = params.outcome;

if (childSessionId && isEmbeddedAgentRunActive(childSessionId)) {

const settled = await waitForEmbeddedAgentRunEnd(childSessionId, settleTimeoutMs);

if (!settled && isEmbeddedAgentRunActive(childSessionId)) {

shouldDeleteChildSession = false;

// Keep delete cleanup retryable until the active child can be removed.

if (outcome?.status !== "timeout" || params.cleanup === "delete") {

return false;

}

}

}

if (!reply && params.waitForCompletion !== false) {

const wait = await waitForSubagentRunOutcome(params.childRunId, settleTimeoutMs);

const applied = applySubagentWaitOutcome({

wait,

outcome,

startedAt: params.startedAt,

endedAt: params.endedAt,

});

outcome = applied.outcome;

params.startedAt = applied.startedAt;

params.endedAt = applied.endedAt;

}

if (!outcome) {

outcome = { status: "unknown" };

}

const failedTerminalOutcome = outcome.status === "error";

const allowFailedOutputCapture =

!failedTerminalOutcome || (!params.roundOneReply && !params.fallbackReply);

if (failedTerminalOutcome) {

reply = undefined;

}

let requesterDepth = getSubagentDepthFromSessionStore(targetRequesterSessionKey);

const requesterIsInternalSession = () =>

requesterDepth >= 1 || isCronSessionKey(targetRequesterSessionKey);

let childCompletionFindings: string | undefined;

let subagentRegistryRuntime:

| Awaited<ReturnType<typeof loadSubagentRegistryRuntime>>

| undefined;

try {

subagentRegistryRuntime = await subagentAnnounceDeps.loadSubagentRegistryRuntime();

if (

requesterDepth >= 1 &&

subagentRegistryRuntime.shouldIgnorePostCompletionAnnounceForSession(

targetRequesterSessionKey,

)

) {

return true;

}

const pendingChildDescendantRuns = Math.max(

0,

subagentRegistryRuntime.countPendingDescendantRuns(params.childSessionKey),

);

if (pendingChildDescendantRuns > 0 && announceType !== "cron job") {

shouldDeleteChildSession = false;

return false;

}

if (typeof subagentRegistryRuntime.listSubagentRunsForRequester === "function") {

const directChildren = subagentRegistryRuntime.listSubagentRunsForRequester(

params.childSessionKey,

{

requesterRunId: params.childRunId,

},

);

if (Array.isArray(directChildren) && directChildren.length > 0) {

childCompletionFindings = buildChildCompletionFindings(

dedupeLatestChildCompletionRows(

filterCurrentDirectChildCompletionRows(directChildren, {

requesterSessionKey: params.childSessionKey,

getLatestSubagentRunByChildSessionKey:

subagentRegistryRuntime.getLatestSubagentRunByChildSessionKey,

}),

),

);

}

}

} catch {

// Best-effort only.

}

const announceId = buildAnnounceIdFromChildRun({

childSessionKey: params.childSessionKey,

childRunId: params.childRunId,

});

const childRunAlreadyWoken = isWakeContinuationRun(params.childRunId);

if (

params.wakeOnDescendantSettle === true &&

childCompletionFindings?.trim() &&

!childRunAlreadyWoken

) {

const wakeAnnounceId = buildAnnounceIdFromChildRun({

childSessionKey: params.childSessionKey,

childRunId: stripWakeRunSuffixes(params.childRunId),

});

const woke = await wakeSubagentRunAfterDescendants({

runId: params.childRunId,

childSessionKey: params.childSessionKey,

taskLabel: params.label || params.task || "task",

findings: childCompletionFindings,

announceId: wakeAnnounceId,

signal: params.signal,

});

if (woke) {

shouldDeleteChildSession = false;

return true;

}

}

if (!childCompletionFindings) {

const fallbackReply = failedTerminalOutcome

? undefined

: normalizeOptionalString(params.fallbackReply);

const fallbackIsSilent =

Boolean(fallbackReply) &&

(isAnnounceSkip(fallbackReply) || isSilentReplyText(fallbackReply, SILENT_REPLY_TOKEN));

if (!reply && allowFailedOutputCapture) {

reply = await readSubagentOutput(params.childSessionKey, outcome);

}

if (!reply?.trim() && allowFailedOutputCapture) {

reply = await readLatestSubagentOutputWithRetry({

sessionKey: params.childSessionKey,

maxWaitMs: params.timeoutMs,

outcome,

});

}

if (!reply?.trim() && fallbackReply && !fallbackIsSilent) {

reply = fallbackReply;

}

// A worker can finish just after the first wait request timed out.

// If we already have real completion content, do one cached recheck so

// the final completion event prefers the authoritative terminal state.

// This is best-effort; if the recheck fails, keep the known timeout

// outcome instead of dropping the announcement entirely.

if (outcome?.status === "timeout" && reply?.trim() && params.waitForCompletion !== false) {

try {

const rechecked = await waitForSubagentRunOutcome(params.childRunId, 0);

const applied = applySubagentWaitOutcome({

wait: rechecked,

outcome,

startedAt: params.startedAt,

endedAt: params.endedAt,

});

outcome = applied.outcome;

params.startedAt = applied.startedAt;

params.endedAt = applied.endedAt;

} catch {

// Best-effort recheck; keep the existing timeout outcome on failure.

}

}

if (isAnnounceSkip(reply) || isSilentReplyText(reply, SILENT_REPLY_TOKEN)) {

if (fallbackReply && !fallbackIsSilent) {

const cleaned = stripAndClassifyReply(fallbackReply);

if (cleaned === null) {

return true;

}

reply = cleaned;

} else {

return true;

}

} else if (reply) {

const cleaned = stripAndClassifyReply(reply);

if (cleaned === null) {

if (fallbackReply && !fallbackIsSilent) {

const cleanedFallback = stripAndClassifyReply(fallbackReply);

if (cleanedFallback === null) {

return true;

}

reply = cleanedFallback;

} else {

return true;

}

} else {

reply = cleaned;

}

}

}

if (!outcome) {

outcome = { status: "unknown" };

}

// Build status label

const statusLabel =

outcome.status === "ok"

? "completed; ready for parent review"

: outcome.status === "timeout"

? "timed out"

: outcome.status === "error"

? `failed: ${outcome.error || "unknown error"}`

: "finished with unknown status";

const taskLabel = params.label || params.task || "task";

const announceSessionId = childSessionId || "unknown";

const findings = childCompletionFindings || reply || "(no output)";

let requesterIsSubagent = requesterIsInternalSession();

if (requesterIsSubagent) {

const {

isSubagentSessionRunActive,

resolveRequesterForChildSession,

shouldIgnorePostCompletionAnnounceForSession,

} = subagentRegistryRuntime ?? (await loadSubagentRegistryRuntime());

if (!isSubagentSessionRunActive(targetRequesterSessionKey)) {

if (shouldIgnorePostCompletionAnnounceForSession(targetRequesterSessionKey)) {

return true;

}

const parentSessionEntry = loadSessionEntryByKey(targetRequesterSessionKey);

const parentSessionAlive = hasUsableSessionEntry(parentSessionEntry);

if (!parentSessionAlive) {

const fallback = resolveRequesterForChildSession(targetRequesterSessionKey);

if (!fallback?.requesterSessionKey) {

shouldDeleteChildSession = false;

return false;

}

targetRequesterSessionKey = fallback.requesterSessionKey;

targetRequesterOrigin =

normalizeDeliveryContext(fallback.requesterOrigin) ?? targetRequesterOrigin;

requesterDepth = getSubagentDepthFromSessionStore(targetRequesterSessionKey);

requesterIsSubagent = requesterIsInternalSession();

}

}

}

const replyInstruction = buildAnnounceReplyInstruction({

requesterIsSubagent,

announceType,

expectsCompletionMessage,

});

const statsLine = await buildCompactAnnounceStatsLine({

sessionKey: params.childSessionKey,

startedAt: params.startedAt,

endedAt: params.endedAt,

});

const internalEvents: AgentInternalEvent[] = [

{

type: "task_completion",

source: announceType === "cron job" ? "cron" : "subagent",

childSessionKey: params.childSessionKey,

childSessionId: announceSessionId,

announceType,

taskLabel,

status: outcome.status,

statusLabel,

result: findings,

statsLine,

replyInstruction,

},

];

const triggerMessage = buildAnnounceSteerMessage(internalEvents);

// Send to the requester session. For nested subagents this is an internal

// follow-up injection (deliver=false) so the orchestrator receives it.

let directOrigin = targetRequesterOrigin;

if (!requesterIsSubagent) {

const { entry } = loadRequesterSessionEntry(targetRequesterSessionKey);

directOrigin = resolveAnnounceOrigin(entry, targetRequesterOrigin);

}

const completionDirectOrigin =

expectsCompletionMessage && !requesterIsSubagent

? await resolveSubagentCompletionOrigin({

childSessionKey: params.childSessionKey,

requesterSessionKey: targetRequesterSessionKey,

requesterOrigin: directOrigin,

childRunId: params.childRunId,

spawnMode: params.spawnMode,

expectsCompletionMessage,

})

: targetRequesterOrigin;

const directIdempotencyKey = buildAnnounceIdempotencyKey(announceId);

const delivery = await deliverSubagentAnnouncement({

requesterSessionKey: targetRequesterSessionKey,

announceId,

triggerMessage,

steerMessage: triggerMessage,

internalEvents,

summaryLine: taskLabel,

requesterSessionOrigin: targetRequesterOrigin,

requesterOrigin:

expectsCompletionMessage && !requesterIsSubagent

? completionDirectOrigin

: targetRequesterOrigin,

completionDirectOrigin,

directOrigin,

sourceSessionKey: params.childSessionKey,

sourceChannel: INTERNAL_MESSAGE_CHANNEL,

sourceTool: "subagent_announce",

targetRequesterSessionKey,

requesterIsSubagent,

expectsCompletionMessage,

bestEffortDeliver: params.bestEffortDeliver,

directIdempotencyKey,

signal: params.signal,

});

params.onDeliveryResult?.(delivery);

didAnnounce = delivery.delivered || delivery.terminal === true;

if (!delivery.delivered && delivery.path === "direct" && delivery.error) {

defaultRuntime.log(

`[warn] Subagent completion direct announce failed for run ${params.childRunId}: ${delivery.error}`,

);

}

} catch (err) {

defaultRuntime.error?.(`Subagent announce failed: ${String(err)}`);

// Best-effort follow-ups; ignore failures to avoid breaking the caller response.

} finally {

// Patch label after all writes complete

if (params.label) {

try {

await subagentAnnounceDeps.callGateway({

method: "sessions.patch",

params: { key: params.childSessionKey, label: params.label },

timeoutMs: 10_000,

});

} catch {

// Best-effort

}

}

if (shouldDeleteChildSession) {

await deleteSubagentSessionForCleanup({

callGateway: subagentAnnounceDeps.callGateway,

childSessionKey: params.childSessionKey,

spawnMode: params.spawnMode,

});

}

}

return didAnnounce;

}

export const testing = {

setDepsForTest(

overrides?: Partial<SubagentAnnounceDeps> & {

callGateway?: typeof callGateway;

},

) {

const callGatewayOverride = overrides?.callGateway;

const dispatchGatewayMethodInProcessOverride =

overrides?.dispatchGatewayMethodInProcess ??

(callGatewayOverride

? ((async (method, agentParams, options) =>

await callGatewayOverride({

method,

params: agentParams,

expectFinal: options?.expectFinal,

timeoutMs: options?.timeoutMs,

})) satisfies typeof dispatchGatewayMethodInProcess)

: undefined);

subagentAnnounceDeps = overrides

? {

...defaultSubagentAnnounceDeps,

...overrides,

...(dispatchGatewayMethodInProcessOverride

? { dispatchGatewayMethodInProcess: dispatchGatewayMethodInProcessOverride }

: {}),

}

: defaultSubagentAnnounceDeps;

},

};

export { testing as __testing };