feat: add dynamic variables and startup result handling in batch run schemas

- Introduced dynamic_variables field in BatchRunCreate schema to support flexible input types. - Added BatchStartupResult schema to encapsulate startup state, including greeting and auto-confirmed messages. - Updated BatchRunCaseOut schema to include startup results, enhancing the structure for test run outputs. - Enhanced test run orchestration to prepare and validate dynamic configurations, ensuring isolated variable management across test cases. - Implemented changes in the frontend to accommodate new data structures for improved test case execution and result tracking.
2026-08-12 16:55:25 +08:00
parent 79a239489a
commit 7560f9a525
12 changed files with 1464 additions and 1004 deletions
@@ -16,6 +16,7 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.frames.frames import (
    BotStoppedSpeakingFrame,
    CancelFrame,
+    DataFrame,
    EndFrame,
    ErrorFrame,
    FunctionCallInProgressFrame,
@@ -103,10 +104,31 @@ class RawTurnResult:
    tool_calls: list[RawToolCall]


+@dataclass
+class TextStartupResult:
+    """Headless startup choices applied before the scripted user turns."""
+
+    greeting: str = ""
+    assistant_outputs: list[str] = field(default_factory=list)
+    auto_confirmed_messages: list[dict[str, Any]] = field(default_factory=list)
+
+
@dataclass
 class TextCaseResult:
    turns: list[RawTurnResult]
    transcript: list[dict[str, str]]
+    startup: TextStartupResult = field(default_factory=TextStartupResult)
+
+
+TurnCompleteCallback = Callable[[RawTurnResult], Awaitable[None]]
+StartupCompleteCallback = Callable[[TextStartupResult], Awaitable[None]]
+
+
+@dataclass
+class _StartupBoundaryFrame(DataFrame):
+    """Ordered marker queued behind every frame emitted by startup."""
+
+    reached: asyncio.Event


@dataclass
@@ -139,6 +161,26 @@ class _TextCaptureProcessor(FrameProcessor):
        self.error: TestExecutionError | None = None
        self.on_fixed_speech_complete: Callable[[], Awaitable[None]] | None = None

+    def fail(self, frame: ErrorFrame) -> None:
+        """Release every wait even when Pipecat reports the error upstream."""
+
+        message = frame.error or "Pipeline 执行失败"
+        unmocked_tool = "UNMOCKED_TOOL_CALL" in message
+        self.error = TestExecutionError(
+            code="UNMOCKED_TOOL_CALL" if unmocked_tool else "PIPELINE_ERROR",
+            message=message,
+            stage=(
+                "tool"
+                if unmocked_tool
+                else "model"
+                if frame.processor
+                else "pipeline"
+            ),
+            retryable=not unmocked_tool,
+        )
+        self.window.pending_tools.clear()
+        self.window.touch()
+
    async def process_frame(self, frame, direction: FrameDirection):
        await super().process_frame(frame, direction)

@@ -196,12 +238,9 @@ class _TextCaptureProcessor(FrameProcessor):
            self.window.pending_tools.discard(call.id)
            self.window.touch()
        elif isinstance(frame, ErrorFrame):
-            self.error = TestExecutionError(
-                code="PIPELINE_ERROR",
-                message=frame.error or "Pipeline 执行失败",
-                stage="model" if frame.processor else "pipeline",
-                retryable=True,
-            )
+            self.fail(frame)
+        elif isinstance(frame, _StartupBoundaryFrame):
+            frame.reached.set()
            self.window.touch()
        elif isinstance(frame, (EndFrame, CancelFrame)):
            self.window.touch()
@@ -281,6 +320,41 @@ async def _wait_for_quiescence(
        await asyncio.sleep(0.05)


+async def _wait_for_startup(
+    worker: PipelineWorker,
+    capture: _TextCaptureProcessor,
+    runner_task: asyncio.Task[None],
+    activity_marker: int,
+) -> None:
+    """Wait until startup-queued frames and any entry LLM reply are complete."""
+
+    boundary = asyncio.Event()
+    await worker.queue_frame(_StartupBoundaryFrame(reached=boundary))
+    boundary_waiter = asyncio.create_task(boundary.wait())
+    done, _pending = await asyncio.wait(
+        {boundary_waiter, runner_task},
+        timeout=WORKER_START_TIMEOUT_SECONDS,
+        return_when=asyncio.FIRST_COMPLETED,
+    )
+    if not done:
+        boundary_waiter.cancel()
+        raise TestExecutionError(
+            code="STARTUP_DRAIN_TIMEOUT",
+            message="开场阶段未能进入稳定状态",
+            stage="pipeline",
+            retryable=True,
+        )
+    if boundary_waiter not in done:
+        boundary_waiter.cancel()
+    await asyncio.gather(boundary_waiter, return_exceptions=True)
+    await _wait_for_quiescence(
+        capture,
+        runner_task,
+        activity_marker,
+        require_activity=False,
+    )
+
+
 class TextPipelineRunner:
    """Execute all fixed input turns in one isolated, transportless pipeline."""

@@ -290,10 +364,17 @@ class TextPipelineRunner:
        definition: TestCaseDefinition,
        *,
        timeout_seconds: int,
+        on_turn_complete: TurnCompleteCallback | None = None,
+        on_startup_complete: StartupCompleteCallback | None = None,
    ) -> TextCaseResult:
        try:
            async with asyncio.timeout(timeout_seconds):
-                return await self._run(cfg, definition)
+                return await self._run(
+                    cfg,
+                    definition,
+                    on_turn_complete=on_turn_complete,
+                    on_startup_complete=on_startup_complete,
+                )
        except TimeoutError as exc:
            raise TestExecutionError(
                code="PIPELINE_TIMEOUT",
@@ -306,6 +387,9 @@ class TextPipelineRunner:
        self,
        cfg: AssistantConfig,
        definition: TestCaseDefinition,
+        *,
+        on_turn_complete: TurnCompleteCallback | None,
+        on_startup_complete: StartupCompleteCallback | None,
    ) -> TextCaseResult:
        if cfg.runtimeMode != "pipeline":
            raise TestExecutionError(
@@ -396,7 +480,12 @@ class TextPipelineRunner:
        async def on_pipeline_started(_worker, _frame):
            worker_started.set()

+        @worker.event_handler("on_pipeline_error")
+        async def on_pipeline_error(_worker, frame):
+            capture.fail(frame)
+
        executor_holder: dict[str, MockToolExecutor] = {}
+        text_client_tools = TextClientToolPort()

        def create_test_executor(store) -> MockToolExecutor:
            executor = MockToolExecutor(store)
@@ -520,7 +609,7 @@ class TextPipelineRunner:
                set_tools=set_tools,
                call_end=call_end,
                session_id=f"test_{uuid4().hex}",
-                client_tools=TextClientToolPort(),
+                client_tools=text_client_tools,
                worker=worker,
                context_aggregator=aggregators,
                transport=None,
@@ -536,6 +625,13 @@ class TextPipelineRunner:
        )
        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.
+        greeting = str(await brain.greeting(cfg) or "").strip()
+        if greeting and brain.spec.owns_context:
+            brain.prepare_greeting_context(greeting, context)
+
        @assistant_aggregator.event_handler("on_assistant_turn_started")
        async def on_assistant_turn_started(_aggregator):
            capture.window.active_assistant += 1
@@ -591,12 +687,21 @@ class TextPipelineRunner:
            marker = capture.window.activity_count
            await brain.on_connected(greeting_pending=False)
            await brain.on_client_ready()
-            await _wait_for_quiescence(
+            await _wait_for_startup(
+                worker,
                capture,
                runner_task,
                marker,
-                require_activity=False,
            )
+            startup = TextStartupResult(
+                greeting=greeting,
+                assistant_outputs=list(capture.outputs),
+                auto_confirmed_messages=list(
+                    text_client_tools.auto_confirmed_messages
+                ),
+            )
+            if on_startup_complete is not None:
+                await on_startup_complete(startup)
            # Startup messages initialize the session but are not a scripted
            # user turn result in the fixed-input editor contract.
            capture.outputs.clear()
@@ -620,6 +725,8 @@ class TextPipelineRunner:
                    index,
                )
                results.append(result)
+                if on_turn_complete is not None:
+                    await on_turn_complete(result)

            transcript = [
                {
@@ -630,7 +737,11 @@ class TextPipelineRunner:
                if message.get("role") in {"user", "assistant"}
                and str(message.get("content") or "").strip()
            ]
-            return TextCaseResult(turns=results, transcript=transcript)
+            return TextCaseResult(
+                turns=results,
+                transcript=transcript,
+                startup=startup,
+            )
        except TestExecutionError:
            raise
        except asyncio.CancelledError: