docs(milestone): complete v0.1 foundation

---ci---
phase: 0
milestone: v0.1
status: complete
---/ci---

scripts/e2e_smoke.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""End-to-end smoke test (TASK-05-06) — also runnable as a pytest test.
+
+Verifies the full v0.1 loop without live API keys (uses the heuristic
+classifier + a fake LLM for the debrief):
+  start session → simulate 2-3 turns → trigger a branch (classifier) →
+  end session → generate debrief → assert:
+    - debrief non-empty
+    - session + turns + cost logged in SQLite
+    - latency < budget (or logged if exceeded — we log a synthetic value)
+
+Run:
+    python scripts/e2e_smoke.py
+    # or
+    pytest tests/test_e2e.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+# Make the project importable when run from the repo root.
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from db.store import PraxisStore, HARDCODED_LEARNER_ID
+from server.scenarios.loader import load
+from server.scenarios.classifier import classify_branch_sync_heuristic
+from server.scenarios.runtime import build_runtime
+from server.session_recorder import SessionRecorder
+from server.debrief import generate_debrief
+from server.guardrails.customer_service import CustomerServiceGuardrail
+from server.services.base import LLMProvider, LLMStreamChunk
+
+
+class _StubDebriefLLM(LLMProvider):
+    """A stub LLMProvider that returns a canned debrief (no API key needed)."""
+
+    name = "stub-debrief"
+    roleplay_model = "gemma4:cloud"
+    debrief_model = "deepseek-v4-flash:cloud"
+
+    async def chat(self, messages, *, stream=True, model=None, no_think=False):
+        yield LLMStreamChunk(content="You did well acknowledging the customer.", is_first=True)
+
+    async def chat_full(self, messages, *, model=None, no_think=False):
+        return (
+            "- What you did well: you acknowledged the customer's frustration and "
+            "offered a concrete refund.\n"
+            "- What to improve: confirm next steps explicitly.\n"
+            "- Next step: practice the empathy-first opening.",
+            {"output_tokens": 60, "model": model or self.debrief_model},
+        )
+
+
+async def run_e2e(db_path: Path | str | None = None) -> dict:
+    """Run the full e2e smoke sequence; return a result dict for assertions."""
+    if db_path is None:
+        tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+        tmp.close()
+        db_path = tmp.name
+
+    store = PraxisStore(db_path)
+    await store.init()
+
+    # 1. Load the scenario.
+    scenario = load("customer_service_refund_ca_v01")
+    runtime = build_runtime(scenario)
+    assert scenario.failure_mode == "escalates_unresolved", "failure_mode field present (D-009)"
+
+    # 2. Start a session.
+    recorder = SessionRecorder(store, scenario_id=scenario.id)
+    session_id = await recorder.start()
+
+    # 3. Simulate 3 turns (accept-resolution path).
+    turns = [
+        {"role": "assistant", "tts_text": scenario.setup.opening_line, "latency_ms": None},
+        {"role": "user", "asr_text": "I'm really sorry you're frustrated. I can offer a full refund right now.", "latency_ms": 420.0},
+        {"role": "assistant", "tts_text": "A refund? Okay, that's something.", "latency_ms": 510.0},
+        {"role": "user", "asr_text": "Let me confirm the next steps for you.", "latency_ms": 380.0},
+    ]
+    for t in turns:
+        await recorder.log_turn(
+            role=t["role"],
+            asr_text=t.get("asr_text"),
+            tts_text=t.get("tts_text"),
+            latency_ms=t.get("latency_ms"),
+        )
+    recorder.add_audio_minutes(1.2)
+
+    # 4. Classify the branch (R7, offline — heuristic fallback, no API key).
+    learner_turn_texts = [t["asr_text"] for t in turns if t["role"] == "user"]
+    branch_id = classify_branch_sync_heuristic(scenario, learner_turn_texts)
+    runtime.set_branch(branch_id)
+    recorder.set_branch_path([branch_id])
+
+    # 5. Generate the debrief (stub LLM — no API key needed).
+    llm = _StubDebriefLLM()
+    guardrail = CustomerServiceGuardrail()
+    debrief_text, _usage = await generate_debrief(
+        llm, scenario,
+        branch_id=branch_id,
+        outcome=runtime.outcome,
+        debrief_focus=runtime.debrief_focus(),
+        learner_turns=[
+            {"role": t["role"], "asr_text": t.get("asr_text"), "tts_text": t.get("tts_text")}
+            for t in turns
+        ],
+        guardrail=guardrail,
+    )
+    recorder.add_debrief_tokens(input_tokens=150, output_tokens=60)
+
+    # 6. End the session (derives cost + writes outcome + debrief + progress).
+    breakdown = await recorder.end(
+        outcome=runtime.outcome,
+        tts_provider=os.environ.get("PRAXIS_TTS", "cartesia"),
+        debrief_text=debrief_text,
+    )
+
+    # 7. Assert DB state.
+    sess = await store.get_session(session_id)
+    db_turns = await store.get_turns(session_id)
+    assert sess is not None, "session row exists"
+    assert sess.outcome == runtime.outcome, f"outcome matches branch: {sess.outcome}"
+    assert sess.branch_path == [branch_id], "branch path logged"
+    assert sess.cost_estimated_cents is not None and sess.cost_estimated_cents >= 0, "cost non-null"
+    assert sess.debrief_text == debrief_text, "debrief text persisted"
+    assert len(db_turns) == len(turns), f"all {len(turns)} turns logged"
+    assert breakdown.derived_cents >= 0, "cost breakdown derived"
+
+    # Synthetic latency (real latency comes from the live pipeline; here we
+    # log the max turn latency as a proxy and check against the budget).
+    max_latency = max((t.get("latency_ms") or 0) for t in turns)
+    budget = 600.0
+    within_budget = max_latency <= budget
+
+    return {
+        "session_id": session_id,
+        "branch_id": branch_id,
+        "outcome": sess.outcome,
+        "turns_logged": len(db_turns),
+        "cost_cents": sess.cost_estimated_cents,
+        "debrief_chars": len(sess.debrief_text or ""),
+        "max_latency_ms": max_latency,
+        "within_budget": within_budget,
+        "budget_ms": budget,
+    }
+
+
+def main() -> int:
+    result = asyncio.run(run_e2e())
+    print("\n" + "=" * 60)
+    print("E2E SMOKE TEST — PASSED")
+    print("=" * 60)
+    for k, v in result.items():
+        print(f"  {k}: {v}")
+    print()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/probe_cartesia.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python3
+"""R2 probe — Cartesia Sonic TTS first-audio-byte latency.
+
+Per PLAN.md SLICE-01 TASK-01-03: WebSocket to Cartesia Sonic, send a sample text
+chunk, measure first-audio-byte latency over 20 iterations; log min/median/p95.
+
+Exit code 0 in all cases:
+  - If CARTESIA_API_KEY is missing, print KEY_MISSING and exit 0.
+  - If present, run the live probe and print a latency table.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import statistics
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+try:
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:  # pragma: no cover
+    pass
+
+
+def _banner(msg: str) -> None:
+    print("\n" + "=" * 72)
+    print(msg)
+    print("=" * 72 + "\n")
+
+
+def _require_key() -> str | None:
+    key = os.environ.get("CARTESIA_API_KEY", "").strip()
+    if not key:
+        _banner(
+            "KEY_MISSING — CARTESIA_API_KEY not set.\n"
+            "  Cannot run live Cartesia probe. Probe infrastructure is built\n"
+            "  and ready; live measurements are pending API key provisioning.\n"
+            "  Set CARTESIA_API_KEY in .env (see .env.example) and re-run."
+        )
+        return None
+    return key
+
+
+SAMPLE_TEXT = (
+    "Hi, I received my order yesterday and the item is cracked. "
+    "I want my money back."
+)
+
+CARTESIA_WS_URL = "wss://api.cartesia.ai/tts/websocket"
+DEFAULT_VOICE_ID = "a3536a36-1d18-4efb-a95a-7c44b7b5e384"
+
+
+async def _probe_once(api_key: str, voice_id: str, model_id: str) -> float | None:
+    """Open Cartesia WS, request TTS, return ms-to-first-audio-byte."""
+    import websockets
+
+    headers = [("x-api-key", api_key), ("cartesia-version", "2024-06-10")]
+    t0 = time.perf_counter()
+    first_audio_ms: float | None = None
+
+    try:
+        async with websockets.connect(
+            CARTESIA_WS_URL, additional_headers=headers, open_timeout=10
+        ) as ws:
+            req = {
+                "model_id": model_id,
+                "transcript": SAMPLE_TEXT,
+                "voice": {"id": voice_id},
+                "output_format": {
+                    "container": "raw",
+                    "encoding": "pcm_s16le",
+                    "sample_rate": 24000,
+                },
+                "stream": True,
+            }
+            await ws.send(json.dumps(req))
+            # Read frames until we get the first audio chunk.
+            while True:
+                msg = await asyncio.wait_for(ws.recv(), timeout=10)
+                if isinstance(msg, (bytes, bytearray)):
+                    first_audio_ms = (time.perf_counter() - t0) * 1000.0
+                    break
+                # JSON control messages (e.g. done) — ignore until audio.
+                if isinstance(msg, str):
+                    data = json.loads(msg)
+                    if data.get("type") == "done":
+                        break
+    except Exception as exc:  # pragma: no cover - network/auth errors
+        print(f"  [probe] Cartesia connection failed: {exc}")
+        return None
+
+    return first_audio_ms
+
+
+async def run_live(api_key: str, iterations: int, voice_id: str, model_id: str) -> list[float]:
+    samples: list[float] = []
+    print(f"  Running {iterations} Cartesia Sonic iterations (voice={voice_id})...")
+    for i in range(iterations):
+        ms = await _probe_once(api_key, voice_id, model_id)
+        if ms is not None:
+            samples.append(ms)
+            print(f"  [{i + 1:2d}/{iterations}] first-audio: {ms:6.1f} ms")
+        else:
+            print(f"  [{i + 1:2d}/{iterations}] no audio received (skipped)")
+        await asyncio.sleep(0.3)
+    return samples
+
+
+def _summarize(samples: list[float], label: str) -> dict:
+    if not samples:
+        print(f"\n  {label}: no samples collected.\n")
+        return {"label": label, "n": 0}
+    s = sorted(samples)
+    p95 = s[int(0.95 * (len(s) - 1))]
+    row = {
+        "label": label,
+        "n": len(s),
+        "min_ms": round(min(s), 1),
+        "median_ms": round(statistics.median(s), 1),
+        "p95_ms": round(p95, 1),
+        "mean_ms": round(statistics.mean(s), 1),
+    }
+    print(
+        f"  {label}: n={row['n']} min={row['min_ms']:.1f} "
+        f"median={row['median_ms']:.1f} p95={row['p95_ms']:.1f} "
+        f"mean={row['mean_ms']:.1f} (ms)"
+    )
+    return row
+
+
+async def amain() -> int:
+    parser = argparse.ArgumentParser(description="R2 Cartesia Sonic latency probe")
+    parser.add_argument("--iterations", type=int, default=20)
+    parser.add_argument("--voice-id", default=os.environ.get("CARTESIA_VOICE_ID", DEFAULT_VOICE_ID))
+    parser.add_argument("--model-id", default="sonic-2")
+    parser.add_argument("--out", default=None)
+    args = parser.parse_args()
+
+    _banner("R2 PROBE — Cartesia Sonic first-audio-byte latency")
+    api_key = _require_key()
+    if api_key is None:
+        return 0
+
+    samples = await run_live(api_key, args.iterations, args.voice_id, args.model_id)
+    summary = _summarize(samples, "cartesia_sonic_first_audio")
+    print()
+    if args.out:
+        Path(args.out).write_text(json.dumps(summary, indent=2))
+        print(f"  Wrote {args.out}")
+    return 0
+
+
+def main() -> int:
+    return asyncio.run(amain())
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/probe_deepgram.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""R1 probe — Deepgram Nova-3 streaming ASR first-partial-transcript latency.
+
+Per PLAN.md SLICE-01 TASK-01-02: measure first-partial-transcript latency from a
+sample audio file (synthesized PCM) over 20 iterations; log min/median/p95.
+
+Exit code 0 in all cases:
+  - If DEEPGRAM_API_KEY is missing, print a clear KEY_MISSING banner and exit 0
+    (the probe infrastructure is the deliverable; live numbers come when keys
+    are provisioned).
+  - If the key is present, run the live probe and print a latency table.
+
+Usage:
+    python scripts/probe_deepgram.py [--iterations N] [--model nova-3]
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import statistics
+import sys
+import time
+from pathlib import Path
+
+# Make the project importable when run from the repo root.
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+try:
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:  # pragma: no cover - dotenv is a declared dep
+    pass
+
+
+def _banner(msg: str) -> None:
+    print("\n" + "=" * 72)
+    print(msg)
+    print("=" * 72 + "\n")
+
+
+def _require_key() -> str | None:
+    """Return the Deepgram API key or None (with a printed banner if missing)."""
+    key = os.environ.get("DEEPGRAM_API_KEY", "").strip()
+    if not key:
+        _banner(
+            "KEY_MISSING — DEEPGRAM_API_KEY not set.\n"
+            "  Cannot run live Deepgram probe. Probe infrastructure is built\n"
+            "  and ready; live measurements are pending API key provisioning.\n"
+            "  Set DEEPGRAM_API_KEY in .env (see .env.example) and re-run."
+        )
+        return None
+    return key
+
+
+def _synth_pcm(duration_s: float = 2.0, sample_rate: int = 16000) -> bytes:
+    """Synthesize a short mono 16-bit PCM buffer (silence + a low tone).
+
+    Deepgram needs real audio frames; we generate a recognizable signal so the
+    streaming endpoint returns a partial. The exact transcript content is not
+    the point — the *latency to first partial* is.
+    """
+    import math
+    import struct
+
+    n = int(duration_s * sample_rate)
+    frames = bytearray()
+    for i in range(n):
+        # 220 Hz tone for the first 1.5s, then silence — a clearly voiced segment.
+        if i < int(1.5 * sample_rate):
+            sample = int(16000 * math.sin(2 * math.pi * 220 * i / sample_rate))
+        else:
+            sample = 0
+        frames += struct.pack("<h", sample)
+    return bytes(frames)
+
+
+DEEPGRAM_WS_URL = "wss://api.deepgram.com/v1/listen"
+
+
+async def _probe_once(api_key: str, model: str, pcm: bytes, sample_rate: int) -> float | None:
+    """Open a Deepgram streaming WebSocket, send PCM, return ms-to-first-partial.
+
+    Uses the raw Deepgram streaming WebSocket API (not the SDK) so the probe is
+    independent of SDK version churn and measures the actual network path.
+    """
+    import websockets
+
+    params = (
+        f"?model={model}&language=en&encoding=linear16&channels=1"
+        f"&sample_rate={sample_rate}&interim_results=true&endpointing=300"
+    )
+    headers = [("Authorization", f"Token {api_key}")]
+    t0 = time.perf_counter()
+    first_partial_ms: float | None = None
+
+    try:
+        async with websockets.connect(
+            DEEPGRAM_WS_URL + params, additional_headers=headers, open_timeout=10
+        ) as ws:
+            # Send in small chunks to mimic real streaming.
+            chunk = 3200  # 100ms of 16kHz mono 16-bit
+            for i in range(0, len(pcm), chunk):
+                await ws.send(pcm[i : i + chunk])
+                await asyncio.sleep(0.02)
+            # Wait for the first transcript message.
+            try:
+                while True:
+                    msg = await asyncio.wait_for(ws.recv(), timeout=5)
+                    if isinstance(msg, str):
+                        data = json.loads(msg)
+                        if data.get("type") == "Results":
+                            channel = data.get("channel", {})
+                            alts = channel.get("alternatives", [])
+                            if alts and alts[0].get("transcript", "").strip():
+                                first_partial_ms = (time.perf_counter() - t0) * 1000.0
+                                break
+            except asyncio.TimeoutError:
+                pass
+            # Signal close.
+            try:
+                await ws.send(json.dumps({"type": "CloseStream"}))
+            except Exception:
+                pass
+    except Exception as exc:  # pragma: no cover - network/auth errors
+        print(f"  [probe] Deepgram connection failed: {exc}")
+        return None
+
+    return first_partial_ms
+
+
+async def run_live(api_key: str, iterations: int, model: str) -> list[float]:
+    sample_rate = 16000
+    pcm = _synth_pcm(duration_s=2.0, sample_rate=sample_rate)
+    samples: list[float] = []
+    print(f"  Running {iterations} Deepgram Nova-3 iterations (model={model})...")
+    for i in range(iterations):
+        ms = await _probe_once(api_key, model, pcm, sample_rate)
+        if ms is not None:
+            samples.append(ms)
+            print(f"  [{i + 1:2d}/{iterations}] first-partial: {ms:6.1f} ms")
+        else:
+            print(f"  [{i + 1:2d}/{iterations}] no partial received (skipped)")
+        await asyncio.sleep(0.3)
+    return samples
+
+
+def _summarize(samples: list[float], label: str) -> dict:
+    if not samples:
+        print(f"\n  {label}: no samples collected.\n")
+        return {"label": label, "n": 0}
+    s = sorted(samples)
+    p95 = s[int(0.95 * (len(s) - 1))]
+    row = {
+        "label": label,
+        "n": len(s),
+        "min_ms": round(min(s), 1),
+        "median_ms": round(statistics.median(s), 1),
+        "p95_ms": round(p95, 1),
+        "mean_ms": round(statistics.mean(s), 1),
+    }
+    print(
+        f"  {label}: n={row['n']} min={row['min_ms']:.1f} "
+        f"median={row['median_ms']:.1f} p95={row['p95_ms']:.1f} "
+        f"mean={row['mean_ms']:.1f} (ms)"
+    )
+    return row
+
+
+async def amain() -> int:
+    parser = argparse.ArgumentParser(description="R1 Deepgram Nova-3 latency probe")
+    parser.add_argument("--iterations", type=int, default=20)
+    parser.add_argument("--model", default=os.environ.get("DEEPGRAM_MODEL", "nova-3"))
+    parser.add_argument("--out", default=None, help="optional JSON results path")
+    args = parser.parse_args()
+
+    _banner("R1 PROBE — Deepgram Nova-3 first-partial-transcript latency")
+    api_key = _require_key()
+    if api_key is None:
+        return 0
+
+    samples = await run_live(api_key, args.iterations, args.model)
+    summary = _summarize(samples, "deepgram_nova3_first_partial")
+    print()
+    if args.out:
+        Path(args.out).write_text(json.dumps(summary, indent=2))
+        print(f"  Wrote {args.out}")
+    return 0
+
+
+def main() -> int:
+    return asyncio.run(amain())
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/probe_e2e.py

@@ -0,0 +1,348 @@
+#!/usr/bin/env python3
+"""R4 probe — integrated three-hop end-to-end latency.
+
+Per PLAN.md SLICE-01 TASK-01-05: feed a sample ASR transcript → Ollama
+gemma4:cloud streaming → Cartesia TTS streaming; measure end-to-end
+(transcript-in → first-audio-out). Run 10 iterations. Also measure the same
+path with Piper self-hosted (if Piper can be stood up locally; otherwise note
+as pending and pre-stage in SLICE-02).
+
+Exit code 0 in all cases:
+  - If OLLAMA_API_KEY or CARTESIA_API_KEY is missing, print KEY_MISSING and
+    exit 0 (the probe infrastructure is the deliverable).
+  - If present, run the live integrated probe and print e2e latency.
+
+The Piper leg is invoked only if PRAXIS_TTS=piper is set AND a Piper voice
+model is available; otherwise it is documented as pre-staged (R4 mitigation).
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import statistics
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+try:
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:  # pragma: no cover
+    pass
+
+
+def _banner(msg: str) -> None:
+    print("\n" + "=" * 72)
+    print(msg)
+    print("=" * 72 + "\n")
+
+
+def _missing(keys: list[str]) -> None:
+    _banner(
+        "KEY_MISSING — " + ", ".join(keys) + " not set.\n"
+        "  Cannot run live integrated e2e probe. Probe infrastructure is built\n"
+        "  and ready; live measurements are pending API key provisioning.\n"
+        "  Set the missing key(s) in .env (see .env.example) and re-run."
+    )
+
+
+CHAT_URL = os.environ.get("OLLAMA_CHAT_URL", "https://ollama.com/api/chat")
+CARTESIA_WS_URL = "wss://api.cartesia.ai/tts/websocket"
+ROLEPLAY_MODEL = os.environ.get("OLLAMA_ROLEPLAY_MODEL", "gemma4:cloud")
+DEFAULT_VOICE_ID = "a3536a36-1d18-4efb-a95a-7c44b7b5e384"
+
+# The "transcript-in" — a realistic ASR final transcript from the learner.
+SAMPLE_TRANSCRIPT = "Hi, I want to help you with your order. What happened?"
+SYSTEM_PROMPT = (
+    "You are Jordan, a customer who received a damaged product. "
+    "You are frustrated but not abusive. Stay in character. Keep responses "
+    "to 1-2 sentences."
+)
+
+
+async def _ollama_first_token(api_key: str) -> tuple[str | None, float | None, str | None]:
+    """Stream Ollama gemma4:cloud, return (full_text, ttft_ms, error)."""
+    import httpx
+
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    body = {
+        "model": ROLEPLAY_MODEL,
+        "messages": [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": SAMPLE_TRANSCRIPT},
+        ],
+        "stream": True,
+    }
+    t0 = time.perf_counter()
+    ttft_ms: float | None = None
+    chunks: list[str] = []
+
+    try:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            async with client.stream("POST", CHAT_URL, headers=headers, json=body) as resp:
+                if resp.status_code != 200:
+                    text = await resp.aread()
+                    return None, None, f"HTTP {resp.status_code}: {text[:200]!r}"
+                async for line in resp.aiter_lines():
+                    if not line:
+                        continue
+                    try:
+                        chunk = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+                    content = chunk.get("message", {}).get("content", "")
+                    if content:
+                        if ttft_ms is None:
+                            ttft_ms = (time.perf_counter() - t0) * 1000.0
+                        chunks.append(content)
+    except Exception as exc:  # pragma: no cover
+        return None, None, f"connection error: {exc}"
+
+    return "".join(chunks), ttft_ms, None
+
+
+async def _cartesia_first_audio(
+    api_key: str, text: str, voice_id: str, model_id: str
+) -> tuple[float | None, str | None]:
+    """Send text to Cartesia WS, return (first_audio_ms_from_t0, error)."""
+    import websockets
+
+    headers = [("x-api-key", api_key), ("cartesia-version", "2024-06-10")]
+    t0 = time.perf_counter()
+    first_audio_ms: float | None = None
+
+    try:
+        async with websockets.connect(
+            CARTESIA_WS_URL, additional_headers=headers, open_timeout=10
+        ) as ws:
+            req = {
+                "model_id": model_id,
+                "transcript": text,
+                "voice": {"id": voice_id},
+                "output_format": {
+                    "container": "raw",
+                    "encoding": "pcm_s16le",
+                    "sample_rate": 24000,
+                },
+                "stream": True,
+            }
+            await ws.send(json.dumps(req))
+            while True:
+                msg = await asyncio.wait_for(ws.recv(), timeout=10)
+                if isinstance(msg, (bytes, bytearray)):
+                    first_audio_ms = (time.perf_counter() - t0) * 1000.0
+                    break
+                if isinstance(msg, str):
+                    data = json.loads(msg)
+                    if data.get("type") == "done":
+                        break
+    except Exception as exc:  # pragma: no cover
+        return None, f"cartesia error: {exc}"
+
+    return first_audio_ms, None
+
+
+async def _piper_first_audio(text: str) -> tuple[float | None, str | None]:
+    """Synthesize via Piper self-hosted, return (first_audio_ms, error).
+
+    Piper pre-staging note (R4 mitigation): Piper is pre-staged as the
+    production v0.1 TTS fallback per ARCHITECTURE.md. The voice model must be
+    downloaded separately (see docs/latency-report.md). If not available,
+    returns an error string that the caller documents as pending.
+    """
+    try:
+        from piper import PiperVoice  # type: ignore
+    except ImportError:
+        return None, "piper-tts not installed (pre-staged for SLICE-02)"
+
+    model_path = os.environ.get("PIPER_VOICE_MODEL", "")
+    if not model_path or not Path(model_path).exists():
+        return None, "PIPER_VOICE_MODEL not set or file missing (pre-staged for SLICE-02)"
+
+    import io
+
+    t0 = time.perf_counter()
+    try:
+        voice = PiperVoice.load(model_path)
+        wav_bytes = io.BytesIO()
+        for chunk in voice.synthesize(text):
+            wav_bytes.write(chunk.audio_int16_bytes)
+        first_audio_ms = (time.perf_counter() - t0) * 1000.0
+        return first_audio_ms, None
+    except Exception as exc:  # pragma: no cover
+        return None, f"piper error: {exc}"
+
+
+async def _e2e_once_cartesia(ollama_key: str, cartesia_key: str, voice_id: str, model_id: str) -> dict:
+    """Run the integrated ASR-transcript → Ollama → Cartesia path once."""
+    t_start = time.perf_counter()
+    text, ttft_ms, llm_err = await _ollama_first_token(ollama_key)
+    if llm_err or not text:
+        return {"ok": False, "error": llm_err or "empty LLM output", "ttft_ms": None}
+    tts_ms, tts_err = await _cartesia_first_audio(cartesia_key, text, voice_id, model_id)
+    if tts_err or tts_ms is None:
+        return {"ok": False, "error": tts_err or "no TTS audio", "ttft_ms": ttft_ms}
+    e2e_ms = (time.perf_counter() - t_start) * 1000.0
+    return {
+        "ok": True,
+        "ttft_ms": ttft_ms,
+        "tts_first_audio_ms": tts_ms,
+        "e2e_ms": e2e_ms,
+        "llm_text": text[:80],
+    }
+
+
+async def _e2e_once_piper(ollama_key: str) -> dict:
+    """Run the integrated ASR-transcript → Ollama → Piper path once."""
+    t_start = time.perf_counter()
+    text, ttft_ms, llm_err = await _ollama_first_token(ollama_key)
+    if llm_err or not text:
+        return {"ok": False, "error": llm_err or "empty LLM output", "ttft_ms": None}
+    tts_ms, tts_err = await _piper_first_audio(text)
+    if tts_err or tts_ms is None:
+        return {"ok": False, "error": tts_err or "no TTS audio", "ttft_ms": ttft_ms, "piper_pending": True}
+    e2e_ms = (time.perf_counter() - t_start) * 1000.0
+    return {
+        "ok": True,
+        "ttft_ms": ttft_ms,
+        "tts_first_audio_ms": tts_ms,
+        "e2e_ms": e2e_ms,
+        "llm_text": text[:80],
+    }
+
+
+def _summarize(samples: list[float], label: str) -> dict:
+    if not samples:
+        print(f"  {label}: no samples collected.")
+        return {"label": label, "n": 0}
+    s = sorted(samples)
+    p95 = s[int(0.95 * (len(s) - 1))]
+    row = {
+        "label": label,
+        "n": len(s),
+        "min_ms": round(min(s), 1),
+        "median_ms": round(statistics.median(s), 1),
+        "p95_ms": round(p95, 1),
+        "mean_ms": round(statistics.mean(s), 1),
+    }
+    print(
+        f"  {label}: n={row['n']} min={row['min_ms']:.1f} "
+        f"median={row['median_ms']:.1f} p95={row['p95_ms']:.1f} "
+        f"mean={row['mean_ms']:.1f} (ms)"
+    )
+    return row
+
+
+async def amain() -> int:
+    parser = argparse.ArgumentParser(description="R4 integrated e2e latency probe")
+    parser.add_argument("--iterations", type=int, default=10)
+    parser.add_argument("--voice-id", default=os.environ.get("CARTESIA_VOICE_ID", DEFAULT_VOICE_ID))
+    parser.add_argument("--cartesia-model", default="sonic-2")
+    parser.add_argument("--out", default=None)
+    parser.add_argument("--piper", action="store_true", help="also run the Piper leg")
+    args = parser.parse_args()
+
+    _banner("R4 PROBE — integrated three-hop e2e (transcript → Ollama → TTS)")
+    ollama_key = os.environ.get("OLLAMA_API_KEY", "").strip()
+    cartesia_key = os.environ.get("CARTESIA_API_KEY", "").strip()
+
+    missing = []
+    if not ollama_key:
+        missing.append("OLLAMA_API_KEY")
+    if not cartesia_key:
+        missing.append("CARTESIA_API_KEY")
+    if missing:
+        _missing(missing)
+        return 0
+
+    # ── Cartesia leg ────────────────────────────────────────────────────────
+    print(f"\n  Cartesia leg — {args.iterations} iterations:")
+    e2e_samples: list[float] = []
+    ttft_samples: list[float] = []
+    tts_samples: list[float] = []
+    for i in range(args.iterations):
+        r = await _e2e_once_cartesia(ollama_key, cartesia_key, args.voice_id, args.cartesia_model)
+        if r.get("ok"):
+            e2e_samples.append(r["e2e_ms"])
+            ttft_samples.append(r["ttft_ms"])
+            tts_samples.append(r["tts_first_audio_ms"])
+            print(f"  [{i + 1:2d}/{args.iterations}] e2e={r['e2e_ms']:.1f}ms "
+                  f"(llm_ttft={r['ttft_ms']:.1f}, tts={r['tts_first_audio_ms']:.1f})")
+        else:
+            print(f"  [{i + 1:2d}/{args.iterations}] error: {r.get('error')}")
+        await asyncio.sleep(0.5)
+
+    print()
+    e2e_summary = _summarize(e2e_samples, "e2e_cartesia")
+    ttft_summary = _summarize(ttft_samples, "e2e_cartesia_llm_ttft")
+    tts_summary = _summarize(tts_samples, "e2e_cartesia_tts_first_audio")
+
+    # ── Piper leg (optional / pre-staged) ───────────────────────────────────
+    piper_summary: dict = {}
+    if args.piper:
+        print(f"\n  Piper leg — {args.iterations} iterations:")
+        p_e2e: list[float] = []
+        p_ttft: list[float] = []
+        p_tts: list[float] = []
+        for i in range(args.iterations):
+            r = await _e2e_once_piper(ollama_key)
+            if r.get("ok"):
+                p_e2e.append(r["e2e_ms"])
+                p_ttft.append(r["ttft_ms"])
+                p_tts.append(r["tts_first_audio_ms"])
+                print(f"  [{i + 1:2d}/{args.iterations}] e2e={r['e2e_ms']:.1f}ms")
+            elif r.get("piper_pending"):
+                print(f"  [{i + 1:2d}/{args.iterations}] Piper pre-staged (pending voice model) — skipping")
+                break
+            else:
+                print(f"  [{i + 1:2d}/{args.iterations}] error: {r.get('error')}")
+            await asyncio.sleep(0.5)
+        print()
+        piper_summary = _summarize(p_e2e, "e2e_piper")
+    else:
+        print("\n  Piper leg not requested (--piper). Piper is pre-staged as the R4 "
+              "mitigation per ARCHITECTURE.md; live Piper measurement pending "
+              "voice-model provisioning (see docs/latency-report.md).")
+
+    # ── Budget comparison ───────────────────────────────────────────────────
+    budget = 600.0
+    print(f"\n  Latency budget: {budget:.0f}ms")
+    if e2e_samples:
+        med = statistics.median(e2e_samples)
+        over = med > budget
+        print(f"  Cartesia median e2e: {med:.1f}ms — {'OVER' if over else 'WITHIN'} budget "
+              f"(delta {med - budget:+.1f}ms)")
+    if piper_summary.get("n"):
+        # type: ignore
+        med = piper_summary.get("median_ms")
+        if med:
+            over = med > budget
+            print(f"  Piper   median e2e: {med:.1f}ms — {'OVER' if over else 'WITHIN'} budget "
+                  f"(delta {med - budget:+.1f}ms)")
+
+    print()
+    if args.out:
+        result = {
+            "e2e_cartesia": e2e_summary,
+            "e2e_cartesia_llm_ttft": ttft_summary,
+            "e2e_cartesia_tts_first_audio": tts_summary,
+            "e2e_piper": piper_summary,
+            "budget_ms": budget,
+        }
+        Path(args.out).write_text(json.dumps(result, indent=2))
+        print(f"  Wrote {args.out}")
+    return 0
+
+
+def main() -> int:
+    return asyncio.run(amain())
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/probe_ollama.py

@@ -0,0 +1,226 @@
+#!/usr/bin/env python3
+"""R3 probe — Ollama Cloud direct-API time-to-first-token (TTFT).
+
+Per PLAN.md SLICE-01 TASK-01-04: direct API call to https://ollama.com/api/chat
+with OLLAMA_API_KEY bearer, model gemma4:cloud, stream=True, measure TTFT over
+20 iterations; also probe deepseek-v4-flash:cloud no-think mode TTFT. Log
+min/median/p95 + any throttle events (R5).
+
+Exit code 0 in all cases:
+  - If OLLAMA_API_KEY is missing, print KEY_MISSING and exit 0.
+  - If present, run the live probe for both models and print TTFT tables.
+
+R6 note: this probe also confirms the Ollama Cloud direct API is callable with a
+bearer token (R6). If it returns 401/403, that is recorded as a throttle/auth
+event, not a crash.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import statistics
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+try:
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:  # pragma: no cover
+    pass
+
+
+def _banner(msg: str) -> None:
+    print("\n" + "=" * 72)
+    print(msg)
+    print("=" * 72 + "\n")
+
+
+def _require_key() -> str | None:
+    key = os.environ.get("OLLAMA_API_KEY", "").strip()
+    if not key:
+        _banner(
+            "KEY_MISSING — OLLAMA_API_KEY not set.\n"
+            "  Cannot run live Ollama Cloud probe. Probe infrastructure is built\n"
+            "  and ready; live measurements are pending API key provisioning.\n"
+            "  Set OLLAMA_API_KEY in .env (see .env.example) and re-run."
+        )
+        return None
+    return key
+
+
+CHAT_URL = os.environ.get("OLLAMA_CHAT_URL", "https://ollama.com/api/chat")
+
+ROLEPLAY_MODEL = os.environ.get("OLLAMA_ROLEPLAY_MODEL", "gemma4:cloud")
+DEBRIEF_MODEL = os.environ.get("OLLAMA_DEBRIEF_MODEL", "deepseek-v4-flash:cloud")
+
+# A short role-play prompt that should produce a fast first token.
+ROLEPLAY_MESSAGES = [
+    {
+        "role": "system",
+        "content": (
+            "You are Jordan, a customer who received a damaged product. "
+            "You are frustrated but not abusive. Stay in character. Keep "
+            "responses to 1-2 sentences."
+        ),
+    },
+    {"role": "user", "content": "Hi, I want to help you with your order. What happened?"},
+]
+
+# Debrief prompt — no_think mode for latency (D-020).
+DEBRIEF_MESSAGES = [
+    {
+        "role": "system",
+        "content": (
+            "You are a coaching mentor. Produce a concise (3-bullet) debrief "
+            "about the learner's customer-service performance. "
+            "Do not reason step-by-step; respond directly."
+        ),
+    },
+    {
+        "role": "user",
+        "content": "The learner said: 'I'm sorry you're upset. I can offer a refund.'",
+    },
+]
+
+
+async def _probe_once(
+    api_key: str, model: str, messages: list[dict], no_think: bool
+) -> tuple[float | None, str | None]:
+    """Call Ollama Cloud /api/chat streaming, return (ttft_ms, error_or_none)."""
+    import httpx
+
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    body: dict = {"model": model, "messages": messages, "stream": True}
+    if no_think:
+        # Ollama no-think mode for deepseek-v4-flash:cloud (D-020).
+        body["think"] = False
+
+    t0 = time.perf_counter()
+    ttft_ms: float | None = None
+
+    try:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            async with client.stream(
+                "POST", CHAT_URL, headers=headers, json=body
+            ) as resp:
+                if resp.status_code != 200:
+                    text = await resp.aread()
+                    return None, f"HTTP {resp.status_code}: {text[:200]!r}"
+                async for line in resp.aiter_lines():
+                    if not line:
+                        continue
+                    try:
+                        chunk = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+                    msg = chunk.get("message", {})
+                    content = msg.get("content", "")
+                    if content and ttft_ms is None:
+                        ttft_ms = (time.perf_counter() - t0) * 1000.0
+                        break
+    except Exception as exc:  # pragma: no cover - network errors
+        return None, f"connection error: {exc}"
+
+    return ttft_ms, None
+
+
+async def run_model(
+    api_key: str, model: str, messages: list[dict], iterations: int, label: str, no_think: bool
+) -> tuple[list[float], list[str]]:
+    samples: list[float] = []
+    errors: list[str] = []
+    print(f"  Running {iterations} iterations for {label} (model={model}, no_think={no_think})...")
+    for i in range(iterations):
+        ms, err = await _probe_once(api_key, model, messages, no_think)
+        if ms is not None:
+            samples.append(ms)
+            print(f"  [{i + 1:2d}/{iterations}] TTFT: {ms:6.1f} ms")
+        else:
+            errors.append(err or "unknown")
+            print(f"  [{i + 1:2d}/{iterations}] error: {err}")
+        await asyncio.sleep(0.5)
+    return samples, errors
+
+
+def _summarize(samples: list[float], label: str) -> dict:
+    if not samples:
+        print(f"\n  {label}: no samples collected.\n")
+        return {"label": label, "n": 0}
+    s = sorted(samples)
+    p95 = s[int(0.95 * (len(s) - 1))]
+    row = {
+        "label": label,
+        "n": len(s),
+        "min_ms": round(min(s), 1),
+        "median_ms": round(statistics.median(s), 1),
+        "p95_ms": round(p95, 1),
+        "mean_ms": round(statistics.mean(s), 1),
+    }
+    print(
+        f"  {label}: n={row['n']} min={row['min_ms']:.1f} "
+        f"median={row['median_ms']:.1f} p95={row['p95_ms']:.1f} "
+        f"mean={row['mean_ms']:.1f} (ms)"
+    )
+    return row
+
+
+async def amain() -> int:
+    parser = argparse.ArgumentParser(description="R3 Ollama Cloud TTFT probe")
+    parser.add_argument("--iterations", type=int, default=20)
+    parser.add_argument("--out", default=None)
+    args = parser.parse_args()
+
+    _banner("R3 PROBE — Ollama Cloud direct-API time-to-first-token")
+    api_key = _require_key()
+    if api_key is None:
+        return 0
+
+    # R6: confirms the direct API + bearer works for the role-play model.
+    rp_samples, rp_errors = await run_model(
+        api_key, ROLEPLAY_MODEL, ROLEPLAY_MESSAGES, args.iterations,
+        "ollama_gemma4_cloud_ttft", no_think=False,
+    )
+    rp_summary = _summarize(rp_samples, "ollama_gemma4_cloud_ttft")
+
+    print()
+    # Debrief model with no_think (D-020).
+    db_samples, db_errors = await run_model(
+        api_key, DEBRIEF_MODEL, DEBRIEF_MESSAGES, args.iterations,
+        "ollama_deepseek_v4_flash_nothink_ttft", no_think=True,
+    )
+    db_summary = _summarize(db_samples, "ollama_deepseek_v4_flash_nothink_ttft")
+
+    # R5: log throttle events (any error could indicate throttling/auth).
+    all_errors = rp_errors + db_errors
+    if all_errors:
+        print(f"\n  R5 — {len(all_errors)} error/throttle event(s) recorded:")
+        for e in all_errors[:10]:
+            print(f"    - {e}")
+    else:
+        print("\n  R5 — no throttle/auth events recorded.")
+
+    print()
+    if args.out:
+        result = {
+            "roleplay": rp_summary,
+            "debrief": db_summary,
+            "errors": all_errors,
+        }
+        Path(args.out).write_text(json.dumps(result, indent=2))
+        print(f"  Wrote {args.out}")
+    return 0
+
+
+def main() -> int:
+    return asyncio.run(amain())
+
+
+if __name__ == "__main__":
+    sys.exit(main())