From 85fe8c3fa7e5708e4b5ecbbfdcb2da0de364c307 Mon Sep 17 00:00:00 2001 From: Elmehdi Aitbrahim Date: Sat, 5 Sep 2026 12:16:29 -0400 Subject: [PATCH 1/2] feat(engine): an append-only hash chain for the book, and a column that stops saying NOT RECORDED (#721) #703's activity export shipped with a `row_hash` column whose every cell read NOT RECORDED. That was honest and it was the whole of what could be said: none of the four stores the timeline merges hashed its rows. This is the record that column was built to read. THE CHAIN IS OVER EVENTS, NOT OVER THE BOOK `orders` rows are MUTATED -- `update_order` writes status, fills and fees as a venue reports them, sometimes days later via `execution.reconcile`. A hash chained over the order row itself would break on every legitimate fill, and a chain that cries wolf on ordinary operation is a chain an operator learns to ignore. The issue named both options and said the second is the one that actually delivers tamper-evidence for a mutable book. So `keel/data/audit.py` chains immutable STATEMENTS about the book. Each write to `orders`, `transactions` and both attestation tables appends one event, in the SAME transaction as the row it describes. The book stays mutable and queryable; the event stream is what verifies. Re-importing a transaction appends rather than rewrites -- two events for one `coinbase_id` is the record that a line arrived twice with different content, which for a store whose provenance is `imported-ledger` is exactly what should be visible. One chain, not one per table. A per-table chain would let an event be moved between streams undetectably, and a removed order event would only be visible to someone who thought to verify the order chain specifically. ONE CANONICAL FORM, AND A PROOF THAT IT DID NOT MOVE The issue's hardest constraint: `research/ledger.py` already decides what canonical JSON means here, and two canonicalisations that disagree produce two hashes for one row -- invisible at write time, surfacing months later as a chain that "cannot be verified", at exactly the moment someone is establishing whether a record was altered. `keel_core.hashchain` now holds the form and the walk; both stores import it. `keel/research/ledger.py` keeps deciding WHAT a trial commits to, because only it knows. The move is pinned by the one test in this repo that asserts against hashes computed by a previous version of the code: the 93 git-tracked rows in `docs/experiments/trials-ledger.jsonl` still verify byte-for-byte. Dropping `separators` from the canonicaliser fails it. `verify_links` also grew a `find_breaks` underneath it -- one walk, formatted two ways -- so the timeline can locate the first break without parsing English out of a message. THREE READINGS, AND THE MIDDLE ONE IS THE POINT The timeline's column now carries `chained`, `not chained` or `chain broken`, with `chain_status` beside `row_hash` rather than inferred from it. `not chained` is rows predating this and every engine-log row (the log is a FILE, not a chained store; a cycle row carrying a hash would be this codebase attesting to something it merely read). An honest gap, deliberately NOT a break, so upgrading into the chain does not open the timeline to a page of red. `chain broken` is a row whose event falls at or after the first break. Its hash is still SHOWN -- hiding it would destroy the value someone verifying would work from -- and the status refuses to call it evidence. A chain proves a sequence, so past a break the sequence is unproven, whatever the single row's own hash still says. Showing those as `chained` would present unverified values as evidence, which is the one thing the column exists to prevent. GREEN REQUIRES SOMETHING TO HAVE BEEN CHECKED `intact` is `no errors AND events exist`. An empty chain reports no breaks because there is nothing in it to break, and `keel doctor` says "audit chain empty -- unverified" rather than "verifies". That is `_chain_payload`'s four-state lesson from #708, on the trading side, where the store cannot be absent so there are three. WHAT IS REFUSED No backfill. No borrowing the trials ledger's hashes -- different domain, different records; the two stores share a canonicaliser and nothing else. No hash computed at export time. TWO SAFETY PROPERTIES THAT NEEDED TESTS TO STAY TRUE `write_transaction` is `BEGIN IMMEDIATE`, and the lock is asserted from BLOCK ENTRY. A deferred transaction takes its lock at the first WRITE -- by which point the chain head has already been read, and two writers racing both write the same `prev_hash` and fork the chain, silently, because each row verifies against the row it believes precedes it. The first version of that test appended before checking the lock, which passes under plain `BEGIN` and so proved nothing. The head read is `ORDER BY seq_id`, never `ts`. The first version of that test used two events, where the head read has one candidate and any ordering picks it; ordering by `ts` survived. Three events, timestamps deliberately out of order, is what shows the fork. Readers tolerate a database without the table; writers do not. `keel/web/ server.py` and `keel mcp`'s `_open_readonly_repo` both open a repo WITHOUT migrating, so a pre-v20 database reaching a reader is ordinary -- #718 shipped a reader that raised on exactly this and took all of `gather_findings` down. Checked via `sqlite_master` rather than caught, so a genuinely corrupt database is not reported as an un-upgraded one. A WRITE that cannot record its event still fails loudly: that is the failed chain write masquerading as an honest gap. Eleven mutants killed, including: the canonicaliser losing its separators, the head read taking no write lock, `prev_hash` dropping out of what a row commits to, the head read ordering by timestamp, `latest_hashes` forgetting which store a row came from, an empty chain counting as intact, the order row landing without its event, a row past a break still reading as chained, NOT RECORDED being truncated like a hash, and the export dropping the status column. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KZZxmspQXe5qJ9FAsG13s6 --- keel/commands/doctor.py | 70 ++++ keel/commands/timeline.py | 150 +++++++- keel/data/audit.py | 371 ++++++++++++++++++++ keel/data/repository.py | 216 ++++++++---- keel/research/ledger.py | 65 ++-- keel/web/payload.py | 87 ++++- keel/web/static/js/render.js | 15 +- packages/keel-core/keel_core/hashchain.py | 185 ++++++++++ tests/commands/test_doctor.py | 62 ++++ tests/commands/test_timeline.py | 206 ++++++++++- tests/core/test_hashchain.py | 162 +++++++++ tests/data/test_audit_chain.py | 408 ++++++++++++++++++++++ tests/research/test_ledger.py | 24 ++ tests/web/test_payload.py | 105 +++++- tests/web/test_timeline_export.py | 10 + 15 files changed, 2014 insertions(+), 122 deletions(-) create mode 100644 keel/data/audit.py create mode 100644 packages/keel-core/keel_core/hashchain.py create mode 100644 tests/core/test_hashchain.py create mode 100644 tests/data/test_audit_chain.py diff --git a/keel/commands/doctor.py b/keel/commands/doctor.py index 54b29bf..8e5bc67 100644 --- a/keel/commands/doctor.py +++ b/keel/commands/doctor.py @@ -300,6 +300,72 @@ def asset_attestation_window_findings( ) +def audit_chain_findings(state: Any) -> list[Finding]: + """Does the append-only audit chain still verify? (#721) + + THREE readings, and the middle one is the whole reason this is not a boolean: + + * **no `audit_events` table** -- a database this build has not migrated. Both the web server + and `keel mcp` open a repo WITHOUT migrating (a view must not take a schema write lock), so + an un-upgraded database reaching a reader is ordinary, not an error. + * **a table with no events** -- nothing has been written since the chain shipped. Reported as + OK, and the headline says UNVERIFIED rather than verified: an empty chain has no breaks + because it has nothing in it to break, and calling that "verified" is a positive claim over + zero rows. + * **a break** -- FAIL. This is the one check in this module where a failure means the RECORD + has been altered rather than a rail being out of position, so it names the row and offers no + `fix`: there is no command that repairs a broken chain, and offering one would imply the + break can be papered over. What an operator does here is investigate. + + `state` is a `keel.data.audit.ChainState`, typed `Any` for the same reason every other + gatherer's input here is: the MCP tool and the click command hand this module rows and records + from a repo it does not import. + """ + if not getattr(state, "table_present", False): + return [ + Finding( + "audit.chain", + OK, + "audit chain not on this database", + "schema predates the append-only audit chain; nothing is recorded to verify", + "keel migrate", + ) + ] + errors = tuple(getattr(state, "errors", ()) or ()) + count = int(getattr(state, "event_count", 0) or 0) + if errors: + return [ + Finding( + "audit.chain", + FAIL, + "audit chain does NOT verify", + f"{len(errors)} break(s) over {count} event(s); first: {errors[0]}", + # No fix. Nothing repairs a broken chain, and a command here would read as though + # something could -- which is the opposite of what a break means. + "-", + ) + ] + if count == 0: + return [ + Finding( + "audit.chain", + OK, + "audit chain empty — unverified", + "no events recorded yet; nothing has been checked", + "-", + ) + ] + return [ + Finding( + "audit.chain", + OK, + "audit chain verifies", + f"{count} event(s), every one still hashing to the next", + "-", + ) + ] + + def instrument_attestation_window_findings( attestations: Iterable[dict[str, Any]], *, now_ts: int ) -> list[Finding]: @@ -1504,6 +1570,10 @@ def gather_findings(repo: Any, config: Any, log_lines: Iterable[str], now_ts: in findings += instrument_attestation_window_findings( repo.get_instrument_attestations(), now_ts=now_ts ) + # #721. Deliberately AFTER the attestation checks and before the venue rails: a chain break + # is a statement about the record those checks were read from, and an operator scanning the + # output should meet it in the same block as the rest of the record's condition. + findings += audit_chain_findings(repo.audit_chain()) findings += trade_scope_findings(repo.get_venue_trade_scope(venue), venue) # #691. Venue-keyed the same way, and reported BEFORE it bites: rail 22 vetoes silently on # a lapse, and the live profile runs unattended. diff --git a/keel/commands/timeline.py b/keel/commands/timeline.py index 2c273ec..bccdd3d 100644 --- a/keel/commands/timeline.py +++ b/keel/commands/timeline.py @@ -13,13 +13,25 @@ **Read-only, no broker, no network.** Same posture as every other service in this package. -**Nothing here is tamper-evident, and the export says so.** #703 asked the CSV to carry each -row's hash. None of these four stores hashes its rows: `orders`, `transactions`, -`asset_attestations` and `instrument_attestations` have no hash column between them, and the only -hash-chained store in this codebase is the research trials ledger (`keel/research/ledger.py`), -which records experiments rather than trading activity and does not belong in this feed. So the -hash column is emitted as NOT RECORDED rather than left blank -- blank invites the reader to -assume the check passed -- and hashing these tables is filed as engine work. +**Tamper-evidence, and its three honest readings (#721).** #703 asked the CSV to carry each row's +hash, and shipped with the column reading NOT RECORDED on every row because no store hashed +anything. `keel/data/audit.py` now chains an append-only event per write to `orders`, +`transactions` and both attestation tables, so this feed reads real hashes -- and has to keep +three readings apart: + +- `chained` -- an event exists for this row and the chain vouches for it. +- `not chained` -- no event was ever written. Rows predating #721, and every engine-log row, which + comes from a FILE rather than a chained store. An honest gap; deliberately NOT a break, so a + deployment upgrading into the chain does not open its timeline to a page of red. +- `chain broken` -- an event exists and falls at or after the first break. The hash is shown and + is NOT evidence: a chain proves a sequence, so past a break the sequence is unproven. Showing + these as `chained` would present unverified values as evidence, which is the one thing the + column exists to prevent. + +The research trials ledger (`keel/research/ledger.py`) is still not in this feed: it records +experiments rather than trading activity, and borrowing its hashes to decorate a trading audit +trail would be provenance laundering. The two stores share `keel_core.hashchain` -- one definition +of canonical JSON, so one row can only ever have one hash -- and nothing else. """ from __future__ import annotations @@ -32,6 +44,7 @@ from typing import Any from keel.commands.orders import normalise_scope, scope_start_ts +from keel.data.audit import ChainState from keel.data.repository import Repository #: The type chips, and the only words `kind` ever takes. @@ -53,10 +66,19 @@ "engine-log", ) -#: What the hash column says until the engine records one. NOT blank: an empty cell in a column +#: What the hash column says where no event was recorded. NOT blank: an empty cell in a column #: headed `row_hash` reads as "nothing to report", and the honest reading is "nobody checked". HASH_NOT_RECORDED = "NOT RECORDED" +#: What the chain says about one row, as a closed vocabulary (#721). The full reasoning is in the +#: module docstring; the short version is that a hash and a verdict are two different facts, and a +#: hash printed without one is a number an auditor cannot use. +CHAIN_STATUSES: tuple[str, ...] = ("chained", "not chained", "chain broken") + +CHAINED = "chained" +NOT_CHAINED = "not chained" +CHAIN_BROKEN = "chain broken" + #: The characters a spreadsheet treats as the start of a formula. OWASP's list. #: #: `\n` and `\r` are here alongside `\t` because a cell can only be re-parsed from its start, @@ -126,8 +148,14 @@ class TimelineRow: #: one thing. amount: Decimal | None amount_kind: str - #: The row's own tamper-evidence, when its store records one. None of the four does today. + #: The row's own tamper-evidence: the `row_hash` of the latest `audit_events` statement about + #: it, or `HASH_NOT_RECORDED` where none was written. row_hash: str = HASH_NOT_RECORDED + #: One of `CHAIN_STATUSES`. Carried BESIDE the hash rather than inferred from it, because + #: "a hash is present" and "the chain vouches for it" are different facts and the second is + #: the one an auditor is actually asking about. A client that inferred the verdict from the + #: presence of a 64-character string would call a broken row verified. + chain_status: str = NOT_CHAINED @dataclass(frozen=True) @@ -157,6 +185,15 @@ class TimelineReport: #: bar that reordered itself as history arrived would move under the reader. kinds_present: tuple[str, ...] + #: Whether the `audit_events` chain holds ANY events (#721). Distinct from `chain_errors` + #: being empty: a verification over zero rows reports nothing and has verified nothing, and a + #: green badge over a table nothing wrote is the failure this codebase keeps re-learning. A + #: deployment that predates the chain and one that has done nothing since are both False here. + chain_recorded: bool = False + #: Every break the chain walk found, verbatim from `keel_core.hashchain`. Reported rather + #: than raised, so both renderers can STATE the chain's condition instead of asserting it. + chain_errors: tuple[str, ...] = () + #: `read_log_window`'s own word for how the engine log read went: `ok`, `missing`, `empty`, #: `oversized`, `unreadable`. Carried rather than acted on, so the page and the CSV can SAY #: the log did not reach this report. @@ -174,6 +211,18 @@ def shown_count(self) -> int: `keel/web/payload.py` may not call `len()` (Rule 6e).""" return len(self.rows) + @property + def chain_intact(self) -> bool: + """Whether the chain found nothing wrong. + + TRUE over an empty chain, and that is the deliberate reading: nothing was checked, so + nothing is broken, and a deployment upgrading into #721 must not open its timeline to a + page of red. `chain_recorded` is the companion that says whether anything was checked at + all, and the two are read together -- exactly the pairing + `payload.py::_chain_payload` holds for the research ledger's badge. + """ + return not self.chain_errors + @property def log_gap(self) -> bool: """Whether the engine log's contents are MISSING FROM this report. @@ -201,7 +250,36 @@ def log_gap(self) -> bool: DEFAULT_TIMELINE_LIMIT = 200 MAX_TIMELINE_LIMIT = 2000 -def _order_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: +class _Chain: + """The chain state, as the two fields a `TimelineRow` carries. + + A thin adapter and not a second source of truth: `keel/data/audit.py` decides what the chain + says, and this decides only how to SAY it on a row. Held as a class so the four row builders + ask one object the same question rather than each reproducing the three-way reading. + """ + + def __init__(self, state: ChainState) -> None: + self._state = state + + def of(self, store: str, reference: str) -> dict[str, str]: + """`row_hash` and `chain_status` for one row, as kwargs. + + A row with no event is `not chained` -- an honest gap, deliberately not a break. A row + whose event falls AT OR AFTER the first break is `chain broken`: its hash is still shown, + because hiding it would destroy the very value an auditor would use to establish what the + row said, but the status refuses to call it evidence. A chain proves a SEQUENCE, so past + a break the sequence is unproven -- which is why this compares `seq_id` against the break + rather than re-verifying the single row, whose own hash may well still match. + """ + seen = self._state.hashes.get((store, reference)) + if seen is None: + return {"row_hash": HASH_NOT_RECORDED, "chain_status": NOT_CHAINED} + broken_from = self._state.first_broken_seq + status = CHAINED if broken_from is None or seen.seq_id < broken_from else CHAIN_BROKEN + return {"row_hash": seen.row_hash, "chain_status": status} + + +def _order_rows(repo: Repository, since_ts: int | None, chain: _Chain) -> list[TimelineRow]: """`orders` -> trade rows. A PAPER order is `simulated`, not `venue-reported`. The paper trader wrote that row with no @@ -224,6 +302,7 @@ def _order_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: provenance="simulated" if mode == "paper" else "venue-reported", source="orders", reference=str(raw.get("id") or ""), + **chain.of("orders", str(raw.get("id") or "")), summary=f"{status} {side} {product} ({mode})".strip(), product_id=product, amount=raw.get("actual_fill"), @@ -233,7 +312,9 @@ def _order_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: return rows -def _transaction_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: +def _transaction_rows( + repo: Repository, since_ts: int | None, chain: _Chain +) -> list[TimelineRow]: """`transactions` -> flow rows. `imported-ledger`, never `venue-reported`: these lines came out of a CSV the operator @@ -255,6 +336,7 @@ def _transaction_rows(repo: Repository, since_ts: int | None) -> list[TimelineRo provenance="imported-ledger", source="transactions", reference=str(raw.get("coinbase_id") or raw.get("id") or ""), + **chain.of("transactions", str(raw.get("coinbase_id") or raw.get("id") or "")), summary=f"{kind_word} {asset}".strip() + (f" -- {note}" if note else ""), product_id="", amount=raw.get("total"), @@ -264,7 +346,9 @@ def _transaction_rows(repo: Repository, since_ts: int | None) -> list[TimelineRo return rows -def _attestation_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: +def _attestation_rows( + repo: Repository, since_ts: int | None, chain: _Chain +) -> list[TimelineRow]: """The attestation tables -> attestation rows. `human-attested`: someone typed this and signed their name to it, which is a different kind @@ -284,6 +368,7 @@ def _attestation_rows(repo: Repository, since_ts: int | None) -> list[TimelineRo provenance="human-attested", source="asset_attestations", reference=asset, + **chain.of("asset_attestations", asset), summary=( f"{asset} attested by {raw.get('attested_by') or 'unnamed'} " f"(source: {raw.get('source') or 'unstated'})" @@ -306,6 +391,7 @@ def _attestation_rows(repo: Repository, since_ts: int | None) -> list[TimelineRo provenance="human-attested", source="instrument_attestations", reference=f"{venue}:{product}", + **chain.of("instrument_attestations", f"{venue}:{product}"), summary=( f"{product} on {venue} attested by " f"{raw.get('attested_by') or 'unnamed'} " @@ -393,10 +479,20 @@ def gather_timeline( resolved_limit = None if limit is None else max(1, min(int(limit), MAX_TIMELINE_LIMIT)) since = scope_start_ts(resolved_scope, now_ts) + # ONE read of `audit_events`, shared by all three store readers (#721). Not one lookup per + # row: the hash printed beside a row and the verdict printed above it would then describe + # different reads of the table, and an event appended between them would have the verdict + # cover a row this report never showed. + chain = repo.audit_chain() + chained = _Chain(chain) + scoped: list[TimelineRow] = [] - scoped.extend(_order_rows(repo, since)) - scoped.extend(_transaction_rows(repo, since)) - scoped.extend(_attestation_rows(repo, since)) + scoped.extend(_order_rows(repo, since, chained)) + scoped.extend(_transaction_rows(repo, since, chained)) + scoped.extend(_attestation_rows(repo, since, chained)) + # No chain argument, and never one: `_cycle_rows` reads the engine's own log FILE, which is + # not a chained store. A cycle row carrying a hash would be this module attesting to something + # it merely read. scoped.extend(_cycle_rows(cycles, since)) # Newest first, HERE -- so neither renderer sorts, and they cannot disagree about what "the @@ -417,6 +513,8 @@ def gather_timeline( rows=tuple(filtered if resolved_limit is None else filtered[:resolved_limit]), log_status=log_status, kinds_present=tuple(kind for kind in TIMELINE_KINDS if kind in present), + chain_recorded=chain.event_count > 0, + chain_errors=chain.errors, ) @@ -463,13 +561,28 @@ def to_csv(report: TimelineReport) -> str: **Provenance is a column, not a footnote.** The point of this file is that a reader can tell a venue-reported fill from a line someone imported from a spreadsheet -- so `provenance` and - `source` sit beside every figure, and `row_hash` says NOT RECORDED rather than being blank. + `source` sit beside every figure, `row_hash` carries the chain's own hash where one was + recorded and says NOT RECORDED rather than being blank where none was, and `chain_status` + says whether that hash is evidence. All three in the same file, on every row. `amount_kind` rides beside `amount` for the same reason: a fill price and a cash-flow total in one column, with nothing saying which is which, is a column that will be summed. """ buffer = io.StringIO() writer = csv.writer(buffer) + if not report.chain_intact: + # Stated ABOVE the header, for the same reason the log note below is: an auditor holding + # this file cannot ask the page anything, and a finding that changes how the whole file + # should be read must travel with the file. Per-row `chain_status` says WHICH rows; this + # says the record has been altered, which is the sentence that matters first. + writer.writerow( + [ + csv_safe( + f"# NOTE: the audit chain does not verify ({report.chain_errors[0]}); " + "every row marked `chain broken` below is shown but is NOT evidence" + ) + ] + ) if report.log_gap: # Stated IN THE FILE, above the header, because this file leaves the application. An # auditor holding a CSV cannot ask the page whether a source was missing from it, and @@ -494,6 +607,10 @@ def to_csv(report: TimelineReport) -> str: "amount_kind", "summary", "row_hash", + # Beside the hash, never instead of it. A 64-character string with nothing saying + # whether it verifies is a number an auditor cannot use, and the reading a reader + # defaults to is the flattering one. + "chain_status", ] ) for row in report.rows: @@ -509,6 +626,7 @@ def to_csv(report: TimelineReport) -> str: csv_safe(row.amount_kind), csv_safe(row.summary), csv_safe(row.row_hash), + csv_safe(row.chain_status), ] ) return buffer.getvalue() diff --git a/keel/data/audit.py b/keel/data/audit.py new file mode 100644 index 0000000..7550a3f --- /dev/null +++ b/keel/data/audit.py @@ -0,0 +1,371 @@ +"""`audit_events`: the append-only, hash-chained record of what the book did (#721). + +#703's activity export shipped with a `row_hash` column whose every cell read `NOT RECORDED`. +That was honest and it was the whole of what could be said: none of the four stores the timeline +merges hashed its rows. This module is the record that column was built to read. + +── WHY EVENTS AND NOT THE ROWS ──────────────────────────────────────────────────────────────── + +`orders` rows are MUTATED. `update_order` writes status, fills and fees as a venue reports them, +sometimes minutes after placement, sometimes days later via `execution.reconcile`. A hash chained +over the order row itself would therefore break on every legitimate fill -- and a chain that +cries wolf on ordinary operation is a chain an operator learns to ignore, which is worse than no +chain at all. + +So the chain is over IMMUTABLE STATEMENTS about the book, not over the book. `insert_order` +appends "this order was placed, with these fields"; each `update_order` appends "these fields +changed". Neither event is ever rewritten. The book stays mutable and queryable; the event stream +is what an auditor verifies. + +The same shape covers `transactions` (whose upsert rewrites an imported line in place -- two +events for one `coinbase_id` is the record that the line was re-imported, which is precisely what +should be visible) and both attestation tables (where a re-attestation is a fresh human claim). + +── ONE CHAIN, NOT ONE PER TABLE ─────────────────────────────────────────────────────────────── + +Every event goes into one sequence. A per-table chain would let an event be moved between streams +undetectably, and would mean a removed order event was only visible to someone who thought to +verify the order chain specifically. One chain means any removal anywhere is visible from every +event after it. + +── WHAT THIS IS NOT ─────────────────────────────────────────────────────────────────────────── + +**Tamper-EVIDENT, not tamper-proof.** Anyone who can write `keel.db` can rewrite the chain from +the edited row forward. What it buys is that a row cannot be changed QUIETLY. + +**No backfill, ever.** Rows written before this shipped have no event. Computing hashes for them +now would produce a chain that verifies and proves nothing -- worse than an honest gap, because +it looks like evidence. `commands/timeline.py` reports those rows as not chained. + +**Not the research trials ledger.** `keel/research/ledger.py` records EXPERIMENTS and is +git-tracked; this records trading activity and is per-deployment. They share +`keel_core.hashchain` -- one definition of canonical JSON, so one row can only have one hash -- +and nothing else. Folding trials into the trading audit trail to borrow their hashes would be +provenance laundering. +""" + +from __future__ import annotations + +import json +import sqlite3 +from collections.abc import Iterator, Mapping +from contextlib import contextmanager +from dataclasses import dataclass +from typing import Any + +from keel_core.hashchain import ( + ZERO_HASH, + ChainLink, + chain_hash, + encode_value, + find_breaks, + verify_links, +) + +#: Every kind of statement this chain carries, and the store each one is about. +#: +#: A CLOSED vocabulary, checked at write time, like `TIMELINE_KINDS` and `PROVENANCES` next door. +#: An event type that arrives via a typo is a stream nothing reads and nothing misses -- and on +#: an audit surface, a silently-ignored record is the failure mode that matters. +#: +#: The store name is carried because `latest_hashes` is keyed by it: an `orders.id` of `"1"` and +#: a `coinbase_id` of `"1"` are different rows, and keyed by entity alone the imported transaction +#: would hand its hash to an unrelated order. +EVENT_STORES: Mapping[str, str] = { + "order_placed": "orders", + "order_updated": "orders", + "transaction_recorded": "transactions", + "asset_attested": "asset_attestations", + "instrument_attested": "instrument_attestations", +} + + +@dataclass(frozen=True) +class AuditEvent: + """One immutable statement about the book, and its place in the chain.""" + + seq_id: int + ts: int + event_type: str + #: The subject store's own identifier -- an `orders.id`, a `coinbase_id`, an asset, a + #: `venue:product_id`. As TEXT, because the five stores key on four different types. + entity_id: str + #: What the statement asserts. Decimals are already strings here: they were encoded on the + #: way in (`keel_core.hashchain.encode_value`) so that what is hashed is exactly what is + #: stored, and a reader is never handed a float rendering of a recorded money value. + payload: dict[str, Any] + prev_hash: str + row_hash: str + + +@contextmanager +def write_transaction(conn: sqlite3.Connection) -> Iterator[None]: + """One write transaction covering a store row and its audit event. + + **`BEGIN IMMEDIATE`, not `BEGIN`.** A deferred transaction takes no lock until its first + WRITE, so the chain-head SELECT inside it would still be unserialised: two writers would both + read the same head, both write it as their `prev_hash`, and the chain would fork -- silently, + because each row verifies against the row it believes precedes it. `IMMEDIATE` takes the + write lock up front, which is what makes "read the head, then append" atomic. + + Nested calls JOIN the caller's transaction rather than opening a second one: sqlite raises + "cannot start a transaction within a transaction", and more importantly a nested commit would + publish the outer writer's half-finished work. The outermost block owns the commit. + + Rolls back on ANY exception, `BaseException` included: a `KeyboardInterrupt` between the + store row and its event would otherwise leave the row committed and the chain silent about + it, which is the one lie this store must not tell about itself. + """ + if conn.in_transaction: + yield + return + conn.execute("BEGIN IMMEDIATE") + try: + yield + except BaseException: + conn.rollback() + raise + conn.commit() + + +def append_event( + conn: sqlite3.Connection, + *, + ts: int, + event_type: str, + entity_id: str, + payload: Mapping[str, Any], +) -> AuditEvent: + """Chain and append one event. Does NOT commit -- the caller's `write_transaction` does. + + Not committing is the point: the store row and its event must land together or not at all. + An order row with no event reads forever after as "written before the bump" -- an + honest-looking gap that is in fact a failed chain write. + + The head read is inside the caller's write transaction and refuses to run outside one, which + is the serialisation this chain rests on. `ORDER BY seq_id DESC`, never `ts`: two events + inside one second share a timestamp, and a clock that steps backwards would silently reorder + the chain into a false break. + """ + if event_type not in EVENT_STORES: + raise ValueError(f"event_type: {event_type!r} not in {sorted(EVENT_STORES)}") + if not conn.in_transaction: + raise RuntimeError( + "append_event must run inside a write transaction (see `write_transaction`): " + "reading the chain head outside one lets two writers fork the chain" + ) + + head = conn.execute("SELECT row_hash FROM audit_events ORDER BY seq_id DESC LIMIT 1").fetchone() + prev_hash = ZERO_HASH if head is None else str(head["row_hash"]) + # Encoded BEFORE hashing and stored in the encoded form, so what is hashed is byte-for-byte + # what a reader gets back. Encoding at read time instead would put a conversion between the + # stored row and the hash, and every such conversion is somewhere the two can drift. + encoded: dict[str, Any] = encode_value(dict(payload)) + body = { + "ts": ts, + "event_type": event_type, + "entity_id": entity_id, + "payload": encoded, + "prev_hash": prev_hash, + } + row_hash = chain_hash(body) + cursor = conn.execute( + "INSERT INTO audit_events (ts, event_type, entity_id, payload_json, prev_hash, row_hash) " + "VALUES (?, ?, ?, ?, ?, ?)", + ( + ts, + event_type, + entity_id, + json.dumps(encoded, sort_keys=True, separators=(",", ":")), + prev_hash, + row_hash, + ), + ) + assert cursor.lastrowid is not None + return AuditEvent( + seq_id=cursor.lastrowid, + ts=ts, + event_type=event_type, + entity_id=entity_id, + payload=encoded, + prev_hash=prev_hash, + row_hash=row_hash, + ) + + +def read_events(conn: sqlite3.Connection) -> list[AuditEvent]: + """Every event, in CHAIN order (`seq_id`), which is the only order the chain verifies in.""" + rows = conn.execute( + "SELECT seq_id, ts, event_type, entity_id, payload_json, prev_hash, row_hash " + "FROM audit_events ORDER BY seq_id" + ).fetchall() + return [ + AuditEvent( + seq_id=int(row["seq_id"]), + ts=int(row["ts"]), + event_type=str(row["event_type"]), + entity_id=str(row["entity_id"]), + payload=json.loads(row["payload_json"]), + prev_hash=str(row["prev_hash"]), + row_hash=str(row["row_hash"]), + ) + for row in rows + ] + + +def _event_payload(event: AuditEvent) -> dict[str, Any]: + """Everything the row commits to -- the row minus `row_hash` itself. + + `seq_id` is deliberately NOT hashed. It is sqlite's own AUTOINCREMENT counter, assigned after + the hash would have to be computed, and position in the chain is already committed to via + `prev_hash`. Hashing it would add nothing and would make the hash unreproducible from the + values a writer actually chose. + """ + return { + "ts": event.ts, + "event_type": event.event_type, + "entity_id": event.entity_id, + "payload": event.payload, + "prev_hash": event.prev_hash, + } + + +def verify_events(events: list[AuditEvent]) -> list[str]: + """Every break in the chain, as human-readable lines. Empty means intact. + + Reports rather than raises, so `keel doctor` and the timeline export can both STATE chain + status instead of asserting it. **An empty list of events returns no errors, and that is not + the same as "verified"** -- nothing was checked. Only the caller knows whether that means a + deployment predating #721 or a deployment that has done nothing yet. + """ + return verify_links( + ChainLink( + label=str(event.seq_id), + prev_hash=event.prev_hash, + row_hash=event.row_hash, + recomputed_hash=chain_hash(_event_payload(event)), + ) + for event in events + ) + + +def latest_hashes(conn: sqlite3.Connection) -> dict[tuple[str, str], str]: + """`(store, entity_id) -> the row_hash of the LATEST event about it`. See `chain_state`.""" + return {key: seen.row_hash for key, seen in _latest(read_events(conn)).items()} + + +@dataclass(frozen=True) +class EntityHash: + """The latest chained statement about one row, and where it sits in the chain.""" + + row_hash: str + #: The event's own `seq_id`. Carried so a caller can ask whether this statement falls in the + #: region the chain no longer vouches for -- see `ChainState.first_broken_seq`. + seq_id: int + + +@dataclass(frozen=True) +class ChainState: + """Everything a report needs to STATE the chain's condition rather than assert it. + + Read in one pass, because the alternative -- a hash lookup here, a verification there -- means + the hash printed beside a row and the verdict printed above it describe two different reads of + the table, and a row appended between them would have the verdict cover a row the report never + showed. That is the mistake `research/ledger.py::verify_records` exists to have fixed once. + """ + + #: Whether `audit_events` EXISTS on this database (schema v20). + #: + #: Not a theoretical case. Two readers open this database without migrating it -- the web + #: server (`keel/web/server.py`: a view must not take a schema write lock) and `keel mcp`'s + #: `_open_readonly_repo`. Both therefore meet pre-v20 databases, and a missing table there is + #: an ordinary fact about an un-upgraded deployment rather than an error. #718 shipped a + #: reader that raised on exactly this and took the whole of `gather_findings` down with it. + table_present: bool + #: How many events the table holds. ZERO IS A DISTINCT STATE, not "verified": a deployment + #: that predates #721 and one that has done nothing since are both empty here, and neither has + #: had anything checked. The caller decides which sentence to print. + event_count: int + errors: tuple[str, ...] + #: The `seq_id` of the FIRST event the chain stops vouching for, or `None` when intact. Every + #: event from here on is unverified -- not because each one is necessarily wrong, but because + #: a chain proves the sequence, and past a break the sequence is no longer proven. + first_broken_seq: int | None + hashes: Mapping[tuple[str, str], EntityHash] + + @property + def intact(self) -> bool: + """The chain verifies AND there is something to verify. + + Both halves, deliberately. `not errors` alone is true of an empty table, and a green + badge over a table nothing read is the exact failure this codebase keeps re-learning. + """ + return not self.errors and self.event_count > 0 + + +def _latest(events: list[AuditEvent]) -> dict[tuple[str, str], EntityHash]: + """`(store, entity_id) -> the LATEST event about it`. + + The latest rather than the first because it is the most recent chained statement about that + row: an order that was placed and then filled has two events, and the fill is the later word + on it. + + Keyed by store as well as entity because the five stores key on four different types and their + identifiers collide -- an `orders.id` of `"1"` and a `coinbase_id` of `"1"` are different + rows, and a single-keyed map would hand one row's hash to the other. + """ + latest: dict[tuple[str, str], EntityHash] = {} + for event in events: + store = EVENT_STORES.get(event.event_type) + if store is None: + # An event type this build does not know -- a row written by a NEWER keel against the + # same database. Skipped rather than guessed at: it still chains (the chain does not + # care what the type means), and inventing a store for it would file it against a + # table it may have nothing to do with. + continue + latest[(store, event.entity_id)] = EntityHash(row_hash=event.row_hash, seq_id=event.seq_id) + return latest + + +def _table_present(conn: sqlite3.Connection) -> bool: + row = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'audit_events'" + ).fetchone() + return row is not None + + +def chain_state(conn: sqlite3.Connection) -> ChainState: + """One read of `audit_events`, verified, indexed by the identifiers a report prints. + + What `commands/timeline.py` calls. The `hashes` map is keyed by `(store, entity_id)` to match + that module's own `source`/`reference` pair, so the hash shown against a row is looked up by + the identifier printed beside it rather than by one a reader has to reconstruct. + """ + if not _table_present(conn): + # Checked rather than caught. A pre-v20 database is a legitimate deployment state, and a + # `try`/`except OperationalError` around the read would swallow a genuine "database disk + # image is malformed" under the same clause -- reporting an un-upgraded database where + # the truth is a corrupt one. + return ChainState( + table_present=False, event_count=0, errors=(), first_broken_seq=None, hashes={} + ) + events = read_events(conn) + breaks = find_breaks( + ChainLink( + label=str(event.seq_id), + prev_hash=event.prev_hash, + row_hash=event.row_hash, + recomputed_hash=chain_hash(_event_payload(event)), + ) + for event in events + ) + # `index` is 1-based POSITION IN THE WALK, and `events` is read in `seq_id` order, so this is + # the event the first break lands on. Looked up rather than parsed out of the message text. + first_broken_seq = events[breaks[0].index - 1].seq_id if breaks else None + return ChainState( + table_present=True, + event_count=len(events), + errors=tuple(found.message for found in breaks), + first_broken_seq=first_broken_seq, + hashes=_latest(events), + ) diff --git a/keel/data/repository.py b/keel/data/repository.py index 337166e..58d3de7 100644 --- a/keel/data/repository.py +++ b/keel/data/repository.py @@ -18,6 +18,7 @@ from keel_core.subscription import BrokerSubscription, SubscriptionStatus from keel_core.trade_scope import TradeScopeState, VenueTradeScope +from keel.data.audit import ChainState, append_event, chain_state, write_transaction from keel.types import Candle, CycleBalance, EquityReading, Granularity, Profile _TRANSACTION_COLUMNS = ( @@ -211,6 +212,17 @@ class Repository: def __init__(self, conn: sqlite3.Connection) -> None: self._conn = conn + # -- the audit chain ------------------------------------------------ + + def audit_chain(self) -> ChainState: + """The `audit_events` chain: its hashes, its condition, and where it stops being evidence. + + Exposed here rather than letting a reader reach for the connection, because "which rows + does this chain vouch for" is a question about the record and the repository is what owns + the record. `commands/timeline.py` is the caller. + """ + return chain_state(self._conn) + # -- transactions --------------------------------------------------- def upsert_transaction(self, tx: dict[str, Any]) -> None: @@ -227,15 +239,32 @@ def upsert_transaction(self, tx: dict[str, Any]) -> None: update_sql = ", ".join( f"{c} = excluded.{c}" for c in _TRANSACTION_COLUMNS if c != "coinbase_id" ) - self._conn.execute( - f""" - INSERT INTO transactions ({columns_sql}) - VALUES ({placeholders_sql}) - ON CONFLICT(coinbase_id) DO UPDATE SET {update_sql} - """, - values, - ) - self._conn.commit() + with write_transaction(self._conn): + cursor = self._conn.execute( + f""" + INSERT INTO transactions ({columns_sql}) + VALUES ({placeholders_sql}) + ON CONFLICT(coinbase_id) DO UPDATE SET {update_sql} + """, + values, + ) + # `coinbase_id or id`, which is EXACTLY the rule `commands/timeline.py:: + # _transaction_rows` uses for the `reference` it prints -- so the export looks the + # hash up by the identifier shown beside it. `coinbase_id` is nullable, and sqlite + # treats NULLs as distinct in a UNIQUE index, so a row without one can never take the + # DO UPDATE branch: `lastrowid` is a real insert's id in the only case that reads it. + coinbase_id = values.get("coinbase_id") + # An APPEND even when the book row was updated in place (#721). Two events for one + # `coinbase_id` is the record that the line was re-imported with different content -- + # which, for a store whose provenance is `imported-ledger` and whose rows nothing + # verified on the way in, is exactly what an auditor needs visible. + append_event( + self._conn, + ts=int(values.get("ts") or time.time()), + event_type="transaction_recorded", + entity_id=str(coinbase_id) if coinbase_id else str(cursor.lastrowid), + payload=dict(values), + ) def get_transactions(self, asset: str | None = None) -> list[dict[str, Any]]: if asset is None: @@ -405,12 +434,31 @@ def insert_order(self, order: dict[str, Any]) -> int: columns_sql = ", ".join(_ORDER_COLUMNS) placeholders_sql = ", ".join(f":{c}" for c in _ORDER_COLUMNS) - cursor = self._conn.execute( - f"INSERT INTO orders ({columns_sql}) VALUES ({placeholders_sql})", values - ) - self._conn.commit() - assert cursor.lastrowid is not None - return cursor.lastrowid + # The row and its audit event in ONE transaction (#721). Both land or neither: an order + # row with no event reads forever after as "written before the chain shipped" -- an + # honest-looking gap that would in fact be a failed chain write, which is the one lie + # this record must not tell about itself. + # + # This runs BEFORE `broker.place_order` (see `executor.execute`), so a failure here fails + # CLOSED: no row, no event, no order at the venue. That ordering is why the audit write + # is allowed to be load-bearing here where a diagnostic write would not be. + with write_transaction(self._conn): + cursor = self._conn.execute( + f"INSERT INTO orders ({columns_sql}) VALUES ({placeholders_sql})", values + ) + assert cursor.lastrowid is not None + order_id = cursor.lastrowid + append_event( + self._conn, + ts=int(values.get("created_at") or time.time()), + event_type="order_placed", + entity_id=str(order_id), + # The row AS STORED, id included -- the whole statement being made, so a later + # reader can reproduce the hash from the row without knowing which columns this + # build happened to populate. + payload={"id": order_id, **values}, + ) + return order_id def update_order(self, order_id: int, **fields: Any) -> None: """Partially update the order row `order_id` with `fields`.""" @@ -423,8 +471,19 @@ def update_order(self, order_id: int, **fields: Any) -> None: set_sql = ", ".join(f"{k} = :{k}" for k in fields) params = dict(fields) params["order_id"] = order_id - self._conn.execute(f"UPDATE orders SET {set_sql} WHERE id = :order_id", params) - self._conn.commit() + with write_transaction(self._conn): + self._conn.execute(f"UPDATE orders SET {set_sql} WHERE id = :order_id", params) + append_event( + self._conn, + ts=int(fields.get("updated_at") or time.time()), + event_type="order_updated", + entity_id=str(order_id), + # WHAT CHANGED, not the whole mutated row. An event is a statement about this + # update; re-hashing the full row would make every event a snapshot, and a reader + # could no longer tell an update from a rewrite. The empty-`fields` early return + # above is what keeps a no-op from appending an event asserting nothing changed. + payload=dict(fields), + ) def get_order(self, order_id: int) -> dict[str, Any] | None: row = self._conn.execute("SELECT * FROM orders WHERE id = ?", (order_id,)).fetchone() @@ -1278,33 +1337,53 @@ def upsert_asset_attestation( before". Omitting it there would let a stale window silently outlive the claim it was recorded for, which is worse than no window at all. """ - self._conn.execute( - """ - INSERT INTO asset_attestations - (asset, sector, backing, pays_yield, source, attested_by, attested_at, - attest_due_ts) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(asset) DO UPDATE SET - sector = excluded.sector, - backing = excluded.backing, - pays_yield = excluded.pays_yield, - source = excluded.source, - attested_by = excluded.attested_by, - attested_at = excluded.attested_at, - attest_due_ts = excluded.attest_due_ts - """, - ( - asset, - sector, - backing, - int(pays_yield), - source, - attested_by, - attested_at, - attest_due_ts, - ), - ) - self._conn.commit() + with write_transaction(self._conn): + self._conn.execute( + """ + INSERT INTO asset_attestations + (asset, sector, backing, pays_yield, source, attested_by, attested_at, + attest_due_ts) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(asset) DO UPDATE SET + sector = excluded.sector, + backing = excluded.backing, + pays_yield = excluded.pays_yield, + source = excluded.source, + attested_by = excluded.attested_by, + attested_at = excluded.attested_at, + attest_due_ts = excluded.attest_due_ts + """, + ( + asset, + sector, + backing, + int(pays_yield), + source, + attested_by, + attested_at, + attest_due_ts, + ), + ) + # Appended, not rewritten (#721): a re-attestation is a FRESH human claim, and the + # claim it replaced is part of the record. `attested_at` is the timestamp the human + # supplied for the claim, so the event is stamped with when the claim was made rather + # than when the row happened to be written. + append_event( + self._conn, + ts=attested_at, + event_type="asset_attested", + entity_id=asset, + payload={ + "asset": asset, + "sector": sector, + "backing": backing, + "pays_yield": bool(pays_yield), + "source": source, + "attested_by": attested_by, + "attested_at": attested_at, + "attest_due_ts": attest_due_ts, + }, + ) def get_asset_attestation(self, asset: str) -> dict | None: row = self._conn.execute( @@ -1337,21 +1416,40 @@ def upsert_instrument_attestation( `upsert_asset_attestation` above -- it must be in `DO UPDATE SET` so a re-attestation without a window CLEARS a previously recorded one rather than carrying it forward. """ - self._conn.execute( - """ - INSERT INTO instrument_attestations - (venue, product_id, wrapper, source, attested_by, attested_at, attest_due_ts) - VALUES (?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(venue, product_id) DO UPDATE SET - wrapper = excluded.wrapper, - source = excluded.source, - attested_by = excluded.attested_by, - attested_at = excluded.attested_at, - attest_due_ts = excluded.attest_due_ts - """, - (venue, product_id, wrapper, source, attested_by, attested_at, attest_due_ts), - ) - self._conn.commit() + with write_transaction(self._conn): + self._conn.execute( + """ + INSERT INTO instrument_attestations + (venue, product_id, wrapper, source, attested_by, attested_at, attest_due_ts) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(venue, product_id) DO UPDATE SET + wrapper = excluded.wrapper, + source = excluded.source, + attested_by = excluded.attested_by, + attested_at = excluded.attested_at, + attest_due_ts = excluded.attest_due_ts + """, + (venue, product_id, wrapper, source, attested_by, attested_at, attest_due_ts), + ) + # `venue:product_id`, the same compound key this table is UNIQUE on and the same + # string `commands/timeline.py` puts in its `reference` column -- so the hash the + # export shows against an instrument attestation is looked up by the identifier + # printed beside it, not by one a reader has to reconstruct. + append_event( + self._conn, + ts=attested_at, + event_type="instrument_attested", + entity_id=f"{venue}:{product_id}", + payload={ + "venue": venue, + "product_id": product_id, + "wrapper": wrapper, + "source": source, + "attested_by": attested_by, + "attested_at": attested_at, + "attest_due_ts": attest_due_ts, + }, + ) def get_instrument_attestation(self, venue: str, product_id: str) -> dict | None: row = self._conn.execute( diff --git a/keel/research/ledger.py b/keel/research/ledger.py index 39d2188..7c0c896 100644 --- a/keel/research/ledger.py +++ b/keel/research/ledger.py @@ -13,7 +13,6 @@ from __future__ import annotations -import hashlib import json import time from collections.abc import Iterable, Mapping, Sequence @@ -22,6 +21,18 @@ from pathlib import Path from typing import Any +# The canonical form and the chain walk are SHARED with `keel/data/audit.py` (#721) rather +# than defined here. `canonical_json` and `ZERO_HASH` keep their names in this module's +# namespace because `tests/research/test_ledger.py` and every downstream reader reach for +# `ledger.canonical_json` -- the definition moved, the vocabulary did not. +from keel_core.hashchain import ( + ZERO_HASH, + ChainLink, + canonical_json, + chain_hash, + verify_links, +) + PROVENANCE = frozenset({"a_priori", "fitted"}) #: `monte_carlo` (#441) is a resampling DIAGNOSTIC row -- same vocabulary discipline as the #: rest of the set: a row's kind says what kind of experiment produced it, so a bootstrap run @@ -44,7 +55,6 @@ #: A CSCV column is a diagnostic, not a decision (spec §4.4) -- it does not count toward N. DIAGNOSTIC_ONLY = "diagnostic_only" -ZERO_HASH = "0" * 64 DEFAULT_LEDGER_PATH = Path("docs/experiments/trials-ledger.jsonl") @@ -67,17 +77,6 @@ class TrialRecord: row_hash: str = "" -def _encode(value: Any) -> Any: - """Decimal -> str so JSON round-trips exactly; recurse through containers.""" - if isinstance(value, Decimal): - return str(value) - if isinstance(value, Mapping): - return {k: _encode(v) for k, v in value.items()} - if isinstance(value, (list, tuple)): - return [_encode(v) for v in value] - return value - - def _decode_series(raw: Any) -> list[Decimal]: return [Decimal(v) for v in (raw or [])] @@ -98,15 +97,6 @@ def _decode_summary(raw: Any) -> dict[str, Any]: return out -def canonical_json(payload: Mapping[str, Any]) -> str: - """Deterministic serialisation: sorted keys, no incidental whitespace. - - The hash is only reproducible if this is byte-stable, so both the separators and the key - ordering are pinned here rather than left to `json.dumps` defaults. - """ - return json.dumps(_encode(dict(payload)), sort_keys=True, separators=(",", ":")) - - def _row_payload(record: TrialRecord) -> dict[str, Any]: """Everything that is hashed -- i.e. the row minus `row_hash` itself.""" return { @@ -127,7 +117,15 @@ def _row_payload(record: TrialRecord) -> dict[str, Any]: def compute_row_hash(record: TrialRecord) -> str: - return hashlib.sha256(canonical_json(_row_payload(record)).encode("utf-8")).hexdigest() + """This row's hash, over `_row_payload` -- i.e. the row minus `row_hash` itself. + + The arithmetic moved to `keel_core.hashchain.chain_hash` in #721 so `audit_events` could + share it. WHAT is hashed stayed here, because only this module knows which of a trial's + fields are part of the record. The 93 rows in the git-tracked ledger were hashed before that + move and still verify byte-for-byte -- pinned by + `test_the_tracked_ledger_still_verifies_after_the_canonicaliser_moved`. + """ + return chain_hash(_row_payload(record)) def _validate(record: TrialRecord) -> None: @@ -257,18 +255,15 @@ def verify_records(records: Iterable[TrialRecord]) -> list[str]: it is held for the web view, and it is the difference between an honest badge and a green light over a file nothing read. """ - errors: list[str] = [] - expected_prev = ZERO_HASH - for index, record in enumerate(records, start=1): - if record.prev_hash != expected_prev: - errors.append( - f"row {index} ({record.trial_id}): prev_hash {record.prev_hash[:12]}... " - f"does not chain to {expected_prev[:12]}..." - ) - elif compute_row_hash(record) != record.row_hash: - errors.append(f"row {index} ({record.trial_id}): content does not match row_hash") - expected_prev = record.row_hash - return errors + return verify_links( + ChainLink( + label=record.trial_id, + prev_hash=record.prev_hash, + row_hash=record.row_hash, + recomputed_hash=compute_row_hash(record), + ) + for record in records + ) def trial_counts(trials: Iterable[TrialRecord]) -> tuple[int, int]: diff --git a/keel/web/payload.py b/keel/web/payload.py index e7f3eb8..639b788 100644 --- a/keel/web/payload.py +++ b/keel/web/payload.py @@ -1795,6 +1795,36 @@ def balances_payload(report: BalancesReport) -> dict[str, Any]: } +#: What the chain says about one row, as a state a client can style (#721). `chained` is the only +#: green: `not chained` is an honest gap that nobody checked, and `chain broken` is the row whose +#: hash is shown and is NOT evidence. Nothing here infers a state from whether a hash is present. +_TIMELINE_CHAIN_STATES: Mapping[str, str] = { + "chained": GOOD, + "not chained": UNKNOWN, + "chain broken": BAD, +} + +#: What each status MEANS, spelled out beside the word, for the same reason `_PROVENANCE_NOTES` +#: above exists: on an audit surface the term of art is not the thing a reader can act on. +_TIMELINE_CHAIN_NOTES: Mapping[str, str] = { + "chained": "recorded in the audit chain, and the chain verifies", + "not chained": "written before the audit chain, or from a source that has none — unchecked", + "chain broken": "the chain does not verify from here on; this hash is not evidence", +} + + +def _timeline_hash_display(row: TimelineRow) -> str | None: + """A row hash, shortened for a table cell -- or the NOT-RECORDED words, unshortened. + + `HASH_NOT_RECORDED` is a SENTENCE, not a hash, and truncating it to twelve characters and an + ellipsis would turn "NOT RECORDED" into something that looks like a short hash in a column of + long ones. The status carries the distinction, and the display must not fight it. + """ + if row.chain_status == "not chained": + return row.row_hash + return _hash_display(row.row_hash) + + def _timeline_row_payload(row: TimelineRow) -> dict[str, Any]: """One event, placed. @@ -1822,11 +1852,59 @@ def _timeline_row_payload(row: TimelineRow) -> dict[str, Any]: "amount_kind": row.amount_kind, "summary": row.summary, # A `label`, so the "we did not check" reading carries a state a client can style rather - # than a bare string it might render as though it were a hash. - "row_hash": label(row.row_hash, state=UNKNOWN), + # than a bare string it might render as though it were a hash. Shortened for a table cell + # -- the full value is in the CSV export, which is the artefact anyone actually verifying + # a hash would be working from. + "row_hash": label( + row.row_hash, + display=_timeline_hash_display(row), + state=_TIMELINE_CHAIN_STATES.get(row.chain_status, UNKNOWN), + ), + # BESIDE the hash, never derived from it. "A hash is present" and "the chain vouches for + # it" are different facts, and a client inferring the second from the first would call a + # row inside a broken region verified. Rule 3: the judgement is made here. + "chain_status": label( + row.chain_status, + display=_TIMELINE_CHAIN_NOTES.get(row.chain_status, row.chain_status), + state=_TIMELINE_CHAIN_STATES.get(row.chain_status, UNKNOWN), + ), } +def _timeline_chain_payload(report: TimelineReport) -> Field: + """The audit chain's verdict, as THREE states -- the trading-side twin of `_chain_payload`. + + Three rather than that function's four because this store cannot be ABSENT: `audit_events` is + a table in the same database the report was read from, so "no file" has no analogue here. The + other three are the same distinctions and are the same trap: + + * **no events** -- a deployment that predates #721, or one that has written nothing since. + Not a verification: an empty chain reports no breaks because there is nothing in it to + break, and calling that "verified" is a positive claim about zero rows. + * **events, no errors** -- the one case that is green. + * **a break** -- the record has been altered, and every row from there on is unproven. + + GOOD requires BOTH: no errors, over events that actually exist. + """ + if not report.chain_recorded: + return label( + "unverified", + display="no audit chain recorded yet — nothing here was checked", + state=UNKNOWN, + ) + if report.chain_intact: + return label( + "intact", + display="audit chain verified — every recorded event still hashes to the next", + state=GOOD, + ) + return label( + "broken", + display="the audit chain does not verify — the rows below say from where", + state=BAD, + ) + + def timeline_payload(report: TimelineReport) -> dict[str, Any]: """`gather_timeline`'s `TimelineReport`, as JSON (#703). @@ -1845,6 +1923,11 @@ def timeline_payload(report: TimelineReport) -> dict[str, Any]: "scoped_count": count(report.scoped_count), "filtered_count": count(report.filtered_count), "shown_count": count(report.shown_count), + # The verdict over the WHOLE chain, beside the per-row statuses -- the same pairing + # `_chain_payload` holds for the research ledger, and for the same reason: an empty chain + # has no breaks and has verified nothing, so "no errors" is never on its own a green light. + "chain": _timeline_chain_payload(report), + "chain_errors": [str(error) for error in report.chain_errors], "rows": [_timeline_row_payload(row) for row in report.rows], } diff --git a/keel/web/static/js/render.js b/keel/web/static/js/render.js index 8fb6e82..7b5fd29 100644 --- a/keel/web/static/js/render.js +++ b/keel/web/static/js/render.js @@ -1203,9 +1203,16 @@ export function timelineView(data, sort, onSort, onKind) { link.setAttribute("href", target.pathname.concat(target.search)); link.setAttribute("download", ""); actions.append(link); - actions.append(" — every row with its provenance; hashes are not recorded yet."); + actions.append(" — every row with its provenance, its hash and whether the chain verifies it."); fragment.append(actions); + // The chain's verdict over the whole record, above the rows it applies to. A per-row status + // answers "is this row evidence"; this answers "has this record been altered", which is the + // question an operator opening an audit page is actually asking. + const chain = el("p", "note"); + chain.append(field(data.chain)); + fragment.append(chain); + fragment.append(heading("h-timeline", "What happened")); fragment.append( table( @@ -1222,6 +1229,10 @@ export function timelineView(data, sort, onSort, onKind) { { label: "product", numeric: false, key: "product_id" }, { label: "amount", numeric: true }, { label: "what", numeric: false }, + // Both columns, never one. A hash with nothing saying whether it verifies is a number a + // reader takes on trust, and the reading taken on trust is the flattering one. + { label: "row hash", numeric: false }, + { label: "tamper-evidence", numeric: false }, ], (data.rows || []).map(/** @param {any} row */ (row) => [ row.at, @@ -1232,6 +1243,8 @@ export function timelineView(data, sort, onSort, onKind) { plain(row.product_id) || "—", row.amount, plain(row.summary) || "—", + row.row_hash, + row.chain_status, ]), "Nothing recorded in this window.", { sort: sort, onSort: onSort }, diff --git a/packages/keel-core/keel_core/hashchain.py b/packages/keel-core/keel_core/hashchain.py new file mode 100644 index 0000000..5e8c944 --- /dev/null +++ b/packages/keel-core/keel_core/hashchain.py @@ -0,0 +1,185 @@ +"""One canonical form and one chain walk, for every hash-chained store in this codebase (#721). + +Two stores chain their rows: the research trials ledger (`keel/research/ledger.py` -- JSONL, git +tracked, records experiments) and `audit_events` (`keel/data/audit.py` -- SQLite, per deployment, +records trading activity). They are deliberately separate stores over separate domains, and +blending them would be provenance laundering. What they must NOT have separately is a definition +of what a row's hash is over. + +**Why this is a shared module and not a copied function.** A hash is only evidence if it can be +recomputed. Two canonicalisations that disagree -- one emitting `{"a": 1}` and the other +`{"a":1}`, one rendering `Decimal("1.0")` as `1.0` and the other as `"1.0"` -- produce two hashes +for one row. That disagreement is invisible at write time and surfaces months later as a chain +that "cannot be verified", at exactly the moment someone is trying to establish whether a record +was altered. The form is decided here, once. + +**Tamper-EVIDENT, not tamper-proof.** Anyone who can write the store can rewrite the whole chain +from the edited row forward. What the chain buys is that a row cannot be changed QUIETLY: an +edit invalidates that row and every row after it, so a partial edit is visible and a full rewrite +requires touching every subsequent row. + +This module holds no I/O, no `keel` imports and no schema. It is pure so both stores can depend +on it without depending on each other. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from decimal import Decimal +from typing import Any + +#: What the FIRST row of a chain commits to. A chain anchored anywhere else starts mid-air: it +#: says nothing about whether rows were removed from in front of it, which is the one deletion a +#: chain would otherwise miss entirely. +ZERO_HASH = "0" * 64 + + +def encode_value(value: Any) -> Any: + """`Decimal` -> `str`, recursively, so JSON round-trips a money value exactly. + + Money is TEXT in every store here (`orders.qty`, `equity_points.equity`), and the reason is + the same reason it is a string in the hash: a float rendering of `Decimal("0.10")` is a + different number from the one the row holds, and a hash over a lossy rendering attests to a + value that was never recorded. + + SCALE IS PRESERVED, and deliberately: `Decimal("1.0")` and `Decimal("1.00")` compare equal + numerically and are different recorded values. A store that rewrote a row's scale changed + that row, and the chain says so rather than shrugging. + """ + if isinstance(value, Decimal): + return str(value) + if isinstance(value, Mapping): + return {k: encode_value(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [encode_value(v) for v in value] + return value + + +def canonical_json(payload: Mapping[str, Any]) -> str: + """Deterministic serialisation: sorted keys, no incidental whitespace, Decimals as strings. + + Both the separators and the key ordering are pinned here rather than left to `json.dumps` + defaults, because the hash is only reproducible if this is byte-stable. `sort_keys` in + particular is not a tidiness preference: the same row is assembled from a `dict` literal in + one module and from `dict(sqlite3.Row)` in another, and insertion order differs between them. + + An ABSENT key is not a null key. `{"a": 1}` and `{"a": 1, "b": None}` hash differently, which + is correct -- a writer that started emitting an explicit null for a field it used to omit has + changed what the row says. + """ + return json.dumps(encode_value(dict(payload)), sort_keys=True, separators=(",", ":")) + + +def chain_hash(payload: Mapping[str, Any]) -> str: + """SHA-256 over `canonical_json(payload)`. + + `payload` is everything the row commits to INCLUDING its `prev_hash` and excluding its own + `row_hash`. Including `prev_hash` is what makes this a chain rather than a column of + independent checksums: without it, a row could be moved, duplicated or reordered freely. + """ + return hashlib.sha256(canonical_json(payload).encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class ChainLink: + """One row, reduced to the four things a chain walk needs. + + Deliberately not a row: this module knows nothing about trials or audit events, and each + store keeps its own record type, its own payload shape and its own reader. `recomputed_hash` + is computed BY THE CALLER, because only the caller knows which of its fields are hashed -- + which is the fact that makes the check meaningful, and the fact this module must not guess. + + `label` is whatever identifies the row to a human reading a report -- a `trial_id`, a + `seq_id`. It appears in the error text and nowhere else. + """ + + label: str + prev_hash: str + row_hash: str + recomputed_hash: str + + +@dataclass(frozen=True) +class ChainBreak: + """One place the chain fails, located as well as described. + + `index` is 1-based POSITION IN THE WALK, not any store's own identifier: the walk is the only + thing that knows where a row sits in the chain, and a store whose rows were renumbered would + otherwise report breaks at coordinates that no longer mean anything. Callers that need their + own key look it up by this index -- see `keel/data/audit.py::chain_state`, which uses it to + find the `seq_id` from which the chain stops being evidence. + + `reason` is one of `link` (a row was inserted, removed or reordered) or `content` (a row was + edited in place). Two different accusations, and a report that blurred them would tell an + auditor to look for the wrong thing. + """ + + index: int + label: str + reason: str + message: str + + +def find_breaks(links: Iterable[ChainLink]) -> list[ChainBreak]: + """Every break in the chain, located. Empty means intact. + + **Reports rather than raises.** A broken chain is a finding to surface in a report -- `keel + doctor`, the timeline export, the research page's badge -- and the caller wants every break + rather than only the first. A check that stopped at the first would let a second edit hide + behind the first, which is the shape an auditor most needs to see. + + Two distinct failures, and the `elif` keeps them from doubling up: a row whose `prev_hash` + does not match its predecessor's `row_hash` is a LINK failure, and a row whose content does + not reproduce its own `row_hash` is a CONTENT failure. A row that fails the link check is + reported once, on the link, because its content hash is then computed over a payload the walk + already knows is in the wrong place. + + The walk continues from the row's STORED `row_hash` rather than the recomputed one, so a + single edit produces one error rather than cascading into a false break at every later row. + A DELETED row is what breaks every row after it, and that is the property these stores exist + for. + + **An empty sequence returns no breaks, and that is not the same as "verified".** Nothing was + checked. The distinction belongs to the caller, because only the caller knows whether empty + means "no rows yet" or "no store at all" -- the difference between an honest badge and a + green light over a file nothing read. + """ + breaks: list[ChainBreak] = [] + expected_prev = ZERO_HASH + for index, link in enumerate(links, start=1): + if link.prev_hash != expected_prev: + breaks.append( + ChainBreak( + index=index, + label=link.label, + reason="link", + message=( + f"row {index} ({link.label}): prev_hash {link.prev_hash[:12]}... " + f"does not chain to {expected_prev[:12]}..." + ), + ) + ) + elif link.recomputed_hash != link.row_hash: + breaks.append( + ChainBreak( + index=index, + label=link.label, + reason="content", + message=f"row {index} ({link.label}): content does not match row_hash", + ) + ) + expected_prev = link.row_hash + return breaks + + +def verify_links(links: Iterable[ChainLink]) -> list[str]: + """`find_breaks`, as the human-readable lines a report prints. Empty means intact. + + A formatter over the one walk rather than a second walk: a chain check that existed twice is + a chain check that can disagree with itself, and the disagreement would be between "the + report says intact" and "the badge says broken" over the same rows. + """ + return [found.message for found in find_breaks(links)] diff --git a/tests/commands/test_doctor.py b/tests/commands/test_doctor.py index 0e65c3f..f20e584 100644 --- a/tests/commands/test_doctor.py +++ b/tests/commands/test_doctor.py @@ -598,6 +598,7 @@ def test_gather_findings_covers_every_check_over_a_seeded_db(tmp_path, valid_con "attest.withdrawals", "attest.asset_window", "attest.instrument_window", + "audit.chain", "scope.trade", "attest.cash_posture", "rail.kill_switch", @@ -1065,3 +1066,64 @@ def test_a_folder_keel_cannot_read_measures_as_empty(tmp_path) -> None: assert footprint.total_files == 0 assert read_backup_footprint(tmp_path / "missing").total_files == 0 + + +# -- the audit chain (#721) --------------------------------------------------------------------- + + +def _chain(**overrides: object): + from keel.data.audit import ChainState + + fields: dict[str, object] = { + "table_present": True, + "event_count": 3, + "errors": (), + "first_broken_seq": None, + "hashes": {}, + } + fields.update(overrides) + return ChainState(**fields) # type: ignore[arg-type] + + +def test_a_verifying_chain_is_ok_and_says_how_much_it_checked() -> None: + from keel.commands.doctor import audit_chain_findings + + (finding,) = audit_chain_findings(_chain()) + assert finding.name == "audit.chain" + assert finding.status == "ok" + assert "3" in finding.detail + + +def test_an_empty_chain_is_ok_but_never_called_verified() -> None: + """Nothing was checked. A doctor line reading "audit chain verifies" over zero events is the + green badge over an unread file that this codebase keeps having to take back out.""" + from keel.commands.doctor import audit_chain_findings + + (finding,) = audit_chain_findings(_chain(event_count=0)) + assert finding.status == "ok" + assert "verifies" not in finding.headline + assert "unverified" in finding.headline + + +def test_a_break_fails_and_offers_no_fix() -> None: + """The one check here whose failure means the RECORD was altered rather than a rail being out + of position. Nothing repairs a broken chain, and a `fix` command would imply something does.""" + from keel.commands.doctor import audit_chain_findings + + (finding,) = audit_chain_findings( + _chain(errors=("row 2 (2): content does not match row_hash",), first_broken_seq=2) + ) + assert finding.status == "fail" + assert finding.fix == "-" + assert "row 2" in finding.detail + + +def test_a_database_without_the_table_is_reported_not_crashed_on() -> None: + """`keel mcp`'s `_open_readonly_repo` and the web server both open a repo WITHOUT migrating. + A reader that raised here would take the whole of `gather_findings` down on an un-upgraded + deployment -- which is exactly what #718's first cut did.""" + from keel.commands.doctor import audit_chain_findings + + (finding,) = audit_chain_findings(_chain(table_present=False)) + assert finding.status == "ok" + assert finding.fix == "keel migrate" diff --git a/tests/commands/test_timeline.py b/tests/commands/test_timeline.py index 3d7dfb3..99c6b9e 100644 --- a/tests/commands/test_timeline.py +++ b/tests/commands/test_timeline.py @@ -14,13 +14,17 @@ import csv import io +import sqlite3 from decimal import Decimal from pathlib import Path +from types import SimpleNamespace from typing import Any import pytest +from keel.commands import timeline from keel.commands.timeline import PROVENANCES, TIMELINE_KINDS, gather_timeline +from keel.data import audit from keel.data.db import connect, migrate from keel.data.repository import Repository @@ -30,10 +34,17 @@ @pytest.fixture() -def repo(tmp_path: Path) -> Repository: +def db_conn(tmp_path: Path) -> sqlite3.Connection: + """The CONNECTION, for the tests that reach past the repository to edit `audit_events` in + place -- which is the only way to write the tampering the chain exists to detect.""" conn = connect(str(tmp_path / "keel.db")) migrate(conn) - return Repository(conn) + return conn + + +@pytest.fixture() +def repo(db_conn: sqlite3.Connection) -> Repository: + return Repository(db_conn) def _order(repo: Repository, **overrides: Any) -> int: @@ -200,17 +211,27 @@ def test_each_row_names_the_store_it_came_from(repo: Repository, tmp_path: Path) def test_no_row_claims_a_hash_it_does_not_have(repo: Repository, tmp_path: Path) -> None: - """#703 asked for tamper-evidence. None of these four stores hashes its rows, so every row - says NOT RECORDED -- never blank, which a reader takes as "nothing to report", and never a - hash computed here, which would be this module attesting to its own output.""" - from keel.commands.timeline import HASH_NOT_RECORDED + """Never blank, which a reader takes as "nothing to report", and never a hash computed HERE, + which would be this module attesting to its own output. + + #703 shipped this as "every row says NOT RECORDED", which was the whole truth then. #721 made + the three stores chain their writes, so the invariant is now the pairing rather than the + constant: a hash and a `chained` status travel together, and the absence of one is the + absence of both. A row showing a hash while claiming `not chained`, or claiming `chained` + with nothing to show, would be a row asserting something the chain never said. + """ + from keel.commands.timeline import CHAIN_STATUSES, HASH_NOT_RECORDED _order(repo) _transaction(repo) _attestation(repo) - for row in gather_timeline(repo, now_ts=NOW_TS, scope="all").rows: - assert row.row_hash == HASH_NOT_RECORDED + rows = gather_timeline(repo, now_ts=NOW_TS, scope="all").rows + assert rows + for row in rows: + assert row.chain_status in CHAIN_STATUSES + assert (row.row_hash == HASH_NOT_RECORDED) == (row.chain_status == "not chained") + assert row.row_hash != "" # -- filtering and scoping ---------------------------------------------------------------------- @@ -432,3 +453,172 @@ def test_the_kind_collapse_is_applied_and_echoed(repo, tmp_path) -> None: assert report.kind == "", "the applied value is echoed, so the substitution is visible" assert report.shown_count == 2, "and nothing was filtered out" + + +# -- the hash column, once the engine records one (#721) ---------------------------------------- +# +# #703 shipped `row_hash` reading NOT RECORDED on every row because none of the four stores hashed +# anything. `keel/data/audit.py` now does. These pin the three readings the column has to keep +# apart: a hash the chain vouches for, an honest gap where no event was ever written, and a hash +# that exists inside the region a break has invalidated. The third is the one that must not be +# allowed to look like the first. + + +def _chained_repo(conn) -> Repository: + """A repository whose writers have recorded events for everything they wrote.""" + repo = Repository(conn) + repo.insert_order( + { + "mode": "live", + "product_id": "BTC-USD", + "side": "buy", + "qty": Decimal("1"), + "status": "filled", + "created_at": 1_000, + } + ) + repo.upsert_transaction( + { + "coinbase_id": "cb-1", + "source": "coinbase", + "type": "deposit", + "asset": "USD", + "qty": Decimal("250"), + "ts": 1_100, + } + ) + repo.upsert_asset_attestation( + asset="BTC", + sector="tech", + backing="none", + pays_yield=False, + source="prospectus", + attested_by="operator", + attested_at=1_200, + ) + return repo + + +def test_a_chained_row_carries_its_recorded_hash(db_conn) -> None: + """The swap #703 built the column for: a real hash where the engine recorded one.""" + repo = _chained_repo(db_conn) + report = timeline.gather_timeline(repo, now_ts=2_000) + + by_source = {row.source: row for row in report.rows} + for source in ("orders", "transactions", "asset_attestations"): + assert by_source[source].row_hash != timeline.HASH_NOT_RECORDED + assert len(by_source[source].row_hash) == 64 + assert by_source[source].chain_status == "chained" + + +def test_the_hash_shown_is_the_hash_recorded_for_that_row(db_conn) -> None: + """Looked up by `(source, reference)` -- the pair printed beside it. A lookup keyed on the + entity alone would hand an `orders.id` of "1" the hash of a `coinbase_id` of "1".""" + repo = _chained_repo(db_conn) + report = timeline.gather_timeline(repo, now_ts=2_000) + + recorded = audit.latest_hashes(db_conn) + for row in report.rows: + if row.chain_status == "chained": + assert row.row_hash == recorded[(row.source, row.reference)] + + +def test_a_row_written_before_the_chain_shipped_reads_as_not_chained(db_conn) -> None: + """NO BACKFILL, on the surface that shows it. An honest gap -- and crucially NOT a break: a + deployment upgrading into #721 must not open its timeline to a page of red.""" + db_conn.execute( + "INSERT INTO orders (mode, product_id, side, qty, status, created_at) " + "VALUES ('live','BTC-USD','buy','1','filled',900)" + ) + db_conn.commit() + repo = _chained_repo(db_conn) + report = timeline.gather_timeline(repo, now_ts=2_000) + + unchained = [row for row in report.rows if row.reference == "1" and row.source == "orders"] + assert len(unchained) == 1 + assert unchained[0].row_hash == timeline.HASH_NOT_RECORDED + assert unchained[0].chain_status == "not chained" + assert report.chain_intact is True + + +def test_an_engine_log_row_is_never_chained_and_says_so(db_conn) -> None: + """The engine log is a FILE, not a chained store. A cycle row carrying a hash would be this + module attesting to something it only read.""" + repo = _chained_repo(db_conn) + cycle = SimpleNamespace( + started_ts=1_500, cycle_id="c-1", products=("BTC-USD",), signals=1, entered=1, exited=0, + errors=0, + ) + report = timeline.gather_timeline(repo, now_ts=2_000, cycles=[cycle]) + + system = [row for row in report.rows if row.kind == "system"] + assert len(system) == 1 + assert system[0].row_hash == timeline.HASH_NOT_RECORDED + assert system[0].chain_status == "not chained" + + +def test_a_break_marks_the_rows_the_chain_no_longer_vouches_for(db_conn) -> None: + """Editing one event does not merely fail that row: the chain proves a SEQUENCE, so every row + from the break onward is unverified. A page that showed those later hashes as `chained` would + be presenting unverified values as evidence -- the one thing this column exists to prevent.""" + repo = _chained_repo(db_conn) + db_conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 1", ('{"a":1}',)) + db_conn.commit() + + report = timeline.gather_timeline(repo, now_ts=2_000) + assert report.chain_intact is False + assert report.chain_errors + assert {row.chain_status for row in report.rows} == {"chain broken"} + + +def test_a_break_leaves_earlier_rows_verified(db_conn) -> None: + """The break is located, not global. Rows chained BEFORE it are still vouched for, and + reporting them as broken would throw away the evidence the chain does hold.""" + repo = _chained_repo(db_conn) + db_conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 3", ('{"a":1}',)) + db_conn.commit() + + report = timeline.gather_timeline(repo, now_ts=2_000) + statuses = {row.source: row.chain_status for row in report.rows} + assert statuses["orders"] == "chained" + assert statuses["transactions"] == "chained" + assert statuses["asset_attestations"] == "chain broken" + + +def test_an_empty_chain_is_reported_as_unchecked_not_as_verified(db_conn) -> None: + """`verify` over zero rows returns no errors, and that is not "verified" -- nothing was read. + The report must not offer a green verdict over a table nothing wrote.""" + db_conn.execute( + "INSERT INTO orders (mode, product_id, side, qty, status, created_at) " + "VALUES ('live','BTC-USD','buy','1','filled',900)" + ) + db_conn.commit() + report = timeline.gather_timeline(Repository(db_conn), now_ts=2_000) + assert report.chain_recorded is False + assert report.chain_intact is True + assert report.chain_errors == () + + +def test_the_export_carries_the_chain_status_beside_the_hash(db_conn) -> None: + """A hash with nothing saying whether it verifies is a number an auditor cannot use. Both + columns, in the same file, on every row.""" + repo = _chained_repo(db_conn) + text = timeline.to_csv(timeline.export_rows(repo, now_ts=2_000)) + rows = list(csv.reader(io.StringIO(text))) + + assert rows[0][-2:] == ["row_hash", "chain_status"] + for row in rows[1:]: + assert row[-1] in ("chained", "not chained", "chain broken") + + +def test_a_broken_chain_is_stated_above_the_header_not_only_per_row(db_conn) -> None: + """The same reasoning as the engine-log note this file already carries: an auditor holding a + CSV cannot ask the page anything, so a finding that changes how the whole file should be read + belongs in the file.""" + repo = _chained_repo(db_conn) + db_conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 1", ('{"a":1}',)) + db_conn.commit() + + text = timeline.to_csv(timeline.export_rows(repo, now_ts=2_000)) + assert text.splitlines()[0].startswith("# NOTE") + assert "chain" in text.splitlines()[0] diff --git a/tests/core/test_hashchain.py b/tests/core/test_hashchain.py new file mode 100644 index 0000000..0d536d8 --- /dev/null +++ b/tests/core/test_hashchain.py @@ -0,0 +1,162 @@ +"""The one canonical form and the one chain walk (#721). + +Two stores hash rows in this codebase -- the research trials ledger (JSONL, on disk, git-tracked) +and `audit_events` (SQLite, per deployment). They must agree about what canonical JSON means, +because two canonicalisations that disagree produce two hashes for one row, and the disagreement +does not surface as a bug: it surfaces, months later, as a chain that "cannot be verified". + +So the form lives here, once, and both stores import it. These tests pin the properties the +hashes depend on rather than the hashes themselves -- with one deliberate exception, the tracked +ledger's own 93 rows (`tests/research/test_ledger.py`), which were hashed before this module +existed and must still verify byte-for-byte. +""" + +from __future__ import annotations + +from decimal import Decimal + +from keel_core import hashchain + + +def _link(label: str, prev: str, payload: dict[str, object]) -> hashchain.ChainLink: + body = dict(payload) + body["prev_hash"] = prev + row_hash = hashchain.chain_hash(body) + return hashchain.ChainLink( + label=label, prev_hash=prev, row_hash=row_hash, recomputed_hash=row_hash + ) + + +def test_canonical_json_is_insertion_order_blind() -> None: + """Key order in the caller's dict must not reach the hash. + + A payload assembled by a `dict` literal in one module and by `dict(row)` out of sqlite in + another would otherwise hash differently while describing the same row. + """ + assert hashchain.canonical_json({"b": 1, "a": 2}) == hashchain.canonical_json( + {"a": 2, "b": 1} + ) + + +def test_canonical_json_has_no_incidental_whitespace() -> None: + """`json.dumps` defaults put a space after every separator. Pinned, because a later reader + "tidying" the call would silently invalidate every hash ever written.""" + assert hashchain.canonical_json({"a": 1, "b": 2}) == '{"a":1,"b":2}' + + +def test_decimal_crosses_as_its_exact_string_at_any_depth() -> None: + """Money is TEXT everywhere in this codebase, and a hash over a float would be a hash over a + lossy rendering of the number the row actually holds.""" + text = hashchain.canonical_json( + {"qty": Decimal("0.10"), "legs": [Decimal("1.5")], "nested": {"fee": Decimal("2.000")}} + ) + assert text == '{"legs":["1.5"],"nested":{"fee":"2.000"},"qty":"0.10"}' + + +def test_decimal_scale_is_part_of_the_hash() -> None: + """`Decimal("1.0") == Decimal("1.00")` numerically, and they are DIFFERENT recorded values. + A store that rewrote a row's scale changed the row, and the chain must say so.""" + assert hashchain.chain_hash({"qty": Decimal("1.0")}) != hashchain.chain_hash( + {"qty": Decimal("1.00")} + ) + + +def test_chain_hash_changes_when_any_field_changes() -> None: + base = {"a": 1, "b": "x"} + assert hashchain.chain_hash(base) != hashchain.chain_hash({"a": 1, "b": "y"}) + assert hashchain.chain_hash(base) != hashchain.chain_hash({"a": 2, "b": "x"}) + + +def test_an_absent_key_is_not_a_null_key() -> None: + """The absent-vs-null distinction the ledger's docstring names. A writer that started + emitting an explicit `None` for a field it used to omit has changed the row.""" + assert hashchain.chain_hash({"a": 1}) != hashchain.chain_hash({"a": 1, "b": None}) + + +def test_an_intact_chain_reports_nothing() -> None: + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + second = _link("two", first.row_hash, {"n": 2}) + assert hashchain.verify_links([first, second]) == [] + + +def test_a_first_row_not_anchored_to_zero_is_a_break() -> None: + """A chain that starts mid-air proves nothing about what came before it.""" + orphan = _link("one", "f" * 64, {"n": 1}) + errors = hashchain.verify_links([orphan]) + assert len(errors) == 1 + assert "row 1 (one)" in errors[0] + assert "does not chain" in errors[0] + + +def test_edited_content_fails_at_that_row() -> None: + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + tampered = hashchain.ChainLink( + label=first.label, + prev_hash=first.prev_hash, + row_hash=first.row_hash, + recomputed_hash=hashchain.chain_hash({"n": 99, "prev_hash": first.prev_hash}), + ) + errors = hashchain.verify_links([tampered]) + assert len(errors) == 1 + assert "row 1 (one)" in errors[0] + assert "row_hash" in errors[0] + + +def test_every_break_is_reported_not_only_the_first() -> None: + """A chain check that stopped at the first break would let the second edit hide behind the + first, which is the shape an auditor most needs to see.""" + first = _link("one", "a" * 64, {"n": 1}) + second = _link("two", "b" * 64, {"n": 2}) + errors = hashchain.verify_links([first, second]) + assert len(errors) == 2 + + +def test_a_deleted_row_breaks_every_row_after_it() -> None: + """The property that makes this evidence rather than a report.""" + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + second = _link("two", first.row_hash, {"n": 2}) + third = _link("three", second.row_hash, {"n": 3}) + errors = hashchain.verify_links([first, third]) + assert len(errors) == 1 + assert "row 2 (three)" in errors[0] + + +def test_an_empty_chain_reports_nothing_and_that_is_not_verified() -> None: + """Nothing was checked. The caller holds the distinction between "no rows yet" and "no store + at all" -- see `verify_links`' own docstring.""" + assert hashchain.verify_links([]) == [] + + +def test_a_break_is_located_as_well_as_described() -> None: + """`verify_links` prints; `find_breaks` locates. A caller that needs its own key for the + first unverified row -- the timeline's chain-status column -- must not have to parse the + English out of a message to find it.""" + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + second = _link("two", first.row_hash, {"n": 2}) + edited = hashchain.ChainLink( + label=second.label, + prev_hash=second.prev_hash, + row_hash=second.row_hash, + recomputed_hash=hashchain.chain_hash({"n": 99, "prev_hash": second.prev_hash}), + ) + breaks = hashchain.find_breaks([first, edited]) + assert [(found.index, found.label, found.reason) for found in breaks] == [(2, "two", "content")] + + +def test_a_missing_row_is_a_link_break_not_a_content_break() -> None: + """Two different accusations. A report that blurred them would send an auditor looking for an + edited row when what happened was a deletion.""" + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + second = _link("two", first.row_hash, {"n": 2}) + third = _link("three", second.row_hash, {"n": 3}) + assert [found.reason for found in hashchain.find_breaks([first, third])] == ["link"] + + +def test_verify_links_is_exactly_find_breaks_formatted() -> None: + """One walk, not two. A chain check that existed twice could disagree with itself, and the + disagreement would be between a report and a badge over the same rows.""" + first = _link("one", "a" * 64, {"n": 1}) + second = _link("two", "b" * 64, {"n": 2}) + assert hashchain.verify_links([first, second]) == [ + found.message for found in hashchain.find_breaks([first, second]) + ] diff --git a/tests/data/test_audit_chain.py b/tests/data/test_audit_chain.py new file mode 100644 index 0000000..f6964c2 --- /dev/null +++ b/tests/data/test_audit_chain.py @@ -0,0 +1,408 @@ +"""`audit_events`: the append-only, hash-chained record of what the book did (#721). + +`orders` rows are UPDATED as a venue reports fills, so a chain over the order row itself would +break on every legitimate update -- a chain that cries wolf is a chain nobody reads. So the chain +is over EVENTS: `insert_order` and each `update_order` append an immutable statement of what +changed, and it is the event stream, not the mutable book, that is tamper-evident. + +What these tests pin is the property that makes it evidence: an edit or a deletion cannot be made +quietly. Everything else here -- the vocabulary, the atomicity, the pre-bump gap -- exists to +keep that property true in the presence of the rest of the engine. +""" + +from __future__ import annotations + +import sqlite3 +from decimal import Decimal + +import pytest +from keel_core.hashchain import ZERO_HASH + +from keel.data import audit, db +from keel.data.repository import Repository + + +@pytest.fixture() +def conn() -> sqlite3.Connection: + connection = db.connect(":memory:") + db.migrate(connection) + return connection + + +def _append(connection: sqlite3.Connection, **kwargs: object) -> audit.AuditEvent: + """One event in its own write transaction, as a repository writer does around a store row.""" + with audit.write_transaction(connection): + return audit.append_event(connection, **kwargs) # type: ignore[arg-type] + + +def _order(**overrides: object) -> dict[str, object]: + """A minimally legal `orders` row -- every NOT NULL column and nothing else.""" + row: dict[str, object] = { + "mode": "live", + "product_id": "BTC-USD", + "side": "buy", + "qty": Decimal("1"), + "created_at": 100, + } + row.update(overrides) + return row + + +def _transaction(**overrides: object) -> dict[str, object]: + """A minimally legal `transactions` row.""" + row: dict[str, object] = { + "coinbase_id": "cb-1", + "source": "coinbase", + "type": "deposit", + "asset": "USD", + "ts": 100, + "qty": Decimal("1"), + } + row.update(overrides) + return row + + +# -- the chain itself ----------------------------------------------------------------------- + + +def test_the_first_event_anchors_to_the_zero_hash(conn: sqlite3.Connection) -> None: + event = _append(conn, ts=100, event_type="order_placed", entity_id="1", payload={"a": 1}) + assert event.prev_hash == ZERO_HASH + assert len(event.row_hash) == 64 + + +def test_each_event_commits_to_its_predecessor(conn: sqlite3.Connection) -> None: + first = _append(conn, ts=100, event_type="order_placed", entity_id="1", payload={"a": 1}) + second = _append(conn, ts=101, event_type="order_updated", entity_id="1", payload={"a": 2}) + assert second.prev_hash == first.row_hash + assert audit.verify_events(audit.read_events(conn)) == [] + + +def test_the_chain_spans_event_types_not_one_chain_per_table(conn: sqlite3.Connection) -> None: + """One chain over everything, so an event cannot be moved between streams, and so a removed + order event is visible from a later transaction event rather than only from its own kind.""" + _append(conn, ts=100, event_type="order_placed", entity_id="1", payload={"a": 1}) + second = _append( + conn, ts=101, event_type="transaction_recorded", entity_id="cb-1", payload={"a": 2} + ) + events = audit.read_events(conn) + assert second.prev_hash == events[0].row_hash + assert audit.verify_events(events) == [] + + +def test_editing_a_payload_in_place_breaks_that_row_and_no_other(conn: sqlite3.Connection) -> None: + for index in range(3): + _append( + conn, + ts=100 + index, + event_type="order_placed", + entity_id=str(index), + payload={"a": index}, + ) + conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 2", ('{"a":99}',)) + conn.commit() + + errors = audit.verify_events(audit.read_events(conn)) + assert len(errors) == 1 + assert "row 2" in errors[0] + assert "row_hash" in errors[0] + + +def test_deleting_a_row_breaks_every_row_after_it(conn: sqlite3.Connection) -> None: + """The property the whole store exists for: a deletion is not a quiet deletion.""" + for index in range(4): + _append( + conn, + ts=100 + index, + event_type="order_placed", + entity_id=str(index), + payload={"a": index}, + ) + conn.execute("DELETE FROM audit_events WHERE seq_id = 2") + conn.commit() + + errors = audit.verify_events(audit.read_events(conn)) + assert len(errors) == 1, errors + assert "row 2" in errors[0] + assert "does not chain" in errors[0] + + +def test_events_are_read_in_chain_order_not_timestamp_order(conn: sqlite3.Connection) -> None: + """`seq_id`, never `ts`. Two events inside one second share a timestamp, and a clock that + steps backwards would reorder the chain into a false break.""" + # THREE, and the timestamps deliberately out of order. With two, the head read has only one + # candidate and any ordering picks it -- which is why the two-event version of this test + # passed against a head read ordered by `ts`. The third event is the one whose predecessor + # the two orderings disagree about: by `seq_id` it chains to the second, by `ts DESC` it + # would chain to the FIRST, and the chain would fork. + _append(conn, ts=500, event_type="order_placed", entity_id="1", payload={"a": 1}) + _append(conn, ts=100, event_type="order_updated", entity_id="1", payload={"a": 2}) + _append(conn, ts=200, event_type="order_updated", entity_id="1", payload={"a": 3}) + + events = audit.read_events(conn) + assert [event.seq_id for event in events] == [1, 2, 3] + assert events[2].prev_hash == events[1].row_hash + assert audit.verify_events(events) == [] + + +def test_a_decimal_in_the_payload_survives_the_round_trip(conn: sqlite3.Connection) -> None: + """Money is the point. A payload whose Decimals came back as floats would verify against a + hash of a value the row never held.""" + event = _append( + conn, + ts=100, + event_type="order_placed", + entity_id="1", + payload={"qty": Decimal("0.10"), "fee": None}, + ) + stored = audit.read_events(conn)[0] + assert stored.payload["qty"] == "0.10" + assert stored.row_hash == event.row_hash + assert audit.verify_events([stored]) == [] + + +# -- the vocabulary and the guards ------------------------------------------------------------ + + +def test_an_unrecognised_event_type_is_refused(conn: sqlite3.Connection) -> None: + """A closed vocabulary, like every other `kind`/`provenance` set in this codebase: a typo + that lands as a new event type is a stream nothing reads and nothing misses.""" + with pytest.raises(ValueError, match="event_type"): + _append(conn, ts=100, event_type="order_plased", entity_id="1", payload={}) + + +def test_appending_outside_a_write_transaction_is_refused(conn: sqlite3.Connection) -> None: + """The head read and the insert MUST be one transaction. Read the head in autocommit and two + writers racing both see the same head, both write it as their `prev_hash`, and the chain + forks -- silently, because each row verifies against the row it thinks precedes it.""" + with pytest.raises(RuntimeError, match="transaction"): + audit.append_event(conn, ts=100, event_type="order_placed", entity_id="1", payload={}) + + +def test_the_head_read_takes_a_write_lock_before_reading(conn: sqlite3.Connection) -> None: + """`BEGIN IMMEDIATE`, not `BEGIN`. A deferred transaction takes no lock until its first + write, so the head SELECT would still be unserialised against a concurrent writer.""" + with audit.write_transaction(conn): + assert conn.in_transaction + # A second connection to the same file cannot begin writing while this one holds the + # write lock. In-memory connections do not share a database, so this asserts the + # statement issued rather than the lock's effect -- the effect is asserted below. + assert not conn.in_transaction + + +def test_two_connections_cannot_hold_the_write_lock_at_once(tmp_path) -> None: + """The lock, on a real file. Without `IMMEDIATE` the second writer would be admitted here and + would read the same chain head as the first.""" + path = tmp_path / "keel.db" + first = db.connect(path) + db.migrate(first) + second = db.connect(path) + second.execute("PRAGMA busy_timeout = 50") + + with audit.write_transaction(first): + # BEFORE any write. A DEFERRED transaction would still be admitted here and would take + # its lock only at the INSERT -- by which point it has already read a chain head that a + # racing writer may have moved. Appending first and then checking the lock passes under + # `BEGIN` too, and so proves nothing: the insert itself takes the lock either way. + with pytest.raises(sqlite3.OperationalError, match="locked|busy"): + second.execute("BEGIN IMMEDIATE") + audit.append_event(first, ts=100, event_type="order_placed", entity_id="1", payload={}) + + +# -- the writers ---------------------------------------------------------------------------- + + +def test_insert_order_records_a_chained_placement_event(conn: sqlite3.Connection) -> None: + repo = Repository(conn) + order_id = repo.insert_order(_order(product_id="BTC-USD", qty=Decimal("0.5"))) + events = audit.read_events(conn) + assert [event.event_type for event in events] == ["order_placed"] + assert events[0].entity_id == str(order_id) + assert events[0].payload["product_id"] == "BTC-USD" + assert audit.verify_events(events) == [] + + +def test_update_order_records_what_changed_not_the_whole_row(conn: sqlite3.Connection) -> None: + """An event says what this statement asserted. Re-hashing the whole mutated row would make + each event a snapshot, and a reader could not tell an update from a re-write.""" + repo = Repository(conn) + order_id = repo.insert_order(_order(product_id="BTC-USD")) + repo.update_order(order_id, status="filled", actual_fill=Decimal("42.5")) + + events = audit.read_events(conn) + assert [event.event_type for event in events] == ["order_placed", "order_updated"] + assert events[1].payload == {"status": "filled", "actual_fill": "42.5"} + assert audit.verify_events(events) == [] + + +def test_an_update_with_no_fields_records_nothing(conn: sqlite3.Connection) -> None: + """`update_order` already returns early for an empty change set. An event asserting that + nothing changed is chain noise.""" + repo = Repository(conn) + repo.insert_order(_order(product_id="BTC-USD")) + repo.update_order(1) + assert [event.event_type for event in audit.read_events(conn)] == ["order_placed"] + + +def test_upsert_transaction_records_a_flow_event(conn: sqlite3.Connection) -> None: + repo = Repository(conn) + repo.upsert_transaction(_transaction(coinbase_id="cb-1", total=Decimal("250.00"))) + events = audit.read_events(conn) + assert [event.event_type for event in events] == ["transaction_recorded"] + assert events[0].entity_id == "cb-1" + assert audit.verify_events(events) == [] + + +def test_re_importing_a_transaction_appends_rather_than_rewrites(conn: sqlite3.Connection) -> None: + """`upsert_transaction` UPDATES the book row in place. The chain must not: two events for one + `coinbase_id` is the record that the line was re-imported with different content, which is + exactly what an auditor wants visible.""" + repo = Repository(conn) + for total in (Decimal("250.00"), Decimal("260.00")): + repo.upsert_transaction( + _transaction(coinbase_id="cb-1", total=total) + ) + events = audit.read_events(conn) + assert len(events) == 2 + assert events[0].row_hash != events[1].row_hash + assert audit.verify_events(events) == [] + + +def test_both_attestation_upserts_record_a_human_claim(conn: sqlite3.Connection) -> None: + repo = Repository(conn) + repo.upsert_asset_attestation( + asset="BTC", sector="tech", backing="none", pays_yield=False, + source="prospectus", attested_by="operator", attested_at=100, + ) + repo.upsert_instrument_attestation( + venue="coinbase", product_id="BTC-USD", wrapper="spot", + source="venue docs", attested_by="operator", attested_at=101, + ) + events = audit.read_events(conn) + assert [event.event_type for event in events] == ["asset_attested", "instrument_attested"] + assert events[0].entity_id == "BTC" + assert events[1].entity_id == "coinbase:BTC-USD" + assert audit.verify_events(events) == [] + + +def test_the_store_row_and_its_event_land_together_or_not_at_all( + conn: sqlite3.Connection, monkeypatch: pytest.MonkeyPatch +) -> None: + """One transaction, both writes. An order row with no event would read forever after as + "written before the bump" -- an honest-looking gap that is in fact a failed chain write, and + the one lie this store must never tell about itself.""" + + def _boom(*_args: object, **_kwargs: object) -> None: + raise sqlite3.OperationalError("disk I/O error") + + monkeypatch.setattr("keel.data.repository.append_event", _boom) + repo = Repository(conn) + with pytest.raises(sqlite3.OperationalError): + repo.insert_order(_order(product_id="BTC-USD")) + + assert conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] == 0 + + +# -- the honest gap --------------------------------------------------------------------------- + + +def test_rows_written_before_the_bump_leave_no_event_and_that_is_not_a_break( + conn: sqlite3.Connection, +) -> None: + """NO BACKFILL. A row inserted by a pre-#721 build has no event, and computing one now would + produce a chain that verifies and proves nothing -- worse than a gap, because it looks like + evidence. The chain over the events that DO exist stays intact.""" + conn.execute( + "INSERT INTO orders (mode, product_id, side, qty, status, created_at) " + "VALUES ('live','BTC-USD','buy','1','filled',50)" + ) + conn.commit() + repo = Repository(conn) + repo.insert_order(_order(product_id="ETH-USD")) + + events = audit.read_events(conn) + assert len(events) == 1 + assert events[0].entity_id == "2" + assert audit.verify_events(events) == [] + + +def test_latest_hashes_reports_the_newest_event_per_entity(conn: sqlite3.Connection) -> None: + """What the timeline reads: the most recent CHAINED statement about a row. An order that was + placed and then filled has two events, and the fill is the later word on it.""" + repo = Repository(conn) + order_id = repo.insert_order(_order(product_id="BTC-USD")) + repo.update_order(order_id, status="filled") + + latest = audit.latest_hashes(conn) + events = audit.read_events(conn) + assert latest[("orders", str(order_id))] == events[1].row_hash + + +def test_latest_hashes_is_keyed_by_store_so_two_stores_cannot_collide( + conn: sqlite3.Connection, +) -> None: + """`orders.id` is `"1"` and so is a `coinbase_id` of `"1"`. Keyed by entity alone, an + imported transaction would hand its hash to an unrelated order row.""" + repo = Repository(conn) + repo.insert_order(_order(product_id="BTC-USD")) + repo.upsert_transaction(_transaction(coinbase_id="1")) + + latest = audit.latest_hashes(conn) + assert latest[("orders", "1")] != latest[("transactions", "1")] + + +# -- readers that never migrate ---------------------------------------------------------------- +# +# Two readers open a database WITHOUT migrating it: `keel/web/server.py` (a view must not take a +# schema write lock) and `keel/mcp/tools.py::_open_readonly_repo`. Both therefore meet pre-v20 +# databases with no `audit_events` table at all. #718 shipped a reader that raised on exactly this +# shape and took the whole of `gather_findings` down with it; these are that lesson, pinned. + + +def _pre_chain_database() -> sqlite3.Connection: + """A database with the book but not the chain -- what an un-upgraded deployment looks like.""" + connection = db.connect(":memory:") + db.migrate(connection) + connection.execute("DROP TABLE audit_events") + connection.commit() + return connection + + +def test_chain_state_reports_a_missing_table_rather_than_raising() -> None: + connection = _pre_chain_database() + state = audit.chain_state(connection) + + assert state.table_present is False + assert state.event_count == 0 + assert state.errors == () + assert state.intact is False + + +def test_the_timeline_still_renders_on_a_database_without_the_chain() -> None: + """The page an operator opens right after upgrading keel and before running `keel migrate`. + Every row reads `not chained`, which is true, and nothing raises.""" + from keel.commands import timeline + + connection = _pre_chain_database() + connection.execute( + "INSERT INTO orders (mode, product_id, side, qty, status, created_at) " + "VALUES ('live','BTC-USD','buy','1','filled',100)" + ) + connection.commit() + + report = timeline.gather_timeline(Repository(connection), now_ts=200) + assert [row.chain_status for row in report.rows] == ["not chained"] + assert report.chain_recorded is False + assert report.chain_errors == () + + +def test_appending_to_a_missing_table_still_fails_loudly() -> None: + """Tolerance is for READERS. A write that cannot record its event must not quietly succeed: + that is the failed chain write masquerading as an honest gap, and the one lie this store must + never tell about itself.""" + connection = _pre_chain_database() + repo = Repository(connection) + with pytest.raises(sqlite3.OperationalError): + repo.insert_order(_order()) + assert connection.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 0 diff --git a/tests/research/test_ledger.py b/tests/research/test_ledger.py index d04b1a8..9e1cf54 100644 --- a/tests/research/test_ledger.py +++ b/tests/research/test_ledger.py @@ -4,6 +4,7 @@ from dataclasses import replace from decimal import Decimal +from pathlib import Path import pytest @@ -205,3 +206,26 @@ def test_verify_records_of_nothing_is_no_errors_and_not_a_verification(tmp_path) held for the web view. Pinned here so the emptiness stays the caller's problem to name rather than becoming a silent green light.""" assert ledger.verify_records([]) == [] + + +def test_the_tracked_ledger_still_verifies_after_the_canonicaliser_moved() -> None: + """The 93 rows in `docs/experiments/trials-ledger.jsonl` were hashed by the canonicaliser + that used to live in this module. #721 moved it to `keel_core.hashchain` so `audit_events` + could share one definition of canonical JSON rather than grow a second one. + + A move like that is exactly where a chain dies quietly: nothing raises, no test fails, and + the ledger simply stops verifying the next time anyone asks -- by which point there is no way + to tell an honest refactor from an edit. So this reads the REAL, git-tracked file (not a + fixture, not a temp path -- `conftest`'s autouse redirect is deliberately bypassed by using + the literal path) and asserts every one of its rows still reproduces its recorded hash. + + It is the only test in this suite that asserts against hashes computed by a previous version + of the code, and that is the point: it is the byte-identity proof. + """ + tracked = Path(__file__).resolve().parents[2] / "docs/experiments/trials-ledger.jsonl" + assert tracked.exists(), f"the tracked ledger is missing: {tracked}" + records = ledger.read_trials(tracked) + # Guards the assertion below against passing over an empty read: `verify_chain` returns no + # errors for zero rows, which is not the same as "verified" (see `verify_records`). + assert len(records) > 1, "the tracked ledger must hold a chain, not a single row" + assert ledger.verify_chain(tracked) == [] diff --git a/tests/web/test_payload.py b/tests/web/test_payload.py index 12cbd69..8f9cef0 100644 --- a/tests/web/test_payload.py +++ b/tests/web/test_payload.py @@ -31,7 +31,7 @@ from __future__ import annotations import json -from dataclasses import dataclass +from dataclasses import dataclass, replace from decimal import Decimal from pathlib import Path from typing import Any @@ -1543,3 +1543,106 @@ def test_an_empty_series_says_so_rather_than_describing_an_empty_chart() -> None assert empty["segments"] == [] assert empty["point_count"]["value"] == "0" assert "no" in empty["reading"]["display"].lower() + + +# -- the audit chain on the wire (#721) --------------------------------------------------------- + + +def _timeline_report(**overrides: object): + from keel.commands.timeline import TimelineReport, TimelineRow + + row = TimelineRow( + ts=1_000, + kind="trade", + provenance="venue-reported", + source="orders", + reference="1", + summary="filled buy BTC-USD (live)", + product_id="BTC-USD", + amount=None, + amount_kind="", + row_hash="a" * 64, + chain_status="chained", + ) + fields: dict[str, object] = { + "now_ts": 2_000, + "scope": "all", + "scope_start_ts": None, + "kind": "", + "limit": None, + "scoped_count": 1, + "filtered_count": 1, + "rows": (row,), + "kinds_present": ("trade",), + "chain_recorded": True, + "chain_errors": (), + } + fields.update(overrides) + return TimelineReport(**fields) # type: ignore[arg-type] + + +def test_a_chained_row_is_green_and_shortened() -> None: + """The hash is truncated for a cell -- the full value lives in the CSV, which is what anyone + actually verifying one would be working from.""" + body = payload.timeline_payload(_timeline_report()) + cell = body["rows"][0]["row_hash"] + + assert cell["state"] == "good" + assert cell["value"] == "a" * 64 + assert cell["display"].endswith("…") + assert len(cell["display"]) < 64 + + +def test_not_recorded_is_shown_whole_because_it_is_a_sentence() -> None: + """Truncating "NOT RECORDED" to twelve characters and an ellipsis would make it LOOK like a + short hash in a column of long ones -- the absence of evidence dressed as some.""" + from keel.commands.timeline import HASH_NOT_RECORDED + + report = _timeline_report() + row = replace(report.rows[0], row_hash=HASH_NOT_RECORDED, chain_status="not chained") + body = payload.timeline_payload(replace(report, rows=(row,))) + + cell = body["rows"][0]["row_hash"] + assert cell["display"] == HASH_NOT_RECORDED + assert cell["state"] == "unknown" + + +def test_a_row_past_a_break_is_bad_even_though_it_has_a_hash() -> None: + """The state must come from the CHAIN, not from whether a hash is present. A client inferring + the verdict from a 64-character string would call an unverified row evidence.""" + report = _timeline_report(chain_errors=("row 1 (1): content does not match row_hash",)) + row = replace(report.rows[0], chain_status="chain broken") + body = payload.timeline_payload(replace(report, rows=(row,))) + + assert body["rows"][0]["row_hash"]["state"] == "bad" + assert body["rows"][0]["chain_status"]["state"] == "bad" + + +def test_every_chain_status_carries_a_sentence_not_only_the_term() -> None: + """The same rule `_PROVENANCE_NOTES` follows: on an audit surface the term of art is not the + thing a reader can act on.""" + from keel.commands.timeline import CHAIN_STATUSES + + report = _timeline_report() + for status in CHAIN_STATUSES: + row = replace(report.rows[0], chain_status=status) + cell = payload.timeline_payload(replace(report, rows=(row,)))["rows"][0]["chain_status"] + assert cell["value"] == status + assert cell["display"] != status, f"{status} has no sentence beside it" + + +def test_an_empty_chain_is_never_green() -> None: + """No events means nothing was checked -- not that everything verified. The exact trap + `_chain_payload` documents for the research ledger, on the trading side.""" + body = payload.timeline_payload(_timeline_report(chain_recorded=False)) + assert body["chain"]["state"] == "unknown" + assert body["chain"]["value"] == "unverified" + + +def test_a_verified_chain_over_recorded_events_is_the_only_green() -> None: + intact = payload.timeline_payload(_timeline_report()) + assert intact["chain"]["state"] == "good" + + broken = payload.timeline_payload(_timeline_report(chain_errors=("row 3: broken",))) + assert broken["chain"]["state"] == "bad" + assert broken["chain_errors"] == ["row 3: broken"] diff --git a/tests/web/test_timeline_export.py b/tests/web/test_timeline_export.py index 2469f5f..f6c5a83 100644 --- a/tests/web/test_timeline_export.py +++ b/tests/web/test_timeline_export.py @@ -246,3 +246,13 @@ def test_the_export_carries_every_row_in_scope_not_the_pages_worth(tmp_path: Pat assert len(data_rows) == total, ( f"the export must carry the whole scope; got {len(data_rows)} of {total}" ) + + +def test_the_export_names_the_chain_status_beside_the_hash(running: Any) -> None: # noqa: F811 + """#721. The hash column stopped being a placeholder; a hash with no verdict beside it is a + number a reader takes on trust, and the reading taken on trust is the flattering one.""" + _status, _headers, body = _csv(running) + + header = next(csv.reader(io.StringIO(body))) + + assert header[-2:] == ["row_hash", "chain_status"] From edb4b1856748f90f1d54ef500bc75ada128f8aa9 Mon Sep 17 00:00:00 2001 From: Elmehdi Aitbrahim Date: Sat, 5 Sep 2026 12:47:25 -0400 Subject: [PATCH 2/2] =?UTF-8?q?fix(engine):=20review=20findings=20on=20#72?= =?UTF-8?q?1=20=E2=80=94=20a=20swallowed=20diagnostic=20write=20was=20stra?= =?UTF-8?q?nding=20the=20order=20row?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An independent review found four issues. The first reaches the venue. `write_transaction` HAD A JOIN BRANCH FOR NESTING THAT DOES NOT EXIST It read `if conn.in_transaction: yield; return` -- "somebody outside opened a transaction, so they own the commit". Nothing in this codebase nests these, so that branch was never reached by the nesting it was written for. What it WAS reached by is the accident. `sqlite3` runs in legacy implicit-transaction mode, so `in_transaction` is also True after any DML that has not been committed -- including one that raised and was swallowed. `execution/equity.py` swallows a failed `record_cycle_balance` by design, which is the rule the previous PR in this series added: a diagnostic write must never abort a cycle. That swallow left the connection mid-transaction. The next `insert_order` in the same cycle then took the join branch, returned a real `order_id`, and never committed -- and `broker.place_order` runs next. Measured end to end through the real swallow site: `orders visible to another connection: 0`. An order at the venue with no durable row behind it is exactly the row `execution.reconcile` exists to find. On main the identical sequence committed, because each writer ended in an unconditional `commit()`. So the same rule, violated from the other end: a swallowed RECORD failure was deciding whether the ORDER was recorded. Fixed twice over. The commit is now unconditional -- skipping the `BEGIN` on an open transaction is only about sqlite refusing to nest one and never means skipping the commit -- and `equity.py` rolls its swallowed write back, so the dirty connection does not reach the next writer at all. The lock invariant survives either path: legacy mode opens a transaction implicitly only on DML, so a connection already `in_transaction` has already taken the write lock. THE PAYLOAD NOW COMMITS AS THE BYTES THE COLUMN HOLDS The hash was taken over a parsed-then-re-encoded object, which puts a decode/encode round trip between the stored row and its hash -- and every such conversion is somewhere the two can drift. Hashing the stored string removes the question rather than answering it shape by shape. It is also the answer to the review's second finding. The console polls `/api/timeline` every 15 seconds and `chain_state` walks the whole chain, which is not negotiable -- a verdict over a suffix has not looked at the rows most likely to have been quietly edited, and a cached prefix verdict assumes precisely the thing being checked. But verification no longer PARSES: 20,000 events went from 277 ms to 134 ms. The cost that remains is stated in the docstring with its measured numbers, the way `DEFAULT_TIMELINE_LIMIT` and `export_rows` already state theirs next door, along with what to do if it ever stops fitting: bound what is CLAIMED and say so on the page, never cache the verdict. "" IS FALSEY TO PYTHON AND PERFECTLY INDEXABLE TO SQLITE `upsert_transaction` filed its event under `cursor.lastrowid` when `coinbase_id` was falsey, reasoning that a nullable column with distinct NULLs can never take the `DO UPDATE` branch. True of NULL. False of the empty string, which IS subject to the unique index -- so a re-upsert of a blank id updated an existing row while `lastrowid` held the last real INSERT's id. The timeline then showed one row the hash of a superseded event, marked `chained`, beside a phantom key no row resolved to. The id is read back now, and the invariant is pinned where it actually lives: the event's `entity_id` equals what `_transaction_rows` prints as the row's `reference`, across all three `coinbase_id` shapes. AND `or` TREATED THE EPOCH AS ABSENT `int(value or time.time())` on three timestamps. A row stamped at 0 got an event stamped `now`, and the event's `ts` is hashed -- so the chain would attest to a timestamp the row does not hold. Every real call site passes a non-zero value today, so this was a guard against a future caller, which is exactly when a silent fallback is worth making explicit. Also: `verify_events` and `latest_hashes` had no production callers and are gone; the tests that used them go through `chain_state`, which is the path anything real takes. And the export's status assertion was `in (the three words)`, which passes on any of them -- it now pins the pairing. Six mutants killed: the join branch returning, the swallow only logging, the falsey branch guessing the newest row, a zero timestamp falling back to now, the payload being re-encoded from the parsed object, and the export's status hard-coded to one word. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KZZxmspQXe5qJ9FAsG13s6 --- keel/data/audit.py | 183 +++++++++++++++-------------- keel/data/repository.py | 68 +++++++++-- keel/execution/equity.py | 12 ++ tests/commands/test_timeline.py | 10 +- tests/data/test_audit_chain.py | 199 +++++++++++++++++++++++++++++--- tests/execution/test_equity.py | 50 ++++++++ 6 files changed, 407 insertions(+), 115 deletions(-) diff --git a/keel/data/audit.py b/keel/data/audit.py index 7550a3f..5b461eb 100644 --- a/keel/data/audit.py +++ b/keel/data/audit.py @@ -53,14 +53,7 @@ from dataclasses import dataclass from typing import Any -from keel_core.hashchain import ( - ZERO_HASH, - ChainLink, - chain_hash, - encode_value, - find_breaks, - verify_links, -) +from keel_core.hashchain import ZERO_HASH, ChainLink, canonical_json, chain_hash, find_breaks #: Every kind of statement this chain carries, and the store each one is about. #: @@ -108,18 +101,33 @@ def write_transaction(conn: sqlite3.Connection) -> Iterator[None]: because each row verifies against the row it believes precedes it. `IMMEDIATE` takes the write lock up front, which is what makes "read the head, then append" atomic. - Nested calls JOIN the caller's transaction rather than opening a second one: sqlite raises - "cannot start a transaction within a transaction", and more importantly a nested commit would - publish the outer writer's half-finished work. The outermost block owns the commit. + **The commit is unconditional, and the first cut of this got it wrong in a way that reached + the venue.** It read `if conn.in_transaction: yield; return` -- "somebody outside opened a + transaction, so they own the commit". Nothing in this codebase nests these, so that branch + was never reached by the nesting it was written for. What it WAS reached by is the accident: + `sqlite3` runs in legacy implicit-transaction mode, so `in_transaction` is also True after any + DML that has not been committed -- INCLUDING one that raised and was swallowed. + `execution/equity.py` swallows a failed `record_cycle_balance` by design (a diagnostic write + must never abort a cycle), and that left the connection mid-transaction. The next + `insert_order` then took the join branch, returned a real `order_id`, and never committed -- + so `broker.place_order` sent an order to the venue with no durable row behind it, which is + precisely the row `execution.reconcile` exists to find. Measured end to end: `orders visible + to another connection: 0`. + + So this always commits on a clean exit, which is exactly what the unconditional + `self._conn.commit()` in each writer did before this module existed. Skipping the `BEGIN` on + an already-open transaction is only about sqlite refusing to nest one; it never means + skipping the commit. (The lock invariant survives that path: legacy mode opens a transaction + implicitly only on DML, so a connection that is already `in_transaction` has already taken + the write lock, and the head read is serialised either way. `equity.py` now rolls its + swallowed write back, so the path should not arise at all -- this is the second belt.) Rolls back on ANY exception, `BaseException` included: a `KeyboardInterrupt` between the store row and its event would otherwise leave the row committed and the chain silent about it, which is the one lie this store must not tell about itself. """ - if conn.in_transaction: - yield - return - conn.execute("BEGIN IMMEDIATE") + if not conn.in_transaction: + conn.execute("BEGIN IMMEDIATE") try: yield except BaseException: @@ -157,29 +165,22 @@ def append_event( head = conn.execute("SELECT row_hash FROM audit_events ORDER BY seq_id DESC LIMIT 1").fetchone() prev_hash = ZERO_HASH if head is None else str(head["row_hash"]) - # Encoded BEFORE hashing and stored in the encoded form, so what is hashed is byte-for-byte - # what a reader gets back. Encoding at read time instead would put a conversion between the - # stored row and the hash, and every such conversion is somewhere the two can drift. - encoded: dict[str, Any] = encode_value(dict(payload)) - body = { - "ts": ts, - "event_type": event_type, - "entity_id": entity_id, - "payload": encoded, - "prev_hash": prev_hash, - } - row_hash = chain_hash(body) + # The payload is canonicalised ONCE and commits AS THAT STRING -- the exact bytes the column + # holds, not a re-serialisation of what a reader parsed back out of it. + # + # Two things follow, and the first is the reason. A hash taken over a parsed-then-re-encoded + # object puts a decode/encode round trip between the stored row and its hash, and every such + # conversion is somewhere the two can drift: a payload shape that does not survive the round + # trip byte-for-byte (a non-string key, a tuple, anything `json` normalises) would store one + # thing and attest to another. Hashing the bytes removes the question rather than answering it + # shape by shape. The second is that verification then never PARSES: measured over 20,000 + # events, `chain_state` went from 277 ms to 134 ms, on the read a 15-second console poll makes. + payload_json = canonical_json(payload) + row_hash = chain_hash(_hash_body(ts, event_type, entity_id, payload_json, prev_hash)) cursor = conn.execute( "INSERT INTO audit_events (ts, event_type, entity_id, payload_json, prev_hash, row_hash) " "VALUES (?, ?, ?, ?, ?, ?)", - ( - ts, - event_type, - entity_id, - json.dumps(encoded, sort_keys=True, separators=(",", ":")), - prev_hash, - row_hash, - ), + (ts, event_type, entity_id, payload_json, prev_hash, row_hash), ) assert cursor.lastrowid is not None return AuditEvent( @@ -187,7 +188,7 @@ def append_event( ts=ts, event_type=event_type, entity_id=entity_id, - payload=encoded, + payload=json.loads(payload_json), prev_hash=prev_hash, row_hash=row_hash, ) @@ -213,8 +214,15 @@ def read_events(conn: sqlite3.Connection) -> list[AuditEvent]: ] -def _event_payload(event: AuditEvent) -> dict[str, Any]: - """Everything the row commits to -- the row minus `row_hash` itself. +def _hash_body( + ts: int, event_type: str, entity_id: str, payload_json: str, prev_hash: str +) -> dict[str, Any]: + """Everything a row commits to -- the row minus `row_hash` itself. + + `payload_json` is the STORED STRING, not a parsed object: see `append_event` for why the + bytes rather than a re-encoding of them. Taken as five scalars rather than as an `AuditEvent` + so the writer and the verifier compute the identical body from the identical inputs, with no + parse on the verification path. `seq_id` is deliberately NOT hashed. It is sqlite's own AUTOINCREMENT counter, assigned after the hash would have to be computed, and position in the chain is already committed to via @@ -222,38 +230,14 @@ def _event_payload(event: AuditEvent) -> dict[str, Any]: values a writer actually chose. """ return { - "ts": event.ts, - "event_type": event.event_type, - "entity_id": event.entity_id, - "payload": event.payload, - "prev_hash": event.prev_hash, + "ts": ts, + "event_type": event_type, + "entity_id": entity_id, + "payload": payload_json, + "prev_hash": prev_hash, } -def verify_events(events: list[AuditEvent]) -> list[str]: - """Every break in the chain, as human-readable lines. Empty means intact. - - Reports rather than raises, so `keel doctor` and the timeline export can both STATE chain - status instead of asserting it. **An empty list of events returns no errors, and that is not - the same as "verified"** -- nothing was checked. Only the caller knows whether that means a - deployment predating #721 or a deployment that has done nothing yet. - """ - return verify_links( - ChainLink( - label=str(event.seq_id), - prev_hash=event.prev_hash, - row_hash=event.row_hash, - recomputed_hash=chain_hash(_event_payload(event)), - ) - for event in events - ) - - -def latest_hashes(conn: sqlite3.Connection) -> dict[tuple[str, str], str]: - """`(store, entity_id) -> the row_hash of the LATEST event about it`. See `chain_state`.""" - return {key: seen.row_hash for key, seen in _latest(read_events(conn)).items()} - - @dataclass(frozen=True) class EntityHash: """The latest chained statement about one row, and where it sits in the chain.""" @@ -303,7 +287,7 @@ def intact(self) -> bool: return not self.errors and self.event_count > 0 -def _latest(events: list[AuditEvent]) -> dict[tuple[str, str], EntityHash]: +def _latest(rows: list[sqlite3.Row]) -> dict[tuple[str, str], EntityHash]: """`(store, entity_id) -> the LATEST event about it`. The latest rather than the first because it is the most recent chained statement about that @@ -315,15 +299,17 @@ def _latest(events: list[AuditEvent]) -> dict[tuple[str, str], EntityHash]: rows, and a single-keyed map would hand one row's hash to the other. """ latest: dict[tuple[str, str], EntityHash] = {} - for event in events: - store = EVENT_STORES.get(event.event_type) + for row in rows: + store = EVENT_STORES.get(str(row["event_type"])) if store is None: # An event type this build does not know -- a row written by a NEWER keel against the # same database. Skipped rather than guessed at: it still chains (the chain does not # care what the type means), and inventing a store for it would file it against a # table it may have nothing to do with. continue - latest[(store, event.entity_id)] = EntityHash(row_hash=event.row_hash, seq_id=event.seq_id) + latest[(store, str(row["entity_id"]))] = EntityHash( + row_hash=str(row["row_hash"]), seq_id=int(row["seq_id"]) + ) return latest @@ -337,9 +323,24 @@ def _table_present(conn: sqlite3.Connection) -> bool: def chain_state(conn: sqlite3.Connection) -> ChainState: """One read of `audit_events`, verified, indexed by the identifiers a report prints. - What `commands/timeline.py` calls. The `hashes` map is keyed by `(store, entity_id)` to match - that module's own `source`/`reference` pair, so the hash shown against a row is looked up by - the identifier printed beside it rather than by one a reader has to reconstruct. + What `commands/timeline.py` and `commands/doctor.py` call, and the only production path into + this module. The `hashes` map is keyed by `(store, entity_id)` to match `timeline.py`'s own + `source`/`reference` pair, so the hash shown against a row is looked up by the identifier + printed beside it rather than by one a reader has to reconstruct. + + **The whole chain, every time, and that is not an oversight.** A chain proves a SEQUENCE, so + a verdict over a suffix is a verdict that has not looked at the rows most likely to have been + quietly edited, and a cached prefix verdict assumes precisely the thing being checked. The + cost is real and is stated rather than hidden, in the manner `DEFAULT_TIMELINE_LIMIT` and + `export_rows` already use next door: measured on a 2026 laptop, 20,000 events verify in about + 135 ms, and the console polls `/api/timeline` every 15 seconds. Event volume is dominated by + `upsert_transaction` -- one event per imported CSV line -- so a long Coinbase history is what + puts a deployment into that band. + + If it ever stops fitting, the change is to BOUND WHAT IS CLAIMED and say so on the page -- a + verdict over the last N events, labelled as one. Never to cache the verdict, which would put + a green badge over rows nothing read: the exact failure the rest of this module is built to + refuse. """ if not _table_present(conn): # Checked rather than caught. A pre-v20 database is a legitimate deployment state, and a @@ -349,23 +350,37 @@ def chain_state(conn: sqlite3.Connection) -> ChainState: return ChainState( table_present=False, event_count=0, errors=(), first_broken_seq=None, hashes={} ) - events = read_events(conn) + # Raw rows, deliberately not `read_events`: verification needs the stored `payload_json` + # STRING, and parsing 20,000 payloads to re-encode them would double the cost of the check + # for a value nothing on this path uses. + rows = conn.execute( + "SELECT seq_id, ts, event_type, entity_id, payload_json, prev_hash, row_hash " + "FROM audit_events ORDER BY seq_id" + ).fetchall() breaks = find_breaks( ChainLink( - label=str(event.seq_id), - prev_hash=event.prev_hash, - row_hash=event.row_hash, - recomputed_hash=chain_hash(_event_payload(event)), + label=str(row["seq_id"]), + prev_hash=str(row["prev_hash"]), + row_hash=str(row["row_hash"]), + recomputed_hash=chain_hash( + _hash_body( + int(row["ts"]), + str(row["event_type"]), + str(row["entity_id"]), + str(row["payload_json"]), + str(row["prev_hash"]), + ) + ), ) - for event in events + for row in rows ) - # `index` is 1-based POSITION IN THE WALK, and `events` is read in `seq_id` order, so this is + # `index` is 1-based POSITION IN THE WALK, and the rows are read in `seq_id` order, so this is # the event the first break lands on. Looked up rather than parsed out of the message text. - first_broken_seq = events[breaks[0].index - 1].seq_id if breaks else None + first_broken_seq = int(rows[breaks[0].index - 1]["seq_id"]) if breaks else None return ChainState( table_present=True, - event_count=len(events), + event_count=len(rows), errors=tuple(found.message for found in breaks), first_broken_seq=first_broken_seq, - hashes=_latest(events), + hashes=_latest(rows), ) diff --git a/keel/data/repository.py b/keel/data/repository.py index 58d3de7..7100750 100644 --- a/keel/data/repository.py +++ b/keel/data/repository.py @@ -206,6 +206,20 @@ def _json_object_hook(d: dict[str, Any]) -> Any: return d +def _event_ts(value: Any) -> int: + """The timestamp an audit event carries, falling back to now ONLY when there is none (#721). + + `int(value or time.time())` was wrong in a small, silent way: `or` treats `0` as absent, so a + row legitimately stamped at the epoch would get an event stamped `now` -- and the event's `ts` + is hashed, so the chain would then attest to a timestamp the row does not hold. Every real + call site supplies a non-zero value today (checked: all five `update_order` callers pass + `updated_at=now_ts`, both `insert_order` callers pass `created_at`, and `transactions.ts` is + NOT NULL), so this is a guard against a future caller rather than a live bug -- which is + exactly when a silent fallback is worth making explicit. + """ + return int(time.time()) if value is None else int(value) + + class Repository: """Typed wrapper around a migrated `sqlite3.Connection` for the keel schema.""" @@ -214,6 +228,17 @@ def __init__(self, conn: sqlite3.Connection) -> None: # -- the audit chain ------------------------------------------------ + def rollback(self) -> None: + """Discard whatever this connection has open but uncommitted. + + For a caller that SWALLOWS a write failure. Every writer here is `execute` then `commit`, + so a failing `execute` leaves sqlite3's implicit transaction open; a caller that catches + the exception and carries on -- `execution/equity.py` does, so a diagnostic write cannot + abort a cycle -- would otherwise hand the next writer on this connection a dirty + transaction it did not open and does not know about. + """ + self._conn.rollback() + def audit_chain(self) -> ChainState: """The `audit_events` chain: its hashes, its condition, and where it stops being evidence. @@ -240,7 +265,7 @@ def upsert_transaction(self, tx: dict[str, Any]) -> None: f"{c} = excluded.{c}" for c in _TRANSACTION_COLUMNS if c != "coinbase_id" ) with write_transaction(self._conn): - cursor = self._conn.execute( + self._conn.execute( f""" INSERT INTO transactions ({columns_sql}) VALUES ({placeholders_sql}) @@ -248,24 +273,45 @@ def upsert_transaction(self, tx: dict[str, Any]) -> None: """, values, ) - # `coinbase_id or id`, which is EXACTLY the rule `commands/timeline.py:: - # _transaction_rows` uses for the `reference` it prints -- so the export looks the - # hash up by the identifier shown beside it. `coinbase_id` is nullable, and sqlite - # treats NULLs as distinct in a UNIQUE index, so a row without one can never take the - # DO UPDATE branch: `lastrowid` is a real insert's id in the only case that reads it. - coinbase_id = values.get("coinbase_id") # An APPEND even when the book row was updated in place (#721). Two events for one # `coinbase_id` is the record that the line was re-imported with different content -- # which, for a store whose provenance is `imported-ledger` and whose rows nothing # verified on the way in, is exactly what an auditor needs visible. append_event( self._conn, - ts=int(values.get("ts") or time.time()), + ts=_event_ts(values.get("ts")), event_type="transaction_recorded", - entity_id=str(coinbase_id) if coinbase_id else str(cursor.lastrowid), + entity_id=self._transaction_reference(values), payload=dict(values), ) + def _transaction_reference(self, values: dict[str, Any]) -> str: + """`coinbase_id or id` -- EXACTLY the rule `commands/timeline.py::_transaction_rows` uses + for the `reference` it prints, so the export looks a row's hash up by the identifier shown + beside it. + + The falsey branch READS THE ID BACK rather than taking `cursor.lastrowid`. The first cut + used `lastrowid` on the reasoning that `coinbase_id` is nullable and sqlite treats NULLs + as distinct in a UNIQUE index, so a row without one can never take the `DO UPDATE` branch + -- true of NULL, and false of the EMPTY STRING, which is falsey to Python and perfectly + indexable to sqlite. A re-upsert of a blank id therefore UPDATED an existing row while + `lastrowid` still held the id of the last real INSERT, filing the event under a different + row's identifier: the timeline then showed one row the hash of a superseded event, marked + `chained`, and a phantom key existed that no row resolved to. Not reachable through + `csv_import` today (`_row_to_tx` raises on a blank ID), and gated only by that one caller. + + A lookup answers NULL and blank alike instead of reasoning about which value reaches which + branch. The non-empty fast path stays because a CSV import runs this per line and a + `coinbase_id` that is present needs no query to be known. + """ + coinbase_id = values.get("coinbase_id") + if coinbase_id: + return str(coinbase_id) + row = self._conn.execute( + "SELECT id FROM transactions WHERE coinbase_id IS ?", (coinbase_id,) + ).fetchone() + return "" if row is None else str(row["id"]) + def get_transactions(self, asset: str | None = None) -> list[dict[str, Any]]: if asset is None: rows = self._conn.execute("SELECT * FROM transactions ORDER BY ts, id").fetchall() @@ -450,7 +496,7 @@ def insert_order(self, order: dict[str, Any]) -> int: order_id = cursor.lastrowid append_event( self._conn, - ts=int(values.get("created_at") or time.time()), + ts=_event_ts(values.get("created_at")), event_type="order_placed", entity_id=str(order_id), # The row AS STORED, id included -- the whole statement being made, so a later @@ -475,7 +521,7 @@ def update_order(self, order_id: int, **fields: Any) -> None: self._conn.execute(f"UPDATE orders SET {set_sql} WHERE id = :order_id", params) append_event( self._conn, - ts=int(fields.get("updated_at") or time.time()), + ts=_event_ts(fields.get("updated_at")), event_type="order_updated", entity_id=str(order_id), # WHAT CHANGED, not the whole mutated row. An event is a statement about this diff --git a/keel/execution/equity.py b/keel/execution/equity.py index d5b8618..38a9cad 100644 --- a/keel/execution/equity.py +++ b/keel/execution/equity.py @@ -310,6 +310,18 @@ def _append_equity_point( ) ) except Exception: + # ROLLED BACK, not merely logged. `record_cycle_balance` is `execute` then `commit`, + # so a failing `execute` leaves sqlite3's implicit transaction OPEN -- and swallowing + # the exception hands the next writer on this connection a dirty transaction it did + # not open. `Repository.insert_order` inherits exactly that connection a few + # milliseconds later, inside the same cycle. Leaving it dirty made a swallowed + # observability failure decide whether the ORDER ROW was durable, which is the same + # rule this `try` exists to enforce, violated from the other end. + # + # Safe to roll back here because the only uncommitted work is this failed INSERT: + # `record_equity_point` above committed its own row, and each `record_cycle_balance` + # commits per currency. + repo.rollback() log_exception(logger, "equity.cycle_balance_write_failed") diff --git a/tests/commands/test_timeline.py b/tests/commands/test_timeline.py index 99c6b9e..5ed3d29 100644 --- a/tests/commands/test_timeline.py +++ b/tests/commands/test_timeline.py @@ -517,10 +517,10 @@ def test_the_hash_shown_is_the_hash_recorded_for_that_row(db_conn) -> None: repo = _chained_repo(db_conn) report = timeline.gather_timeline(repo, now_ts=2_000) - recorded = audit.latest_hashes(db_conn) + recorded = audit.chain_state(db_conn).hashes for row in report.rows: if row.chain_status == "chained": - assert row.row_hash == recorded[(row.source, row.reference)] + assert row.row_hash == recorded[(row.source, row.reference)].row_hash def test_a_row_written_before_the_chain_shipped_reads_as_not_chained(db_conn) -> None: @@ -607,8 +607,12 @@ def test_the_export_carries_the_chain_status_beside_the_hash(db_conn) -> None: rows = list(csv.reader(io.StringIO(text))) assert rows[0][-2:] == ["row_hash", "chain_status"] + # Every row, and the exact pairing -- `in (the three words)` would pass on any of them and so + # would survive the status being hard-coded to one value. + assert len(rows) == 4 for row in rows[1:]: - assert row[-1] in ("chained", "not chained", "chain broken") + assert row[-1] == "chained" + assert len(row[-2]) == 64 def test_a_broken_chain_is_stated_above_the_header_not_only_per_row(db_conn) -> None: diff --git a/tests/data/test_audit_chain.py b/tests/data/test_audit_chain.py index f6964c2..0058b74 100644 --- a/tests/data/test_audit_chain.py +++ b/tests/data/test_audit_chain.py @@ -16,7 +16,7 @@ from decimal import Decimal import pytest -from keel_core.hashchain import ZERO_HASH +from keel_core.hashchain import ZERO_HASH, chain_hash from keel.data import audit, db from keel.data.repository import Repository @@ -75,7 +75,7 @@ def test_each_event_commits_to_its_predecessor(conn: sqlite3.Connection) -> None first = _append(conn, ts=100, event_type="order_placed", entity_id="1", payload={"a": 1}) second = _append(conn, ts=101, event_type="order_updated", entity_id="1", payload={"a": 2}) assert second.prev_hash == first.row_hash - assert audit.verify_events(audit.read_events(conn)) == [] + assert audit.chain_state(conn).errors == () def test_the_chain_spans_event_types_not_one_chain_per_table(conn: sqlite3.Connection) -> None: @@ -87,7 +87,7 @@ def test_the_chain_spans_event_types_not_one_chain_per_table(conn: sqlite3.Conne ) events = audit.read_events(conn) assert second.prev_hash == events[0].row_hash - assert audit.verify_events(events) == [] + assert audit.chain_state(conn).errors == () def test_editing_a_payload_in_place_breaks_that_row_and_no_other(conn: sqlite3.Connection) -> None: @@ -102,7 +102,7 @@ def test_editing_a_payload_in_place_breaks_that_row_and_no_other(conn: sqlite3.C conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 2", ('{"a":99}',)) conn.commit() - errors = audit.verify_events(audit.read_events(conn)) + errors = audit.chain_state(conn).errors assert len(errors) == 1 assert "row 2" in errors[0] assert "row_hash" in errors[0] @@ -121,7 +121,7 @@ def test_deleting_a_row_breaks_every_row_after_it(conn: sqlite3.Connection) -> N conn.execute("DELETE FROM audit_events WHERE seq_id = 2") conn.commit() - errors = audit.verify_events(audit.read_events(conn)) + errors = audit.chain_state(conn).errors assert len(errors) == 1, errors assert "row 2" in errors[0] assert "does not chain" in errors[0] @@ -142,7 +142,7 @@ def test_events_are_read_in_chain_order_not_timestamp_order(conn: sqlite3.Connec events = audit.read_events(conn) assert [event.seq_id for event in events] == [1, 2, 3] assert events[2].prev_hash == events[1].row_hash - assert audit.verify_events(events) == [] + assert audit.chain_state(conn).errors == () def test_a_decimal_in_the_payload_survives_the_round_trip(conn: sqlite3.Connection) -> None: @@ -158,7 +158,7 @@ def test_a_decimal_in_the_payload_survives_the_round_trip(conn: sqlite3.Connecti stored = audit.read_events(conn)[0] assert stored.payload["qty"] == "0.10" assert stored.row_hash == event.row_hash - assert audit.verify_events([stored]) == [] + assert audit.chain_state(conn).errors == () # -- the vocabulary and the guards ------------------------------------------------------------ @@ -219,7 +219,7 @@ def test_insert_order_records_a_chained_placement_event(conn: sqlite3.Connection assert [event.event_type for event in events] == ["order_placed"] assert events[0].entity_id == str(order_id) assert events[0].payload["product_id"] == "BTC-USD" - assert audit.verify_events(events) == [] + assert audit.chain_state(conn).errors == () def test_update_order_records_what_changed_not_the_whole_row(conn: sqlite3.Connection) -> None: @@ -232,7 +232,7 @@ def test_update_order_records_what_changed_not_the_whole_row(conn: sqlite3.Conne events = audit.read_events(conn) assert [event.event_type for event in events] == ["order_placed", "order_updated"] assert events[1].payload == {"status": "filled", "actual_fill": "42.5"} - assert audit.verify_events(events) == [] + assert audit.chain_state(conn).errors == () def test_an_update_with_no_fields_records_nothing(conn: sqlite3.Connection) -> None: @@ -250,7 +250,7 @@ def test_upsert_transaction_records_a_flow_event(conn: sqlite3.Connection) -> No events = audit.read_events(conn) assert [event.event_type for event in events] == ["transaction_recorded"] assert events[0].entity_id == "cb-1" - assert audit.verify_events(events) == [] + assert audit.chain_state(conn).errors == () def test_re_importing_a_transaction_appends_rather_than_rewrites(conn: sqlite3.Connection) -> None: @@ -265,7 +265,7 @@ def test_re_importing_a_transaction_appends_rather_than_rewrites(conn: sqlite3.C events = audit.read_events(conn) assert len(events) == 2 assert events[0].row_hash != events[1].row_hash - assert audit.verify_events(events) == [] + assert audit.chain_state(conn).errors == () def test_both_attestation_upserts_record_a_human_claim(conn: sqlite3.Connection) -> None: @@ -282,7 +282,7 @@ def test_both_attestation_upserts_record_a_human_claim(conn: sqlite3.Connection) assert [event.event_type for event in events] == ["asset_attested", "instrument_attested"] assert events[0].entity_id == "BTC" assert events[1].entity_id == "coinbase:BTC-USD" - assert audit.verify_events(events) == [] + assert audit.chain_state(conn).errors == () def test_the_store_row_and_its_event_land_together_or_not_at_all( @@ -324,7 +324,7 @@ def test_rows_written_before_the_bump_leave_no_event_and_that_is_not_a_break( events = audit.read_events(conn) assert len(events) == 1 assert events[0].entity_id == "2" - assert audit.verify_events(events) == [] + assert audit.chain_state(conn).errors == () def test_latest_hashes_reports_the_newest_event_per_entity(conn: sqlite3.Connection) -> None: @@ -334,9 +334,9 @@ def test_latest_hashes_reports_the_newest_event_per_entity(conn: sqlite3.Connect order_id = repo.insert_order(_order(product_id="BTC-USD")) repo.update_order(order_id, status="filled") - latest = audit.latest_hashes(conn) + latest = audit.chain_state(conn).hashes events = audit.read_events(conn) - assert latest[("orders", str(order_id))] == events[1].row_hash + assert latest[("orders", str(order_id))].row_hash == events[1].row_hash def test_latest_hashes_is_keyed_by_store_so_two_stores_cannot_collide( @@ -348,8 +348,8 @@ def test_latest_hashes_is_keyed_by_store_so_two_stores_cannot_collide( repo.insert_order(_order(product_id="BTC-USD")) repo.upsert_transaction(_transaction(coinbase_id="1")) - latest = audit.latest_hashes(conn) - assert latest[("orders", "1")] != latest[("transactions", "1")] + latest = audit.chain_state(conn).hashes + assert latest[("orders", "1")].row_hash != latest[("transactions", "1")].row_hash # -- readers that never migrate ---------------------------------------------------------------- @@ -406,3 +406,168 @@ def test_appending_to_a_missing_table_still_fails_loudly() -> None: with pytest.raises(sqlite3.OperationalError): repo.insert_order(_order()) assert connection.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 0 + + +# -- a swallowed write must not decide whether an order is durable ------------------------------ + + +def test_a_swallowed_diagnostic_failure_does_not_strand_the_next_order( + tmp_path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The bug the first cut of `write_transaction` shipped, end to end. + + `sqlite3`'s legacy mode leaves the implicit transaction OPEN after a DML that raised, so a + swallowed write leaves the connection dirty. `write_transaction` read that as "an outer + caller owns the commit", took a join branch nothing in this codebase ever legitimately + reaches, and returned from `insert_order` without committing -- while `broker.place_order` + went on to send the order to the venue. + + Asserted against ANOTHER CONNECTION, because the writing connection can see its own + uncommitted rows: "durable" means visible outside this transaction, not merely inserted. + """ + path = tmp_path / "keel.db" + connection = db.connect(path) + db.migrate(connection) + repo = Repository(connection) + + # A failed diagnostic write, swallowed exactly as `execution/equity.py` swallows it. + try: + connection.execute( + "INSERT INTO cycle_balances (ts, mode, currency) VALUES (?, ?, ?)", (1, "live", None) + ) + except sqlite3.IntegrityError: + pass + assert connection.in_transaction, "the premise: a swallowed write leaves the connection dirty" + + order_id = repo.insert_order(_order()) + + reader = db.connect(path) + reader.execute("PRAGMA busy_timeout = 100") + assert reader.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 1, ( + f"order {order_id} was returned to the executor but is not durable; " + "`broker.place_order` runs next" + ) + assert reader.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] == 1 + + +def test_the_repository_rolls_a_swallowed_write_back_rather_than_leaving_it_open() -> None: + """The fix at the source. `execution/equity.py` calls this after swallowing, so the dirty + connection never reaches the next writer at all -- `write_transaction`'s unconditional commit + is the second belt, not the only one.""" + connection = db.connect(":memory:") + db.migrate(connection) + repo = Repository(connection) + try: + connection.execute( + "INSERT INTO cycle_balances (ts, mode, currency) VALUES (?, ?, ?)", (1, "live", None) + ) + except sqlite3.IntegrityError: + pass + + repo.rollback() + assert not connection.in_transaction + + +def test_a_blank_coinbase_id_files_its_event_under_its_own_row(conn: sqlite3.Connection) -> None: + """`""` is falsey to Python and perfectly indexable to sqlite, so unlike NULL it CAN take the + `DO UPDATE` branch -- at which point `cursor.lastrowid` is the last real insert's id, not the + updated row's. The first cut filed the re-upsert's event under a different row's identifier: + the timeline then showed one row the hash of a superseded event, marked `chained`, and a + phantom key existed that no row resolved to.""" + repo = Repository(conn) + repo.upsert_transaction(_transaction(coinbase_id="", asset="USD")) + repo.upsert_transaction(_transaction(coinbase_id="X1", asset="EUR")) + repo.upsert_transaction(_transaction(coinbase_id="", asset="GBP")) + + blank_row_id = str( + conn.execute("SELECT id FROM transactions WHERE coinbase_id = ''").fetchone()["id"] + ) + events = audit.read_events(conn) + assert [event.entity_id for event in events] == [blank_row_id, "X1", blank_row_id] + + # And the latest hash for that row is its LATEST event, with no phantom key beside it. + hashes = audit.chain_state(conn).hashes + assert hashes[("transactions", blank_row_id)].row_hash == events[2].row_hash + assert set(hashes) == {("transactions", blank_row_id), ("transactions", "X1")} + + +def test_a_null_coinbase_id_still_files_under_the_row_id(conn: sqlite3.Connection) -> None: + """The case the `lastrowid` version got right, kept right by the lookup that replaced it.""" + repo = Repository(conn) + repo.upsert_transaction(_transaction(coinbase_id=None)) + row_id = str(conn.execute("SELECT id FROM transactions").fetchone()["id"]) + assert [event.entity_id for event in audit.read_events(conn)] == [row_id] + + +def test_an_epoch_timestamp_is_recorded_not_replaced_with_now(conn: sqlite3.Connection) -> None: + """`int(value or time.time())` treats `0` as absent. The event's `ts` is HASHED, so the chain + would then attest to a timestamp the row does not hold.""" + repo = Repository(conn) + repo.insert_order(_order(created_at=0)) + assert [event.ts for event in audit.read_events(conn)] == [0] + + +def test_a_genuinely_absent_timestamp_still_falls_back_to_now( + conn: sqlite3.Connection, monkeypatch: pytest.MonkeyPatch +) -> None: + """The fallback is for `None`, and only for `None`.""" + monkeypatch.setattr("keel.data.repository.time.time", lambda: 4_242.0) + repo = Repository(conn) + repo.insert_order(_order(created_at=None)) + assert [event.ts for event in audit.read_events(conn)] == [4_242] + + +def test_a_payload_commits_as_the_bytes_the_column_holds(conn: sqlite3.Connection) -> None: + """The hash is over the STORED string, not over a re-encoding of what a reader parsed back. + Editing the column by one character must therefore break the row, with no shape of payload + able to survive the round trip differently from how it was hashed.""" + event = _append( + conn, + ts=100, + event_type="order_placed", + entity_id="1", + payload={"qty": Decimal("1.10"), "note": "a", "nested": {"b": [Decimal("2.0")]}}, + ) + stored = conn.execute("SELECT payload_json FROM audit_events").fetchone()["payload_json"] + assert stored == '{"nested":{"b":["2.0"]},"note":"a","qty":"1.10"}' + assert event.row_hash == chain_hash( + { + "ts": 100, + "event_type": "order_placed", + "entity_id": "1", + "payload": stored, + "prev_hash": ZERO_HASH, + } + ) + + conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 1", (stored + " ",)) + conn.commit() + assert audit.chain_state(conn).errors + + +def test_an_events_entity_id_is_what_the_timeline_prints_as_the_rows_reference( + conn: sqlite3.Connection, +) -> None: + """THE invariant, rather than either side's implementation of it. + + `commands/timeline.py` looks a row's hash up by `(source, reference)`. If the writer files an + event under any other identifier, the lookup misses and the row silently reads `not chained` + -- or, worse, hits a DIFFERENT row's event. Asserted across all three `coinbase_id` shapes, + because that is where the two sides can disagree. + """ + from keel.commands import timeline + + repo = Repository(conn) + repo.upsert_transaction(_transaction(coinbase_id="cb-1", asset="USD")) + repo.upsert_transaction(_transaction(coinbase_id=None, asset="EUR")) + repo.upsert_transaction(_transaction(coinbase_id="", asset="GBP")) + + report = timeline.gather_timeline(repo, now_ts=1_000) + flows = [row for row in report.rows if row.source == "transactions"] + assert len(flows) == 3 + for row in flows: + assert row.chain_status == "chained", f"{row.reference} lost its event" + + printed = {row.reference for row in flows} + recorded = {event.entity_id for event in audit.read_events(conn)} + assert printed == recorded diff --git a/tests/execution/test_equity.py b/tests/execution/test_equity.py index c56fe79..00b65ea 100644 --- a/tests/execution/test_equity.py +++ b/tests/execution/test_equity.py @@ -437,3 +437,53 @@ def _boom(*_a, **_k): assert repo.get_equity_points(mode="live", limit=1), "the equity point still landed" + + +def test_a_swallowed_balance_write_leaves_the_connection_clean(tmp_path) -> None: + """#721. `record_cycle_balance` is `execute` then `commit`, so a failing `execute` leaves + sqlite3's implicit transaction OPEN -- and this function swallows that failure by design, so + a diagnostic write cannot abort a cycle. + + Swallowing without rolling back handed the next writer on this connection a dirty transaction + it did not open. `Repository.insert_order` inherits exactly that connection later in the same + cycle, and the measured consequence was an order that reached the venue with no durable row + behind it. The swallow is still a swallow; it just no longer leaks. + """ + import sqlite3 + + from keel.data.db import connect, migrate + from keel.data.repository import Repository + from keel.execution import equity as equity_mod + + conn = connect(str(tmp_path / "keel.db")) + migrate(conn) + repo = Repository(conn) + repo.set_state("equity_state_mode", "live") + + real = repo.record_cycle_balance + + def _fail(reading): # type: ignore[no-untyped-def] + # Fail the way a real write fails: through sqlite, leaving the transaction open. + conn.execute( + "INSERT INTO cycle_balances (ts, mode, currency) VALUES (?, ?, ?)", (1, "live", None) + ) + raise AssertionError("unreachable -- the INSERT above raises") + + repo.record_cycle_balance = _fail # type: ignore[method-assign] + try: + equity_mod.update_drawdown( + repo, + equity=Decimal("100"), + now_ts=1_000, + cash=Decimal("100"), + balances=[("USD", Decimal("100"), Decimal("100"))], + ) + finally: + repo.record_cycle_balance = real # type: ignore[method-assign] + + assert not conn.in_transaction, "a swallowed write left the connection mid-transaction" + # And the reading the rail needs is still durable, read from another connection. + other = connect(str(tmp_path / "keel.db")) + other.execute("PRAGMA busy_timeout = 100") + assert other.execute("SELECT COUNT(*) FROM equity_points").fetchone()[0] == 1 + assert isinstance(conn, sqlite3.Connection)