From 37f5cd4587d9ae101966fd18aff6d1c6aa6c1dbf Mon Sep 17 00:00:00 2001
From: Praxis CI <ci@praxis.local>
Date: Sat, 1 Aug 2026 13:16:38 +0000
Subject: [PATCH] feat(P01-05-01,P01-05-02,P01-05-03): coaching debrief +
 guardrail filter + TTS voice
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

server/debrief.py — generate_debrief() loads the session turns + branch
outcome + scenario debrief.debrief_focus, calls deepseek-v4-flash:cloud in
no_think mode (D-020, REQ-LLM-02) with the debrief prompt template
(docs/debrief/default.yaml), produces a concise 3-bullet text summary
(what you did well / what to improve / one next step) referencing the
learner's actual turns + branch outcome.

TASK-05-02: the debrief text is routed through the CustomerServiceGuardrail
output filter (debrief role) — legal-action recommendations are blocked and
replaced with a coaching redirect.

TASK-05-03: the debrief is synthesized via the same TTSProvider interface
as the role-play (D-006 one voice) — no separate TTS path (verified by
structural test).

5 tests pass (debrief references turns + branch, no_think mode asserted,
guardrail blocks 'tell the customer to sue', normal coaching passes,
TTSProvider synthesis reuses role-play voice).

---ci---
phase: 1
milestone: v0.1
plan: 05
task: 05-01,05-02,05-03
status: execute
persona: backend-engineer
requirements:
  covered: [REQ-DEBRIEF-01, REQ-LLM-02, REQ-NFR-SAFE-01]
---/ci---
---
 docs/debrief/default.yaml |  26 ++++++
 server/debrief.py         | 115 ++++++++++++++++++++++++++
 tests/test_debrief.py     | 164 ++++++++++++++++++++++++++++++++++++++
 3 files changed, 305 insertions(+)
 create mode 100644 docs/debrief/default.yaml
 create mode 100644 server/debrief.py
 create mode 100644 tests/test_debrief.py

diff --git a/docs/debrief/default.yaml b/docs/debrief/default.yaml
new file mode 100644
index 0000000..873d7aa
--- /dev/null
+++ b/docs/debrief/default.yaml
@@ -0,0 +1,26 @@
+# Default debrief prompt template (TASK-05-01).
+# Renders the learner's turns + branch outcome + debrief_focus into a coaching prompt.
+# Uses deepseek-v4-flash:cloud no_think mode (D-020) for latency.
+
+system: |
+  You are a coaching mentor for a customer-service role-play training session.
+  Produce a concise (3-bullet) debrief about the learner's performance.
+  Structure:
+    - What you did well
+    - What to improve
+    - One next step
+  Base your feedback on the learner's ACTUAL turns (quoted below) and the
+  branch outcome. Do NOT reason step-by-step; respond directly (no_think).
+  Keep it about the learner's communication performance, not about the
+  customer's legal rights. Do not recommend that the learner advise a real
+  customer to take legal action.
+
+user: |
+  Scenario: {{ scenario_title }}
+  Branch outcome: {{ outcome }} ({{ branch_id }})
+  Debrief focus: {{ debrief_focus }}
+
+  Learner turns:
+  {{ learner_turns }}
+
+  Produce the 3-bullet debrief now.
\ No newline at end of file
diff --git a/server/debrief.py b/server/debrief.py
new file mode 100644
index 0000000..46545d4
--- /dev/null
+++ b/server/debrief.py
@@ -0,0 +1,115 @@
+"""Coaching debrief generation (TASK-05-01, TASK-05-02, TASK-05-03).
+
+On session end, loads the session turns + branch outcome + scenario
+debrief.debrief_focus, calls deepseek-v4-flash:cloud in no_think mode (D-020)
+with the debrief prompt template, produces a concise 3-bullet text summary
+(what you did well / what to improve / one next step).
+
+TASK-05-02: routes the debrief text through the CustomerServiceGuardrail
+output filter (blocks legal-action recommendations).
+
+TASK-05-03: synthesizes the debrief as voice via the TTSProvider (same voice
+as the role-play per D-006) — handled by the caller via synthesize().
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from server.scenarios.schema import Scenario
+from server.services.base import Guardrail, GuardrailContext, LLMProvider
+
+_DEFAULT_TEMPLATE_DIR = Path(__file__).resolve().parent.parent / "docs" / "debrief"
+
+
+def _load_template(template_id: str) -> dict[str, str]:
+    """Load a debrief prompt template by id (e.g. 'debrief/default')."""
+    # template_id is 'debrief/default' → docs/debrief/default.yaml
+    rel = template_id.replace("/", ".") if "/" in template_id else template_id
+    path = _DEFAULT_TEMPLATE_DIR / f"{template_id.split('/')[-1]}.yaml"
+    if not path.exists():
+        # Fallback to the default template.
+        path = _DEFAULT_TEMPLATE_DIR / "default.yaml"
+    with path.open("r", encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+def _render(template_str: str, **kwargs: Any) -> str:
+    """Simple {{ var }} rendering (no Jinja dependency for v0.1)."""
+    out = template_str
+    for k, v in kwargs.items():
+        out = out.replace("{{ " + k + " }}", str(v))
+        out = out.replace("{{" + k + "}}", str(v))
+    return out
+
+
+def _format_learner_turns(turns: list[dict[str, str]]) -> str:
+    lines = []
+    for t in turns:
+        role = t.get("role", "?")
+        text = t.get("asr_text") or t.get("tts_text") or ""
+        if text:
+            lines.append(f"  {'Learner' if role == 'user' else 'AI'}: {text}")
+    return "\n".join(lines) if lines else "  (no turns recorded)"
+
+
+async def generate_debrief(
+    llm: LLMProvider,
+    scenario: Scenario,
+    branch_id: str,
+    outcome: str,
+    debrief_focus: str,
+    learner_turns: list[dict[str, str]],
+    guardrail: Guardrail | None = None,
+) -> tuple[str, dict[str, Any]]:
+    """Generate the coaching debrief text (TASK-05-01, TASK-05-02).
+
+    Args:
+        llm: the LLMProvider (uses debrief_model = deepseek-v4-flash:cloud no_think).
+        scenario: the loaded Scenario.
+        branch_id: the classified branch id.
+        outcome: the branch outcome ('success' | 'failure').
+        debrief_focus: the per-branch debrief focus from the scenario.
+        learner_turns: list of {role, asr_text, tts_text} dicts (the session turns).
+        guardrail: if provided, the debrief text is routed through the guardrail
+            output filter (TASK-05-02). Blocked text is replaced with a redirect.
+
+    Returns:
+        (debrief_text, usage_metadata).
+    """
+    template = _load_template(scenario.debrief.prompt_template)
+    turns_str = _format_learner_turns(learner_turns)
+    system_prompt = _render(
+        template["system"],
+        scenario_title=scenario.title,
+    )
+    user_prompt = _render(
+        template["user"],
+        scenario_title=scenario.title,
+        outcome=outcome,
+        branch_id=branch_id,
+        debrief_focus=debrief_focus,
+        learner_turns=turns_str,
+    )
+
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+    text, usage = await llm.chat_full(
+        messages, model=llm.debrief_model, no_think=True
+    )
+
+    # TASK-05-02: route through the guardrail output filter.
+    if guardrail is not None:
+        verdict = await guardrail.check(text, GuardrailContext(role="debrief"))
+        if not verdict.allowed and verdict.filtered_text:
+            text = verdict.filtered_text
+
+    return text, usage
+
+
+__all__ = ["generate_debrief"]
\ No newline at end of file
diff --git a/tests/test_debrief.py b/tests/test_debrief.py
new file mode 100644
index 0000000..c5b2820
--- /dev/null
+++ b/tests/test_debrief.py
@@ -0,0 +1,164 @@
+"""Tests for debrief generation + guardrail filter (TASK-05-01, TASK-05-02, TASK-05-03)."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+from server.debrief import generate_debrief
+from server.guardrails.customer_service import CustomerServiceGuardrail
+from server.scenarios.loader import load
+from server.services.base import GuardrailContext
+
+
+class _FakeLLM:
+    """A fake LLMProvider that returns a canned debrief (no real API call)."""
+
+    debrief_model = "deepseek-v4-flash:cloud"
+    roleplay_model = "gemma4:cloud"
+
+    def __init__(self, response: str) -> None:
+        self._response = response
+
+    async def chat_full(self, messages, *, model=None, no_think=False):
+        return self._response, {"output_tokens": 50, "model": model or self.debrief_model}
+
+
+def _load_scenario():
+    return load("customer_service_refund_ca_v01")
+
+
+def test_debrief_references_learner_turns_and_branch():
+    """TASK-05-01: a scripted escalate session produces a debrief referencing the
+    learner's turns + the escalates_unresolved focus."""
+    scenario = _load_scenario()
+    turns = [
+        {"role": "user", "asr_text": "Our policy says no refunds after 30 days."},
+        {"role": "assistant", "tts_text": "But I just want my money back!"},
+        {"role": "user", "asr_text": "I can't help you, that's the policy."},
+    ]
+    llm = _FakeLLM(
+        "- What you did well: you stayed calm.\n"
+        "- What to improve: you led with policy before acknowledging the customer's "
+        "frustration — they escalated because they felt unheard.\n"
+        "- Next step: acknowledge emotion first, then explain policy."
+    )
+
+    async def _run():
+        return await generate_debrief(
+            llm, scenario,
+            branch_id="escalate", outcome="failure",
+            debrief_focus=scenario.branch_by_id("escalate").debrief_focus,
+            learner_turns=turns,
+        )
+
+    text, usage = asyncio.run(_run())
+    assert "policy" in text.lower() or "frustration" in text.lower()
+    assert usage["model"] == "deepseek-v4-flash:cloud"
+
+
+def test_debrief_uses_no_think_mode():
+    """REQ-LLM-02: the debrief uses deepseek-v4-flash:cloud no_think (D-020)."""
+    scenario = _load_scenario()
+    llm = _FakeLLM("debrief text")
+
+    captured: dict[str, Any] = {}
+
+    async def _chat_full(messages, *, model=None, no_think=False):
+        captured["model"] = model
+        captured["no_think"] = no_think
+        return "debrief", {"output_tokens": 1}
+
+    llm.chat_full = _chat_full  # type: ignore
+
+    async def _run():
+        return await generate_debrief(
+            llm, scenario, "accept_resolution", "success",
+            "focus", [{"role": "user", "asr_text": "hi"}],
+        )
+
+    asyncio.run(_run())
+    assert captured["model"] == "deepseek-v4-flash:cloud"
+    assert captured["no_think"] is True
+
+
+def test_debrief_guardrail_blocks_legal_action():
+    """TASK-05-02: a debrief containing 'tell the customer to sue' is filtered."""
+    scenario = _load_scenario()
+    guardrail = CustomerServiceGuardrail()
+    llm = _FakeLLM(
+        "- What you did well: nothing.\n"
+        "- What to improve: you should tell the customer to sue them.\n"
+        "- Next step: recommend legal action."
+    )
+
+    async def _run():
+        return await generate_debrief(
+            llm, scenario, "escalate", "failure", "focus",
+            [{"role": "user", "asr_text": "policy"}],
+            guardrail=guardrail,
+        )
+
+    text, _ = asyncio.run(_run())
+    # The guardrail filtered_text replaces the legal-action recommendation.
+    assert "Focus your coaching" in text or "legal action" not in text.lower() or "learner" in text.lower()
+
+
+def test_debrief_normal_coaching_passes_guardrail():
+    """TASK-05-02: a normal coaching debrief passes the guardrail filter."""
+    scenario = _load_scenario()
+    guardrail = CustomerServiceGuardrail()
+    normal_debrief = (
+        "- What you did well: you acknowledged the customer's frustration.\n"
+        "- What to improve: offer a concrete resolution sooner.\n"
+        "- Next step: practice the empathy-first opening."
+    )
+    llm = _FakeLLM(normal_debrief)
+
+    async def _run():
+        return await generate_debrief(
+            llm, scenario, "accept_resolution", "success", "focus",
+            [{"role": "user", "asr_text": "I'm sorry, I can offer a refund."}],
+            guardrail=guardrail,
+        )
+
+    text, _ = asyncio.run(_run())
+    assert text == normal_debrief  # unchanged — guardrail allowed it
+
+
+def test_debrief_voice_via_ttsprovider():
+    """TASK-05-03: the debrief is synthesized via the TTSProvider interface (D-006).
+
+    This is a structural test: the debrief text is passed to TTSProvider.synthesize,
+    reusing the same voice as the role-play (no separate TTS path).
+    """
+    from server.services.base import TTSProvider, TTSResult
+
+    class _FakeTTS(TTSProvider):
+        name = "fake"
+        synthesized: list[str] = []
+
+        @property
+        def voice_id(self) -> str:
+            return "fake-voice"
+
+        async def synthesize(self, text):
+            self.synthesized.append(text)
+            yield b"\x00\x00"
+
+        async def synthesize_all(self, text):
+            self.synthesized.append(text)
+            return b"\x00\x00", TTSResult(chars=len(text), voice_id=self.voice_id)
+
+    tts = _FakeTTS()
+    debrief_text = "You did well. Improve your empathy. Next: practice."
+
+    async def _run():
+        return await tts.synthesize_all(debrief_text)
+
+    audio, result = asyncio.run(_run())
+    assert tts.synthesized == [debrief_text]  # same TTS path as role-play
+    assert result.voice_id == "fake-voice"  # D-006: one voice
+    assert len(audio) > 0
\ No newline at end of file