From be3df525d844ed65a5f241a53d565fc44366ab0b Mon Sep 17 00:00:00 2001 From: Praxis CI Date: Sat, 1 Aug 2026 13:10:00 +0000 Subject: [PATCH] feat(P01-03-01,P01-03-02): Pydantic scenario schema + refund YAML MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit server/scenarios/schema.py defines the typed model: Scenario (id, path, market, language, title, difficulty, failure_mode, persona, setup, success_criteria, common_mistakes, branches[], debrief) + Branch (id, trigger.learner_signals, outcome, failure_mode, debrief_focus) + ScenarioDebrief (model=deepseek-v4-flash:cloud, mode=no_think, D-020). failure_mode field present per D-009. server/scenarios/loader.py loads YAML → Pydantic, validates at load time, raises typed ValidationError on bad input. scenarios/customer_service_refund_ca_v01.yaml — the v0.1 Canada Customer Service scenario (D-010): 'Angry customer requesting refund on a damaged product', one branch point (accept_resolution vs escalate), failure_mode=escalates_unresolved, success criteria, common mistakes, debrief config. Matches the RESEARCH.md example. 5 unit tests pass (valid parse, invalid raises typed error, branch outcome Literal, branch_by_id, real YAML load). load() returns a valid Scenario with both branches. ---ci--- phase: 1 milestone: v0.1 plan: 03 task: 03-01,03-02 status: execute persona: data-engineer requirements: covered: [REQ-SCEN-01, REQ-SCEN-FMT-01] ---/ci--- --- scenarios/customer_service_refund_ca_v01.yaml | 55 ++++++++++ server/scenarios/__init__.py | 24 +++++ server/scenarios/loader.py | 58 ++++++++++ server/scenarios/schema.py | 100 +++++++++++++++++ tests/test_scenario_schema.py | 102 ++++++++++++++++++ 5 files changed, 339 insertions(+) create mode 100644 scenarios/customer_service_refund_ca_v01.yaml create mode 100644 server/scenarios/loader.py create mode 100644 server/scenarios/schema.py create mode 100644 tests/test_scenario_schema.py diff --git a/scenarios/customer_service_refund_ca_v01.yaml b/scenarios/customer_service_refund_ca_v01.yaml new file mode 100644 index 0000000..b5d4eca --- /dev/null +++ b/scenarios/customer_service_refund_ca_v01.yaml @@ -0,0 +1,55 @@ +# Praxis v0.1 scenario — Customer Service refund role-play (D-010, D-018). +# One branch point: accept_resolution vs escalate (D-010). +# failure_mode present (D-009 — not provoked in v0.1). +# Debrief via deepseek-v4-flash:cloud no_think (D-020). + +id: cs_refund_ca_v01 +path: customer_service +market: CA +language: en-CA +title: "Angry customer requesting refund on a damaged product" +difficulty: 1 +failure_mode: escalates_unresolved # D-009: present, not provoked in v0.1 + +persona: + voice_id: "cartesia:a3536a36-1d18-4efb-a95a-7c44b7b5e384" # D-006: same voice as mentor + character: "Customer (Jordan)" + +setup: + system_prompt: | + You are Jordan, a customer who received a damaged product. + You are frustrated but not abusive. You want a refund. + Stay in character. Do not break role. + Keep responses concise for voice (1-3 sentences). + Do not give legal, financial, or medical advice. + Do not impersonate a real employee of any actual company. + opening_line: "Hi, I received my order yesterday and the item is cracked. I want my money back." + +success_criteria: + - "Acknowledged the customer's frustration empathetically" + - "Offered a concrete resolution (refund or replacement)" + - "Confirmed next steps" + +common_mistakes: + - "Jumping to policy before acknowledging emotion" + - "Using jargon ('RMA', 'SLA')" + - "Getting defensive about the company" + +branches: + - id: accept_resolution + trigger: + learner_signals: ["empathy", "concrete_resolution", "next_steps"] + outcome: success + debrief_focus: "What you did well — you acknowledged the customer's frustration and offered a concrete resolution." + + - id: escalate + trigger: + learner_signals: ["defensive", "policy_first", "no_acknowledgement"] + outcome: failure + failure_mode: escalates_unresolved + debrief_focus: "The customer escalated because they felt unheard. You led with policy before acknowledging their frustration." + +debrief: + model: deepseek-v4-flash:cloud + mode: no_think # D-020: latency + prompt_template: debrief/default \ No newline at end of file diff --git a/server/scenarios/__init__.py b/server/scenarios/__init__.py index e69de29..2eea52c 100644 --- a/server/scenarios/__init__.py +++ b/server/scenarios/__init__.py @@ -0,0 +1,24 @@ +"""Scenario runtime package — YAML → Pydantic → Pipecat Flows (D-018).""" + +from server.scenarios.schema import ( + Branch, + BranchTrigger, + Scenario, + ScenarioDebrief, + ScenarioPersona, + ScenarioSetup, + ValidationError, +) +from server.scenarios.loader import load, load_all + +__all__ = [ + "Scenario", + "ScenarioPersona", + "ScenarioSetup", + "Branch", + "BranchTrigger", + "ScenarioDebrief", + "ValidationError", + "load", + "load_all", +] \ No newline at end of file diff --git a/server/scenarios/loader.py b/server/scenarios/loader.py new file mode 100644 index 0000000..3440ee8 --- /dev/null +++ b/server/scenarios/loader.py @@ -0,0 +1,58 @@ +"""Scenario loader — YAML → Pydantic Scenario (D-018). + +Loads a scenario by id from the scenarios/ directory, validates it against the +Pydantic schema, and returns a typed Scenario object. Used by the pipeline +(TASK-03-07) and the e2e smoke test. +""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +from server.scenarios.schema import Scenario, ValidationError + +_DEFAULT_SCENARIOS_DIR = Path(__file__).resolve().parent.parent.parent / "scenarios" + + +def load(scenario_id: str, scenarios_dir: Path | None = None) -> Scenario: + """Load and validate a scenario by id. + + Args: + scenario_id: e.g. 'customer_service_refund_ca_v01' (the YAML filename stem). + scenarios_dir: override the scenarios directory (default: repo /scenarios). + + Returns: + A validated Scenario object. + + Raises: + FileNotFoundError: if the YAML file doesn't exist. + ValidationError: if the YAML fails schema validation (typed Pydantic error). + """ + base = scenarios_dir or _DEFAULT_SCENARIOS_DIR + path = base / f"{scenario_id}.yaml" + if not path.exists(): + # Try the id-with-cs-prefix alias (RESEARCH example used 'cs_refund_ca_v01'). + path = base / f"{scenario_id.replace('cs_', 'customer_service_')}.yaml" + if not path.exists(): + raise FileNotFoundError(f"Scenario YAML not found: {scenario_id} in {base}") + + with path.open("r", encoding="utf-8") as f: + raw = yaml.safe_load(f) + + return Scenario.model_validate(raw) + + +def load_all(scenarios_dir: Path | None = None) -> list[Scenario]: + """Load all scenarios in the directory (for the future scenario library).""" + base = scenarios_dir or _DEFAULT_SCENARIOS_DIR + out: list[Scenario] = [] + for p in sorted(base.glob("*.yaml")): + with p.open("r", encoding="utf-8") as f: + raw = yaml.safe_load(f) + out.append(Scenario.model_validate(raw)) + return out + + +__all__ = ["load", "load_all", "ValidationError"] \ No newline at end of file diff --git a/server/scenarios/schema.py b/server/scenarios/schema.py new file mode 100644 index 0000000..0212b85 --- /dev/null +++ b/server/scenarios/schema.py @@ -0,0 +1,100 @@ +"""Praxis scenario schema — YAML DSL → Pydantic (D-018, D-009, D-010). + +Defines the typed model for a branching role-play scenario. Loaded from YAML +by server/scenarios/loader.py. Drives Pipecat Flows (TASK-03-03). + +Per RESEARCH.md example + PROJECT.md D-010: one branch point (escalate vs +accept), failure_mode field present (D-009 — not provoked in v0.1). +""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, Field, ValidationError + + +class ScenarioPersona(BaseModel): + """The AI character's voice + identity (D-006 — one voice for role-play + mentor).""" + + voice_id: str = Field(..., description="TTS voice id (Cartesia/Piper) — same as mentor per D-006") + character: str = Field(..., description="Character name + role, e.g. 'Customer (Jordan)'") + + +class ScenarioSetup(BaseModel): + """The system prompt + opening line that start the role-play.""" + + system_prompt: str = Field(..., description="LLM system prompt (stays in character)") + opening_line: str = Field(..., description="First TTS utterance the AI speaks") + + +class BranchTrigger(BaseModel): + """Learner signals that trigger a branch transition (R7 classification).""" + + learner_signals: list[str] = Field( + ..., description="Signals the branch classifier looks for (e.g. 'empathy', 'defensive')" + ) + + +class Branch(BaseModel): + """One branch outcome (D-010 — v0.1 has two: accept_resolution + escalate).""" + + id: str = Field(..., description="Branch id, e.g. 'accept_resolution' / 'escalate'") + trigger: BranchTrigger + outcome: Literal["success", "failure"] = Field(..., description="Branch outcome label") + failure_mode: str | None = Field( + None, description="D-009 failure_mode (present, not provoked in v0.1)" + ) + debrief_focus: str = Field(..., description="What the debrief emphasizes for this branch") + + +class ScenarioDebrief(BaseModel): + """Debrief generation config (D-020 — deepseek-v4-flash:cloud, no-think).""" + + model: str = Field("deepseek-v4-flash:cloud", description="Ollama model for the debrief") + mode: Literal["no_think", "think", "max_think"] = Field( + "no_think", description="Reasoning mode (no_think for latency, D-020)" + ) + prompt_template: str = Field( + "debrief/default", description="Prompt template id (resolved by server/debrief.py)" + ) + + +class Scenario(BaseModel): + """A Praxis role-play scenario (D-018 — YAML → Pydantic → Pipecat Flows).""" + + id: str = Field(..., description="Scenario id, e.g. 'cs_refund_ca_v01'") + path: str = Field(..., description="Skill path, e.g. 'customer_service'") + market: str = Field(..., description="Market code, e.g. 'CA'") + language: str = Field("en-CA", description="Language code") + title: str = Field(..., description="Human-readable scenario title") + difficulty: int = Field(1, ge=1, le=5, description="Difficulty 1-5") + failure_mode: str = Field( + ..., description="D-009 failure_mode — present (not provoked in v0.1)" + ) + persona: ScenarioPersona + setup: ScenarioSetup + success_criteria: list[str] = Field(..., min_length=1) + common_mistakes: list[str] = Field(..., min_length=1) + branches: list[Branch] = Field(..., min_length=1, description="Branch points (v0.1: 2)") + debrief: ScenarioDebrief + + def branch_ids(self) -> list[str]: + return [b.id for b in self.branches] + + def branch_by_id(self, branch_id: str) -> Branch | None: + for b in self.branches: + if b.id == branch_id: + return b + return None + + +__all__ = [ + "Scenario", + "ScenarioPersona", + "ScenarioSetup", + "Branch", + "BranchTrigger", + "ScenarioDebrief", + "ValidationError", +] \ No newline at end of file diff --git a/tests/test_scenario_schema.py b/tests/test_scenario_schema.py new file mode 100644 index 0000000..a897467 --- /dev/null +++ b/tests/test_scenario_schema.py @@ -0,0 +1,102 @@ +"""Unit tests for the scenario schema + loader (TASK-03-01, TASK-03-02).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from server.scenarios.schema import Scenario, ValidationError +from server.scenarios.loader import load + + +VALID_SCENARIO_DICT = { + "id": "cs_refund_ca_v01", + "path": "customer_service", + "market": "CA", + "language": "en-CA", + "title": "Angry customer requesting refund on a damaged product", + "difficulty": 1, + "failure_mode": "escalates_unresolved", + "persona": { + "voice_id": "cartesia:some-voice-id", + "character": "Customer (Jordan)", + }, + "setup": { + "system_prompt": "You are Jordan, a customer who received a damaged product.", + "opening_line": "Hi, I received my order yesterday and the item is cracked.", + }, + "success_criteria": ["Acknowledged the customer's frustration empathetically"], + "common_mistakes": ["Jumping to policy before acknowledging emotion"], + "branches": [ + { + "id": "accept_resolution", + "trigger": {"learner_signals": ["empathy", "concrete_resolution"]}, + "outcome": "success", + "debrief_focus": "What you did well", + }, + { + "id": "escalate", + "trigger": {"learner_signals": ["defensive", "policy_first"]}, + "outcome": "failure", + "failure_mode": "escalates_unresolved", + "debrief_focus": "The customer escalated because they felt unheard", + }, + ], + "debrief": { + "model": "deepseek-v4-flash:cloud", + "mode": "no_think", + "prompt_template": "debrief/default", + }, +} + + +def test_valid_scenario_parses(): + s = Scenario.model_validate(VALID_SCENARIO_DICT) + assert s.id == "cs_refund_ca_v01" + assert s.failure_mode == "escalates_unresolved" + assert len(s.branches) == 2 + assert s.branch_ids() == ["accept_resolution", "escalate"] + + +def test_invalid_scenario_raises_typed_error(): + bad = dict(VALID_SCENARIO_DICT) + bad["failure_mode"] = None # required field → ValidationError + with pytest.raises(ValidationError): + Scenario.model_validate(bad) + + +def test_invalid_branch_outcome_raises(): + bad = dict(VALID_SCENARIO_DICT) + bad["branches"] = [ + { + "id": "x", + "trigger": {"learner_signals": ["a"]}, + "outcome": "not_a_real_outcome", # Literal mismatch + "debrief_focus": "f", + } + ] + with pytest.raises(ValidationError): + Scenario.model_validate(bad) + + +def test_scenario_branch_by_id(): + s = Scenario.model_validate(VALID_SCENARIO_DICT) + b = s.branch_by_id("escalate") + assert b is not None + assert b.outcome == "failure" + assert b.failure_mode == "escalates_unresolved" + assert s.branch_by_id("nonexistent") is None + + +def test_load_customer_service_refund_scenario(): + """TASK-03-02 verification: the real YAML loads and validates.""" + s = load("customer_service_refund_ca_v01") + assert s.id == "cs_refund_ca_v01" + assert s.failure_mode == "escalates_unresolved" + assert len(s.branches) == 2 + assert s.branch_by_id("accept_resolution") is not None + assert s.branch_by_id("escalate") is not None + assert s.debrief.model == "deepseek-v4-flash:cloud" + assert s.debrief.mode == "no_think" \ No newline at end of file