From f07bc48a921bfbede19daf1b54df001a6508f332 Mon Sep 17 00:00:00 2001
From: hudsonaikins-crown <hudson.aikins@blanoire.co>
Date: Thu, 10 Sep 2026 19:25:25 -0400
Subject: [PATCH 1/7] NRCL-104 Add read-only paper journal queries

---
 docs/trading/durable-paper.mdx |  43 +++++++-
 neural/paper_query.py          | 179 +++++++++++++++++++++++++++++++++
 tests/test_paper_query.py      | 156 ++++++++++++++++++++++++++++
 3 files changed, 377 insertions(+), 1 deletion(-)
 create mode 100644 neural/paper_query.py
 create mode 100644 tests/test_paper_query.py

diff --git a/docs/trading/durable-paper.mdx b/docs/trading/durable-paper.mdx
index 058f85f5..8db76554 100644
--- a/docs/trading/durable-paper.mdx
+++ b/docs/trading/durable-paper.mdx
@@ -62,8 +62,49 @@ available through `inspect`. Queued model-1 jobs fail explicitly as unsupported;
 submit the original strategy, recording and assumptions again to run model 2.
 Old results are not silently recalculated or relabeled with the new model.
 
+## Read-only experiment queries
+
+The experimental `neural.paper_query.PaperJournal` API lets consumers use saved
+experiments without reading Neural's tables or starting a worker:
+
+```python
+from neural.paper_query import PaperJournal
+
+journal = PaperJournal("/private/path/paper.sqlite3")
+page = journal.history(offset=0, limit=25)  # jobs, has_more; no result traces
+if page["jobs"]:
+    job_id = page["jobs"][0]["id"]
+    saved = journal.inspect(job_id)  # id, status, error, config, result
+    original_bytes = journal.recording(job_id)
+```
+
+Queries open an existing journal in SQLite read-only mode, without acquiring a
+writer reservation, creating a queue, migrating its schema or running simulation.
+Missing files raise `FileNotFoundError`; unknown jobs, invalid stored data and
+unsupported journal versions raise `ValueError`. History orders by insertion,
+newest first; offset is 0–100,000 and page size is 1–100. No timestamps are invented.
+Reading can still report database-busy if SQLite holds an exclusive lock.
+
+`recording(job_id)` verifies the stored configuration/job identity and the original
+recording's size and digest before returning its exact bytes. Inspection preserves
+historical results as recorded; it does not recalculate them or certify their
+correctness. The caller selects the runtime archive and verifies its wheel and
+environment identity. This API does not authenticate users or grant data rights.
+
+`compare(first_id, second_id)` returns two completed jobs with derived
+`total_fees`, `fill_count` and `rejection_count` in their result objects. Recording,
+model, market/outcome, cash, fees and simulation limits must match; strategy
+thresholds and risk limits may differ. Summaries use exact decimals and never
+write back to the journal. Missing historical traces reject comparison explicitly.
+
+`compare_markets(first_id, second_id)` separately returns `ids`, supplied `markets`
+and a sports `comparison`. It can inspect cross-venue contract metadata, but never
+compares performance or execution costs. Missing sports metadata rejects the
+request. Keep the existing `PaperJobs` write/worker API for submission and processing;
+its methods and the journal format are unchanged.
+
 ```sh
-pytest tests/test_paper_worker.py -q
+pytest tests/test_paper_worker.py tests/test_paper_query.py -q
 ```
 
 Tests exercise CLI submission/inspection, concurrent processes and process death
diff --git a/neural/paper_query.py b/neural/paper_query.py
new file mode 100644
index 00000000..b049269a
--- /dev/null
+++ b/neural/paper_query.py
@@ -0,0 +1,179 @@
+"""Read-only views of Neural's local paper journal; never run or migrate jobs."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sqlite3
+from collections.abc import Iterator
+from contextlib import contextmanager
+from decimal import Decimal, localcontext
+from pathlib import Path
+from typing import Any
+
+from neural.paper_worker import APPLICATION_ID, MAX_INPUT_BYTES
+from neural.sports import SportsMarket, compare_sports_markets
+from neural.strategy import StrategySpec, _decimal
+
+
+class PaperJournal:
+    """Inspect an existing v1 journal without a writer lock or worker startup.
+
+    Missing files raise FileNotFoundError; malformed/unsupported journals and
+    inputs raise ValueError. Historical model results remain opaque on reads.
+    The caller owns the journal path and selects the verified runtime archive.
+    """
+
+    def __init__(self, database: str | Path):
+        if str(database) in ("", ":memory:"):
+            raise ValueError("paper journal requires an existing database file")
+        self.database = Path(database).resolve()
+
+    @contextmanager
+    def _connection(self) -> Iterator[sqlite3.Connection]:
+        if not self.database.exists():
+            raise FileNotFoundError("paper job database not found")
+        db = None
+        try:
+            db = sqlite3.connect(self.database.as_uri() + "?mode=ro", uri=True, timeout=5)
+            db.row_factory = sqlite3.Row
+            db.execute("PRAGMA query_only=ON")
+            db.execute("BEGIN")
+            if (
+                db.execute("PRAGMA application_id").fetchone()[0] != APPLICATION_ID
+                or db.execute("PRAGMA user_version").fetchone()[0] != 1
+            ):
+                raise ValueError("unsupported paper job database")
+            yield db
+        except sqlite3.Error as exc:
+            raise ValueError("cannot read paper job database") from exc
+        finally:
+            if db is not None:
+                db.close()
+
+    @staticmethod
+    def _view(row: sqlite3.Row) -> dict[str, Any]:
+        try:
+            if not isinstance(row["config"], str):
+                raise ValueError("stored job configuration is invalid")
+            config = json.loads(row["config"])
+            if (
+                not isinstance(config, dict)
+                or hashlib.sha256(row["config"].encode()).hexdigest() != row["id"]
+                or row["status"] not in ("queued", "completed", "failed")
+            ):
+                raise ValueError("stored job input integrity mismatch")
+            view = {key: row[key] for key in ("id", "status", "error")}
+            view["config"] = config
+            if "result" in row.keys():
+                result = json.loads(row["result"]) if row["result"] is not None else None
+                if (result is not None and not isinstance(result, dict)) or (
+                    (row["status"] == "completed") != (result is not None)
+                ):
+                    raise ValueError("stored job result is invalid")
+                view["result"] = result
+            return view
+        except (TypeError, json.JSONDecodeError) as exc:
+            raise ValueError("stored job data is invalid") from exc
+
+    @classmethod
+    def _job(cls, db: sqlite3.Connection, identity: str) -> dict[str, Any]:
+        if not isinstance(identity, str):
+            raise ValueError("job identity must be a string")
+        row = db.execute(
+            "SELECT id,status,error,config,result FROM jobs WHERE id=?", (identity,)
+        ).fetchone()
+        if row is None:
+            raise ValueError("unknown paper job")
+        return cls._view(row)
+
+    def history(self, *, offset: int = 0, limit: int = 25) -> dict[str, Any]:
+        """Newest inserted jobs first, without result traces or invented timestamps."""
+        if type(offset) is not int or not 0 <= offset <= 100000:
+            raise ValueError("offset must be an integer from 0 to 100000")
+        if type(limit) is not int or not 1 <= limit <= 100:
+            raise ValueError("limit must be an integer from 1 to 100")
+        with self._connection() as db:
+            rows = db.execute(
+                "SELECT id,status,error,config FROM jobs ORDER BY rowid DESC LIMIT ? OFFSET ?",
+                (limit + 1, offset),
+            ).fetchall()
+            return {
+                "jobs": [self._view(row) for row in rows[:limit]],
+                "has_more": len(rows) > limit,
+            }
+
+    def inspect(self, identity: str) -> dict[str, Any]:
+        """Return saved configuration and result; do not execute or reinterpret it."""
+        with self._connection() as db:
+            return self._job(db, identity)
+
+    def recording(self, identity: str) -> bytes:
+        """Return original input bytes after checking size and the saved digest."""
+        with self._connection() as db:
+            job = self._job(db, identity)
+            raw = db.execute(
+                "SELECT CASE WHEN length(recording) BETWEEN 1 AND ? THEN recording END "
+                "FROM jobs WHERE id=?",
+                (MAX_INPUT_BYTES, identity),
+            ).fetchone()[0]
+            if not isinstance(raw, bytes) or hashlib.sha256(raw).hexdigest() != job["config"].get(
+                "recording_sha256"
+            ):
+                raise ValueError("stored job recording integrity mismatch")
+            return raw
+
+    def _pair(self, first: str, second: str) -> list[dict[str, Any]]:
+        if first == second:
+            raise ValueError("choose two distinct completed experiments")
+        with self._connection() as db:
+            jobs = [self._job(db, identity) for identity in (first, second)]
+        if any(job["status"] != "completed" for job in jobs):
+            raise ValueError("choose two distinct completed experiments")
+        return jobs
+
+    @staticmethod
+    def _assumptions(job: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
+        config = dict(job["config"])
+        strategy = config.pop("strategy", None)
+        StrategySpec.from_dict(strategy)
+        return config, {
+            key: strategy[key]
+            for key in ("schema_version", "kind", "venue", "market_id", "outcome")
+        }
+
+    def compare(self, first: str, second: str) -> list[dict[str, Any]]:
+        """Compare strategy variants only when their recorded inputs/assumptions match."""
+        jobs = self._pair(first, second)
+        if self._assumptions(jobs[0]) != self._assumptions(jobs[1]):
+            raise ValueError(
+                "comparison requires the same recording, model, market, outcome, "
+                "cash, fees and simulation limits"
+            )
+        for job in jobs:
+            trace = job["result"].get("trace")
+            if not isinstance(trace, list) or any(
+                not isinstance(event, dict) or not isinstance(event.get("action"), str)
+                for event in trace
+            ):
+                raise ValueError("saved comparison trace unavailable or invalid")
+            with localcontext() as context:
+                context.prec = 100
+                total_fees = sum(
+                    (_decimal(event.get("fees", "0"), "fees") for event in trace), Decimal(0)
+                )
+            job["result"]["total_fees"] = str(total_fees)
+            job["result"]["fill_count"] = sum(event["action"] == "fill" for event in trace)
+            job["result"]["rejection_count"] = sum(event["action"] == "reject" for event in trace)
+        return jobs
+
+    def compare_markets(self, first: str, second: str) -> dict[str, Any]:
+        """Inspect sports terms separately; this never compares execution costs/PnL."""
+        jobs = self._pair(first, second)
+        markets = [job["result"].get("sports_market") for job in jobs]
+        if any(market is None for market in markets):
+            raise ValueError(
+                "market terms unavailable: both recordings need sports contract metadata"
+            )
+        comparison = compare_sports_markets(*(SportsMarket.from_dict(market) for market in markets))
+        return {"comparison": comparison.to_dict(), "ids": [first, second], "markets": markets}
diff --git a/tests/test_paper_query.py b/tests/test_paper_query.py
new file mode 100644
index 00000000..4f64c524
--- /dev/null
+++ b/tests/test_paper_query.py
@@ -0,0 +1,156 @@
+"""Read-only consumer behavior over current, historical and damaged journals."""
+
+import json
+import sqlite3
+from dataclasses import replace
+from decimal import localcontext
+
+import pytest
+
+import neural.paper_worker as worker
+from neural.paper_query import PaperJournal
+from tests.test_recordings import inputs
+
+
+def enqueue(jobs, *, venue="polymarket-us", entry="0.45", **assumptions):
+    spec, path = inputs(venue)
+    identity = jobs.submit(
+        replace(spec, entry_price=entry),
+        path,
+        **{"initial_cash": "10", "fee_per_contract": "0.01", **assumptions},
+    )
+    return identity, path
+
+
+def test_reopen_compare_and_source_bytes_across_synthetic_venues(tmp_path):
+    database = tmp_path / "journal.sqlite3"
+    jobs = worker.PaperJobs(database)
+    identities = []
+    for venue in ("kalshi", "polymarket-us"):
+        for entry in ("0.45", "0.46"):
+            identity, source = enqueue(jobs, venue=venue, entry=entry)
+            identities.append(identity)
+            assert jobs.run_next()["status"] == "completed"
+    before = database.read_bytes()
+    journal = PaperJournal(database)
+    page = journal.history(limit=2)
+    assert [job["id"] for job in page["jobs"]] == list(reversed(identities[2:]))
+    assert page["has_more"] is True
+    assert "result" not in page["jobs"][0]
+    assert journal.history(offset=2, limit=2)["has_more"] is False
+    assert journal.history(offset=4) == {"jobs": [], "has_more": False}
+    first, second = identities[2:]
+    saved = journal.inspect(first)
+    assert saved["config"]["strategy"]["venue"] == "polymarket_us"
+    assert saved["result"] == jobs.inspect(first)["result"]
+    assert journal.recording(first) == source.read_bytes()
+    with localcontext() as context:
+        context.prec = 2
+        compared = journal.compare(first, second)
+    assert [job["id"] for job in compared] == [first, second]
+    for job in compared:
+        assert job["result"]["cash"] == "10.52"
+        assert job["result"]["realized_pnl"] == "0.52"
+        assert job["result"]["total_fees"] == "0.04"
+        assert job["result"]["fill_count"] == 2
+        assert job["result"]["rejection_count"] == 0
+    assert "total_fees" not in journal.inspect(first)["result"]
+    with pytest.raises(ValueError, match="same recording"):
+        journal.compare(identities[0], first)
+    terms = journal.compare_markets(identities[0], first)
+    assert terms["comparison"]["status"] == "compatible"  # Supplied synthetic rules only.
+    assert terms["ids"] == [identities[0], first]
+    assert database.read_bytes() == before
+
+
+def test_reads_do_not_acquire_writer_lock_or_initialize_journal(tmp_path):
+    database = tmp_path / "read only?#.sqlite3"
+    jobs = worker.PaperJobs(database)
+    identity, _ = enqueue(jobs)
+    journal = PaperJournal(database)
+    before = database.read_bytes()
+    with jobs._transaction():
+        assert journal.inspect(identity)["status"] == "queued"
+        assert journal.history()["jobs"][0]["id"] == identity
+        assert journal.recording(identity)
+    assert database.read_bytes() == before
+    missing = tmp_path / "missing" / "never-created.sqlite3"
+    with pytest.raises(FileNotFoundError, match="not found"):
+        PaperJournal(missing).history()
+    assert not missing.parent.exists()
+
+
+def test_historical_models_stay_opaque_and_read_only(tmp_path, monkeypatch):
+    database = tmp_path / "old.sqlite3"
+    jobs = worker.PaperJobs(database)
+    monkeypatch.setattr(worker, "PAPER_MODEL", "neural-paper/1")
+    first, source = enqueue(jobs)
+    second, _ = enqueue(jobs, entry="0.46")
+    historical = {"model": "neural-paper/1", "cash": "7.123456789012345678"}
+    with jobs._transaction() as db:
+        db.execute("UPDATE jobs SET status='completed',result=?", (json.dumps(historical),))
+    before = database.read_bytes()
+    journal = PaperJournal(database)
+    assert journal.inspect(first)["result"] == historical
+    assert journal.recording(first) == source.read_bytes()
+    with pytest.raises(ValueError, match="trace unavailable"):
+        journal.compare(first, second)
+    with pytest.raises(ValueError, match="terms unavailable"):
+        journal.compare_markets(first, second)
+    assert database.read_bytes() == before
+
+
+@pytest.mark.parametrize(
+    "change", [{"initial_cash": "11"}, {"fee_per_contract": "0.02"}, {"max_events": 1000}]
+)
+def test_incompatible_assumptions_and_unfinished_jobs_reject(tmp_path, change):
+    jobs = worker.PaperJobs(tmp_path / "jobs.sqlite3")
+    first, _ = enqueue(jobs)
+    second, _ = enqueue(jobs, **change)
+    journal = PaperJournal(jobs.database)
+    with pytest.raises(ValueError, match="completed"):
+        journal.compare(first, second)
+    jobs.run_next()
+    jobs.run_next()
+    with pytest.raises(ValueError, match="same recording"):
+        journal.compare(first, second)
+    with pytest.raises(ValueError, match="distinct"):
+        journal.compare(first, first)
+    with pytest.raises(ValueError, match="unknown"):
+        journal.inspect("missing")
+
+
+@pytest.mark.parametrize("offset,limit", [(-1, 25), (True, 25), (100001, 25), (0, 0), (0, 101)])
+def test_pagination_is_bounded_before_opening_storage(tmp_path, offset, limit):
+    database = tmp_path / "absent.sqlite3"
+    with pytest.raises(ValueError):
+        PaperJournal(database).history(offset=offset, limit=limit)
+    assert not database.exists()
+
+
+@pytest.mark.parametrize("damage", ["config", "recording", "result", "version", "not_sqlite"])
+def test_damaged_journals_fail_without_repairing_them(tmp_path, damage):
+    database = tmp_path / "jobs.sqlite3"
+    jobs = worker.PaperJobs(database)
+    identity, _ = enqueue(jobs)
+    jobs.run_next()
+    if damage == "not_sqlite":
+        database.write_bytes(b"not a SQLite journal")
+    else:
+        with sqlite3.connect(database) as db:
+            if damage == "config":
+                db.execute("UPDATE jobs SET config=?", ('{"model":"changed"}',))
+            elif damage == "recording":
+                db.execute("UPDATE jobs SET recording=?", (b"changed",))
+            elif damage == "result":
+                db.execute("UPDATE jobs SET result=?", ("[]",))
+            else:
+                db.execute("PRAGMA user_version=99")
+    before = database.read_bytes()
+    journal = PaperJournal(database)
+    with pytest.raises(ValueError):
+        if damage == "recording":
+            journal.recording(identity)
+        else:
+            journal.inspect(identity)
+    assert database.read_bytes() == before

From 05c533b23b5efc082152dac76a42ed24af819b1d Mon Sep 17 00:00:00 2001
From: hudsonaikins-crown <hudson.aikins@blanoire.co>
Date: Thu, 10 Sep 2026 19:29:39 -0400
Subject: [PATCH 2/7] NRCL-104 Document journal query invariants

---
 neural/paper_query.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/neural/paper_query.py b/neural/paper_query.py
index b049269a..af2e2f90 100644
--- a/neural/paper_query.py
+++ b/neural/paper_query.py
@@ -25,12 +25,14 @@ class PaperJournal:
     """
 
     def __init__(self, database: str | Path):
+        """Select an existing local journal path without opening or creating it."""
         if str(database) in ("", ":memory:"):
             raise ValueError("paper journal requires an existing database file")
         self.database = Path(database).resolve()
 
     @contextmanager
     def _connection(self) -> Iterator[sqlite3.Connection]:
+        """Open a read snapshot and reject journals outside the supported schema."""
         if not self.database.exists():
             raise FileNotFoundError("paper job database not found")
         db = None
@@ -53,6 +55,7 @@ def _connection(self) -> Iterator[sqlite3.Connection]:
 
     @staticmethod
     def _view(row: sqlite3.Row) -> dict[str, Any]:
+        """Decode a saved row while checking configuration identity and state."""
         try:
             if not isinstance(row["config"], str):
                 raise ValueError("stored job configuration is invalid")
@@ -78,6 +81,7 @@ def _view(row: sqlite3.Row) -> dict[str, Any]:
 
     @classmethod
     def _job(cls, db: sqlite3.Connection, identity: str) -> dict[str, Any]:
+        """Read one complete saved job from the caller's snapshot."""
         if not isinstance(identity, str):
             raise ValueError("job identity must be a string")
         row = db.execute(
@@ -124,6 +128,7 @@ def recording(self, identity: str) -> bytes:
             return raw
 
     def _pair(self, first: str, second: str) -> list[dict[str, Any]]:
+        """Read two distinct completed jobs from a single consistent snapshot."""
         if first == second:
             raise ValueError("choose two distinct completed experiments")
         with self._connection() as db:
@@ -134,6 +139,7 @@ def _pair(self, first: str, second: str) -> list[dict[str, Any]]:
 
     @staticmethod
     def _assumptions(job: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
+        """Separate fixed replay inputs from tunable strategy thresholds and caps."""
         config = dict(job["config"])
         strategy = config.pop("strategy", None)
         StrategySpec.from_dict(strategy)

From 918beaf44296db5216384b69ec7437f4a70005f4 Mon Sep 17 00:00:00 2001
From: hudsonaikins-crown <hudson.aikins@blanoire.co>
Date: Thu, 10 Sep 2026 19:36:34 -0400
Subject: [PATCH 3/7] NRCL-104 Preserve derived fee precision and SQLite
 contention

---
 docs/trading/durable-paper.mdx |  3 +-
 neural/paper_query.py          | 25 ++++++++++++----
 tests/test_paper_query.py      | 52 ++++++++++++++++++++++++++++++++++
 3 files changed, 74 insertions(+), 6 deletions(-)

diff --git a/docs/trading/durable-paper.mdx b/docs/trading/durable-paper.mdx
index 8db76554..bcd2d083 100644
--- a/docs/trading/durable-paper.mdx
+++ b/docs/trading/durable-paper.mdx
@@ -83,7 +83,8 @@ writer reservation, creating a queue, migrating its schema or running simulation
 Missing files raise `FileNotFoundError`; unknown jobs, invalid stored data and
 unsupported journal versions raise `ValueError`. History orders by insertion,
 newest first; offset is 0–100,000 and page size is 1–100. No timestamps are invented.
-Reading can still report database-busy if SQLite holds an exclusive lock.
+Reading preserves `sqlite3.OperationalError` for database-busy contention if SQLite
+holds an exclusive lock, so callers can distinguish it from invalid stored data.
 
 `recording(job_id)` verifies the stored configuration/job identity and the original
 recording's size and digest before returning its exact bytes. Inspection preserves
diff --git a/neural/paper_query.py b/neural/paper_query.py
index af2e2f90..5675f55f 100644
--- a/neural/paper_query.py
+++ b/neural/paper_query.py
@@ -4,6 +4,7 @@
 
 import hashlib
 import json
+import re
 import sqlite3
 from collections.abc import Iterator
 from contextlib import contextmanager
@@ -13,7 +14,7 @@
 
 from neural.paper_worker import APPLICATION_ID, MAX_INPUT_BYTES
 from neural.sports import SportsMarket, compare_sports_markets
-from neural.strategy import StrategySpec, _decimal
+from neural.strategy import StrategySpec, _decimal_text
 
 
 class PaperJournal:
@@ -48,6 +49,11 @@ def _connection(self) -> Iterator[sqlite3.Connection]:
                 raise ValueError("unsupported paper job database")
             yield db
         except sqlite3.Error as exc:
+            if isinstance(exc, sqlite3.OperationalError) and str(exc) in (
+                "database is locked",
+                "database table is locked",
+            ):
+                raise
             raise ValueError("cannot read paper job database") from exc
         finally:
             if db is not None:
@@ -165,14 +171,23 @@ def compare(self, first: str, second: str) -> list[dict[str, Any]]:
                 raise ValueError("saved comparison trace unavailable or invalid")
             with localcontext() as context:
                 context.prec = 100
-                total_fees = sum(
-                    (_decimal(event.get("fees", "0"), "fees") for event in trace), Decimal(0)
-                )
-            job["result"]["total_fees"] = str(total_fees)
+                total_fees = sum((self._trace_fee(event) for event in trace), Decimal(0))
+            job["result"]["total_fees"] = _decimal_text(total_fees)
             job["result"]["fill_count"] = sum(event["action"] == "fill" for event in trace)
             job["result"]["rejection_count"] = sum(event["action"] == "reject" for event in trace)
         return jobs
 
+    @staticmethod
+    def _trace_fee(event: dict[str, Any]) -> Decimal:
+        """Parse products of two 18-digit inputs without truncating their scale."""
+        value = event.get("fees", "0")
+        if (
+            not isinstance(value, str)
+            or re.fullmatch(r"(0|[1-9][0-9]{0,35})(\.[0-9]{1,36})?", value) is None
+        ):
+            raise ValueError("saved trace fees must be a finite nonnegative decimal product")
+        return Decimal(value)
+
     def compare_markets(self, first: str, second: str) -> dict[str, Any]:
         """Inspect sports terms separately; this never compares execution costs/PnL."""
         jobs = self._pair(first, second)
diff --git a/tests/test_paper_query.py b/tests/test_paper_query.py
index 4f64c524..6bd5504d 100644
--- a/tests/test_paper_query.py
+++ b/tests/test_paper_query.py
@@ -100,6 +100,58 @@ def test_historical_models_stay_opaque_and_read_only(tmp_path, monkeypatch):
     assert database.read_bytes() == before
 
 
+def test_derived_fees_preserve_product_scale(tmp_path):
+    database = tmp_path / "tiny-fees.sqlite3"
+    jobs = worker.PaperJobs(database)
+    spec, recording = inputs()
+    ids = []
+    for entry in ("0.45", "0.46"):
+        ids.append(
+            jobs.submit(
+                replace(spec, entry_price=entry, quantity="0.01"),
+                recording,
+                initial_cash="10",
+                fee_per_contract="0.000000000000000001",
+            )
+        )
+        assert jobs.run_next()["status"] == "completed"
+    before = database.read_bytes()
+    for job in PaperJournal(jobs.database).compare(*ids):
+        assert job["result"]["total_fees"] == "0.00000000000000000002"
+        assert job["result"]["fill_count"] == 2
+    assert database.read_bytes() == before
+
+
+@pytest.mark.parametrize("fee", ["NaN", "Infinity", "-0.01", 0.01, "0." + "0" * 36 + "1"])
+def test_invalid_saved_fees_reject_comparison(tmp_path, fee):
+    jobs = worker.PaperJobs(tmp_path / "invalid-fees.sqlite3")
+    first, _ = enqueue(jobs)
+    second, _ = enqueue(jobs, entry="0.46")
+    jobs.run_next()
+    jobs.run_next()
+    result = jobs.inspect(first)["result"]
+    result["trace"][0]["fees"] = fee
+    with sqlite3.connect(jobs.database) as db:
+        db.execute("UPDATE jobs SET result=? WHERE id=?", (json.dumps(result), first))
+    with pytest.raises(ValueError, match="saved trace fees"):
+        PaperJournal(jobs.database).compare(first, second)
+
+
+def test_exclusive_lock_stays_distinguishable_from_invalid_data(tmp_path, monkeypatch):
+    database = tmp_path / "busy.sqlite3"
+    worker.PaperJobs(database)
+    before = database.read_bytes()
+    connect = sqlite3.connect
+    monkeypatch.setattr(
+        sqlite3, "connect", lambda *args, **kwargs: connect(*args, **{**kwargs, "timeout": 0})
+    )
+    with connect(database) as writer:
+        writer.execute("BEGIN EXCLUSIVE")
+        with pytest.raises(sqlite3.OperationalError, match="database is locked"):
+            PaperJournal(database).history()
+    assert database.read_bytes() == before
+
+
 @pytest.mark.parametrize(
     "change", [{"initial_cash": "11"}, {"fee_per_contract": "0.02"}, {"max_events": 1000}]
 )

From 0a1e5fe73409903b67cad0866b6d74f50b816f0a Mon Sep 17 00:00:00 2001
From: hudsonaikins-crown <hudson.aikins@blanoire.co>
Date: Thu, 10 Sep 2026 20:05:42 -0400
Subject: [PATCH 4/7] NRCL-104 Preserve fair comparison inputs and result
 identities

---
 docs/trading/durable-paper.mdx |  5 +++--
 neural/paper_query.py          | 12 +++++++-----
 tests/test_paper_query.py      | 23 +++++++++++++++--------
 3 files changed, 25 insertions(+), 15 deletions(-)

diff --git a/docs/trading/durable-paper.mdx b/docs/trading/durable-paper.mdx
index bcd2d083..2b468f97 100644
--- a/docs/trading/durable-paper.mdx
+++ b/docs/trading/durable-paper.mdx
@@ -93,8 +93,9 @@ correctness. The caller selects the runtime archive and verifies its wheel and
 environment identity. This API does not authenticate users or grant data rights.
 
 `compare(first_id, second_id)` returns two completed jobs with derived
-`total_fees`, `fill_count` and `rejection_count` in their result objects. Recording,
-model, market/outcome, cash, fees and simulation limits must match; strategy
+`total_fees`, `fill_count` and `rejection_count` in a separate `summary` object on
+each job. Saved `result` objects and their `result_id` values remain unchanged. Recording,
+model, market/outcome, quantity, cash, fees and simulation limits must match; strategy
 thresholds and risk limits may differ. Summaries use exact decimals and never
 write back to the journal. Missing historical traces reject comparison explicitly.
 
diff --git a/neural/paper_query.py b/neural/paper_query.py
index 5675f55f..4ebd159c 100644
--- a/neural/paper_query.py
+++ b/neural/paper_query.py
@@ -151,7 +151,7 @@ def _assumptions(job: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
         StrategySpec.from_dict(strategy)
         return config, {
             key: strategy[key]
-            for key in ("schema_version", "kind", "venue", "market_id", "outcome")
+            for key in ("schema_version", "kind", "venue", "market_id", "outcome", "quantity")
         }
 
     def compare(self, first: str, second: str) -> list[dict[str, Any]]:
@@ -160,7 +160,7 @@ def compare(self, first: str, second: str) -> list[dict[str, Any]]:
         if self._assumptions(jobs[0]) != self._assumptions(jobs[1]):
             raise ValueError(
                 "comparison requires the same recording, model, market, outcome, "
-                "cash, fees and simulation limits"
+                "quantity, cash, fees and simulation limits"
             )
         for job in jobs:
             trace = job["result"].get("trace")
@@ -172,9 +172,11 @@ def compare(self, first: str, second: str) -> list[dict[str, Any]]:
             with localcontext() as context:
                 context.prec = 100
                 total_fees = sum((self._trace_fee(event) for event in trace), Decimal(0))
-            job["result"]["total_fees"] = _decimal_text(total_fees)
-            job["result"]["fill_count"] = sum(event["action"] == "fill" for event in trace)
-            job["result"]["rejection_count"] = sum(event["action"] == "reject" for event in trace)
+            job["summary"] = {
+                "total_fees": _decimal_text(total_fees),
+                "fill_count": sum(event["action"] == "fill" for event in trace),
+                "rejection_count": sum(event["action"] == "reject" for event in trace),
+            }
         return jobs
 
     @staticmethod
diff --git a/tests/test_paper_query.py b/tests/test_paper_query.py
index 6bd5504d..948ddf93 100644
--- a/tests/test_paper_query.py
+++ b/tests/test_paper_query.py
@@ -1,5 +1,6 @@
 """Read-only consumer behavior over current, historical and damaged journals."""
 
+import hashlib
 import json
 import sqlite3
 from dataclasses import replace
@@ -12,10 +13,10 @@
 from tests.test_recordings import inputs
 
 
-def enqueue(jobs, *, venue="polymarket-us", entry="0.45", **assumptions):
+def enqueue(jobs, *, venue="polymarket-us", entry="0.45", quantity=None, **assumptions):
     spec, path = inputs(venue)
     identity = jobs.submit(
-        replace(spec, entry_price=entry),
+        replace(spec, entry_price=entry, quantity=quantity or spec.quantity),
         path,
         **{"initial_cash": "10", "fee_per_contract": "0.01", **assumptions},
     )
@@ -51,9 +52,14 @@ def test_reopen_compare_and_source_bytes_across_synthetic_venues(tmp_path):
     for job in compared:
         assert job["result"]["cash"] == "10.52"
         assert job["result"]["realized_pnl"] == "0.52"
-        assert job["result"]["total_fees"] == "0.04"
-        assert job["result"]["fill_count"] == 2
-        assert job["result"]["rejection_count"] == 0
+        assert job["summary"]["total_fees"] == "0.04"
+        assert job["summary"]["fill_count"] == 2
+        assert job["summary"]["rejection_count"] == 0
+        assert job["result"] == jobs.inspect(job["id"])["result"]
+        report = dict(job["result"])
+        result_id = report.pop("result_id")
+        encoded = json.dumps(report, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+        assert hashlib.sha256(encoded.encode()).hexdigest() == result_id
     assert "total_fees" not in journal.inspect(first)["result"]
     with pytest.raises(ValueError, match="same recording"):
         journal.compare(identities[0], first)
@@ -117,8 +123,8 @@ def test_derived_fees_preserve_product_scale(tmp_path):
         assert jobs.run_next()["status"] == "completed"
     before = database.read_bytes()
     for job in PaperJournal(jobs.database).compare(*ids):
-        assert job["result"]["total_fees"] == "0.00000000000000000002"
-        assert job["result"]["fill_count"] == 2
+        assert job["summary"]["total_fees"] == "0.00000000000000000002"
+        assert job["summary"]["fill_count"] == 2
     assert database.read_bytes() == before
 
 
@@ -153,7 +159,8 @@ def test_exclusive_lock_stays_distinguishable_from_invalid_data(tmp_path, monkey
 
 
 @pytest.mark.parametrize(
-    "change", [{"initial_cash": "11"}, {"fee_per_contract": "0.02"}, {"max_events": 1000}]
+    "change",
+    [{"initial_cash": "11"}, {"fee_per_contract": "0.02"}, {"max_events": 1000}, {"quantity": "1"}],
 )
 def test_incompatible_assumptions_and_unfinished_jobs_reject(tmp_path, change):
     jobs = worker.PaperJobs(tmp_path / "jobs.sqlite3")

From 3aa452f93e5310dab078939bcc5e02805c2bd7f7 Mon Sep 17 00:00:00 2001
From: hudsonaikins-crown <hudson.aikins@blanoire.co>
Date: Thu, 10 Sep 2026 20:50:48 -0400
Subject: [PATCH 5/7] NRCL-104 Reject incomplete fees and malformed result
 storage

---
 neural/paper_query.py     |  4 +++-
 tests/test_paper_query.py | 23 ++++++++++++++++++++++-
 2 files changed, 25 insertions(+), 2 deletions(-)

diff --git a/neural/paper_query.py b/neural/paper_query.py
index 4ebd159c..aa0de7c8 100644
--- a/neural/paper_query.py
+++ b/neural/paper_query.py
@@ -75,6 +75,8 @@ def _view(row: sqlite3.Row) -> dict[str, Any]:
             view = {key: row[key] for key in ("id", "status", "error")}
             view["config"] = config
             if "result" in row.keys():
+                if row["result"] is not None and not isinstance(row["result"], str):
+                    raise ValueError("stored job result is invalid")
                 result = json.loads(row["result"]) if row["result"] is not None else None
                 if (result is not None and not isinstance(result, dict)) or (
                     (row["status"] == "completed") != (result is not None)
@@ -182,7 +184,7 @@ def compare(self, first: str, second: str) -> list[dict[str, Any]]:
     @staticmethod
     def _trace_fee(event: dict[str, Any]) -> Decimal:
         """Parse products of two 18-digit inputs without truncating their scale."""
-        value = event.get("fees", "0")
+        value = event.get("fees") if event["action"] == "fill" else event.get("fees", "0")
         if (
             not isinstance(value, str)
             or re.fullmatch(r"(0|[1-9][0-9]{0,35})(\.[0-9]{1,36})?", value) is None
diff --git a/tests/test_paper_query.py b/tests/test_paper_query.py
index 948ddf93..48b75669 100644
--- a/tests/test_paper_query.py
+++ b/tests/test_paper_query.py
@@ -158,6 +158,23 @@ def test_exclusive_lock_stays_distinguishable_from_invalid_data(tmp_path, monkey
     assert database.read_bytes() == before
 
 
+def test_fill_without_fee_rejects_comparison(tmp_path):
+    database = tmp_path / "missing-fee.sqlite3"
+    jobs = worker.PaperJobs(database)
+    first, _ = enqueue(jobs)
+    second, _ = enqueue(jobs, entry="0.46")
+    jobs.run_next()
+    jobs.run_next()
+    result = jobs.inspect(first)["result"]
+    next(event for event in result["trace"] if event["action"] == "fill").pop("fees")
+    with sqlite3.connect(jobs.database) as db:
+        db.execute("UPDATE jobs SET result=? WHERE id=?", (json.dumps(result), first))
+    before = database.read_bytes()
+    with pytest.raises(ValueError, match="saved trace fees"):
+        PaperJournal(jobs.database).compare(first, second)
+    assert database.read_bytes() == before
+
+
 @pytest.mark.parametrize(
     "change",
     [{"initial_cash": "11"}, {"fee_per_contract": "0.02"}, {"max_events": 1000}, {"quantity": "1"}],
@@ -187,7 +204,9 @@ def test_pagination_is_bounded_before_opening_storage(tmp_path, offset, limit):
     assert not database.exists()
 
 
-@pytest.mark.parametrize("damage", ["config", "recording", "result", "version", "not_sqlite"])
+@pytest.mark.parametrize(
+    "damage", ["config", "recording", "result", "result_bytes", "version", "not_sqlite"]
+)
 def test_damaged_journals_fail_without_repairing_them(tmp_path, damage):
     database = tmp_path / "jobs.sqlite3"
     jobs = worker.PaperJobs(database)
@@ -203,6 +222,8 @@ def test_damaged_journals_fail_without_repairing_them(tmp_path, damage):
                 db.execute("UPDATE jobs SET recording=?", (b"changed",))
             elif damage == "result":
                 db.execute("UPDATE jobs SET result=?", ("[]",))
+            elif damage == "result_bytes":
+                db.execute("UPDATE jobs SET result=?", (b"\xff",))
             else:
                 db.execute("PRAGMA user_version=99")
     before = database.read_bytes()

From 22d88006b73c2ae709fbb7b23118fd7a5cb486db Mon Sep 17 00:00:00 2001
From: hudsonaikins-crown <hudson.aikins@blanoire.co>
Date: Thu, 10 Sep 2026 21:06:20 -0400
Subject: [PATCH 6/7] NRCL-104 Harden stored data and archive read boundaries

---
 docs/trading/durable-paper.mdx |  7 ++++-
 neural/paper_query.py          | 34 ++++++++++++++++++------
 tests/test_paper_query.py      | 48 ++++++++++++++++++++++++++++++++++
 3 files changed, 80 insertions(+), 9 deletions(-)

diff --git a/docs/trading/durable-paper.mdx b/docs/trading/durable-paper.mdx
index 2b468f97..51e923a1 100644
--- a/docs/trading/durable-paper.mdx
+++ b/docs/trading/durable-paper.mdx
@@ -85,9 +85,14 @@ unsupported journal versions raise `ValueError`. History orders by insertion,
 newest first; offset is 0–100,000 and page size is 1–100. No timestamps are invented.
 Reading preserves `sqlite3.OperationalError` for database-busy contention if SQLite
 holds an exclusive lock, so callers can distinguish it from invalid stored data.
+Only rollback-journal databases are supported. WAL-format journals are rejected
+before SQLite opens them, preventing creation of WAL/SHM sidecars. The caller must
+not change journal mode or replace the database during a read.
 
 `recording(job_id)` verifies the stored configuration/job identity and the original
-recording's size and digest before returning its exact bytes. Inspection preserves
+recording's size and digest before returning its exact bytes.
+Results are not decoded when recovering recording bytes; a damaged output does not
+prevent recovery of an intact, verified input. Inspection preserves
 historical results as recorded; it does not recalculate them or certify their
 correctness. The caller selects the runtime archive and verifies its wheel and
 environment identity. This API does not authenticate users or grant data rights.
diff --git a/neural/paper_query.py b/neural/paper_query.py
index aa0de7c8..149934d0 100644
--- a/neural/paper_query.py
+++ b/neural/paper_query.py
@@ -36,6 +36,10 @@ def _connection(self) -> Iterator[sqlite3.Connection]:
         """Open a read snapshot and reject journals outside the supported schema."""
         if not self.database.exists():
             raise FileNotFoundError("paper job database not found")
+        with self.database.open("rb") as source:
+            header = source.read(20)
+        if header.startswith(b"SQLite format 3\x00") and header[18:20] != b"\x01\x01":
+            raise ValueError("WAL paper journals are unsupported; use a rollback-journal archive")
         db = None
         try:
             db = sqlite3.connect(self.database.as_uri() + "?mode=ro", uri=True, timeout=5)
@@ -59,13 +63,18 @@ def _connection(self) -> Iterator[sqlite3.Connection]:
             if db is not None:
                 db.close()
 
+    @staticmethod
+    def _reject_constant(value: str) -> Any:
+        """Reject nonstandard JSON numbers that cannot be written by the worker."""
+        raise ValueError("stored job JSON contains a non-finite constant")
+
     @staticmethod
     def _view(row: sqlite3.Row) -> dict[str, Any]:
         """Decode a saved row while checking configuration identity and state."""
         try:
             if not isinstance(row["config"], str):
                 raise ValueError("stored job configuration is invalid")
-            config = json.loads(row["config"])
+            config = json.loads(row["config"], parse_constant=PaperJournal._reject_constant)
             if (
                 not isinstance(config, dict)
                 or hashlib.sha256(row["config"].encode()).hexdigest() != row["id"]
@@ -73,11 +82,19 @@ def _view(row: sqlite3.Row) -> dict[str, Any]:
             ):
                 raise ValueError("stored job input integrity mismatch")
             view = {key: row[key] for key in ("id", "status", "error")}
+            if (row["error"] is not None and not isinstance(row["error"], str)) or (
+                (row["status"] == "failed") != (row["error"] is not None)
+            ):
+                raise ValueError("stored job error is invalid")
             view["config"] = config
             if "result" in row.keys():
                 if row["result"] is not None and not isinstance(row["result"], str):
                     raise ValueError("stored job result is invalid")
-                result = json.loads(row["result"]) if row["result"] is not None else None
+                result = (
+                    json.loads(row["result"], parse_constant=PaperJournal._reject_constant)
+                    if row["result"] is not None
+                    else None
+                )
                 if (result is not None and not isinstance(result, dict)) or (
                     (row["status"] == "completed") != (result is not None)
                 ):
@@ -88,13 +105,14 @@ def _view(row: sqlite3.Row) -> dict[str, Any]:
             raise ValueError("stored job data is invalid") from exc
 
     @classmethod
-    def _job(cls, db: sqlite3.Connection, identity: str) -> dict[str, Any]:
-        """Read one complete saved job from the caller's snapshot."""
+    def _job(
+        cls, db: sqlite3.Connection, identity: str, *, include_result: bool = True
+    ) -> dict[str, Any]:
+        """Read one saved job, optionally excluding its unrelated output."""
         if not isinstance(identity, str):
             raise ValueError("job identity must be a string")
-        row = db.execute(
-            "SELECT id,status,error,config,result FROM jobs WHERE id=?", (identity,)
-        ).fetchone()
+        fields = "id,status,error,config,result" if include_result else "id,status,error,config"
+        row = db.execute(f"SELECT {fields} FROM jobs WHERE id=?", (identity,)).fetchone()
         if row is None:
             raise ValueError("unknown paper job")
         return cls._view(row)
@@ -123,7 +141,7 @@ def inspect(self, identity: str) -> dict[str, Any]:
     def recording(self, identity: str) -> bytes:
         """Return original input bytes after checking size and the saved digest."""
         with self._connection() as db:
-            job = self._job(db, identity)
+            job = self._job(db, identity, include_result=False)
             raw = db.execute(
                 "SELECT CASE WHEN length(recording) BETWEEN 1 AND ? THEN recording END "
                 "FROM jobs WHERE id=?",
diff --git a/tests/test_paper_query.py b/tests/test_paper_query.py
index 48b75669..ca7b3a16 100644
--- a/tests/test_paper_query.py
+++ b/tests/test_paper_query.py
@@ -204,6 +204,54 @@ def test_pagination_is_bounded_before_opening_storage(tmp_path, offset, limit):
     assert not database.exists()
 
 
+@pytest.mark.parametrize(
+    "result", ['{"cash": NaN}', '{"cash": Infinity}', '{"cash": -Infinity}', "{"]
+)
+def test_invalid_result_does_not_prevent_verified_recording_recovery(tmp_path, result):
+    database = tmp_path / "damaged-output.sqlite3"
+    jobs = worker.PaperJobs(database)
+    identity, source = enqueue(jobs)
+    jobs.run_next()
+    with sqlite3.connect(database) as db:
+        db.execute("UPDATE jobs SET result=? WHERE id=?", (result, identity))
+    before = database.read_bytes()
+    journal = PaperJournal(database)
+    with pytest.raises(ValueError):
+        journal.inspect(identity)
+    assert journal.recording(identity) == source.read_bytes()
+    assert database.read_bytes() == before
+
+
+@pytest.mark.parametrize("status,error", [("failed", b"bad"), ("failed", None), ("queued", "bad")])
+def test_invalid_error_state_rejects_history_and_inspection(tmp_path, status, error):
+    database = tmp_path / "bad-error.sqlite3"
+    jobs = worker.PaperJobs(database)
+    identity, _ = enqueue(jobs)
+    with sqlite3.connect(database) as db:
+        db.execute("PRAGMA ignore_check_constraints=ON")
+        db.execute("UPDATE jobs SET status=?,error=?", (status, error))
+    before = database.read_bytes()
+    journal = PaperJournal(database)
+    with pytest.raises(ValueError, match="stored job error"):
+        journal.inspect(identity)
+    with pytest.raises(ValueError, match="stored job error"):
+        journal.history()
+    assert database.read_bytes() == before
+
+
+def test_wal_archive_rejected_without_creating_sidecars(tmp_path):
+    database = tmp_path / "wal.sqlite3"
+    worker.PaperJobs(database)
+    db = sqlite3.connect(database)
+    db.execute("PRAGMA journal_mode=WAL")
+    db.close()
+    before = {path.name: path.read_bytes() for path in tmp_path.iterdir()}
+    assert set(before) == {database.name}
+    with pytest.raises(ValueError, match="WAL paper journals"):
+        PaperJournal(database).history()
+    assert {path.name: path.read_bytes() for path in tmp_path.iterdir()} == before
+
+
 @pytest.mark.parametrize(
     "damage", ["config", "recording", "result", "result_bytes", "version", "not_sqlite"]
 )

From 37474320e471416e4a8551525d97fa84705d9f97 Mon Sep 17 00:00:00 2001
From: hudsonaikins-crown <hudson.aikins@blanoire.co>
Date: Thu, 10 Sep 2026 21:16:38 -0400
Subject: [PATCH 7/7] NRCL-104 Validate finite numbers and saved history
 consistency

---
 neural/paper_query.py     | 33 +++++++++++++++++++++++++++++----
 tests/test_paper_query.py | 29 +++++++++++++++++++++++++++--
 2 files changed, 56 insertions(+), 6 deletions(-)

diff --git a/neural/paper_query.py b/neural/paper_query.py
index 149934d0..64b7e1e7 100644
--- a/neural/paper_query.py
+++ b/neural/paper_query.py
@@ -4,6 +4,7 @@
 
 import hashlib
 import json
+import math
 import re
 import sqlite3
 from collections.abc import Iterator
@@ -68,13 +69,25 @@ def _reject_constant(value: str) -> Any:
         """Reject nonstandard JSON numbers that cannot be written by the worker."""
         raise ValueError("stored job JSON contains a non-finite constant")
 
+    @staticmethod
+    def _finite_float(value: str) -> float:
+        """Decode standard JSON numbers without allowing floating-point overflow."""
+        number = float(value)
+        if not math.isfinite(number):
+            raise ValueError("stored job JSON contains a non-finite number")
+        return number
+
     @staticmethod
     def _view(row: sqlite3.Row) -> dict[str, Any]:
         """Decode a saved row while checking configuration identity and state."""
         try:
             if not isinstance(row["config"], str):
                 raise ValueError("stored job configuration is invalid")
-            config = json.loads(row["config"], parse_constant=PaperJournal._reject_constant)
+            config = json.loads(
+                row["config"],
+                parse_constant=PaperJournal._reject_constant,
+                parse_float=PaperJournal._finite_float,
+            )
             if (
                 not isinstance(config, dict)
                 or hashlib.sha256(row["config"].encode()).hexdigest() != row["id"]
@@ -87,11 +100,19 @@ def _view(row: sqlite3.Row) -> dict[str, Any]:
             ):
                 raise ValueError("stored job error is invalid")
             view["config"] = config
+            if "has_result" in row.keys() and (
+                (row["status"] == "completed") != bool(row["has_result"])
+            ):
+                raise ValueError("stored job result is invalid")
             if "result" in row.keys():
                 if row["result"] is not None and not isinstance(row["result"], str):
                     raise ValueError("stored job result is invalid")
                 result = (
-                    json.loads(row["result"], parse_constant=PaperJournal._reject_constant)
+                    json.loads(
+                        row["result"],
+                        parse_constant=PaperJournal._reject_constant,
+                        parse_float=PaperJournal._finite_float,
+                    )
                     if row["result"] is not None
                     else None
                 )
@@ -125,7 +146,8 @@ def history(self, *, offset: int = 0, limit: int = 25) -> dict[str, Any]:
             raise ValueError("limit must be an integer from 1 to 100")
         with self._connection() as db:
             rows = db.execute(
-                "SELECT id,status,error,config FROM jobs ORDER BY rowid DESC LIMIT ? OFFSET ?",
+                "SELECT id,status,error,config,result IS NOT NULL AS has_result "
+                "FROM jobs ORDER BY rowid DESC LIMIT ? OFFSET ?",
                 (limit + 1, offset),
             ).fetchall()
             return {
@@ -208,7 +230,10 @@ def _trace_fee(event: dict[str, Any]) -> Decimal:
             or re.fullmatch(r"(0|[1-9][0-9]{0,35})(\.[0-9]{1,36})?", value) is None
         ):
             raise ValueError("saved trace fees must be a finite nonnegative decimal product")
-        return Decimal(value)
+        fee = Decimal(value)
+        if event["action"] != "fill" and fee != 0:
+            raise ValueError("saved trace fees require a fill event")
+        return fee
 
     def compare_markets(self, first: str, second: str) -> dict[str, Any]:
         """Inspect sports terms separately; this never compares execution costs/PnL."""
diff --git a/tests/test_paper_query.py b/tests/test_paper_query.py
index ca7b3a16..4ea3bf80 100644
--- a/tests/test_paper_query.py
+++ b/tests/test_paper_query.py
@@ -128,7 +128,7 @@ def test_derived_fees_preserve_product_scale(tmp_path):
     assert database.read_bytes() == before
 
 
-@pytest.mark.parametrize("fee", ["NaN", "Infinity", "-0.01", 0.01, "0." + "0" * 36 + "1"])
+@pytest.mark.parametrize("fee", ["NaN", "Infinity", "-0.01", 0.01, "0." + "0" * 36 + "1", "0.01"])
 def test_invalid_saved_fees_reject_comparison(tmp_path, fee):
     jobs = worker.PaperJobs(tmp_path / "invalid-fees.sqlite3")
     first, _ = enqueue(jobs)
@@ -205,7 +205,15 @@ def test_pagination_is_bounded_before_opening_storage(tmp_path, offset, limit):
 
 
 @pytest.mark.parametrize(
-    "result", ['{"cash": NaN}', '{"cash": Infinity}', '{"cash": -Infinity}', "{"]
+    "result",
+    [
+        '{"cash": NaN}',
+        '{"cash": Infinity}',
+        '{"cash": -Infinity}',
+        '{"cash": 1e999}',
+        '{"cash": -1e999}',
+        "{",
+    ],
 )
 def test_invalid_result_does_not_prevent_verified_recording_recovery(tmp_path, result):
     database = tmp_path / "damaged-output.sqlite3"
@@ -252,6 +260,23 @@ def test_wal_archive_rejected_without_creating_sidecars(tmp_path):
     assert {path.name: path.read_bytes() for path in tmp_path.iterdir()} == before
 
 
+@pytest.mark.parametrize(
+    "status,result,error",
+    [("completed", None, None), ("queued", "{}", None), ("failed", "{}", "failure")],
+)
+def test_history_rejects_invalid_result_presence(tmp_path, status, result, error):
+    database = tmp_path / "bad-state.sqlite3"
+    jobs = worker.PaperJobs(database)
+    enqueue(jobs)
+    with sqlite3.connect(database) as db:
+        db.execute("PRAGMA ignore_check_constraints=ON")
+        db.execute("UPDATE jobs SET status=?,result=?,error=?", (status, result, error))
+    before = database.read_bytes()
+    with pytest.raises(ValueError, match="stored job result"):
+        PaperJournal(database).history()
+    assert database.read_bytes() == before
+
+
 @pytest.mark.parametrize(
     "damage", ["config", "recording", "result", "result_bytes", "version", "not_sqlite"]
 )