github/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-05-27 19:20:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(agents): remove send path from subagent announce (again) -- routes all completions through agent path so parent LLM processes results. Third time fixing this regression.

Browse Source
This commit is contained in:
Tyler Yust
2026-02-21 12:53:37 -08:00
parent 853ae626fa
commit c20fe0346d
3 changed files with 104 additions and 518 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
src/agents/subagent-announce.format.e2e.test.ts
View File

@@ -374,7 +374,6 @@ describe("subagent announce formatting", () => {
requesterDisplayKey: "main",
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
});
expect(didAnnounce).toBe(true);
@@ -415,7 +414,6 @@ describe("subagent announce formatting", () => {
requesterDisplayKey: "main",
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
});
expect(didAnnounce).toBe(true);
@@ -479,7 +477,6 @@ describe("subagent announce formatting", () => {
requesterDisplayKey: "main",
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
spawnMode: "session",
});
@@ -564,7 +561,6 @@ describe("subagent announce formatting", () => {
accountId: "acct-1",
},
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
spawnMode: "session",
}),
runSubagentAnnounceFlow({
@@ -578,7 +574,6 @@ describe("subagent announce formatting", () => {
accountId: "acct-1",
},
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
spawnMode: "session",
}),
]);
@@ -618,7 +613,6 @@ describe("subagent announce formatting", () => {
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
...defaultOutcomeAnnounce,
outcome: { status: "error", error: "boom" },
expectsCompletionMessage: true,
spawnMode: "session",
});
@@ -655,7 +649,6 @@ describe("subagent announce formatting", () => {
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
...defaultOutcomeAnnounce,
outcome: { status: "timeout" },
expectsCompletionMessage: true,
});
expect(didAnnounce).toBe(true);
@@ -692,7 +685,6 @@ describe("subagent announce formatting", () => {
requesterDisplayKey: "main",
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
});
expect(didAnnounce).toBe(true);
@@ -730,7 +722,6 @@ describe("subagent announce formatting", () => {
threadId: 99,
},
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
});
expect(didAnnounce).toBe(true);
@@ -766,7 +757,6 @@ describe("subagent announce formatting", () => {
threadId: "777",
},
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
spawnMode: "session",
});
@@ -783,7 +773,6 @@ describe("subagent announce formatting", () => {
},
childRunId: "run-direct-thread-bound",
spawnMode: "session",
expectsCompletionMessage: true,
},
{
runId: "run-direct-thread-bound",
@@ -824,7 +813,6 @@ describe("subagent announce formatting", () => {
accountId: "acct-1",
},
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
spawnMode: "session",
});
@@ -852,7 +840,6 @@ describe("subagent announce formatting", () => {
accountId: "acct-1",
},
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
spawnMode: "session",
});
@@ -885,7 +872,6 @@ describe("subagent announce formatting", () => {
accountId: "acct-1",
},
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
spawnMode: "session",
});
@@ -921,7 +907,6 @@ describe("subagent announce formatting", () => {
threadId: "999",
},
...defaultOutcomeAnnounce,
expectsCompletionMessage: true,
spawnMode: "session",
});
@@ -1067,7 +1052,6 @@ describe("subagent announce formatting", () => {
childRunId: "run-completion-direct-fallback",
requesterSessionKey: "main",
requesterDisplayKey: "main",
expectsCompletionMessage: true,
...defaultOutcomeAnnounce,
});
@@ -1106,7 +1090,6 @@ describe("subagent announce formatting", () => {
childRunId: "run-completion-direct-fail",
requesterSessionKey: "main",
requesterDisplayKey: "main",
expectsCompletionMessage: true,
...defaultOutcomeAnnounce,
});
@@ -1137,7 +1120,6 @@ describe("subagent announce formatting", () => {
requesterSessionKey: "agent:main:main",
requesterDisplayKey: "main",
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
expectsCompletionMessage: true,
...defaultOutcomeAnnounce,
});
@@ -1171,7 +1153,6 @@ describe("subagent announce formatting", () => {
requesterSessionKey: "agent:main:main",
requesterDisplayKey: "main",
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
expectsCompletionMessage: true,
...defaultOutcomeAnnounce,
});
@@ -1200,7 +1181,6 @@ describe("subagent announce formatting", () => {
requesterSessionKey: "agent:main:main",
requesterDisplayKey: "main",
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
expectsCompletionMessage: true,
...defaultOutcomeAnnounce,
});
@@ -1405,7 +1385,6 @@ describe("subagent announce formatting", () => {
requesterSessionKey: "agent:main:subagent:orchestrator",
requesterOrigin: { channel: "whatsapp", accountId: "acct-123", to: "+1555" },
requesterDisplayKey: "agent:main:subagent:orchestrator",
expectsCompletionMessage: true,
...defaultOutcomeAnnounce,
});
@@ -1510,7 +1489,6 @@ describe("subagent announce formatting", () => {
childRunId: "run-parent-completion",
requesterSessionKey: "agent:main:main",
requesterDisplayKey: "main",
expectsCompletionMessage: true,
...defaultOutcomeAnnounce,
});
@@ -1652,7 +1630,6 @@ describe("subagent announce formatting", () => {
startedAt: 10,
endedAt: 20,
outcome: { status: "ok" },
expectsCompletionMessage: true,
});
expect(didAnnounce).toBe(false);

581
src/agents/subagent-announce.ts
View File

@@ -14,7 +14,6 @@ import type { ConversationRef } from "../infra/outbound/session-binding-service.
import { getGlobalHookRunner } from "../plugins/hook-runner-global.js";
import { normalizeAccountId, normalizeMainKey } from "../routing/session-key.js";
import { defaultRuntime } from "../runtime.js";
import { extractTextFromChatContent } from "../shared/chat-content.js";
import {
type DeliveryContext,
deliveryContextFromSession,
@@ -36,198 +35,27 @@ import { type AnnounceQueueItem, enqueueAnnounce } from "./subagent-announce-que
import { getSubagentDepthFromSessionStore } from "./subagent-depth.js";
import type { SpawnSubagentMode } from "./subagent-spawn.js";
import { readLatestAssistantReply } from "./tools/agent-step.js";
import { sanitizeTextContent, extractAssistantText } from "./tools/sessions-helpers.js";
type ToolResultMessage = {
role?: unknown;
content?: unknown;
};
type SubagentDeliveryPath = "queued" | "steered" | "direct" | "none";
type SubagentAnnounceDeliveryResult = {
delivered: boolean;
path: SubagentDeliveryPath;
error?: string;
};
function buildCompletionDeliveryMessage(params: {
findings: string;
subagentName: string;
spawnMode?: SpawnSubagentMode;
outcome?: SubagentRunOutcome;
}): string {
const findingsText = params.findings.trim();
const hasFindings = findingsText.length > 0 && findingsText !== "(no output)";
const header = (() => {
if (params.outcome?.status === "error") {
return params.spawnMode === "session"
? `❌ Subagent ${params.subagentName} failed this task (session remains active)`
: `❌ Subagent ${params.subagentName} failed`;
}
if (params.outcome?.status === "timeout") {
return params.spawnMode === "session"
? `⏱️ Subagent ${params.subagentName} timed out on this task (session remains active)`
: `⏱️ Subagent ${params.subagentName} timed out`;
}
return params.spawnMode === "session"
? `✅ Subagent ${params.subagentName} completed this task (session remains active)`
: `✅ Subagent ${params.subagentName} finished`;
})();
if (!hasFindings) {
return header;
}
return `${header}\n\n${findingsText}`;
}
function summarizeDeliveryError(error: unknown): string {
if (error instanceof Error) {
return error.message || "error";
}
if (typeof error === "string") {
return error;
}
if (error === undefined || error === null) {
return "unknown error";
}
try {
return JSON.stringify(error);
} catch {
return "error";
}
}
function extractToolResultText(content: unknown): string {
if (typeof content === "string") {
return sanitizeTextContent(content);
}
if (content && typeof content === "object" && !Array.isArray(content)) {
const obj = content as {
text?: unknown;
output?: unknown;
content?: unknown;
result?: unknown;
error?: unknown;
summary?: unknown;
};
if (typeof obj.text === "string") {
return sanitizeTextContent(obj.text);
}
if (typeof obj.output === "string") {
return sanitizeTextContent(obj.output);
}
if (typeof obj.content === "string") {
return sanitizeTextContent(obj.content);
}
if (typeof obj.result === "string") {
return sanitizeTextContent(obj.result);
}
if (typeof obj.error === "string") {
return sanitizeTextContent(obj.error);
}
if (typeof obj.summary === "string") {
return sanitizeTextContent(obj.summary);
}
}
if (!Array.isArray(content)) {
return "";
}
const joined = extractTextFromChatContent(content, {
sanitizeText: sanitizeTextContent,
normalizeText: (text) => text,
joinWith: "\n",
});
return joined?.trim() ?? "";
}
function extractInlineTextContent(content: unknown): string {
if (!Array.isArray(content)) {
return "";
}
return (
extractTextFromChatContent(content, {
sanitizeText: sanitizeTextContent,
normalizeText: (text) => text.trim(),
joinWith: "",
}) ?? ""
);
}
function extractSubagentOutputText(message: unknown): string {
if (!message || typeof message !== "object") {
return "";
}
const role = (message as { role?: unknown }).role;
const content = (message as { content?: unknown }).content;
if (role === "assistant") {
const assistantText = extractAssistantText(message);
if (assistantText) {
return assistantText;
}
if (typeof content === "string") {
return sanitizeTextContent(content);
}
if (Array.isArray(content)) {
return extractInlineTextContent(content);
}
return "";
}
if (role === "toolResult" || role === "tool") {
return extractToolResultText((message as ToolResultMessage).content);
}
if (role == null) {
if (typeof content === "string") {
return sanitizeTextContent(content);
}
if (Array.isArray(content)) {
return extractInlineTextContent(content);
}
}
return "";
}
async function readLatestSubagentOutput(sessionKey: string): Promise<string | undefined> {
try {
const latestAssistant = await readLatestAssistantReply({
sessionKey,
limit: 50,
});
if (latestAssistant?.trim()) {
return latestAssistant;
}
} catch {
// Best-effort: fall back to richer history parsing below.
}
const history = await callGateway<{ messages?: Array<unknown> }>({
method: "chat.history",
params: { sessionKey, limit: 50 },
});
const messages = Array.isArray(history?.messages) ? history.messages : [];
for (let i = messages.length - 1; i >= 0; i -= 1) {
const msg = messages[i];
const text = extractSubagentOutputText(msg);
if (text) {
return text;
}
}
return undefined;
}
async function readLatestSubagentOutputWithRetry(params: {
async function readLatestAssistantReplyWithRetry(params: {
sessionKey: string;
initialReply?: string;
maxWaitMs: number;
}): Promise<string | undefined> {
const RETRY_INTERVAL_MS = 100;
const deadline = Date.now() + Math.max(0, Math.min(params.maxWaitMs, 15_000));
let result: string | undefined;
while (Date.now() < deadline) {
result = await readLatestSubagentOutput(params.sessionKey);
if (result?.trim()) {
return result;
}
await new Promise((resolve) => setTimeout(resolve, RETRY_INTERVAL_MS));
let reply = params.initialReply?.trim() ? params.initialReply : undefined;
if (reply) {
return reply;
}
return result;
const deadline = Date.now() + Math.max(0, Math.min(params.maxWaitMs, 15_000));
while (Date.now() < deadline) {
await new Promise((resolve) => setTimeout(resolve, RETRY_INTERVAL_MS));
const latest = await readLatestAssistantReply({ sessionKey: params.sessionKey });
if (latest?.trim()) {
return latest;
}
}
return reply;
}
async function waitForSubagentOutputChange(params: {
@@ -243,7 +71,10 @@ async function waitForSubagentOutputChange(params: {
const deadline = Date.now() + Math.max(0, Math.min(params.maxWaitMs, 5_000));
let latest = params.baselineReply;
while (Date.now() < deadline) {
const next = await readLatestSubagentOutput(params.sessionKey);
const next = await readLatestAssistantReply({
sessionKey: params.sessionKey,
limit: 50,
});
if (next?.trim()) {
latest = next;
if (next.trim() !== baseline) {
@@ -359,17 +190,13 @@ function resolveAnnounceOrigin(
return mergeDeliveryContext(normalizedRequester, entryForMerge);
}
async function resolveSubagentCompletionOrigin(params: {
async function resolveSubagentDeliveryOrigin(params: {
childSessionKey: string;
requesterSessionKey: string;
requesterOrigin?: DeliveryContext;
childRunId?: string;
spawnMode?: SpawnSubagentMode;
expectsCompletionMessage: boolean;
}): Promise<{
origin?: DeliveryContext;
routeMode: "bound" | "fallback" | "hook";
}> {
}): Promise<DeliveryContext | undefined> {
const requesterOrigin = normalizeDeliveryContext(params.requesterOrigin);
const requesterConversation = (() => {
const channel = requesterOrigin?.channel?.trim().toLowerCase();
@@ -404,30 +231,24 @@ async function resolveSubagentCompletionOrigin(params: {
to: `channel:${route.binding.conversation.conversationId}`,
threadId: route.binding.conversation.conversationId,
};
return {
// Bound target is authoritative; requester hints fill only missing fields.
origin: mergeDeliveryContext(boundOrigin, requesterOrigin),
routeMode: "bound",
};
// Bound target is authoritative; requester hints fill only missing fields.
return mergeDeliveryContext(boundOrigin, requesterOrigin);
}
const hookRunner = getGlobalHookRunner();
if (!hookRunner?.hasHooks("subagent_delivery_target")) {
return {
origin: requesterOrigin,
routeMode: "fallback",
};
return requesterOrigin;
}
try {
const hookEvent = {
childSessionKey: params.childSessionKey,
requesterSessionKey: params.requesterSessionKey,
requesterOrigin,
childRunId: params.childRunId,
spawnMode: params.spawnMode,
};
const result = await hookRunner.runSubagentDeliveryTarget(
{
childSessionKey: params.childSessionKey,
requesterSessionKey: params.requesterSessionKey,
requesterOrigin,
childRunId: params.childRunId,
spawnMode: params.spawnMode,
expectsCompletionMessage: params.expectsCompletionMessage,
},
hookEvent as Parameters<typeof hookRunner.runSubagentDeliveryTarget>[0],
{
runId: params.childRunId,
childSessionKey: params.childSessionKey,
@@ -436,27 +257,15 @@ async function resolveSubagentCompletionOrigin(params: {
);
const hookOrigin = normalizeDeliveryContext(result?.origin);
if (!hookOrigin) {
return {
origin: requesterOrigin,
routeMode: "fallback",
};
return requesterOrigin;
}
if (hookOrigin.channel && !isDeliverableMessageChannel(hookOrigin.channel)) {
return {
origin: requesterOrigin,
routeMode: "fallback",
};
return requesterOrigin;
}
// Hook-provided origin should override requester defaults when present.
return {
origin: mergeDeliveryContext(hookOrigin, requesterOrigin),
routeMode: "hook",
};
return mergeDeliveryContext(hookOrigin, requesterOrigin);
} catch {
return {
origin: requesterOrigin,
routeMode: "fallback",
};
return requesterOrigin;
}
}
@@ -578,210 +387,45 @@ async function maybeQueueSubagentAnnounce(params: {
return "none";
}
function queueOutcomeToDeliveryResult(
outcome: "steered" | "queued" | "none",
): SubagentAnnounceDeliveryResult {
if (outcome === "steered") {
return {
delivered: true,
path: "steered",
};
}
if (outcome === "queued") {
return {
delivered: true,
path: "queued",
};
}
return {
delivered: false,
path: "none",
};
}
async function sendSubagentAnnounceDirectly(params: {
targetRequesterSessionKey: string;
triggerMessage: string;
completionMessage?: string;
expectsCompletionMessage: boolean;
completionRouteMode?: "bound" | "fallback" | "hook";
spawnMode?: SpawnSubagentMode;
directIdempotencyKey: string;
completionDirectOrigin?: DeliveryContext;
directOrigin?: DeliveryContext;
requesterIsSubagent: boolean;
}): Promise<SubagentAnnounceDeliveryResult> {
}): Promise<void> {
// ⚠️ CRITICAL: DO NOT ADD A "send" PATH HERE
// All subagent completions MUST route through method: "agent" so the parent
// session's LLM can process the result. Using method: "send" bypasses the
// LLM entirely and breaks: subagent orchestration, cron subagents, nested
// subagent chains, and any workflow that depends on the parent processing
// results. This has been broken and reverted 3 times already.
// Contact @tyler6204 before making ANY changes to this function.
const cfg = loadConfig();
const canonicalRequesterSessionKey = resolveRequesterStoreKey(
cfg,
params.targetRequesterSessionKey,
);
try {
const completionDirectOrigin = normalizeDeliveryContext(params.completionDirectOrigin);
const completionChannelRaw =
typeof completionDirectOrigin?.channel === "string"
? completionDirectOrigin.channel.trim()
: "";
const completionChannel =
completionChannelRaw && isDeliverableMessageChannel(completionChannelRaw)
? completionChannelRaw
: "";
const completionTo =
typeof completionDirectOrigin?.to === "string" ? completionDirectOrigin.to.trim() : "";
const hasCompletionDirectTarget =
!params.requesterIsSubagent && Boolean(completionChannel) && Boolean(completionTo);
if (
params.expectsCompletionMessage &&
hasCompletionDirectTarget &&
params.completionMessage?.trim()
) {
const forceBoundSessionDirectDelivery =
params.spawnMode === "session" &&
(params.completionRouteMode === "bound" || params.completionRouteMode === "hook");
let shouldSendCompletionDirectly = true;
if (!forceBoundSessionDirectDelivery) {
let activeDescendantRuns = 0;
try {
const { countActiveDescendantRuns } = await import("./subagent-registry.js");
activeDescendantRuns = Math.max(
0,
countActiveDescendantRuns(canonicalRequesterSessionKey),
);
} catch {
// Best-effort only; when unavailable keep historical direct-send behavior.
}
// Keep non-bound completion announcements coordinated via requester
// session routing while sibling/descendant runs are still active.
if (activeDescendantRuns > 0) {
shouldSendCompletionDirectly = false;
}
}
if (shouldSendCompletionDirectly) {
const completionThreadId =
completionDirectOrigin?.threadId != null && completionDirectOrigin.threadId !== ""
? String(completionDirectOrigin.threadId)
: undefined;
await callGateway({
method: "send",
params: {
channel: completionChannel,
to: completionTo,
accountId: completionDirectOrigin?.accountId,
threadId: completionThreadId,
sessionKey: canonicalRequesterSessionKey,
message: params.completionMessage,
idempotencyKey: params.directIdempotencyKey,
},
timeoutMs: 15_000,
});
return {
delivered: true,
path: "direct",
};
}
}
const directOrigin = normalizeDeliveryContext(params.directOrigin);
const threadId =
directOrigin?.threadId != null && directOrigin.threadId !== ""
? String(directOrigin.threadId)
: undefined;
await callGateway({
method: "agent",
params: {
sessionKey: canonicalRequesterSessionKey,
message: params.triggerMessage,
deliver: !params.requesterIsSubagent,
channel: params.requesterIsSubagent ? undefined : directOrigin?.channel,
accountId: params.requesterIsSubagent ? undefined : directOrigin?.accountId,
to: params.requesterIsSubagent ? undefined : directOrigin?.to,
threadId: params.requesterIsSubagent ? undefined : threadId,
idempotencyKey: params.directIdempotencyKey,
},
expectFinal: true,
timeoutMs: 15_000,
});
return {
delivered: true,
path: "direct",
};
} catch (err) {
return {
delivered: false,
path: "direct",
error: summarizeDeliveryError(err),
};
}
}
async function deliverSubagentAnnouncement(params: {
requesterSessionKey: string;
announceId?: string;
triggerMessage: string;
completionMessage?: string;
summaryLine?: string;
requesterOrigin?: DeliveryContext;
completionDirectOrigin?: DeliveryContext;
directOrigin?: DeliveryContext;
targetRequesterSessionKey: string;
requesterIsSubagent: boolean;
expectsCompletionMessage: boolean;
completionRouteMode?: "bound" | "fallback" | "hook";
spawnMode?: SpawnSubagentMode;
directIdempotencyKey: string;
}): Promise<SubagentAnnounceDeliveryResult> {
// Non-completion mode mirrors historical behavior: try queued/steered delivery first,
// then (only if not queued) attempt direct delivery.
if (!params.expectsCompletionMessage) {
const queueOutcome = await maybeQueueSubagentAnnounce({
requesterSessionKey: params.requesterSessionKey,
announceId: params.announceId,
triggerMessage: params.triggerMessage,
summaryLine: params.summaryLine,
requesterOrigin: params.requesterOrigin,
});
const queued = queueOutcomeToDeliveryResult(queueOutcome);
if (queued.delivered) {
return queued;
}
}
// Completion-mode uses direct send first so manual spawns can return immediately
// in the common ready-to-deliver case.
const direct = await sendSubagentAnnounceDirectly({
targetRequesterSessionKey: params.targetRequesterSessionKey,
triggerMessage: params.triggerMessage,
completionMessage: params.completionMessage,
directIdempotencyKey: params.directIdempotencyKey,
completionDirectOrigin: params.completionDirectOrigin,
completionRouteMode: params.completionRouteMode,
spawnMode: params.spawnMode,
directOrigin: params.directOrigin,
requesterIsSubagent: params.requesterIsSubagent,
expectsCompletionMessage: params.expectsCompletionMessage,
const directOrigin = normalizeDeliveryContext(params.directOrigin);
const threadId =
directOrigin?.threadId != null && directOrigin.threadId !== ""
? String(directOrigin.threadId)
: undefined;
await callGateway({
method: "agent",
params: {
sessionKey: canonicalRequesterSessionKey,
message: params.triggerMessage,
deliver: !params.requesterIsSubagent,
channel: params.requesterIsSubagent ? undefined : directOrigin?.channel,
accountId: params.requesterIsSubagent ? undefined : directOrigin?.accountId,
to: params.requesterIsSubagent ? undefined : directOrigin?.to,
threadId: params.requesterIsSubagent ? undefined : threadId,
idempotencyKey: params.directIdempotencyKey,
},
expectFinal: true,
timeoutMs: 15_000,
});
if (direct.delivered || !params.expectsCompletionMessage) {
return direct;
}
// If completion path failed direct delivery, try queueing as a fallback so the
// report can still be delivered once the requester session is idle.
const queueOutcome = await maybeQueueSubagentAnnounce({
requesterSessionKey: params.requesterSessionKey,
announceId: params.announceId,
triggerMessage: params.triggerMessage,
summaryLine: params.summaryLine,
requesterOrigin: params.requesterOrigin,
});
if (queueOutcome === "steered" || queueOutcome === "queued") {
return queueOutcomeToDeliveryResult(queueOutcome);
}
return direct;
}
function loadSessionEntryByKey(sessionKey: string) {
@@ -895,7 +539,6 @@ function buildAnnounceReplyInstruction(params: {
remainingActiveSubagentRuns: number;
requesterIsSubagent: boolean;
announceType: SubagentAnnounceType;
expectsCompletionMessage?: boolean;
}): string {
if (params.remainingActiveSubagentRuns > 0) {
const activeRunsLabel = params.remainingActiveSubagentRuns === 1 ? "run" : "runs";
@@ -904,9 +547,6 @@ function buildAnnounceReplyInstruction(params: {
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 user delivery. Convert the result above into your normal assistant voice and send that user-facing update now. Keep this internal context private (don't mention system/log/stats/session details or announce type).`;
}
return `A completed ${params.announceType} is ready for user delivery. Convert the result above into your normal assistant voice and send that user-facing update now. Keep this internal context private (don't mention system/log/stats/session details or announce type), and do not copy the system message verbatim. Reply ONLY: ${SILENT_REPLY_TOKEN} if this exact result was already delivered to the user in this same turn.`;
}
@@ -926,11 +566,10 @@ export async function runSubagentAnnounceFlow(params: {
label?: string;
outcome?: SubagentRunOutcome;
announceType?: SubagentAnnounceType;
expectsCompletionMessage?: boolean;
spawnMode?: SpawnSubagentMode;
[key: string]: unknown;
}): Promise<boolean> {
let didAnnounce = false;
const expectsCompletionMessage = params.expectsCompletionMessage === true;
let shouldDeleteChildSession = params.cleanup === "delete";
try {
let targetRequesterSessionKey = params.requesterSessionKey;
@@ -991,26 +630,22 @@ export async function runSubagentAnnounceFlow(params: {
outcome = { status: "timeout" };
}
}
reply = await readLatestSubagentOutput(params.childSessionKey);
reply = await readLatestAssistantReply({ sessionKey: params.childSessionKey });
}
if (!reply) {
reply = await readLatestSubagentOutput(params.childSessionKey);
reply = await readLatestAssistantReply({ sessionKey: params.childSessionKey });
}
if (!reply?.trim()) {
reply = await readLatestSubagentOutputWithRetry({
reply = await readLatestAssistantReplyWithRetry({
sessionKey: params.childSessionKey,
initialReply: reply,
maxWaitMs: params.timeoutMs,
});
}
if (
!expectsCompletionMessage &&
!reply?.trim() &&
childSessionId &&
isEmbeddedPiRunActive(childSessionId)
) {
if (!reply?.trim() && childSessionId && isEmbeddedPiRunActive(childSessionId)) {
// Avoid announcing "(no output)" while the child run is still producing output.
shouldDeleteChildSession = false;
return false;
@@ -1057,10 +692,8 @@ export async function runSubagentAnnounceFlow(params: {
// Build instructional message for main agent
const announceType = params.announceType ?? "subagent task";
const taskLabel = params.label || params.task || "task";
const subagentName = resolveAgentIdFromSessionKey(params.childSessionKey);
const announceSessionId = childSessionId || "unknown";
const findings = reply || "(no output)";
let completionMessage = "";
let triggerMessage = "";
let requesterIsSubagent = requesterDepth >= 1;
@@ -1116,19 +749,12 @@ export async function runSubagentAnnounceFlow(params: {
remainingActiveSubagentRuns,
requesterIsSubagent,
announceType,
expectsCompletionMessage,
});
const statsLine = await buildCompactAnnounceStatsLine({
sessionKey: params.childSessionKey,
startedAt: params.startedAt,
endedAt: params.endedAt,
});
completionMessage = buildCompletionDeliveryMessage({
findings,
subagentName,
spawnMode: params.spawnMode,
outcome,
});
const internalSummaryMessage = [
`[System Message] [sessionId: ${announceSessionId}] A ${announceType} "${taskLabel}" just ${statusLabel}.`,
"",
@@ -1150,50 +776,45 @@ export async function runSubagentAnnounceFlow(params: {
const { entry } = loadRequesterSessionEntry(targetRequesterSessionKey);
directOrigin = resolveAnnounceOrigin(entry, targetRequesterOrigin);
}
const completionResolution =
expectsCompletionMessage && !requesterIsSubagent
? await resolveSubagentCompletionOrigin({
childSessionKey: params.childSessionKey,
requesterSessionKey: targetRequesterSessionKey,
requesterOrigin: directOrigin,
childRunId: params.childRunId,
spawnMode: params.spawnMode,
expectsCompletionMessage,
})
: {
origin: targetRequesterOrigin,
routeMode: "fallback" as const,
};
const completionDirectOrigin = completionResolution.origin;
let deliveryOrigin = directOrigin;
if (!requesterIsSubagent) {
const resolvedOrigin = await resolveSubagentDeliveryOrigin({
childSessionKey: params.childSessionKey,
requesterSessionKey: targetRequesterSessionKey,
requesterOrigin: directOrigin,
childRunId: params.childRunId,
spawnMode: params.spawnMode,
});
deliveryOrigin = resolvedOrigin ?? deliveryOrigin;
}
const queueOutcome = await maybeQueueSubagentAnnounce({
requesterSessionKey: targetRequesterSessionKey,
announceId,
triggerMessage,
summaryLine: taskLabel,
requesterOrigin: deliveryOrigin,
});
if (queueOutcome === "steered") {
didAnnounce = true;
return true;
}
if (queueOutcome === "queued") {
didAnnounce = true;
return true;
}
// Use a deterministic idempotency key so the gateway dedup cache
// catches duplicates if this announce is also queued by the gateway-
// level message queue while the main session is busy (#17122).
const directIdempotencyKey = buildAnnounceIdempotencyKey(announceId);
const delivery = await deliverSubagentAnnouncement({
requesterSessionKey: targetRequesterSessionKey,
announceId,
triggerMessage,
completionMessage,
summaryLine: taskLabel,
requesterOrigin:
expectsCompletionMessage && !requesterIsSubagent
? completionDirectOrigin
: targetRequesterOrigin,
completionDirectOrigin,
directOrigin,
await sendSubagentAnnounceDirectly({
targetRequesterSessionKey,
requesterIsSubagent,
expectsCompletionMessage: expectsCompletionMessage,
completionRouteMode: completionResolution.routeMode,
spawnMode: params.spawnMode,
triggerMessage,
directIdempotencyKey,
directOrigin: deliveryOrigin,
requesterIsSubagent,
});
didAnnounce = delivery.delivered;
if (!delivery.delivered && delivery.path === "direct" && delivery.error) {
defaultRuntime.error?.(
`Subagent completion direct announce failed for run ${params.childRunId}: ${delivery.error}`,
);
}
didAnnounce = true;
} catch (err) {
defaultRuntime.error?.(`Subagent announce failed: ${String(err)}`);
// Best-effort follow-ups; ignore failures to avoid breaking the caller response.

18
src/agents/subagent-registry.ts
View File

@@ -167,12 +167,7 @@ async function completeSubagentRun(params: {
entry,
reason: params.reason,
});
const shouldDeferEndedHook =
shouldEmitEndedHook &&
params.triggerCleanup &&
entry.expectsCompletionMessage === true &&
!suppressedForSteerRestart;
if (!shouldDeferEndedHook && shouldEmitEndedHook) {
if (shouldEmitEndedHook) {
await emitSubagentEndedHookForRun({
entry,
reason: params.reason,
@@ -210,7 +205,6 @@ function startSubagentAnnounceCleanupFlow(runId: string, entry: SubagentRunRecor
label: entry.label,
outcome: entry.outcome,
spawnMode: entry.spawnMode,
expectsCompletionMessage: entry.expectsCompletionMessage,
}).then((didAnnounce) => {
void finalizeSubagentCleanup(runId, entry.cleanup, didAnnounce);
});
@@ -245,11 +239,7 @@ function resumeSubagentRun(runId: string) {
const now = Date.now();
const delayMs = resolveAnnounceRetryDelayMs(entry.announceRetryCount ?? 0);
const earliestRetryAt = (entry.lastAnnounceRetryAt ?? 0) + delayMs;
if (
entry.expectsCompletionMessage === true &&
entry.lastAnnounceRetryAt &&
now < earliestRetryAt
) {
if (entry.lastAnnounceRetryAt && now < earliestRetryAt) {
const waitMs = Math.max(1, earliestRetryAt - now);
setTimeout(() => {
resumeSubagentRun(runId);
@@ -493,7 +483,6 @@ async function emitCompletionEndedHookIfNeeded(
reason: SubagentLifecycleEndedReason,
) {
if (
entry.expectsCompletionMessage === true &&
shouldEmitEndedHookForRun({
entry,
reason,
@@ -679,8 +668,8 @@ export function registerSubagentRun(params: {
label?: string;
model?: string;
runTimeoutSeconds?: number;
expectsCompletionMessage?: boolean;
spawnMode?: "run" | "session";
[key: string]: unknown;
}) {
const now = Date.now();
const cfg = loadConfig();
@@ -699,7 +688,6 @@ export function registerSubagentRun(params: {
requesterDisplayKey: params.requesterDisplayKey,
task: params.task,
cleanup: params.cleanup,
expectsCompletionMessage: params.expectsCompletionMessage,
spawnMode,
label: params.label,
model: params.model,