Repair legacy timestamps in `NativeDB` merges

Keep incoming provider frames strict while dropping non-positive rows
from persisted history before merge validation.

This lets known-bad MNQ history repair itself on the next valid IB
frame without losing valid old bars or publishing noncanonical indexes.

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

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

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

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

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

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

piker/storage/nativedb.py

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

tests/test_storage_nativedb.py

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