diff --git a/backend/routes/test_runs.py b/backend/routes/test_runs.py index 64caa3b..2dbc237 100644 --- a/backend/routes/test_runs.py +++ b/backend/routes/test_runs.py @@ -154,6 +154,14 @@ async def create_test_run( raise HTTPException(422, f"用例“{row.name}”校验失败: {exc}") from exc definitions_by_id[row.id] = definition has_image_input = has_image_input or bool(image_asset_ids(definition)) + for simulation in definition.tool_simulations: + tool = tools_by_id.get(simulation.tool_id) + if tool is None or tool.function_name != simulation.function_name: + raise HTTPException( + 422, + f"用例“{row.name}”引用的工具 {simulation.function_name} " + "不属于当前被测助手", + ) for turn in definition.turns: for behavior in turn.behaviors: if behavior.type != "tool_call": diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index 219b2c9..43ebab2 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -22,7 +22,7 @@ from models import ( RuntimeToolBinding, ) from services.node_specs import copy_graph, graph_references -from sqlalchemy import select +from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession @@ -135,24 +135,56 @@ def _runtime_tool_ids( ) +def _implicit_startup_tool_names(startup: dict | None) -> list[str]: + """Return platform client tools used by built-in startup experiences. + + A confirmation opening is configured directly on the assistant instead of + through an AssistantToolBinding, but text tests still need to simulate the + underlying client interaction explicitly. + """ + + value = startup or {} + opening_mode = value.get("opening_mode") or value.get("openingMode") + opening_message = value.get("opening_message") or value.get("openingMessage") + if opening_mode == "confirmation" and isinstance(opening_message, dict): + return ["show_message"] + return [] + + async def _tools_for( session: AsyncSession, assistant: Assistant, tool_ids: list[str], + implicit_function_names: list[str] | None = None, ) -> list[RuntimeTool]: - if assistant.type not in {"prompt", "workflow"} or not tool_ids: + function_names = list(dict.fromkeys(implicit_function_names or [])) + if assistant.type not in {"prompt", "workflow"} or not ( + tool_ids or function_names + ): return [] + identity_filters = [] + if tool_ids: + identity_filters.append(Tool.id.in_(tool_ids)) + if function_names: + identity_filters.append(Tool.function_name.in_(function_names)) rows = ( await session.execute( select(Tool) .where( - Tool.id.in_(tool_ids), + or_(*identity_filters), Tool.status == "active", ) ) ).scalars().all() tool_by_id = {tool.id: tool for tool in rows} + tool_by_function = {tool.function_name: tool for tool in rows} tools = [tool_by_id[tool_id] for tool_id in tool_ids if tool_id in tool_by_id] + seen_tool_ids = {tool.id for tool in tools} + for function_name in function_names: + tool = tool_by_function.get(function_name) + if tool is not None and tool.id not in seen_tool_ids: + tools.append(tool) + seen_tool_ids.add(tool.id) server_ids = { str(tool.mcp_server_id) for tool in tools @@ -268,7 +300,12 @@ async def resolve_runtime_config( prompt_tool_bindings = await _bound_tools(session, assistant.id) runtime_tool_ids = _runtime_tool_ids(prompt_tool_bindings, assistant.startup or {}) - runtime_tools = await _tools_for(session, assistant, runtime_tool_ids) + runtime_tools = await _tools_for( + session, + assistant, + runtime_tool_ids, + _implicit_startup_tool_names(assistant.startup or {}), + ) return AssistantConfig( name=assistant.name, diff --git a/backend/services/test_runs/mock_tools.py b/backend/services/test_runs/mock_tools.py index c20911d..b3dda60 100644 --- a/backend/services/test_runs/mock_tools.py +++ b/backend/services/test_runs/mock_tools.py @@ -1,4 +1,4 @@ -"""Test-only tool executor that never reaches external side effects.""" +"""Strict, deterministic tool simulations for text test runs.""" from __future__ import annotations @@ -9,25 +9,141 @@ from typing import Any from models import RuntimeTool, RuntimeToolBinding from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutionError, ToolExecutor -from test_schemas import ToolCallExpectedBehavior +from test_schemas import ToolSimulation, ToolSimulationResponse, ToolSimulationStep + + +TOOL_SIMULATION_ERROR_CODES = { + "UNMOCKED_TOOL_CALL", + "TOOL_SIMULATION_EXHAUSTED", + "TOOL_SIMULATION_SCOPE_MISMATCH", +} + + +class ToolSimulationError(ToolExecutionError): + """A non-retryable test configuration error raised at tool-call time.""" + + def __init__(self, code: str, message: str): + self.code = code + super().__init__(f"{code}: {message}") + + +class ToolSimulationResolver: + """Resolve one case's fixed or ordered simulated tool responses.""" + + def __init__(self, simulations: list[ToolSimulation]): + self._by_tool_id = {item.tool_id: item for item in simulations} + self._by_function = {item.function_name: item for item in simulations} + self._positions = {item.tool_id: 0 for item in simulations} + self._phase = "startup" + self._turn_id: str | None = None + self._failure: ToolSimulationError | None = None + + def enter_startup(self) -> None: + self._phase = "startup" + self._turn_id = None + + def enter_turn(self, turn_id: str) -> None: + self._phase = "turn" + self._turn_id = turn_id + + async def resolve( + self, + *, + tool_id: str | None, + function_name: str, + ) -> dict[str, Any]: + simulation = ( + self._by_tool_id.get(tool_id or "") + or self._by_function.get(function_name) + ) + if simulation is None: + raise self._record_failure( + "UNMOCKED_TOOL_CALL", + f"工具 {function_name} 未配置工具模拟", + ) + + response: ToolSimulationResponse | ToolSimulationStep + if simulation.mode == "fixed": + response = simulation.response + else: + position = self._positions[simulation.tool_id] + if position >= len(simulation.steps): + raise self._record_failure( + "TOOL_SIMULATION_EXHAUSTED", + f"工具 {function_name} 的模拟响应已耗尽(第 {position + 1} 次调用)", + ) + step = simulation.steps[position] + if not self._scope_matches(step): + raise self._record_failure( + "TOOL_SIMULATION_SCOPE_MISMATCH", + ( + f"工具 {function_name} 的第 {position + 1} 条模拟步骤" + f"期望发生在{self._scope_label(step)}," + f"实际发生在{self._current_scope_label()}" + ), + ) + response = step + self._positions[simulation.tool_id] = position + 1 + + if response.delay_ms: + await asyncio.sleep(response.delay_ms / 1000) + return self._normalized_result(response) + + @property + def failure(self) -> ToolSimulationError | None: + return self._failure + + def raise_if_failed(self) -> None: + """Re-raise errors swallowed by a higher-level client message stage.""" + + if self._failure is not None: + raise self._failure + + def _record_failure(self, code: str, message: str) -> ToolSimulationError: + error = ToolSimulationError(code, message) + if self._failure is None: + self._failure = error + return error + + def _scope_matches(self, step: ToolSimulationStep) -> bool: + if step.scope.phase == "any": + return True + if step.scope.phase == "startup": + return self._phase == "startup" + return self._phase == "turn" and step.scope.turn_id == self._turn_id + + @staticmethod + def _scope_label(step: ToolSimulationStep) -> str: + if step.scope.phase == "any": + return "任意阶段" + if step.scope.phase == "startup": + return "启动阶段" + return f"轮次 {step.scope.turn_id}" + + def _current_scope_label(self) -> str: + if self._phase == "startup": + return "启动阶段" + return f"轮次 {self._turn_id or 'unknown'}" + + @staticmethod + def _normalized_result(response: ToolSimulationResponse) -> dict[str, Any]: + payload = json.loads(response.body) + status = "ok" if response.outcome == "success" else "error" + if isinstance(payload, dict): + return {**payload, "status": status} + return {"status": status, "data": payload} class MockToolExecutor(ToolExecutor): - """Resolve the active turn's configured mocks and reject everything else.""" + """Execute external runtime tools exclusively through one test resolver.""" - def __init__(self, store: DynamicVariableStore): - super().__init__(store) - self._mocks_by_tool_id: dict[str, ToolCallExpectedBehavior] = {} - self._mocks_by_function: dict[str, ToolCallExpectedBehavior] = {} - self.unmocked_calls: list[str] = [] - - def set_turn_behaviors( + def __init__( self, - behaviors: list[ToolCallExpectedBehavior], - ) -> None: - self._mocks_by_tool_id = {item.tool_id: item for item in behaviors} - self._mocks_by_function = {item.function_name: item for item in behaviors} - self.unmocked_calls = [] + store: DynamicVariableStore, + resolver: ToolSimulationResolver, + ): + super().__init__(store) + self._resolver = resolver async def execute( self, @@ -37,26 +153,10 @@ class MockToolExecutor(ToolExecutor): binding: RuntimeToolBinding | None = None, result_assignments: dict[str, str] | None = None, ) -> dict[str, Any]: - behavior = self._mocks_by_tool_id.get(tool.id) or self._mocks_by_function.get( - tool.function_name + result = await self._resolver.resolve( + tool_id=tool.id, + function_name=tool.function_name, ) - if behavior is None: - self.unmocked_calls.append(tool.function_name) - raise ToolExecutionError( - f"UNMOCKED_TOOL_CALL: 工具 {tool.function_name} 未配置 Mock 返回值" - ) - - delay_seconds = behavior.mock_response.delay_ms / 1000 - if delay_seconds: - await asyncio.sleep(delay_seconds) - - payload = json.loads(behavior.mock_response.body) - status = "ok" if behavior.mock_response.outcome == "success" else "error" - if isinstance(payload, dict): - result: dict[str, Any] = dict(payload) - result["status"] = status - else: - result = {"status": status, "data": payload} if result.get("status") != "ok": return {**result, "updated_variables": []} return self._apply_result_assignments( @@ -71,13 +171,16 @@ class MockToolExecutor(ToolExecutor): class TextClientToolPort: - """Automatically acknowledge built-in message stages in text tests.""" + """Return configured client-tool results without a browser transport.""" - def __init__(self) -> None: + def __init__(self, resolver: ToolSimulationResolver) -> None: + self._resolver = resolver self.auto_confirmed_messages: list[dict[str, Any]] = [] - def supports(self, function_name: str) -> bool: - return function_name == "show_message" + def supports(self, _function_name: str) -> bool: + # Test runs advertise the virtual client, then fail strictly in call(). + # Returning False would silently skip an unconfigured client tool. + return True async def call( self, @@ -85,12 +188,31 @@ class TextClientToolPort: arguments: dict[str, Any], **_kwargs: Any, ) -> dict[str, Any]: - if function_name == "show_message": - self.auto_confirmed_messages.append(dict(arguments)) - return { - "status": "ok", - "data": {"action": "confirmed", "arguments": arguments}, - } - raise ToolExecutionError( - f"UNMOCKED_CLIENT_TOOL: 客户端工具 {function_name} 不能脱离 Mock 执行" + resolved = await self._resolver.resolve( + tool_id=None, + function_name=function_name, ) + status = str(resolved.get("status") or "error") + payload = ( + resolved.get("data") + if set(resolved).issubset({"status", "data"}) and "data" in resolved + else {key: value for key, value in resolved.items() if key != "status"} + ) + # ClientToolPort mirrors the browser transport envelope. The JSON + # configured in the editor remains the client handler's root object. + result: dict[str, Any] = {"status": status, "data": payload} + if ( + status != "ok" + and isinstance(payload, dict) + and payload.get("message") is not None + ): + result["message"] = str(payload["message"]) + data = result.get("data") + action = data.get("action") if isinstance(data, dict) else None + if ( + function_name == "show_message" + and result.get("status") == "ok" + and action == "confirmed" + ): + self.auto_confirmed_messages.append(dict(arguments)) + return result diff --git a/backend/services/test_runs/text_runner.py b/backend/services/test_runs/text_runner.py index 45c232d..a7ef68e 100644 --- a/backend/services/test_runs/text_runner.py +++ b/backend/services/test_runs/text_runner.py @@ -21,7 +21,6 @@ from pipecat.frames.frames import ( ErrorFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, - LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesAppendFrame, @@ -54,7 +53,13 @@ from services.pipecat.processors import ( from services.pipecat.service_factory import config_with_resource from services.pipecat.workflow_services import build_workflow_llm_switcher from services.test_runs.errors import TestExecutionError -from services.test_runs.mock_tools import MockToolExecutor, TextClientToolPort +from services.test_runs.mock_tools import ( + TOOL_SIMULATION_ERROR_CODES, + MockToolExecutor, + TextClientToolPort, + ToolSimulationError, + ToolSimulationResolver, +) from services.vision import build_image_user_message from test_schemas import ContextTurn, FixedInputTurn, TestCaseDefinition, TestImageRef @@ -197,18 +202,21 @@ class _TextCaptureProcessor(FrameProcessor): """Release every wait even when Pipecat reports the error upstream.""" message = frame.error or "Pipeline 执行失败" - unmocked_tool = "UNMOCKED_TOOL_CALL" in message + simulation_code = next( + (code for code in TOOL_SIMULATION_ERROR_CODES if code in message), + None, + ) self.error = TestExecutionError( - code="UNMOCKED_TOOL_CALL" if unmocked_tool else "PIPELINE_ERROR", + code=simulation_code or "PIPELINE_ERROR", message=message, stage=( "tool" - if unmocked_tool + if simulation_code else "model" if frame.processor else "pipeline" ), - retryable=not unmocked_tool, + retryable=simulation_code is None, ) self.window.pending_tools.clear() self.window.touch() @@ -573,14 +581,12 @@ class TextPipelineRunner: async def on_pipeline_error(_worker, frame): capture.fail(frame) - executor_holder: dict[str, MockToolExecutor] = {} - text_client_tools = TextClientToolPort() + simulation_resolver = ToolSimulationResolver(definition.tool_simulations) + text_client_tools = TextClientToolPort(simulation_resolver) workflow_vision_scope: dict[str, Any] = {"enabled": False} def create_test_executor(store) -> MockToolExecutor: - executor = MockToolExecutor(store) - executor_holder["executor"] = executor - return executor + return MockToolExecutor(store, simulation_resolver) async def queue_call_end(reason: str) -> None: logger.debug(f"文本测试收到会话结束请求: {reason}") @@ -738,8 +744,6 @@ class TextPipelineRunner: tool_executor_factory=create_test_executor, ), ) - executor = executor_holder["executor"] - # A transportless text run cannot synthesize audio, but the configured # greeting has already completed before scripted inputs begin. Mirror # production's provider-safe context fact so the model does not repeat it. @@ -772,25 +776,11 @@ class TextPipelineRunner: # response is therefore the equivalent of playback ending. await call_end.finish() - all_tool_behaviors = [ - behavior - for turn in definition.turns - for behavior in turn.behaviors - if behavior.type == "tool_call" - ] - executor.set_turn_behaviors(all_tool_behaviors) - runner_task: asyncio.Task[None] | None = None try: startup_runtime_record_marker = len(capture.runtime_records) + simulation_resolver.enter_startup() await brain.run_preflight() - if executor.unmocked_calls: - raise TestExecutionError( - code="UNMOCKED_TOOL_CALL", - message="启动阶段调用了未配置 Mock 的工具: " - + "、".join(executor.unmocked_calls), - stage="tool", - ) await runner.add_workers(worker) runner_task = asyncio.create_task( @@ -803,12 +793,14 @@ class TextPipelineRunner: marker = capture.window.activity_count await brain.on_connected(greeting_pending=False) await brain.on_client_ready() + simulation_resolver.raise_if_failed() await _wait_for_startup( worker, capture, runner_task, marker, ) + simulation_resolver.raise_if_failed() startup_greeting = _resolve_startup_greeting( greeting, capture.opening_speech_outputs, @@ -844,7 +836,7 @@ class TextPipelineRunner: worker, runner_task, capture, - executor, + simulation_resolver, turn, index, image_assets=image_assets, @@ -873,9 +865,24 @@ class TextPipelineRunner: ) except TestExecutionError: raise + except ToolSimulationError as exc: + raise TestExecutionError( + code=exc.code, + message=str(exc), + stage="tool", + retryable=False, + ) from exc except asyncio.CancelledError: raise except Exception as exc: + if simulation_resolver.failure is not None: + error = simulation_resolver.failure + raise TestExecutionError( + code=error.code, + message=str(error), + stage="tool", + retryable=False, + ) from exc raise TestExecutionError( code="PIPELINE_EXECUTION_FAILED", message=str(exc) or type(exc).__name__, @@ -894,17 +901,14 @@ class TextPipelineRunner: worker: PipelineWorker, runner_task: asyncio.Task[None], capture: _TextCaptureProcessor, - executor: MockToolExecutor, + simulation_resolver: ToolSimulationResolver, turn: FixedInputTurn, index: int, *, image_assets: dict[str, bytes], workflow_vision_scope: dict[str, Any], ) -> RawTurnResult: - tool_behaviors = [ - behavior for behavior in turn.behaviors if behavior.type == "tool_call" - ] - executor.set_turn_behaviors(tool_behaviors) + simulation_resolver.enter_turn(turn.id) activity_marker = capture.window.activity_count output_marker = len(capture.outputs) tool_marker = len(capture.tool_calls) @@ -942,13 +946,7 @@ class TextPipelineRunner: LLMMessagesAppendFrame(messages=[user_message], run_llm=True) ) await _wait_for_quiescence(capture, runner_task, activity_marker) - if executor.unmocked_calls: - raise TestExecutionError( - code="UNMOCKED_TOOL_CALL", - message="本轮调用了未配置 Mock 的工具: " - + "、".join(executor.unmocked_calls), - stage="tool", - ) + simulation_resolver.raise_if_failed() return RawTurnResult( id=turn.id, index=index, diff --git a/backend/test_schemas.py b/backend/test_schemas.py index 1fb1c77..0c5e8b3 100644 --- a/backend/test_schemas.py +++ b/backend/test_schemas.py @@ -77,7 +77,7 @@ class ToolParamAssertion(CamelModel): return self -class ToolMockResponse(CamelModel): +class ToolSimulationResponse(CamelModel): outcome: Literal["success", "error"] = "success" body: str = Field(default='{"status":"ok"}', max_length=100_000) delay_ms: int = Field(default=0, ge=0, le=60_000) @@ -93,6 +93,63 @@ class ToolMockResponse(CamelModel): return self +class AnyToolSimulationScope(CamelModel): + phase: Literal["any"] = "any" + + +class StartupToolSimulationScope(CamelModel): + phase: Literal["startup"] = "startup" + + +class TurnToolSimulationScope(CamelModel): + phase: Literal["turn"] = "turn" + turn_id: str = Field(min_length=1, max_length=128) + + +ToolSimulationScope = Annotated[ + Union[ + AnyToolSimulationScope, + StartupToolSimulationScope, + TurnToolSimulationScope, + ], + Field(discriminator="phase"), +] + + +class ToolSimulationStep(ToolSimulationResponse): + id: str = Field(min_length=1, max_length=128) + scope: ToolSimulationScope = Field(default_factory=AnyToolSimulationScope) + + +class ToolSimulationBase(CamelModel): + tool_id: str = Field(min_length=1, max_length=40) + function_name: str = Field(min_length=1, max_length=128) + + @field_validator("tool_id", "function_name") + @classmethod + def strip_tool_identity(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("必须选择有效的工具") + return value + + +class FixedToolSimulation(ToolSimulationBase): + mode: Literal["fixed"] = "fixed" + response: ToolSimulationResponse = Field(default_factory=ToolSimulationResponse) + + +class SequenceToolSimulation(ToolSimulationBase): + mode: Literal["sequence"] = "sequence" + steps: list[ToolSimulationStep] = Field(min_length=1, max_length=10_000) + + +ToolSimulation = Annotated[ + Union[FixedToolSimulation, SequenceToolSimulation], + Field(discriminator="mode"), +] + + class ReplyExpectedBehavior(CamelModel): id: str = Field(min_length=1, max_length=128) type: Literal["reply"] = "reply" @@ -122,7 +179,6 @@ class ToolCallExpectedBehavior(CamelModel): min_calls: int = Field(default=1, ge=0, le=100) max_calls: int | None = Field(default=None, ge=0, le=100) param_assertions: list[ToolParamAssertion] = Field(default_factory=list, max_length=50) - mock_response: ToolMockResponse = Field(default_factory=ToolMockResponse) @model_validator(mode="after") def validate_call_expectation(self): @@ -191,14 +247,176 @@ class OverallCriterion(CamelModel): class TestCaseDefinition(CamelModel): context_turns: list[ContextTurn] = Field(default_factory=list, max_length=100) turns: list[FixedInputTurn] = Field(min_length=1, max_length=100) + tool_simulations: list[ToolSimulation] = Field(default_factory=list, max_length=100) overall_criteria: list[OverallCriterion] = Field(default_factory=list, max_length=50) + @model_validator(mode="before") + @classmethod + def migrate_legacy_tool_mocks(cls, raw): + """Normalize the former per-assertion mockResponse contract on read.""" + + if not isinstance(raw, dict): + return raw + if "tool_simulations" in raw or "toolSimulations" in raw: + return raw + + groups: dict[tuple[str, str], list[dict[str, Any]]] = {} + for turn_index, turn in enumerate(raw.get("turns") or []): + if not isinstance(turn, dict): + continue + turn_id = str(turn.get("id") or "").strip() + for behavior_index, behavior in enumerate(turn.get("behaviors") or []): + if not isinstance(behavior, dict): + continue + if behavior.get("type") != "tool_call": + continue + if behavior.get("expectation") != "called": + continue + mock = behavior.get("mock_response") or behavior.get("mockResponse") + if not isinstance(mock, dict): + continue + tool_id = str( + behavior.get("tool_id") or behavior.get("toolId") or "" + ).strip() + function_name = str( + behavior.get("function_name") + or behavior.get("functionName") + or "" + ).strip() + if not tool_id or not function_name: + continue + response = { + "outcome": str(mock.get("outcome") or "success"), + "body": str(mock.get("body") or '{"status":"ok"}'), + "delayMs": int( + mock.get("delay_ms") + if mock.get("delay_ms") is not None + else mock.get("delayMs") or 0 + ), + } + groups.setdefault((tool_id, function_name), []).append( + { + "response": response, + "turnId": turn_id, + "turnIndex": turn_index, + "behaviorIndex": behavior_index, + "minCalls": max( + 1, + int( + behavior.get("min_calls") + if behavior.get("min_calls") is not None + else behavior.get("minCalls") or 1 + ), + ), + } + ) + + migrated: list[dict[str, Any]] = [] + for (tool_id, function_name), entries in groups.items(): + response_signatures = { + json.dumps(entry["response"], ensure_ascii=False, sort_keys=True) + for entry in entries + } + if len(response_signatures) == 1: + migrated.append( + { + "toolId": tool_id, + "functionName": function_name, + "mode": "fixed", + "response": entries[0]["response"], + } + ) + continue + + steps: list[dict[str, Any]] = [] + for entry in entries: + for repeat_index in range(entry["minCalls"]): + steps.append( + { + "id": ( + f"legacy_step_{entry['turnIndex'] + 1}_" + f"{entry['behaviorIndex'] + 1}_{repeat_index + 1}" + ), + "scope": { + "phase": "turn", + "turnId": entry["turnId"], + }, + **entry["response"], + } + ) + migrated.append( + { + "toolId": tool_id, + "functionName": function_name, + "mode": "sequence", + "steps": steps, + } + ) + + normalized = dict(raw) + normalized["toolSimulations"] = migrated + return normalized + @model_validator(mode="after") def validate_has_expectation(self): if not any(turn.behaviors for turn in self.turns) and not self.overall_criteria: raise ValueError("至少配置一项有效的预期行为或整体评估标准") return self + @model_validator(mode="after") + def validate_tool_simulations(self): + tool_ids: set[str] = set() + step_ids: set[str] = set() + turn_ids = {turn.id for turn in self.turns} + simulations_by_tool: dict[str, ToolSimulation] = {} + + for simulation in self.tool_simulations: + if simulation.tool_id in tool_ids: + raise ValueError( + f"工具 {simulation.function_name} 只能配置一条工具模拟" + ) + tool_ids.add(simulation.tool_id) + simulations_by_tool[simulation.tool_id] = simulation + if simulation.mode != "sequence": + continue + for step in simulation.steps: + if step.id in step_ids: + raise ValueError(f"工具模拟步骤 ID 重复: {step.id}") + step_ids.add(step.id) + if step.scope.phase == "turn" and step.scope.turn_id not in turn_ids: + raise ValueError( + f"工具模拟步骤引用了不存在的轮次: {step.scope.turn_id}" + ) + + for turn in self.turns: + for behavior in turn.behaviors: + if behavior.type != "tool_call" or behavior.expectation != "called": + continue + simulation = simulations_by_tool.get(behavior.tool_id) + if simulation is None: + raise ValueError( + f"工具 {behavior.function_name} 配置了应调用断言,但缺少工具模拟" + ) + if simulation.function_name != behavior.function_name: + raise ValueError( + f"工具 {behavior.function_name} 的断言与模拟不匹配" + ) + if simulation.mode == "fixed": + continue + covers_turn = any( + step.scope.phase == "any" + or ( + step.scope.phase == "turn" + and step.scope.turn_id == turn.id + ) + for step in simulation.steps + ) + if not covers_turn: + raise ValueError( + f"工具 {behavior.function_name} 缺少覆盖本轮的模拟步骤" + ) + return self + @model_validator(mode="after") def validate_tool_context_pairs(self): pending: list[tuple[str | None, str]] = [] @@ -275,6 +493,7 @@ class TestCaseWrite(CamelModel): input_mode: TestCaseInputMode = "fixed_script_text" context_turns: list[ContextTurn] = Field(default_factory=list, max_length=100) turns: list[FixedInputTurn] = Field(min_length=1, max_length=100) + tool_simulations: list[ToolSimulation] = Field(default_factory=list, max_length=100) overall_criteria: list[OverallCriterion] = Field(default_factory=list, max_length=50) @field_validator("name") @@ -292,6 +511,7 @@ class TestCaseWrite(CamelModel): TestCaseDefinition( context_turns=self.context_turns, turns=self.turns, + tool_simulations=self.tool_simulations, overall_criteria=self.overall_criteria, ) return self @@ -300,6 +520,7 @@ class TestCaseWrite(CamelModel): return TestCaseDefinition( context_turns=self.context_turns, turns=self.turns, + tool_simulations=self.tool_simulations, overall_criteria=self.overall_criteria, ) diff --git a/backend/tests/test_debug_auto_test.py b/backend/tests/test_debug_auto_test.py index 45a3fb1..4121990 100644 --- a/backend/tests/test_debug_auto_test.py +++ b/backend/tests/test_debug_auto_test.py @@ -7,7 +7,11 @@ from pipecat.processors.aggregators.llm_context import LLMContext from services.brains import BrainRuntime, build_brain from services.brains.base import GREETING_CONTEXT_MARKER from services.runtime_variables import DynamicVariableStore, prepare_dynamic_config -from services.test_runs.mock_tools import MockToolExecutor, TextClientToolPort +from services.test_runs.mock_tools import ( + MockToolExecutor, + TextClientToolPort, + ToolSimulationResolver, +) from services.test_runs.text_runner import ( _TextCaptureProcessor, _opening_speech_from_message, @@ -16,7 +20,7 @@ from services.test_runs.text_runner import ( _wait_for_startup, _wait_for_quiescence, ) -from test_schemas import BatchRunCreate, BatchStartupResult +from test_schemas import BatchRunCreate, BatchStartupResult, FixedToolSimulation class FakeLLM: @@ -154,8 +158,24 @@ class DebugAutoTestContractTests(unittest.TestCase): "message_started", ) - def test_text_client_port_records_automatic_message_confirmation(self): - port = TextClientToolPort() + def test_text_client_port_records_configured_message_confirmation(self): + resolver = ToolSimulationResolver( + [ + FixedToolSimulation.model_validate( + { + "toolId": "tool_message", + "functionName": "show_message", + "mode": "fixed", + "response": { + "outcome": "success", + "body": '{"action":"confirmed"}', + "delayMs": 0, + }, + } + ) + ] + ) + port = TextClientToolPort(resolver) result = asyncio.run( port.call( @@ -230,12 +250,9 @@ class DebugAutoTestRuntimeTests(unittest.IsolatedAsyncioTestCase): brain = build_brain(cfg) llm = FakeLLM() call_end = FakeCallEnd() - executor_holder = {} def create_executor(store): - executor = MockToolExecutor(store) - executor_holder["executor"] = executor - return executor + return MockToolExecutor(store, ToolSimulationResolver([])) await brain.setup( cfg, @@ -256,7 +273,6 @@ class DebugAutoTestRuntimeTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(call_end.ending) self.assertEqual(call_end.reason, "测试完成") self.assertEqual(params.result["action"], "ending_call") - self.assertEqual(executor_holder["executor"].unmocked_calls, []) async def test_text_capture_keeps_runtime_records_for_auto_test_ui(self): event = { diff --git a/backend/tests/test_startup_actions.py b/backend/tests/test_startup_actions.py index 3ccfcd4..95fb80f 100644 --- a/backend/tests/test_startup_actions.py +++ b/backend/tests/test_startup_actions.py @@ -7,7 +7,11 @@ from fastapi import HTTPException from routes.assistants import _validate_startup_actions from models import RuntimeToolBinding from schemas import AssistantUpsert -from services.config_resolver import _runtime_tool_ids +from services.config_resolver import ( + _implicit_startup_tool_names, + _runtime_tool_ids, + _tools_for, +) def startup_body( @@ -57,6 +61,25 @@ class FakeSession: return self.tool if self.tool is not None and tool_id == self.tool.id else None +class FakeToolQueryResult: + def __init__(self, tools): + self.tools = tools + + def scalars(self): + return self + + def all(self): + return self.tools + + +class FakeToolCatalogSession: + def __init__(self, tools): + self.tools = tools + + async def execute(self, _statement): + return FakeToolQueryResult(self.tools) + + class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): def test_runtime_pool_includes_bound_and_startup_only_tools(self): self.assertEqual( @@ -75,6 +98,51 @@ class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): ["conversation_tool", "shared_tool", "opening_tool"], ) + def test_confirmation_opening_declares_implicit_show_message_tool(self): + self.assertEqual( + _implicit_startup_tool_names( + { + "opening_mode": "confirmation", + "opening_message": { + "title": "重要提示", + "message": "请确认已阅读。", + }, + } + ), + ["show_message"], + ) + self.assertEqual( + _implicit_startup_tool_names( + { + "openingMode": "playback", + "openingMessage": {"message": "无需确认"}, + } + ), + [], + ) + + async def test_implicit_show_message_enters_runtime_pool_without_binding(self): + tool = SimpleNamespace( + id="tool_show_message_default", + name="显示确认消息", + function_name="show_message", + type="client", + description="显示确认弹窗", + definition={"config": {}}, + secrets={}, + mcp_server_id=None, + ) + + runtime_tools = await _tools_for( + FakeToolCatalogSession([tool]), + SimpleNamespace(type="prompt"), + [], + ["show_message"], + ) + + self.assertEqual([item.id for item in runtime_tools], [tool.id]) + self.assertEqual(runtime_tools[0].function_name, "show_message") + def test_realtime_rejects_startup_actions(self): with self.assertRaisesRegex(ValueError, "Realtime"): AssistantUpsert( diff --git a/backend/tests/test_tool_simulations.py b/backend/tests/test_tool_simulations.py new file mode 100644 index 0000000..2a2f6c7 --- /dev/null +++ b/backend/tests/test_tool_simulations.py @@ -0,0 +1,312 @@ +import unittest + +from pydantic import ValidationError + +from services.message_policy import MESSAGE_CONFIRMATION +from services.message_stage import ( + MessageDisplaySpec, + MessageStageRunner, + MessageStageSpec, +) +from services.test_runs.mock_tools import ( + TextClientToolPort, + ToolSimulationError, + ToolSimulationResolver, +) +from test_schemas import TestCaseDefinition + + +def definition_with_simulation(simulation: dict) -> TestCaseDefinition: + return TestCaseDefinition.model_validate( + { + "turns": [ + { + "id": "turn_1", + "userInput": "查询订单", + "behaviors": [ + { + "id": "behavior_1", + "type": "tool_call", + "toolId": "tool_order", + "functionName": "query_order", + "expectation": "called", + "minCalls": 1, + "maxCalls": 1, + "paramAssertions": [], + } + ], + } + ], + "toolSimulations": [simulation], + } + ) + + +class ToolSimulationSchemaTests(unittest.TestCase): + def test_called_assertion_requires_simulation(self): + with self.assertRaisesRegex(ValidationError, "缺少工具模拟"): + TestCaseDefinition.model_validate( + { + "turns": [ + { + "id": "turn_1", + "userInput": "查询订单", + "behaviors": [ + { + "id": "behavior_1", + "type": "tool_call", + "toolId": "tool_order", + "functionName": "query_order", + "expectation": "called", + "minCalls": 1, + "paramAssertions": [], + } + ], + } + ] + } + ) + + def test_sequence_turn_scope_must_reference_existing_turn(self): + with self.assertRaisesRegex(ValidationError, "不存在的轮次"): + definition_with_simulation( + { + "toolId": "tool_order", + "functionName": "query_order", + "mode": "sequence", + "steps": [ + { + "id": "step_1", + "scope": {"phase": "turn", "turnId": "missing"}, + "outcome": "success", + "body": "{}", + "delayMs": 0, + } + ], + } + ) + + def test_legacy_equal_responses_become_fixed_simulation(self): + definition = TestCaseDefinition.model_validate( + { + "turns": [ + { + "id": "turn_1", + "userInput": "第一次查询", + "behaviors": [ + { + "id": "behavior_1", + "type": "tool_call", + "toolId": "tool_order", + "functionName": "query_order", + "expectation": "called", + "minCalls": 1, + "paramAssertions": [], + "mockResponse": { + "outcome": "success", + "body": '{"orderId":"A1"}', + "delayMs": 0, + }, + } + ], + }, + { + "id": "turn_2", + "userInput": "再次查询", + "behaviors": [ + { + "id": "behavior_2", + "type": "tool_call", + "toolId": "tool_order", + "functionName": "query_order", + "expectation": "called", + "minCalls": 1, + "paramAssertions": [], + "mockResponse": { + "outcome": "success", + "body": '{"orderId":"A1"}', + "delayMs": 0, + }, + } + ], + }, + ] + } + ) + + self.assertEqual(definition.tool_simulations[0].mode, "fixed") + dumped = definition.model_dump(mode="json", by_alias=True) + self.assertNotIn("mockResponse", dumped["turns"][0]["behaviors"][0]) + + def test_legacy_different_responses_become_scoped_sequence(self): + definition = TestCaseDefinition.model_validate( + { + "turns": [ + { + "id": "turn_1", + "userInput": "第一次查询", + "behaviors": [ + { + "id": "behavior_1", + "type": "tool_call", + "toolId": "tool_order", + "functionName": "query_order", + "expectation": "called", + "minCalls": 2, + "paramAssertions": [], + "mockResponse": { + "outcome": "success", + "body": '{"orderId":"A1"}', + "delayMs": 0, + }, + } + ], + }, + { + "id": "turn_2", + "userInput": "再次查询", + "behaviors": [ + { + "id": "behavior_2", + "type": "tool_call", + "toolId": "tool_order", + "functionName": "query_order", + "expectation": "called", + "minCalls": 1, + "paramAssertions": [], + "mockResponse": { + "outcome": "error", + "body": '{"message":"not found"}', + "delayMs": 0, + }, + } + ], + }, + ] + } + ) + + simulation = definition.tool_simulations[0] + self.assertEqual(simulation.mode, "sequence") + self.assertEqual(len(simulation.steps), 3) + self.assertEqual(simulation.steps[0].scope.turn_id, "turn_1") + self.assertEqual(simulation.steps[2].scope.turn_id, "turn_2") + + +class ToolSimulationResolverTests(unittest.IsolatedAsyncioTestCase): + async def test_client_message_stage_keeps_unmocked_failure_for_runner(self): + resolver = ToolSimulationResolver([]) + stage = MessageStageRunner(TextClientToolPort(resolver)) + + result = await stage.run( + MessageStageSpec( + display=MessageDisplaySpec(title="确认", message="请确认"), + completion_policy=MESSAGE_CONFIRMATION, + ) + ) + + self.assertFalse(result.succeeded) + self.assertIn("UNMOCKED_TOOL_CALL", result.error or "") + with self.assertRaisesRegex(ToolSimulationError, "UNMOCKED_TOOL_CALL"): + resolver.raise_if_failed() + + async def test_fixed_response_can_be_reused(self): + definition = definition_with_simulation( + { + "toolId": "tool_order", + "functionName": "query_order", + "mode": "fixed", + "response": { + "outcome": "success", + "body": '{"orderId":"A1"}', + "delayMs": 0, + }, + } + ) + resolver = ToolSimulationResolver(definition.tool_simulations) + resolver.enter_turn("turn_1") + + first = await resolver.resolve( + tool_id="tool_order", function_name="query_order" + ) + second = await resolver.resolve( + tool_id="tool_order", function_name="query_order" + ) + + self.assertEqual(first, {"orderId": "A1", "status": "ok"}) + self.assertEqual(second, first) + + async def test_sequence_consumes_client_choices_then_exhausts(self): + definition = definition_with_simulation( + { + "toolId": "tool_order", + "functionName": "query_order", + "mode": "sequence", + "steps": [ + { + "id": "step_1", + "scope": {"phase": "turn", "turnId": "turn_1"}, + "outcome": "success", + "body": '{"action":"selected","value":"basic"}', + "delayMs": 0, + }, + { + "id": "step_2", + "scope": {"phase": "any"}, + "outcome": "success", + "body": '{"action":"dismissed"}', + "delayMs": 0, + }, + ], + } + ) + resolver = ToolSimulationResolver(definition.tool_simulations) + resolver.enter_turn("turn_1") + port = TextClientToolPort(resolver) + + selected = await port.call("query_order", {}) + dismissed = await port.call("query_order", {}) + + self.assertEqual(selected["data"]["value"], "basic") + self.assertEqual(dismissed["data"]["action"], "dismissed") + with self.assertRaisesRegex( + ToolSimulationError, "TOOL_SIMULATION_EXHAUSTED" + ): + await port.call("query_order", {}) + + async def test_missing_and_scope_mismatch_are_distinct(self): + definition = definition_with_simulation( + { + "toolId": "tool_order", + "functionName": "query_order", + "mode": "sequence", + "steps": [ + { + "id": "step_1", + "scope": {"phase": "turn", "turnId": "turn_1"}, + "outcome": "success", + "body": "{}", + "delayMs": 0, + } + ], + } + ) + resolver = ToolSimulationResolver(definition.tool_simulations) + resolver.enter_startup() + + with self.assertRaisesRegex( + ToolSimulationError, "TOOL_SIMULATION_SCOPE_MISMATCH" + ): + await resolver.resolve( + tool_id="tool_order", function_name="query_order" + ) + with self.assertRaisesRegex( + ToolSimulationError, "TOOL_SIMULATION_SCOPE_MISMATCH" + ): + resolver.raise_if_failed() + with self.assertRaisesRegex(ToolSimulationError, "UNMOCKED_TOOL_CALL"): + await resolver.resolve(tool_id="missing", function_name="missing") + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/src/components/assistant-editor/debug-auto-test.tsx b/frontend/src/components/assistant-editor/debug-auto-test.tsx index c769528..6ddeb69 100644 --- a/frontend/src/components/assistant-editor/debug-auto-test.tsx +++ b/frontend/src/components/assistant-editor/debug-auto-test.tsx @@ -36,11 +36,12 @@ import { } from "@/components/ui/popover"; import { SearchInput } from "@/components/ui/search-input"; import { Tabs, TabsList, TabsTrigger } from "@/components/ui/tabs"; -import type { - BatchEvaluationResult, - BatchRunCase, - BatchRunSnapshot, - BatchStartupResult, +import { + testExecutionErrorLabel, + type BatchEvaluationResult, + type BatchRunCase, + type BatchRunSnapshot, + type BatchStartupResult, } from "@/data/batch-run"; import { getTestCaseValidationMessage, @@ -422,7 +423,7 @@ export function AutoTestLivePanel({ ) { return ( ); @@ -504,7 +505,7 @@ export function AutoTestLivePanel({ {state.phase === "error" && state.runCase?.executionError && (
- {state.runCase.executionError.code} + {testExecutionErrorLabel(state.runCase.executionError.code)} · {state.runCase.executionError.message}
diff --git a/frontend/src/components/batch-test/batch-case-detail-drawer.tsx b/frontend/src/components/batch-test/batch-case-detail-drawer.tsx index ddd3ed5..3db8304 100644 --- a/frontend/src/components/batch-test/batch-case-detail-drawer.tsx +++ b/frontend/src/components/batch-test/batch-case-detail-drawer.tsx @@ -23,6 +23,7 @@ import { BatchCaseStatusBadge } from "@/components/batch-test/batch-case-status" import { Button } from "@/components/ui/button"; import { firstEvaluationFailureReason, + testExecutionErrorLabel, type BatchEvaluationResult, type BatchRunCase, type BatchToolCallRecord, @@ -115,7 +116,7 @@ export function BatchCaseDetailDrawer({
- {item.executionError.code} + {testExecutionErrorLabel(item.executionError.code)} {item.executionError.stage} diff --git a/frontend/src/components/pages/TestCasesPage.tsx b/frontend/src/components/pages/TestCasesPage.tsx index 5fe9b36..cf57b67 100644 --- a/frontend/src/components/pages/TestCasesPage.tsx +++ b/frontend/src/components/pages/TestCasesPage.tsx @@ -56,6 +56,7 @@ import { DEFAULT_INPUT_MODE, getTestCaseValidationMessage, normalizeOverallCriteria, + prepareFixedToolSimulations, TEST_CASE_INPUT_MODE_LABEL, TEST_CASE_INPUT_MODE_SHORT_LABEL, type TestCase, @@ -420,6 +421,7 @@ function caseToDraft(item: TestCase): CaseEditorDraft { inputMode: item.inputMode, contextTurns: item.contextTurns.map((turn) => ({ ...turn })), turns: cloneTurns(item.turns), + toolSimulations: prepareFixedToolSimulations(item.toolSimulations), overallCriteria: cloneOverallCriteria(item.overallCriteria), }; } @@ -434,6 +436,7 @@ function createEmptyCaseDraft(): CaseEditorDraft { inputMode: DEFAULT_INPUT_MODE, contextTurns: [], turns: [createEmptyFixedInputTurn()], + toolSimulations: [], overallCriteria: [], }; } @@ -625,6 +628,7 @@ function SuiteDetailView({ inputMode: draft.inputMode, contextTurns: draft.contextTurns, turns: draft.turns, + toolSimulations: draft.toolSimulations, overallCriteria: normalizeOverallCriteria(draft.overallCriteria), }; diff --git a/frontend/src/components/test-cases/next-reply-editor.tsx b/frontend/src/components/test-cases/next-reply-editor.tsx index 2585f60..9188f10 100644 --- a/frontend/src/components/test-cases/next-reply-editor.tsx +++ b/frontend/src/components/test-cases/next-reply-editor.tsx @@ -55,6 +55,10 @@ import { SelectValue, } from "@/components/ui/select"; import { Textarea } from "@/components/ui/textarea"; +import { + ToolSimulationEditor, + type EditorToolOption, +} from "@/components/test-cases/tool-simulation-editor"; import { ASSERTION_TYPE_LABEL, KEYWORD_MATCH_MODE_LABEL, @@ -76,6 +80,7 @@ import { type TestImageRef, type TestCaseInputMode, type ToolCallExpectedBehavior, + type ToolSimulation, type ToolParamAssertion, type ToolParamMatchMode, } from "@/data/test-suites"; @@ -87,6 +92,7 @@ export type CaseEditorDraft = { inputMode: TestCaseInputMode; contextTurns: ContextTurn[]; turns: FixedInputTurn[]; + toolSimulations: ToolSimulation[]; overallCriteria: OverallCriterion[]; }; @@ -104,15 +110,7 @@ function contextPlaceholder(role: ContextRole): string { return '{\n "status": "accepted"\n}'; } -/** 编辑器内使用的工具选项(来自 Agent 工具库) */ -export type EditorToolOption = { - id: string; - functionName: string; - label: string; - parameters: { name: string; description: string }[]; -}; - -type EditorSectionId = "context" | "main" | "overall"; +type EditorSectionId = "context" | "main" | "tools" | "overall"; function editorSections(): { id: EditorSectionId; @@ -121,6 +119,7 @@ function editorSections(): { return [ { id: "context", label: "对话上下文" }, { id: "main", label: "固定输入" }, + { id: "tools", label: "工具模拟" }, { id: "overall", label: "整体评估" }, ]; } @@ -156,6 +155,7 @@ function toEditorToolOption(tool: Tool): EditorToolOption { id: tool.id, functionName: tool.functionName || tool.name, label: tool.name || tool.functionName, + type: tool.type, parameters: toolParametersFromDefinition(tool), }; } @@ -175,6 +175,7 @@ export function NextReplyEditorBody({ const sectionRefs = useRef>({ context: null, main: null, + tools: null, overall: null, }); const selectedAnchorRef = useRef(null); @@ -230,7 +231,12 @@ export function NextReplyEditorBody({ const containerTop = scrollContainer.getBoundingClientRect().top; const activationLine = containerTop + 24; - const sectionIds: EditorSectionId[] = ["context", "main", "overall"]; + const sectionIds: EditorSectionId[] = [ + "context", + "main", + "tools", + "overall", + ]; let nextSection: EditorSectionId = sectionIds[0]; for (const sectionId of sectionIds) { @@ -394,7 +400,10 @@ export function NextReplyEditorBody({ function removeTurn(turnId: string) { if (draft.turns.length <= 1) return; - updateTurns(draft.turns.filter((turn) => turn.id !== turnId)); + onChange({ + ...draft, + turns: draft.turns.filter((turn) => turn.id !== turnId), + }); } function handleTurnDragEnd(event: DragEndEvent) { @@ -681,6 +690,28 @@ export function NextReplyEditorBody({ +
{ + sectionRefs.current.tools = element; + }} + className="scroll-mt-3 space-y-3" + > + } + title="工具模拟" + description="为外部工具配置固定返回;每次调用复用同一结果,缺少模拟时测试立即停止。" + > + + onChange({ ...draft, toolSimulations }) + } + /> + +
+
{ sectionRefs.current.overall = element; @@ -1094,7 +1125,7 @@ function SortableFixedInputTurn({ > @@ -1140,7 +1171,7 @@ function ExpectedBehaviorCard({ className={cn( "group/card w-full rounded-xl border px-3 py-2.5 text-left transition-colors hover:bg-canvas-soft/60", behavior.type === "tool_call" - ? "border-violet-500/25 bg-violet-500/[0.04]" + ? "border-hairline bg-surface-strong/25" : "border-hairline bg-canvas-soft/30", validationMessage && "border-destructive/40", )} @@ -1168,7 +1199,7 @@ function ExpectedBehaviorCard({ className={cn( "rounded-xl border px-3 py-2.5", behavior.type === "tool_call" - ? "border-violet-500/30 bg-violet-500/[0.04]" + ? "border-hairline bg-surface-strong/25" : "border-hairline bg-canvas-soft/40", validationMessage && "border-destructive/40", )} @@ -1184,7 +1215,7 @@ function ExpectedBehaviorCard({ {behavior.type === "reply" ? ( ) : ( - + )} {behavior.type === "reply" ? "回复要求" : "工具调用"} @@ -1257,7 +1288,7 @@ function BehaviorSummary({ behavior }: { behavior: ExpectedBehavior }) { return (
- + {behavior.expectation === "not_called" ? "禁止调用 · " : ""} {name}
@@ -1712,87 +1743,9 @@ function ToolCallBehaviorEditor({
)} - {selectedTool && behavior.expectation === "called" && ( -
-
-
- Mock 工具响应 -
-

- 工具被调用后将此结果注入 pipeline,使下一轮可稳定复现。 -

-
-
- -
- { - const next = Number(event.target.value); - onChange({ - ...behavior, - mockResponse: { - ...behavior.mockResponse, - delayMs: Number.isFinite(next) - ? Math.min(60000, Math.max(0, Math.round(next))) - : 0, - }, - }); - }} - aria-label="Mock 工具响应延迟" - className="h-8 border-hairline-strong bg-background text-xs" - /> - ms -
-
-