mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-20 09:47:53 +00:00
fix(agents): restore ask_user roundtrip in Gateway chats (#110961)
* fix(agents): restore ask_user channel roundtrip * test(qa): derive ask_user proof from answers * test(qa): isolate mock reply directives * test(qa): wait for complete ask_user reply * fix(agents): fail closed on stale ask_user prompts * chore: remove release-owned changelog entry * fix(agents): satisfy ask_user CI contracts
This commit is contained in:
Peter Steinberger
@@ -20,12 +20,12 @@ sidebarTitle: "Tools and custom providers"
|
||||
Local onboarding defaults new local configs to `tools.profile: "coding"` when unset (existing explicit profiles are preserved).
|
||||
</Note>
|
||||
|
||||
| Profile | Includes |
|
||||
| ----------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `minimal` | `session_status` only |
|
||||
| `coding` | `group:fs`, `group:runtime`, `group:web`, `group:sessions`, `group:memory`, `cron`, `get_goal`, `create_goal`, `update_goal`, `update_plan`, `skill_workshop`, `image`, `image_generate`, `music_generate`, `video_generate` |
|
||||
| `messaging` | `group:messaging`, `sessions_list`, `sessions_history`, `sessions_send`, `session_status` |
|
||||
| `full` | No restriction (same as unset) |
|
||||
| Profile | Includes |
|
||||
| ----------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `minimal` | `session_status` only |
|
||||
| `coding` | `group:fs`, `group:runtime`, `group:web`, `group:sessions`, `group:memory`, `cron`, `get_goal`, `create_goal`, `update_goal`, `update_plan`, `ask_user`, `skill_workshop`, `image`, `image_generate`, `music_generate`, `video_generate` |
|
||||
| `messaging` | `group:messaging`, `sessions_list`, `sessions_history`, `sessions_send`, `session_status`, `ask_user` |
|
||||
| `full` | No restriction (same as unset) |
|
||||
|
||||
`coding` and `messaging` also implicitly allow `bundle-mcp` (configured MCP servers).
|
||||
|
||||
@@ -42,7 +42,7 @@ Local onboarding defaults new local configs to `tools.profile: "coding"` when un
|
||||
| `group:automation` | `heartbeat_respond`, `cron`, `gateway` |
|
||||
| `group:messaging` | `message` |
|
||||
| `group:nodes` | `nodes`, `computer` |
|
||||
| `group:agents` | `agents_list`, `get_goal`, `create_goal`, `update_goal`, `update_plan`, `skill_workshop` |
|
||||
| `group:agents` | `agents_list`, `get_goal`, `create_goal`, `update_goal`, `update_plan`, `ask_user`, `skill_workshop` |
|
||||
| `group:media` | `image`, `image_generate`, `music_generate`, `video_generate`, `tts` |
|
||||
| `group:openclaw` | All built-in tools above except `read`/`write`/`edit`/`apply_patch`/`exec`/`process`/`canvas` (excludes plugin tools) |
|
||||
| `group:plugins` | Tools owned by loaded plugins, including configured MCP servers exposed through `bundle-mcp` |
|
||||
|
||||
@@ -61,6 +61,16 @@ export function buildAssistantText(
|
||||
? extractLatestToolOutput(input)
|
||||
: "");
|
||||
const toolJson = parseToolOutputJson(scenarioToolOutput);
|
||||
const structuredToolText = Array.isArray(toolJson?.content)
|
||||
? toolJson.content
|
||||
.map((entry) =>
|
||||
entry && typeof entry === "object" && !Array.isArray(entry)
|
||||
? (entry as { text?: unknown }).text
|
||||
: undefined,
|
||||
)
|
||||
.filter((value): value is string => typeof value === "string")
|
||||
.join("\n")
|
||||
: "";
|
||||
const userTexts = extractAllUserTexts(input);
|
||||
const allInputText = extractAllRequestTexts(input, body);
|
||||
const rememberedFact = extractRememberedFact(userTexts);
|
||||
@@ -80,9 +90,12 @@ export function buildAssistantText(
|
||||
: "";
|
||||
const promptExactReplyDirective = extractExactReplyDirective(prompt);
|
||||
const promptExactMarkerDirective = extractExactMarkerDirective(prompt);
|
||||
const allUserText = userTexts.join("\n");
|
||||
const userExactReplyDirective =
|
||||
promptExactReplyDirective ?? extractExactReplyDirective(allUserText);
|
||||
const userExactMarkerDirective =
|
||||
promptExactMarkerDirective ?? extractExactMarkerDirective(allUserText);
|
||||
const exactReplyDirective = promptExactReplyDirective ?? extractExactReplyDirective(allInputText);
|
||||
const exactMarkerDirective =
|
||||
promptExactMarkerDirective ?? extractExactMarkerDirective(allInputText);
|
||||
const whatsAppLocationMarker = shouldUseWhatsAppLocationMarker(prompt)
|
||||
? extractWhatsAppLocationMarkerDirective(allInputText)
|
||||
: "";
|
||||
@@ -151,11 +164,11 @@ export function buildAssistantText(
|
||||
if (/\bmarker\b/i.test(allInputText) && promptExactReplyDirective) {
|
||||
return promptExactReplyDirective;
|
||||
}
|
||||
if (/\bmarker\b/i.test(allInputText) && exactMarkerDirective) {
|
||||
return exactMarkerDirective;
|
||||
if (/\bmarker\b/i.test(allInputText) && userExactMarkerDirective) {
|
||||
return userExactMarkerDirective;
|
||||
}
|
||||
if (/\bmarker\b/i.test(allInputText) && exactReplyDirective) {
|
||||
return exactReplyDirective;
|
||||
if (/\bmarker\b/i.test(allInputText) && userExactReplyDirective) {
|
||||
return userExactReplyDirective;
|
||||
}
|
||||
if (promptExactReplyDirective) {
|
||||
return promptExactReplyDirective;
|
||||
@@ -321,6 +334,24 @@ export function buildAssistantText(
|
||||
}
|
||||
return "";
|
||||
}
|
||||
const askUserResult = structuredToolText || toolOutput;
|
||||
const askUserDeploy = /^Deploy:\s*(.+)$/m.exec(askUserResult)?.[1]?.trim();
|
||||
const askUserChecks = /^Checks:\s*(.+)$/m
|
||||
.exec(askUserResult)?.[1]
|
||||
?.split(",")
|
||||
.map((value) => value.replace(/\s*\(Recommended\)\s*$/, "").trim())
|
||||
.filter(Boolean)
|
||||
.join(",");
|
||||
const askUserNote = /^Note:\s*(.+)$/m.exec(askUserResult)?.[1]?.trim();
|
||||
if (
|
||||
toolOutput &&
|
||||
/"status"\s*:\s*"answered"/.test(askUserResult) &&
|
||||
askUserDeploy &&
|
||||
askUserChecks &&
|
||||
askUserNote
|
||||
) {
|
||||
return `ASK-USER-ROUNDTRIP-OK | deploy=${askUserDeploy} | checks=${askUserChecks} | note=${askUserNote}`;
|
||||
}
|
||||
if (
|
||||
toolOutput &&
|
||||
(QA_TOOL_SEARCH_PROMPT_RE.test(allInputText) ||
|
||||
|
||||
@@ -215,6 +215,42 @@ export function buildQaToolSearchArgs(
|
||||
if (targetTool === "message") {
|
||||
return { action: "send", message: "runtime parity message fixture" };
|
||||
}
|
||||
if (targetTool === "ask_user") {
|
||||
return {
|
||||
questions: [
|
||||
{
|
||||
id: "deploy_target",
|
||||
header: "Deploy",
|
||||
question: "Where should this deploy?",
|
||||
options: [
|
||||
{ label: "Staging (Recommended)", description: "Safer default" },
|
||||
{ label: "Production", description: "Ship to users" },
|
||||
],
|
||||
},
|
||||
{
|
||||
id: "checks",
|
||||
header: "Checks",
|
||||
question: "Which checks should run?",
|
||||
options: [
|
||||
{ label: "Unit (Recommended)", description: "Fast focused coverage" },
|
||||
{ label: "E2E", description: "Full user-path coverage" },
|
||||
{ label: "Lint", description: "Static checks" },
|
||||
],
|
||||
multiSelect: true,
|
||||
},
|
||||
{
|
||||
id: "release_note",
|
||||
header: "Note",
|
||||
question: "Which release note label should be used?",
|
||||
options: [
|
||||
{ label: "Routine (Recommended)", description: "Standard release note" },
|
||||
{ label: "Urgent", description: "Highlight prominently" },
|
||||
],
|
||||
},
|
||||
],
|
||||
timeoutSeconds: 60,
|
||||
};
|
||||
}
|
||||
if (targetTool === "session_status") {
|
||||
return { sessionKey: "current" };
|
||||
}
|
||||
|
||||
@@ -4600,6 +4600,44 @@ describe("qa mock openai server", () => {
|
||||
expect(outputText(await response.json())).toBe(`FAKE_PLUGIN_OK ${targetTool}`);
|
||||
});
|
||||
|
||||
it("derives ask_user QA summaries from the returned answers", async () => {
|
||||
const server = await startMockServer();
|
||||
const response = await postResponses(server, {
|
||||
stream: false,
|
||||
input: [
|
||||
{
|
||||
role: "system",
|
||||
content: [
|
||||
{
|
||||
type: "input_text",
|
||||
text: "Nothing to say: entire reply exactly NO_REPLY",
|
||||
},
|
||||
],
|
||||
},
|
||||
makeUserInput(
|
||||
"QA routing marker: tool search qa check target=ask_user. Ask structured questions, then summarize their actual answers.",
|
||||
),
|
||||
{
|
||||
type: "function_call_output",
|
||||
call_id: "call_ask_user_1",
|
||||
output: JSON.stringify({
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: 'Deploy: Canary\nChecks: Lint, Unit (Recommended)\nNote: weekend-only\n\n{"status":"answered"}',
|
||||
},
|
||||
],
|
||||
}),
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(outputText(await response.json())).toBe(
|
||||
"ASK-USER-ROUNDTRIP-OK | deploy=Canary | checks=Lint,Unit | note=weekend-only",
|
||||
);
|
||||
});
|
||||
|
||||
it("plans QA tool-search failure calls with denied-input args", async () => {
|
||||
const server = await startMockServer();
|
||||
|
||||
|
||||
@@ -164,6 +164,11 @@ async function buildResponsesPayload(
|
||||
const toolJson = parseToolOutputJson(scenarioToolOutput);
|
||||
const promptExactReplyDirective = extractExactReplyDirective(prompt);
|
||||
const promptExactMarkerDirective = extractExactMarkerDirective(prompt);
|
||||
const allUserText = extractAllUserTexts(input).join("\n");
|
||||
const userExactReplyDirective =
|
||||
promptExactReplyDirective ?? extractExactReplyDirective(allUserText);
|
||||
const userExactMarkerDirective =
|
||||
promptExactMarkerDirective ?? extractExactMarkerDirective(allUserText);
|
||||
const exactReplyDirective = promptExactReplyDirective ?? extractExactReplyDirective(allInputText);
|
||||
const exactMarkerDirective =
|
||||
promptExactMarkerDirective ?? extractExactMarkerDirective(allInputText);
|
||||
@@ -728,11 +733,11 @@ async function buildResponsesPayload(
|
||||
if (/\bmarker\b/i.test(allInputText) && promptExactReplyDirective) {
|
||||
return buildAssistantEvents(promptExactReplyDirective);
|
||||
}
|
||||
if (/\bmarker\b/i.test(allInputText) && exactMarkerDirective) {
|
||||
return buildAssistantEvents(exactMarkerDirective);
|
||||
if (/\bmarker\b/i.test(allInputText) && userExactMarkerDirective) {
|
||||
return buildAssistantEvents(userExactMarkerDirective);
|
||||
}
|
||||
if (/\bmarker\b/i.test(allInputText) && exactReplyDirective) {
|
||||
return buildAssistantEvents(exactReplyDirective);
|
||||
if (/\bmarker\b/i.test(allInputText) && userExactReplyDirective) {
|
||||
return buildAssistantEvents(userExactReplyDirective);
|
||||
}
|
||||
if (QA_SKILL_WORKSHOP_REVIEW_PROMPT_RE.test(allInputText)) {
|
||||
return buildAssistantEvents(
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
title: Ask user structured question roundtrip
|
||||
|
||||
scenario:
|
||||
id: runtime-tool-ask-user
|
||||
surface: runtime-tool
|
||||
coverage:
|
||||
primary:
|
||||
- tools.ask-user
|
||||
secondary:
|
||||
- channels.qa-channel
|
||||
- gateway.chat-apis
|
||||
objective: Verify the agent can ask three structured questions, pause, receive mixed answer shapes, and resume with the selected values.
|
||||
successCriteria:
|
||||
- The model calls ask_user exactly once with single-select, multi-select, and free-text-capable questions.
|
||||
- QA Channel delivers the structured prompt while the agent run remains blocked.
|
||||
- Gateway resolution resumes the same run with the canonical selected values.
|
||||
- The final reply proves the agent received the selected single, multiple, and free-text answers.
|
||||
docsRefs:
|
||||
- docs/tools/ask-user.md
|
||||
- docs/channels/qa-channel.md
|
||||
- docs/concepts/qa-e2e-automation.md
|
||||
codeRefs:
|
||||
- src/agents/tools/ask-user-tool.ts
|
||||
- src/gateway/server-methods/question.ts
|
||||
- extensions/qa-lab/src/providers/mock-openai/mock-openai-tooling.ts
|
||||
- extensions/qa-lab/src/suite-runtime-agent-process.ts
|
||||
execution:
|
||||
kind: flow
|
||||
summary: Send an inbound chat turn, observe its ask_user prompt through QA Channel, resolve three answer shapes, and verify the resumed final reply.
|
||||
config:
|
||||
conversationId: qa-ask-user-roundtrip
|
||||
promptNeedle: Where should this deploy?
|
||||
expectedFinal: ASK-USER-ROUNDTRIP-OK | deploy=Production | checks=Unit,E2E | note=night-shift
|
||||
prompt: >-
|
||||
Structured question QA check. Call ask_user exactly once with these three questions:
|
||||
deploy_target asks "Where should this deploy?" with Staging (Recommended) first and Production second;
|
||||
checks asks "Which checks should run?" with Unit (Recommended), E2E, and Lint and allows multiple selections;
|
||||
release_note asks "Which release note label should be used?" with Routine (Recommended) and Urgent.
|
||||
Use a 60 second timeout. After the tool returns, read its actual answers and reply on one line
|
||||
using this format: `ASK-USER-ROUNDTRIP-OK | deploy=<deploy_target> | checks=<checks comma-separated without spaces or Recommended suffixes> | note=<release_note>`.
|
||||
Never invent or predict the answers.
|
||||
QA routing marker: tool search qa check target=ask_user.
|
||||
|
||||
flow:
|
||||
steps:
|
||||
- name: asks, blocks, resolves, and resumes with structured answers
|
||||
actions:
|
||||
- call: waitForGatewayHealthy
|
||||
args:
|
||||
- ref: env
|
||||
- 60000
|
||||
- call: waitForQaChannelReady
|
||||
args:
|
||||
- ref: env
|
||||
- 60000
|
||||
- call: reset
|
||||
- set: requestCursorBefore
|
||||
value:
|
||||
expr: "env.mock ? (await fetchJson(`${env.mock.baseUrl}/debug/request-cursor`)).cursor : 0"
|
||||
- set: startIndex
|
||||
value:
|
||||
expr: state.getSnapshot().messages.length
|
||||
- sendInbound:
|
||||
conversation:
|
||||
id:
|
||||
ref: config.conversationId
|
||||
kind: direct
|
||||
senderId:
|
||||
ref: config.conversationId
|
||||
senderName: QA Operator
|
||||
text:
|
||||
ref: config.prompt
|
||||
- call: waitForCondition
|
||||
saveAs: pendingQuestion
|
||||
args:
|
||||
- lambda:
|
||||
async: true
|
||||
expr: "(await env.gateway.call('question.list', {})).questions.find((question) => question.status === 'pending' && question.questions.some((entry) => entry.question === config.promptNeedle))"
|
||||
- expr: liveTurnTimeoutMs(env, 45000)
|
||||
- expr: "env.providerMode === 'mock-openai' ? 100 : 250"
|
||||
- call: waitForCondition
|
||||
saveAs: promptMessage
|
||||
args:
|
||||
- lambda:
|
||||
expr: "state.getSnapshot().messages.slice(startIndex).find((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === config.conversationId && candidate.text.includes(config.promptNeedle))"
|
||||
- expr: liveTurnTimeoutMs(env, 45000)
|
||||
- expr: "env.providerMode === 'mock-openai' ? 100 : 250"
|
||||
- assert:
|
||||
expr: "pendingQuestion.questions.length === 3 && pendingQuestion.questions[0].id === 'deploy_target' && pendingQuestion.questions[1].multiSelect === true && pendingQuestion.questions.every((question) => question.isOther === true)"
|
||||
message:
|
||||
expr: "`ask_user question shape drifted: ${JSON.stringify(pendingQuestion.questions)}`"
|
||||
- set: resolution
|
||||
value:
|
||||
expr: >-
|
||||
env.gateway.call('question.resolve', {
|
||||
id: pendingQuestion.id,
|
||||
answers: {
|
||||
answers: {
|
||||
deploy_target: { answers: ['Production'] },
|
||||
checks: { answers: ['Unit (Recommended)', 'E2E'] },
|
||||
release_note: { answers: ['night-shift'] },
|
||||
},
|
||||
},
|
||||
resolvedBy: 'qa-lab',
|
||||
})
|
||||
- assert:
|
||||
expr: "resolution.status === 'answered'"
|
||||
message:
|
||||
expr: "`ask_user resolution failed: ${JSON.stringify(resolution)}`"
|
||||
- call: waitForCondition
|
||||
saveAs: resumedProviderRequest
|
||||
args:
|
||||
- lambda:
|
||||
async: true
|
||||
expr: "env.mock ? (await fetchJson(`${env.mock.baseUrl}/debug/requests?after=${requestCursorBefore}`)).find((request) => String(request.toolOutput ?? '').includes('Deploy:')) : true"
|
||||
- expr: liveTurnTimeoutMs(env, 45000)
|
||||
- expr: "env.providerMode === 'mock-openai' ? 100 : 250"
|
||||
- assert:
|
||||
expr: "!env.mock || (String(resumedProviderRequest.toolOutput).includes('Production') && String(resumedProviderRequest.toolOutput).includes('Unit (Recommended)') && String(resumedProviderRequest.toolOutput).includes('E2E') && String(resumedProviderRequest.toolOutput).includes('night-shift'))"
|
||||
message:
|
||||
expr: "`ask_user answers did not reach the resumed provider request: ${JSON.stringify({ toolOutput: resumedProviderRequest?.toolOutput })}`"
|
||||
- call: waitForCondition
|
||||
saveAs: finalMessage
|
||||
args:
|
||||
- lambda:
|
||||
expr: "state.getSnapshot().messages.slice(startIndex).find((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === config.conversationId && candidate !== promptMessage && candidate.text === config.expectedFinal)"
|
||||
- expr: liveTurnTimeoutMs(env, 45000)
|
||||
- expr: "env.providerMode === 'mock-openai' ? 100 : 250"
|
||||
- assert:
|
||||
expr: "finalMessage.text === config.expectedFinal"
|
||||
message:
|
||||
expr: "`ask_user resumed with unexpected final text: ${JSON.stringify(finalMessage.text)}`"
|
||||
- set: terminalQuestion
|
||||
value:
|
||||
expr: "env.gateway.call('question.get', { id: pendingQuestion.id })"
|
||||
- assert:
|
||||
expr: "resolution.status === 'answered' && terminalQuestion.question.status === 'answered'"
|
||||
message:
|
||||
expr: "`ask_user question did not terminalize as answered: ${JSON.stringify({ resolution, terminalQuestion })}`"
|
||||
detailsExpr: finalMessage.text
|
||||
@@ -47,6 +47,7 @@ async function activateAskUserPrompt(toolCallId: string, args: unknown) {
|
||||
let resolveAnswer: ((value: { status: "cancelled" }) => void) | undefined;
|
||||
const tool = createAskUserTool({
|
||||
sessionKey: "agent:unit-session",
|
||||
runId: "run-test",
|
||||
gatewayCall: async (method, _opts, params) => {
|
||||
if (method === "question.request") {
|
||||
if (!params || typeof params !== "object" || !("id" in params)) {
|
||||
|
||||
@@ -156,11 +156,18 @@ function readUpdatePlanResult(
|
||||
function buildAskUserPromptPayload(
|
||||
toolCallId: string,
|
||||
sessionKey: string | undefined,
|
||||
runId: string,
|
||||
args: unknown,
|
||||
) {
|
||||
try {
|
||||
const { questions } = normalizeAskUserParams(args);
|
||||
const reservation = reserveAskUserPromptDelivery({ toolCallId, sessionKey, questions });
|
||||
const { questions, timeoutSeconds } = normalizeAskUserParams(args);
|
||||
const reservation = reserveAskUserPromptDelivery({
|
||||
toolCallId,
|
||||
sessionKey,
|
||||
runId,
|
||||
questions,
|
||||
timeoutSeconds,
|
||||
});
|
||||
if (!reservation) {
|
||||
return undefined;
|
||||
}
|
||||
@@ -973,11 +980,11 @@ export function handleToolExecutionStart(
|
||||
const startToolName = normalizeToolName(evt.toolName);
|
||||
const askUserPromptReservation =
|
||||
startToolName === "ask_user" && ctx.params.onToolResult
|
||||
? buildAskUserPromptPayload(evt.toolCallId, ctx.params.sessionKey, evt.args)
|
||||
? buildAskUserPromptPayload(evt.toolCallId, ctx.params.sessionKey, ctx.params.runId, evt.args)
|
||||
: undefined;
|
||||
const cancelAskUserPromptReservation = () => {
|
||||
if (askUserPromptReservation) {
|
||||
cancelAskUserPromptDelivery(evt.toolCallId, ctx.params.sessionKey);
|
||||
cancelAskUserPromptDelivery(evt.toolCallId, ctx.params.sessionKey, ctx.params.runId);
|
||||
}
|
||||
};
|
||||
const continueAfterBlockReplyFlush = (): void | Promise<void> => {
|
||||
@@ -1366,7 +1373,7 @@ export async function handleToolExecutionEnd(
|
||||
const hideFromChannelProgress = evt.hideFromChannelProgress === true;
|
||||
const toolCallId = evt.toolCallId;
|
||||
if (toolName === "ask_user") {
|
||||
cancelAskUserPromptDelivery(toolCallId, ctx.params.sessionKey);
|
||||
cancelAskUserPromptDelivery(toolCallId, ctx.params.sessionKey, ctx.params.runId);
|
||||
}
|
||||
const runId = ctx.params.runId;
|
||||
const isError = evt.isError;
|
||||
|
||||
@@ -591,6 +591,7 @@ export function createOpenClawTools(
|
||||
createAskUserTool({
|
||||
agentId: sessionAgentId,
|
||||
sessionKey: options?.runSessionKey ?? options?.agentSessionKey,
|
||||
runId: options?.runId,
|
||||
}),
|
||||
]
|
||||
: []),
|
||||
|
||||
@@ -58,6 +58,7 @@ describe("tool-catalog", () => {
|
||||
"create_goal",
|
||||
"update_goal",
|
||||
"update_plan",
|
||||
"ask_user",
|
||||
"skill_workshop",
|
||||
"image",
|
||||
"image_generate",
|
||||
@@ -83,6 +84,7 @@ describe("tool-catalog", () => {
|
||||
"subagents",
|
||||
"session_status",
|
||||
"message",
|
||||
"ask_user",
|
||||
"bundle-mcp",
|
||||
]);
|
||||
expect(requirePolicyAllow("minimal")).toEqual(["session_status"]);
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
* for OpenClaw-owned tools.
|
||||
*/
|
||||
import {
|
||||
ASK_USER_TOOL_DISPLAY_SUMMARY,
|
||||
CRON_TOOL_DISPLAY_SUMMARY,
|
||||
EXEC_TOOL_DISPLAY_SUMMARY,
|
||||
PROCESS_TOOL_DISPLAY_SUMMARY,
|
||||
@@ -390,6 +391,14 @@ const CORE_TOOL_DEFINITIONS: CoreToolDefinition[] = [
|
||||
profiles: ["coding"],
|
||||
includeInOpenClawGroup: true,
|
||||
},
|
||||
{
|
||||
id: "ask_user",
|
||||
label: "ask_user",
|
||||
description: ASK_USER_TOOL_DISPLAY_SUMMARY,
|
||||
sectionId: "agents",
|
||||
profiles: ["coding", "messaging"],
|
||||
includeInOpenClawGroup: true,
|
||||
},
|
||||
{
|
||||
id: "skill_workshop",
|
||||
label: "skill_workshop",
|
||||
|
||||
@@ -3,11 +3,13 @@ import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import type { UserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.types.js";
|
||||
import { steerActiveSessionWithOptionalDeliveryWait } from "../embedded-agent-runner/run/attempt.queue-message.js";
|
||||
import {
|
||||
cancelAskUserPromptDelivery,
|
||||
createAskUserTool,
|
||||
isAskUserPromptActive,
|
||||
isAskUserPromptPending,
|
||||
normalizeAskUserParams,
|
||||
reserveAskUserPromptDelivery,
|
||||
settleAskUserPromptDelivery,
|
||||
waitForAskUserPromptReady,
|
||||
} from "./ask-user-tool.js";
|
||||
import { resetPendingAskUserQuestionsForTest } from "./ask-user-tool.test-support.js";
|
||||
|
||||
@@ -101,6 +103,161 @@ describe("ask_user normalization", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("ask_user prompt delivery", () => {
|
||||
it("uses the Gateway record when the executor has isolated runtime state", async () => {
|
||||
const questions = normalizeAskUserParams(validArgs).questions;
|
||||
const reservation = reserveAskUserPromptDelivery({
|
||||
toolCallId: "call-isolated-runtime",
|
||||
sessionKey: "agent:main:isolated-runtime",
|
||||
questions,
|
||||
});
|
||||
if (!reservation) {
|
||||
throw new Error("expected prompt reservation");
|
||||
}
|
||||
const gateway = gatewayStub(async (method, _opts, params) => {
|
||||
expect(method).toBe("question.list");
|
||||
expect(params).toEqual({});
|
||||
return { questions: [{ id: reservation.questionId, status: "pending" }] };
|
||||
});
|
||||
|
||||
await expect(waitForAskUserPromptReady(reservation.questionId, gateway.call)).resolves.toEqual(
|
||||
questions,
|
||||
);
|
||||
await expect(isAskUserPromptPending(reservation.questionId, gateway.call)).resolves.toBe(true);
|
||||
});
|
||||
|
||||
it("rejects prompt delivery after the Gateway question terminalizes", async () => {
|
||||
const questions = normalizeAskUserParams(validArgs).questions;
|
||||
const reservation = reserveAskUserPromptDelivery({
|
||||
toolCallId: "call-terminal-before-delivery",
|
||||
sessionKey: "agent:main:terminal-before-delivery",
|
||||
questions,
|
||||
});
|
||||
if (!reservation) {
|
||||
throw new Error("expected prompt reservation");
|
||||
}
|
||||
const gateway = gatewayStub(async () => ({
|
||||
questions: [{ id: reservation.questionId, status: "expired" }],
|
||||
}));
|
||||
|
||||
await expect(isAskUserPromptPending(reservation.questionId, gateway.call)).resolves.toBe(false);
|
||||
});
|
||||
|
||||
it("retries a transient Gateway revalidation failure before delivering", async () => {
|
||||
const questions = normalizeAskUserParams(validArgs).questions;
|
||||
const reservation = reserveAskUserPromptDelivery({
|
||||
toolCallId: "call-revalidation-failure",
|
||||
sessionKey: "agent:main:revalidation-failure",
|
||||
questions,
|
||||
});
|
||||
if (!reservation) {
|
||||
throw new Error("expected prompt reservation");
|
||||
}
|
||||
let attempts = 0;
|
||||
const gateway = gatewayStub(async () => {
|
||||
attempts += 1;
|
||||
if (attempts === 1) {
|
||||
throw new Error("temporary Gateway disconnect");
|
||||
}
|
||||
return { questions: [{ id: reservation.questionId, status: "pending" }] };
|
||||
});
|
||||
|
||||
await expect(isAskUserPromptPending(reservation.questionId, gateway.call)).resolves.toBe(true);
|
||||
expect(attempts).toBe(2);
|
||||
});
|
||||
|
||||
it("drops a prompt when local terminalization wins during Gateway revalidation", async () => {
|
||||
const questions = normalizeAskUserParams(validArgs).questions;
|
||||
const toolCallId = "call-revalidation-terminal";
|
||||
const sessionKey = "agent:main:revalidation-terminal";
|
||||
const reservation = reserveAskUserPromptDelivery({ toolCallId, sessionKey, questions });
|
||||
if (!reservation) {
|
||||
throw new Error("expected prompt reservation");
|
||||
}
|
||||
const gateway = gatewayStub(async () => {
|
||||
cancelAskUserPromptDelivery(toolCallId, sessionKey);
|
||||
return { questions: [{ id: reservation.questionId, status: "pending" }] };
|
||||
});
|
||||
|
||||
await expect(isAskUserPromptPending(reservation.questionId, gateway.call)).resolves.toBe(false);
|
||||
});
|
||||
|
||||
it("expires prompt revalidation while the Gateway lookup remains stalled", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const questions = normalizeAskUserParams(validArgs).questions;
|
||||
const reservation = reserveAskUserPromptDelivery({
|
||||
toolCallId: "call-revalidation-stalled",
|
||||
sessionKey: "agent:main:revalidation-stalled",
|
||||
questions,
|
||||
timeoutSeconds: 30,
|
||||
});
|
||||
if (!reservation) {
|
||||
throw new Error("expected prompt reservation");
|
||||
}
|
||||
const gateway = gatewayStub(async () => await new Promise(() => {}));
|
||||
|
||||
const pending = isAskUserPromptPending(reservation.questionId, gateway.call);
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
await expect(pending).resolves.toBe(false);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("shares prompt readiness across bundled module instances", async () => {
|
||||
const runId = "run-split-module";
|
||||
const toolCallId = "call-split-module";
|
||||
const questions = normalizeAskUserParams(validArgs).questions;
|
||||
const reservation = reserveAskUserPromptDelivery({
|
||||
toolCallId,
|
||||
runId,
|
||||
sessionKey: "agent:main:subscriber-session",
|
||||
questions,
|
||||
});
|
||||
if (!reservation) {
|
||||
throw new Error("expected prompt reservation");
|
||||
}
|
||||
let finishWait: ((value: unknown) => void) | undefined;
|
||||
const gateway = gatewayStub(async (method, _opts, params) => {
|
||||
if (method === "question.request") {
|
||||
return { id: params.id };
|
||||
}
|
||||
if (method === "question.waitAnswer") {
|
||||
return await new Promise((resolve) => {
|
||||
finishWait = resolve;
|
||||
});
|
||||
}
|
||||
throw new Error(`unexpected method ${method}`);
|
||||
});
|
||||
|
||||
vi.resetModules();
|
||||
const isolatedModule = await import("./ask-user-tool.js");
|
||||
const pending = isolatedModule
|
||||
.createAskUserTool({
|
||||
runId,
|
||||
sessionKey: "agent:main:executor-session",
|
||||
gatewayCall: gateway.call,
|
||||
})
|
||||
.execute(toolCallId, validArgs);
|
||||
await vi.waitFor(() =>
|
||||
expect(gateway.mock.mock.calls.some(([method]) => method === "question.request")).toBe(true),
|
||||
);
|
||||
let readyQuestions: unknown;
|
||||
void waitForAskUserPromptReady(reservation.questionId).then((value) => {
|
||||
readyQuestions = value;
|
||||
});
|
||||
await vi.waitFor(() => expect(readyQuestions).toEqual(questions));
|
||||
|
||||
settleAskUserPromptDelivery(reservation.questionId);
|
||||
finishWait?.({
|
||||
status: "answered",
|
||||
answers: { answers: { deploy_target: { answers: ["Production"] } } },
|
||||
});
|
||||
await expect(pending).resolves.toMatchObject({ details: { status: "answered" } });
|
||||
});
|
||||
});
|
||||
|
||||
describe("ask_user execution", () => {
|
||||
it("returns answered details plus readable answer lines", async () => {
|
||||
const answers = { answers: { deploy_target: { answers: ["Staging (Recommended)"] } } };
|
||||
@@ -305,7 +462,13 @@ describe("ask_user execution", () => {
|
||||
finishRegistration?.({ id: reservation.questionId });
|
||||
|
||||
await expect(pending).rejects.toThrow("stop before registration completed");
|
||||
expect(isAskUserPromptActive(reservation.questionId)).toBe(false);
|
||||
expect(
|
||||
reserveAskUserPromptDelivery({
|
||||
toolCallId: "call-after-late-registration-abort",
|
||||
sessionKey,
|
||||
questions: normalizeAskUserParams(validArgs).questions,
|
||||
}),
|
||||
).toBeDefined();
|
||||
});
|
||||
|
||||
it("suppresses a stale prompt when an image arrives during registration", async () => {
|
||||
@@ -510,8 +673,14 @@ describe("ask_user execution", () => {
|
||||
|
||||
controller.abort(new Error("stop during delivery"));
|
||||
|
||||
expect(isAskUserPromptActive(reservation.questionId)).toBe(false);
|
||||
await expect(pending).rejects.toThrow("stop during delivery");
|
||||
expect(
|
||||
reserveAskUserPromptDelivery({
|
||||
toolCallId: "call-after-delivery-abort",
|
||||
sessionKey,
|
||||
questions: normalizeAskUserParams(validArgs).questions,
|
||||
}),
|
||||
).toBeDefined();
|
||||
expect(gateway.mock).toHaveBeenCalledWith(
|
||||
"question.resolve",
|
||||
{ timeoutMs: 10_000 },
|
||||
|
||||
@@ -16,6 +16,7 @@ const DEFAULT_ASK_USER_TIMEOUT_SECONDS = 900;
|
||||
const MIN_ASK_USER_TIMEOUT_SECONDS = 30;
|
||||
const MAX_ASK_USER_TIMEOUT_SECONDS = 3600;
|
||||
const ASK_USER_RPC_GRACE_MS = 10_000;
|
||||
const ASK_USER_PROMPT_RECHECK_MS = 50;
|
||||
const QUESTION_ID_PATTERN = /^[a-z][a-z0-9_]*$/;
|
||||
const TERMINAL_QUESTION_ERROR_REASONS = new Set([
|
||||
"QUESTION_ALREADY_TERMINAL",
|
||||
@@ -80,6 +81,7 @@ type AskUserQuestionState = {
|
||||
questionId: string;
|
||||
sessionKey: string;
|
||||
questions: QuestionRequestQuestion[];
|
||||
expiresAtMs: number;
|
||||
phase: AskUserQuestionPhase;
|
||||
gatewayCall?: AskUserGatewayCall;
|
||||
answer?: Promise<QuestionWaitAnswerResult>;
|
||||
@@ -87,7 +89,19 @@ type AskUserQuestionState = {
|
||||
waiters: Set<() => void>;
|
||||
};
|
||||
|
||||
const askUserQuestions = new Map<string, AskUserQuestionState>();
|
||||
const ASK_USER_QUESTIONS_KEY = Symbol.for("openclaw.askUserQuestions");
|
||||
const askUserGlobal = globalThis as Record<PropertyKey, unknown>;
|
||||
// Tool execution and subscriber delivery can live in separate production bundles.
|
||||
// Keep one process registry or prompt readiness never reaches the delivery waiter.
|
||||
const askUserQuestions = (() => {
|
||||
const existing = askUserGlobal[ASK_USER_QUESTIONS_KEY];
|
||||
if (existing instanceof Map) {
|
||||
return existing as Map<string, AskUserQuestionState>;
|
||||
}
|
||||
const questions = new Map<string, AskUserQuestionState>();
|
||||
askUserGlobal[ASK_USER_QUESTIONS_KEY] = questions;
|
||||
return questions;
|
||||
})();
|
||||
|
||||
type NormalizedAskUserParams = {
|
||||
questions: QuestionRequestQuestion[];
|
||||
@@ -192,8 +206,9 @@ export function normalizeAskUserParams(value: unknown): NormalizedAskUserParams
|
||||
}
|
||||
|
||||
/** Stable client-generated gateway question id shared with tool-start delivery. */
|
||||
function buildAskUserQuestionId(toolCallId: string, sessionKey?: string): string {
|
||||
const identity = `${sessionKey?.trim() ?? ""}\0${toolCallId}`;
|
||||
function buildAskUserQuestionId(toolCallId: string, sessionKey?: string, runId?: string): string {
|
||||
const owner = runId?.trim() || sessionKey?.trim() || "";
|
||||
const identity = `${owner}\0${toolCallId}`;
|
||||
return `ask_${createHash("sha256").update(identity).digest("hex").slice(0, 32)}`;
|
||||
}
|
||||
|
||||
@@ -254,13 +269,15 @@ async function waitForQuestionChange(
|
||||
export function reserveAskUserPromptDelivery(params: {
|
||||
toolCallId: string;
|
||||
sessionKey?: string;
|
||||
runId?: string;
|
||||
questions: QuestionRequestQuestion[];
|
||||
timeoutSeconds?: number;
|
||||
}): { questionId: string } | undefined {
|
||||
const sessionKey = askUserSessionKey(params.sessionKey);
|
||||
if (findAskUserQuestionForSession(sessionKey)) {
|
||||
return undefined;
|
||||
}
|
||||
const questionId = buildAskUserQuestionId(params.toolCallId, params.sessionKey);
|
||||
const questionId = buildAskUserQuestionId(params.toolCallId, params.sessionKey, params.runId);
|
||||
if (askUserQuestions.has(questionId)) {
|
||||
return undefined;
|
||||
}
|
||||
@@ -268,6 +285,7 @@ export function reserveAskUserPromptDelivery(params: {
|
||||
questionId,
|
||||
sessionKey,
|
||||
questions: params.questions,
|
||||
expiresAtMs: Date.now() + (params.timeoutSeconds ?? DEFAULT_ASK_USER_TIMEOUT_SECONDS) * 1_000,
|
||||
phase: { kind: "reserved" },
|
||||
waiters: new Set(),
|
||||
});
|
||||
@@ -277,6 +295,7 @@ export function reserveAskUserPromptDelivery(params: {
|
||||
/** Waits until policy-accepted tool execution has registered the gateway question. */
|
||||
export async function waitForAskUserPromptReady(
|
||||
questionId: string,
|
||||
gatewayCall: AskUserGatewayCall = callGatewayTool,
|
||||
): Promise<QuestionRequestQuestion[] | undefined> {
|
||||
const state = askUserQuestions.get(questionId);
|
||||
if (!state) {
|
||||
@@ -291,11 +310,84 @@ export async function waitForAskUserPromptReady(
|
||||
) {
|
||||
return state.questions;
|
||||
}
|
||||
await waitForQuestionChange(state);
|
||||
try {
|
||||
const status = await readAskUserQuestionStatus(questionId, gatewayCall);
|
||||
if (status === "pending") {
|
||||
// The executor may live in another JS realm or process. The Gateway record
|
||||
// is the cross-runtime readiness boundary when local state cannot signal.
|
||||
return state.questions;
|
||||
}
|
||||
if (typeof status === "string") {
|
||||
return undefined;
|
||||
}
|
||||
} catch {
|
||||
// Registration and local Gateway credentials may still be coming online.
|
||||
// Local state can win on the next pass; isolated runtimes retry the record.
|
||||
}
|
||||
await new Promise<void>((resolve) => {
|
||||
setTimeout(resolve, 50);
|
||||
});
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
async function readAskUserQuestionStatus(
|
||||
questionId: string,
|
||||
gatewayCall: AskUserGatewayCall,
|
||||
): Promise<string | undefined> {
|
||||
const result = await gatewayCall("question.list", { timeoutMs: ASK_USER_RPC_GRACE_MS }, {});
|
||||
const questions =
|
||||
result && typeof result === "object" && !Array.isArray(result)
|
||||
? (result as { questions?: unknown }).questions
|
||||
: undefined;
|
||||
const question = Array.isArray(questions)
|
||||
? questions.find(
|
||||
(candidate) =>
|
||||
candidate &&
|
||||
typeof candidate === "object" &&
|
||||
!Array.isArray(candidate) &&
|
||||
(candidate as { id?: unknown }).id === questionId,
|
||||
)
|
||||
: undefined;
|
||||
const status =
|
||||
question && typeof question === "object" && !Array.isArray(question)
|
||||
? (question as { status?: unknown }).status
|
||||
: undefined;
|
||||
return typeof status === "string" ? status : undefined;
|
||||
}
|
||||
|
||||
type AskUserPromptStatusRead =
|
||||
| { kind: "status"; status: string | undefined }
|
||||
| { kind: "error" }
|
||||
| { kind: "expired" };
|
||||
|
||||
async function readAskUserQuestionStatusBeforeExpiry(
|
||||
questionId: string,
|
||||
expiresAtMs: number,
|
||||
gatewayCall: AskUserGatewayCall,
|
||||
): Promise<AskUserPromptStatusRead> {
|
||||
const remainingMs = expiresAtMs - Date.now();
|
||||
if (remainingMs <= 0) {
|
||||
return { kind: "expired" };
|
||||
}
|
||||
return await new Promise<AskUserPromptStatusRead>((resolve) => {
|
||||
let settled = false;
|
||||
const finish = (result: AskUserPromptStatusRead) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
clearTimeout(expiryTimer);
|
||||
resolve(result);
|
||||
};
|
||||
const expiryTimer = setTimeout(() => finish({ kind: "expired" }), remainingMs);
|
||||
void readAskUserQuestionStatus(questionId, gatewayCall).then(
|
||||
(status) => finish({ kind: "status", status }),
|
||||
() => finish({ kind: "error" }),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/** Opens prompt delivery after question.request succeeds. */
|
||||
function markAskUserPromptReady(questionId: string, questions: QuestionRequestQuestion[]): void {
|
||||
const state = askUserQuestions.get(questionId);
|
||||
@@ -318,14 +410,65 @@ export function settleAskUserPromptDelivery(questionId: string, error?: unknown)
|
||||
);
|
||||
}
|
||||
|
||||
/** Returns whether a question-associated prompt still belongs to a blocking ask_user call. */
|
||||
export function isAskUserPromptActive(questionId: string): boolean {
|
||||
return askUserQuestions.has(questionId);
|
||||
/** Rechecks the Gateway immediately before exposing an answerable prompt. */
|
||||
export async function isAskUserPromptPending(
|
||||
questionId: string,
|
||||
gatewayCall: AskUserGatewayCall = callGatewayTool,
|
||||
): Promise<boolean> {
|
||||
const state = askUserQuestions.get(questionId);
|
||||
if (!state) {
|
||||
return false;
|
||||
}
|
||||
while (askUserQuestions.get(questionId) === state) {
|
||||
if (state.phase.kind === "resolving" || state.phase.kind === "prompt-failed") {
|
||||
return false;
|
||||
}
|
||||
const read = await readAskUserQuestionStatusBeforeExpiry(
|
||||
questionId,
|
||||
state.expiresAtMs,
|
||||
gatewayCall,
|
||||
);
|
||||
if (read.kind === "expired") {
|
||||
return false;
|
||||
}
|
||||
// Cancellation can win while the Gateway request is in flight. Recheck local
|
||||
// ownership before trusting an older remote `pending` snapshot.
|
||||
const currentState = askUserQuestions.get(questionId);
|
||||
if (
|
||||
currentState !== state ||
|
||||
currentState.phase.kind === "resolving" ||
|
||||
currentState.phase.kind === "prompt-failed"
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (read.kind === "status" && read.status === "pending") {
|
||||
return true;
|
||||
}
|
||||
if (read.kind === "status" && typeof read.status === "string") {
|
||||
return false;
|
||||
}
|
||||
if (read.kind === "error") {
|
||||
// Keep the prompt private until Gateway state is authoritative again.
|
||||
// Failing open here can expose a stale question after remote terminalization.
|
||||
}
|
||||
const remainingMs = state.expiresAtMs - Date.now();
|
||||
if (remainingMs <= 0) {
|
||||
return false;
|
||||
}
|
||||
await new Promise<void>((resolve) => {
|
||||
setTimeout(resolve, Math.min(ASK_USER_PROMPT_RECHECK_MS, remainingMs));
|
||||
});
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/** Releases a tool-start reservation when policy rejects execution. */
|
||||
export function cancelAskUserPromptDelivery(toolCallId: string, sessionKey?: string): void {
|
||||
releaseAskUserQuestion(buildAskUserQuestionId(toolCallId, sessionKey));
|
||||
export function cancelAskUserPromptDelivery(
|
||||
toolCallId: string,
|
||||
sessionKey?: string,
|
||||
runId?: string,
|
||||
): void {
|
||||
releaseAskUserQuestion(buildAskUserQuestionId(toolCallId, sessionKey, runId));
|
||||
}
|
||||
|
||||
function answeredResult(questions: readonly QuestionRequestQuestion[], answers: QuestionAnswers) {
|
||||
@@ -399,6 +542,7 @@ if (process.env.VITEST || process.env.NODE_ENV === "test") {
|
||||
export function createAskUserTool(params: {
|
||||
agentId?: string;
|
||||
sessionKey?: string;
|
||||
runId?: string;
|
||||
gatewayCall?: AskUserGatewayCall;
|
||||
}): AnyAgentTool {
|
||||
const gatewayCall: AskUserGatewayCall = params.gatewayCall ?? callGatewayTool;
|
||||
@@ -409,7 +553,7 @@ export function createAskUserTool(params: {
|
||||
description: describeAskUserTool(),
|
||||
parameters: AskUserToolSchema,
|
||||
execute: async (toolCallId, args, signal) => {
|
||||
const questionId = buildAskUserQuestionId(toolCallId, params.sessionKey);
|
||||
const questionId = buildAskUserQuestionId(toolCallId, params.sessionKey, params.runId);
|
||||
let normalized: NormalizedAskUserParams;
|
||||
try {
|
||||
signal?.throwIfAborted();
|
||||
@@ -435,12 +579,14 @@ export function createAskUserTool(params: {
|
||||
questionId,
|
||||
sessionKey,
|
||||
questions: normalized.questions,
|
||||
expiresAtMs: Date.now() + timeoutMs,
|
||||
phase: { kind: "registering" },
|
||||
gatewayCall,
|
||||
waiters: new Set(),
|
||||
} satisfies AskUserQuestionState);
|
||||
state.sessionKey = sessionKey;
|
||||
state.questions = normalized.questions;
|
||||
state.expiresAtMs = Date.now() + timeoutMs;
|
||||
state.gatewayCall = gatewayCall;
|
||||
transitionAskUserQuestion(state, { kind: "registering" });
|
||||
askUserQuestions.set(questionId, state);
|
||||
|
||||
@@ -4,6 +4,7 @@ import type { OpenClawConfig } from "../../config/config.js";
|
||||
import type { MsgContext } from "../templating.js";
|
||||
import type { GetReplyOptions, ReplyPayload } from "../types.js";
|
||||
import {
|
||||
askUserMocks,
|
||||
createDispatcher,
|
||||
emptyConfig,
|
||||
hookMocks,
|
||||
@@ -402,6 +403,60 @@ describe("dispatchReplyFromConfig", () => {
|
||||
expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("delivers ask_user prompts when verbose tool progress is disabled", async () => {
|
||||
setNoAbort();
|
||||
const payload = {
|
||||
text: "Question for you: Where should this deploy?",
|
||||
channelData: { askUser: { questionId: "question-owned-by-agent-runtime" } },
|
||||
} satisfies ReplyPayload;
|
||||
const dispatcher = createDispatcher();
|
||||
const ctx = buildTestCtx({ Provider: "telegram", ChatType: "direct" });
|
||||
const replyResolver = async (
|
||||
_ctx: MsgContext,
|
||||
opts?: GetReplyOptions,
|
||||
_cfg?: OpenClawConfig,
|
||||
) => {
|
||||
await requireToolResultHandler(opts?.onToolResult)(payload);
|
||||
return undefined;
|
||||
};
|
||||
|
||||
await dispatchReplyFromConfig({ ctx, cfg: emptyConfig, dispatcher, replyResolver });
|
||||
expect(dispatcher.sendToolResult).toHaveBeenCalledWith(payload);
|
||||
const toolDeliveryOrder =
|
||||
vi.mocked(dispatcher.sendToolResult).mock.invocationCallOrder[0] ?? Number.NEGATIVE_INFINITY;
|
||||
expect(
|
||||
vi
|
||||
.mocked(dispatcher.waitForIdle)
|
||||
.mock.invocationCallOrder.some((order) => order > toolDeliveryOrder),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("drops ask_user prompts that terminalize before dispatcher delivery", async () => {
|
||||
setNoAbort();
|
||||
askUserMocks.isAskUserPromptPending.mockResolvedValue(false);
|
||||
const payload = {
|
||||
text: "Question for you: Where should this deploy?",
|
||||
channelData: { askUser: { questionId: "question-terminal-before-delivery" } },
|
||||
} satisfies ReplyPayload;
|
||||
const dispatcher = createDispatcher();
|
||||
const ctx = buildTestCtx({ Provider: "telegram", ChatType: "direct" });
|
||||
const replyResolver = async (
|
||||
_ctx: MsgContext,
|
||||
opts?: GetReplyOptions,
|
||||
_cfg?: OpenClawConfig,
|
||||
) => {
|
||||
await requireToolResultHandler(opts?.onToolResult)(payload);
|
||||
return undefined;
|
||||
};
|
||||
|
||||
await dispatchReplyFromConfig({ ctx, cfg: emptyConfig, dispatcher, replyResolver });
|
||||
|
||||
expect(askUserMocks.isAskUserPromptPending).toHaveBeenCalledWith(
|
||||
"question-terminal-before-delivery",
|
||||
);
|
||||
expect(dispatcher.sendToolResult).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not synthesize hidden text-only tool summaries into TTS media", async () => {
|
||||
setNoAbort();
|
||||
ttsMocks.state.synthesizeToolAudio = true;
|
||||
|
||||
@@ -40,6 +40,9 @@ const mocks = vi.hoisted(() => ({
|
||||
const globalMocks = vi.hoisted(() => ({
|
||||
logVerbose: vi.fn(),
|
||||
}));
|
||||
const askUserMocks = vi.hoisted(() => ({
|
||||
isAskUserPromptPending: vi.fn(async () => true),
|
||||
}));
|
||||
const diagnosticMocks = vi.hoisted(() => ({
|
||||
logMessageDispatchCompleted: vi.fn(),
|
||||
logMessageDispatchStarted: vi.fn(),
|
||||
@@ -369,6 +372,7 @@ export {
|
||||
acpManagerRuntimeMocks,
|
||||
acpMocks,
|
||||
agentEventMocks,
|
||||
askUserMocks,
|
||||
diagnosticMocks,
|
||||
globalMocks,
|
||||
hookMocks,
|
||||
@@ -438,6 +442,10 @@ vi.mock("../../globals.js", async (importOriginal) => {
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("../../agents/tools/ask-user-tool.js", () => ({
|
||||
isAskUserPromptPending: askUserMocks.isAskUserPromptPending,
|
||||
}));
|
||||
|
||||
vi.mock("../../logging/diagnostic.js", () => ({
|
||||
logMessageDispatchCompleted: diagnosticMocks.logMessageDispatchCompleted,
|
||||
logMessageDispatchStarted: diagnosticMocks.logMessageDispatchStarted,
|
||||
@@ -688,6 +696,7 @@ export function createDispatcher(): ReplyDispatcher {
|
||||
}
|
||||
|
||||
export function resetPluginTtsAndThreadMocks() {
|
||||
askUserMocks.isAskUserPromptPending.mockReset().mockResolvedValue(true);
|
||||
pluginConversationBindingMocks.shownFallbackNoticeBindingIds.clear();
|
||||
ttsMocks.state.synthesizeFinalAudio = false;
|
||||
ttsMocks.state.synthesizeToolAudio = false;
|
||||
|
||||
@@ -30,7 +30,7 @@ import {
|
||||
} from "../../agents/subagent-capabilities.js";
|
||||
import { isToolAllowedByPolicies } from "../../agents/tool-policy-match.js";
|
||||
import { mergeAlsoAllowPolicy, resolveToolProfilePolicy } from "../../agents/tool-policy.js";
|
||||
import { isAskUserPromptActive } from "../../agents/tools/ask-user-tool.js";
|
||||
import { isAskUserPromptPending } from "../../agents/tools/ask-user-tool.js";
|
||||
import {
|
||||
resolveConversationBindingRecord,
|
||||
touchConversationBindingRecord,
|
||||
@@ -1400,13 +1400,13 @@ async function dispatchReplyFromConfigInner(
|
||||
const askUser = payload.channelData?.askUser;
|
||||
return askUser && typeof askUser === "object" && !Array.isArray(askUser);
|
||||
};
|
||||
const isInactiveAskUserPayload = (payload: ReplyPayload) => {
|
||||
const readAskUserQuestionId = (payload: ReplyPayload) => {
|
||||
const askUser = payload.channelData?.askUser;
|
||||
if (!askUser || typeof askUser !== "object" || Array.isArray(askUser)) {
|
||||
return false;
|
||||
return undefined;
|
||||
}
|
||||
const questionId = (askUser as { questionId?: unknown }).questionId;
|
||||
return typeof questionId === "string" && !isAskUserPromptActive(questionId);
|
||||
return typeof questionId === "string" ? questionId : undefined;
|
||||
};
|
||||
const shouldSuppressLateTextOnlyToolProgress = (payload: ReplyPayload) => {
|
||||
if (!finalReplyDeliveryStarted) {
|
||||
@@ -2251,18 +2251,12 @@ async function dispatchReplyFromConfigInner(
|
||||
if (isDispatchOperationAborted()) {
|
||||
return;
|
||||
}
|
||||
if (isInactiveAskUserPayload(payload)) {
|
||||
return;
|
||||
}
|
||||
await waitForPendingDirectBlockReplyDelivery(
|
||||
getDispatchAbortOperation()?.abortSignal,
|
||||
);
|
||||
if (isDispatchOperationAborted()) {
|
||||
return;
|
||||
}
|
||||
if (isInactiveAskUserPayload(payload)) {
|
||||
return;
|
||||
}
|
||||
markInboundDedupeReplayUnsafe();
|
||||
// Buffered commentary preceded this tool; land it before the summary.
|
||||
await flushPendingCommentaryProgress();
|
||||
@@ -2305,7 +2299,8 @@ async function dispatchReplyFromConfigInner(
|
||||
if (
|
||||
shouldSuppressProgressDelivery() &&
|
||||
!isFastModeAutoProgressDelivery &&
|
||||
!isForcedToolProgress
|
||||
!isForcedToolProgress &&
|
||||
!hasAskUserPayload(payload)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
@@ -2334,9 +2329,6 @@ async function dispatchReplyFromConfigInner(
|
||||
if (isDispatchOperationAborted()) {
|
||||
return;
|
||||
}
|
||||
if (isInactiveAskUserPayload(deliveryPayload)) {
|
||||
return;
|
||||
}
|
||||
if (
|
||||
shouldSuppressLateTextOnlyToolProgress(deliveryPayload) &&
|
||||
!isFastModeAutoProgressPayload(deliveryPayload) &&
|
||||
@@ -2354,18 +2346,40 @@ async function dispatchReplyFromConfigInner(
|
||||
) {
|
||||
const hasMedia =
|
||||
resolveSendableOutboundReplyParts(deliveryPayload).hasMedia;
|
||||
if (!hasMedia && !hasExecApprovalPayload(deliveryPayload)) {
|
||||
if (
|
||||
!hasMedia &&
|
||||
!hasExecApprovalPayload(deliveryPayload) &&
|
||||
!hasAskUserPayload(deliveryPayload)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (deliveryPayload.isError === true) {
|
||||
markVisibleToolErrorProgress();
|
||||
}
|
||||
const askUserQuestionId = readAskUserQuestionId(deliveryPayload);
|
||||
if (
|
||||
askUserQuestionId !== undefined &&
|
||||
!(await isAskUserPromptPending(askUserQuestionId))
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (isDispatchOperationAborted()) {
|
||||
return;
|
||||
}
|
||||
if (shouldRouteToOriginating) {
|
||||
await sendPayloadAsync(deliveryPayload, undefined, false);
|
||||
} else {
|
||||
markInboundDedupeReplayUnsafe();
|
||||
dispatcher.sendToolResult(deliveryPayload);
|
||||
const delivered = dispatcher.sendToolResult(deliveryPayload);
|
||||
if (delivered && hasAskUserPayload(deliveryPayload)) {
|
||||
// ask_user blocks until this callback resolves; drain its prompt now
|
||||
// or the answerable UI can remain queued behind the blocked agent run.
|
||||
await waitForReplyDispatcherIdle(
|
||||
dispatcher,
|
||||
getDispatchAbortOperation()?.abortSignal,
|
||||
);
|
||||
}
|
||||
}
|
||||
};
|
||||
return run();
|
||||
|
||||
@@ -1626,6 +1626,7 @@ surfaces:
|
||||
runtime.codex-native-workspace.read,
|
||||
runtime.prompt-compatibility,
|
||||
runtime.tool-continuity,
|
||||
tools.ask-user,
|
||||
tools.apply-patch,
|
||||
tools.edit,
|
||||
tools.evidence,
|
||||
|
||||
Reference in New Issue
Block a user