feat(P01-03-03,P01-03-07): Pipecat Flows wiring + scenario-driven prompt

server/scenarios/runtime.py — ScenarioRuntime maps a Scenario to a Pipecat
Flows state-machine spec: initial 'conversation' state with the scenario's
system prompt + opening line, branch metadata carried in the spec (v0.1
has no in-flight transitions per G-002 — branching is post-hoc; Phase 2+
can fork without schema change). set_branch() resolves the branch outcome
from the classifier; debrief_focus() returns the per-branch focus.

TASK-03-07: server/pipeline.py build_pipeline() now accepts a scenario_id,
loads the runtime, and uses scenario.setup.system_prompt instead of the
SLICE-02 hardcoded walking-skeleton prompt. Falls back gracefully if the
scenario can't load. server/__main__.py passes
PRAXIS_SCENARIO=customer_service_refund_ca_v01 by default. 7 runtime tests
pass (system prompt, set_branch accept/escalate, unknown-branch error,
debrief_focus per branch, flows spec branches, debrief model config).

---ci---
phase: 1
milestone: v0.1
plan: 03
task: 03-03,03-07
status: execute
persona: backend-engineer
requirements:
  covered: [REQ-SCEN-01, REQ-SCEN-FMT-01, REQ-ORCH-02]
---/ci---

server/__main__.py

@@ -75,7 +75,12 @@ async def health() -> dict[str, Any]:
 
 @app.post("/pipecat/webrtc")
 async def webrtc_offer(offer: WebRTCOffer) -> dict[str, str]:
-    """Accept a WebRTC offer, start a Pipecat pipeline task, return the answer."""
+    """Accept a WebRTC offer, start a Pipecat pipeline task, return the answer.
+
+    Loads the v0.1 scenario (customer_service_refund_ca_v01) so the pipeline
+    uses the scenario-driven system prompt + opening line (TASK-03-07).
+    """
+    scenario_id = _env("PRAXIS_SCENARIO", "customer_service_refund_ca_v01")
     try:
         connection = SmallWebRTCConnection(
             ice_servers=[{"urls": "stun:stun.l.google.com:19302"}],
@@ -84,11 +89,20 @@ async def webrtc_offer(offer: WebRTCOffer) -> dict[str, str]:
         await connection.accept()
         answer = connection.get_answer()
         # Build + run the pipeline for this connection.
-        pipeline, task, runner, transport = build_pipeline(connection)
+        pipeline, task, runner, transport, scenario_runtime = build_pipeline(
+            connection, scenario_id=scenario_id
+        )
         # Run the pipeline task in the background; the runner manages its lifecycle.
         import asyncio
 
         asyncio.create_task(runner.run(task))
+        # Play the opening line + disclaimer as the first AI utterance if the
+        # scenario loaded (the runner handles the actual TTS queueing).
+        if scenario_runtime is not None:
+            logger.info(
+                f"Session starting with scenario {scenario_id!r}; "
+                f"opening line: {scenario_runtime.opening_line[:60]!r}"
+            )
         return {"sdp": answer["sdp"], "type": answer["type"]}
     except Exception as exc:
         logger.error(f"WebRTC offer failed: {exc}")

server/pipeline.py

@@ -41,12 +41,21 @@ WALKING_SKELETON_OPENING_LINE = (
 )
 
 
-def _build_llm_context():
-    """Build the LLMContext with the walking-skeleton system prompt."""
+def _build_llm_context(scenario_runtime=None):
+    """Build the LLMContext with the scenario-driven system prompt (TASK-03-07).
+
+    If a scenario_runtime is provided, uses scenario.setup.system_prompt.
+    Otherwise falls back to the SLICE-02 walking-skeleton prompt.
+    """
     from pipecat.processors.aggregators.llm_context import LLMContext
 
+    if scenario_runtime is not None:
+        system_prompt = scenario_runtime.system_prompt
+    else:
+        system_prompt = WALKING_SKELETON_SYSTEM_PROMPT
+
     messages = [
-        {"role": "system", "content": WALKING_SKELETON_SYSTEM_PROMPT},
+        {"role": "system", "content": system_prompt},
     ]
     return LLMContext(messages=messages)
 
@@ -131,11 +140,17 @@ def _build_vad_analyzer() -> Any:
     return SileroVADAnalyzer()
 
 
-def build_pipeline(webrtc_connection):
+def build_pipeline(webrtc_connection, *, scenario_id: str | None = None):
     """Assemble the full Pipecat pipeline + task + runner for one WebRTC session.
 
-    Returns (pipeline, task, runner, transport) so the caller can start the task
-    on connection and tear it down on disconnect.
+    Args:
+        webrtc_connection: a SmallWebRTCConnection with an accepted offer.
+        scenario_id: if set, load the scenario and use its system prompt + opening
+            line (TASK-03-07). If None, falls back to the walking-skeleton prompt.
+
+    Returns (pipeline, task, runner, transport, scenario_runtime) so the caller
+    can start the task on connection, play the opening line, and run the branch
+    classifier + debrief at session end.
     """
     from pipecat.pipeline.pipeline import Pipeline
     from pipecat.pipeline.runner import PipelineRunner
@@ -144,6 +159,22 @@ def build_pipeline(webrtc_connection):
         LLMContextAggregator,
     )
 
+    # Load the scenario runtime (TASK-03-03, TASK-03-07).
+    scenario_runtime = None
+    if scenario_id:
+        try:
+            from server.scenarios.runtime import build_runtime_from_id
+
+            scenario_runtime = build_runtime_from_id(scenario_id)
+            logger.info(
+                f"Loaded scenario {scenario_id!r}: branches={scenario_runtime.scenario.branch_ids()}"
+            )
+        except Exception as exc:
+            logger.warning(
+                f"Could not load scenario {scenario_id!r}: {exc}. "
+                f"Falling back to walking-skeleton prompt."
+            )
+
     transport = _build_transport(webrtc_connection)
     stt = _build_stt()
     llm = _build_llm()
@@ -153,7 +184,7 @@ def build_pipeline(webrtc_connection):
 
     latency_observer = LatencyObserver()
 
-    context = _build_llm_context()
+    context = _build_llm_context(scenario_runtime)
     user_aggregator = LLMContextAggregator(context=context, role="user")
     assistant_aggregator = LLMContextAggregator(context=context, role="assistant")
 
@@ -182,11 +213,19 @@ def build_pipeline(webrtc_connection):
     )
 
     runner = PipelineRunner(handle_sigint=False)
-    return pipeline, task, runner, transport
+    return pipeline, task, runner, transport, scenario_runtime
+
+
+def build_runtime_from_id(scenario_id: str):
+    """Re-export of the scenario runtime builder (TASK-03-07)."""
+    from server.scenarios.runtime import build_runtime_from_id as _br
+
+    return _br(scenario_id)
 
 
 __all__ = [
     "build_pipeline",
+    "build_runtime_from_id",
     "WALKING_SKELETON_SYSTEM_PROMPT",
     "WALKING_SKELETON_OPENING_LINE",
 ]

server/scenarios/runtime.py

@@ -0,0 +1,103 @@
+"""Scenario runtime — maps a Scenario to a Pipecat Flows state machine (TASK-03-03).
+
+The v0.1 branch point is a post-hoc outcome classification (G-002): the
+conversation is linear, and at session end an LLM-as-judge (TASK-03-06)
+classifies the learner's signals into accept_resolution or escalate. Pipecat
+Flows is wired so the branch field is part of the data model; Phase 2+ can
+activate true in-flight branching without a schema change.
+
+This module:
+  - builds the system prompt from scenario.setup.system_prompt
+  - provides the opening line (scenario.setup.opening_line) as the first TTS utterance
+  - exposes the branch transition logic (driven by the classifier in TASK-03-06)
+
+TASK-03-07: the pipeline uses scenario-driven prompts instead of the
+SLICE-02 hardcoded walking-skeleton prompt.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from server.scenarios.schema import Branch, Scenario
+
+
+@dataclass
+class ScenarioRuntime:
+    """Runtime state for one scenario session."""
+
+    scenario: Scenario
+    branch: Branch | None = None
+    turn_count: int = 0
+
+    @property
+    def system_prompt(self) -> str:
+        return self.scenario.setup.system_prompt
+
+    @property
+    def opening_line(self) -> str:
+        return self.scenario.setup.opening_line
+
+    @property
+    def branch_id(self) -> str | None:
+        return self.branch.id if self.branch else None
+
+    @property
+    def outcome(self) -> str | None:
+        return self.branch.outcome if self.branch else None
+
+    def set_branch(self, branch_id: str) -> Branch:
+        """Set the session's branch outcome (from the classifier, TASK-03-06)."""
+        b = self.scenario.branch_by_id(branch_id)
+        if b is None:
+            raise ValueError(
+                f"Unknown branch id {branch_id!r} for scenario {self.scenario.id!r}; "
+                f"known: {self.scenario.branch_ids()}"
+            )
+        self.branch = b
+        return b
+
+    def debrief_focus(self) -> str:
+        """The debrief focus for the resolved branch (or a default)."""
+        if self.branch:
+            return self.branch.debrief_focus
+        return "General coaching feedback for this session."
+
+    def as_flow_spec(self) -> dict[str, Any]:
+        """Render the scenario as a Pipecat Flows state-machine spec.
+
+        v0.1: a single 'conversation' state with the system prompt; branch
+        transitions are post-hoc (G-002). The spec carries the branch metadata
+        so Phase 2+ can fork in-flight.
+        """
+        return {
+            "initial_state": "conversation",
+            "states": {
+                "conversation": {
+                    "system_prompt": self.system_prompt,
+                    "opening_line": self.opening_line,
+                    "branches": [
+                        {"id": b.id, "outcome": b.outcome,
+                         "learner_signals": b.trigger.learner_signals}
+                        for b in self.scenario.branches
+                    ],
+                },
+            },
+            "transitions": [],  # v0.1: no in-flight transitions (G-002)
+        }
+
+
+def build_runtime(scenario: Scenario) -> ScenarioRuntime:
+    """Construct a ScenarioRuntime for the given scenario."""
+    return ScenarioRuntime(scenario=scenario)
+
+
+def build_runtime_from_id(scenario_id: str) -> ScenarioRuntime:
+    """Load + build a runtime by scenario id (convenience for the pipeline)."""
+    from server.scenarios.loader import load
+
+    return build_runtime(load(scenario_id))
+
+
+__all__ = ["ScenarioRuntime", "build_runtime", "build_runtime_from_id"]