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

@@ -104,9 +104,6 @@ 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`
@@ -189,12 +186,8 @@ 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
 ```
 
@@ -203,9 +196,6 @@ 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

@@ -1,42 +0,0 @@
----
-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

@@ -1,32 +0,0 @@
----
-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

@@ -1,40 +0,0 @@
----
-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

@@ -1,27 +0,0 @@
----
-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

@@ -1,39 +0,0 @@
----
-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

@@ -1,27 +0,0 @@
----
-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

@@ -1,40 +0,0 @@
----
-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

@@ -1,25 +0,0 @@
----
-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

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

piker/storage/_audit.py

@@ -1,652 +0,0 @@
-# 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,9 +19,7 @@ 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 (
@@ -33,25 +31,20 @@ 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
@@ -60,65 +53,6 @@ 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(
@@ -160,170 +94,6 @@ 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,14 +514,6 @@ 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

@@ -1,936 +0,0 @@
-#!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

@@ -1,564 +0,0 @@
-'''
-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

@@ -1,81 +0,0 @@
-'''
-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

@@ -1,554 +0,0 @@
-'''
-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,47 +131,6 @@ 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: