Skip to content

Commit b9a871d

Browse files
cavit99cavit99
committed
fix: stop A2A source-reply mirror duplicates
1 parent 1f3ea6f commit b9a871d

11 files changed

Lines changed: 932 additions & 36 deletions

File tree

‎extensions/qa-lab/src/providers/mock-openai/server.test.ts‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -467,6 +467,75 @@ describe("qa mock openai server", () => {
467467
expect(responseBody).toContain('"text":"QA-FINAL-ONLY-STREAMING-OK"');
468468
});
469469

470+
it("plans sessions_send for the A2A message-tool mirror proof scenario", async () => {
471+
const server = await startMockServer();
472+
const prompt =
473+
'qa a2a message-tool mirror check. sessionKey="agent:qa:a2a-target". exact marker: `QA-A2A-MIRROR-OK`';
474+
475+
const toolPlan = await expectResponsesJson(server, {
476+
stream: false,
477+
model: "gpt-5.5",
478+
tools: [{ type: "function", name: "sessions_send" }],
479+
input: [makeUserInput(prompt)],
480+
});
481+
482+
const args = outputToolArgs(toolPlan);
483+
expect(outputItem(toolPlan).type).toBe("function_call");
484+
expect(outputItem(toolPlan).name).toBe("sessions_send");
485+
expect(args).toMatchObject({
486+
sessionKey: "agent:qa:a2a-target",
487+
timeoutSeconds: 0,
488+
});
489+
expect(String(args.message)).toContain("qa group visible reply tool check");
490+
expect(String(args.message)).toContain("QA-A2A-MIRROR-OK");
491+
492+
const debugResponse = await fetch(`${server.baseUrl}/debug/last-request`);
493+
expect(debugResponse.status).toBe(200);
494+
const debugPayload = requireRecord(await debugResponse.json(), "debug request");
495+
expect(debugPayload.plannedToolName).toBe("sessions_send");
496+
expect(debugPayload.plannedToolArgs).toMatchObject({
497+
sessionKey: "agent:qa:a2a-target",
498+
timeoutSeconds: 0,
499+
});
500+
501+
const final = await expectResponsesJson(server, {
502+
stream: false,
503+
model: "gpt-5.5",
504+
tools: [{ type: "function", name: "sessions_send" }],
505+
input: [
506+
makeUserInput(prompt),
507+
{
508+
type: "function_call_output",
509+
call_id: "call_mock_sessions_send_fixture",
510+
output: JSON.stringify({ status: "accepted", delivery: { mode: "announce" } }),
511+
},
512+
],
513+
});
514+
expect(outputText(final)).toBe("");
515+
516+
const targetToolPlan = await expectResponsesJson(server, {
517+
stream: false,
518+
model: "gpt-5.5",
519+
tools: [
520+
{ type: "function", name: "sessions_send" },
521+
{ type: "function", name: "message" },
522+
],
523+
input: [
524+
makeUserInput(prompt),
525+
makeUserInput(
526+
"qa group visible reply tool check. Use the visible room reply path. exact marker: `QA-A2A-MIRROR-OK`",
527+
),
528+
],
529+
});
530+
531+
expect(outputItem(targetToolPlan).type).toBe("function_call");
532+
expect(outputItem(targetToolPlan).name).toBe("message");
533+
expect(outputToolArgs(targetToolPlan)).toMatchObject({
534+
action: "send",
535+
message: "QA-A2A-MIRROR-OK",
536+
});
537+
});
538+
470539
it("emits deterministic text deltas for generic streaming QA prompts", async () => {
471540
const server = await startMockServer();
472541

‎extensions/qa-lab/src/providers/mock-openai/server.ts‎

Lines changed: 31 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -169,13 +169,13 @@ const QA_BLOCK_STREAMING_PROMPT_RE = /block streaming qa check/i;
169169
const QA_TOOL_PROGRESS_ERROR_PROMPT_RE = /tool progress error qa check/i;
170170
const QA_TOOL_PROGRESS_PROMPT_RE = /tool progress qa check/i;
171171
const QA_GROUP_VISIBLE_REPLY_TOOL_PROMPT_RE = /qa group visible reply tool check/i;
172+
const QA_A2A_MESSAGE_TOOL_MIRROR_PROMPT_RE = /qa a2a message-tool mirror check/i;
172173
const QA_GROUP_MESSAGE_UNAVAILABLE_FALLBACK_PROMPT_RE =
173174
/qa group message unavailable fallback check/i;
174175
const QA_STRANDED_FINAL_RECOVERY_PROMPT_RE = /qa stranded final recovery check/i;
175176
const QA_STRANDED_FINAL_RETRY_FAILURE_PROMPT_RE = /qa stranded final retry failure check/i;
176177
const QA_STRANDED_FINAL_RETRY_PROMPT_RE = /you did not call message\(action=send\)/i;
177-
const QA_STRANDED_FINAL_RETRY_FAILURE_MARKER =
178-
"QA-STRANDED-RETRY-FAIL-RAW";
178+
const QA_STRANDED_FINAL_RETRY_FAILURE_MARKER = "QA-STRANDED-RETRY-FAIL-RAW";
179179
const QA_TELEGRAM_CURRENT_SESSION_STATUS_PROMPT_RE = /telegram current session_status qa check/i;
180180
const QA_TELEGRAM_STREAM_SINGLE_MARKER = "QA-TELEGRAM-STREAM-SINGLE-OK";
181181
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<string, unknown> |
12921292
};
12931293
}
12941294

1295+
function buildQaA2aMessageToolMirrorSessionsSendArgs(text: string): Record<string, unknown> | null {
1296+
if (!QA_A2A_MESSAGE_TOOL_MIRROR_PROMPT_RE.test(text)) {
1297+
return null;
1298+
}
1299+
const sessionKey =
1300+
extractQuotedToolArg(text, "sessionKey") ?? extractBareToolArg(text, "sessionKey");
1301+
if (!sessionKey) {
1302+
return null;
1303+
}
1304+
const marker =
1305+
extractExactMarkerDirective(text) ??
1306+
extractExactReplyDirective(text) ??
1307+
"QA-A2A-MESSAGE-TOOL-MIRROR-OK";
1308+
return {
1309+
sessionKey,
1310+
message: `qa group visible reply tool check. Use the visible room reply path. exact marker: \`${marker}\``,
1311+
timeoutSeconds: 0,
1312+
};
1313+
}
1314+
12951315
function extractToolErrorForNamedCall(params: {
12961316
input: ResponsesInputItem[];
12971317
name: string;
@@ -2569,6 +2589,15 @@ async function buildResponsesPayload(
25692589
}
25702590
return buildAssistantEvents(buildStrandedFinalRecoveryText());
25712591
}
2592+
if (QA_A2A_MESSAGE_TOOL_MIRROR_PROMPT_RE.test(prompt)) {
2593+
if (toolOutput) {
2594+
return buildAssistantEvents("");
2595+
}
2596+
const sessionsSendArgs = buildQaA2aMessageToolMirrorSessionsSendArgs(prompt);
2597+
if (sessionsSendArgs && hasDeclaredTool(body, "sessions_send")) {
2598+
return buildToolCallEventsWithArgs("sessions_send", sessionsSendArgs);
2599+
}
2600+
}
25722601
if (QA_GROUP_VISIBLE_REPLY_TOOL_PROMPT_RE.test(allInputText)) {
25732602
const marker = exactMarkerDirective ?? exactReplyDirective ?? "QA-GROUP-TOOL-OK";
25742603
if (!toolOutput && hasDeclaredTool(body, "message")) {

‎qa/scenarios/channels/a2a-message-tool-mirror-dedupe.yaml‎

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
title: A2A message-tool mirror dedupe
2+
3+
scenario:
4+
id: a2a-message-tool-mirror-dedupe
5+
surface: channel
6+
coverage:
7+
primary:
8+
- runtime.delivery
9+
secondary:
10+
- channels.qa-channel
11+
- tools.message
12+
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.
13+
gatewayConfigPatch:
14+
messages:
15+
groupChat:
16+
visibleReplies: message_tool
17+
session:
18+
agentToAgent:
19+
maxPingPongTurns: 0
20+
tools:
21+
sessions:
22+
visibility: all
23+
agentToAgent:
24+
enabled: true
25+
successCriteria:
26+
- Source agent receives a synthetic qa-channel group turn.
27+
- Source agent calls sessions_send against its real qa-channel group session.
28+
- Target run calls message(action=send) under the source-reply path.
29+
- The requester group sees the marker exactly once, with no duplicate during the post-delivery window.
30+
docsRefs:
31+
- docs/channels/qa-channel.md
32+
- docs/concepts/qa-e2e-automation.md
33+
codeRefs:
34+
- src/agents/tools/sessions-send-tool.ts
35+
- src/agents/tools/sessions-send-tool.a2a.ts
36+
- src/agents/run-wait.ts
37+
- src/shared/transcript-only-openclaw-assistant.ts
38+
execution:
39+
kind: flow
40+
summary: Run a real QA Gateway/qa-channel A2A source-reply turn and assert the message-tool delivery mirror is not announced again.
41+
config:
42+
requiredProviderMode: mock-openai
43+
requiredChannelDriver: qa-channel
44+
conversationId: qa-a2a-mirror-room
45+
conversationTitle: QA A2A Mirror Room
46+
promptSnippet: qa a2a message-tool mirror check
47+
targetPromptSnippet: qa group visible reply tool check
48+
expectedMarker: QA-A2A-MESSAGE-TOOL-MIRROR-OK
49+
duplicateWindowMs: 8000
50+
51+
flow:
52+
steps:
53+
- name: delivers target message-tool reply once without mirror re-announce
54+
actions:
55+
- assert:
56+
expr: "env.providerMode === config.requiredProviderMode"
57+
message: this seeded scenario is mock-openai only
58+
- call: waitForGatewayHealthy
59+
args:
60+
- ref: env
61+
- 60000
62+
- call: waitForQaChannelReady
63+
args:
64+
- ref: env
65+
- 60000
66+
- call: reset
67+
- set: startIndex
68+
value:
69+
expr: state.getSnapshot().messages.length
70+
- set: targetSessionKey
71+
value:
72+
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 })"
73+
- set: requestCountBefore
74+
value:
75+
expr: "env.mock ? (await fetchJson(`${env.mock.baseUrl}/debug/requests`)).length : 0"
76+
- call: state.addInboundMessage
77+
args:
78+
- conversation:
79+
id:
80+
expr: config.conversationId
81+
kind: group
82+
title:
83+
expr: config.conversationTitle
84+
senderId: alice
85+
senderName: Alice
86+
text:
87+
expr: "`@openclaw ${config.promptSnippet}. sessionKey=\"${targetSessionKey}\". Use sessions_send once with timeoutSeconds=0. The target must reply visibly with exact marker: \\`${config.expectedMarker}\\``"
88+
- call: waitForCondition
89+
saveAs: sourceSessionsSendRequest
90+
args:
91+
- lambda:
92+
async: true
93+
params: []
94+
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"
95+
- expr: liveTurnTimeoutMs(env, 60000)
96+
- 500
97+
- call: waitForCondition
98+
saveAs: targetMessageToolRequest
99+
args:
100+
- lambda:
101+
async: true
102+
params: []
103+
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"
104+
- expr: liveTurnTimeoutMs(env, 90000)
105+
- 500
106+
- call: waitForOutboundMessage
107+
saveAs: outbound
108+
args:
109+
- ref: state
110+
- lambda:
111+
params: [candidate]
112+
expr: "candidate.conversation.id === config.conversationId && candidate.conversation.kind === 'group' && candidate.direction === 'outbound' && String(candidate.text ?? '').includes(config.expectedMarker)"
113+
- expr: liveTurnTimeoutMs(env, 90000)
114+
- sinceIndex:
115+
ref: startIndex
116+
- call: sleep
117+
args:
118+
- expr: config.duplicateWindowMs
119+
- set: snapshot
120+
value:
121+
expr: state.getSnapshot()
122+
- set: matchingOutbound
123+
value:
124+
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))"
125+
- assert:
126+
expr: matchingOutbound.length === 1
127+
message:
128+
expr: "`expected exactly one requester-visible A2A marker after duplicate window, saw ${matchingOutbound.length}; transcript=${formatTransportTranscript(state, { conversationId: config.conversationId })}`"
129+
- set: scenarioRequests
130+
value:
131+
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 })) : []"
132+
- assert:
133+
expr: "!env.mock || scenarioRequests.filter((request) => request.plannedToolName === 'sessions_send' && request.plannedToolArgs?.sessionKey === targetSessionKey && request.plannedToolArgs?.timeoutSeconds === 0).length === 1"
134+
message:
135+
expr: "`expected exactly one source sessions_send plan for the target session; requests=${JSON.stringify(scenarioRequests)}`"
136+
- assert:
137+
expr: "!env.mock || scenarioRequests.filter((request) => request.plannedToolName === 'message' && request.plannedToolArgs?.action === 'send' && request.plannedToolArgs?.message === config.expectedMarker).length === 1"
138+
message:
139+
expr: "`expected exactly one target message(action=send) plan for the marker; requests=${JSON.stringify(scenarioRequests)}`"
140+
detailsExpr: "`outbound=${outbound.conversation.kind}:${outbound.conversation.id}:${outbound.text}; sourceTool=${JSON.stringify(sourceSessionsSendRequest?.plannedToolArgs ?? {})}; targetTool=${JSON.stringify(targetMessageToolRequest?.plannedToolArgs ?? {})}; duplicateWindowMs=${config.duplicateWindowMs}`"

0 commit comments

Comments
 (0)