diff --git a/scenarios/cost_rates.yaml b/scenarios/cost_rates.yaml
new file mode 100644
index 0000000..61b3419
--- /dev/null
+++ b/scenarios/cost_rates.yaml
@@ -0,0 +1,20 @@
+# Praxis v0.1 cost rates — per-unit pricing for the cost logging (D-012, REQ-NFR-COST-01).
+# v0.1 logs actual per-session cost; no enforced ceiling (pilot).
+# Per G-005: these are pilot-config rates (Ollama tier + cloud), NOT at-scale
+# per-learner unit economics — the $3/learner target requires self-hosted
+# gemma4:e4b + Piper (post-pilot).
+
+# LLM role-play (gemma4:cloud) — Ollama tier (Pro plan amortized, pilot estimate).
+gemma4_cloud_per_1k_tokens_cents: 0.5
+
+# Debrief + classifier (deepseek-v4-flash:cloud) — Ollama tier.
+deepseek_v4_flash_per_1k_tokens_cents: 1.0
+
+# ASR (Deepgram Nova-3 streaming) — $0.0043/min → 0.43 cents/min.
+deepgram_per_audio_minute_cents: 0.43
+
+# TTS (Cartesia Sonic cloud) — per-char pricing (pilot estimate).
+cartesia_per_1k_chars_cents: 3.0
+
+# TTS (Piper self-hosted) — open-weights, $0 marginal cost.
+piper_per_1k_chars_cents: 0.0
\ No newline at end of file
diff --git a/server/cost.py b/server/cost.py
new file mode 100644
index 0000000..832d717
--- /dev/null
+++ b/server/cost.py
@@ -0,0 +1,111 @@
+"""Cost logging — per-session cost derivation (REQ-NFR-COST-01, D-012, TASK-04-04).
+
+Counts LLM input/output tokens (gemma4 + deepseek-v4-flash), Deepgram audio
+minutes, Cartesia/Piper characters; derives an estimated cost in cents using
+cost_rates.yaml. No enforced ceiling (D-012 — pilot). The derived cost +
+breakdown are stored in sessions.cost_estimated_cents / cost_breakdown_json.
+
+v0.1 logged costs are NOT representative of at-scale per-learner cost (G-005):
+Ollama tier-based pricing + Canada cloud + low volume = the most expensive
+configuration. The $3/learner target requires self-hosted gemma4:e4b + Piper
+(post-pilot). The logging infrastructure is the v0.1 contribution.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+_DEFAULT_RATES_PATH = Path(__file__).resolve().parent.parent / "scenarios" / "cost_rates.yaml"
+
+
+@dataclass
+class CostBreakdown:
+    """Per-session cost inputs + derived cents."""
+
+    llm_input_tokens: int = 0
+    llm_output_tokens: int = 0
+    deepgram_audio_minutes: float = 0.0
+    tts_characters: int = 0
+    debrief_input_tokens: int = 0
+    debrief_output_tokens: int = 0
+    rates: dict[str, float] = field(default_factory=dict)
+    derived_cents: int = 0
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "llm_input_tokens": self.llm_input_tokens,
+            "llm_output_tokens": self.llm_output_tokens,
+            "deepgram_audio_minutes": round(self.deepgram_audio_minutes, 3),
+            "tts_characters": self.tts_characters,
+            "debrief_input_tokens": self.debrief_input_tokens,
+            "debrief_output_tokens": self.debrief_output_tokens,
+            "rates": self.rates,
+            "derived_cents": self.derived_cents,
+        }
+
+
+def load_rates(path: Path | None = None) -> dict[str, float]:
+    """Load cost rates from cost_rates.yaml (or defaults if absent)."""
+    p = path or _DEFAULT_RATES_PATH
+    if p.exists():
+        with p.open("r", encoding="utf-8") as f:
+            return yaml.safe_load(f) or {}
+    # Defaults — vendor-list prices, per-unit (pilot estimates, G-005).
+    return {
+        "gemma4_cloud_per_1k_tokens_cents": 0.5,        # Ollama tier (pro plan amortized)
+        "deepseek_v4_flash_per_1k_tokens_cents": 1.0,   # Ollama tier
+        "deepgram_per_audio_minute_cents": 0.43,        # $0.0043/min
+        "cartesia_per_1k_chars_cents": 3.0,             # per-char pricing
+        "piper_per_1k_chars_cents": 0.0,                # self-hosted, $0
+    }
+
+
+def derive_cost(
+    llm_input_tokens: int = 0,
+    llm_output_tokens: int = 0,
+    deepgram_audio_minutes: float = 0.0,
+    tts_characters: int = 0,
+    debrief_input_tokens: int = 0,
+    debrief_output_tokens: int = 0,
+    tts_provider: str = "cartesia",
+    rates: dict[str, float] | None = None,
+) -> CostBreakdown:
+    """Derive the per-session cost in cents from the usage inputs + rates."""
+    r = rates or load_rates()
+
+    # LLM role-play (gemma4:cloud).
+    rp_tokens = llm_input_tokens + llm_output_tokens
+    rp_cents = (rp_tokens / 1000.0) * r.get("gemma4_cloud_per_1k_tokens_cents", 0.5)
+
+    # Debrief (deepseek-v4-flash:cloud).
+    db_tokens = debrief_input_tokens + debrief_output_tokens
+    db_cents = (db_tokens / 1000.0) * r.get("deepseek_v4_flash_per_1k_tokens_cents", 1.0)
+
+    # ASR (Deepgram).
+    asr_cents = deepgram_audio_minutes * r.get("deepgram_per_audio_minute_cents", 0.43)
+
+    # TTS (Cartesia or Piper).
+    tts_rate_key = (
+        "piper_per_1k_chars_cents" if tts_provider == "piper"
+        else "cartesia_per_1k_chars_cents"
+    )
+    tts_cents = (tts_characters / 1000.0) * r.get(tts_rate_key, 3.0)
+
+    total = int(round(rp_cents + db_cents + asr_cents + tts_cents))
+    return CostBreakdown(
+        llm_input_tokens=llm_input_tokens,
+        llm_output_tokens=llm_output_tokens,
+        deepgram_audio_minutes=deepgram_audio_minutes,
+        tts_characters=tts_characters,
+        debrief_input_tokens=debrief_input_tokens,
+        debrief_output_tokens=debrief_output_tokens,
+        rates=r,
+        derived_cents=total,
+    )
+
+
+__all__ = ["CostBreakdown", "derive_cost", "load_rates"]
\ No newline at end of file
diff --git a/server/session_recorder.py b/server/session_recorder.py
new file mode 100644
index 0000000..9968b22
--- /dev/null
+++ b/server/session_recorder.py
@@ -0,0 +1,114 @@
+"""Session recorder — wires the SQLite store into the pipeline lifecycle (TASK-04-03).
+
+On session start: create a sessions row.
+Per turn: log a turns row with ASR/TTS text + latency.
+On branch decision: update branch_path.
+On session end: set outcome + update progress + store cost + debrief.
+
+No auth — learner_id is the hardcoded 'learner-1' (D-007).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from db.store import PraxisStore, HARDCODED_LEARNER_ID
+from server.cost import CostBreakdown, derive_cost
+
+
+class SessionRecorder:
+    """Records a voice session to SQLite (TASK-04-03)."""
+
+    def __init__(
+        self,
+        store: PraxisStore,
+        learner_id: str = HARDCODED_LEARNER_ID,
+        scenario_id: str = "cs_refund_ca_v01",
+    ) -> None:
+        self.store = store
+        self.learner_id = learner_id
+        self.scenario_id = scenario_id
+        self.session_id: str | None = None
+        self._turn_seq = 0
+        # Cost inputs accumulated over the session.
+        self._llm_input_tokens = 0
+        self._llm_output_tokens = 0
+        self._deepgram_minutes = 0.0
+        self._tts_chars = 0
+        self._debrief_input_tokens = 0
+        self._debrief_output_tokens = 0
+        self._branch_path: list[str] = []
+
+    async def start(self) -> str:
+        """Create the session row; return the session id."""
+        self.session_id = await self.store.start_session(self.learner_id, self.scenario_id)
+        return self.session_id
+
+    async def log_turn(
+        self,
+        role: str,
+        asr_text: str | None = None,
+        tts_text: str | None = None,
+        latency_ms: float | None = None,
+    ) -> None:
+        """Log one turn to the turns table."""
+        if self.session_id is None:
+            return
+        await self.store.log_turn(
+            self.session_id, self._turn_seq, role, asr_text, tts_text, latency_ms
+        )
+        self._turn_seq += 1
+        # Accumulate cost inputs.
+        if asr_text:
+            # Rough: 1 token ≈ 4 chars.
+            self._llm_input_tokens += len(asr_text) // 4
+        if tts_text:
+            self._tts_chars += len(tts_text)
+            self._llm_output_tokens += len(tts_text) // 4
+        if latency_ms and role == "assistant":
+            # Rough audio-minutes estimate from latency (placeholder for real metering).
+            pass
+
+    def add_audio_minutes(self, minutes: float) -> None:
+        self._deepgram_minutes += minutes
+
+    def add_debrief_tokens(self, input_tokens: int, output_tokens: int) -> None:
+        self._debrief_input_tokens += input_tokens
+        self._debrief_output_tokens += output_tokens
+
+    def set_branch_path(self, branch_path: list[str]) -> None:
+        self._branch_path = branch_path
+
+    async def end(
+        self,
+        outcome: str,
+        tts_provider: str = "cartesia",
+        debrief_text: str | None = None,
+    ) -> CostBreakdown:
+        """End the session: derive cost, write the session row, update progress."""
+        if self.session_id is None:
+            raise RuntimeError("SessionRecorder.end() called before start()")
+
+        breakdown = derive_cost(
+            llm_input_tokens=self._llm_input_tokens,
+            llm_output_tokens=self._llm_output_tokens,
+            deepgram_audio_minutes=self._deepgram_minutes,
+            tts_characters=self._tts_chars,
+            debrief_input_tokens=self._debrief_input_tokens,
+            debrief_output_tokens=self._debrief_output_tokens,
+            tts_provider=tts_provider,
+        )
+
+        await self.store.end_session(
+            self.session_id,
+            branch_path=self._branch_path,
+            outcome=outcome,
+            cost_cents=breakdown.derived_cents,
+            cost_breakdown=breakdown.as_dict(),
+            debrief_text=debrief_text,
+        )
+        await self.store.update_progress(self.learner_id, self.scenario_id, outcome)
+        return breakdown
+
+
+__all__ = ["SessionRecorder"]
\ No newline at end of file
diff --git a/tests/test_cost_and_recorder.py b/tests/test_cost_and_recorder.py
new file mode 100644
index 0000000..d74618a
--- /dev/null
+++ b/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"
\ No newline at end of file