Add raw OHLCV audit to `NativeDB`

Inspect Parquet bytes directly without opening storage or normalizing
malformed evidence.

Deats,
- report canonical schema, timestamp, index and finite-value state
- retain exact cadence aggregates with bounded endpoint details
- snapshot bytes before parsing and clean up partial copy failures
- reject unsafe FQMEs, links, collisions and non-regular sources

Prompt-IO: ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.md

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

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

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

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

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

piker/storage/_audit.py

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

piker/storage/cli.py

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

tests/test_storage_audit.py

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