feat(P01-04-03,P01-04-04): wire store into pipeline + per-session cost logging

server/cost.py — derive_cost() counts LLM input/output tokens (gemma4 +
deepseek-v4-flash), Deepgram audio minutes, Cartesia/Piper characters;
derives estimated cents from scenarios/cost_rates.yaml. No enforced ceiling
(D-012 pilot). CostBreakdown dataclass carries the breakdown dict stored in
sessions.cost_breakdown_json. Per G-005: v0.1 logged costs are pilot-config
(Ollama tier + cloud), NOT at-scale /learner economics — that requires
self-hosted gemma4:e4b + Piper (post-pilot).

server/session_recorder.py — SessionRecorder wires the SQLite store into the
pipeline lifecycle: start() creates a session row, log_turn() writes turns
with ASR/TTS text + latency + accumulates cost inputs, set_branch_path(),
end() derives cost + writes outcome + debrief + updates progress. No auth —
learner_id is the hardcoded learner-1 (D-007).

7 tests pass (derive_cost basic/piper-zero/breakdown-dict, load_rates yaml,
no-enforced-ceiling, recorder full lifecycle with DB assertions, progress
updated).

---ci---
phase: 1
milestone: v0.1
plan: 04
task: 04-03,04-04
status: execute
persona: backend-engineer,data-engineer
requirements:
  covered: [REQ-STATE-01, REQ-NFR-COST-01]
---/ci---

tests/test_cost_and_recorder.py

@@ -0,0 +1,128 @@
+"""Tests for cost logging (TASK-04-04) + session recorder (TASK-04-03)."""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+import pytest
+
+from server.cost import CostBreakdown, derive_cost, load_rates
+from server.session_recorder import SessionRecorder
+from db.store import PraxisStore, HARDCODED_LEARNER_ID
+
+
+# ─── TASK-04-04: cost logging ────────────────────────────────────────────────
+
+
+def test_derive_cost_basic():
+    """derive_cost produces a non-null cents value from usage inputs."""
+    b = derive_cost(
+        llm_input_tokens=500,
+        llm_output_tokens=200,
+        deepgram_audio_minutes=2.0,
+        tts_characters=800,
+        debrief_input_tokens=300,
+        debrief_output_tokens=150,
+        tts_provider="cartesia",
+    )
+    assert b.derived_cents > 0
+    assert b.llm_input_tokens == 500
+    assert b.tts_characters == 800
+
+
+def test_derive_cost_piper_zero_tts():
+    """Piper self-hosted TTS is $0 marginal cost (R4 mitigation, post-pilot path)."""
+    b = derive_cost(tts_characters=10000, tts_provider="piper")
+    # Piper rate is 0.0 per 1k chars → TTS contributes 0.
+    assert b.derived_cents == 0
+
+
+def test_derive_cost_breakdown_dict():
+    b = derive_cost(llm_input_tokens=1000, llm_output_tokens=500)
+    d = b.as_dict()
+    assert d["llm_input_tokens"] == 1000
+    assert d["derived_cents"] > 0
+    assert "rates" in d
+
+
+def test_load_rates_from_yaml():
+    """cost_rates.yaml is present and loadable."""
+    rates = load_rates()
+    assert "gemma4_cloud_per_1k_tokens_cents" in rates
+    assert rates["piper_per_1k_chars_cents"] == 0.0
+
+
+def test_cost_no_enforced_ceiling():
+    """D-012: v0.1 has no enforced cost ceiling (pilot). A high-cost session
+    is still logged, not rejected."""
+    b = derive_cost(
+        llm_input_tokens=1_000_000,
+        llm_output_tokens=500_000,
+        deepgram_audio_minutes=600.0,
+        tts_characters=2_000_000,
+    )
+    # No ceiling — just a (large) number.
+    assert b.derived_cents > 0
+
+
+# ─── TASK-04-03: session recorder wiring ─────────────────────────────────────
+
+
+@pytest.fixture
+def tmp_db(tmp_path: Path) -> Path:
+    return tmp_path / "test_recorder.db"
+
+
+def test_session_recorder_full_lifecycle(tmp_db: Path):
+    """TASK-04-03: start → log turns → set branch → end → DB has session + turns + progress."""
+    store = PraxisStore(tmp_db)
+
+    async def _run():
+        await store.init()
+        rec = SessionRecorder(store, scenario_id="cs_refund_ca_v01")
+        sid = await rec.start()
+        await rec.log_turn("assistant", tts_text="Hi, I want a refund.", latency_ms=None)
+        await rec.log_turn("user", asr_text="I'm sorry, I can offer a refund.", latency_ms=450.0)
+        await rec.log_turn("assistant", tts_text="Okay, what's the issue?", latency_ms=520.0)
+        rec.add_audio_minutes(1.5)
+        rec.add_debrief_tokens(input_tokens=200, output_tokens=100)
+        rec.set_branch_path(["accept_resolution"])
+        breakdown = await rec.end(outcome="success", debrief_text="You did well.")
+        sess = await store.get_session(sid)
+        turns = await store.get_turns(sid)
+        return sess, turns, breakdown
+
+    sess, turns, breakdown = asyncio.run(_run())
+    assert sess is not None
+    assert sess.outcome == "success"
+    assert sess.branch_path == ["accept_resolution"]
+    assert sess.cost_estimated_cents is not None and sess.cost_estimated_cents > 0
+    assert sess.debrief_text == "You did well."
+    assert len(turns) == 3
+    assert breakdown.derived_cents > 0
+
+
+def test_session_recorder_progress_updated(tmp_db: Path):
+    """TASK-04-03: end() updates the progress table (attempts + last_outcome)."""
+    store = PraxisStore(tmp_db)
+
+    async def _run():
+        await store.init()
+        rec = SessionRecorder(store, scenario_id="cs_refund_ca_v01")
+        await rec.start()
+        await rec.log_turn("user", asr_text="policy says no refunds")
+        rec.set_branch_path(["escalate"])
+        await rec.end(outcome="failure")
+
+        async with store._connect() as db:
+            cur = await db.execute(
+                "SELECT attempts, last_outcome FROM progress WHERE learner_id = ? AND scenario_id = ?",
+                (HARDCODED_LEARNER_ID, "cs_refund_ca_v01"),
+            )
+            return await cur.fetchone()
+
+    row = asyncio.run(_run())
+    assert row is not None
+    assert row[0] == 1
+    assert row[1] == "failure"