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 a83c01d0ac1d..f43f7850a9e4 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.test.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.test.ts @@ -467,6 +467,75 @@ describe("qa mock openai server", () => { expect(responseBody).toContain('"text":"QA-FINAL-ONLY-STREAMING-OK"'); }); + it("plans sessions_send for the A2A message-tool mirror proof scenario", async () => { + const server = await startMockServer(); + const prompt = + 'qa a2a message-tool mirror check. sessionKey="agent:qa:a2a-target". exact marker: `QA-A2A-MIRROR-OK`'; + + const toolPlan = await expectResponsesJson(server, { + stream: false, + model: "gpt-5.5", + tools: [{ type: "function", name: "sessions_send" }], + input: [makeUserInput(prompt)], + }); + + const args = outputToolArgs(toolPlan); + expect(outputItem(toolPlan).type).toBe("function_call"); + expect(outputItem(toolPlan).name).toBe("sessions_send"); + expect(args).toMatchObject({ + sessionKey: "agent:qa:a2a-target", + timeoutSeconds: 0, + }); + expect(String(args.message)).toContain("qa group visible reply tool check"); + expect(String(args.message)).toContain("QA-A2A-MIRROR-OK"); + + const debugResponse = await fetch(`${server.baseUrl}/debug/last-request`); + expect(debugResponse.status).toBe(200); + const debugPayload = requireRecord(await debugResponse.json(), "debug request"); + expect(debugPayload.plannedToolName).toBe("sessions_send"); + expect(debugPayload.plannedToolArgs).toMatchObject({ + sessionKey: "agent:qa:a2a-target", + timeoutSeconds: 0, + }); + + const final = await expectResponsesJson(server, { + stream: false, + model: "gpt-5.5", + tools: [{ type: "function", name: "sessions_send" }], + input: [ + makeUserInput(prompt), + { + type: "function_call_output", + call_id: "call_mock_sessions_send_fixture", + output: JSON.stringify({ status: "accepted", delivery: { mode: "announce" } }), + }, + ], + }); + expect(outputText(final)).toBe(""); + + const targetToolPlan = await expectResponsesJson(server, { + stream: false, + model: "gpt-5.5", + tools: [ + { type: "function", name: "sessions_send" }, + { type: "function", name: "message" }, + ], + input: [ + makeUserInput(prompt), + makeUserInput( + "qa group visible reply tool check. Use the visible room reply path. exact marker: `QA-A2A-MIRROR-OK`", + ), + ], + }); + + expect(outputItem(targetToolPlan).type).toBe("function_call"); + expect(outputItem(targetToolPlan).name).toBe("message"); + expect(outputToolArgs(targetToolPlan)).toMatchObject({ + action: "send", + message: "QA-A2A-MIRROR-OK", + }); + }); + it("emits deterministic text deltas for generic streaming QA prompts", async () => { const server = await startMockServer(); diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index 4897e368ea32..1a608203da66 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.ts @@ -169,13 +169,13 @@ const QA_BLOCK_STREAMING_PROMPT_RE = /block streaming qa check/i; const QA_TOOL_PROGRESS_ERROR_PROMPT_RE = /tool progress error qa check/i; const QA_TOOL_PROGRESS_PROMPT_RE = /tool progress qa check/i; const QA_GROUP_VISIBLE_REPLY_TOOL_PROMPT_RE = /qa group visible reply tool check/i; +const QA_A2A_MESSAGE_TOOL_MIRROR_PROMPT_RE = /qa a2a message-tool mirror check/i; const QA_GROUP_MESSAGE_UNAVAILABLE_FALLBACK_PROMPT_RE = /qa group message unavailable fallback check/i; const QA_STRANDED_FINAL_RECOVERY_PROMPT_RE = /qa stranded final recovery check/i; const QA_STRANDED_FINAL_RETRY_FAILURE_PROMPT_RE = /qa stranded final retry failure check/i; const QA_STRANDED_FINAL_RETRY_PROMPT_RE = /you did not call message\(action=send\)/i; -const QA_STRANDED_FINAL_RETRY_FAILURE_MARKER = - "QA-STRANDED-RETRY-FAIL-RAW"; +const QA_STRANDED_FINAL_RETRY_FAILURE_MARKER = "QA-STRANDED-RETRY-FAIL-RAW"; const QA_TELEGRAM_CURRENT_SESSION_STATUS_PROMPT_RE = /telegram current session_status qa check/i; const QA_TELEGRAM_STREAM_SINGLE_MARKER = "QA-TELEGRAM-STREAM-SINGLE-OK"; const QA_TELEGRAM_LONG_FINAL_THREE_CHUNK_PROMPT_RE = /telegram long final three chunk qa check/i; @@ -1292,6 +1292,26 @@ function buildExplicitSessionsSpawnArgs(text: string): Record | }; } +function buildQaA2aMessageToolMirrorSessionsSendArgs(text: string): Record | null { + if (!QA_A2A_MESSAGE_TOOL_MIRROR_PROMPT_RE.test(text)) { + return null; + } + const sessionKey = + extractQuotedToolArg(text, "sessionKey") ?? extractBareToolArg(text, "sessionKey"); + if (!sessionKey) { + return null; + } + const marker = + extractExactMarkerDirective(text) ?? + extractExactReplyDirective(text) ?? + "QA-A2A-MESSAGE-TOOL-MIRROR-OK"; + return { + sessionKey, + message: `qa group visible reply tool check. Use the visible room reply path. exact marker: \`${marker}\``, + timeoutSeconds: 0, + }; +} + function extractToolErrorForNamedCall(params: { input: ResponsesInputItem[]; name: string; @@ -2569,6 +2589,15 @@ async function buildResponsesPayload( } return buildAssistantEvents(buildStrandedFinalRecoveryText()); } + if (QA_A2A_MESSAGE_TOOL_MIRROR_PROMPT_RE.test(prompt)) { + if (toolOutput) { + return buildAssistantEvents(""); + } + const sessionsSendArgs = buildQaA2aMessageToolMirrorSessionsSendArgs(prompt); + if (sessionsSendArgs && hasDeclaredTool(body, "sessions_send")) { + return buildToolCallEventsWithArgs("sessions_send", sessionsSendArgs); + } + } if (QA_GROUP_VISIBLE_REPLY_TOOL_PROMPT_RE.test(allInputText)) { const marker = exactMarkerDirective ?? exactReplyDirective ?? "QA-GROUP-TOOL-OK"; if (!toolOutput && hasDeclaredTool(body, "message")) { diff --git a/qa/scenarios/channels/a2a-message-tool-mirror-dedupe.yaml b/qa/scenarios/channels/a2a-message-tool-mirror-dedupe.yaml new file mode 100644 index 000000000000..dc9c7d82e965 --- /dev/null +++ b/qa/scenarios/channels/a2a-message-tool-mirror-dedupe.yaml @@ -0,0 +1,140 @@ +title: A2A message-tool mirror dedupe + +scenario: + id: a2a-message-tool-mirror-dedupe + surface: channel + coverage: + primary: + - runtime.delivery + secondary: + - channels.qa-channel + - tools.message + objective: Verify a sessions_send A2A turn whose nested target run replies through message(action=send) delivers once to the requester channel and does not re-announce the delivery-mirror transcript row. + gatewayConfigPatch: + messages: + groupChat: + visibleReplies: message_tool + session: + agentToAgent: + maxPingPongTurns: 0 + tools: + sessions: + visibility: all + agentToAgent: + enabled: true + successCriteria: + - Source agent receives a synthetic qa-channel group turn. + - Source agent calls sessions_send against its real qa-channel group session. + - Target run calls message(action=send) under the source-reply path. + - The requester group sees the marker exactly once, with no duplicate during the post-delivery window. + docsRefs: + - docs/channels/qa-channel.md + - docs/concepts/qa-e2e-automation.md + codeRefs: + - src/agents/tools/sessions-send-tool.ts + - src/agents/tools/sessions-send-tool.a2a.ts + - src/agents/run-wait.ts + - src/shared/transcript-only-openclaw-assistant.ts + execution: + kind: flow + summary: Run a real QA Gateway/qa-channel A2A source-reply turn and assert the message-tool delivery mirror is not announced again. + config: + requiredProviderMode: mock-openai + requiredChannelDriver: qa-channel + conversationId: qa-a2a-mirror-room + conversationTitle: QA A2A Mirror Room + promptSnippet: qa a2a message-tool mirror check + targetPromptSnippet: qa group visible reply tool check + expectedMarker: QA-A2A-MESSAGE-TOOL-MIRROR-OK + duplicateWindowMs: 8000 + +flow: + steps: + - name: delivers target message-tool reply once without mirror re-announce + actions: + - assert: + expr: "env.providerMode === config.requiredProviderMode" + message: this seeded scenario is mock-openai only + - call: waitForGatewayHealthy + args: + - ref: env + - 60000 + - call: waitForQaChannelReady + args: + - ref: env + - 60000 + - call: reset + - set: startIndex + value: + expr: state.getSnapshot().messages.length + - set: targetSessionKey + value: + expr: "buildAgentSessionKey({ agentId: 'qa', channel: 'qa-channel', accountId: 'default', peer: { kind: 'group', id: `group:${config.conversationId}` }, dmScope: env.cfg.session?.dmScope, identityLinks: env.cfg.session?.identityLinks })" + - set: requestCountBefore + value: + expr: "env.mock ? (await fetchJson(`${env.mock.baseUrl}/debug/requests`)).length : 0" + - call: state.addInboundMessage + args: + - conversation: + id: + expr: config.conversationId + kind: group + title: + expr: config.conversationTitle + senderId: alice + senderName: Alice + text: + expr: "`@openclaw ${config.promptSnippet}. sessionKey=\"${targetSessionKey}\". Use sessions_send once with timeoutSeconds=0. The target must reply visibly with exact marker: \\`${config.expectedMarker}\\``" + - call: waitForCondition + saveAs: sourceSessionsSendRequest + args: + - lambda: + async: true + params: [] + expr: "env.mock ? (await fetchJson(`${env.mock.baseUrl}/debug/requests`)).slice(requestCountBefore).find((request) => String(request.allInputText ?? '').includes(config.promptSnippet) && request.plannedToolName === 'sessions_send' && request.plannedToolArgs?.sessionKey === targetSessionKey && request.plannedToolArgs?.timeoutSeconds === 0) : true" + - expr: liveTurnTimeoutMs(env, 60000) + - 500 + - call: waitForCondition + saveAs: targetMessageToolRequest + args: + - lambda: + async: true + params: [] + expr: "env.mock ? (await fetchJson(`${env.mock.baseUrl}/debug/requests`)).slice(requestCountBefore).find((request) => String(request.allInputText ?? '').includes(config.targetPromptSnippet) && request.plannedToolName === 'message' && request.plannedToolArgs?.action === 'send' && request.plannedToolArgs?.message === config.expectedMarker) : true" + - expr: liveTurnTimeoutMs(env, 90000) + - 500 + - call: waitForOutboundMessage + saveAs: outbound + args: + - ref: state + - lambda: + params: [candidate] + expr: "candidate.conversation.id === config.conversationId && candidate.conversation.kind === 'group' && candidate.direction === 'outbound' && String(candidate.text ?? '').includes(config.expectedMarker)" + - expr: liveTurnTimeoutMs(env, 90000) + - sinceIndex: + ref: startIndex + - call: sleep + args: + - expr: config.duplicateWindowMs + - set: snapshot + value: + expr: state.getSnapshot() + - set: matchingOutbound + value: + expr: "snapshot.messages.slice(startIndex).filter((message) => message.direction === 'outbound' && message.conversation.id === config.conversationId && message.conversation.kind === 'group' && String(message.text ?? '').includes(config.expectedMarker))" + - assert: + expr: matchingOutbound.length === 1 + message: + expr: "`expected exactly one requester-visible A2A marker after duplicate window, saw ${matchingOutbound.length}; transcript=${formatTransportTranscript(state, { conversationId: config.conversationId })}`" + - set: scenarioRequests + value: + expr: "env.mock ? (await fetchJson(`${env.mock.baseUrl}/debug/requests`)).slice(requestCountBefore).map((request) => ({ prompt: String(request.prompt ?? '').slice(0, 220), plannedToolName: request.plannedToolName ?? null, plannedToolArgs: request.plannedToolArgs ?? null, toolOutput: request.toolOutput ? String(request.toolOutput).slice(0, 220) : null })) : []" + - assert: + expr: "!env.mock || scenarioRequests.filter((request) => request.plannedToolName === 'sessions_send' && request.plannedToolArgs?.sessionKey === targetSessionKey && request.plannedToolArgs?.timeoutSeconds === 0).length === 1" + message: + expr: "`expected exactly one source sessions_send plan for the target session; requests=${JSON.stringify(scenarioRequests)}`" + - assert: + expr: "!env.mock || scenarioRequests.filter((request) => request.plannedToolName === 'message' && request.plannedToolArgs?.action === 'send' && request.plannedToolArgs?.message === config.expectedMarker).length === 1" + message: + expr: "`expected exactly one target message(action=send) plan for the marker; requests=${JSON.stringify(scenarioRequests)}`" + detailsExpr: "`outbound=${outbound.conversation.kind}:${outbound.conversation.id}:${outbound.text}; sourceTool=${JSON.stringify(sourceSessionsSendRequest?.plannedToolArgs ?? {})}; targetTool=${JSON.stringify(targetMessageToolRequest?.plannedToolArgs ?? {})}; duplicateWindowMs=${config.duplicateWindowMs}`" diff --git a/src/agents/run-wait.test.ts b/src/agents/run-wait.test.ts index 21bfa71ab1db..9fce135bd367 100644 --- a/src/agents/run-wait.test.ts +++ b/src/agents/run-wait.test.ts @@ -76,7 +76,7 @@ function expectAgentWaitRequest( describe("readLatestAssistantReply", () => { beforeEach(() => { - callGatewayMock.mockClear(); + callGatewayMock.mockReset(); testing.setDepsForTest({ callGateway: async (opts) => await callGatewayMock(opts), }); @@ -117,6 +117,90 @@ describe("readLatestAssistantReply", () => { expect(result).toBe("older output"); }); + it("skips trailing transcript-only OpenClaw assistant mirrors for normal latest-reply reads", async () => { + callGatewayMock.mockResolvedValue({ + messages: [ + { + role: "assistant", + content: [{ type: "text", text: "real worker reply" }], + timestamp: 10, + }, + { + role: "assistant", + content: [{ type: "text", text: "already delivered through message tool" }], + openclawMessageToolMirror: { + toolName: "message", + toolCallId: "call-message-send", + }, + timestamp: 11, + }, + { + role: "assistant", + provider: "openclaw", + model: "gateway-injected", + content: [{ type: "text", text: "gateway notice" }], + timestamp: 12, + }, + ], + }); + + const result = await readLatestAssistantReply({ sessionKey: "agent:main:child" }); + + expect(result).toBe("real worker reply"); + }); + + it("skips trailing inter-session input rows for normal latest-reply reads", async () => { + callGatewayMock.mockResolvedValue({ + messages: [ + { + role: "assistant", + content: [{ type: "text", text: "older worker reply" }], + timestamp: 10, + }, + { + role: "assistant", + content: [{ type: "text", text: "forwarded sessions_send prompt" }], + provenance: { + kind: "inter_session", + sourceSessionKey: "agent:main:source", + sourceTool: "sessions_send", + }, + timestamp: 11, + }, + ], + }); + + const result = await readLatestAssistantReply({ sessionKey: "agent:main:target" }); + + expect(result).toBe("older worker reply"); + }); + + it("stops at trailing transcript artifacts for waited reply extraction", async () => { + callGatewayMock.mockResolvedValue({ + messages: [ + { + role: "assistant", + content: [{ type: "text", text: "older worker reply" }], + timestamp: 10, + }, + { + role: "assistant", + provider: "openclaw", + model: "gateway-injected", + content: [{ type: "text", text: "gateway notice" }], + timestamp: 11, + }, + ], + }); + + const result = await readLatestAssistantReplySnapshot({ + sessionKey: "agent:main:target", + stopAtTranscriptArtifact: true, + }); + + expect(result).toEqual({}); + }); + it("returns assistant fingerprints for delta comparisons", async () => { callGatewayMock.mockResolvedValue({ messages: [ @@ -194,7 +278,7 @@ describe("readLatestAssistantReply", () => { describe("waitForAgentRun", () => { beforeEach(() => { - callGatewayMock.mockClear(); + callGatewayMock.mockReset(); testing.setDepsForTest({ callGateway: async (opts) => await callGatewayMock(opts), }); @@ -380,7 +464,7 @@ describe("waitForAgentRun", () => { describe("waitForAgentRunAndReadUpdatedAssistantReply", () => { beforeEach(() => { - callGatewayMock.mockClear(); + callGatewayMock.mockReset(); testing.setDepsForTest({ callGateway: async (opts) => await callGatewayMock(opts), }); @@ -416,6 +500,364 @@ describe("waitForAgentRunAndReadUpdatedAssistantReply", () => { }); }); + it("returns undefined when a text-only baseline matches the latest assistant reply", async () => { + callGatewayMock + .mockResolvedValueOnce({ + status: "ok", + }) + .mockResolvedValueOnce({ + messages: [ + { + role: "assistant", + content: [{ type: "text", text: "same reply" }], + timestamp: 42, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-text-baseline", + sessionKey: "agent:main:child", + timeoutMs: 1_000, + baseline: { + text: "same reply", + }, + }); + + expect(result).toEqual({ + status: "ok", + replyText: undefined, + }); + }); + + it("does not treat a message-tool delivery mirror as a new waited reply", async () => { + const baselineMessage = { + role: "assistant", + content: [{ type: "text", text: "previous real reply" }], + timestamp: 41, + }; + callGatewayMock + .mockResolvedValueOnce({ + status: "ok", + }) + .mockResolvedValueOnce({ + messages: [ + baselineMessage, + { + role: "assistant", + provider: "openclaw", + model: "delivery-mirror", + content: [{ type: "text", text: "already delivered source reply" }], + timestamp: 42, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-source-reply", + sessionKey: "agent:main:child", + timeoutMs: 1_000, + baseline: { + text: "previous real reply", + fingerprint: JSON.stringify(baselineMessage), + }, + }); + + expect(result).toEqual({ + status: "ok", + replyText: undefined, + }); + }); + + it("does not treat a projected message-tool mirror as a new waited reply", async () => { + const baselineMessage = { + role: "assistant", + content: [{ type: "text", text: "previous real reply" }], + timestamp: 41, + }; + callGatewayMock + .mockResolvedValueOnce({ + status: "ok", + }) + .mockResolvedValueOnce({ + messages: [ + baselineMessage, + { + role: "assistant", + content: [{ type: "text", text: "already delivered source reply" }], + openclawMessageToolMirror: { + toolName: "message", + toolCallId: "call-message-send", + }, + timestamp: 42, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-projected-source-reply", + sessionKey: "agent:main:child", + timeoutMs: 1_000, + baseline: { + text: "previous real reply", + fingerprint: JSON.stringify(baselineMessage), + }, + }); + + expect(result).toEqual({ + status: "ok", + replyText: undefined, + }); + }); + + it("returns a projected message-tool reply held for outer A2A delivery", async () => { + callGatewayMock.mockResolvedValueOnce({ status: "ok" }).mockResolvedValueOnce({ + messages: [ + { + role: "assistant", + provenance: { + kind: "inter_session", + sourceSessionKey: "agent:main:source", + sourceTool: "sessions_send", + }, + content: [{ type: "text", text: "forwarded request" }], + __openclaw: { seq: 41 }, + timestamp: 41, + }, + { + role: "assistant", + content: [{ type: "text", text: "source reply awaiting delivery" }], + openclawMessageToolMirror: { + toolName: "message", + toolCallId: "call-message-send", + sourceReplySink: "internal-ui", + sourceMessageSeq: 42, + }, + timestamp: 42, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-internal-source-reply", + sessionKey: "agent:worker:main", + timeoutMs: 1_000, + }); + + expect(result).toEqual({ + status: "ok", + replyText: "source reply awaiting delivery", + }); + }); + + it("prefers an internal source reply over a later private final", async () => { + callGatewayMock.mockResolvedValueOnce({ status: "ok" }).mockResolvedValueOnce({ + messages: [ + { + role: "assistant", + provenance: { + kind: "inter_session", + sourceSessionKey: "agent:main:source", + sourceTool: "sessions_send", + }, + content: [{ type: "text", text: "forwarded request" }], + __openclaw: { seq: 41 }, + timestamp: 41, + }, + { + role: "assistant", + content: [{ type: "text", text: "source reply awaiting delivery" }], + openclawMessageToolMirror: { + toolName: "message", + toolCallId: "call-message-send", + sourceReplySink: "internal-ui", + sourceMessageSeq: 42, + }, + timestamp: 42, + }, + { + role: "assistant", + content: [{ type: "text", text: "Done" }], + timestamp: 43, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-internal-source-reply-with-private-final", + sessionKey: "agent:worker:main", + timeoutMs: 1_000, + }); + + expect(result).toEqual({ + status: "ok", + replyText: "source reply awaiting delivery", + }); + }); + + it("does not let a late internal result cross an inter-session turn boundary", async () => { + callGatewayMock.mockResolvedValueOnce({ status: "ok" }).mockResolvedValueOnce({ + messages: [ + { + role: "assistant", + provenance: { + kind: "inter_session", + sourceSessionKey: "agent:main:source", + sourceTool: "sessions_send", + }, + content: [{ type: "text", text: "new forwarded request" }], + __openclaw: { seq: 42 }, + timestamp: 42, + }, + { + role: "assistant", + content: [{ type: "text", text: "stale source reply" }], + openclawMessageToolMirror: { + toolName: "message", + toolCallId: "call-message-before-request", + sourceReplySink: "internal-ui", + sourceMessageSeq: 41, + }, + timestamp: 41, + }, + { + role: "assistant", + content: [{ type: "text", text: "fresh reply" }], + timestamp: 43, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-after-late-internal-source-reply", + sessionKey: "agent:worker:main", + timeoutMs: 1_000, + }); + + expect(result).toEqual({ + status: "ok", + replyText: "fresh reply", + }); + }); + + it("does not return a private final written after a message-tool delivery mirror", async () => { + callGatewayMock.mockResolvedValueOnce({ status: "ok" }).mockResolvedValueOnce({ + messages: [ + { + role: "assistant", + provenance: { + kind: "inter_session", + sourceSessionKey: "agent:main:source", + sourceTool: "sessions_send", + }, + content: [{ type: "text", text: "forwarded request" }], + timestamp: 41, + }, + { + role: "assistant", + content: [{ type: "text", text: "already delivered source reply" }], + openclawMessageToolMirror: { + toolName: "message", + toolCallId: "call-message-send", + }, + timestamp: 42, + }, + { + role: "assistant", + content: [{ type: "text", text: "Done" }], + timestamp: 43, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-source-reply-with-private-final", + sessionKey: "agent:main:child", + timeoutMs: 1_000, + }); + + expect(result).toEqual({ + status: "ok", + replyText: undefined, + }); + }); + + it("does not let an older turn's message-tool mirror suppress a fresh reply", async () => { + callGatewayMock.mockResolvedValueOnce({ status: "ok" }).mockResolvedValueOnce({ + messages: [ + { + role: "assistant", + content: [{ type: "text", text: "older delivered reply" }], + openclawMessageToolMirror: { + toolName: "message", + toolCallId: "call-older-message-send", + }, + timestamp: 40, + }, + { + role: "assistant", + provenance: { + kind: "inter_session", + sourceSessionKey: "agent:main:source", + sourceTool: "sessions_send", + }, + content: [{ type: "text", text: "new forwarded request" }], + timestamp: 41, + }, + { + role: "assistant", + content: [{ type: "text", text: "fresh reply" }], + timestamp: 42, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-after-older-source-reply", + sessionKey: "agent:main:child", + timeoutMs: 1_000, + }); + + expect(result).toEqual({ + status: "ok", + replyText: "fresh reply", + }); + }); + + it("does not resurrect an older reply when only a delivery mirror is newer", async () => { + callGatewayMock + .mockResolvedValueOnce({ + status: "ok", + }) + .mockResolvedValueOnce({ + messages: [ + { + role: "assistant", + content: [{ type: "text", text: "stale previous reply" }], + timestamp: 41, + }, + { + role: "assistant", + provider: "openclaw", + model: "delivery-mirror", + content: [{ type: "text", text: "already delivered source reply" }], + timestamp: 42, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-source-reply-without-baseline", + sessionKey: "agent:main:child", + timeoutMs: 1_000, + }); + + expect(result).toEqual({ + status: "ok", + replyText: undefined, + }); + }); + it("returns the new assistant text when the fingerprint changes", async () => { callGatewayMock .mockResolvedValueOnce({ @@ -446,11 +888,52 @@ describe("waitForAgentRunAndReadUpdatedAssistantReply", () => { replyText: "fresh reply", }); }); + + it("preserves successful wait metadata when returning an updated reply", async () => { + callGatewayMock + .mockResolvedValueOnce({ + status: "ok", + startedAt: 100, + endedAt: 200, + stopReason: "completed", + yielded: true, + providerStarted: true, + }) + .mockResolvedValueOnce({ + messages: [ + { + role: "assistant", + content: [{ type: "text", text: "fresh reply" }], + timestamp: 99, + }, + ], + }); + + const result = await waitForAgentRunAndReadUpdatedAssistantReply({ + runId: "run-with-metadata", + sessionKey: "agent:main:child", + timeoutMs: 1_000, + baseline: { + text: "older reply", + fingerprint: "old-fingerprint", + }, + }); + + expect(result).toEqual({ + status: "ok", + startedAt: 100, + endedAt: 200, + stopReason: "completed", + yielded: true, + providerStarted: true, + replyText: "fresh reply", + }); + }); }); describe("waitForAgentRunsToDrain", () => { beforeEach(() => { - callGatewayMock.mockClear(); + callGatewayMock.mockReset(); testing.setDepsForTest({ callGateway: async (opts) => await callGatewayMock(opts), }); diff --git a/src/agents/run-wait.ts b/src/agents/run-wait.ts index f4a7720555a6..2709c8d54d01 100644 --- a/src/agents/run-wait.ts +++ b/src/agents/run-wait.ts @@ -6,6 +6,7 @@ import { addTimerTimeoutGraceMs, asDateTimestampMs, + asPositiveSafeInteger, clampTimerTimeoutMs, parseFiniteNumber, resolveDateTimestampMs, @@ -14,6 +15,11 @@ import { import { callGateway } from "../gateway/call.js"; import { formatErrorMessage } from "../infra/errors.js"; import { normalizeBlockedLivenessWaitStatus } from "../shared/agent-liveness.js"; +import { + isOpenClawInternalSourceReplyMirrorAssistantMessage, + isOpenClawMessageToolMirrorAssistantMessage, + isTranscriptOnlyOpenClawAssistantMessage, +} from "../shared/transcript-only-openclaw-assistant.js"; import { buildAgentRunTerminalOutcomeFromWaitResult, type AgentRunTerminalOutcome, @@ -160,34 +166,174 @@ function normalizePendingRunIds(runIds: Iterable): string[] { return [...seen]; } -function resolveLatestAssistantReplySnapshot(messages: unknown[]): AssistantReplySnapshot { +function isWaitedReplyTranscriptArtifact(message: unknown): boolean { + return ( + isTranscriptOnlyOpenClawAssistantMessage(message) || + isOpenClawMessageToolMirrorAssistantMessage(message) || + isInterSessionInputMessage(message) + ); +} + +function isInterSessionInputMessage(message: unknown): boolean { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return false; + } + const provenance = (message as { provenance?: unknown }).provenance; + return ( + Boolean(provenance) && + typeof provenance === "object" && + !Array.isArray(provenance) && + (provenance as { kind?: unknown }).kind === "inter_session" + ); +} + +function isWaitedReplyTurnBoundary(message: unknown): boolean { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return false; + } + return (message as { role?: unknown }).role === "user" || isInterSessionInputMessage(message); +} + +function snapshotAssistantReply(message: unknown): AssistantReplySnapshot | undefined { + const text = extractAssistantText(message); + if (!text?.trim()) { + return undefined; + } + let fingerprint: string | undefined; + try { + fingerprint = JSON.stringify(message); + } catch { + fingerprint = text; + } + return { text, fingerprint }; +} + +function readTranscriptMessageSeq(message: unknown): number | undefined { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return undefined; + } + const meta = (message as { __openclaw?: unknown })["__openclaw"]; + if (!meta || typeof meta !== "object" || Array.isArray(meta)) { + return undefined; + } + return asPositiveSafeInteger((meta as { seq?: unknown }).seq); +} + +function readInternalSourceReplyMessageSeq(message: unknown): number | undefined { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return undefined; + } + const marker = (message as { openclawMessageToolMirror?: unknown }).openclawMessageToolMirror; + if (!marker || typeof marker !== "object" || Array.isArray(marker)) { + return undefined; + } + return asPositiveSafeInteger((marker as { sourceMessageSeq?: unknown }).sourceMessageSeq); +} + +function resolveLatestAssistantReplySnapshot( + messages: unknown[], + opts?: { stopAtTranscriptArtifact?: boolean }, +): AssistantReplySnapshot { + let latestReply: AssistantReplySnapshot = {}; + const internalSourceReplies: Array<{ + snapshot: AssistantReplySnapshot; + sourceMessageSeq?: number; + }> = []; + let sawTranscriptArtifact = false; for (let i = messages.length - 1; i >= 0; i -= 1) { const candidate = messages[i]; if (!candidate || typeof candidate !== "object") { continue; } + if (opts?.stopAtTranscriptArtifact === true && isWaitedReplyTurnBoundary(candidate)) { + const boundarySeq = readTranscriptMessageSeq(candidate); + const currentInternalSourceReply = boundarySeq + ? internalSourceReplies.find( + (reply) => reply.sourceMessageSeq !== undefined && reply.sourceMessageSeq > boundarySeq, + ) + : undefined; + if (currentInternalSourceReply) { + return currentInternalSourceReply.snapshot; + } + if (!boundarySeq && internalSourceReplies.length > 0) { + sawTranscriptArtifact = true; + } + internalSourceReplies.length = 0; + break; + } if ((candidate as { role?: unknown }).role !== "assistant") { continue; } - const text = extractAssistantText(candidate); - if (!text?.trim()) { + if ( + opts?.stopAtTranscriptArtifact === true && + isOpenClawInternalSourceReplyMirrorAssistantMessage(candidate) + ) { + // Internal source replies still need the outer A2A flow to deliver them. + // The source seq prevents a late old result from crossing a new turn. + const snapshot = snapshotAssistantReply(candidate); + const sourceMessageSeq = readInternalSourceReplyMessageSeq(candidate); + if (snapshot) { + internalSourceReplies.push({ snapshot, sourceMessageSeq }); + } + if (!sourceMessageSeq) { + sawTranscriptArtifact = true; + } continue; } - let fingerprint: string | undefined; - try { - fingerprint = JSON.stringify(candidate); - } catch { - fingerprint = text; + if (isWaitedReplyTranscriptArtifact(candidate)) { + if (opts?.stopAtTranscriptArtifact === true) { + sawTranscriptArtifact = true; + } + continue; + } + const snapshot = snapshotAssistantReply(candidate); + if (!snapshot) { + continue; + } + if (opts?.stopAtTranscriptArtifact !== true) { + return snapshot; + } + if (!latestReply.text) { + latestReply = snapshot; } - return { text, fingerprint }; } - return {}; + if (opts?.stopAtTranscriptArtifact === true) { + if (internalSourceReplies.length > 0) { + sawTranscriptArtifact = true; + } + if (sawTranscriptArtifact) { + return {}; + } + } + return latestReply; +} + +export function hasUpdatedAssistantReplySnapshot( + latestReply: AssistantReplySnapshot, + baseline: AssistantReplySnapshot | undefined, +): boolean { + if (!latestReply.text) { + return false; + } + if (!baseline) { + return true; + } + if (baseline.fingerprint !== undefined) { + return latestReply.fingerprint !== baseline.fingerprint; + } + if (baseline.text !== undefined) { + return latestReply.text !== baseline.text; + } + return true; } /** Read the latest non-tool assistant message for a session. */ export async function readLatestAssistantReplySnapshot(params: { sessionKey: string; limit?: number; + // Waited reply paths stop at transcript artifacts so they do not resurrect + // an older assistant message as a fresh post-run reply. + stopAtTranscriptArtifact?: boolean; callGateway?: GatewayCaller; }): Promise { const history = await (params.callGateway ?? runWaitDeps.callGateway)<{ @@ -198,6 +344,7 @@ export async function readLatestAssistantReplySnapshot(params: { }); return resolveLatestAssistantReplySnapshot( stripToolMessages(Array.isArray(history?.messages) ? history.messages : []), + { stopAtTranscriptArtifact: params.stopAtTranscriptArtifact }, ); } @@ -272,15 +419,14 @@ export async function waitForAgentRunAndReadUpdatedAssistantReply(params: { const latestReply = await readLatestAssistantReplySnapshot({ sessionKey: params.sessionKey, limit: params.limit, + stopAtTranscriptArtifact: true, callGateway: params.callGateway, }); - const baselineFingerprint = params.baseline?.fingerprint; - const replyText = - latestReply.text && (!baselineFingerprint || latestReply.fingerprint !== baselineFingerprint) - ? latestReply.text - : undefined; + const replyText = hasUpdatedAssistantReplySnapshot(latestReply, params.baseline) + ? latestReply.text + : undefined; return { - status: "ok", + ...wait, replyText, }; } diff --git a/src/agents/tools/agent-step.test.ts b/src/agents/tools/agent-step.test.ts index 133f8a2550ef..a604926b3f05 100644 --- a/src/agents/tools/agent-step.test.ts +++ b/src/agents/tools/agent-step.test.ts @@ -132,4 +132,75 @@ describe("runAgentStep", () => { expect(ingress?.sourceReplyDeliveryMode).toBe("message_tool_only"); expect(ingress?.transcriptMessage).toBe(""); }); + + it("does not return failed transcript-mode output as an announce reply", async () => { + const agentCommandFromIngress = vi.fn(async () => ({ + payloads: [ + { + text: "⚠️ Agent couldn't generate a response. Please try again.", + mediaUrl: null, + isError: true, + }, + ], + meta: { + durationMs: 1, + error: { + kind: "incomplete_turn" as const, + message: "Agent couldn't generate a response.", + fallbackSafe: true, + terminalPresentation: false, + }, + }, + })); + testing.setDepsForTest({ + agentCommandFromIngress, + callGateway: async (): Promise => ({ runId: "unused" }) as T, + }); + + await expect( + runAgentStep({ + sessionKey: "agent:main:subagent:child", + message: "internal announce step", + transcriptMessage: "", + extraSystemPrompt: "announce only", + timeoutMs: 10_000, + }), + ).resolves.toBeUndefined(); + + expect(bundleMcpRuntimeMocks.retireSessionMcpRuntimeForSessionKey).toHaveBeenCalledWith({ + sessionKey: "agent:main:subagent:child", + reason: "nested-agent-step-complete", + }); + }); + + it("returns trusted terminal presentations from incomplete transcript turns", async () => { + const presentation = + "The read-only lookup completed successfully.\n\n⚠️ Agent couldn't generate a response. Please try again."; + const agentCommandFromIngress = vi.fn(async () => ({ + payloads: [{ text: presentation, mediaUrl: null, isError: true }], + meta: { + durationMs: 1, + error: { + kind: "incomplete_turn" as const, + message: "Agent couldn't generate a response.", + fallbackSafe: true, + terminalPresentation: true, + }, + }, + })); + testing.setDepsForTest({ + agentCommandFromIngress, + callGateway: async (): Promise => ({ runId: "unused" }) as T, + }); + + await expect( + runAgentStep({ + sessionKey: "agent:main:subagent:child", + message: "internal announce step", + transcriptMessage: "", + extraSystemPrompt: "announce only", + timeoutMs: 10_000, + }), + ).resolves.toBe(presentation); + }); }); diff --git a/src/agents/tools/agent-step.ts b/src/agents/tools/agent-step.ts index 486eff5f6028..95dfb23d3cd2 100644 --- a/src/agents/tools/agent-step.ts +++ b/src/agents/tools/agent-step.ts @@ -28,7 +28,18 @@ let agentStepDeps: { } = defaultAgentStepDeps; function extractAgentCommandReply(result: unknown): string | undefined { - const payloads = (result as { payloads?: unknown } | undefined)?.payloads; + const candidate = result as { meta?: { error?: unknown }; payloads?: unknown } | null | undefined; + const error = + candidate?.meta?.error && + typeof candidate.meta.error === "object" && + !Array.isArray(candidate.meta.error) + ? (candidate.meta.error as { kind?: unknown; terminalPresentation?: unknown }) + : undefined; + // Plain incomplete-turn output is a control failure; trusted terminal tool presentations remain deliverable. + if (error?.kind === "incomplete_turn" && error.terminalPresentation !== true) { + return undefined; + } + const payloads = candidate?.payloads; if (!Array.isArray(payloads)) { return undefined; } diff --git a/src/agents/tools/sessions-send-tool.a2a.test.ts b/src/agents/tools/sessions-send-tool.a2a.test.ts index 6257c7d42ccf..3dbdaed45266 100644 --- a/src/agents/tools/sessions-send-tool.a2a.test.ts +++ b/src/agents/tools/sessions-send-tool.a2a.test.ts @@ -16,9 +16,9 @@ vi.mock("../../gateway/call.js", () => ({ })); vi.mock("../run-wait.js", async (importOriginal) => { - const { isRecoverableAgentWaitError } = await importOriginal(); + const actual = await importOriginal(); return { - isRecoverableAgentWaitError, + ...actual, waitForAgentRun: vi.fn().mockResolvedValue({ status: "ok" }), readLatestAssistantReplySnapshot: vi.fn().mockResolvedValue({ text: "Test announce reply", @@ -227,6 +227,25 @@ describe("runSessionsSendA2AFlow announce delivery", () => { expect(gatewayCalls.find((call) => call.method === "send")).toBeUndefined(); }); + it("delivers a legitimate reply that quotes incomplete-turn text", async () => { + const reply = 'The log says "Agent couldn\'t generate a response", but the retry succeeded.'; + + await runSessionsSendA2AFlow({ + targetSessionKey: "agent:main:discord:channel:target-room", + displayKey: "agent:main:discord:channel:target-room", + message: "Diagnose the failed turn", + announceTimeoutMs: 10_000, + maxPingPongTurns: 2, + requesterSessionKey: "agent:main:discord:channel:target-room", + requesterChannel: "discord", + roundOneReply: reply, + }); + + expect(runAgentStep).not.toHaveBeenCalled(); + const sendCall = requireGatewayCall("send"); + expect((sendCall.params as Record).message).toBe(reply); + }); + it("keeps the announce decider for same-session sends from a different channel", async () => { vi.mocked(runAgentStep).mockResolvedValueOnce("ANNOUNCE_SKIP"); @@ -247,6 +266,22 @@ describe("runSessionsSendA2AFlow announce delivery", () => { expect(gatewayCalls.find((call) => call.method === "send")).toBeUndefined(); }); + it("does not run the announce decider for same-session sends without an announce target", async () => { + await runSessionsSendA2AFlow({ + targetSessionKey: "agent:main:main", + displayKey: "agent:main:main", + message: "Test message", + announceTimeoutMs: 10_000, + maxPingPongTurns: 2, + requesterSessionKey: "agent:main:main", + requesterChannel: "qa-channel", + roundOneReply: "Already delivered through the source message tool", + }); + + expect(runAgentStep).not.toHaveBeenCalled(); + expect(gatewayCalls.find((call) => call.method === "send")).toBeUndefined(); + }); + it.each([ { source: "deliveryContext.accountId", @@ -472,6 +507,31 @@ describe("runSessionsSendA2AFlow announce delivery", () => { }); }); + it("does not inject a delayed reply that matches a text-only baseline", async () => { + vi.mocked(readLatestAssistantReplySnapshot).mockResolvedValueOnce({ + text: "same reply", + fingerprint: "same-reply-new-fingerprint", + }); + + await runSessionsSendA2AFlow({ + targetSessionKey: "agent:main:discord:group:dev", + displayKey: "agent:main:discord:group:dev", + message: "Test message", + announceTimeoutMs: 10_000, + maxPingPongTurns: 2, + requesterSessionKey: "agent:main:discord:group:req", + requesterChannel: "discord", + baseline: { + text: "same reply", + }, + waitRunId: "run-delayed", + }); + + expect(firstMockArg(vi.mocked(waitForAgentRun), "agent run wait").runId).toBe("run-delayed"); + expect(runAgentStep).not.toHaveBeenCalled(); + expect(gatewayCalls.find((call) => call.method === "send")).toBeUndefined(); + }); + it.each(["NO_REPLY", "HEARTBEAT_OK", "ANNOUNCE_SKIP"])( "suppresses exact announce control reply %s before channel delivery", async (announceReply) => { diff --git a/src/agents/tools/sessions-send-tool.a2a.ts b/src/agents/tools/sessions-send-tool.a2a.ts index 85cf4a8e9f57..7ae9154026bb 100644 --- a/src/agents/tools/sessions-send-tool.a2a.ts +++ b/src/agents/tools/sessions-send-tool.a2a.ts @@ -12,6 +12,7 @@ import { resolveNestedAgentLaneForSession } from "../lanes.js"; import { type AgentWaitResult, type AssistantReplySnapshot, + hasUpdatedAssistantReplySnapshot, isRecoverableAgentWaitError, readLatestAssistantReplySnapshot, waitForAgentRun, @@ -107,14 +108,12 @@ export async function runSessionsSendA2AFlow(params: { if (wait.status === "ok") { const latestSnapshot = await readLatestAssistantReplySnapshot({ sessionKey: params.targetSessionKey, + stopAtTranscriptArtifact: true, callGateway: sessionsSendA2ADeps.callGateway, }); - const baselineFingerprint = params.baseline?.fingerprint; - primaryReply = - latestSnapshot.text && - (!baselineFingerprint || latestSnapshot.fingerprint !== baselineFingerprint) - ? latestSnapshot.text - : undefined; + primaryReply = hasUpdatedAssistantReplySnapshot(latestSnapshot, params.baseline) + ? latestSnapshot.text + : undefined; latestReply = primaryReply; } else { if ( @@ -155,14 +154,13 @@ export async function runSessionsSendA2AFlow(params: { // A same-session send is a human-facing source-channel reply, not a true // agent-to-agent announcement. Asking the same session to decide whether to - // announce can learn stale ANNOUNCE_SKIP patterns from its own history and - // silently drop a normal channel response. - if ( + // announce can re-run the same prompt and duplicate source-reply side effects. + const sameSessionSourceReply = + params.requesterSessionKey && params.requesterSessionKey === params.targetSessionKey; + const canDirectDeliverSameSessionReply = announceTarget && - params.requesterSessionKey && - params.requesterSessionKey === params.targetSessionKey && - params.requesterChannel === announceTarget.channel - ) { + (!params.requesterChannel || params.requesterChannel === announceTarget.channel); + if (sameSessionSourceReply && canDirectDeliverSameSessionReply) { if (params.waitRunId && !params.roundOneReply && !params.baseline) { return; } @@ -173,6 +171,9 @@ export async function runSessionsSendA2AFlow(params: { }); return; } + if (sameSessionSourceReply && !announceTarget) { + return; + } if ( params.maxPingPongTurns > 0 && diff --git a/src/agents/tools/sessions-send-tool.ts b/src/agents/tools/sessions-send-tool.ts index 279b13f719aa..a5f4f1ad5aa9 100644 --- a/src/agents/tools/sessions-send-tool.ts +++ b/src/agents/tools/sessions-send-tool.ts @@ -555,7 +555,7 @@ export function createSessionsSendTool(opts?: { }); } - const requesterSessionKey = opts?.agentSessionKey; + const requesterSessionKey = opts?.agentSessionKey ? effectiveRequesterKey : undefined; const requesterChannel = opts?.agentChannel; const sameSessionA2A = requesterSessionKey === resolvedKey; const isIsolatedCronRequester = isCronRunSessionKey(requesterSessionKey); @@ -597,14 +597,14 @@ export function createSessionsSendTool(opts?: { : undefined; const agentMessageContext = buildAgentToAgentMessageContext({ - requesterSessionKey: opts?.agentSessionKey, - requesterChannel: opts?.agentChannel, + requesterSessionKey, + requesterChannel, targetSessionKey: displayKey, }); const inputProvenance = { kind: "inter_session" as const, - sourceSessionKey: opts?.agentSessionKey, - sourceChannel: opts?.agentChannel, + sourceSessionKey: requesterSessionKey, + sourceChannel: requesterChannel, sourceTool: "sessions_send", }; const sendParams = { diff --git a/src/agents/tools/sessions.test.ts b/src/agents/tools/sessions.test.ts index 3a0551b8e375..f24b6eebcb0c 100644 --- a/src/agents/tools/sessions.test.ts +++ b/src/agents/tools/sessions.test.ts @@ -1195,6 +1195,55 @@ describe("sessions_send gating", () => { expect(flowParams?.baseline?.text).toBe("older reply from a previous run"); }); + it("canonicalizes aliased requester keys for same-session A2A delivery", async () => { + const { runSessionsSendA2AFlow } = await import("./sessions-send-tool.a2a.js"); + vi.mocked(runSessionsSendA2AFlow).mockClear(); + const tool = createSessionsSendTool({ + agentSessionKey: "main", + agentChannel: MAIN_AGENT_CHANNEL, + config: { + session: { scope: "per-sender", mainKey: MAIN_AGENT_SESSION_KEY }, + tools: { agentToAgent: { enabled: false } }, + } as never, + }); + const staleAssistantMessage = { + role: "assistant", + content: [{ type: "text", text: "older reply from a previous run" }], + timestamp: 20, + }; + + callGatewayMock.mockImplementation(async (opts: unknown) => { + const request = opts as { method?: string }; + if (request.method === "sessions.list") { + return { + path: "/tmp/sessions.json", + sessions: [{ key: MAIN_AGENT_SESSION_KEY, kind: "direct" }], + }; + } + if (request.method === "chat.history") { + return { messages: [staleAssistantMessage] }; + } + if (request.method === "agent") { + return { runId: "run-alias-fire-and-forget", acceptedAt: 123 }; + } + return {}; + }); + + const result = await tool.execute("call-aliased-fire-and-forget-same-session", { + sessionKey: MAIN_AGENT_SESSION_KEY, + message: "ping", + timeoutSeconds: 0, + }); + + const details = requireDetails(result); + expect(details.status).toBe("accepted"); + expect(details.sessionKey).toBe("main"); + const flowParams = vi.mocked(runSessionsSendA2AFlow).mock.calls[0]?.[0]; + expect(flowParams?.requesterSessionKey).toBe(MAIN_AGENT_SESSION_KEY); + expect(flowParams?.targetSessionKey).toBe(MAIN_AGENT_SESSION_KEY); + expect(flowParams?.baseline?.text).toBe("older reply from a previous run"); + }); + it("accepts fire-and-forget same-session sends when baseline history is unavailable", async () => { const { runSessionsSendA2AFlow } = await import("./sessions-send-tool.a2a.js"); vi.mocked(runSessionsSendA2AFlow).mockClear(); diff --git a/src/gateway/chat-display-projection.ts b/src/gateway/chat-display-projection.ts index 3936760a9229..24fb96b5330c 100644 --- a/src/gateway/chat-display-projection.ts +++ b/src/gateway/chat-display-projection.ts @@ -1,7 +1,10 @@ // Gateway chat display projection. // Converts raw transcript messages into bounded Control UI/history display records. import { createHash } from "node:crypto"; -import { asFiniteNumber } from "@openclaw/normalization-core/number-coercion"; +import { + asFiniteNumber, + asPositiveSafeInteger, +} from "@openclaw/normalization-core/number-coercion"; import { asOptionalRecord as readRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; @@ -39,6 +42,7 @@ type PendingMessageToolVisibleReply = { completionAnchor?: Record; deliveryMirrorAnchor?: Record; deliveryMirrorIndex?: number; + sourceReplySink?: "internal-ui"; succeeded: boolean; }; @@ -928,15 +932,25 @@ function isSuccessfulMessageToolResultPayload(message: Record): return ok !== false; } +function readMessageToolSourceReplySink( + message: Record, +): "internal-ui" | undefined { + const details = readRecord(message.details); + return details?.sourceReplySink === "internal-ui" ? "internal-ui" : undefined; +} + function buildMessageToolVisibleReplyMirror( pending: PendingMessageToolVisibleReply, ): Record { + const sourceMessageSeq = asPositiveSafeInteger(readRecord(pending.anchor["__openclaw"])?.seq); const mirror: Record = { role: "assistant", content: [{ type: "text", text: pending.text }], openclawMessageToolMirror: { toolName: "message", ...(pending.toolCallId ? { toolCallId: pending.toolCallId } : {}), + ...(pending.sourceReplySink ? { sourceReplySink: pending.sourceReplySink } : {}), + ...(pending.sourceReplySink && sourceMessageSeq ? { sourceMessageSeq } : {}), }, }; for (const field of ["timestamp", "createdAt", "agentId"] as const) { @@ -1054,6 +1068,10 @@ function mirrorMessageToolVisibleReplies(messages: unknown[]): unknown[] { for (const item of pending) { if (!item.succeeded && isSuccessfulMessageToolResult(record, item)) { item.succeeded = true; + const sourceReplySink = readMessageToolSourceReplySink(record); + if (sourceReplySink) { + item.sourceReplySink = sourceReplySink; + } item.completionAnchor = item.deliveryMirrorAnchor ?? record; if (item.deliveryMirrorAnchor) { if (typeof item.deliveryMirrorIndex === "number") { diff --git a/src/gateway/server-methods/server-methods.test.ts b/src/gateway/server-methods/server-methods.test.ts index 461561565f39..29a1a7e16dd4 100644 --- a/src/gateway/server-methods/server-methods.test.ts +++ b/src/gateway/server-methods/server-methods.test.ts @@ -1432,6 +1432,7 @@ describe("projectRecentChatDisplayMessages", () => { args: { action: "send", message: "visible via message tool" }, }, ], + __openclaw: { seq: 1 }, timestamp: 1, }, { @@ -1442,6 +1443,7 @@ describe("projectRecentChatDisplayMessages", () => { sourceSessionKey: "agent:main:webchat:source", sourceTool: "sessions_send", }, + __openclaw: { seq: 2 }, timestamp: 2, }, { @@ -1449,6 +1451,7 @@ describe("projectRecentChatDisplayMessages", () => { toolName: "message", toolCallId: "call-message", content: JSON.stringify({ ok: true }), + details: { sourceReplySink: "internal-ui" }, timestamp: 3, }, { @@ -1469,6 +1472,7 @@ describe("projectRecentChatDisplayMessages", () => { args: { action: "send", message: "visible via message tool" }, }, ], + __openclaw: { seq: 1 }, timestamp: 1, }, { @@ -1480,6 +1484,7 @@ describe("projectRecentChatDisplayMessages", () => { sourceSessionKey: "agent:main:webchat:source", sourceTool: "sessions_send", }, + __openclaw: { seq: 2 }, timestamp: 2, }, { @@ -1495,6 +1500,8 @@ describe("projectRecentChatDisplayMessages", () => { openclawMessageToolMirror: { toolName: "message", toolCallId: "call-message", + sourceReplySink: "internal-ui", + sourceMessageSeq: 1, }, timestamp: 1, }, diff --git a/src/gateway/server.chat.gateway-server-chat.test.ts b/src/gateway/server.chat.gateway-server-chat.test.ts index 7fd093b2bc62..2fc833adc0fb 100644 --- a/src/gateway/server.chat.gateway-server-chat.test.ts +++ b/src/gateway/server.chat.gateway-server-chat.test.ts @@ -870,6 +870,64 @@ describe("gateway server chat", () => { ).toBe(true); }); + test("chat.history marks message-tool replies held for internal source delivery", async () => { + const replyText = "Forward this source reply."; + const historyMessages = await loadChatHistoryWithMessages([ + { + role: "assistant", + content: [ + { + type: "toolCall", + id: "call-message-internal-source", + name: "message", + arguments: { + action: "send", + message: replyText, + }, + }, + ], + timestamp: 1, + }, + { + role: "toolResult", + toolName: "message", + toolCallId: "call-message-internal-source", + content: [{ type: "text", text: "Sent visible reply via internal-ui." }], + details: { + status: "ok", + deliveryStatus: "sent", + sourceReplySink: "internal-ui", + }, + timestamp: 2, + }, + { + role: "assistant", + content: [{ type: "text", text: "NO_REPLY" }], + timestamp: 3, + }, + ]); + + const visibleAssistantMessages = historyMessages.filter((message) => { + if (!message || typeof message !== "object") { + return false; + } + const entry = message as { role?: unknown }; + return entry.role === "assistant" && extractFirstTextBlock(message) !== undefined; + }); + expect(visibleAssistantMessages).toEqual([ + expect.objectContaining({ + role: "assistant", + content: [{ type: "text", text: replyText }], + openclawMessageToolMirror: { + toolName: "message", + toolCallId: "call-message-internal-source", + sourceReplySink: "internal-ui", + sourceMessageSeq: 1, + }, + }), + ]); + }); + test("chat.history hides raw delivery-mirror rows but keeps message-tool mirrors", async () => { const replyText = "One visible send."; const historyMessages = await loadChatHistoryWithMessages([ diff --git a/src/gateway/server.sessions-send.test.ts b/src/gateway/server.sessions-send.test.ts index c23073cdabdc..bdc22b63e98c 100644 --- a/src/gateway/server.sessions-send.test.ts +++ b/src/gateway/server.sessions-send.test.ts @@ -292,6 +292,184 @@ describe("sessions_send gateway loopback", () => { } }, ); + + it( + "does not re-announce a trailing message-tool delivery mirror after a waited A2A run", + { timeout: SESSION_SEND_E2E_TIMEOUT_MS }, + async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-sessions-send-mirror-")); + const sessionKey = "agent:main:whatsapp:direct:peer-1"; + const sessionId = "sess-whatsapp-mirror"; + const sessionFile = path.join(dir, `${sessionId}.jsonl`); + const runId = `run-message-tool-mirror-${Date.now()}-${Math.random().toString(36).slice(2)}`; + const deliveredReply = "already delivered source reply"; + const sendCalls: Array<{ + to?: string; + text?: string; + accountId?: string | null; + threadId?: string | number | null; + }> = []; + setTestPluginRegistry( + createTestRegistry([ + { + pluginId: "whatsapp", + source: "test", + plugin: createOutboundTestPlugin({ + id: "whatsapp", + label: "WhatsApp", + outbound: { + deliveryMode: "direct", + resolveTarget: ({ to }) => { + const target = to?.trim(); + return target + ? { ok: true, to: target } + : { ok: false, error: new Error("missing target") }; + }, + sendText: async (ctx) => { + sendCalls.push({ + to: ctx.to, + text: ctx.text, + accountId: ctx.accountId, + threadId: ctx.threadId, + }); + return { channel: "whatsapp", messageId: "wa-duplicate-proof-msg" }; + }, + }, + messaging: { + normalizeTarget: (raw) => raw, + }, + }), + }, + ]), + ); + + testState.sessionStorePath = path.join(dir, "sessions.json"); + try { + await writeSessionStore({ + entries: { + [sessionKey]: { + sessionId, + sessionFile, + updatedAt: Date.now(), + deliveryContext: { + channel: "whatsapp", + to: "peer-1", + }, + origin: { + provider: "whatsapp", + accountId: "work", + threadId: "thread-77", + }, + }, + }, + }); + await fs.writeFile( + sessionFile, + [ + { + message: { + role: "assistant", + content: [{ type: "text", text: "previous real reply" }], + timestamp: 1, + }, + }, + { + message: { + role: "assistant", + content: [ + { + type: "toolCall", + id: "call-message-duplicate-proof", + name: "message", + arguments: { + action: "send", + message: deliveredReply, + }, + }, + ], + timestamp: 2, + }, + }, + { + message: { + role: "toolResult", + toolName: "message", + toolCallId: "call-message-duplicate-proof", + content: { ok: true, messageId: "24271", chatId: "peer-1" }, + timestamp: 3, + }, + }, + { + message: { + role: "assistant", + provider: "openclaw", + model: "delivery-mirror", + content: [{ type: "text", text: deliveredReply }], + timestamp: 4, + }, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n") + "\n", + "utf-8", + ); + + const { callGateway } = await import("./call.js"); + const history = await callGateway<{ messages?: unknown[] }>({ + method: "chat.history", + params: { sessionKey, limit: 10 }, + timeoutMs: 5_000, + }); + expect(history.messages).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + role: "assistant", + content: expect.arrayContaining([ + expect.objectContaining({ type: "text", text: deliveredReply }), + ]), + openclawMessageToolMirror: expect.objectContaining({ + toolName: "message", + toolCallId: "call-message-duplicate-proof", + }), + }), + ]), + ); + + const startedAt = Date.now(); + emitAgentEvent({ + runId, + stream: "lifecycle", + data: { phase: "start", startedAt }, + }); + emitAgentEvent({ + runId, + stream: "lifecycle", + data: { phase: "end", startedAt, endedAt: Date.now() }, + }); + agentStepTesting.setDepsForTest({ + agentCommandFromIngress: async () => ({ + payloads: [{ text: "SHOULD_NOT_SEND", mediaUrl: null }], + meta: { durationMs: 1 }, + }), + }); + + await runSessionsSendA2AFlow({ + targetSessionKey: sessionKey, + displayKey: sessionKey, + message: "proof ping", + announceTimeoutMs: 5_000, + maxPingPongTurns: 0, + waitRunId: runId, + }); + + expect(sendCalls).toEqual([]); + } finally { + agentStepTesting.setDepsForTest(); + testState.sessionStorePath = undefined; + await fs.rm(dir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + } + }, + ); }); describe("sessions_send label lookup", () => { diff --git a/src/shared/transcript-only-openclaw-assistant.ts b/src/shared/transcript-only-openclaw-assistant.ts index 6a842a0571c6..f5ceb080d8b9 100644 --- a/src/shared/transcript-only-openclaw-assistant.ts +++ b/src/shared/transcript-only-openclaw-assistant.ts @@ -30,6 +30,27 @@ export function isTranscriptOnlyOpenClawAssistantMessage(message: unknown): bool ); } +export function isOpenClawMessageToolMirrorAssistantMessage(message: unknown): boolean { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return false; + } + const entry = message as { role?: unknown; openclawMessageToolMirror?: unknown }; + return entry.role === "assistant" && entry.openclawMessageToolMirror !== undefined; +} + +export function isOpenClawInternalSourceReplyMirrorAssistantMessage(message: unknown): boolean { + if (!isOpenClawMessageToolMirrorAssistantMessage(message)) { + return false; + } + const marker = (message as { openclawMessageToolMirror?: unknown }).openclawMessageToolMirror; + return ( + Boolean(marker) && + typeof marker === "object" && + !Array.isArray(marker) && + (marker as { sourceReplySink?: unknown }).sourceReplySink === "internal-ui" + ); +} + export function isOpenClawDeliveryMirrorAssistantMessage(message: unknown): boolean { if (!message || typeof message !== "object" || Array.isArray(message)) { return false;