mirror of
https://github.com/openclaw/openclaw.git
synced 2026-04-18 19:37:28 +00:00
fix(agents): make manual subagent completion announce deterministic
This commit is contained in:
@@ -8,6 +8,7 @@ type RequesterResolution = {
|
||||
} | null;
|
||||
|
||||
const agentSpy = vi.fn(async (_req: AgentCallRequest) => ({ runId: "run-main", status: "ok" }));
|
||||
const sendSpy = vi.fn(async (_req: AgentCallRequest) => ({ runId: "send-main", status: "ok" }));
|
||||
const sessionsDeleteSpy = vi.fn((_req: AgentCallRequest) => undefined);
|
||||
const readLatestAssistantReplyMock = vi.fn(
|
||||
async (_sessionKey?: string): Promise<string | undefined> => "raw subagent reply",
|
||||
@@ -64,6 +65,9 @@ vi.mock("../gateway/call.js", () => ({
|
||||
if (typed.method === "agent") {
|
||||
return await agentSpy(typed);
|
||||
}
|
||||
if (typed.method === "send") {
|
||||
return await sendSpy(typed);
|
||||
}
|
||||
if (typed.method === "agent.wait") {
|
||||
return { status: "error", startedAt: 10, endedAt: 20, error: "boom" };
|
||||
}
|
||||
@@ -109,6 +113,7 @@ vi.mock("../config/config.js", async (importOriginal) => {
|
||||
describe("subagent announce formatting", () => {
|
||||
beforeEach(() => {
|
||||
agentSpy.mockClear();
|
||||
sendSpy.mockClear();
|
||||
sessionsDeleteSpy.mockClear();
|
||||
embeddedRunMock.isEmbeddedPiRunActive.mockReset().mockReturnValue(false);
|
||||
embeddedRunMock.isEmbeddedPiRunStreaming.mockReset().mockReturnValue(false);
|
||||
@@ -329,6 +334,85 @@ describe("subagent announce formatting", () => {
|
||||
expect(msg).toContain("step-139");
|
||||
});
|
||||
|
||||
it("sends deterministic completion message directly for manual spawn completion", async () => {
|
||||
const { runSubagentAnnounceFlow } = await import("./subagent-announce.js");
|
||||
sessionStore = {
|
||||
"agent:main:subagent:test": {
|
||||
sessionId: "child-session-direct",
|
||||
inputTokens: 12,
|
||||
outputTokens: 34,
|
||||
totalTokens: 46,
|
||||
},
|
||||
"agent:main:main": {
|
||||
sessionId: "requester-session",
|
||||
},
|
||||
};
|
||||
chatHistoryMock.mockResolvedValueOnce({
|
||||
messages: [{ role: "assistant", content: [{ type: "text", text: "final answer: 2" }] }],
|
||||
});
|
||||
|
||||
const didAnnounce = await runSubagentAnnounceFlow({
|
||||
childSessionKey: "agent:main:subagent:test",
|
||||
childRunId: "run-direct-completion",
|
||||
requesterSessionKey: "agent:main:main",
|
||||
requesterDisplayKey: "main",
|
||||
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
|
||||
...defaultOutcomeAnnounce,
|
||||
expectsCompletionMessage: true,
|
||||
});
|
||||
|
||||
expect(didAnnounce).toBe(true);
|
||||
expect(sendSpy).toHaveBeenCalledTimes(1);
|
||||
expect(agentSpy).not.toHaveBeenCalled();
|
||||
const call = sendSpy.mock.calls[0]?.[0] as { params?: Record<string, unknown> };
|
||||
const rawMessage = call?.params?.message;
|
||||
const msg = typeof rawMessage === "string" ? rawMessage : "";
|
||||
expect(call?.params?.channel).toBe("discord");
|
||||
expect(call?.params?.to).toBe("channel:12345");
|
||||
expect(call?.params?.sessionKey).toBe("agent:main:main");
|
||||
expect(msg).toContain("[System Message]");
|
||||
expect(msg).toContain('subagent task "do thing"');
|
||||
expect(msg).toContain("Result:");
|
||||
expect(msg).toContain("final answer: 2");
|
||||
expect(msg).toContain("Stats:");
|
||||
expect(msg).not.toContain("Convert the result above into your normal assistant voice");
|
||||
});
|
||||
|
||||
it("ignores stale session thread hints for manual completion direct-send", async () => {
|
||||
const { runSubagentAnnounceFlow } = await import("./subagent-announce.js");
|
||||
sessionStore = {
|
||||
"agent:main:subagent:test": {
|
||||
sessionId: "child-session-direct-thread",
|
||||
},
|
||||
"agent:main:main": {
|
||||
sessionId: "requester-session-thread",
|
||||
lastChannel: "discord",
|
||||
lastTo: "channel:stale",
|
||||
lastThreadId: 42,
|
||||
},
|
||||
};
|
||||
chatHistoryMock.mockResolvedValueOnce({
|
||||
messages: [{ role: "assistant", content: [{ type: "text", text: "done" }] }],
|
||||
});
|
||||
|
||||
const didAnnounce = await runSubagentAnnounceFlow({
|
||||
childSessionKey: "agent:main:subagent:test",
|
||||
childRunId: "run-direct-stale-thread",
|
||||
requesterSessionKey: "agent:main:main",
|
||||
requesterDisplayKey: "main",
|
||||
requesterOrigin: { channel: "discord", to: "channel:12345", accountId: "acct-1" },
|
||||
...defaultOutcomeAnnounce,
|
||||
expectsCompletionMessage: true,
|
||||
});
|
||||
|
||||
expect(didAnnounce).toBe(true);
|
||||
expect(sendSpy).toHaveBeenCalledTimes(1);
|
||||
expect(agentSpy).not.toHaveBeenCalled();
|
||||
const call = sendSpy.mock.calls[0]?.[0] as { params?: Record<string, unknown> };
|
||||
expect(call?.params?.channel).toBe("discord");
|
||||
expect(call?.params?.to).toBe("channel:12345");
|
||||
});
|
||||
|
||||
it("steers announcements into an active run when queue mode is steer", async () => {
|
||||
const { runSubagentAnnounceFlow } = await import("./subagent-announce.js");
|
||||
embeddedRunMock.isEmbeddedPiRunActive.mockReturnValue(true);
|
||||
|
||||
@@ -404,26 +404,68 @@ function queueOutcomeToDeliveryResult(
|
||||
async function sendSubagentAnnounceDirectly(params: {
|
||||
targetRequesterSessionKey: string;
|
||||
triggerMessage: string;
|
||||
completionMessage?: string;
|
||||
expectsCompletionMessage: boolean;
|
||||
directIdempotencyKey: string;
|
||||
completionDirectOrigin?: DeliveryContext;
|
||||
directOrigin?: DeliveryContext;
|
||||
requesterIsSubagent: boolean;
|
||||
}): Promise<SubagentAnnounceDeliveryResult> {
|
||||
try {
|
||||
const completionDirectOrigin = normalizeDeliveryContext(params.completionDirectOrigin);
|
||||
const completionChannel =
|
||||
typeof completionDirectOrigin?.channel === "string"
|
||||
? completionDirectOrigin.channel.trim()
|
||||
: "";
|
||||
const completionTo =
|
||||
typeof completionDirectOrigin?.to === "string" ? completionDirectOrigin.to.trim() : "";
|
||||
const completionHasThreadHint =
|
||||
completionDirectOrigin?.threadId != null &&
|
||||
String(completionDirectOrigin.threadId).trim() !== "";
|
||||
const hasCompletionDirectTarget =
|
||||
!params.requesterIsSubagent && Boolean(completionChannel) && Boolean(completionTo);
|
||||
|
||||
if (
|
||||
params.expectsCompletionMessage &&
|
||||
hasCompletionDirectTarget &&
|
||||
!completionHasThreadHint &&
|
||||
params.completionMessage?.trim()
|
||||
) {
|
||||
await callGateway({
|
||||
method: "send",
|
||||
params: {
|
||||
channel: completionChannel,
|
||||
to: completionTo,
|
||||
accountId: completionDirectOrigin?.accountId,
|
||||
sessionKey: params.targetRequesterSessionKey,
|
||||
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: params.targetRequesterSessionKey,
|
||||
message: params.triggerMessage,
|
||||
deliver: !params.requesterIsSubagent,
|
||||
channel: params.requesterIsSubagent ? undefined : params.directOrigin?.channel,
|
||||
accountId: params.requesterIsSubagent ? undefined : params.directOrigin?.accountId,
|
||||
to: params.requesterIsSubagent ? undefined : params.directOrigin?.to,
|
||||
threadId:
|
||||
!params.requesterIsSubagent &&
|
||||
params.directOrigin?.threadId != null &&
|
||||
params.directOrigin.threadId !== ""
|
||||
? String(params.directOrigin.threadId)
|
||||
: undefined,
|
||||
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,
|
||||
@@ -443,12 +485,14 @@ async function sendSubagentAnnounceDirectly(params: {
|
||||
}
|
||||
}
|
||||
|
||||
async function deliverSubagentCompletionAnnouncement(params: {
|
||||
async function deliverSubagentAnnouncement(params: {
|
||||
requesterSessionKey: string;
|
||||
announceId?: string;
|
||||
triggerMessage: string;
|
||||
completionMessage?: string;
|
||||
summaryLine?: string;
|
||||
requesterOrigin?: DeliveryContext;
|
||||
completionDirectOrigin?: DeliveryContext;
|
||||
directOrigin?: DeliveryContext;
|
||||
targetRequesterSessionKey: string;
|
||||
requesterIsSubagent: boolean;
|
||||
@@ -476,7 +520,10 @@ async function deliverSubagentCompletionAnnouncement(params: {
|
||||
const direct = await sendSubagentAnnounceDirectly({
|
||||
targetRequesterSessionKey: params.targetRequesterSessionKey,
|
||||
triggerMessage: params.triggerMessage,
|
||||
completionMessage: params.completionMessage,
|
||||
expectsCompletionMessage: params.expectsCompletionMessage,
|
||||
directIdempotencyKey: params.directIdempotencyKey,
|
||||
completionDirectOrigin: params.completionDirectOrigin,
|
||||
directOrigin: params.directOrigin,
|
||||
requesterIsSubagent: params.requesterIsSubagent,
|
||||
});
|
||||
@@ -761,6 +808,7 @@ export async function runSubagentAnnounceFlow(params: {
|
||||
const taskLabel = params.label || params.task || "task";
|
||||
const announceSessionId = childSessionId || "unknown";
|
||||
const findings = reply || "(no output)";
|
||||
let completionMessage = "";
|
||||
let triggerMessage = "";
|
||||
|
||||
let requesterDepth = getSubagentDepthFromSessionStore(targetRequesterSessionKey);
|
||||
@@ -824,39 +872,20 @@ export async function runSubagentAnnounceFlow(params: {
|
||||
startedAt: params.startedAt,
|
||||
endedAt: params.endedAt,
|
||||
});
|
||||
triggerMessage = [
|
||||
completionMessage = [
|
||||
`[System Message] [sessionId: ${announceSessionId}] A ${announceType} "${taskLabel}" just ${statusLabel}.`,
|
||||
"",
|
||||
"Result:",
|
||||
findings,
|
||||
"",
|
||||
statsLine,
|
||||
"",
|
||||
replyInstruction,
|
||||
].join("\n");
|
||||
triggerMessage = [completionMessage, "", replyInstruction].join("\n");
|
||||
|
||||
const announceId = buildAnnounceIdFromChildRun({
|
||||
childSessionKey: params.childSessionKey,
|
||||
childRunId: params.childRunId,
|
||||
});
|
||||
if (!expectsCompletionMessage) {
|
||||
const queued = await maybeQueueSubagentAnnounce({
|
||||
requesterSessionKey: targetRequesterSessionKey,
|
||||
announceId,
|
||||
triggerMessage,
|
||||
summaryLine: taskLabel,
|
||||
requesterOrigin: targetRequesterOrigin,
|
||||
});
|
||||
if (queued === "steered") {
|
||||
didAnnounce = true;
|
||||
return true;
|
||||
}
|
||||
if (queued === "queued") {
|
||||
didAnnounce = true;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
// 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;
|
||||
@@ -868,12 +897,14 @@ export async function runSubagentAnnounceFlow(params: {
|
||||
// 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 deliverSubagentCompletionAnnouncement({
|
||||
const delivery = await deliverSubagentAnnouncement({
|
||||
requesterSessionKey: targetRequesterSessionKey,
|
||||
announceId,
|
||||
triggerMessage,
|
||||
completionMessage,
|
||||
summaryLine: taskLabel,
|
||||
requesterOrigin: targetRequesterOrigin,
|
||||
completionDirectOrigin: targetRequesterOrigin,
|
||||
directOrigin,
|
||||
targetRequesterSessionKey,
|
||||
requesterIsSubagent,
|
||||
|
||||
Reference in New Issue
Block a user