From 922a5df88d5b6d2ef7aa7eb932f9edb843c30e65 Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 28 Jul 2026 19:41:55 -0400 Subject: [PATCH] Add `xonsh` backfill qualification workflow Drive known-bad, fresh, restart and append checks from a private disposable NativeDB root. Deats, - bind child config and imports to the marked worktree root - preserve checksum-matched reports, snapshots and case metadata - compare timestamp coverage, seam values and persisted structure - document manual chart runs, gap triage and the expansion matrix Prompt-IO: ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../run-tests/test-harness-reference.md | 7 + .../20260728T003247Z_ed85721c_prompt_io.md | 42 + ...20260728T003247Z_ed85721c_prompt_io.raw.md | 32 + docs/manual_backfill_qualification.rst | 285 ++++++ snippets/nativedb_backfill_audit.xsh | 936 ++++++++++++++++++ tests/test_backfill_audit_snippet.py | 564 +++++++++++ 6 files changed, 1866 insertions(+) create mode 100644 ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.raw.md create mode 100644 docs/manual_backfill_qualification.rst create mode 100644 snippets/nativedb_backfill_audit.xsh create mode 100644 tests/test_backfill_audit_snippet.py diff --git a/.claude/skills/run-tests/test-harness-reference.md b/.claude/skills/run-tests/test-harness-reference.md index 4bb0c601..412a08db 100644 --- a/.claude/skills/run-tests/test-harness-reference.md +++ b/.claude/skills/run-tests/test-harness-reference.md @@ -104,6 +104,8 @@ default. Deterministic or local first-pass targets: - `tests/test_watchlists.py` +- `tests/test_storage_audit.py` +- `tests/test_backfill_audit_snippet.py` - `tests/test_accounting.py::test_account_file_default_empty` - `tests/test_services.py::test_runtime_boot` - `tests/test_services.py::test_datad_spawn` @@ -188,6 +190,9 @@ tests/ test_feeds.py live Binance/Kraken feeds and shared memory test_questrade.py obsolete credentialed tests; skipped test_services.py pikerd/datad/feed/EMS actor lifecycle + test_storage_audit.py read-only NativeDB audit and JSON CLI + test_backfill_audit_snippet.py + disposable xonsh qualification helpers test_watchlists.py deterministic watchlist JSON operations ``` @@ -196,6 +201,8 @@ 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/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 | diff --git a/ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.md b/ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.md new file mode 100644 index 00000000..450f6433 --- /dev/null +++ b/ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.md @@ -0,0 +1,42 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +session: 24fb9765-a550-4570-8350-f0fc9b7e17db +timestamp: 2026-07-28T00:32:47Z +git_ref: ed85721c +scope: code +substantive: true +raw_file: 20260728T003247Z_ed85721c_prompt_io.raw.md +--- + +## Prompt + +Build a safe xonsh workflow for qualifying NativeDB and reverse-backfill +correctness against a known gappy IB chart, beginning with +`mnq.cme.20260918`. Cover baseline, fresh, restart, append, and expansion +testing through implementation, adversarial review, and verification. + +## Response summary + +Implemented a private disposable xonsh qualification workflow. Evidence is +tied to a fixed case, report checksum, snapshot checksum, worktree import, +and phase metadata. Added explicit path and symlink safety, source +replacement handling, xonsh child environments, and lifecycle preservation +comparisons. + +## Files changed + +- `snippets/nativedb_backfill_audit.xsh` - disposable qualification helpers +- `docs/manual_backfill_qualification.rst` - operator lifecycle runbook +- `tests/test_backfill_audit_snippet.py` - xonsh and comparison regressions +- `.claude/skills/run-tests/test-harness-reference.md` + - qualification test-harness guidance +- `ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.raw.md` + - unedited response record +- `ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.md` + - provenance metadata and response summary + +## Human edits + +None - generated changes have not been edited by the human. diff --git a/ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.raw.md b/ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.raw.md new file mode 100644 index 00000000..4f642e74 --- /dev/null +++ b/ai/prompt-io/opencode/20260728T003247Z_ed85721c_prompt_io.raw.md @@ -0,0 +1,32 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +timestamp: 2026-07-28T00:32:47Z +git_ref: ed85721c +diff_cmd: git diff HEAD~1..HEAD +--- + +> `git diff HEAD~1..HEAD -- snippets/nativedb_backfill_audit.xsh docs/manual_backfill_qualification.rst tests/test_backfill_audit_snippet.py .claude/skills/run-tests/test-harness-reference.md` + +Added sourceable xonsh helpers that create a private disposable root, bind +child processes to the reviewed worktree, capture checksum-bound phase +evidence, safely seed and clear test history, and compare lifecycle +preservation. + +The workflow rejects path traversal, pre-existing symlinks, output +collisions, mismatched case/report/snapshot evidence, and same-phase +comparison. The accompanying runbook covers known-bad replay, fresh +backfill, restart, append, gap triage, and expansion cases. + +Verification generated with the patch: + +- xonsh qualification helper regressions: passed +- xonsh sourceability: passed +- Ruff: passed +- `git diff --check`: passed + +Adversarial review drove additional evidence binding, worktree executable, +environment propagation, source replacement, and descriptor-level safety +fixes. Residual concurrent mutation risk is stated explicitly in the private +disposable-root runbook. diff --git a/docs/manual_backfill_qualification.rst b/docs/manual_backfill_qualification.rst new file mode 100644 index 00000000..7f67ae54 --- /dev/null +++ b/docs/manual_backfill_qualification.rst @@ -0,0 +1,285 @@ +Manual Backfill Qualification +============================= + +Purpose +------- + +This runbook separates provider coverage, durable NativeDB state, +shared-memory hydration, and chart rendering while reproducing gappy +history. Work one FQME and timeframe at a time. Preserve the original +Parquet first, then qualify both repair of that evidence and a completely +fresh backfill. + +The read-only command used throughout is:: + + piker store audit FQME --period 60 + piker store audit FQME --period 60 --json + +It reads raw Parquet directly. It does not open a storage client, +canonicalize data, publish cache state, attach SHM, or rewrite history. +Positive gaps are deliberately unclassified; an observed interval is not +automatically corruption or an expected venue closure. + +.. warning:: + + Do not run qualification against normal user storage. The helper below + requires a marked disposable root before it will seed, archive, or clear + a Parquet. ``piker store anal`` and ``piker store ldshm`` are interactive + and potentially mutating; they are not audit commands. + + Keep the disposable root private and do not rename, replace, or symlink + files under it while a helper is running. Stop the chart cleanly before + every audit or comparison phase. + +Load The Xonsh Helpers +---------------------- + +Run from the repository root in the same xonsh which will launch the chart +or daemon:: + + source snippets/nativedb_backfill_audit.xsh + +Choose a disposable root and copy only the provider configuration needed +for the case:: + + run_root = p'/tmp/piker-backfill/mnq-20260918-60s' + source_config = p'~/.config/piker'.expanduser() + bfq_init(run_root, source_config) + bfq_record_case( + run_root, + fqme='mnq.cme.20260918.ib', + period_s=60, + provider='ib', + symptom='gappy chart during reverse backfill', + bad_start_utc=None, + bad_end_utc=None, + ) + +``bfq_init()`` sets ``XDG_CONFIG_HOME`` in the current xonsh, so every +subsequent ``piker`` and ``pikerd`` child inherits the disposable config. +The effective storage path is under ``RUN_ROOT/xdg/piker/nativedb``. +It also records the exact repository root and ``piker`` executable in the +qualification marker, prepends that root to child ``PYTHONPATH``, and +verifies that the bound executable exposes ``piker store audit``. Audits +fail if that executable imports another tree. + +The copied ``brokers.toml`` can contain credentials. Keep the run root +private and remove it manually after evidence is no longer needed. + +Repository Preflight +-------------------- + +Before every daemon start, verify the code and config inherited by child +processes:: + + git status --short --branch + git rev-parse HEAD + python -c 'import piker; print(piker.__file__)' + python -c 'import os; print(os.environ["XDG_CONFIG_HOME"])' + +Stop any daemon from the previous phase cleanly. Use a separate registry +address in the disposable ``conf.toml`` when a normal Piker stack remains +active. + +Known-Bad Replay +---------------- + +Copy the original file; never move it out of normal storage:: + + fqme = 'mnq.cme.20260918.ib' + period_s = 60 + source_parquet = ( + p'~/.config/piker/nativedb'.expanduser() + /f'{fqme}.ohlcv{period_s}s.parquet' + ) + bfq_seed(run_root, fqme, period_s, source_parquet) + bfq_audit(run_root, 'baseline', fqme, period_s) + +The baseline phase writes:: + + evidence/baseline.json + evidence/baseline.parquet + evidence/baseline.manual.json + +The JSON checksum must match the read-only snapshot, and the manual sidecar +binds that report to its phase and case. Comparison rechecks those bindings +before reading either phase. The report records raw schema, row and +timestamp counts, non-finite values, ordering, duplicate excess, canonical +index state, and every retained positive-gap detail. Files with invalid +Parquet encoding can not produce a JSON report, but ``--snapshot`` still +preserves their exact bytes before parse failure. + +Launch the chart manually from the same shell and capture logs:: + + piker -l info chart mnq.cme.20260918.ib \ + 2>&1 | tee @(run_root / 'logs' / 'repair-chart.log') + +After a clean stop:: + + bfq_audit( + run_root, + 'repair', + fqme, + period_s, + require_structural=True, + ) + bfq_compare(run_root, 'baseline', 'repair') + +The comparison fails if structural validity regresses, any original +timestamp disappears, chronological endpoints shrink, timestamps become +duplicated or unordered, indexes cease to be canonical, or OHLCV becomes +non-finite. Provider replacement of matching seam rows is reported +separately as ``changed_common_timestamps`` for operator review. + +Fresh Backfill +-------------- + +Yes, fresh qualification means saving the known-bad evidence and then +starting with no active Parquet in the disposable NativeDB. Archive only +the disposable file:: + + bfq_clear_for_fresh(run_root, fqme, period_s) + +Confirm the printed active path is absent, launch the same chart, wait for +bounded provider exhaustion or the intended coverage, then stop cleanly:: + + bfq_audit( + run_root, + 'fresh', + fqme, + period_s, + require_structural=True, + ) + +Do not compare ``baseline`` to ``fresh`` as a timestamp-preservation gate: +provider limits may intentionally produce a different initial extent. +Treat ``fresh`` as the baseline for restart and append qualification. + +Restart And Append +------------------ + +Restart the same chart against the fresh persisted file, stop cleanly, and +record:: + + bfq_audit( + run_root, + 'restart', + fqme, + period_s, + require_structural=True, + ) + bfq_compare(run_root, 'fresh', 'restart') + +Leave a subsequent run active long enough to append new samples, stop, and +record:: + + bfq_audit( + run_root, + 'append', + fqme, + period_s, + require_structural=True, + ) + bfq_compare( + run_root, + 'restart', + 'append', + require_newer=True, + ) + +The comparison JSON reports gaps added and removed. Expected closures stay +visible until a venue-aware classifier proves their session alignment. + +Gap Layer Triage +---------------- + +Use the persisted report and chart together: + +=============================== ========================================= +Observation Suspected layer +=============================== ========================================= +Parquet contiguous, chart gappy SHM hydration, sampling, or chart render +Parquet gap, provider has bars provider-delta or storage merge +Provider and Parquet both gappy provider omission or venue closure +Gap vanishes after restart SHM publication or cache lifecycle +Old timestamps disappear destructive persistence regression +Newest seam duplicates conflict or dedupe policy +=============================== ========================================= + +Always record exact UTC gap endpoints. The audit's numeric epoch is +authoritative; the ``*_utc`` fields are deterministic display values. + +Expansion Matrix +---------------- + +Expand only after the fixed FQME passes fresh, restart, and append. Use a +new disposable root and case record for every row: + +==================== ========== ======================================== +Case Period Coverage model +==================== ========== ======================================== +Original IB future 60 seconds CME maintenance and weekend closures +Original IB future 1 second high-frequency provider limits +Second CME future 60 seconds contract-specific history +IB equity 60 seconds regular overnight and weekend closure +IB FX 60 seconds 24x5 session behavior +Crypto spot/perp 60 seconds continuous-session control +Expired future 60 seconds history-only and expiry behavior +==================== ========== ======================================== + +A sourceable matrix can remain ordinary Python data in xonsh:: + + cases = [ + { + 'name': 'mnq-60s', + 'fqme': 'mnq.cme.20260918.ib', + 'period_s': 60, + 'provider': 'ib', + }, + { + 'name': 'mnq-1s', + 'fqme': 'mnq.cme.20260918.ib', + 'period_s': 1, + 'provider': 'ib', + }, + ] + for case in cases: + case_root = p'/tmp/piker-backfill' / case['name'] + bfq_init(case_root, source_config) + bfq_record_case( + case_root, + fqme=case['fqme'], + period_s=case['period_s'], + provider=case['provider'], + symptom='manual expansion case', + ) + +This loop prepares isolated case roots only. Run the complete seed or +fresh, chart, audit, restart, append, and compare lifecycle in each root; +directory creation alone is not qualification. + +The chart path may allocate companion 1-second and 60-second files. Audit +the selected case path explicitly and retain companion files as secondary +evidence; do not imply that the live request isolated one timeframe. + +Interruption Qualification +-------------------------- + +After normal restart is stable, interrupt one phase with graceful Ctrl-C, +restart, and compare against the last clean evidence. Provider disconnect, +SIGTERM, and deterministic write-barrier failures come next. Avoid timing a +SIGKILL by hand; add an explicit failpoint before qualifying atomic replace +boundaries. + +Result Interpretation +--------------------- + +``structural_ok`` means the raw persisted file satisfies durable schema, +finite values, positive unique ordered timestamps, canonical index, and +period-aligned gaps. ``gap_free`` means no positive gaps were observed when +at least two valid timestamps made cadence verifiable; otherwise it is +``null``. ``qualification_ok`` combines both and is intentionally strict. + +A structurally valid session-market file can have ``gap_free = false``. +Those intervals require provider/session evidence before being marked as +expected closures. Unknown calendar state must stay unknown. diff --git a/snippets/nativedb_backfill_audit.xsh b/snippets/nativedb_backfill_audit.xsh new file mode 100644 index 00000000..d56ed938 --- /dev/null +++ b/snippets/nativedb_backfill_audit.xsh @@ -0,0 +1,936 @@ +#!env xonsh +''' +Sourceable helpers for disposable NativeDB backfill qualification. + +Usage: + + source snippets/nativedb_backfill_audit.xsh + +Every mutating helper requires a marker created by ``bfq_init()`` and +refuses to operate outside that disposable root. +Keep the private run root free from concurrent manual changes. + +''' +from datetime import ( + UTC, + datetime, +) +from hashlib import sha256 +import json +import os +from pathlib import Path +import shutil +import subprocess +from typing import BinaryIO + +import numpy as np +import polars as pl + + +_bfq_marker: str = '.piker-backfill-qualification' +_bfq_schema: str = 'piker.backfill-qualification/v1' +_bfq_audit_schema: str = 'piker.nativedb.audit/v1' + + +def _bfq_process_env() -> dict[str, str]: + ''' + Return the effective environment inherited by xonsh children. + + ''' + env: dict[str, str] = os.environ.copy() + try: + from xonsh.built_ins import XSH + except ImportError: + return env + if XSH.env is not None: + env.update(XSH.env.detype()) + return env + + +def _bfq_setenv(name: str, value: str) -> None: + ''' + Update both Python and xonsh subprocess environments. + + ''' + os.environ[name] = value + try: + from xonsh.built_ins import XSH + except ImportError: + return + if XSH.env is not None: + XSH.env[name] = value + + +def _bfq_metadata(root: Path) -> dict: + marker: Path = root / _bfq_marker + if ( + not marker.is_file() + or + marker.is_symlink() + ): + raise RuntimeError( + f'Qualification marker missing or unsafe: {marker}' + ) + metadata: dict = json.loads(marker.read_text()) + if metadata.get('schema') != _bfq_schema: + raise RuntimeError(f'Unknown qualification marker: {marker}') + if Path(metadata['root']).resolve() != root: + raise RuntimeError( + f'Qualification marker root mismatch: {marker}' + ) + return metadata + + +def _bfq_root(run_root: str|Path) -> Path: + root = Path(run_root).expanduser().resolve() + _bfq_metadata(root) + return root + + +def _bfq_fqme(fqme: str) -> str: + if ( + fqme in {'.', '..'} + or + Path(fqme).name != fqme + or + '/' in fqme + or + '\\' in fqme + ): + raise ValueError(f'Unsafe FQME path component: {fqme!r}') + return fqme + + +def _bfq_paths( + run_root: str|Path, + fqme: str, + period_s: int, + +) -> tuple[Path, Path, Path, Path]: + root: Path = _bfq_root(run_root) + fqme = _bfq_fqme(fqme) + xdg: Path = root / 'xdg' + evidence: Path = root / 'evidence' + nativedb: Path = xdg / 'piker' / 'nativedb' + parquet: Path = nativedb / f'{fqme}.ohlcv{period_s}s.parquet' + for path in (xdg, evidence, nativedb, parquet): + resolved: Path = path.resolve() + if not resolved.is_relative_to(root): + raise RuntimeError( + f'Qualification path escaped run root: {resolved}' + ) + if path.is_symlink(): + raise RuntimeError(f'Refusing symlinked path: {path}') + return root, xdg, evidence, parquet + + +def _bfq_hash_file(file: BinaryIO) -> str: + digest = sha256() + file.seek(0) + for chunk in iter( + lambda: file.read(1024 * 1024), + b'', + ): + digest.update(chunk) + return digest.hexdigest() + + +def _bfq_hash(path: Path) -> str: + with path.open('rb') as file: + return _bfq_hash_file(file) + + +def _bfq_read_bytes(path: Path) -> bytes: + descriptor: int = os.open( + path, + os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW, + ) + with os.fdopen(descriptor, 'rb') as file: + return file.read() + + +def _bfq_copy_exclusive( + source: Path, + destination: Path, + mode: int|None = None, +) -> None: + ''' + Copy bytes to a new non-symlink path without replacement. + + ''' + if destination.is_symlink(): + raise RuntimeError( + f'Refusing symlinked output: {destination}' + ) + created: bool = False + try: + with source.open('rb') as source_file: + with destination.open('xb') as destination_file: + created = True + if mode is not None: + os.fchmod(destination_file.fileno(), mode) + shutil.copyfileobj(source_file, destination_file) + except BaseException: + if created: + destination.unlink(missing_ok=True) + raise + + +def _bfq_phase(phase: str) -> str: + safe: str = phase.replace('-', '').replace('_', '') + if not safe.isalnum(): + raise ValueError( + 'Phase names may contain letters, numbers, - and _ only' + ) + return phase + + +def bfq_init( + run_root: str|Path, + source_config: str|Path|None = None, + validate_piker: bool = True, +) -> Path: + ''' + Create a disposable XDG root and optionally copy provider config. + + ''' + root = Path(run_root).expanduser().resolve() + marker: Path = root / _bfq_marker + marker_exists: bool = ( + marker.exists() + or + marker.is_symlink() + ) + if ( + root.exists() + and + not marker_exists + and + any(root.iterdir()) + ): + raise RuntimeError( + f'Refusing non-empty unmarked directory: {root}' + ) + + if marker_exists: + metadata: dict = _bfq_metadata(root) + else: + metadata = {} + + xdg: Path = root / 'xdg' + piker_conf: Path = xdg / 'piker' + nativedb: Path = piker_conf / 'nativedb' + evidence: Path = root / 'evidence' + logs: Path = root / 'logs' + for path in (root, xdg, piker_conf, nativedb, evidence, logs): + if path.is_symlink(): + raise RuntimeError(f'Refusing symlinked path: {path}') + if not path.resolve().is_relative_to(root): + raise RuntimeError( + f'Path escaped qualification root: {path}' + ) + path.mkdir(parents=True, exist_ok=True, mode=0o700) + path.chmod(0o700) + + if not marker_exists: + process_env: dict[str, str] = _bfq_process_env() + repo_root = subprocess.run( + ['git', 'rev-parse', '--show-toplevel'], + check=True, + capture_output=True, + env=process_env, + text=True, + ).stdout.strip() + piker_executable: str|None = shutil.which( + 'piker', + path=process_env.get('PATH'), + ) + if piker_executable is None: + raise RuntimeError('No `piker` executable found in PATH') + marker_data: dict = { + 'schema': _bfq_schema, + 'root': str(root), + 'repo_root': str(Path(repo_root).resolve()), + 'piker_executable': str( + Path(piker_executable).resolve() + ), + } + metadata = marker_data + + if validate_piker: + process_env = _bfq_process_env() + repo_root = metadata['repo_root'] + pythonpath: list[str] = [repo_root] + pythonpath.extend( + path + for path in process_env.get('PYTHONPATH', '').split( + os.pathsep + ) + if path and path != repo_root + ) + process_env['PYTHONPATH'] = os.pathsep.join(pythonpath) + proc = subprocess.run( + [ + metadata['piker_executable'], + 'store', + 'audit', + '--help', + ], + check=False, + capture_output=True, + cwd=metadata['repo_root'], + env=process_env, + text=True, + ) + if proc.returncode: + raise RuntimeError( + f'Bound `piker` lacks `store audit`:\n' + f'{metadata["piker_executable"]}\n' + f'{proc.stderr}' + ) + + if not marker_exists: + with marker.open('x') as marker_file: + os.fchmod(marker_file.fileno(), 0o600) + marker_file.write( + json.dumps(metadata, indent=2, sort_keys=True) + +'\n' + ) + + if source_config is not None: + source = Path(source_config).expanduser().resolve() + for name in ('conf.toml', 'brokers.toml'): + src: Path = source / name + dst: Path = piker_conf / name + if src.is_file() and not dst.exists(): + shutil.copy2(src, dst) + if name == 'brokers.toml': + with dst.open('rb') as config_file: + os.fchmod(config_file.fileno(), 0o600) + + bfq_use(root) + print( + f'Disposable qualification root ready:\n' + f' root: {root}\n' + f' XDG_CONFIG_HOME: {root / "xdg"}\n' + ) + return root + + +def bfq_use(run_root: str|Path) -> Path: + ''' + Point xonsh and future children at the disposable root. + + ''' + root: Path = _bfq_root(run_root) + metadata: dict = _bfq_metadata(root) + xdg: Path = root / 'xdg' + repo_root: str = metadata['repo_root'] + process_env: dict[str, str] = _bfq_process_env() + pythonpath: list[str] = [repo_root] + pythonpath.extend( + path + for path in process_env.get('PYTHONPATH', '').split( + os.pathsep + ) + if path and path != repo_root + ) + _bfq_setenv('XDG_CONFIG_HOME', str(xdg)) + _bfq_setenv('PYTHONPATH', os.pathsep.join(pythonpath)) + print(f'XDG_CONFIG_HOME={xdg}') + return xdg + + +def bfq_record_case( + run_root: str|Path, + fqme: str, + period_s: int, + provider: str, + symptom: str, + bad_start_utc: str|None = None, + bad_end_utc: str|None = None, +) -> Path: + ''' + Save read-only operator inputs used to identify one case. + + ''' + root: Path = _bfq_root(run_root) + fqme = _bfq_fqme(fqme) + path: Path = root / 'case.json' + if path.is_symlink(): + raise RuntimeError(f'Refusing symlinked case record: {path}') + if path.exists(): + raise FileExistsError(f'Case record already exists: {path}') + case: dict = { + 'fqme': fqme, + 'period_s': period_s, + 'provider': provider, + 'symptom': symptom, + 'bad_start_utc': bad_start_utc, + 'bad_end_utc': bad_end_utc, + 'created_at_utc': datetime.now(tz=UTC).isoformat(), + } + with path.open('x') as case_file: + os.fchmod(case_file.fileno(), 0o444) + case_file.write( + json.dumps(case, indent=2, sort_keys=True) + '\n' + ) + print(f'Wrote case record: {path}') + return path + + +def bfq_seed( + run_root: str|Path, + fqme: str, + period_s: int, + source_parquet: str|Path, +) -> Path: + ''' + Copy a known-bad Parquet into disposable storage and evidence. + + ''' + ( + root, + _, + evidence, + parquet, + ) = _bfq_paths(run_root, fqme, period_s) + source = Path(source_parquet).expanduser().resolve() + if not source.is_file(): + raise FileNotFoundError(source) + if parquet.exists(): + raise FileExistsError(parquet) + + evidence_seed: Path = evidence / 'seed.parquet' + if evidence_seed.exists(): + raise FileExistsError(evidence_seed) + _bfq_copy_exclusive(source, evidence_seed, mode=0o444) + _bfq_copy_exclusive(evidence_seed, parquet) + print( + f'Seeded disposable NativeDB:\n' + f' source: {source}\n' + f' evidence: {evidence_seed}\n' + f' active: {parquet}\n' + f' root: {root}\n' + ) + return parquet + + +def bfq_clear_for_fresh( + run_root: str|Path, + fqme: str, + period_s: int, +) -> Path|None: + ''' + Remove only disposable active history after archiving it. + + ''' + ( + _, + _, + evidence, + parquet, + ) = _bfq_paths(run_root, fqme, period_s) + if not parquet.exists(): + print(f'Already fresh; no active Parquet: {parquet}') + return None + + stamp: str = datetime.now(tz=UTC).strftime( + '%Y%m%dT%H%M%S%fZ' + ) + archived: Path = evidence / f'pre-fresh-{stamp}.parquet' + if ( + archived.exists() + or + archived.is_symlink() + ): + raise FileExistsError(archived) + os.link(parquet, archived) + with archived.open('rb') as archived_file: + os.fchmod(archived_file.fileno(), 0o444) + parquet.unlink() + print( + f'Archived disposable history for fresh backfill:\n' + f' archived: {archived}\n' + f' active path now absent: {parquet}\n' + ) + return archived + + +def bfq_audit( + run_root: str|Path, + phase: str, + fqme: str, + period_s: int, + max_gaps: int = 200, + require_structural: bool = False, +) -> dict: + ''' + Save JSON plus a checksum-matched Parquet snapshot for one phase. + + ''' + phase = _bfq_phase(phase) + ( + root, + _, + evidence, + parquet, + ) = _bfq_paths(run_root, fqme, period_s) + bfq_use(root) + if not parquet.is_file(): + raise FileNotFoundError(parquet) + + report_path: Path = evidence / f'{phase}.json' + snapshot_path: Path = evidence / f'{phase}.parquet' + manual_path: Path = evidence / f'{phase}.manual.json' + if any( + path.exists() + for path in (report_path, snapshot_path, manual_path) + ): + raise FileExistsError( + f'Phase evidence already exists: {phase}' + ) + + metadata: dict = _bfq_metadata(root) + piker_cmd: str = metadata['piker_executable'] + + proc = subprocess.run( + [ + piker_cmd, + 'store', + 'audit', + fqme, + '--period', + str(period_s), + '--max-gaps', + str(max_gaps), + '--output', + str(report_path), + '--snapshot', + str(snapshot_path), + '--json', + ], + check=False, + capture_output=True, + env=_bfq_process_env(), + text=True, + ) + if proc.returncode: + raise RuntimeError( + f'`piker store audit` failed ({proc.returncode}):\n' + f'{proc.stderr}' + ) + report: dict = json.loads(report_path.read_text()) + report_hash: str = _bfq_hash(report_path) + snapshot_hash: str = _bfq_hash(snapshot_path) + if snapshot_hash != report['source']['sha256']: + raise RuntimeError('Audit snapshot checksum mismatch') + + git_head = subprocess.run( + ['git', 'rev-parse', 'HEAD'], + cwd=metadata['repo_root'], + check=False, + capture_output=True, + text=True, + ).stdout.strip() + piker_file = Path(report['runtime']['piker_file']).resolve() + repo_root = Path(metadata['repo_root']).resolve() + if not piker_file.is_relative_to(repo_root): + raise RuntimeError( + f'`piker` imported outside reviewed repo: {piker_file}' + ) + manual: dict = { + 'phase': phase, + 'run_root': str(root), + 'snapshot_path': snapshot_path.name, + 'report_sha256': report_hash, + 'git_head': git_head, + 'piker_file': str(piker_file), + 'piker_executable': piker_cmd, + 'xdg_config_home': os.environ['XDG_CONFIG_HOME'], + } + with manual_path.open('x') as manual_file: + os.fchmod(manual_file.fileno(), 0o444) + manual_file.write( + json.dumps(manual, indent=2, sort_keys=True) + '\n' + ) + with report_path.open('rb') as report_file: + os.fchmod(report_file.fileno(), 0o444) + + result: dict = report['result'] + print( + f'Audit saved: {report_path}\n' + f' snapshot: {snapshot_path}\n' + f' rows: {report["rows"]["count"]}\n' + f' structural_ok: {result["structural_ok"]}\n' + f' gaps: {report["gaps"]["count"]}\n' + ) + if ( + require_structural + and + not result['structural_ok'] + ): + raise AssertionError( + f'Phase is structurally invalid: {report_path}' + ) + return report + + +def bfq_compare( + run_root: str|Path, + before_phase: str, + after_phase: str, + require_newer: bool = False, +) -> dict: + ''' + Compare evidence and fail on core preservation regressions. + + ''' + before_phase = _bfq_phase(before_phase) + after_phase = _bfq_phase(after_phase) + if before_phase == after_phase: + raise ValueError('Comparison phases must differ') + root: Path = _bfq_root(run_root) + evidence: Path = root / 'evidence' + if ( + evidence.is_symlink() + or + not evidence.is_dir() + or + not evidence.resolve().is_relative_to(root) + ): + raise RuntimeError(f'Unsafe evidence directory: {evidence}') + case_path: Path = root / 'case.json' + if ( + case_path.is_symlink() + or + not case_path.is_file() + ): + raise RuntimeError( + f'Qualification case missing: {case_path}' + ) + case: dict = json.loads(case_path.read_text()) + + def load_phase(phase: str) -> tuple[dict, pl.DataFrame]: + report_path: Path = evidence / f'{phase}.json' + snapshot_path: Path = evidence / f'{phase}.parquet' + manual_path: Path = evidence / f'{phase}.manual.json' + for path in (report_path, snapshot_path, manual_path): + if ( + path.is_symlink() + or + not path.is_file() + or + not path.resolve().is_relative_to(root) + ): + raise RuntimeError(f'Unsafe phase evidence: {path}') + + report_bytes: bytes = _bfq_read_bytes(report_path) + manual_bytes: bytes = _bfq_read_bytes(manual_path) + report: dict = json.loads(report_bytes) + manual: dict = json.loads(manual_bytes) + if report.get('audit_schema') != _bfq_audit_schema: + raise RuntimeError( + f'Unknown audit schema: {report_path}' + ) + if ( + report.get('fqme') != case.get('fqme') + or + report.get('period_s') != case.get('period_s') + ): + raise RuntimeError( + f'Phase does not match qualification case: {phase}' + ) + if ( + manual.get('phase') != phase + or + manual.get('run_root') != str(root) + or + manual.get('snapshot_path') != snapshot_path.name + ): + raise RuntimeError(f'Manual metadata mismatch: {phase}') + report_hash: str = sha256(report_bytes).hexdigest() + if report_hash != manual.get('report_sha256'): + raise RuntimeError( + f'Audit report checksum mismatch: {phase}' + ) + descriptor: int = os.open( + snapshot_path, + os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW, + ) + with os.fdopen(descriptor, 'rb') as snapshot_file: + snapshot_stat = os.fstat(snapshot_file.fileno()) + snapshot_hash: str = _bfq_hash_file(snapshot_file) + if snapshot_hash != report['source']['sha256']: + raise RuntimeError( + f'Audit snapshot checksum mismatch: {phase}' + ) + source_size: int = report['source']['size_bytes'] + if snapshot_stat.st_size != source_size: + raise RuntimeError( + f'Audit snapshot size mismatch: {phase}' + ) + snapshot_file.seek(0) + frame: pl.DataFrame = pl.read_parquet(snapshot_file) + return report, frame + + before, before_frame = load_phase(before_phase) + after, after_frame = load_phase(after_phase) + + limitations: list[str] = [] + + def valid_timestamps( + frame: pl.DataFrame, + label: str, + ) -> set|None: + if 'time' not in frame.columns: + limitations.append(f'{label}_time_missing') + return None + time_dtype = frame['time'].dtype + if not ( + time_dtype.is_integer() + or + time_dtype.is_float() + ): + limitations.append(f'{label}_time_nonnumeric') + return None + + timestamps: set = set() + for timestamp in frame['time'].drop_nulls().to_list(): + if ( + isinstance(timestamp, float) + and + not np.isfinite(timestamp) + ): + continue + if timestamp <= 0: + continue + timestamps.add(timestamp) + return timestamps + + before_times: set|None = valid_timestamps( + before_frame, + 'before', + ) + after_times: set|None = valid_timestamps( + after_frame, + 'after', + ) + missing_times: list = [] + if ( + before_times is not None + and + after_times is not None + ): + missing_times = sorted(before_times - after_times) + + value_fields: list[str] = [ + 'open', + 'high', + 'low', + 'close', + 'volume', + ] + changed_times: list|None = None + seam_columns: set[str] = set(value_fields) + seam_columns.add('time') + seam_ready: bool = bool( + seam_columns.issubset(before_frame.columns) + and + seam_columns.issubset(after_frame.columns) + and + before_times is not None + and + after_times is not None + ) + if seam_ready: + before_latest = before_frame.unique( + subset='time', + keep='last', + ).filter( + pl.col('time').is_not_null() + & + pl.col('time').is_finite() + & + (pl.col('time') > 0) + ).select([ + 'time', + pl.struct(value_fields).hash().alias('value_hash'), + ]) + after_latest = after_frame.unique( + subset='time', + keep='last', + ).filter( + pl.col('time').is_not_null() + & + pl.col('time').is_finite() + & + (pl.col('time') > 0) + ).select([ + 'time', + pl.struct(value_fields).hash().alias('value_hash'), + ]) + common = before_latest.join( + after_latest, + on='time', + how='inner', + suffix='_after', + ) + changed_times = sorted( + common + .filter( + pl.col('value_hash') + != + pl.col('value_hash_after') + ) + ['time'] + .to_list() + ) + else: + limitations.append('common_value_comparison_unavailable') + + before_gaps: set[tuple] = { + ( + gap['left_timestamp'], + gap['right_timestamp'], + ) + for gap in before['gaps']['intervals'] + } + after_gaps: set[tuple] = { + ( + gap['left_timestamp'], + gap['right_timestamp'], + ) + for gap in after['gaps']['intervals'] + } + before_min = min(before_times) if before_times else None + after_min = min(after_times) if after_times else None + before_max = max(before_times) if before_times else None + after_max = max(after_times) if after_times else None + checks: dict[str, bool] = { + 'after_structural_ok': after['result']['structural_ok'], + 'no_timestamps_lost': bool( + before_times is not None + and + after_times is not None + and + not missing_times + ), + 'minimum_not_later': bool( + before_min is not None + and + after_min is not None + and + after_min + <= + before_min + ), + 'maximum_not_earlier': bool( + before_max is not None + and + after_max is not None + and + after_max + >= + before_max + ), + 'timestamps_unique': ( + after['timestamps'].get('duplicate_excess') == 0 + ), + 'timestamps_ordered': ( + after['timestamps'].get('non_positive_delta') == 0 + ), + 'index_canonical': after['index']['canonical_from_zero'], + 'ohlcv_finite': after['values']['all_finite'], + } + if require_newer: + checks['maximum_advanced'] = bool( + before_max is not None + and + after_max is not None + and + after_max > before_max + ) + + gap_details_truncated: bool = bool( + before['gaps']['details_truncated'] + or + after['gaps']['details_truncated'] + ) + gaps_added: list|None = None + gaps_removed: list|None = None + if not gap_details_truncated: + gaps_added = sorted(after_gaps - before_gaps) + gaps_removed = sorted(before_gaps - after_gaps) + comparison: dict = { + 'before_phase': before_phase, + 'after_phase': after_phase, + 'checks': checks, + 'passed': all(checks.values()), + 'before_rows': before['rows']['count'], + 'after_rows': after['rows']['count'], + 'missing_timestamp_count': len(missing_times), + 'missing_timestamps': missing_times[:200], + 'missing_timestamps_truncated': len(missing_times) > 200, + 'before_duplicate_excess': ( + before['timestamps'].get('duplicate_excess') + ), + 'after_duplicate_excess': ( + after['timestamps'].get('duplicate_excess') + ), + 'changed_common_row_count': ( + len(changed_times) + if changed_times is not None + else None + ), + 'changed_common_timestamps': ( + changed_times[:200] + if changed_times is not None + else None + ), + 'changed_common_rows_truncated': bool( + changed_times is not None + and + len(changed_times) > 200 + ), + 'gaps_added': gaps_added, + 'gaps_removed': gaps_removed, + 'gap_comparison_truncated': gap_details_truncated, + 'limitations': limitations, + } + output: Path = ( + evidence + /f'compare-{before_phase}-to-{after_phase}.json' + ) + if output.is_symlink(): + raise RuntimeError(f'Refusing symlinked output: {output}') + if output.exists(): + raise FileExistsError(output) + with output.open('x') as output_file: + os.fchmod(output_file.fileno(), 0o444) + output_file.write( + json.dumps( + comparison, + allow_nan=False, + indent=2, + sort_keys=True, + ) + +'\n' + ) + + print(f'Comparison saved: {output}') + for check, passed in checks.items(): + print(f' {"PASS" if passed else "FAIL"}: {check}') + print( + f' changed common rows: ' + f'{None if changed_times is None else len(changed_times)}\n' + f' gaps added: ' + f'{None if gaps_added is None else len(gaps_added)}\n' + f' gaps removed: ' + f'{None if gaps_removed is None else len(gaps_removed)}\n' + ) + if not comparison['passed']: + raise AssertionError( + f'Qualification comparison failed: {output}' + ) + return comparison diff --git a/tests/test_backfill_audit_snippet.py b/tests/test_backfill_audit_snippet.py new file mode 100644 index 00000000..a0ca454f --- /dev/null +++ b/tests/test_backfill_audit_snippet.py @@ -0,0 +1,564 @@ +''' +Operator backfill-audit helper regressions. + +''' +from hashlib import sha256 +import json +from pathlib import Path +import runpy +import subprocess + +import numpy as np +import polars as pl +import pytest + +from piker.storage._audit import audit_ohlcv_parquet + + +def load_helpers() -> dict: + ''' + Load the sourceable xonsh file as its pure-Python namespace. + + ''' + path = ( + Path(__file__).parents[1] + /'snippets' + /'nativedb_backfill_audit.xsh' + ) + return runpy.run_path(path) + + +def mk_frame( + times: tuple[int, ...], + closes: tuple[float, ...]|None = None, +) -> pl.DataFrame: + ''' + Build canonical history for evidence comparisons. + + ''' + size: int = len(times) + closes = closes or tuple(float(value) for value in times) + return pl.DataFrame({ + 'index': pl.Series(np.arange(size), dtype=pl.Int64), + 'time': pl.Series(times, dtype=pl.Int64), + 'open': pl.Series(np.arange(size), dtype=pl.Float64), + 'high': pl.Series(np.arange(size) + 1, dtype=pl.Float64), + 'low': pl.Series(np.arange(size), dtype=pl.Float64), + 'close': pl.Series(closes, dtype=pl.Float64), + 'volume': pl.Series(np.arange(size), dtype=pl.Float64), + }) + + +def init_case( + helpers: dict, + run_root: Path, +) -> None: + ''' + Initialize one fixed synthetic qualification case. + + ''' + helpers['bfq_init'](run_root, validate_piker=False) + helpers['bfq_record_case']( + run_root, + fqme='x.test', + period_s=60, + provider='fake', + symptom='synthetic lifecycle regression', + ) + + +def write_phase_evidence( + run_root: Path, + phase: str, + path: Path, + fqme: str = 'x.test', +) -> None: + ''' + Write checksum-bound report and manual metadata for a snapshot. + + ''' + report: dict = audit_ohlcv_parquet( + path, + fqme=fqme, + period_s=60, + ) + evidence: Path = run_root / 'evidence' + report_path: Path = evidence / f'{phase}.json' + report_path.write_text( + json.dumps(report, indent=2, sort_keys=True) + '\n' + ) + manual: dict = { + 'phase': phase, + 'run_root': str(run_root.resolve()), + 'snapshot_path': path.name, + 'report_sha256': sha256( + report_path.read_bytes() + ).hexdigest(), + } + (evidence / f'{phase}.manual.json').write_text( + json.dumps(manual, indent=2, sort_keys=True) + '\n' + ) + + +def test_disposable_seed_is_archived_before_fresh( + tmp_path: Path, +) -> None: + ''' + Fresh qualification must never remove the operator's source file. + + The workflow starts from known-bad production evidence but needs + an empty disposable NativeDB for a full provider backfill. + Initialize a marked root, seed by copy, then clear for fresh and + prove both the original and evidence copy survive while only the + disposable active path is moved into the evidence directory. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + source: Path = tmp_path / 'known-bad.parquet' + mk_frame((60, 120)).write_parquet(source) + + helpers['bfq_init'](run_root, validate_piker=False) + active: Path = helpers['bfq_seed']( + run_root, + 'x.test', + 60, + source, + ) + archived: Path = helpers['bfq_clear_for_fresh']( + run_root, + 'x.test', + 60, + ) + + assert source.is_file() + assert (run_root / 'evidence' / 'seed.parquet').is_file() + assert archived.is_file() + assert not active.exists() + + +def test_seed_active_copy_comes_from_captured_evidence( + tmp_path: Path, +) -> None: + ''' + Source replacement must not split baseline and replay bytes. + + Production writers replace Parquet atomically, so copying the + source twice can capture one generation as evidence and replay + another. Replace the source after the first copy, then prove the + the active disposable file still derives + from the captured seed rather than reopening the changed source. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + source: Path = tmp_path / 'source.parquet' + original: bytes + mk_frame((60, 120)).write_parquet(source) + original = source.read_bytes() + helpers['bfq_init'](run_root, validate_piker=False) + real_copy = helpers['_bfq_copy_exclusive'] + copied: int = 0 + + def replace_after_capture( + src: Path, + dst: Path, + mode: int|None = None, + ) -> None: + nonlocal copied + real_copy(src, dst, mode=mode) + copied += 1 + if copied == 1: + mk_frame((60, 180)).write_parquet(source) + + helpers['bfq_seed'].__globals__[ + '_bfq_copy_exclusive' + ] = replace_after_capture + active: Path = helpers['bfq_seed']( + run_root, + 'x.test', + 60, + source, + ) + + assert source.read_bytes() != original + assert active.read_bytes() == original + assert ( + run_root / 'evidence' / 'seed.parquet' + ).read_bytes() == original + + +def test_seed_copy_failure_removes_partial_active_file( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Failed replay copy must not leave an active partial Parquet. + + Seed evidence can complete before the disposable active copy runs + out of space. A partial active path blocks retry and lets a chart + mistake truncated bytes for history. Fail the second copy + after a short write and prove active storage is removed while the + completed seed remains exact. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + source: Path = tmp_path / 'source.parquet' + mk_frame((60, 120)).write_parquet(source) + before: bytes = source.read_bytes() + helpers['bfq_init'](run_root, validate_piker=False) + real_copy = helpers['shutil'].copyfileobj + copies: int = 0 + + def fail_second(source_file, destination_file) -> None: + nonlocal copies + copies += 1 + if copies == 2: + destination_file.write(source_file.read(10)) + raise OSError('simulated full filesystem') + real_copy(source_file, destination_file) + + monkeypatch.setattr( + helpers['shutil'], + 'copyfileobj', + fail_second, + ) + with pytest.raises(OSError, match='full filesystem'): + helpers['bfq_seed']( + run_root, + 'x.test', + 60, + source, + ) + + active: Path = ( + run_root + /'xdg/piker/nativedb/x.test.ohlcv60s.parquet' + ) + assert not active.exists() + assert ( + run_root / 'evidence' / 'seed.parquet' + ).read_bytes() == before + + +def test_helpers_reject_paths_outside_marked_root( + tmp_path: Path, +) -> None: + ''' + A marker must not authorize traversal or symlinked paths. + + FQME text becomes part of the active Parquet filename and fresh + reset moves that path. Prove parent traversal is rejected, then + replace the evidence directory with a symlink and prove path + construction refuses to cross the disposable-root boundary. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + helpers['bfq_init'](run_root, validate_piker=False) + + with pytest.raises(ValueError, match='Unsafe FQME'): + helpers['_bfq_paths']( + run_root, + '../../outside', + 60, + ) + + evidence: Path = run_root / 'evidence' + outside: Path = tmp_path / 'outside' + outside.mkdir() + evidence.rmdir() + evidence.symlink_to(outside, target_is_directory=True) + with pytest.raises(RuntimeError, match='symlinked path'): + helpers['bfq_init']( + run_root, + validate_piker=False, + ) + with pytest.raises(RuntimeError, match='escaped run root'): + helpers['_bfq_paths'](run_root, 'x.test', 60) + with pytest.raises(RuntimeError, match='Unsafe evidence'): + helpers['bfq_compare'](run_root, 'before', 'after') + + +def test_helpers_reject_dangling_evidence_symlinks( + tmp_path: Path, +) -> None: + ''' + Final evidence names must not follow dangling external symlinks. + + ``Path.exists()`` is false for a dangling symlink, which can make + a check-then-write path create its external target. Link the case + record and seed outside a marked root. Prove both writers reject + the final component before creating external bytes. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + helpers['bfq_init'](run_root, validate_piker=False) + outside_case: Path = tmp_path / 'outside-case.json' + case_path: Path = run_root / 'case.json' + case_path.symlink_to(outside_case) + with pytest.raises(RuntimeError, match='symlinked case'): + helpers['bfq_record_case']( + run_root, + fqme='x.test', + period_s=60, + provider='fake', + symptom='unsafe path test', + ) + assert not outside_case.exists() + + case_path.unlink() + source: Path = tmp_path / 'source.parquet' + mk_frame((60, 120)).write_parquet(source) + outside_seed: Path = tmp_path / 'outside-seed.parquet' + seed: Path = run_root / 'evidence' / 'seed.parquet' + seed.symlink_to(outside_seed) + with pytest.raises(RuntimeError, match='symlinked output'): + helpers['bfq_seed']( + run_root, + 'x.test', + 60, + source, + ) + assert not outside_seed.exists() + + +def test_bfq_use_updates_xonsh_subprocess_environment( + tmp_path: Path, +) -> None: + ''' + Disposable XDG selection must reach commands launched by xonsh. + + Updating only ``os.environ`` does not mutate xonsh's environment + mapping, so manual chart commands previously retained normal user + storage. Source the real snippet in xonsh, initialize a run root, + and assert a subprocess sees the disposable XDG path. + + ''' + repo: Path = Path(__file__).parents[1] + snippet: Path = repo / 'snippets' / 'nativedb_backfill_audit.xsh' + run_root: Path = tmp_path / 'run' + code: str = ( + f'source {snippet}; ' + f'bfq_init(p\'{run_root}\', validate_piker=False); ' + 'env' + ) + result = subprocess.run( + ['xonsh', '-c', code], + check=False, + capture_output=True, + cwd=repo, + text=True, + ) + + assert result.returncode == 0, result.stderr + expected: str = f'XDG_CONFIG_HOME={run_root / "xdg"}' + assert result.stdout.count(expected) == 2 + + +def test_init_validates_console_script_against_worktree( + tmp_path: Path, +) -> None: + ''' + An installed console script must import the reviewed worktree. + + A shared virtualenv's ``piker`` entry point can resolve its + editable root checkout, where a new audit command is absent, even + while the operator is in this linked worktree. Run initialization + and prove its PYTHONPATH-bound preflight sees the local command + before it records the executable and repository identity. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + + helpers['bfq_init'](run_root) + + marker: dict = json.loads( + (run_root / '.piker-backfill-qualification').read_text() + ) + assert Path(marker['repo_root']) == Path(__file__).parents[1] + + +def test_phase_compare_detects_preservation_and_seam_changes( + tmp_path: Path, +) -> None: + ''' + Lifecycle comparison must preserve data without hiding seams. + + Restart and append may replace a matching provider seam while old + timestamps remain. Save two read-only snapshots where the second + appends a row and updates a close. Prove checks pass while the + the changed timestamp stays explicit operator evidence. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + init_case(helpers, run_root) + evidence: Path = run_root / 'evidence' + before_path: Path = evidence / 'fresh.parquet' + after_path: Path = evidence / 'restart.parquet' + mk_frame((60, 120), (1, 2)).write_parquet(before_path) + mk_frame((60, 120, 180), (1, 20, 3)).write_parquet(after_path) + + for phase, path in ( + ('fresh', before_path), + ('restart', after_path), + ): + write_phase_evidence(run_root, phase, path) + + comparison: dict = helpers['bfq_compare']( + run_root, + 'fresh', + 'restart', + require_newer=True, + ) + + assert comparison['passed'] is True + assert comparison['missing_timestamp_count'] == 0 + assert comparison['changed_common_row_count'] == 1 + assert comparison['changed_common_timestamps'] == [120] + assert comparison['checks']['maximum_advanced'] is True + with pytest.raises(ValueError, match='phases must differ'): + helpers['bfq_compare'](run_root, 'fresh', 'fresh') + + +def test_repair_compare_allows_invalid_and_duplicate_removal( + tmp_path: Path, +) -> None: + ''' + Repair may remove invalid rows but must preserve valid times. + + Known-bad history may contain zero-epoch and duplicate rows. Both + should disappear during canonical rewrite. Compare + that baseline to clean extended history and prove preservation + ignores invalid epoch and duplicate multiplicity while recording + their before and after counts. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + init_case(helpers, run_root) + evidence: Path = run_root / 'evidence' + before_path: Path = evidence / 'baseline.parquet' + after_path: Path = evidence / 'repair.parquet' + mk_frame((0, 60, 60, 120)).write_parquet(before_path) + mk_frame((60, 120, 180)).write_parquet(after_path) + + for phase, path in ( + ('baseline', before_path), + ('repair', after_path), + ): + write_phase_evidence(run_root, phase, path) + + comparison: dict = helpers['bfq_compare']( + run_root, + 'baseline', + 'repair', + require_newer=True, + ) + + assert comparison['passed'] is True + assert comparison['missing_timestamp_count'] == 0 + assert comparison['before_duplicate_excess'] == 1 + assert comparison['after_duplicate_excess'] == 0 + + +def test_malformed_baseline_produces_failed_comparison_evidence( + tmp_path: Path, +) -> None: + ''' + Comparison must report unauditable baselines instead of crashing. + + Raw audit accepts malformed Parquet so humans can save evidence + before repair. Write a baseline with textual timestamps and no + close column, compare it with canonical history, and prove the + helper writes limitations plus failed checks before raising its + qualification assertion. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + init_case(helpers, run_root) + evidence: Path = run_root / 'evidence' + before_path: Path = evidence / 'broken.parquet' + after_path: Path = evidence / 'repair.parquet' + mk_frame((60, 120)).drop('close').with_columns( + pl.Series('time', ['60', '120']) + ).write_parquet(before_path) + mk_frame((60, 120, 180)).write_parquet(after_path) + + for phase, path in ( + ('broken', before_path), + ('repair', after_path), + ): + write_phase_evidence(run_root, phase, path) + + with pytest.raises(AssertionError): + helpers['bfq_compare']( + run_root, + 'broken', + 'repair', + ) + + comparison: dict = json.loads( + ( + evidence + /'compare-broken-to-repair.json' + ).read_text() + ) + assert comparison['passed'] is False + assert comparison['checks']['no_timestamps_lost'] is False + assert 'before_time_nonnumeric' in comparison['limitations'] + assert ( + 'common_value_comparison_unavailable' + in comparison['limitations'] + ) + + +def test_compare_rejects_modified_or_cross_case_evidence( + tmp_path: Path, +) -> None: + ''' + Lifecycle checks must not combine unrelated or altered artifacts. + + JSON can remain plausible when a phase snapshot is replaced or a + report was generated for another market. Bind each report to its + checksum and fixed case metadata. Prove both changes abort before + abort before a comparison result can be accepted. + + ''' + helpers: dict = load_helpers() + run_root: Path = tmp_path / 'run' + init_case(helpers, run_root) + evidence: Path = run_root / 'evidence' + before: Path = evidence / 'before.parquet' + after: Path = evidence / 'after.parquet' + mk_frame((60, 120)).write_parquet(before) + mk_frame((60, 120, 180)).write_parquet(after) + write_phase_evidence(run_root, 'before', before) + write_phase_evidence(run_root, 'after', after) + + after_report: Path = evidence / 'after.json' + report_bytes: bytes = after_report.read_bytes() + after_report.write_bytes(report_bytes + b'\n') + with pytest.raises(RuntimeError, match='report checksum'): + helpers['bfq_compare'](run_root, 'before', 'after') + after_report.write_bytes(report_bytes) + + mk_frame((60, 180)).write_parquet(before) + with pytest.raises(RuntimeError, match='snapshot checksum'): + helpers['bfq_compare'](run_root, 'before', 'after') + + other: Path = evidence / 'other.parquet' + mk_frame((60, 120)).write_parquet(other) + write_phase_evidence( + run_root, + 'other', + other, + fqme='other.test', + ) + with pytest.raises(RuntimeError, match='does not match'): + helpers['bfq_compare'](run_root, 'other', 'after')