Use IB UTC syntax for history `endDateTime`

Serialize aware reverse-fill boundaries as `YYYYMMDD-HH:MM:SS`
before handing them to `ib_async`.

This avoids Gateway error 10314 from its trailing-`UTC` format and
drops the unfinished EST comparison breakpoint from `Client.bars()`.

Prompt-IO: ai/prompt-io/opencode/20260729T034106Z_922a5df8_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

.claude/skills/run-tests/test-harness-reference.md

@@ -104,6 +104,9 @@ default.
 Deterministic or local first-pass targets:
 
 - `tests/test_watchlists.py`
+- `tests/test_storage_audit.py`
+- `tests/test_backfill_audit_snippet.py`
+- `tests/test_ib_history.py`
 - `tests/test_accounting.py::test_account_file_default_empty`
 - `tests/test_services.py::test_runtime_boot`
 - `tests/test_services.py::test_datad_spawn`
@@ -186,8 +189,12 @@ tests/
   test_dpi_font.py          Qt DPI/font behavior
   test_ems.py               actor, EMS, and paper-position behavior
   test_feeds.py             live Binance/Kraken feeds and shared memory
+  test_ib_history.py        deterministic IB history request formatting
   test_questrade.py         obsolete credentialed tests; skipped
   test_services.py          pikerd/datad/feed/EMS actor lifecycle
+  test_storage_audit.py     read-only NativeDB audit and JSON CLI
+  test_backfill_audit_snippet.py
+                            disposable xonsh qualification helpers
   test_watchlists.py        deterministic watchlist JSON operations
 ```
 
@@ -196,6 +203,9 @@ tests/
 | Changed area | Run first | Caveat |
 |---|---|---|
 | `piker/watchlists/` | `tests/test_watchlists.py` | CLI suite is skipped |
+| `piker/storage/_audit.py`, `piker/storage/cli.py` | `tests/test_storage_audit.py` | direct Typer app, no actor |
+| `snippets/nativedb_backfill_audit.xsh` | `tests/test_backfill_audit_snippet.py` | disposable paths only |
+| `piker/brokers/ib/api.py` history formatting | `tests/test_ib_history.py` | fake client, no network |
 | `piker/config.py` | `test_account_file_default_empty` | root-network test has a known mismatch |
 | `piker/accounting/` | targeted accounting node | some cases use live/configured state |
 | `piker/ui/_style.py`, `piker/ui/qt.py` | `tests/test_dpi_font.py` | GUI/config-isolated opt-in |

ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.md

@@ -0,0 +1,42 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: 24fb9765-a550-4570-8350-f0fc9b7e17db
+timestamp: 2026-07-28T00:32:47Z
+git_ref: ed85721c
+scope: code
+substantive: true
+raw_file: 20260728T003247Z_ed85721c_prompt_io.raw.md
+---
+
+## Prompt
+
+Build a safe xonsh workflow for qualifying NativeDB and reverse-backfill
+correctness against a known gappy IB chart, beginning with
+`mnq.cme.20260918`. Cover baseline, fresh, restart, append, and expansion
+testing through implementation, adversarial review, and verification.
+
+## Response summary
+
+Implemented a private disposable xonsh qualification workflow. Evidence is
+tied to a fixed case, report checksum, snapshot checksum, worktree import,
+and phase metadata. Added explicit path and symlink safety, source
+replacement handling, xonsh child environments, and lifecycle preservation
+comparisons.
+
+## Files changed
+
+- `snippets/nativedb_backfill_audit.xsh` - disposable qualification helpers
+- `docs/manual_backfill_qualification.rst` - operator lifecycle runbook
+- `tests/test_backfill_audit_snippet.py` - xonsh and comparison regressions
+- `.claude/skills/run-tests/test-harness-reference.md`
+  - qualification test-harness guidance
+- `ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not been edited by the human.

ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.raw.md

@@ -0,0 +1,32 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-28T00:32:47Z
+git_ref: ed85721c
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- snippets/nativedb_backfill_audit.xsh docs/manual_backfill_qualification.rst tests/test_backfill_audit_snippet.py .claude/skills/run-tests/test-harness-reference.md`
+
+Added sourceable xonsh helpers that create a private disposable root, bind
+child processes to the reviewed worktree, capture checksum-bound phase
+evidence, safely seed and clear test history, and compare lifecycle
+preservation.
+
+The workflow rejects path traversal, pre-existing symlinks, output
+collisions, mismatched case/report/snapshot evidence, and same-phase
+comparison. The accompanying runbook covers known-bad replay, fresh
+backfill, restart, append, gap triage, and expansion cases.
+
+Verification generated with the patch:
+
+- xonsh qualification helper regressions: passed
+- xonsh sourceability: passed
+- Ruff: passed
+- `git diff --check`: passed
+
+Adversarial review drove additional evidence binding, worktree executable,
+environment propagation, source replacement, and descriptor-level safety
+fixes. Residual concurrent mutation risk is stated explicitly in the private
+disposable-root runbook.

ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.md

@@ -0,0 +1,40 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: 24fb9765-a550-4570-8350-f0fc9b7e17db
+timestamp: 2026-07-28T00:50:34Z
+git_ref: ed85721c
+scope: code
+substantive: true
+raw_file: 20260728T005034Z_ed85721c_prompt_io.raw.md
+---
+
+## Prompt
+
+Add reusable read-only tooling to qualify raw NativeDB Parquet against a
+known gappy IB chart, beginning with `mnq.cme.20260918`. Preserve malformed
+evidence and distinguish structural validity from unclassified positive
+gaps without opening or mutating the storage runtime.
+
+## Response summary
+
+Implemented raw Parquet audit reporting with human and JSON output, bounded
+cadence details, exact integer arithmetic, strict status, collision-safe
+snapshot capture, and source path protections. Regressions cover malformed
+schema and values, timestamp precision, gap arithmetic, corrupt bytes,
+copy failures, symlinks, traversal, and read-only CLI behavior.
+
+## Files changed
+
+- `piker/storage/_audit.py` - raw Parquet audit and snapshot implementation
+- `piker/storage/cli.py` - `piker store audit` command
+- `tests/test_storage_audit.py` - audit and CLI regressions
+- `ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not been edited by the human.

ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.raw.md

@@ -0,0 +1,27 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-28T00:50:34Z
+git_ref: ed85721c
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/storage/_audit.py piker/storage/cli.py tests/test_storage_audit.py`
+
+Added a raw, read-only NativeDB Parquet audit with human and JSON output,
+bounded cadence evidence, exact integer arithmetic, pre-parse snapshot
+capture, and strict qualification status.
+
+The command rejects unsafe FQME paths, symlinked sources and outputs,
+destination collisions, non-regular source nodes, malformed canonical
+schema, non-finite values, invalid timestamps, noncanonical indexes, and
+sub-period cadence. Copy-failure cleanup preserves source bytes while
+completed corrupt snapshots remain available for diagnosis.
+
+Verification generated with the patch:
+
+- storage audit and CLI regressions: passed
+- Ruff: passed
+- `git diff --check`: passed
+- final adversarial review: no findings

ai/prompt-io/opencode/20260728T232718Z_ed85721c_prompt_io.md

@@ -0,0 +1,39 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: 24fb9765-a550-4570-8350-f0fc9b7e17db
+timestamp: 2026-07-28T23:27:18Z
+git_ref: ed85721c
+scope: code
+substantive: true
+raw_file: 20260728T232718Z_ed85721c_prompt_io.raw.md
+---
+
+## Prompt
+
+Diagnose the live `pikerd` failure where backfill publication raised
+`ValueError: OHLCV timestamps must be finite and positive` while merging the
+known gappy MNQ NativeDB baseline with fresh IB history. Determine whether to
+commit the pending audit stack or iterate first.
+
+## Response summary
+
+Traced the primary failure through `publish_latest_frame()` to
+`NativeStorageClient.update_ohlcv()`. The incoming frame was valid, but the
+canonicalized persisted baseline retained its epoch-zero row and caused the
+merged frame to fail validation. Added persisted-row repair before merge and
+a regression using the observed legacy schema characteristics.
+
+## Files changed
+
+- `piker/storage/nativedb.py` - drop non-positive persisted rows during merge
+- `tests/test_storage_nativedb.py` - live MNQ failure-shape regression
+- `ai/prompt-io/opencode/20260728T232718Z_ed85721c_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260728T232718Z_ed85721c_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not been edited by the human.

ai/prompt-io/opencode/20260728T232718Z_ed85721c_prompt_io.raw.md

@@ -0,0 +1,27 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-28T23:27:18Z
+git_ref: ed85721c
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/storage/nativedb.py tests/test_storage_nativedb.py`
+
+Fixed incremental NativeDB updates against legacy persisted history with
+non-positive timestamps. Incoming provider frames remain strictly validated,
+while merge repair discards invalid persisted rows before concat, dedupe,
+ordering, canonical-index rewrite, and atomic publication.
+
+The regression reproduces the observed MNQ baseline shape: one epoch-zero
+row, noncanonical absolute indexes, extra derived columns, and a valid fresh
+IB frame. It proves the invalid row is removed while all valid old and new
+bars survive with canonical indexes.
+
+Verification generated with the patch:
+
+- NativeDB regressions: 18 passed
+- history backfill regressions: 6 passed
+- focused Ruff: passed
+- `git diff --check`: passed

ai/prompt-io/opencode/20260729T034106Z_922a5df8_prompt_io.md

@@ -0,0 +1,40 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: 24fb9765-a550-4570-8350-f0fc9b7e17db
+timestamp: 2026-07-29T03:41:06Z
+git_ref: 922a5df8
+scope: code
+substantive: true
+raw_file: 20260729T034106Z_922a5df8_prompt_io.raw.md
+---
+
+## Prompt
+
+Diagnose the live IB reverse-fill failure where
+`reqHistoricalDataAsync()` raised API error 10314 because its end date,
+time, or timezone was invalid.
+
+## Response summary
+
+Traced the failure to `Client.bars()`, where an unfinished EST conversion
+left the original datetime for `ib_async` to serialize as
+`YYYYMMDD HH:MM:SS UTC`. Converted aware boundaries explicitly to UTC and
+passed IB's accepted `YYYYMMDD-HH:MM:SS` string while retaining a blank
+latest-request boundary. Added exact request-argument regressions.
+
+## Files changed
+
+- `piker/brokers/ib/api.py` - explicit UTC-dash history boundary format
+- `tests/test_ib_history.py` - IB request-format regressions
+- `.claude/skills/run-tests/test-harness-reference.md`
+  - deterministic IB history test mapping
+- `ai/prompt-io/opencode/20260729T034106Z_922a5df8_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260729T034106Z_922a5df8_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not been edited by the human.

ai/prompt-io/opencode/20260729T034106Z_922a5df8_prompt_io.raw.md

@@ -0,0 +1,25 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-29T03:41:06Z
+git_ref: 922a5df8
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py tests/test_ib_history.py .claude/skills/run-tests/test-harness-reference.md`
+
+Fixed IB reverse-history requests to serialize aware end boundaries with
+IB's explicit UTC-dash syntax, `YYYYMMDD-HH:MM:SS`. This bypasses
+`ib_async`'s trailing-`UTC` representation, which the live Gateway rejected
+with API error 10314, and removes the unfinished EST comparison breakpoint.
+
+The regression captures the exact `reqHistoricalDataAsync()` argument for
+latest, stdlib-aware, Pendulum UTC, and America/New_York inputs.
+
+Verification generated with the patch:
+
+- IB history request regressions: 4 passed
+- focused Ruff: passed
+- `git diff --check`: passed
+- adversarial review: no findings

docs/manual_backfill_qualification.rst

@@ -0,0 +1,285 @@
+Manual Backfill Qualification
+=============================
+
+Purpose
+-------
+
+This runbook separates provider coverage, durable NativeDB state,
+shared-memory hydration, and chart rendering while reproducing gappy
+history. Work one FQME and timeframe at a time. Preserve the original
+Parquet first, then qualify both repair of that evidence and a completely
+fresh backfill.
+
+The read-only command used throughout is::
+
+    piker store audit FQME --period 60
+    piker store audit FQME --period 60 --json
+
+It reads raw Parquet directly. It does not open a storage client,
+canonicalize data, publish cache state, attach SHM, or rewrite history.
+Positive gaps are deliberately unclassified; an observed interval is not
+automatically corruption or an expected venue closure.
+
+.. warning::
+
+   Do not run qualification against normal user storage. The helper below
+   requires a marked disposable root before it will seed, archive, or clear
+   a Parquet. ``piker store anal`` and ``piker store ldshm`` are interactive
+   and potentially mutating; they are not audit commands.
+
+   Keep the disposable root private and do not rename, replace, or symlink
+   files under it while a helper is running. Stop the chart cleanly before
+   every audit or comparison phase.
+
+Load The Xonsh Helpers
+----------------------
+
+Run from the repository root in the same xonsh which will launch the chart
+or daemon::
+
+    source snippets/nativedb_backfill_audit.xsh
+
+Choose a disposable root and copy only the provider configuration needed
+for the case::
+
+    run_root = p'/tmp/piker-backfill/mnq-20260918-60s'
+    source_config = p'~/.config/piker'.expanduser()
+    bfq_init(run_root, source_config)
+    bfq_record_case(
+        run_root,
+        fqme='mnq.cme.20260918.ib',
+        period_s=60,
+        provider='ib',
+        symptom='gappy chart during reverse backfill',
+        bad_start_utc=None,
+        bad_end_utc=None,
+    )
+
+``bfq_init()`` sets ``XDG_CONFIG_HOME`` in the current xonsh, so every
+subsequent ``piker`` and ``pikerd`` child inherits the disposable config.
+The effective storage path is under ``RUN_ROOT/xdg/piker/nativedb``.
+It also records the exact repository root and ``piker`` executable in the
+qualification marker, prepends that root to child ``PYTHONPATH``, and
+verifies that the bound executable exposes ``piker store audit``. Audits
+fail if that executable imports another tree.
+
+The copied ``brokers.toml`` can contain credentials. Keep the run root
+private and remove it manually after evidence is no longer needed.
+
+Repository Preflight
+--------------------
+
+Before every daemon start, verify the code and config inherited by child
+processes::
+
+    git status --short --branch
+    git rev-parse HEAD
+    python -c 'import piker; print(piker.__file__)'
+    python -c 'import os; print(os.environ["XDG_CONFIG_HOME"])'
+
+Stop any daemon from the previous phase cleanly. Use a separate registry
+address in the disposable ``conf.toml`` when a normal Piker stack remains
+active.
+
+Known-Bad Replay
+----------------
+
+Copy the original file; never move it out of normal storage::
+
+    fqme = 'mnq.cme.20260918.ib'
+    period_s = 60
+    source_parquet = (
+        p'~/.config/piker/nativedb'.expanduser()
+        /f'{fqme}.ohlcv{period_s}s.parquet'
+    )
+    bfq_seed(run_root, fqme, period_s, source_parquet)
+    bfq_audit(run_root, 'baseline', fqme, period_s)
+
+The baseline phase writes::
+
+    evidence/baseline.json
+    evidence/baseline.parquet
+    evidence/baseline.manual.json
+
+The JSON checksum must match the read-only snapshot, and the manual sidecar
+binds that report to its phase and case. Comparison rechecks those bindings
+before reading either phase. The report records raw schema, row and
+timestamp counts, non-finite values, ordering, duplicate excess, canonical
+index state, and every retained positive-gap detail. Files with invalid
+Parquet encoding can not produce a JSON report, but ``--snapshot`` still
+preserves their exact bytes before parse failure.
+
+Launch the chart manually from the same shell and capture logs::
+
+    piker -l info chart mnq.cme.20260918.ib \
+        2>&1 | tee @(run_root / 'logs' / 'repair-chart.log')
+
+After a clean stop::
+
+    bfq_audit(
+        run_root,
+        'repair',
+        fqme,
+        period_s,
+        require_structural=True,
+    )
+    bfq_compare(run_root, 'baseline', 'repair')
+
+The comparison fails if structural validity regresses, any original
+timestamp disappears, chronological endpoints shrink, timestamps become
+duplicated or unordered, indexes cease to be canonical, or OHLCV becomes
+non-finite. Provider replacement of matching seam rows is reported
+separately as ``changed_common_timestamps`` for operator review.
+
+Fresh Backfill
+--------------
+
+Yes, fresh qualification means saving the known-bad evidence and then
+starting with no active Parquet in the disposable NativeDB. Archive only
+the disposable file::
+
+    bfq_clear_for_fresh(run_root, fqme, period_s)
+
+Confirm the printed active path is absent, launch the same chart, wait for
+bounded provider exhaustion or the intended coverage, then stop cleanly::
+
+    bfq_audit(
+        run_root,
+        'fresh',
+        fqme,
+        period_s,
+        require_structural=True,
+    )
+
+Do not compare ``baseline`` to ``fresh`` as a timestamp-preservation gate:
+provider limits may intentionally produce a different initial extent.
+Treat ``fresh`` as the baseline for restart and append qualification.
+
+Restart And Append
+------------------
+
+Restart the same chart against the fresh persisted file, stop cleanly, and
+record::
+
+    bfq_audit(
+        run_root,
+        'restart',
+        fqme,
+        period_s,
+        require_structural=True,
+    )
+    bfq_compare(run_root, 'fresh', 'restart')
+
+Leave a subsequent run active long enough to append new samples, stop, and
+record::
+
+    bfq_audit(
+        run_root,
+        'append',
+        fqme,
+        period_s,
+        require_structural=True,
+    )
+    bfq_compare(
+        run_root,
+        'restart',
+        'append',
+        require_newer=True,
+    )
+
+The comparison JSON reports gaps added and removed. Expected closures stay
+visible until a venue-aware classifier proves their session alignment.
+
+Gap Layer Triage
+----------------
+
+Use the persisted report and chart together:
+
+===============================  =========================================
+Observation                      Suspected layer
+===============================  =========================================
+Parquet contiguous, chart gappy  SHM hydration, sampling, or chart render
+Parquet gap, provider has bars    provider-delta or storage merge
+Provider and Parquet both gappy  provider omission or venue closure
+Gap vanishes after restart       SHM publication or cache lifecycle
+Old timestamps disappear         destructive persistence regression
+Newest seam duplicates           conflict or dedupe policy
+===============================  =========================================
+
+Always record exact UTC gap endpoints. The audit's numeric epoch is
+authoritative; the ``*_utc`` fields are deterministic display values.
+
+Expansion Matrix
+----------------
+
+Expand only after the fixed FQME passes fresh, restart, and append. Use a
+new disposable root and case record for every row:
+
+====================  ==========  ========================================
+Case                  Period      Coverage model
+====================  ==========  ========================================
+Original IB future    60 seconds  CME maintenance and weekend closures
+Original IB future    1 second    high-frequency provider limits
+Second CME future     60 seconds  contract-specific history
+IB equity             60 seconds  regular overnight and weekend closure
+IB FX                 60 seconds  24x5 session behavior
+Crypto spot/perp      60 seconds  continuous-session control
+Expired future        60 seconds  history-only and expiry behavior
+====================  ==========  ========================================
+
+A sourceable matrix can remain ordinary Python data in xonsh::
+
+    cases = [
+        {
+            'name': 'mnq-60s',
+            'fqme': 'mnq.cme.20260918.ib',
+            'period_s': 60,
+            'provider': 'ib',
+        },
+        {
+            'name': 'mnq-1s',
+            'fqme': 'mnq.cme.20260918.ib',
+            'period_s': 1,
+            'provider': 'ib',
+        },
+    ]
+    for case in cases:
+        case_root = p'/tmp/piker-backfill' / case['name']
+        bfq_init(case_root, source_config)
+        bfq_record_case(
+            case_root,
+            fqme=case['fqme'],
+            period_s=case['period_s'],
+            provider=case['provider'],
+            symptom='manual expansion case',
+        )
+
+This loop prepares isolated case roots only. Run the complete seed or
+fresh, chart, audit, restart, append, and compare lifecycle in each root;
+directory creation alone is not qualification.
+
+The chart path may allocate companion 1-second and 60-second files. Audit
+the selected case path explicitly and retain companion files as secondary
+evidence; do not imply that the live request isolated one timeframe.
+
+Interruption Qualification
+--------------------------
+
+After normal restart is stable, interrupt one phase with graceful Ctrl-C,
+restart, and compare against the last clean evidence. Provider disconnect,
+SIGTERM, and deterministic write-barrier failures come next. Avoid timing a
+SIGKILL by hand; add an explicit failpoint before qualifying atomic replace
+boundaries.
+
+Result Interpretation
+---------------------
+
+``structural_ok`` means the raw persisted file satisfies durable schema,
+finite values, positive unique ordered timestamps, canonical index, and
+period-aligned gaps. ``gap_free`` means no positive gaps were observed when
+at least two valid timestamps made cadence verifiable; otherwise it is
+``null``. ``qualification_ok`` combines both and is intentionally strict.
+
+A structurally valid session-market file can have ``gap_free = false``.
+Those intervals require provider/session evidence before being marked as
+expected closures. Unknown calendar state must stay unknown.

piker/brokers/ib/api.py

@@ -29,7 +29,10 @@ from dataclasses import (
     asdict,
     astuple,
 )
-from datetime import datetime
+from datetime import (
+    UTC,
+    datetime,
+)
 from functools import (
     partial,
 )
@@ -395,15 +398,13 @@ class Client:
             default_dt_duration
         )
 
-        # TODO: maybe remove all this?
         global _enters
         if end_dt is None:
-            end_dt: str = ''
-
+            end_date_time: str = ''
         else:
-            est_end_dt = end_dt.in_tz('EST')
-            if est_end_dt != end_dt:
-                breakpoint()
+            end_date_time = end_dt.astimezone(UTC).strftime(
+                '%Y%m%d-%H:%M:%S'
+            )
 
         _enters += 1
 
@@ -412,7 +413,7 @@ class Client:
 
         kwargs: dict[str, Any] = dict(
             contract=contract,
-            endDateTime=end_dt,
+            endDateTime=end_date_time,
             formatDate=2,
 
             # OHLC sampling values:

piker/storage/_audit.py

@@ -0,0 +1,652 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+'''
+Read-only OHLCV persistence qualification.
+
+'''
+from datetime import (
+    UTC,
+    datetime,
+)
+from hashlib import sha256
+import os
+from pathlib import Path
+import shutil
+from stat import S_ISREG
+from typing import BinaryIO
+
+import numpy as np
+import polars as pl
+
+from piker.data import def_iohlcv_fields
+
+
+AUDIT_SCHEMA: str = 'piker.nativedb.audit/v1'
+_value_fields: tuple[str, ...] = (
+    'open',
+    'high',
+    'low',
+    'close',
+    'volume',
+)
+
+
+def _utc_str(timestamp: float|int|None) -> str|None:
+    '''
+    Render an epoch timestamp as deterministic UTC.
+
+    '''
+    if timestamp is None:
+        return None
+    try:
+        dt: datetime = datetime.fromtimestamp(timestamp, tz=UTC)
+    except (
+        OSError,
+        OverflowError,
+        ValueError,
+    ):
+        return None
+    return dt.isoformat().replace('+00:00', 'Z')
+
+
+def _number(value: float|int) -> float|int:
+    '''
+    Convert a NumPy number to a JSON-native scalar.
+
+    '''
+    if isinstance(value, (int, np.integer)):
+        return int(value)
+    value = float(value)
+    return int(value) if value.is_integer() else value
+
+
+def _sha256(file: BinaryIO) -> str:
+    '''
+    Hash and rewind one already-open file snapshot.
+
+    '''
+    digest = sha256()
+    for chunk in iter(
+        lambda: file.read(1024 * 1024),
+        b'',
+    ):
+        digest.update(chunk)
+    file.seek(0)
+    return digest.hexdigest()
+
+
+def _numeric_counts(series: pl.Series) -> dict[str, int|None]:
+    '''
+    Count null and non-finite values without coercing text.
+
+    '''
+    null_count: int = series.null_count()
+    if not series.dtype.is_numeric():
+        return {
+            'null': null_count,
+            'nan': None,
+            'positive_infinity': None,
+            'negative_infinity': None,
+            'nonfinite': None,
+        }
+
+    numeric: pl.Series = series.cast(pl.Float64, strict=False)
+    nan_count: int = int(numeric.is_nan().sum() or 0)
+    pos_inf_count: int = int(
+        (numeric == float('inf')).sum() or 0
+    )
+    neg_inf_count: int = int(
+        (numeric == float('-inf')).sum() or 0
+    )
+    cast_null_count: int = max(
+        0,
+        numeric.null_count() - null_count,
+    )
+    return {
+        'null': null_count,
+        'nan': nan_count,
+        'positive_infinity': pos_inf_count,
+        'negative_infinity': neg_inf_count,
+        'nonfinite': (
+            null_count
+            + nan_count
+            + pos_inf_count
+            + neg_inf_count
+            + cast_null_count
+        ),
+    }
+
+
+def _numeric_values(
+    series: pl.Series,
+) -> tuple[np.ndarray, np.ndarray]:
+    '''
+    Preserve integer precision and return an explicit valid mask.
+
+    '''
+    if series.dtype.is_integer():
+        valid: np.ndarray = series.is_not_null().to_numpy()
+        values: np.ndarray = series.fill_null(0).to_numpy()
+        return values, valid
+
+    values = series.to_numpy()
+    try:
+        valid = np.isfinite(values)
+    except TypeError:
+        values = series.cast(pl.Float64).to_numpy()
+        valid = np.isfinite(values)
+    return values, valid
+
+
+def _empty_gaps(
+    period_s: int,
+    verifiable: bool = False,
+) -> dict:
+    '''
+    Return the stable gap report shape for unauditable timestamps.
+
+    '''
+    return {
+        'basis': 'sorted_unique_valid_timestamps',
+        'expected_period_s': period_s,
+        'verifiable': verifiable,
+        'count': 0,
+        'aligned_count': 0,
+        'misaligned_count': 0,
+        'subperiod_count': 0,
+        'missing_samples_total': 0,
+        'details_count': 0,
+        'details_truncated': False,
+        'intervals': [],
+        'subperiod_intervals': [],
+    }
+
+
+def _audit_timestamps(
+    df: pl.DataFrame,
+    period_s: int,
+    max_gaps: int,
+
+) -> tuple[dict, dict]:
+    '''
+    Audit physical timestamp ordering and chronological gaps.
+
+    '''
+    if 'time' not in df.columns:
+        return (
+            {
+                'present': False,
+                'numeric': False,
+                'strictly_increasing': False,
+            },
+            _empty_gaps(period_s),
+        )
+
+    series: pl.Series = df['time']
+    counts: dict[str, int|None] = _numeric_counts(series)
+    if not series.dtype.is_numeric():
+        return (
+            {
+                'present': True,
+                'numeric': False,
+                **counts,
+                'strictly_increasing': False,
+            },
+            _empty_gaps(period_s),
+        )
+
+    values, finite = _numeric_values(series)
+    valid: np.ndarray = values[finite]
+    unique: np.ndarray = np.unique(valid)
+    adjacent_valid: np.ndarray = finite[:-1] & finite[1:]
+    deltas: np.ndarray = np.diff(values)
+    physical_deltas: np.ndarray = deltas[adjacent_valid]
+
+    zero_delta_count: int = int(np.count_nonzero(
+        physical_deltas == 0
+    ))
+    negative_delta_count: int = int(np.count_nonzero(
+        physical_deltas < 0
+    ))
+    first: float|int|None = None
+    last: float|int|None = None
+    if values.size:
+        if finite[0]:
+            first = _number(values[0])
+        if finite[-1]:
+            last = _number(values[-1])
+
+    minimum: float|int|None = None
+    maximum: float|int|None = None
+    if valid.size:
+        minimum = _number(np.min(valid))
+        maximum = _number(np.max(valid))
+
+    chronological: np.ndarray = unique[unique > 0]
+    gap_deltas: np.ndarray = np.diff(chronological)
+    gap_indexes: np.ndarray = np.flatnonzero(
+        gap_deltas > period_s
+    )
+    gap_values: np.ndarray = gap_deltas[gap_indexes]
+    integer_gaps: bool = np.issubdtype(
+        gap_values.dtype,
+        np.integer,
+    )
+    if integer_gaps:
+        aligned: np.ndarray = gap_values % period_s == 0
+    else:
+        aligned = np.isclose(
+            gap_values % period_s,
+            0,
+        )
+    aligned_count: int = int(np.count_nonzero(aligned))
+    if integer_gaps:
+        missing_total: int = sum(
+            int(delta) // period_s - 1
+            for delta in gap_values[aligned]
+        )
+    else:
+        missing_total = sum(
+            int(round(float(delta) / period_s)) - 1
+            for delta in gap_values[aligned]
+        )
+    subperiod_indexes: np.ndarray = np.flatnonzero(
+        gap_deltas < period_s
+    )
+    intervals: list[dict] = []
+    for gap_index in gap_indexes[:max_gaps]:
+        left: float = chronological[gap_index]
+        right: float = chronological[gap_index + 1]
+        delta: float = right - left
+        period_multiple: bool = bool(np.isclose(
+            delta % period_s,
+            0,
+        ))
+        missing_samples: int|None = None
+        if period_multiple:
+            missing_samples = (
+                int(delta) // period_s - 1
+                if integer_gaps
+                else int(round(float(delta) / period_s)) - 1
+            )
+
+        left_value: float|int = _number(left)
+        right_value: float|int = _number(right)
+        intervals.append({
+            'left_timestamp': left_value,
+            'right_timestamp': right_value,
+            'left_utc': _utc_str(left_value),
+            'right_utc': _utc_str(right_value),
+            'delta_s': _number(delta),
+            'period_multiple': period_multiple,
+            'missing_samples': missing_samples,
+            'classification': 'unclassified',
+        })
+
+    subperiod_intervals: list[dict] = []
+    remaining_details: int = max(0, max_gaps - len(intervals))
+    for step_index in subperiod_indexes[:remaining_details]:
+        left = _number(chronological[step_index])
+        right = _number(chronological[step_index + 1])
+        subperiod_intervals.append({
+            'left_timestamp': left,
+            'right_timestamp': right,
+            'left_utc': _utc_str(left),
+            'right_utc': _utc_str(right),
+            'delta_s': _number(right - left),
+        })
+
+    gap_count: int = gap_indexes.size
+    gaps: dict = {
+        'basis': 'sorted_unique_valid_timestamps',
+        'expected_period_s': period_s,
+        'verifiable': chronological.size >= 2,
+        'count': gap_count,
+        'aligned_count': aligned_count,
+        'misaligned_count': gap_count - aligned_count,
+        'missing_samples_total': missing_total,
+        'subperiod_count': subperiod_indexes.size,
+        'details_count': (
+            len(intervals)
+            + len(subperiod_intervals)
+        ),
+        'details_truncated': bool(
+            gap_count > len(intervals)
+            or
+            subperiod_indexes.size > len(subperiod_intervals)
+        ),
+        'intervals': intervals,
+        'subperiod_intervals': subperiod_intervals,
+    }
+    timestamps: dict = {
+        'present': True,
+        'numeric': True,
+        'first_in_file': first,
+        'last_in_file': last,
+        'minimum': minimum,
+        'maximum': maximum,
+        'minimum_utc': _utc_str(minimum),
+        'maximum_utc': _utc_str(maximum),
+        **counts,
+        'zero': int(np.count_nonzero(valid == 0)),
+        'negative': int(np.count_nonzero(valid < 0)),
+        'fractional': int(np.count_nonzero(
+            valid != np.floor(valid)
+        )),
+        'unique': unique.size,
+        'duplicate_excess': valid.size - unique.size,
+        'adjacent_zero_delta': zero_delta_count,
+        'negative_delta': negative_delta_count,
+        'non_positive_delta': (
+            zero_delta_count
+            + negative_delta_count
+        ),
+        'strictly_increasing': bool(
+            values.size > 0
+            and
+            values.size == valid.size
+            and
+            np.all(np.diff(values) > 0)
+        ),
+    }
+    return timestamps, gaps
+
+
+def _audit_index(df: pl.DataFrame) -> dict:
+    '''
+    Verify persisted indexes match physical row positions.
+
+    '''
+    if 'index' not in df.columns:
+        return {
+            'present': False,
+            'numeric': False,
+            'canonical_from_zero': False,
+        }
+
+    series: pl.Series = df['index']
+    counts: dict[str, int|None] = _numeric_counts(series)
+    if not series.dtype.is_numeric():
+        return {
+            'present': True,
+            'numeric': False,
+            'dtype': str(series.dtype),
+            **counts,
+            'canonical_from_zero': False,
+        }
+
+    values, finite = _numeric_values(series)
+    expected: np.ndarray = np.arange(df.height)
+    mismatch_count: int = int(np.count_nonzero(
+        ~finite
+        |
+        (values != expected)
+    ))
+    valid: np.ndarray = values[finite]
+    unique_count: int = np.unique(valid).size
+    return {
+        'present': True,
+        'numeric': True,
+        'dtype': str(series.dtype),
+        'first': (
+            _number(values[0])
+            if values.size and finite[0]
+            else None
+        ),
+        'last': (
+            _number(values[-1])
+            if values.size and finite[-1]
+            else None
+        ),
+        **counts,
+        'duplicate_excess': valid.size - unique_count,
+        'non_unit_step': int(np.count_nonzero(
+            np.diff(valid) != 1
+        )),
+        'row_position_mismatch': mismatch_count,
+        'contiguous': bool(
+            values.size > 0
+            and
+            valid.size == values.size
+            and
+            np.all(np.diff(values) == 1)
+        ),
+        'canonical_from_zero': bool(
+            values.size > 0
+            and
+            values.size == expected.size
+            and
+            mismatch_count == 0
+        ),
+    }
+
+
+def audit_ohlcv_frame(
+    df: pl.DataFrame,
+    fqme: str,
+    period_s: int,
+    max_gaps: int = 100,
+
+) -> dict:
+    '''
+    Inspect a raw persisted frame without normalizing evidence.
+
+    '''
+    if period_s < 1:
+        raise ValueError('Audit period must be positive')
+    if max_gaps < 0:
+        raise ValueError('Maximum gap details must be non-negative')
+
+    expected_columns: list[str] = [
+        name
+        for name, _ in def_iohlcv_fields
+    ]
+    expected_dtypes: dict[str, str] = {
+        name: str(pl.Int64 if field_type is int else pl.Float64)
+        for name, field_type in def_iohlcv_fields
+    }
+    actual_dtypes: dict[str, str] = {
+        name: str(dtype)
+        for name, dtype in df.schema.items()
+    }
+    missing: list[str] = sorted(
+        set(expected_columns).difference(df.columns)
+    )
+    extra: list[str] = sorted(
+        set(df.columns).difference(expected_columns)
+    )
+    column_order_ok: bool = df.columns == expected_columns
+    dtypes_ok: bool = all(
+        actual_dtypes.get(name) == dtype
+        for name, dtype in expected_dtypes.items()
+    )
+
+    timestamps, gaps = _audit_timestamps(
+        df,
+        period_s,
+        max_gaps,
+    )
+    index: dict = _audit_index(df)
+
+    nulls: dict[str, int|None] = {}
+    nans: dict[str, int|None] = {}
+    infinities: dict[str, int|None] = {}
+    all_values_finite: bool = True
+    for field in _value_fields:
+        if field not in df.columns:
+            nulls[field] = None
+            nans[field] = None
+            infinities[field] = None
+            all_values_finite = False
+            continue
+
+        counts = _numeric_counts(df[field])
+        nulls[field] = counts['null']
+        nans[field] = counts['nan']
+        pos_inf: int|None = counts['positive_infinity']
+        neg_inf: int|None = counts['negative_infinity']
+        infinities[field] = (
+            None
+            if pos_inf is None or neg_inf is None
+            else pos_inf + neg_inf
+        )
+        if counts['nonfinite'] != 0:
+            all_values_finite = False
+
+    schema_ok: bool = bool(
+        not missing
+        and
+        not extra
+        and
+        column_order_ok
+        and
+        dtypes_ok
+    )
+    violations: list[str] = []
+    if df.is_empty():
+        violations.append('empty_frame')
+    if missing:
+        violations.append('missing_columns')
+    if extra:
+        violations.append('extra_columns')
+    if not column_order_ok:
+        violations.append('column_order')
+    if not dtypes_ok:
+        violations.append('canonical_dtypes')
+    if not all_values_finite:
+        violations.append('nonfinite_ohlcv')
+    if timestamps.get('nonfinite') != 0:
+        violations.append('nonfinite_timestamps')
+    if (
+        timestamps.get('zero', 0)
+        or
+        timestamps.get('negative', 0)
+    ):
+        violations.append('nonpositive_timestamps')
+    if timestamps.get('fractional', 0):
+        violations.append('fractional_timestamps')
+    if timestamps.get('duplicate_excess', 0):
+        violations.append('duplicate_timestamps')
+    if not timestamps.get('strictly_increasing', False):
+        violations.append('timestamp_order')
+    if not index.get('canonical_from_zero', False):
+        violations.append('index_not_canonical')
+    if gaps.get('misaligned_count', 0):
+        violations.append('misaligned_time_gap')
+    if gaps.get('subperiod_count', 0):
+        violations.append('subperiod_time_step')
+
+    warnings: list[str] = []
+    if gaps['count']:
+        warnings.append('positive_time_gaps_unclassified')
+
+    structural_ok: bool = not violations
+    gap_free: bool|None = (
+        gaps['count'] == 0
+        if gaps['verifiable']
+        else None
+    )
+    return {
+        'audit_schema': AUDIT_SCHEMA,
+        'fqme': fqme,
+        'period_s': period_s,
+        'schema': {
+            'columns': df.columns,
+            'dtypes': actual_dtypes,
+            'expected_columns': expected_columns,
+            'expected_dtypes': expected_dtypes,
+            'missing_columns': missing,
+            'extra_columns': extra,
+            'column_order_ok': column_order_ok,
+            'canonical_dtypes_ok': dtypes_ok,
+            'canonical': schema_ok,
+        },
+        'rows': {
+            'count': df.height,
+            'empty': df.is_empty(),
+        },
+        'timestamps': timestamps,
+        'index': index,
+        'values': {
+            'null_by_column': nulls,
+            'nan_by_column': nans,
+            'infinity_by_column': infinities,
+            'all_finite': all_values_finite,
+        },
+        'gaps': gaps,
+        'result': {
+            'structural_ok': structural_ok,
+            'gap_free': gap_free,
+            'qualification_ok': bool(structural_ok and gap_free),
+            'violations': violations,
+            'warnings': warnings,
+        },
+    }
+
+
+def audit_ohlcv_parquet(
+    path: Path,
+    fqme: str,
+    period_s: int,
+    max_gaps: int = 100,
+    snapshot: Path|None = None,
+
+) -> dict:
+    '''
+    Read and audit one Parquet file without touching storage state.
+
+    '''
+    path = path.expanduser().absolute()
+    descriptor: int = os.open(
+        path,
+        os.O_RDONLY
+        | os.O_CLOEXEC
+        | os.O_NOFOLLOW
+        | os.O_NONBLOCK,
+    )
+    stat = os.fstat(descriptor)
+    if not S_ISREG(stat.st_mode):
+        os.close(descriptor)
+        raise ValueError('Audit source must be a regular file')
+    with os.fdopen(descriptor, 'rb') as file:
+        checksum: str = _sha256(file)
+        if snapshot is not None:
+            snapshot_is_symlink: bool = snapshot.is_symlink()
+            snapshot = snapshot.resolve()
+            if snapshot == path:
+                raise ValueError(
+                    'Snapshot path can not replace source'
+                )
+            if snapshot_is_symlink:
+                raise ValueError(
+                    'Snapshot path can not be a symlink'
+                )
+            file.seek(0)
+            created: bool = False
+            try:
+                with snapshot.open('xb') as snapshot_file:
+                    created = True
+                    os.fchmod(snapshot_file.fileno(), 0o444)
+                    shutil.copyfileobj(file, snapshot_file)
+            except BaseException:
+                if created:
+                    snapshot.unlink(missing_ok=True)
+                raise
+        file.seek(0)
+        frame: pl.DataFrame = pl.read_parquet(file)
+    report: dict = audit_ohlcv_frame(
+        frame,
+        fqme,
+        period_s,
+        max_gaps,
+    )
+    generated_at: float = datetime.now(tz=UTC).timestamp()
+    report['generated_at_utc'] = _utc_str(generated_at)
+    report['source'] = {
+        'path': str(path),
+        'size_bytes': stat.st_size,
+        'mtime_ns': stat.st_mtime_ns,
+        'sha256': checksum,
+    }
+    return report

piker/storage/cli.py

@@ -19,7 +19,9 @@ Storage middle-ware CLIs.
 
 """
 from __future__ import annotations
+import json
 from pathlib import Path
+import sys
 import time
 from types import ModuleType
 from typing import (
@@ -31,20 +33,25 @@ import numpy as np
 import tractor
 # import pendulum
 from rich.console import Console
+from rich.table import Table
 import trio
 # from rich.markdown import Markdown
 import typer
 
+import piker as piker_pkg
 from piker.service import open_piker_runtime
 from piker.cli import cli
 from tractor.ipc._shm import ShmArray
 from piker import tsp
+from piker import config
 from . import log
 from . import (
     __tsdbs__,
     open_storage_client,
     StorageClient,
 )
+from ._audit import audit_ohlcv_parquet
+from .nativedb import mk_ohlcv_shm_keyed_filepath
 
 if TYPE_CHECKING:
     from piker.ui._remote_ctl import AnnotCtl
@@ -53,6 +60,65 @@ if TYPE_CHECKING:
 store = typer.Typer()
 
 
+def _render_audit_report(report: dict) -> None:
+    '''
+    Render a compact human summary and explicit gap endpoints.
+
+    '''
+    result: dict = report['result']
+    timestamps: dict = report['timestamps']
+    gaps: dict = report['gaps']
+    source: dict = report['source']
+
+    fqme: str = report['fqme']
+    period_s: int = report['period_s']
+    table = Table(title=f'{fqme} @ {period_s}s')
+    table.add_column('Check')
+    table.add_column('Value')
+    table.add_row('Path', source['path'])
+    table.add_row('SHA-256', source['sha256'])
+    table.add_row('Rows', str(report['rows']['count']))
+    table.add_row('Minimum UTC', str(timestamps.get('minimum_utc')))
+    table.add_row('Maximum UTC', str(timestamps.get('maximum_utc')))
+    table.add_row('Structural OK', str(result['structural_ok']))
+    table.add_row('Gap free', str(result['gap_free']))
+    table.add_row('Gap count', str(gaps['count']))
+    table.add_row('Sub-period steps', str(gaps['subperiod_count']))
+    table.add_row('Gap details truncated', str(
+        gaps['details_truncated']
+    ))
+    violations: list[str] = result['violations']
+    table.add_row(
+        'Violations',
+        ', '.join(violations) if violations else 'none',
+    )
+
+    console = Console()
+    console.print(table)
+    intervals: list[dict] = [
+        *gaps['intervals'],
+        *gaps['subperiod_intervals'],
+    ]
+    if not intervals:
+        return
+
+    gap_table = Table(title='Timestamp cadence deviations')
+    gap_table.add_column('Left UTC')
+    gap_table.add_column('Right UTC')
+    gap_table.add_column('Delta (s)')
+    gap_table.add_column('Missing')
+    gap_table.add_column('Aligned')
+    for gap in intervals:
+        gap_table.add_row(
+            str(gap['left_utc']),
+            str(gap['right_utc']),
+            str(gap['delta_s']),
+            str(gap.get('missing_samples')),
+            str(gap.get('period_multiple')),
+        )
+    console.print(gap_table)
+
+
 @store.command()
 def ls(
     backends: list[str] = typer.Argument(
@@ -94,6 +160,170 @@ def ls(
     trio.run(query_all)
 
 
+@store.command()
+def audit(
+    fqme: str,
+    period: int = typer.Option(
+        60,
+        '--period',
+        min=1,
+        help='Expected sampling period in seconds.',
+    ),
+    json_output: bool = typer.Option(
+        False,
+        '--json',
+        help='Emit machine-readable JSON.',
+    ),
+    output: Path|None = typer.Option(
+        None,
+        '--output',
+        '-o',
+        help='Write JSON to a new file without replacing it.',
+    ),
+    snapshot: Path|None = typer.Option(
+        None,
+        '--snapshot',
+        help='Copy the exact audited bytes to a new file.',
+    ),
+    max_gaps: int = typer.Option(
+        100,
+        '--max-gaps',
+        min=0,
+        help='Maximum gap details to include.',
+    ),
+    strict: bool = typer.Option(
+        False,
+        '--strict',
+        help='Exit nonzero for structural defects or any gap.',
+    ),
+) -> None:
+    '''
+    Audit one NativeDB Parquet without normalizing or mutating it.
+
+    Positive gaps are reported as unclassified evidence. They are not
+    automatically treated as corruption or expected venue closures.
+
+    '''
+    if (
+        fqme in {'.', '..'}
+        or
+        Path(fqme).name != fqme
+        or
+        '/' in fqme
+        or
+        '\\' in fqme
+    ):
+        typer.echo(f'Unsafe FQME path component: {fqme!r}', err=True)
+        raise typer.Exit(code=2)
+    datadir: Path = config.get_conf_dir() / 'nativedb'
+    path: Path = mk_ohlcv_shm_keyed_filepath(
+        fqme,
+        period,
+        datadir,
+    )
+    destinations: list[Path] = [
+        destination
+        for destination in (output, snapshot)
+        if destination is not None
+    ]
+    resolved_source: Path = path.resolve()
+    if path.is_symlink():
+        typer.echo(
+            f'Refusing symlinked audit source: {path}',
+            err=True,
+        )
+        raise typer.Exit(code=2)
+    resolved_destinations: list[Path] = []
+    for destination in destinations:
+        if destination.is_symlink():
+            typer.echo(
+                f'Refusing symlinked output: {destination}',
+                err=True,
+            )
+            raise typer.Exit(code=2)
+        resolved: Path = destination.resolve()
+        if resolved == resolved_source:
+            typer.echo(
+                f'Refusing to replace audited source: {resolved}',
+                err=True,
+            )
+            raise typer.Exit(code=2)
+        if destination.exists():
+            typer.echo(
+                f'Refusing to replace existing output: '
+                f'{destination}',
+                err=True,
+            )
+            raise typer.Exit(code=2)
+        if not destination.parent.is_dir():
+            typer.echo(
+                f'Output directory does not exist: '
+                f'{destination.parent}',
+                err=True,
+            )
+            raise typer.Exit(code=2)
+        resolved_destinations.append(resolved)
+    if len(set(resolved_destinations)) != len(resolved_destinations):
+        typer.echo(
+            'JSON output and snapshot paths must differ',
+            err=True,
+        )
+        raise typer.Exit(code=2)
+
+    try:
+        report: dict = audit_ohlcv_parquet(
+            path,
+            fqme,
+            period,
+            max_gaps,
+            snapshot,
+        )
+    except (
+        FileNotFoundError,
+        OSError,
+        pl.exceptions.PolarsError,
+        ValueError,
+    ) as err:
+        typer.echo(
+            f'Unable to audit {path}: {err}',
+            err=True,
+        )
+        raise typer.Exit(code=2) from err
+
+    report['runtime'] = {
+        'piker_file': str(Path(piker_pkg.__file__).resolve()),
+        'executable': str(Path(sys.argv[0]).resolve()),
+    }
+    payload: str = json.dumps(
+        report,
+        allow_nan=False,
+        indent=2,
+        sort_keys=True,
+    )
+    if output is not None:
+        try:
+            with output.open('x') as output_file:
+                output_file.write(f'{payload}\n')
+        except OSError as err:
+            typer.echo(
+                f'Unable to write audit output: {err}',
+                err=True,
+            )
+            raise typer.Exit(code=2) from err
+
+    if json_output:
+        typer.echo(payload)
+    else:
+        _render_audit_report(report)
+
+    if (
+        strict
+        and
+        not report['result']['qualification_ok']
+    ):
+        raise typer.Exit(code=1)
+
+
 # TODO: like ls but takes in a pattern and matches
 # @store.command()
 # def search(

piker/storage/nativedb.py

@@ -514,6 +514,14 @@ class NativeStorageClient:
             stored: pl.DataFrame = self._canonicalize_ohlcv(
                 pl.read_parquet(path)
             )
+            stored_len: int = stored.height
+            stored = stored.filter(pl.col('time') > 0)
+            dropped: int = stored_len - stored.height
+            if dropped:
+                log.warning(
+                    f'Dropping {dropped} persisted OHLCV row(s) with '
+                    f'non-positive timestamps during merge repair'
+                )
             merged: pl.DataFrame = pl.concat(
                 [stored, incoming],
                 how='diagonal_relaxed',

snippets/nativedb_backfill_audit.xsh

@@ -0,0 +1,936 @@
+#!env xonsh
+'''
+Sourceable helpers for disposable NativeDB backfill qualification.
+
+Usage:
+
+    source snippets/nativedb_backfill_audit.xsh
+
+Every mutating helper requires a marker created by ``bfq_init()`` and
+refuses to operate outside that disposable root.
+Keep the private run root free from concurrent manual changes.
+
+'''
+from datetime import (
+    UTC,
+    datetime,
+)
+from hashlib import sha256
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+from typing import BinaryIO
+
+import numpy as np
+import polars as pl
+
+
+_bfq_marker: str = '.piker-backfill-qualification'
+_bfq_schema: str = 'piker.backfill-qualification/v1'
+_bfq_audit_schema: str = 'piker.nativedb.audit/v1'
+
+
+def _bfq_process_env() -> dict[str, str]:
+    '''
+    Return the effective environment inherited by xonsh children.
+
+    '''
+    env: dict[str, str] = os.environ.copy()
+    try:
+        from xonsh.built_ins import XSH
+    except ImportError:
+        return env
+    if XSH.env is not None:
+        env.update(XSH.env.detype())
+    return env
+
+
+def _bfq_setenv(name: str, value: str) -> None:
+    '''
+    Update both Python and xonsh subprocess environments.
+
+    '''
+    os.environ[name] = value
+    try:
+        from xonsh.built_ins import XSH
+    except ImportError:
+        return
+    if XSH.env is not None:
+        XSH.env[name] = value
+
+
+def _bfq_metadata(root: Path) -> dict:
+    marker: Path = root / _bfq_marker
+    if (
+        not marker.is_file()
+        or
+        marker.is_symlink()
+    ):
+        raise RuntimeError(
+            f'Qualification marker missing or unsafe: {marker}'
+        )
+    metadata: dict = json.loads(marker.read_text())
+    if metadata.get('schema') != _bfq_schema:
+        raise RuntimeError(f'Unknown qualification marker: {marker}')
+    if Path(metadata['root']).resolve() != root:
+        raise RuntimeError(
+            f'Qualification marker root mismatch: {marker}'
+        )
+    return metadata
+
+
+def _bfq_root(run_root: str|Path) -> Path:
+    root = Path(run_root).expanduser().resolve()
+    _bfq_metadata(root)
+    return root
+
+
+def _bfq_fqme(fqme: str) -> str:
+    if (
+        fqme in {'.', '..'}
+        or
+        Path(fqme).name != fqme
+        or
+        '/' in fqme
+        or
+        '\\' in fqme
+    ):
+        raise ValueError(f'Unsafe FQME path component: {fqme!r}')
+    return fqme
+
+
+def _bfq_paths(
+    run_root: str|Path,
+    fqme: str,
+    period_s: int,
+
+) -> tuple[Path, Path, Path, Path]:
+    root: Path = _bfq_root(run_root)
+    fqme = _bfq_fqme(fqme)
+    xdg: Path = root / 'xdg'
+    evidence: Path = root / 'evidence'
+    nativedb: Path = xdg / 'piker' / 'nativedb'
+    parquet: Path = nativedb / f'{fqme}.ohlcv{period_s}s.parquet'
+    for path in (xdg, evidence, nativedb, parquet):
+        resolved: Path = path.resolve()
+        if not resolved.is_relative_to(root):
+            raise RuntimeError(
+                f'Qualification path escaped run root: {resolved}'
+            )
+        if path.is_symlink():
+            raise RuntimeError(f'Refusing symlinked path: {path}')
+    return root, xdg, evidence, parquet
+
+
+def _bfq_hash_file(file: BinaryIO) -> str:
+    digest = sha256()
+    file.seek(0)
+    for chunk in iter(
+        lambda: file.read(1024 * 1024),
+        b'',
+    ):
+        digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _bfq_hash(path: Path) -> str:
+    with path.open('rb') as file:
+        return _bfq_hash_file(file)
+
+
+def _bfq_read_bytes(path: Path) -> bytes:
+    descriptor: int = os.open(
+        path,
+        os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW,
+    )
+    with os.fdopen(descriptor, 'rb') as file:
+        return file.read()
+
+
+def _bfq_copy_exclusive(
+    source: Path,
+    destination: Path,
+    mode: int|None = None,
+) -> None:
+    '''
+    Copy bytes to a new non-symlink path without replacement.
+
+    '''
+    if destination.is_symlink():
+        raise RuntimeError(
+            f'Refusing symlinked output: {destination}'
+        )
+    created: bool = False
+    try:
+        with source.open('rb') as source_file:
+            with destination.open('xb') as destination_file:
+                created = True
+                if mode is not None:
+                    os.fchmod(destination_file.fileno(), mode)
+                shutil.copyfileobj(source_file, destination_file)
+    except BaseException:
+        if created:
+            destination.unlink(missing_ok=True)
+        raise
+
+
+def _bfq_phase(phase: str) -> str:
+    safe: str = phase.replace('-', '').replace('_', '')
+    if not safe.isalnum():
+        raise ValueError(
+            'Phase names may contain letters, numbers, - and _ only'
+        )
+    return phase
+
+
+def bfq_init(
+    run_root: str|Path,
+    source_config: str|Path|None = None,
+    validate_piker: bool = True,
+) -> Path:
+    '''
+    Create a disposable XDG root and optionally copy provider config.
+
+    '''
+    root = Path(run_root).expanduser().resolve()
+    marker: Path = root / _bfq_marker
+    marker_exists: bool = (
+        marker.exists()
+        or
+        marker.is_symlink()
+    )
+    if (
+        root.exists()
+        and
+        not marker_exists
+        and
+        any(root.iterdir())
+    ):
+        raise RuntimeError(
+            f'Refusing non-empty unmarked directory: {root}'
+        )
+
+    if marker_exists:
+        metadata: dict = _bfq_metadata(root)
+    else:
+        metadata = {}
+
+    xdg: Path = root / 'xdg'
+    piker_conf: Path = xdg / 'piker'
+    nativedb: Path = piker_conf / 'nativedb'
+    evidence: Path = root / 'evidence'
+    logs: Path = root / 'logs'
+    for path in (root, xdg, piker_conf, nativedb, evidence, logs):
+        if path.is_symlink():
+            raise RuntimeError(f'Refusing symlinked path: {path}')
+        if not path.resolve().is_relative_to(root):
+            raise RuntimeError(
+                f'Path escaped qualification root: {path}'
+            )
+        path.mkdir(parents=True, exist_ok=True, mode=0o700)
+        path.chmod(0o700)
+
+    if not marker_exists:
+        process_env: dict[str, str] = _bfq_process_env()
+        repo_root = subprocess.run(
+            ['git', 'rev-parse', '--show-toplevel'],
+            check=True,
+            capture_output=True,
+            env=process_env,
+            text=True,
+        ).stdout.strip()
+        piker_executable: str|None = shutil.which(
+            'piker',
+            path=process_env.get('PATH'),
+        )
+        if piker_executable is None:
+            raise RuntimeError('No `piker` executable found in PATH')
+        marker_data: dict = {
+            'schema': _bfq_schema,
+            'root': str(root),
+            'repo_root': str(Path(repo_root).resolve()),
+            'piker_executable': str(
+                Path(piker_executable).resolve()
+            ),
+        }
+        metadata = marker_data
+
+    if validate_piker:
+        process_env = _bfq_process_env()
+        repo_root = metadata['repo_root']
+        pythonpath: list[str] = [repo_root]
+        pythonpath.extend(
+            path
+            for path in process_env.get('PYTHONPATH', '').split(
+                os.pathsep
+            )
+            if path and path != repo_root
+        )
+        process_env['PYTHONPATH'] = os.pathsep.join(pythonpath)
+        proc = subprocess.run(
+            [
+                metadata['piker_executable'],
+                'store',
+                'audit',
+                '--help',
+            ],
+            check=False,
+            capture_output=True,
+            cwd=metadata['repo_root'],
+            env=process_env,
+            text=True,
+        )
+        if proc.returncode:
+            raise RuntimeError(
+                f'Bound `piker` lacks `store audit`:\n'
+                f'{metadata["piker_executable"]}\n'
+                f'{proc.stderr}'
+            )
+
+    if not marker_exists:
+        with marker.open('x') as marker_file:
+            os.fchmod(marker_file.fileno(), 0o600)
+            marker_file.write(
+                json.dumps(metadata, indent=2, sort_keys=True)
+                +'\n'
+            )
+
+    if source_config is not None:
+        source = Path(source_config).expanduser().resolve()
+        for name in ('conf.toml', 'brokers.toml'):
+            src: Path = source / name
+            dst: Path = piker_conf / name
+            if src.is_file() and not dst.exists():
+                shutil.copy2(src, dst)
+                if name == 'brokers.toml':
+                    with dst.open('rb') as config_file:
+                        os.fchmod(config_file.fileno(), 0o600)
+
+    bfq_use(root)
+    print(
+        f'Disposable qualification root ready:\n'
+        f'  root: {root}\n'
+        f'  XDG_CONFIG_HOME: {root / "xdg"}\n'
+    )
+    return root
+
+
+def bfq_use(run_root: str|Path) -> Path:
+    '''
+    Point xonsh and future children at the disposable root.
+
+    '''
+    root: Path = _bfq_root(run_root)
+    metadata: dict = _bfq_metadata(root)
+    xdg: Path = root / 'xdg'
+    repo_root: str = metadata['repo_root']
+    process_env: dict[str, str] = _bfq_process_env()
+    pythonpath: list[str] = [repo_root]
+    pythonpath.extend(
+        path
+        for path in process_env.get('PYTHONPATH', '').split(
+            os.pathsep
+        )
+        if path and path != repo_root
+    )
+    _bfq_setenv('XDG_CONFIG_HOME', str(xdg))
+    _bfq_setenv('PYTHONPATH', os.pathsep.join(pythonpath))
+    print(f'XDG_CONFIG_HOME={xdg}')
+    return xdg
+
+
+def bfq_record_case(
+    run_root: str|Path,
+    fqme: str,
+    period_s: int,
+    provider: str,
+    symptom: str,
+    bad_start_utc: str|None = None,
+    bad_end_utc: str|None = None,
+) -> Path:
+    '''
+    Save read-only operator inputs used to identify one case.
+
+    '''
+    root: Path = _bfq_root(run_root)
+    fqme = _bfq_fqme(fqme)
+    path: Path = root / 'case.json'
+    if path.is_symlink():
+        raise RuntimeError(f'Refusing symlinked case record: {path}')
+    if path.exists():
+        raise FileExistsError(f'Case record already exists: {path}')
+    case: dict = {
+        'fqme': fqme,
+        'period_s': period_s,
+        'provider': provider,
+        'symptom': symptom,
+        'bad_start_utc': bad_start_utc,
+        'bad_end_utc': bad_end_utc,
+        'created_at_utc': datetime.now(tz=UTC).isoformat(),
+    }
+    with path.open('x') as case_file:
+        os.fchmod(case_file.fileno(), 0o444)
+        case_file.write(
+            json.dumps(case, indent=2, sort_keys=True) + '\n'
+        )
+    print(f'Wrote case record: {path}')
+    return path
+
+
+def bfq_seed(
+    run_root: str|Path,
+    fqme: str,
+    period_s: int,
+    source_parquet: str|Path,
+) -> Path:
+    '''
+    Copy a known-bad Parquet into disposable storage and evidence.
+
+    '''
+    (
+        root,
+        _,
+        evidence,
+        parquet,
+    ) = _bfq_paths(run_root, fqme, period_s)
+    source = Path(source_parquet).expanduser().resolve()
+    if not source.is_file():
+        raise FileNotFoundError(source)
+    if parquet.exists():
+        raise FileExistsError(parquet)
+
+    evidence_seed: Path = evidence / 'seed.parquet'
+    if evidence_seed.exists():
+        raise FileExistsError(evidence_seed)
+    _bfq_copy_exclusive(source, evidence_seed, mode=0o444)
+    _bfq_copy_exclusive(evidence_seed, parquet)
+    print(
+        f'Seeded disposable NativeDB:\n'
+        f'  source: {source}\n'
+        f'  evidence: {evidence_seed}\n'
+        f'  active: {parquet}\n'
+        f'  root: {root}\n'
+    )
+    return parquet
+
+
+def bfq_clear_for_fresh(
+    run_root: str|Path,
+    fqme: str,
+    period_s: int,
+) -> Path|None:
+    '''
+    Remove only disposable active history after archiving it.
+
+    '''
+    (
+        _,
+        _,
+        evidence,
+        parquet,
+    ) = _bfq_paths(run_root, fqme, period_s)
+    if not parquet.exists():
+        print(f'Already fresh; no active Parquet: {parquet}')
+        return None
+
+    stamp: str = datetime.now(tz=UTC).strftime(
+        '%Y%m%dT%H%M%S%fZ'
+    )
+    archived: Path = evidence / f'pre-fresh-{stamp}.parquet'
+    if (
+        archived.exists()
+        or
+        archived.is_symlink()
+    ):
+        raise FileExistsError(archived)
+    os.link(parquet, archived)
+    with archived.open('rb') as archived_file:
+        os.fchmod(archived_file.fileno(), 0o444)
+    parquet.unlink()
+    print(
+        f'Archived disposable history for fresh backfill:\n'
+        f'  archived: {archived}\n'
+        f'  active path now absent: {parquet}\n'
+    )
+    return archived
+
+
+def bfq_audit(
+    run_root: str|Path,
+    phase: str,
+    fqme: str,
+    period_s: int,
+    max_gaps: int = 200,
+    require_structural: bool = False,
+) -> dict:
+    '''
+    Save JSON plus a checksum-matched Parquet snapshot for one phase.
+
+    '''
+    phase = _bfq_phase(phase)
+    (
+        root,
+        _,
+        evidence,
+        parquet,
+    ) = _bfq_paths(run_root, fqme, period_s)
+    bfq_use(root)
+    if not parquet.is_file():
+        raise FileNotFoundError(parquet)
+
+    report_path: Path = evidence / f'{phase}.json'
+    snapshot_path: Path = evidence / f'{phase}.parquet'
+    manual_path: Path = evidence / f'{phase}.manual.json'
+    if any(
+        path.exists()
+        for path in (report_path, snapshot_path, manual_path)
+    ):
+        raise FileExistsError(
+            f'Phase evidence already exists: {phase}'
+        )
+
+    metadata: dict = _bfq_metadata(root)
+    piker_cmd: str = metadata['piker_executable']
+
+    proc = subprocess.run(
+        [
+            piker_cmd,
+            'store',
+            'audit',
+            fqme,
+            '--period',
+            str(period_s),
+            '--max-gaps',
+            str(max_gaps),
+            '--output',
+            str(report_path),
+            '--snapshot',
+            str(snapshot_path),
+            '--json',
+        ],
+        check=False,
+        capture_output=True,
+        env=_bfq_process_env(),
+        text=True,
+    )
+    if proc.returncode:
+        raise RuntimeError(
+            f'`piker store audit` failed ({proc.returncode}):\n'
+            f'{proc.stderr}'
+        )
+    report: dict = json.loads(report_path.read_text())
+    report_hash: str = _bfq_hash(report_path)
+    snapshot_hash: str = _bfq_hash(snapshot_path)
+    if snapshot_hash != report['source']['sha256']:
+        raise RuntimeError('Audit snapshot checksum mismatch')
+
+    git_head = subprocess.run(
+        ['git', 'rev-parse', 'HEAD'],
+        cwd=metadata['repo_root'],
+        check=False,
+        capture_output=True,
+        text=True,
+    ).stdout.strip()
+    piker_file = Path(report['runtime']['piker_file']).resolve()
+    repo_root = Path(metadata['repo_root']).resolve()
+    if not piker_file.is_relative_to(repo_root):
+        raise RuntimeError(
+            f'`piker` imported outside reviewed repo: {piker_file}'
+        )
+    manual: dict = {
+        'phase': phase,
+        'run_root': str(root),
+        'snapshot_path': snapshot_path.name,
+        'report_sha256': report_hash,
+        'git_head': git_head,
+        'piker_file': str(piker_file),
+        'piker_executable': piker_cmd,
+        'xdg_config_home': os.environ['XDG_CONFIG_HOME'],
+    }
+    with manual_path.open('x') as manual_file:
+        os.fchmod(manual_file.fileno(), 0o444)
+        manual_file.write(
+            json.dumps(manual, indent=2, sort_keys=True) + '\n'
+        )
+    with report_path.open('rb') as report_file:
+        os.fchmod(report_file.fileno(), 0o444)
+
+    result: dict = report['result']
+    print(
+        f'Audit saved: {report_path}\n'
+        f'  snapshot: {snapshot_path}\n'
+        f'  rows: {report["rows"]["count"]}\n'
+        f'  structural_ok: {result["structural_ok"]}\n'
+        f'  gaps: {report["gaps"]["count"]}\n'
+    )
+    if (
+        require_structural
+        and
+        not result['structural_ok']
+    ):
+        raise AssertionError(
+            f'Phase is structurally invalid: {report_path}'
+        )
+    return report
+
+
+def bfq_compare(
+    run_root: str|Path,
+    before_phase: str,
+    after_phase: str,
+    require_newer: bool = False,
+) -> dict:
+    '''
+    Compare evidence and fail on core preservation regressions.
+
+    '''
+    before_phase = _bfq_phase(before_phase)
+    after_phase = _bfq_phase(after_phase)
+    if before_phase == after_phase:
+        raise ValueError('Comparison phases must differ')
+    root: Path = _bfq_root(run_root)
+    evidence: Path = root / 'evidence'
+    if (
+        evidence.is_symlink()
+        or
+        not evidence.is_dir()
+        or
+        not evidence.resolve().is_relative_to(root)
+    ):
+        raise RuntimeError(f'Unsafe evidence directory: {evidence}')
+    case_path: Path = root / 'case.json'
+    if (
+        case_path.is_symlink()
+        or
+        not case_path.is_file()
+    ):
+        raise RuntimeError(
+            f'Qualification case missing: {case_path}'
+        )
+    case: dict = json.loads(case_path.read_text())
+
+    def load_phase(phase: str) -> tuple[dict, pl.DataFrame]:
+        report_path: Path = evidence / f'{phase}.json'
+        snapshot_path: Path = evidence / f'{phase}.parquet'
+        manual_path: Path = evidence / f'{phase}.manual.json'
+        for path in (report_path, snapshot_path, manual_path):
+            if (
+                path.is_symlink()
+                or
+                not path.is_file()
+                or
+                not path.resolve().is_relative_to(root)
+            ):
+                raise RuntimeError(f'Unsafe phase evidence: {path}')
+
+        report_bytes: bytes = _bfq_read_bytes(report_path)
+        manual_bytes: bytes = _bfq_read_bytes(manual_path)
+        report: dict = json.loads(report_bytes)
+        manual: dict = json.loads(manual_bytes)
+        if report.get('audit_schema') != _bfq_audit_schema:
+            raise RuntimeError(
+                f'Unknown audit schema: {report_path}'
+            )
+        if (
+            report.get('fqme') != case.get('fqme')
+            or
+            report.get('period_s') != case.get('period_s')
+        ):
+            raise RuntimeError(
+                f'Phase does not match qualification case: {phase}'
+            )
+        if (
+            manual.get('phase') != phase
+            or
+            manual.get('run_root') != str(root)
+            or
+            manual.get('snapshot_path') != snapshot_path.name
+        ):
+            raise RuntimeError(f'Manual metadata mismatch: {phase}')
+        report_hash: str = sha256(report_bytes).hexdigest()
+        if report_hash != manual.get('report_sha256'):
+            raise RuntimeError(
+                f'Audit report checksum mismatch: {phase}'
+            )
+        descriptor: int = os.open(
+            snapshot_path,
+            os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW,
+        )
+        with os.fdopen(descriptor, 'rb') as snapshot_file:
+            snapshot_stat = os.fstat(snapshot_file.fileno())
+            snapshot_hash: str = _bfq_hash_file(snapshot_file)
+            if snapshot_hash != report['source']['sha256']:
+                raise RuntimeError(
+                    f'Audit snapshot checksum mismatch: {phase}'
+                )
+            source_size: int = report['source']['size_bytes']
+            if snapshot_stat.st_size != source_size:
+                raise RuntimeError(
+                    f'Audit snapshot size mismatch: {phase}'
+                )
+            snapshot_file.seek(0)
+            frame: pl.DataFrame = pl.read_parquet(snapshot_file)
+        return report, frame
+
+    before, before_frame = load_phase(before_phase)
+    after, after_frame = load_phase(after_phase)
+
+    limitations: list[str] = []
+
+    def valid_timestamps(
+        frame: pl.DataFrame,
+        label: str,
+    ) -> set|None:
+        if 'time' not in frame.columns:
+            limitations.append(f'{label}_time_missing')
+            return None
+        time_dtype = frame['time'].dtype
+        if not (
+            time_dtype.is_integer()
+            or
+            time_dtype.is_float()
+        ):
+            limitations.append(f'{label}_time_nonnumeric')
+            return None
+
+        timestamps: set = set()
+        for timestamp in frame['time'].drop_nulls().to_list():
+            if (
+                isinstance(timestamp, float)
+                and
+                not np.isfinite(timestamp)
+            ):
+                continue
+            if timestamp <= 0:
+                continue
+            timestamps.add(timestamp)
+        return timestamps
+
+    before_times: set|None = valid_timestamps(
+        before_frame,
+        'before',
+    )
+    after_times: set|None = valid_timestamps(
+        after_frame,
+        'after',
+    )
+    missing_times: list = []
+    if (
+        before_times is not None
+        and
+        after_times is not None
+    ):
+        missing_times = sorted(before_times - after_times)
+
+    value_fields: list[str] = [
+        'open',
+        'high',
+        'low',
+        'close',
+        'volume',
+    ]
+    changed_times: list|None = None
+    seam_columns: set[str] = set(value_fields)
+    seam_columns.add('time')
+    seam_ready: bool = bool(
+        seam_columns.issubset(before_frame.columns)
+        and
+        seam_columns.issubset(after_frame.columns)
+        and
+        before_times is not None
+        and
+        after_times is not None
+    )
+    if seam_ready:
+        before_latest = before_frame.unique(
+            subset='time',
+            keep='last',
+        ).filter(
+            pl.col('time').is_not_null()
+            &
+            pl.col('time').is_finite()
+            &
+            (pl.col('time') > 0)
+        ).select([
+            'time',
+            pl.struct(value_fields).hash().alias('value_hash'),
+        ])
+        after_latest = after_frame.unique(
+            subset='time',
+            keep='last',
+        ).filter(
+            pl.col('time').is_not_null()
+            &
+            pl.col('time').is_finite()
+            &
+            (pl.col('time') > 0)
+        ).select([
+            'time',
+            pl.struct(value_fields).hash().alias('value_hash'),
+        ])
+        common = before_latest.join(
+            after_latest,
+            on='time',
+            how='inner',
+            suffix='_after',
+        )
+        changed_times = sorted(
+            common
+            .filter(
+                pl.col('value_hash')
+                !=
+                pl.col('value_hash_after')
+            )
+            ['time']
+            .to_list()
+        )
+    else:
+        limitations.append('common_value_comparison_unavailable')
+
+    before_gaps: set[tuple] = {
+        (
+            gap['left_timestamp'],
+            gap['right_timestamp'],
+        )
+        for gap in before['gaps']['intervals']
+    }
+    after_gaps: set[tuple] = {
+        (
+            gap['left_timestamp'],
+            gap['right_timestamp'],
+        )
+        for gap in after['gaps']['intervals']
+    }
+    before_min = min(before_times) if before_times else None
+    after_min = min(after_times) if after_times else None
+    before_max = max(before_times) if before_times else None
+    after_max = max(after_times) if after_times else None
+    checks: dict[str, bool] = {
+        'after_structural_ok': after['result']['structural_ok'],
+        'no_timestamps_lost': bool(
+            before_times is not None
+            and
+            after_times is not None
+            and
+            not missing_times
+        ),
+        'minimum_not_later': bool(
+            before_min is not None
+            and
+            after_min is not None
+            and
+            after_min
+            <=
+            before_min
+        ),
+        'maximum_not_earlier': bool(
+            before_max is not None
+            and
+            after_max is not None
+            and
+            after_max
+            >=
+            before_max
+        ),
+        'timestamps_unique': (
+            after['timestamps'].get('duplicate_excess') == 0
+        ),
+        'timestamps_ordered': (
+            after['timestamps'].get('non_positive_delta') == 0
+        ),
+        'index_canonical': after['index']['canonical_from_zero'],
+        'ohlcv_finite': after['values']['all_finite'],
+    }
+    if require_newer:
+        checks['maximum_advanced'] = bool(
+            before_max is not None
+            and
+            after_max is not None
+            and
+            after_max > before_max
+        )
+
+    gap_details_truncated: bool = bool(
+        before['gaps']['details_truncated']
+        or
+        after['gaps']['details_truncated']
+    )
+    gaps_added: list|None = None
+    gaps_removed: list|None = None
+    if not gap_details_truncated:
+        gaps_added = sorted(after_gaps - before_gaps)
+        gaps_removed = sorted(before_gaps - after_gaps)
+    comparison: dict = {
+        'before_phase': before_phase,
+        'after_phase': after_phase,
+        'checks': checks,
+        'passed': all(checks.values()),
+        'before_rows': before['rows']['count'],
+        'after_rows': after['rows']['count'],
+        'missing_timestamp_count': len(missing_times),
+        'missing_timestamps': missing_times[:200],
+        'missing_timestamps_truncated': len(missing_times) > 200,
+        'before_duplicate_excess': (
+            before['timestamps'].get('duplicate_excess')
+        ),
+        'after_duplicate_excess': (
+            after['timestamps'].get('duplicate_excess')
+        ),
+        'changed_common_row_count': (
+            len(changed_times)
+            if changed_times is not None
+            else None
+        ),
+        'changed_common_timestamps': (
+            changed_times[:200]
+            if changed_times is not None
+            else None
+        ),
+        'changed_common_rows_truncated': bool(
+            changed_times is not None
+            and
+            len(changed_times) > 200
+        ),
+        'gaps_added': gaps_added,
+        'gaps_removed': gaps_removed,
+        'gap_comparison_truncated': gap_details_truncated,
+        'limitations': limitations,
+    }
+    output: Path = (
+        evidence
+        /f'compare-{before_phase}-to-{after_phase}.json'
+    )
+    if output.is_symlink():
+        raise RuntimeError(f'Refusing symlinked output: {output}')
+    if output.exists():
+        raise FileExistsError(output)
+    with output.open('x') as output_file:
+        os.fchmod(output_file.fileno(), 0o444)
+        output_file.write(
+            json.dumps(
+                comparison,
+                allow_nan=False,
+                indent=2,
+                sort_keys=True,
+            )
+            +'\n'
+        )
+
+    print(f'Comparison saved: {output}')
+    for check, passed in checks.items():
+        print(f'  {"PASS" if passed else "FAIL"}: {check}')
+    print(
+        f'  changed common rows: '
+        f'{None if changed_times is None else len(changed_times)}\n'
+        f'  gaps added: '
+        f'{None if gaps_added is None else len(gaps_added)}\n'
+        f'  gaps removed: '
+        f'{None if gaps_removed is None else len(gaps_removed)}\n'
+    )
+    if not comparison['passed']:
+        raise AssertionError(
+            f'Qualification comparison failed: {output}'
+        )
+    return comparison

tests/test_backfill_audit_snippet.py

@@ -0,0 +1,564 @@
+'''
+Operator backfill-audit helper regressions.
+
+'''
+from hashlib import sha256
+import json
+from pathlib import Path
+import runpy
+import subprocess
+
+import numpy as np
+import polars as pl
+import pytest
+
+from piker.storage._audit import audit_ohlcv_parquet
+
+
+def load_helpers() -> dict:
+    '''
+    Load the sourceable xonsh file as its pure-Python namespace.
+
+    '''
+    path = (
+        Path(__file__).parents[1]
+        /'snippets'
+        /'nativedb_backfill_audit.xsh'
+    )
+    return runpy.run_path(path)
+
+
+def mk_frame(
+    times: tuple[int, ...],
+    closes: tuple[float, ...]|None = None,
+) -> pl.DataFrame:
+    '''
+    Build canonical history for evidence comparisons.
+
+    '''
+    size: int = len(times)
+    closes = closes or tuple(float(value) for value in times)
+    return pl.DataFrame({
+        'index': pl.Series(np.arange(size), dtype=pl.Int64),
+        'time': pl.Series(times, dtype=pl.Int64),
+        'open': pl.Series(np.arange(size), dtype=pl.Float64),
+        'high': pl.Series(np.arange(size) + 1, dtype=pl.Float64),
+        'low': pl.Series(np.arange(size), dtype=pl.Float64),
+        'close': pl.Series(closes, dtype=pl.Float64),
+        'volume': pl.Series(np.arange(size), dtype=pl.Float64),
+    })
+
+
+def init_case(
+    helpers: dict,
+    run_root: Path,
+) -> None:
+    '''
+    Initialize one fixed synthetic qualification case.
+
+    '''
+    helpers['bfq_init'](run_root, validate_piker=False)
+    helpers['bfq_record_case'](
+        run_root,
+        fqme='x.test',
+        period_s=60,
+        provider='fake',
+        symptom='synthetic lifecycle regression',
+    )
+
+
+def write_phase_evidence(
+    run_root: Path,
+    phase: str,
+    path: Path,
+    fqme: str = 'x.test',
+) -> None:
+    '''
+    Write checksum-bound report and manual metadata for a snapshot.
+
+    '''
+    report: dict = audit_ohlcv_parquet(
+        path,
+        fqme=fqme,
+        period_s=60,
+    )
+    evidence: Path = run_root / 'evidence'
+    report_path: Path = evidence / f'{phase}.json'
+    report_path.write_text(
+        json.dumps(report, indent=2, sort_keys=True) + '\n'
+    )
+    manual: dict = {
+        'phase': phase,
+        'run_root': str(run_root.resolve()),
+        'snapshot_path': path.name,
+        'report_sha256': sha256(
+            report_path.read_bytes()
+        ).hexdigest(),
+    }
+    (evidence / f'{phase}.manual.json').write_text(
+        json.dumps(manual, indent=2, sort_keys=True) + '\n'
+    )
+
+
+def test_disposable_seed_is_archived_before_fresh(
+    tmp_path: Path,
+) -> None:
+    '''
+    Fresh qualification must never remove the operator's source file.
+
+    The workflow starts from known-bad production evidence but needs
+    an empty disposable NativeDB for a full provider backfill.
+    Initialize a marked root, seed by copy, then clear for fresh and
+    prove both the original and evidence copy survive while only the
+    disposable active path is moved into the evidence directory.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    source: Path = tmp_path / 'known-bad.parquet'
+    mk_frame((60, 120)).write_parquet(source)
+
+    helpers['bfq_init'](run_root, validate_piker=False)
+    active: Path = helpers['bfq_seed'](
+        run_root,
+        'x.test',
+        60,
+        source,
+    )
+    archived: Path = helpers['bfq_clear_for_fresh'](
+        run_root,
+        'x.test',
+        60,
+    )
+
+    assert source.is_file()
+    assert (run_root / 'evidence' / 'seed.parquet').is_file()
+    assert archived.is_file()
+    assert not active.exists()
+
+
+def test_seed_active_copy_comes_from_captured_evidence(
+    tmp_path: Path,
+) -> None:
+    '''
+    Source replacement must not split baseline and replay bytes.
+
+    Production writers replace Parquet atomically, so copying the
+    source twice can capture one generation as evidence and replay
+    another. Replace the source after the first copy, then prove the
+    the active disposable file still derives
+    from the captured seed rather than reopening the changed source.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    source: Path = tmp_path / 'source.parquet'
+    original: bytes
+    mk_frame((60, 120)).write_parquet(source)
+    original = source.read_bytes()
+    helpers['bfq_init'](run_root, validate_piker=False)
+    real_copy = helpers['_bfq_copy_exclusive']
+    copied: int = 0
+
+    def replace_after_capture(
+        src: Path,
+        dst: Path,
+        mode: int|None = None,
+    ) -> None:
+        nonlocal copied
+        real_copy(src, dst, mode=mode)
+        copied += 1
+        if copied == 1:
+            mk_frame((60, 180)).write_parquet(source)
+
+    helpers['bfq_seed'].__globals__[
+        '_bfq_copy_exclusive'
+    ] = replace_after_capture
+    active: Path = helpers['bfq_seed'](
+        run_root,
+        'x.test',
+        60,
+        source,
+    )
+
+    assert source.read_bytes() != original
+    assert active.read_bytes() == original
+    assert (
+        run_root / 'evidence' / 'seed.parquet'
+    ).read_bytes() == original
+
+
+def test_seed_copy_failure_removes_partial_active_file(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Failed replay copy must not leave an active partial Parquet.
+
+    Seed evidence can complete before the disposable active copy runs
+    out of space. A partial active path blocks retry and lets a chart
+    mistake truncated bytes for history. Fail the second copy
+    after a short write and prove active storage is removed while the
+    completed seed remains exact.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    source: Path = tmp_path / 'source.parquet'
+    mk_frame((60, 120)).write_parquet(source)
+    before: bytes = source.read_bytes()
+    helpers['bfq_init'](run_root, validate_piker=False)
+    real_copy = helpers['shutil'].copyfileobj
+    copies: int = 0
+
+    def fail_second(source_file, destination_file) -> None:
+        nonlocal copies
+        copies += 1
+        if copies == 2:
+            destination_file.write(source_file.read(10))
+            raise OSError('simulated full filesystem')
+        real_copy(source_file, destination_file)
+
+    monkeypatch.setattr(
+        helpers['shutil'],
+        'copyfileobj',
+        fail_second,
+    )
+    with pytest.raises(OSError, match='full filesystem'):
+        helpers['bfq_seed'](
+            run_root,
+            'x.test',
+            60,
+            source,
+        )
+
+    active: Path = (
+        run_root
+        /'xdg/piker/nativedb/x.test.ohlcv60s.parquet'
+    )
+    assert not active.exists()
+    assert (
+        run_root / 'evidence' / 'seed.parquet'
+    ).read_bytes() == before
+
+
+def test_helpers_reject_paths_outside_marked_root(
+    tmp_path: Path,
+) -> None:
+    '''
+    A marker must not authorize traversal or symlinked paths.
+
+    FQME text becomes part of the active Parquet filename and fresh
+    reset moves that path. Prove parent traversal is rejected, then
+    replace the evidence directory with a symlink and prove path
+    construction refuses to cross the disposable-root boundary.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    helpers['bfq_init'](run_root, validate_piker=False)
+
+    with pytest.raises(ValueError, match='Unsafe FQME'):
+        helpers['_bfq_paths'](
+            run_root,
+            '../../outside',
+            60,
+        )
+
+    evidence: Path = run_root / 'evidence'
+    outside: Path = tmp_path / 'outside'
+    outside.mkdir()
+    evidence.rmdir()
+    evidence.symlink_to(outside, target_is_directory=True)
+    with pytest.raises(RuntimeError, match='symlinked path'):
+        helpers['bfq_init'](
+            run_root,
+            validate_piker=False,
+        )
+    with pytest.raises(RuntimeError, match='escaped run root'):
+        helpers['_bfq_paths'](run_root, 'x.test', 60)
+    with pytest.raises(RuntimeError, match='Unsafe evidence'):
+        helpers['bfq_compare'](run_root, 'before', 'after')
+
+
+def test_helpers_reject_dangling_evidence_symlinks(
+    tmp_path: Path,
+) -> None:
+    '''
+    Final evidence names must not follow dangling external symlinks.
+
+    ``Path.exists()`` is false for a dangling symlink, which can make
+    a check-then-write path create its external target. Link the case
+    record and seed outside a marked root. Prove both writers reject
+    the final component before creating external bytes.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    helpers['bfq_init'](run_root, validate_piker=False)
+    outside_case: Path = tmp_path / 'outside-case.json'
+    case_path: Path = run_root / 'case.json'
+    case_path.symlink_to(outside_case)
+    with pytest.raises(RuntimeError, match='symlinked case'):
+        helpers['bfq_record_case'](
+            run_root,
+            fqme='x.test',
+            period_s=60,
+            provider='fake',
+            symptom='unsafe path test',
+        )
+    assert not outside_case.exists()
+
+    case_path.unlink()
+    source: Path = tmp_path / 'source.parquet'
+    mk_frame((60, 120)).write_parquet(source)
+    outside_seed: Path = tmp_path / 'outside-seed.parquet'
+    seed: Path = run_root / 'evidence' / 'seed.parquet'
+    seed.symlink_to(outside_seed)
+    with pytest.raises(RuntimeError, match='symlinked output'):
+        helpers['bfq_seed'](
+            run_root,
+            'x.test',
+            60,
+            source,
+        )
+    assert not outside_seed.exists()
+
+
+def test_bfq_use_updates_xonsh_subprocess_environment(
+    tmp_path: Path,
+) -> None:
+    '''
+    Disposable XDG selection must reach commands launched by xonsh.
+
+    Updating only ``os.environ`` does not mutate xonsh's environment
+    mapping, so manual chart commands previously retained normal user
+    storage. Source the real snippet in xonsh, initialize a run root,
+    and assert a subprocess sees the disposable XDG path.
+
+    '''
+    repo: Path = Path(__file__).parents[1]
+    snippet: Path = repo / 'snippets' / 'nativedb_backfill_audit.xsh'
+    run_root: Path = tmp_path / 'run'
+    code: str = (
+        f'source {snippet}; '
+        f'bfq_init(p\'{run_root}\', validate_piker=False); '
+        'env'
+    )
+    result = subprocess.run(
+        ['xonsh', '-c', code],
+        check=False,
+        capture_output=True,
+        cwd=repo,
+        text=True,
+    )
+
+    assert result.returncode == 0, result.stderr
+    expected: str = f'XDG_CONFIG_HOME={run_root / "xdg"}'
+    assert result.stdout.count(expected) == 2
+
+
+def test_init_validates_console_script_against_worktree(
+    tmp_path: Path,
+) -> None:
+    '''
+    An installed console script must import the reviewed worktree.
+
+    A shared virtualenv's ``piker`` entry point can resolve its
+    editable root checkout, where a new audit command is absent, even
+    while the operator is in this linked worktree. Run initialization
+    and prove its PYTHONPATH-bound preflight sees the local command
+    before it records the executable and repository identity.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+
+    helpers['bfq_init'](run_root)
+
+    marker: dict = json.loads(
+        (run_root / '.piker-backfill-qualification').read_text()
+    )
+    assert Path(marker['repo_root']) == Path(__file__).parents[1]
+
+
+def test_phase_compare_detects_preservation_and_seam_changes(
+    tmp_path: Path,
+) -> None:
+    '''
+    Lifecycle comparison must preserve data without hiding seams.
+
+    Restart and append may replace a matching provider seam while old
+    timestamps remain. Save two read-only snapshots where the second
+    appends a row and updates a close. Prove checks pass while the
+    the changed timestamp stays explicit operator evidence.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    init_case(helpers, run_root)
+    evidence: Path = run_root / 'evidence'
+    before_path: Path = evidence / 'fresh.parquet'
+    after_path: Path = evidence / 'restart.parquet'
+    mk_frame((60, 120), (1, 2)).write_parquet(before_path)
+    mk_frame((60, 120, 180), (1, 20, 3)).write_parquet(after_path)
+
+    for phase, path in (
+        ('fresh', before_path),
+        ('restart', after_path),
+    ):
+        write_phase_evidence(run_root, phase, path)
+
+    comparison: dict = helpers['bfq_compare'](
+        run_root,
+        'fresh',
+        'restart',
+        require_newer=True,
+    )
+
+    assert comparison['passed'] is True
+    assert comparison['missing_timestamp_count'] == 0
+    assert comparison['changed_common_row_count'] == 1
+    assert comparison['changed_common_timestamps'] == [120]
+    assert comparison['checks']['maximum_advanced'] is True
+    with pytest.raises(ValueError, match='phases must differ'):
+        helpers['bfq_compare'](run_root, 'fresh', 'fresh')
+
+
+def test_repair_compare_allows_invalid_and_duplicate_removal(
+    tmp_path: Path,
+) -> None:
+    '''
+    Repair may remove invalid rows but must preserve valid times.
+
+    Known-bad history may contain zero-epoch and duplicate rows. Both
+    should disappear during canonical rewrite. Compare
+    that baseline to clean extended history and prove preservation
+    ignores invalid epoch and duplicate multiplicity while recording
+    their before and after counts.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    init_case(helpers, run_root)
+    evidence: Path = run_root / 'evidence'
+    before_path: Path = evidence / 'baseline.parquet'
+    after_path: Path = evidence / 'repair.parquet'
+    mk_frame((0, 60, 60, 120)).write_parquet(before_path)
+    mk_frame((60, 120, 180)).write_parquet(after_path)
+
+    for phase, path in (
+        ('baseline', before_path),
+        ('repair', after_path),
+    ):
+        write_phase_evidence(run_root, phase, path)
+
+    comparison: dict = helpers['bfq_compare'](
+        run_root,
+        'baseline',
+        'repair',
+        require_newer=True,
+    )
+
+    assert comparison['passed'] is True
+    assert comparison['missing_timestamp_count'] == 0
+    assert comparison['before_duplicate_excess'] == 1
+    assert comparison['after_duplicate_excess'] == 0
+
+
+def test_malformed_baseline_produces_failed_comparison_evidence(
+    tmp_path: Path,
+) -> None:
+    '''
+    Comparison must report unauditable baselines instead of crashing.
+
+    Raw audit accepts malformed Parquet so humans can save evidence
+    before repair. Write a baseline with textual timestamps and no
+    close column, compare it with canonical history, and prove the
+    helper writes limitations plus failed checks before raising its
+    qualification assertion.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    init_case(helpers, run_root)
+    evidence: Path = run_root / 'evidence'
+    before_path: Path = evidence / 'broken.parquet'
+    after_path: Path = evidence / 'repair.parquet'
+    mk_frame((60, 120)).drop('close').with_columns(
+        pl.Series('time', ['60', '120'])
+    ).write_parquet(before_path)
+    mk_frame((60, 120, 180)).write_parquet(after_path)
+
+    for phase, path in (
+        ('broken', before_path),
+        ('repair', after_path),
+    ):
+        write_phase_evidence(run_root, phase, path)
+
+    with pytest.raises(AssertionError):
+        helpers['bfq_compare'](
+            run_root,
+            'broken',
+            'repair',
+        )
+
+    comparison: dict = json.loads(
+        (
+            evidence
+            /'compare-broken-to-repair.json'
+        ).read_text()
+    )
+    assert comparison['passed'] is False
+    assert comparison['checks']['no_timestamps_lost'] is False
+    assert 'before_time_nonnumeric' in comparison['limitations']
+    assert (
+        'common_value_comparison_unavailable'
+        in comparison['limitations']
+    )
+
+
+def test_compare_rejects_modified_or_cross_case_evidence(
+    tmp_path: Path,
+) -> None:
+    '''
+    Lifecycle checks must not combine unrelated or altered artifacts.
+
+    JSON can remain plausible when a phase snapshot is replaced or a
+    report was generated for another market. Bind each report to its
+    checksum and fixed case metadata. Prove both changes abort before
+    abort before a comparison result can be accepted.
+
+    '''
+    helpers: dict = load_helpers()
+    run_root: Path = tmp_path / 'run'
+    init_case(helpers, run_root)
+    evidence: Path = run_root / 'evidence'
+    before: Path = evidence / 'before.parquet'
+    after: Path = evidence / 'after.parquet'
+    mk_frame((60, 120)).write_parquet(before)
+    mk_frame((60, 120, 180)).write_parquet(after)
+    write_phase_evidence(run_root, 'before', before)
+    write_phase_evidence(run_root, 'after', after)
+
+    after_report: Path = evidence / 'after.json'
+    report_bytes: bytes = after_report.read_bytes()
+    after_report.write_bytes(report_bytes + b'\n')
+    with pytest.raises(RuntimeError, match='report checksum'):
+        helpers['bfq_compare'](run_root, 'before', 'after')
+    after_report.write_bytes(report_bytes)
+
+    mk_frame((60, 180)).write_parquet(before)
+    with pytest.raises(RuntimeError, match='snapshot checksum'):
+        helpers['bfq_compare'](run_root, 'before', 'after')
+
+    other: Path = evidence / 'other.parquet'
+    mk_frame((60, 120)).write_parquet(other)
+    write_phase_evidence(
+        run_root,
+        'other',
+        other,
+        fqme='other.test',
+    )
+    with pytest.raises(RuntimeError, match='does not match'):
+        helpers['bfq_compare'](run_root, 'other', 'after')

tests/test_ib_history.py

@@ -0,0 +1,81 @@
+'''
+IB historical-data request regressions.
+
+'''
+from datetime import (
+    UTC,
+    datetime,
+)
+from types import SimpleNamespace
+
+from pendulum import datetime as pdatetime
+import pytest
+import trio
+
+from piker.brokers.ib.api import Client
+
+
+@pytest.mark.parametrize(
+    ('end_dt', 'expected'),
+    [
+        (None, ''),
+        (
+            datetime(2026, 7, 28, 23, 30, tzinfo=UTC),
+            '20260728-23:30:00',
+        ),
+        (
+            pdatetime(2026, 7, 28, 23, 30, tz='UTC'),
+            '20260728-23:30:00',
+        ),
+        (
+            pdatetime(
+                2026,
+                7,
+                28,
+                23,
+                30,
+                tz='America/New_York',
+            ),
+            '20260729-03:30:00',
+        ),
+    ],
+)
+def test_history_end_datetime_uses_ib_utc_format(
+    end_dt: datetime|None,
+    expected: str,
+) -> None:
+    '''
+    Reverse history boundaries must use IB's explicit UTC syntax.
+
+    ``ib_async`` adds a trailing ``UTC`` zone to aware datetimes.
+    The live IB gateway rejected that form with error 10314 while
+    filling an MNQ gap. Invoke ``Client.bars()`` with stdlib and
+    Pendulum boundaries, then capture the request kwargs.
+    Prove each becomes IB's UTC-dash form while latest requests stay
+    blank.
+
+    '''
+    request: dict = {}
+
+    class FakeIB:
+        async def reqHistoricalDataAsync(self, **kwargs) -> list:
+            request.update(kwargs)
+            return []
+
+    client = SimpleNamespace(ib=FakeIB())
+
+    async def find_contracts(fqme: str) -> list:
+        return [SimpleNamespace()]
+
+    client.find_contracts = find_contracts
+
+    async def main() -> None:
+        await Client.bars(
+            client,
+            fqme='mnq.cme.20260918',
+            sample_period_s=60,
+            end_dt=end_dt,
+        )
+
+    trio.run(main)
+    assert request['endDateTime'] == expected

tests/test_storage_audit.py

@@ -0,0 +1,554 @@
+'''
+Read-only NativeDB audit regressions.
+
+'''
+from hashlib import sha256
+import json
+from pathlib import Path
+
+import numpy as np
+import polars as pl
+import pytest
+from typer.testing import CliRunner
+
+from piker import config
+from piker.data import def_iohlcv_fields
+from piker.storage import _audit as audit_mod
+from piker.storage._audit import (
+    audit_ohlcv_frame,
+    audit_ohlcv_parquet,
+)
+from piker.storage.cli import store
+from piker.storage.nativedb import (
+    mk_ohlcv_shm_keyed_filepath,
+)
+
+
+def mk_frame(times: tuple[int, ...]) -> pl.DataFrame:
+    '''
+    Build one canonical persisted frame with finite values.
+
+    '''
+    size: int = len(times)
+    return pl.DataFrame({
+        'index': pl.Series(np.arange(size), dtype=pl.Int64),
+        'time': pl.Series(times, dtype=pl.Int64),
+        'open': pl.Series(np.arange(size) + 1, dtype=pl.Float64),
+        'high': pl.Series(np.arange(size) + 2, dtype=pl.Float64),
+        'low': pl.Series(np.arange(size), dtype=pl.Float64),
+        'close': pl.Series(np.arange(size) + 1, dtype=pl.Float64),
+        'volume': pl.Series(np.arange(size) + 10, dtype=pl.Float64),
+    })
+
+
+def test_audit_separates_structure_from_positive_gaps() -> None:
+    '''
+    Expected closures must not masquerade as structural corruption.
+
+    Persisted CME and session-market history can be canonical while
+    containing positive timestamp gaps. Build a structurally valid
+    frame with one aligned interval and prove validity remains green
+    while exact endpoints and unresolved coverage remain visible.
+
+    '''
+    report = audit_ohlcv_frame(
+        mk_frame((60, 120, 300)),
+        fqme='mnq.cme.20260918.ib',
+        period_s=60,
+    )
+
+    assert report['result'] == {
+        'structural_ok': True,
+        'gap_free': False,
+        'qualification_ok': False,
+        'violations': [],
+        'warnings': ['positive_time_gaps_unclassified'],
+    }
+    assert report['gaps']['count'] == 1
+    assert report['gaps']['missing_samples_total'] == 2
+    assert report['gaps']['intervals'][0] == {
+        'left_timestamp': 120,
+        'right_timestamp': 300,
+        'left_utc': '1970-01-01T00:02:00Z',
+        'right_utc': '1970-01-01T00:05:00Z',
+        'delta_s': 180,
+        'period_multiple': True,
+        'missing_samples': 2,
+        'classification': 'unclassified',
+    }
+
+
+def test_audit_preserves_raw_defect_evidence() -> None:
+    '''
+    Audit must count malformed rows without repairing or sorting.
+
+    A baseline can contain extra provider columns, duplicated and
+    reversed timestamps, a zero epoch, broken indexes, and non-finite
+    values simultaneously. Construct all defects in physical file
+    order and prove stable violations report each layer instead of
+    hiding them through NativeDB canonicalization or dedupe.
+
+    '''
+    frame = mk_frame((60, 60, 30, 0)).with_columns(
+        pl.Series('index', [0, 2, 2, 4]),
+        pl.Series('open', [1, 2, 3, 4], dtype=pl.Int64),
+        pl.Series(
+            'close',
+            [1, float('nan'), 3, 4],
+            dtype=pl.Float64,
+        ),
+        pl.Series(
+            'volume',
+            [1, 2, float('inf'), 4],
+            dtype=pl.Float64,
+        ),
+        pl.Series('count', [1, 1, 1, 1]),
+    )
+    report = audit_ohlcv_frame(
+        frame,
+        fqme='x.test',
+        period_s=60,
+    )
+
+    assert set(report['result']['violations']) == {
+        'extra_columns',
+        'column_order',
+        'canonical_dtypes',
+        'nonfinite_ohlcv',
+        'nonpositive_timestamps',
+        'duplicate_timestamps',
+        'timestamp_order',
+        'index_not_canonical',
+        'subperiod_time_step',
+    }
+    assert report['timestamps']['duplicate_excess'] == 1
+    assert report['timestamps']['adjacent_zero_delta'] == 1
+    assert report['timestamps']['negative_delta'] == 2
+    assert report['index']['row_position_mismatch'] == 2
+    assert report['values']['nan_by_column']['close'] == 1
+    assert report['values']['infinity_by_column']['volume'] == 1
+
+
+def test_gap_aggregates_are_not_truncated_with_details() -> None:
+    '''
+    Limiting JSON detail must not undercount total missing coverage.
+
+    Long-lived session markets can contain thousands of expected
+    closure intervals. Arrange two gaps but request one detail,
+    record, then prove aggregate counts still describe the complete
+    frame while the bounded detail list is marked truncated.
+
+    '''
+    report = audit_ohlcv_frame(
+        mk_frame((60, 180, 300)),
+        fqme='x.test',
+        period_s=60,
+        max_gaps=1,
+    )
+    gaps: dict = report['gaps']
+
+    assert gaps['count'] == 2
+    assert gaps['aligned_count'] == 2
+    assert gaps['missing_samples_total'] == 2
+    assert gaps['details_count'] == 1
+    assert gaps['details_truncated'] is True
+
+
+def test_subperiod_steps_fail_cadence_qualification() -> None:
+    '''
+    Short positive deltas must not bypass expected-period validation.
+
+    Gap detection alone considers deltas larger than the expected
+    period. That let a 30-second step in 60-second data qualify.
+    Arrange that defect and prove it remains separate from positive
+    gaps while making structural qualification fail.
+
+    '''
+    report = audit_ohlcv_frame(
+        mk_frame((60, 90)),
+        fqme='x.test',
+        period_s=60,
+    )
+
+    assert report['gaps']['count'] == 0
+    assert report['gaps']['subperiod_count'] == 1
+    assert report['gaps']['subperiod_intervals'][0]['delta_s'] == 30
+    assert 'subperiod_time_step' in report['result']['violations']
+    assert report['result']['qualification_ok'] is False
+
+
+def test_integer_evidence_preserves_values_above_float_precision(
+) -> None:
+    '''
+    Int64 timestamp and index evidence must not round through float.
+
+    Adjacent integers above ``2**53`` collapse when coerced to
+    Float64, inventing duplicate times and index mismatches. Audit a
+    canonical two-row frame at that boundary and prove exact
+    endpoints, uniqueness, and indexes remain JSON-native integers.
+
+    '''
+    start: int = 2**53
+    frame = mk_frame((start, start + 1)).with_columns(
+        pl.Series('index', [start, start + 1], dtype=pl.Int64)
+    )
+    report = audit_ohlcv_frame(
+        frame,
+        fqme='x.test',
+        period_s=1,
+    )
+
+    assert report['timestamps']['minimum'] == start
+    assert report['timestamps']['maximum'] == start + 1
+    assert report['timestamps']['duplicate_excess'] == 0
+    assert report['timestamps']['strictly_increasing'] is True
+    assert report['index']['first'] == start
+    assert report['index']['last'] == start + 1
+    assert report['index']['duplicate_excess'] == 0
+    assert report['index']['contiguous'] is True
+    assert report['index']['canonical_from_zero'] is False
+
+
+def test_large_integer_gap_counts_remain_exact() -> None:
+    '''
+    Missing-sample totals must not divide Int64 deltas through float.
+
+    A near-Int64 interval exceeds Float64's exact range. Audit it at
+    one-second cadence and prove all counts agree with exact integer
+    division rather than rounded evidence.
+
+    '''
+    right: int = 2**63 - 1
+    report = audit_ohlcv_frame(
+        mk_frame((1, right)),
+        fqme='x.test',
+        period_s=1,
+    )
+    expected: int = right - 2
+
+    assert report['gaps']['missing_samples_total'] == expected
+    assert (
+        report['gaps']['intervals'][0]['missing_samples']
+        ==
+        expected
+    )
+
+
+def test_decimal_columns_remain_auditable_defect_evidence() -> None:
+    '''
+    Numeric but noncanonical Polars dtypes must produce a report.
+
+    Decimal Parquet columns are readable numeric evidence but do not
+    support Polars ``is_nan()``. Cast timestamps and an OHLC field to
+    Decimal and prove audit reports the dtype violation instead of
+    crashing before malformed baseline evidence can be saved.
+
+    '''
+    frame = mk_frame((60, 120)).with_columns(
+        pl.col('time').cast(pl.Decimal(scale=0)),
+        pl.col('close').cast(pl.Decimal(scale=2)),
+    )
+    report = audit_ohlcv_frame(
+        frame,
+        fqme='x.test',
+        period_s=60,
+    )
+
+    assert report['timestamps']['numeric'] is True
+    assert report['timestamps']['strictly_increasing'] is True
+    assert report['values']['all_finite'] is True
+    assert 'canonical_dtypes' in report['result']['violations']
+
+
+def test_empty_frame_has_unverifiable_gap_coverage() -> None:
+    '''
+    Empty storage must not claim valid cadence or index results.
+
+    An empty canonical schema is structurally invalid and has no pair
+    of timestamps from which gap coverage can be inferred. Prove the
+    report retains its stable gap shape while gap-free status remains
+    unknown and canonical index qualification is rejected.
+
+    '''
+    report = audit_ohlcv_frame(
+        mk_frame(()),
+        fqme='x.test',
+        period_s=60,
+    )
+
+    assert report['gaps']['verifiable'] is False
+    assert report['gaps']['aligned_count'] == 0
+    assert report['result']['gap_free'] is None
+    assert report['index']['canonical_from_zero'] is False
+    assert report['result']['qualification_ok'] is False
+
+
+def test_detail_limit_bounds_all_cadence_deviations() -> None:
+    '''
+    One detail budget must cover gaps and short steps together.
+
+    Human output combines both anomaly types. Slicing each list can
+    exceed ``--max-gaps`` and misreport the detail count. Arrange one
+    short step and one long gap with a budget of one and prove the
+    combined detail output remains bounded and marked truncated.
+
+    '''
+    report = audit_ohlcv_frame(
+        mk_frame((60, 90, 210)),
+        fqme='x.test',
+        period_s=60,
+        max_gaps=1,
+    )
+    gaps: dict = report['gaps']
+
+    assert gaps['count'] == 1
+    assert gaps['subperiod_count'] == 1
+    assert gaps['details_count'] == 1
+    assert (
+        len(gaps['intervals'])
+        + len(gaps['subperiod_intervals'])
+    ) == 1
+    assert gaps['details_truncated'] is True
+
+
+def test_store_audit_json_is_read_only(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    The CLI must inspect exact bytes without storage side effects.
+
+    Operational storage openers can create config directories and
+    mutate caches. Point config at an existing disposable NativeDB,
+    invoke the real Typer command, and prove its JSON references the
+    exact file while bytes and modification time stay unchanged.
+
+    '''
+    monkeypatch.setattr(config, '_config_dir', tmp_path)
+    datadir: Path = tmp_path / 'nativedb'
+    datadir.mkdir()
+    fqme: str = 'x.test'
+    path: Path = mk_ohlcv_shm_keyed_filepath(
+        fqme,
+        60,
+        datadir,
+    )
+    mk_frame((60, 120)).write_parquet(path)
+    before_bytes: bytes = path.read_bytes()
+    before_mtime: int = path.stat().st_mtime_ns
+    output: Path = tmp_path / 'audit.json'
+    snapshot: Path = tmp_path / 'audit.parquet'
+
+    result = CliRunner().invoke(
+        store,
+        [
+            'audit',
+            fqme,
+            '--period',
+            '60',
+            '--output',
+            str(output),
+            '--snapshot',
+            str(snapshot),
+            '--json',
+        ],
+    )
+    report: dict = json.loads(result.stdout)
+
+    assert result.exit_code == 0
+    assert report['source']['path'] == str(path)
+    assert report['source']['sha256'] == sha256(
+        before_bytes
+    ).hexdigest()
+    assert report['result']['qualification_ok'] is True
+    assert path.read_bytes() == before_bytes
+    assert path.stat().st_mtime_ns == before_mtime
+    assert snapshot.read_bytes() == before_bytes
+    assert json.loads(output.read_text()) == report
+    assert report['schema']['columns'] == [
+        name
+        for name, _ in def_iohlcv_fields
+    ]
+
+
+def test_store_audit_refuses_output_collisions(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Evidence output must not replace source or existing files.
+
+    The first CLI accepted any ``--output`` path and could overwrite
+    audited Parquet with JSON. Point output at the source, a report,
+    and a directory; prove each exits with original bytes unchanged.
+
+    '''
+    monkeypatch.setattr(config, '_config_dir', tmp_path)
+    datadir: Path = tmp_path / 'nativedb'
+    datadir.mkdir()
+    fqme: str = 'x.test'
+    path: Path = mk_ohlcv_shm_keyed_filepath(
+        fqme,
+        60,
+        datadir,
+    )
+    mk_frame((60, 120)).write_parquet(path)
+    before: bytes = path.read_bytes()
+    runner = CliRunner()
+
+    for output in (path, tmp_path / 'existing.json', tmp_path):
+        if output.name == 'existing.json':
+            output.write_text('keep\n')
+        result = runner.invoke(
+            store,
+            ['audit', fqme, '--output', str(output)],
+        )
+        assert result.exit_code == 2
+
+    assert path.read_bytes() == before
+    assert (tmp_path / 'existing.json').read_text() == 'keep\n'
+
+
+def test_direct_snapshot_collision_preserves_existing_file(
+    tmp_path: Path,
+) -> None:
+    '''
+    Exclusive snapshot races must never unlink another writer's file.
+
+    The helper opens snapshots with ``xb``. Its first cleanup path
+    unlinked the destination when exclusive open failed. Pre-create a
+    destination and prove the collision preserves its existing bytes.
+
+    '''
+    source: Path = tmp_path / 'source.parquet'
+    snapshot: Path = tmp_path / 'snapshot.parquet'
+    mk_frame((60, 120)).write_parquet(source)
+    snapshot.write_bytes(b'existing evidence')
+
+    with pytest.raises(FileExistsError):
+        audit_ohlcv_parquet(
+            source,
+            fqme='x.test',
+            period_s=60,
+            snapshot=snapshot,
+        )
+
+    assert snapshot.read_bytes() == b'existing evidence'
+
+
+def test_invalid_parquet_is_snapshotted_before_parse(
+    tmp_path: Path,
+) -> None:
+    '''
+    Unreadable phase bytes must survive a failed structural audit.
+
+    A truncated or non-Parquet file previously failed during parsing
+    before ``--snapshot`` captured anything, discarding exact failure
+    evidence from repair and restart. Supply invalid bytes to the
+    low-level audit and prove it raises only after creating an exact,
+    read-only snapshot for later diagnosis.
+
+    '''
+    source: Path = tmp_path / 'broken.parquet'
+    snapshot: Path = tmp_path / 'snapshot.parquet'
+    evidence: bytes = b'not a parquet file\n'
+    source.write_bytes(evidence)
+
+    with pytest.raises(pl.exceptions.PolarsError):
+        audit_ohlcv_parquet(
+            source,
+            fqme='x.test',
+            period_s=60,
+            snapshot=snapshot,
+        )
+
+    assert snapshot.read_bytes() == evidence
+    assert snapshot.stat().st_mode & 0o222 == 0
+
+
+def test_snapshot_copy_failure_removes_partial_output(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Copy failure must not leave partial authoritative evidence.
+
+    Snapshot bytes are captured before parsing so malformed evidence
+    survives. Failed copies leave partial output that blocks retry.
+    Raise after a prefix. Prove partial output is removed while the
+    source stays whole.
+
+    '''
+    source: Path = tmp_path / 'source.parquet'
+    snapshot: Path = tmp_path / 'snapshot.parquet'
+    mk_frame((60, 120)).write_parquet(source)
+    before: bytes = source.read_bytes()
+
+    def fail_copy(source_file, snapshot_file) -> None:
+        snapshot_file.write(source_file.read(10))
+        raise OSError('simulated full filesystem')
+
+    monkeypatch.setattr(audit_mod.shutil, 'copyfileobj', fail_copy)
+    with pytest.raises(OSError, match='full filesystem'):
+        audit_ohlcv_parquet(
+            source,
+            fqme='x.test',
+            period_s=60,
+            snapshot=snapshot,
+        )
+
+    assert not snapshot.exists()
+    assert source.read_bytes() == before
+
+
+def test_store_audit_rejects_unsafe_or_symlinked_sources(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    User-controlled FQME text must stay inside NativeDB storage.
+
+    Absolute and parent-relative market names previously flowed into
+    path construction, while a normal market could be a symlink to
+    external bytes. Point config at a disposable NativeDB and prove
+    traversal and source links fail before external bytes change.
+
+    '''
+    monkeypatch.setattr(config, '_config_dir', tmp_path)
+    datadir: Path = tmp_path / 'nativedb'
+    datadir.mkdir()
+    outside: Path = tmp_path / 'outside.parquet'
+    mk_frame((60, 120)).write_parquet(outside)
+    before: bytes = outside.read_bytes()
+    source: Path = mk_ohlcv_shm_keyed_filepath(
+        'x.test',
+        60,
+        datadir,
+    )
+    source.symlink_to(outside)
+    runner = CliRunner()
+
+    for fqme in ('../../outside', str(outside), 'x.test'):
+        result = runner.invoke(store, ['audit', fqme])
+        assert result.exit_code == 2
+
+    assert outside.read_bytes() == before
+
+
+def test_direct_audit_rejects_nonregular_source(
+    tmp_path: Path,
+) -> None:
+    '''
+    Audit must not read devices, directories, or blocking pipes.
+
+    ``O_NOFOLLOW`` rejects links but opens other node types. Pass a
+    a directory to the reader and prove descriptor metadata
+    rejects it before hashing or handing it to the Parquet parser.
+
+    '''
+    with pytest.raises(ValueError, match='regular file'):
+        audit_ohlcv_parquet(
+            tmp_path,
+            fqme='x.test',
+            period_s=60,
+        )

tests/test_storage_nativedb.py

@@ -131,6 +131,47 @@ def test_update_preserves_history_and_resolves_conflicts(
     assert stored['index'].to_list() == [0, 1, 2, 3]
 
 
+def test_update_repairs_nonpositive_persisted_timestamps(
+    tmp_path: Path,
+) -> None:
+    '''
+    Valid incoming history must repair a legacy epoch-zero row.
+
+    The known MNQ baseline contains one zero timestamp plus extra derived
+    columns and noncanonical absolute indexes. ``update_ohlcv()`` used to
+    canonicalize those bytes but retain the zero row, then reject the
+    whole merged frame before fresh IB bars could publish. Write that
+    legacy shape directly, append a valid provider frame, and prove only
+    the invalid persisted row disappears while old and new valid bars
+    survive with canonical indexes.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    legacy = (
+        tsp.np2pl(mk_ohlcv(
+            (0, 60, 120),
+            (0, 1, 2),
+        ))
+        .with_columns(
+            pl.Series('index', [3137800, 3137801, 3137802]),
+            pl.Series('time_prev', [None, 0, 60]),
+        )
+    )
+    path: Path = client.mk_path('x.test', 60)
+    legacy.write_parquet(path)
+
+    run(client.update_ohlcv(
+        'x.test',
+        mk_ohlcv((180,), (3,)),
+        60,
+    ))
+    stored: pl.DataFrame = pl.read_parquet(path)
+
+    assert stored['time'].to_list() == [60, 120, 180]
+    assert stored['close'].to_list() == [1, 2, 3]
+    assert stored['index'].to_list() == [0, 1, 2]
+
+
 def test_write_ohlcv_remains_an_explicit_replacement(
     tmp_path: Path,
 ) -> None: