diff --git a/README.md b/README.md
index fff378b..3e1268b 100644
--- a/README.md
+++ b/README.md
@@ -81,10 +81,19 @@ python3 -m backtest.coach --load data/sample_journal.csv --db metrics.db
 curl -X POST localhost:8000/journal -H 'content-type: application/json' \
   -d '{"symbol":"MNQ","side":"long","entered_at_ns":1760000030000000000,
        "entry_price":21400,"size":1,"notes":"followed my plan","emotion":"calm"}'
-curl 'localhost:8000/journal?limit=5'             # entries joined to the market regime
-curl -X POST 'localhost:8000/journal/analyze'     # Claude behavioral analysis (503 without a key)
+curl 'localhost:8000/journal?limit=5'                 # entries joined to the market regime
+curl 'localhost:8000/journal?since_ns=...&until_ns=...' # bound by entry time (since incl., until excl.)
+curl 'localhost:8000/journal/1'                       # one entry by id (404 if absent)
+curl -X PATCH localhost:8000/journal/1 -H 'content-type: application/json' \
+  -d '{"emotion":"disciplined"}'                       # partial update
+curl -X DELETE 'localhost:8000/journal/1'             # 204 on success, 404 if absent
+curl -X POST 'localhost:8000/journal/analyze'         # Claude behavioral analysis (503 without a key)
 ```
 
+`side` must be `long`/`short`, `size`/`entry_price`/`entered_at_ns` must be positive, and
+`exited_at_ns` (if given) must be ≥ `entered_at_ns` — invalid bodies return 422. If Claude
+declines or returns no structured analysis, `/journal/analyze` returns 502.
+
 Run the checks the same way CI does:
 
 ```bash
@@ -92,6 +101,13 @@ cd engine && cargo fmt --check && cargo clippy --all-targets -- -D warnings && c
 cd backtest && ruff check . && ruff format --check . && pytest
 ```
 
+Verify the whole Phase 1–3 pipeline end to end (tape → engine → metrics → journal → regime
+join; no API key needed):
+
+```bash
+bash scripts/verify_pipeline.sh
+```
+
 ## Design decisions
 
 This project is built in deliberate collaboration with Claude. This section documents
@@ -147,3 +163,17 @@ calls were made by me — as the project progresses.
   design **not** trading advice and **not** price predictions. The Anthropic client is
   dependency-injected, so the analysis is unit-tested against a mock and CI needs neither
   a key nor network access.
+- **Phase 3 polish: validation, CRUD, refusal handling, time filters** (Phase 3, mine):
+  journal entries are validated at the API boundary (`side` ∈ `long`/`short`, positive
+  size/price/timestamps, `exited_at_ns` ≥ `entered_at_ns`) → 422 on bad input. The journal
+  is a full resource (`GET`/`PATCH`/`DELETE /journal/{id}`) with `since_ns`/`until_ns`
+  filters for scoping analysis to a session. Because `messages.parse().parsed_output` is
+  `Optional` (it is `None` on a safety refusal or an empty structured parse), the agent
+  raises `AnalysisUnavailable` and the endpoint maps it to a 502 rather than surfacing a
+  bare `null`. `scripts/verify_pipeline.sh` exercises the whole tape → engine → journal →
+  regime-join chain without an API key.
+- **Nanosecond timestamps deferred as a Phase 4 JSON concern** (Phase 3, mine): the API
+  returns every `*_ns` field as a JSON integer, and these epochs (~1.76e18) exceed
+  JavaScript's safe-integer range (2^53 ≈ 9e15). Rather than change the Phase 1–2 metrics
+  contract now, the Phase 4 dashboard will read them as strings/BigInt; documented in the
+  `backtest.api` module docstring so it is not forgotten.
diff --git a/backtest/src/backtest/api.py b/backtest/src/backtest/api.py
index 967a36d..e6d7469 100644
--- a/backtest/src/backtest/api.py
+++ b/backtest/src/backtest/api.py
@@ -6,15 +6,21 @@
 same database, so entries can be joined to the market regime at entry time. Point
 the app at a database with the METRICS_DB environment variable (default:
 metrics.db in the working directory).
+
+Note on timestamps: every `*_ns` field is serialized as a JSON integer, and these
+nanosecond epochs (~1.76e18) exceed JavaScript's safe-integer range (2^53 ≈ 9e15).
+A JS client (the Phase 4 dashboard) must read them as strings/BigInt to avoid
+precision loss. This is left as a Phase 4 concern to keep the Phase 1–3 contract
+unchanged.
 """
 
 import os
 import sqlite3
 from pathlib import Path
-from typing import Annotated, Any
+from typing import Annotated, Any, Literal
 
 from fastapi import Depends, FastAPI, HTTPException, Query
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, model_validator
 
 from backtest import coach, journal
 
@@ -26,17 +32,51 @@
 class JournalEntryIn(BaseModel):
     """A trade the user is logging to their journal."""
 
-    symbol: str
-    side: str
-    entered_at_ns: int
-    entry_price: float
-    size: int
-    exited_at_ns: int | None = None
-    exit_price: float | None = None
+    symbol: str = Field(min_length=1)
+    side: Literal["long", "short"]
+    entered_at_ns: int = Field(gt=0)
+    entry_price: float = Field(gt=0)
+    size: int = Field(gt=0)
+    exited_at_ns: int | None = Field(default=None, gt=0)
+    exit_price: float | None = Field(default=None, gt=0)
     pnl: float | None = None
     notes: str | None = None
     emotion: str | None = None
 
+    @model_validator(mode="after")
+    def _check_exit_after_entry(self) -> "JournalEntryIn":
+        if self.exited_at_ns is not None and self.exited_at_ns < self.entered_at_ns:
+            raise ValueError("exited_at_ns must be >= entered_at_ns")
+        return self
+
+
+class JournalEntryPatch(BaseModel):
+    """A partial update to a journal entry — only the provided fields change."""
+
+    symbol: str | None = Field(default=None, min_length=1)
+    side: Literal["long", "short"] | None = None
+    entered_at_ns: int | None = Field(default=None, gt=0)
+    entry_price: float | None = Field(default=None, gt=0)
+    size: int | None = Field(default=None, gt=0)
+    exited_at_ns: int | None = Field(default=None, gt=0)
+    exit_price: float | None = Field(default=None, gt=0)
+    pnl: float | None = None
+    notes: str | None = None
+    emotion: str | None = None
+
+    @model_validator(mode="after")
+    def _validate(self) -> "JournalEntryPatch":
+        if not self.model_fields_set:
+            raise ValueError("at least one field must be provided")
+        if (
+            {"entered_at_ns", "exited_at_ns"} <= self.model_fields_set
+            and self.entered_at_ns is not None
+            and self.exited_at_ns is not None
+            and self.exited_at_ns < self.entered_at_ns
+        ):
+            raise ValueError("exited_at_ns must be >= entered_at_ns")
+        return self
+
 
 def get_anthropic_client() -> Any:
     """Provide an Anthropic client, or 503 when no API key is configured.
@@ -113,21 +153,59 @@ def create_journal_entry(entry: JournalEntryIn) -> dict[str, Any]:
 
 
 @app.get("/journal")
-def get_journal(limit: int = Query(default=100, ge=1, le=10_000)) -> list[dict[str, Any]]:
+def get_journal(
+    limit: int = Query(default=100, ge=1, le=10_000),
+    since_ns: int | None = Query(default=None),
+    until_ns: int | None = Query(default=None),
+) -> list[dict[str, Any]]:
     """Return journal entries newest first, each joined to its market regime.
 
+    ``since_ns`` (inclusive) and ``until_ns`` (exclusive) bound the entry time.
     An empty journal (or a database that does not exist yet) returns ``[]`` —
     unlike the metrics endpoints, the journal is created by this layer, so
     "no entries yet" is a normal state rather than a 503.
     """
-    return journal.list_enriched(_db_path(), limit)
+    return journal.list_enriched(_db_path(), limit, since_ns=since_ns, until_ns=until_ns)
+
+
+@app.get("/journal/{entry_id}")
+def get_journal_entry(entry_id: int) -> dict[str, Any]:
+    """Return one journal entry (joined to its market regime), or 404."""
+    entry = journal.get_entry(_db_path(), entry_id)
+    if entry is None:
+        raise HTTPException(status_code=404, detail=f"journal entry {entry_id} not found")
+    return entry
+
+
+@app.patch("/journal/{entry_id}")
+def patch_journal_entry(entry_id: int, patch: JournalEntryPatch) -> dict[str, Any]:
+    """Apply a partial update to a journal entry, returning the updated row."""
+    updated = journal.update_entry(_db_path(), entry_id, patch.model_dump(exclude_unset=True))
+    if updated is None:
+        raise HTTPException(status_code=404, detail=f"journal entry {entry_id} not found")
+    return updated
+
+
+@app.delete("/journal/{entry_id}", status_code=204)
+def delete_journal_entry(entry_id: int) -> None:
+    """Delete a journal entry by id (204 on success, 404 if absent)."""
+    if not journal.delete_entry(_db_path(), entry_id):
+        raise HTTPException(status_code=404, detail=f"journal entry {entry_id} not found")
 
 
 @app.post("/journal/analyze")
 def analyze_journal(
     client: Annotated[Any, Depends(get_anthropic_client)],
     limit: int = Query(default=100, ge=1, le=10_000),
+    since_ns: int | None = Query(default=None),
+    until_ns: int | None = Query(default=None),
 ) -> coach.BehavioralAnalysis:
-    """Run the Claude behavioral-analysis agent over the journal."""
-    entries = journal.list_enriched(_db_path(), limit)
-    return coach.analyze(entries, client=client)
+    """Run the Claude behavioral-analysis agent over the journal.
+
+    Returns 502 when the model declines or yields no structured analysis.
+    """
+    entries = journal.list_enriched(_db_path(), limit, since_ns=since_ns, until_ns=until_ns)
+    try:
+        return coach.analyze(entries, client=client)
+    except coach.AnalysisUnavailable as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
diff --git a/backtest/src/backtest/coach.py b/backtest/src/backtest/coach.py
index 895e013..7a7df64 100644
--- a/backtest/src/backtest/coach.py
+++ b/backtest/src/backtest/coach.py
@@ -53,6 +53,15 @@
 """
 
 
+class AnalysisUnavailable(Exception):
+    """Raised when Claude returns no usable structured analysis.
+
+    Happens on a safety ``refusal`` or when structured parsing yields nothing.
+    The API maps this to a 502 and the CLI reports it, rather than surfacing a
+    bare ``None``.
+    """
+
+
 class BehavioralObservation(BaseModel):
     """One recurring behavioral pattern found in the journal."""
 
@@ -126,7 +135,13 @@ def analyze(
         messages=messages,
         output_format=BehavioralAnalysis,
     )
-    return response.parsed_output
+    analysis = response.parsed_output
+    if analysis is None:
+        stop_reason = getattr(response, "stop_reason", None)
+        raise AnalysisUnavailable(
+            f"the model returned no structured analysis (stop_reason={stop_reason!r})"
+        )
+    return analysis
 
 
 def _format_entry(entry: dict[str, Any]) -> str:
@@ -181,7 +196,11 @@ def main(argv: list[str] | None = None) -> None:
 
     import anthropic  # lazy: keeps the SDK out of the library import path
 
-    analysis = analyze(entries, client=anthropic.Anthropic())
+    try:
+        analysis = analyze(entries, client=anthropic.Anthropic())
+    except AnalysisUnavailable as exc:
+        print(f"analysis unavailable: {exc}")
+        return
     print(_render(analysis))
 
 
diff --git a/backtest/src/backtest/journal.py b/backtest/src/backtest/journal.py
index 5daca07..1b14b16 100644
--- a/backtest/src/backtest/journal.py
+++ b/backtest/src/backtest/journal.py
@@ -13,11 +13,13 @@
 import csv
 import sqlite3
 import time
+from collections.abc import Sequence
 from pathlib import Path
 from typing import Any
 
 # Fields the caller supplies for a new entry. The store manages ``id`` (an
-# autoincrement surrogate key) and ``created_at_ns`` (insertion time).
+# autoincrement surrogate key) and ``created_at_ns`` (insertion time). These are
+# also the only columns an update is allowed to touch.
 _INPUT_COLUMNS = (
     "symbol",
     "side",
@@ -30,12 +32,13 @@
     "notes",
     "emotion",
 )
+_UPDATABLE_COLUMNS = frozenset(_INPUT_COLUMNS)
 
 # Full column set returned when reading a row back.
 _ROW_COLUMNS = ("id", *_INPUT_COLUMNS, "created_at_ns")
 
-# Regime columns attached to each entry by ``list_enriched``. Present (possibly
-# ``None``) on every enriched row so downstream consumers see a stable shape.
+# Regime columns attached to each entry by ``list_enriched`` / ``get_entry``.
+# Present (possibly ``None``) on every enriched row so consumers see a stable shape.
 _REGIME_COLUMNS = (
     "regime_bucket_start_ns",
     "regime_window_ns",
@@ -47,11 +50,13 @@
 _INT_FIELDS = frozenset({"entered_at_ns", "exited_at_ns", "size"})
 _FLOAT_FIELDS = frozenset({"entry_price", "exit_price", "pnl"})
 
-# For each journal entry, pick the metric bucket whose window contains the entry
-# time. When several window sizes coexist, prefer the finest (smallest window),
-# which describes the regime most tightly. A LEFT JOIN leaves the regime columns
-# ``NULL`` when no bucket matches (or the metrics table is empty).
-_ENRICHED_QUERY = f"""
+# SELECT..FROM for a plain read and for a regime-enriched read. The enriched read
+# picks, per entry, the metric bucket whose window contains the entry time; when
+# several window sizes coexist it prefers the finest (smallest) window. A LEFT
+# JOIN leaves the regime columns NULL when no bucket matches (or metrics is empty).
+_PLAIN_SELECT = f"SELECT {', '.join(_ROW_COLUMNS)} FROM journal_entries"
+
+_ENRICHED_SELECT = f"""
 SELECT
     {", ".join("j." + c for c in _ROW_COLUMNS)},
     r.bucket_start_ns     AS regime_bucket_start_ns,
@@ -67,14 +72,8 @@
     ORDER BY m.window_ns ASC, m.bucket_start_ns DESC
     LIMIT 1
 )
-ORDER BY j.entered_at_ns DESC
-LIMIT ?
 """
 
-_PLAIN_QUERY = (
-    f"SELECT {', '.join(_ROW_COLUMNS)} FROM journal_entries ORDER BY entered_at_ns DESC LIMIT ?"
-)
-
 
 def init_db(conn: sqlite3.Connection) -> None:
     """Create the ``journal_entries`` table if it does not already exist."""
@@ -122,18 +121,92 @@ def insert_entry(path: str | Path, entry: dict[str, Any]) -> dict[str, Any]:
     return dict(stored)
 
 
-def list_entries(path: str | Path, limit: int) -> list[dict[str, Any]]:
-    """Return journal entries newest first (by entry time). Empty if none yet."""
-    return _read(path, _PLAIN_QUERY, limit)
+def get_entry(path: str | Path, entry_id: int) -> dict[str, Any] | None:
+    """Return one journal entry (regime-enriched) by id, or ``None`` if absent."""
+    path = Path(path)
+    if not path.exists():
+        return None
+    with sqlite3.connect(path) as conn:
+        conn.row_factory = sqlite3.Row
+        try:
+            if _has_metrics_table(conn):
+                query = f"{_ENRICHED_SELECT} WHERE j.id = ?"
+            else:
+                query = f"{_PLAIN_SELECT} WHERE id = ?"
+            row = conn.execute(query, (entry_id,)).fetchone()
+        except sqlite3.OperationalError:
+            return None
+    return _with_regime(row) if row is not None else None
+
 
+def update_entry(path: str | Path, entry_id: int, fields: dict[str, Any]) -> dict[str, Any] | None:
+    """Apply a partial update to an entry, returning the updated (enriched) row.
 
-def list_enriched(path: str | Path, limit: int) -> list[dict[str, Any]]:
+    Only recognised, mutable columns in ``fields`` are written; ``id`` and
+    ``created_at_ns`` are never touched. Returns ``None`` if the entry does not
+    exist. An empty update is a no-op that returns the current row.
+    """
+    updates = {col: value for col, value in fields.items() if col in _UPDATABLE_COLUMNS}
+    if not updates:
+        return get_entry(path, entry_id)
+    path = Path(path)
+    if not path.exists():
+        return None
+    assignments = ", ".join(f"{col} = ?" for col in updates)
+    with sqlite3.connect(path) as conn:
+        try:
+            cursor = conn.execute(
+                f"UPDATE journal_entries SET {assignments} WHERE id = ?",
+                (*updates.values(), entry_id),
+            )
+        except sqlite3.OperationalError:
+            return None
+        if cursor.rowcount == 0:
+            return None
+    return get_entry(path, entry_id)
+
+
+def delete_entry(path: str | Path, entry_id: int) -> bool:
+    """Delete an entry by id, returning ``True`` if a row was removed."""
+    path = Path(path)
+    if not path.exists():
+        return False
+    with sqlite3.connect(path) as conn:
+        try:
+            cursor = conn.execute("DELETE FROM journal_entries WHERE id = ?", (entry_id,))
+        except sqlite3.OperationalError:
+            return False
+    return cursor.rowcount > 0
+
+
+def list_entries(
+    path: str | Path,
+    limit: int,
+    *,
+    since_ns: int | None = None,
+    until_ns: int | None = None,
+) -> list[dict[str, Any]]:
+    """Return journal entries newest first, optionally bounded by entry time.
+
+    ``since_ns`` is inclusive, ``until_ns`` exclusive. Empty if none / no DB.
+    """
+    query, params = _build_list_query(_PLAIN_SELECT, "entered_at_ns", since_ns, until_ns)
+    return _read(path, query, (*params, limit))
+
+
+def list_enriched(
+    path: str | Path,
+    limit: int,
+    *,
+    since_ns: int | None = None,
+    until_ns: int | None = None,
+) -> list[dict[str, Any]]:
     """Return journal entries newest first, each joined to its market regime.
 
     Every row carries the ``regime_*`` columns; they are ``None`` when the
-    metrics table is missing/empty or no bucket covers the entry time. This is
-    the join that lets the agent correlate behavior with the microstructure
-    context at the time of each trade.
+    metrics table is missing/empty or no bucket covers the entry time. ``since_ns``
+    is inclusive and ``until_ns`` exclusive. This is the join that lets the agent
+    correlate behavior with the microstructure context at the time of each trade.
     """
     path = Path(path)
     if not path.exists():
@@ -141,17 +214,18 @@ def list_enriched(path: str | Path, limit: int) -> list[dict[str, Any]]:
     with sqlite3.connect(path) as conn:
         conn.row_factory = sqlite3.Row
         try:
-            query = _ENRICHED_QUERY if _has_metrics_table(conn) else _PLAIN_QUERY
-            rows = conn.execute(query, (limit,)).fetchall()
+            if _has_metrics_table(conn):
+                query, params = _build_list_query(
+                    _ENRICHED_SELECT, "j.entered_at_ns", since_ns, until_ns
+                )
+            else:
+                query, params = _build_list_query(
+                    _PLAIN_SELECT, "entered_at_ns", since_ns, until_ns
+                )
+            rows = conn.execute(query, (*params, limit)).fetchall()
         except sqlite3.OperationalError:
             return []
-    enriched: list[dict[str, Any]] = []
-    for row in rows:
-        record = dict(row)
-        for column in _REGIME_COLUMNS:
-            record.setdefault(column, None)
-        enriched.append(record)
-    return enriched
+    return [_with_regime(row) for row in rows]
 
 
 def import_csv(csv_path: str | Path, path: str | Path) -> int:
@@ -176,19 +250,41 @@ def import_csv(csv_path: str | Path, path: str | Path) -> int:
     return len(rows)
 
 
-def _read(path: str | Path, query: str, limit: int) -> list[dict[str, Any]]:
+def _build_list_query(
+    select: str, time_column: str, since_ns: int | None, until_ns: int | None
+) -> tuple[str, list[Any]]:
+    clauses: list[str] = []
+    params: list[Any] = []
+    if since_ns is not None:
+        clauses.append(f"{time_column} >= ?")
+        params.append(since_ns)
+    if until_ns is not None:
+        clauses.append(f"{time_column} < ?")
+        params.append(until_ns)
+    where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
+    return f"{select}{where} ORDER BY {time_column} DESC LIMIT ?", params
+
+
+def _read(path: str | Path, query: str, params: Sequence[Any]) -> list[dict[str, Any]]:
     path = Path(path)
     if not path.exists():
         return []
     with sqlite3.connect(path) as conn:
         conn.row_factory = sqlite3.Row
         try:
-            rows = conn.execute(query, (limit,)).fetchall()
+            rows = conn.execute(query, params).fetchall()
         except sqlite3.OperationalError:
             return []
     return [dict(row) for row in rows]
 
 
+def _with_regime(row: sqlite3.Row) -> dict[str, Any]:
+    record = dict(row)
+    for column in _REGIME_COLUMNS:
+        record.setdefault(column, None)
+    return record
+
+
 def _has_metrics_table(conn: sqlite3.Connection) -> bool:
     row = conn.execute(
         "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'metrics'"
diff --git a/backtest/tests/test_api.py b/backtest/tests/test_api.py
index 6013e8d..34a52e1 100644
--- a/backtest/tests/test_api.py
+++ b/backtest/tests/test_api.py
@@ -170,3 +170,87 @@ def test_analyze_without_api_key_returns_503(client, monkeypatch):
     monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
     api.app.dependency_overrides.clear()
     assert client.post("/journal/analyze").status_code == 503
+
+
+@pytest.mark.parametrize(
+    "bad",
+    [
+        {"side": "sideways"},
+        {"size": 0},
+        {"symbol": ""},
+        {"entry_price": 0},
+        {"entered_at_ns": 0},
+        {"entered_at_ns": 2_000, "exited_at_ns": 1_000},  # exit before entry
+    ],
+)
+def test_create_journal_entry_validation(client, bad):
+    body = {
+        "symbol": "MNQ",
+        "side": "long",
+        "entered_at_ns": 1_500_000_000,
+        "entry_price": 21_400.0,
+        "size": 1,
+    }
+    body.update(bad)
+    assert client.post("/journal", json=body).status_code == 422
+
+
+def test_get_journal_entry_by_id(client):
+    created = _post_entry(client, entered_at_ns=1_500_000_000).json()
+    got = client.get(f"/journal/{created['id']}").json()
+    assert got["id"] == created["id"]
+    assert got["regime_ofi"] == 0.5
+
+
+def test_get_journal_entry_404(client):
+    assert client.get("/journal/999").status_code == 404
+
+
+def test_patch_journal_entry(client):
+    created = _post_entry(client).json()
+    response = client.patch(f"/journal/{created['id']}", json={"size": 4, "emotion": "calm"})
+    assert response.status_code == 200
+    body = response.json()
+    assert body["size"] == 4
+    assert body["emotion"] == "calm"
+
+
+def test_patch_journal_entry_404(client):
+    assert client.patch("/journal/999", json={"size": 2}).status_code == 404
+
+
+def test_patch_journal_entry_empty_returns_422(client):
+    created = _post_entry(client).json()
+    assert client.patch(f"/journal/{created['id']}", json={}).status_code == 422
+
+
+def test_patch_journal_entry_invalid_returns_422(client):
+    created = _post_entry(client).json()
+    assert client.patch(f"/journal/{created['id']}", json={"side": "up"}).status_code == 422
+
+
+def test_delete_journal_entry(client):
+    created = _post_entry(client).json()
+    assert client.delete(f"/journal/{created['id']}").status_code == 204
+    assert client.get(f"/journal/{created['id']}").status_code == 404
+
+
+def test_delete_journal_entry_404(client):
+    assert client.delete("/journal/999").status_code == 404
+
+
+def test_get_journal_time_range(client):
+    _post_entry(client, entered_at_ns=1_500_000_000)
+    _post_entry(client, entered_at_ns=2_500_000_000)
+    rows = client.get("/journal", params={"since_ns": 2_000_000_000}).json()
+    assert [r["entered_at_ns"] for r in rows] == [2_500_000_000]
+
+
+def test_analyze_journal_502_when_no_structured_output(client):
+    _post_entry(client)
+    api.app.dependency_overrides[api.get_anthropic_client] = lambda: _FakeClient(None)
+    try:
+        response = client.post("/journal/analyze")
+    finally:
+        api.app.dependency_overrides.clear()
+    assert response.status_code == 502
diff --git a/backtest/tests/test_coach.py b/backtest/tests/test_coach.py
index c163f2c..fd5b1ec 100644
--- a/backtest/tests/test_coach.py
+++ b/backtest/tests/test_coach.py
@@ -1,25 +1,29 @@
+import pytest
+
 from backtest import coach
 from backtest.coach import BehavioralAnalysis, BehavioralObservation
 
 
 class _FakeMessages:
-    def __init__(self, parsed_output):
+    def __init__(self, parsed_output, stop_reason="end_turn"):
         self._parsed_output = parsed_output
+        self._stop_reason = stop_reason
         self.calls = []
 
     def parse(self, **kwargs):
         self.calls.append(kwargs)
-        return _FakeResponse(self._parsed_output)
+        return _FakeResponse(self._parsed_output, self._stop_reason)
 
 
 class _FakeResponse:
-    def __init__(self, parsed_output):
+    def __init__(self, parsed_output, stop_reason="end_turn"):
         self.parsed_output = parsed_output
+        self.stop_reason = stop_reason
 
 
 class FakeClient:
-    def __init__(self, parsed_output):
-        self.messages = _FakeMessages(parsed_output)
+    def __init__(self, parsed_output, stop_reason="end_turn"):
+        self.messages = _FakeMessages(parsed_output, stop_reason)
 
 
 def _canned() -> BehavioralAnalysis:
@@ -83,6 +87,12 @@ def test_analyze_empty_journal_short_circuits():
     assert client.messages.calls == []  # no API call for an empty journal
 
 
+def test_analyze_raises_when_no_structured_output():
+    client = FakeClient(None, stop_reason="refusal")
+    with pytest.raises(coach.AnalysisUnavailable, match="refusal"):
+        coach.analyze([_enriched()], client=client)
+
+
 def test_build_messages_includes_regime_and_notes():
     system, messages = coach.build_messages(
         [_enriched(notes="chased the move", emotion="fomo", regime_ofi=-0.5)]
diff --git a/backtest/tests/test_integration.py b/backtest/tests/test_integration.py
new file mode 100644
index 0000000..bdb53fa
--- /dev/null
+++ b/backtest/tests/test_integration.py
@@ -0,0 +1,99 @@
+"""Hermetic end-to-end test of the journal HTTP flow over a seeded metrics table.
+
+Runs in CI without cargo or an API key: it stands in a `metrics` table for the
+Rust engine's output, drives the real FastAPI app, and mocks the Anthropic
+client. The Rust-in-the-loop smoke test lives in scripts/verify_pipeline.sh.
+"""
+
+import sqlite3
+
+import pytest
+from fastapi.testclient import TestClient
+
+from backtest import api, coach
+
+METRICS_COLUMNS = (
+    "bucket_start_ns, window_ns, buy_volume, sell_volume, total_volume,"
+    " trade_count, ofi, vwap, realized_volatility"
+)
+# Two 1s buckets: [1e9, 2e9) with ofi 0.5, [2e9, 3e9) with ofi -0.5.
+METRIC_ROWS = [
+    (1_000_000_000, 1_000_000_000, 30, 10, 40, 5, 0.5, 21_400.0, 0.001),
+    (2_000_000_000, 1_000_000_000, 5, 15, 20, 4, -0.5, 21_401.0, 0.002),
+]
+
+
+class _FakeResponse:
+    def __init__(self, parsed_output):
+        self.parsed_output = parsed_output
+        self.stop_reason = "end_turn"
+
+
+class _FakeMessages:
+    def __init__(self, parsed_output):
+        self._parsed_output = parsed_output
+
+    def parse(self, **kwargs):
+        return _FakeResponse(self._parsed_output)
+
+
+class _FakeClient:
+    def __init__(self, parsed_output):
+        self.messages = _FakeMessages(parsed_output)
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    db = tmp_path / "metrics.db"
+    with sqlite3.connect(db) as conn:
+        conn.execute(f"CREATE TABLE metrics ({METRICS_COLUMNS})")
+        conn.executemany(f"INSERT INTO metrics VALUES ({', '.join(['?'] * 9)})", METRIC_ROWS)
+    monkeypatch.setenv(api.DB_PATH_ENV, str(db))
+    return TestClient(app=api.app)
+
+
+def _entry(entered_at_ns, side="long"):
+    return {
+        "symbol": "MNQ",
+        "side": side,
+        "entered_at_ns": entered_at_ns,
+        "entry_price": 21_400.0,
+        "size": 1,
+    }
+
+
+def test_full_journal_flow_through_http(client):
+    # Two trades entered in different regime windows.
+    first = client.post("/journal", json=_entry(1_500_000_000, "long")).json()
+    second = client.post("/journal", json=_entry(2_500_000_000, "short")).json()
+
+    listed = client.get("/journal").json()
+    by_id = {row["id"]: row for row in listed}
+    assert by_id[first["id"]]["regime_ofi"] == 0.5
+    assert by_id[second["id"]]["regime_ofi"] == -0.5
+    assert [row["id"] for row in listed] == [second["id"], first["id"]]  # newest first
+
+    # Analyze the whole journal via HTTP with a mocked model.
+    analysis = coach.BehavioralAnalysis(
+        summary="two trades across opposite regimes",
+        observations=[],
+        disclaimer=coach.DISCLAIMER,
+    )
+    api.app.dependency_overrides[api.get_anthropic_client] = lambda: _FakeClient(analysis)
+    try:
+        response = client.post("/journal/analyze")
+    finally:
+        api.app.dependency_overrides.clear()
+    assert response.status_code == 200
+    assert response.json()["summary"] == "two trades across opposite regimes"
+
+
+def test_time_range_filter_through_http(client):
+    client.post("/journal", json=_entry(1_500_000_000))
+    client.post("/journal", json=_entry(2_500_000_000))
+
+    only_first = client.get("/journal", params={"until_ns": 2_000_000_000}).json()
+    assert [row["entered_at_ns"] for row in only_first] == [1_500_000_000]
+
+    only_second = client.get("/journal", params={"since_ns": 2_000_000_000}).json()
+    assert [row["entered_at_ns"] for row in only_second] == [2_500_000_000]
diff --git a/backtest/tests/test_journal.py b/backtest/tests/test_journal.py
index 8276a94..9af553a 100644
--- a/backtest/tests/test_journal.py
+++ b/backtest/tests/test_journal.py
@@ -128,3 +128,92 @@ def test_import_csv(tmp_path):
     assert newest["notes"] is None
     assert oldest["entry_price"] == 21_400.0
     assert oldest["pnl"] == 10.0
+
+
+# --- CRUD by id ----------------------------------------------------------------
+
+
+def test_get_entry_enriched(db):
+    stored = journal.insert_entry(db, _entry(1_500))
+    _seed_metrics(db, [(1000, 1000, 30, 10, 40, 5, 0.5, 21_400.0, 0.001)])
+    got = journal.get_entry(db, stored["id"])
+    assert got["id"] == stored["id"]
+    assert got["regime_ofi"] == 0.5
+
+
+def test_get_entry_missing_returns_none(db):
+    journal.insert_entry(db, _entry(1000))
+    assert journal.get_entry(db, 999) is None
+
+
+def test_get_entry_missing_db_returns_none(tmp_path):
+    assert journal.get_entry(tmp_path / "nope.db", 1) is None
+
+
+def test_update_entry_partial(db):
+    stored = journal.insert_entry(db, _entry(1000, side="long"))
+    updated = journal.update_entry(db, stored["id"], {"size": 5, "emotion": "angry"})
+    assert updated["size"] == 5
+    assert updated["emotion"] == "angry"
+    assert updated["side"] == "long"  # untouched
+    assert journal.get_entry(db, stored["id"])["size"] == 5  # persisted
+
+
+def test_update_entry_ignores_protected_columns(db):
+    stored = journal.insert_entry(db, _entry(1000))
+    updated = journal.update_entry(db, stored["id"], {"id": 999, "created_at_ns": 0, "notes": "x"})
+    assert updated["id"] == stored["id"]
+    assert updated["created_at_ns"] == stored["created_at_ns"]
+    assert updated["notes"] == "x"
+
+
+def test_update_entry_missing_returns_none(db):
+    journal.insert_entry(db, _entry(1000))
+    assert journal.update_entry(db, 999, {"size": 3}) is None
+
+
+def test_update_entry_empty_is_noop(db):
+    stored = journal.insert_entry(db, _entry(1000))
+    same = journal.update_entry(db, stored["id"], {})
+    assert same["id"] == stored["id"]
+    assert same["size"] == stored["size"]
+
+
+def test_delete_entry(db):
+    stored = journal.insert_entry(db, _entry(1000))
+    assert journal.delete_entry(db, stored["id"]) is True
+    assert journal.get_entry(db, stored["id"]) is None
+    assert journal.delete_entry(db, stored["id"]) is False
+
+
+def test_delete_entry_missing_db(tmp_path):
+    assert journal.delete_entry(tmp_path / "nope.db", 1) is False
+
+
+# --- Time-range filters --------------------------------------------------------
+
+
+def test_list_entries_time_range(db):
+    for entered_at_ns in (1000, 2000, 3000):
+        journal.insert_entry(db, _entry(entered_at_ns))
+    since = journal.list_entries(db, 10, since_ns=2000)
+    assert [e["entered_at_ns"] for e in since] == [3000, 2000]  # inclusive
+    until = journal.list_entries(db, 10, until_ns=3000)
+    assert [e["entered_at_ns"] for e in until] == [2000, 1000]  # exclusive
+    both = journal.list_entries(db, 10, since_ns=2000, until_ns=3000)
+    assert [e["entered_at_ns"] for e in both] == [2000]
+
+
+def test_list_enriched_time_range(db):
+    journal.insert_entry(db, _entry(1_500))
+    journal.insert_entry(db, _entry(2_500))
+    _seed_metrics(
+        db,
+        [
+            (1000, 1000, 1, 1, 2, 2, 0.5, 1.0, 0.0),
+            (2000, 1000, 1, 1, 2, 2, -0.5, 2.0, 0.0),
+        ],
+    )
+    rows = journal.list_enriched(db, 10, since_ns=2000)
+    assert [r["entered_at_ns"] for r in rows] == [2_500]
+    assert rows[0]["regime_ofi"] == -0.5
diff --git a/scripts/verify_pipeline.sh b/scripts/verify_pipeline.sh
new file mode 100755
index 0000000..5c28881
--- /dev/null
+++ b/scripts/verify_pipeline.sh
@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+# End-to-end smoke test of the Phase 1-3 pipeline (no Claude API key required):
+#
+#   synthetic tape  ->  Rust engine  ->  metrics.db  ->  journal seed  ->  regime join
+#
+# It proves the three layers work together: the engine writes metrics, the
+# journal is stored in the same DB, and each trade joins to the microstructure
+# regime at its entry time. The behavioral agent (POST /journal/analyze) needs
+# ANTHROPIC_API_KEY and is covered by the mocked unit tests instead.
+#
+# Usage: bash scripts/verify_pipeline.sh   (override the interpreter with PYTHON=...)
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT"
+
+DB="$(mktemp -t tme_metrics.XXXXXX.db)"
+trap 'rm -f "$DB"' EXIT
+
+PY="${PYTHON:-python3}"
+# Make `backtest` importable without an install; journal.py is stdlib-only.
+export PYTHONPATH="$ROOT/backtest/src${PYTHONPATH:+:$PYTHONPATH}"
+
+echo "1/4 generating synthetic tick tape"
+"$PY" data/generate_ticks.py >/dev/null
+
+echo "2/4 running the Rust engine into a temp metrics DB"
+cargo run --quiet --manifest-path engine/Cargo.toml --release -- \
+  --input data/sample_mnq_ticks.csv --db "$DB" --window 1s
+
+echo "3/4 generating and seeding the synthetic journal"
+"$PY" data/generate_journal.py >/dev/null
+"$PY" - "$DB" <<'PYCODE'
+import sys
+
+from backtest import journal
+
+db = sys.argv[1]
+seeded = journal.import_csv("data/sample_journal.csv", db)
+entries = journal.list_enriched(db, limit=1000)
+joined = [e for e in entries if e["regime_ofi"] is not None]
+print(f"    seeded {seeded} entries; {len(joined)}/{len(entries)} joined to a regime")
+if not entries:
+    sys.exit("FAIL: no journal entries after seeding")
+if not joined:
+    sys.exit("FAIL: no journal entry joined to a metric bucket")
+PYCODE
+
+echo "4/4 OK — Phase 1-3 pipeline verified end to end"