Repair IB startup gaps in `ShmArray`

Keep reverse history and stored SHM layout aligned so sparse IB
frames do not expose permanent zero-time reservations.

Deats,
- omit inclusive query endpoints from SHM but retain full storage
- serialize and bound farm resets, blanks, and cancellations
- run the final null sweep after reverse retrieval completes
- require valid timestamp boundaries for synthetic gap rows
- cover the live QQQ 653-row failure and reset interleavings

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

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

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

@@ -107,6 +107,8 @@ Deterministic or local first-pass targets:
 - `tests/test_storage_audit.py`
 - `tests/test_backfill_audit_snippet.py`
 - `tests/test_ib_history.py`
+- `tests/test_history_backfill.py`
+- `tests/test_ldshm.py`
 - `tests/test_accounting.py::test_account_file_default_empty`
 - `tests/test_services.py::test_runtime_boot`
 - `tests/test_services.py::test_datad_spawn`
@@ -190,6 +192,8 @@ tests/
   test_ems.py               actor, EMS, and paper-position behavior
   test_feeds.py             live Binance/Kraken feeds and shared memory
   test_ib_history.py        deterministic IB history request formatting
+  test_history_backfill.py  deterministic history/SHM orchestration
+  test_ldshm.py             ldshm unpublished-slot guard
   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
@@ -205,7 +209,9 @@ tests/
 | `piker/watchlists/` | `tests/test_watchlists.py` | CLI suite is skipped |
 | `piker/storage/_audit.py`, `piker/storage/cli.py` | `tests/test_storage_audit.py` | direct Typer app, no actor |
 | `snippets/nativedb_backfill_audit.xsh` | `tests/test_backfill_audit_snippet.py` | disposable paths only |
-| `piker/brokers/ib/api.py` history formatting | `tests/test_ib_history.py` | fake client, no network |
+| `piker/brokers/ib/api.py`, `feed.py` history | `tests/test_ib_history.py` | fake client, no network |
+| `piker/tsp/_history.py` | `tests/test_history_backfill.py` | fake provider/storage/SHM |
+| `piker/storage/cli.py` ldshm null-slot guard | `tests/test_ldshm.py` | synthetic timestamps, no SHM mutation |
 | `piker/config.py` | `test_account_file_default_empty` | root-network test has a known mismatch |
 | `piker/accounting/` | targeted accounting node | some cases use live/configured state |
 | `piker/ui/_style.py`, `piker/ui/qt.py` | `tests/test_dpi_font.py` | GUI/config-isolated opt-in |

ai/prompt-io/opencode/20260729T041723Z_ce33deb6_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-29T04:17:23Z
+git_ref: ce33deb6
+scope: code
+substantive: true
+raw_file: 20260729T041723Z_ce33deb6_prompt_io.raw.md
+---
+
+## Prompt
+
+Diagnose why the chart still showed a zero first OHLCV row after the
+epoch-zero timestamp repair, causing auto-y-ranging to include zero.
+
+## Response summary
+
+Audited live 60s and 1s NativeDB files and found one exact 60s first-slot
+sentinel at epoch 60 with a completely zero OHLCV payload. Added narrowly
+scoped read and merge repair, matching incoming validation, sentinel-only
+startup handling, canonical reindexing, and explicit audit classification.
+Preserved modern all-zero bars as unclassified evidence rather than assuming
+all zero-priced instruments are corrupt.
+
+## Files changed
+
+- `piker/storage/nativedb.py` - first-slot sentinel repair and validation
+- `piker/storage/_audit.py` - sentinel classification and zero-row evidence
+- `tests/test_storage_nativedb.py` - hydration, merge, and boundary regressions
+- `tests/test_storage_audit.py` - sentinel and modern-zero audit regressions
+- `ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not been edited by the human.

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

@@ -0,0 +1,31 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-29T04:17:23Z
+git_ref: ce33deb6
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/storage/nativedb.py piker/storage/_audit.py tests/test_storage_nativedb.py tests/test_storage_audit.py`
+
+Fixed the remaining legacy first-slot sentinel observed in live MNQ 60s
+history: `time=60` with zero open, high, low, close, and volume. NativeDB now
+repairs the exact sentinel before read hydration and merge, reindexes the
+filtered view, and persists the repaired frame on the next valid update.
+
+The repair remains narrowly scoped. Modern all-zero OHLC bars stay valid for
+spreads or synthetic instruments, null-containing malformed rows still fail
+validation, new writes can not recreate the first-slot sentinel, and a file
+containing only sentinels loads as no history so fresh backfill can proceed.
+
+The raw audit reports total all-zero price rows, classifies first-slot
+sentinels as structural violations, and leaves later all-zero rows visible as
+unclassified warnings.
+
+Verification generated with the patch:
+
+- NativeDB, audit, history, xonsh, and IB regressions: 61 passed
+- focused Ruff: passed
+- `git diff --check`: passed
+- final adversarial review: no findings

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

@@ -0,0 +1,34 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 6409e3bb-1a9a-4ada-b6c4-c554115143e9
+timestamp: 2026-07-29T18:54:06Z
+git_ref: d359b1cb
+scope: code
+substantive: true
+raw_file: 20260729T185406Z_d359b1cb_prompt_io.raw.md
+---
+
+## Prompt
+
+The user reported that QQQ 60-second zeros appeared as a flat line after
+reload, clarified that the outstanding concern was the prior `ldshm`
+failure, and instructed the agent to implement the more focused fix.
+
+## Response summary
+
+Changed `ldshm` to classify invalid SHM snapshots before downstream
+analysis. Cadence is inferred only from observed finite positive steps;
+buffers containing non-finite or non-positive timestamps, or lacking a
+positive cadence, are skipped rather than partially replacement-written
+or reloaded. Added deterministic helper and command-level regressions.
+
+## Files changed
+
+- `piker/storage/cli.py` - inspect cadence and reject invalid SHM snapshots before mutation.
+- `tests/test_ldshm.py` - cover QQQ-like zeros, corrupt timestamps, cadence selection, degenerate buffers, and mutation prevention.
+- `.claude/skills/run-tests/test-harness-reference.md` - register the deterministic regression target.
+
+## Human edits
+
+None - generated output remains uncommitted.

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

@@ -0,0 +1,58 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-07-29T18:54:06Z
+git_ref: d359b1cb
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+The user clarified that the QQQ 60-second chart now displayed a flat
+forward-filled segment, then asked whether the previously reported
+`piker store ldshm` failure had been fixed. After being told it had not,
+the user instructed: "ok but i read above you have a more focussed fix
+todo now? so have at it!"
+
+The implementation focused on the exact failure boundary. It does not
+filter invalid rows into a partial replacement and does not mutate live
+SHM. Instead, `ldshm` inspects one captured Polars snapshot, infers its
+cadence from observed finite positive timestamp steps, and skips the
+entire buffer before analysis, persistence, or reload when any timestamp
+is non-finite or non-positive.
+
+> `git diff HEAD~1..HEAD -- piker/storage/cli.py`
+
+Generated `_shm_period_and_invalid_count()` and integrated its result in
+`ldshm`. The helper selects the most frequent observed positive step,
+using the smallest observed step as the deterministic tie-break. Short
+or cadence-degenerate buffers are skipped. Invalid snapshots are logged
+and cannot reach deduplication, NativeDB replacement, or SHM reload.
+
+> `git diff HEAD~1..HEAD -- tests/test_ldshm.py`
+
+Generated deterministic regressions for the QQQ-shaped interior zero
+run, observed-step cadence selection, all invalid timestamp classes,
+short and degenerate buffers, and command-level prevention of dedupe,
+storage writes, and reload. The command-level test uses an immutable
+Polars snapshot and failing spies, without live SHM mutation or timing.
+
+> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
+
+Added the deterministic `ldshm` regression target and its change-to-test
+mapping to the repository test harness reference.
+
+Verification output:
+
+```text
+..........................................................               [100%]
+58 passed in 0.70s
+```
+
+`git diff --check` passed. Ruff was unavailable in the worktree's
+existing environment. Three adversarial review passes found and drove
+fixes for partial-replacement data loss, invented median timeframes,
+mixed SHM observations, and pre-guard null-analysis failures. The final
+review reported no remaining findings.
+
+The patch intentionally does not claim that `ldshm` is safe to run as a
+general concurrent repair writer. It handles invalid live snapshots by
+refusing mutation rather than weakening NativeDB validation.

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

@@ -0,0 +1,37 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 253ade33-07ba-45b5-8839-be35d17ec164
+timestamp: 2026-07-29T20:40:46Z
+git_ref: ad6c560f
+scope: code
+substantive: true
+raw_file: 20260729T204046Z_ad6c560f_prompt_io.raw.md
+---
+
+## Prompt
+
+The user reported that the QQQ 1-second chart still contained a gap after
+the `ldshm` fix, asked the agent to inspect it, and authorized continued
+implementation after a short pause and shutdown of the live workspace.
+
+## Response summary
+
+Diagnosed a 653-row active RT SHM null reservation caused by inclusive IB
+reverse-query overlaps and startup repair orchestration. Prevented extra
+physical SHM overlap rows, serialized and bounded IB history recovery,
+made blank/error responses terminate, and constrained synthetic repair to
+valid interior timestamp boundaries. Added deterministic concurrency and
+data-layout regressions.
+
+## Files changed
+
+- `piker/brokers/ib/feed.py` - bound and serialize IB history reset/retry behavior.
+- `piker/tsp/_history.py` - prevent SHM overlap skew and safely finalize null repair.
+- `tests/test_ib_history.py` - cover blank, cancellation, reset, and pacing interleavings.
+- `tests/test_history_backfill.py` - cover overlap and null-repair invariants.
+- `.claude/skills/run-tests/test-harness-reference.md` - register deterministic coverage.
+
+## Human edits
+
+None - generated output remains uncommitted.

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

@@ -0,0 +1,72 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-07-29T20:40:46Z
+git_ref: ad6c560f
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+The user reported that QQQ still displayed a gap on the 1-second chart
+after the focused `ldshm` fix and asked the agent to inspect it. The user
+then paused and resumed implementation while shutting down the live piker
+trading workspace.
+
+Read-only live inspection found the active datad generation held 653
+contiguous zero-time rows in QQQ's 1-second RT SHM, while durable Parquet
+contained no zero rows. The run occupied absolute indexes 168148 through
+168800 between timestamps 1785348398 and 1785349051. The stale actor
+generation and active 60-second history were separately identified.
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
+
+Generated history changes that exclude each inclusive reverse-query
+endpoint already present in SHM while retaining the full provider frame
+for NativeDB merge. Startup null repair now runs after reverse backfill,
+uses bounded UI notifications, handles provider no-data without an
+interactive pause, fills only exact interior zero groups, and refuses
+synthetic timestamps unless both valid boundaries establish the expected
+cadence.
+
+> `git diff HEAD~1..HEAD -- piker/brokers/ib/feed.py`
+
+Generated IB history reset changes that make blank responses complete,
+bound repeated cancellation and reset failures per request, serialize
+timeout and pacing farm resets with one actor-global Trio lock, carry
+explicit reset results, and coordinate pacing handoff with queued timeout
+waiters.
+
+> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py`
+
+Generated regressions for inclusive endpoint exclusion, full NativeDB
+provider deltas, strict synthetic gap boundaries, empty-provider local
+fallback, bounded notifications, and preservation of valid margin rows.
+
+> `git diff HEAD~1..HEAD -- tests/test_ib_history.py`
+
+Generated deterministic regressions for blank IB responses, cancellation
+limits, stalled reset cancellation, pacing reset failure limits,
+cross-request reset serialization, timeout-to-pacing handoff, and
+successful handoff timeout restart.
+
+> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
+
+Registered the deterministic history-backfill target and expanded the IB
+history mapping to include `feed.py`.
+
+Verification output:
+
+```text
+........................................................................ [ 93%]
+.....                                                                    [100%]
+77 passed in 2.12s
+```
+
+`git diff --check` passed. Multiple adversarial review rounds identified
+and drove fixes for unbounded IB empty/cancellation/reset paths, stale
+reset ownership, actor-global failure poisoning, concurrent farm resets,
+double-counted pacing handoffs, physical endpoint overlap, and unsafe
+synthetic timestamp boundaries. Final review reported no findings.
+
+Live validation was intentionally deferred because the user shut down the
+trading workspace. A fresh datad/chart startup is required to regenerate
+SHM and verify that QQQ no longer contains the 653-row null run.

piker/brokers/ib/feed.py

@@ -32,7 +32,6 @@ import time
 from typing import (
     Any,
     Callable,
-    TYPE_CHECKING,
 )
 
 from async_generator import aclosing
@@ -75,9 +74,6 @@ from ._util import (
 )
 from .symbols import get_mkt_info
 
-if TYPE_CHECKING:
-    from trio._core._run import Task
-
 log = get_logger(
     name=__name__,
 )
@@ -353,6 +349,7 @@ async def wait_on_data_reset(
         tuple[
             trio.CancelScope,
             trio.Event,
+            list[bool],
         ]
     ] = trio.TASK_STATUS_IGNORED,
 ) -> bool:
@@ -396,8 +393,9 @@ async def wait_on_data_reset(
     client: Client = proxy._aio_ns
 
     done = trio.Event()
+    result: list[bool] = []
     with trio.move_on_after(timeout) as cs:
-        task_status.started((cs, done))
+        task_status.started((cs, done, result))
 
         log.warning(
             'Sending DATA RESET request:\n'
@@ -413,6 +411,7 @@ async def wait_on_data_reset(
                 'NO VNC DETECTED!\n'
                 'Manually press ctrl-alt-f on your IB java app'
             )
+            result.append(False)
             done.set()
             return False
 
@@ -427,6 +426,7 @@ async def wait_on_data_reset(
         ]:
             await ev.wait()
             log.info(f"{name} DATA RESET")
+            result.append(True)
             done.set()
             return True
 
@@ -435,12 +435,12 @@ async def wait_on_data_reset(
             'Data reset task canceled?'
         )
 
+    result.append(False)
     done.set()
     return False
 
 
-_data_resetter_task: Task | None = None
+_data_reset_lock = trio.Lock()
-_failed_resets: int = 0
 
 
 async def get_bars(
@@ -479,8 +479,11 @@ async def get_bars(
     `.ib.api.Client` methods.
 
     '''
-    global _data_resetter_task, _failed_resets
     nodatas_count: int = 0
+    cancelled_count: int = 0
+    failed_resets: int = 0
+    pacing_reset_pending: bool = False
+    pacing_reset_completed: bool = False
 
     data_cs: trio.CancelScope | None = None
     result: tuple[
@@ -493,10 +496,16 @@ async def get_bars(
 
     async def query():
 
-        global _failed_resets
+        nonlocal result, data_cs, end_dt
-        nonlocal result, data_cs, end_dt, nodatas_count
+        nonlocal nodatas_count, cancelled_count, failed_resets
+        nonlocal pacing_reset_pending, pacing_reset_completed
 
-        while _failed_resets < max_failed_resets:
+        while True:
+            if failed_resets >= max_failed_resets:
+                raise DataUnavailable(
+                    f'IB history reset failed {failed_resets} '
+                    f'times for {fqme}'
+                )
             try:
                 (
                     bars,
@@ -528,6 +537,10 @@ async def get_bars(
                     )
                     # NOTE: REQUIRED to pass back value..
                     result = None
+                    failed_resets = 0
+                    result_ready.set()
+                    if data_cs:
+                        data_cs.cancel()
                     return None
 
                 # not enough bars signal, likely due to venue
@@ -587,6 +600,7 @@ async def get_bars(
                     first_dt,
                     last_dt,
                 )
+                failed_resets = 0
 
                 # signal data reset loop parent task
                 result_ready.set()
@@ -642,6 +656,12 @@ async def get_bars(
                         'Query cancelled by IB (:eyeroll:):\n'
                         f'{err.message}'
                     )
+                    cancelled_count += 1
+                    if cancelled_count >= max_failed_resets:
+                        raise DataUnavailable(
+                            f'IB cancelled {cancelled_count} history '
+                            f'queries for {fqme}'
+                        )
                     continue
 
                 elif (
@@ -662,6 +682,7 @@ async def get_bars(
                     )
 
                     client = proxy._aio_ns.ib.client
+                    pacing_reset_pending = True
 
                     # cancel any existing reset task
                     if data_cs:
@@ -669,69 +690,81 @@ async def get_bars(
                         data_cs.cancel()
 
                     # spawn new data reset task
-                    data_cs, reset_done = await tn.start(
+                    async with _data_reset_lock:
-                        partial(
+                        (
-                            wait_on_data_reset,
+                            data_cs,
-                            proxy,
+                            reset_done,
-                            reset_type='connection'
+                            reset_result,
+                        ) = await tn.start(
+                            partial(
+                                wait_on_data_reset,
+                                proxy,
+                                reset_type='connection'
+                            )
                         )
-                    )
+                        await reset_done.wait()
-                    if reset_done:
+                    if reset_result[0]:
-                        _failed_resets = 0
+                        failed_resets = 0
                     else:
-                        _failed_resets += 1
+                        failed_resets += 1
+                    pacing_reset_pending = False
+                    pacing_reset_completed = True
                     continue
 
                 else:
                     raise
 
-    # TODO: make this global across all history task/requests
-    # such that simultaneous symbol queries don't try data resettingn
-    # too fast..
-    unset_resetter: bool = False
     async with (
         tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn
     ):
 
-        # start history request that we allow
+        # Run history until a result or bounded reset failure.
-        # to run indefinitely until a result is acquired
         tn.start_soon(query)
 
-        # start history reset loop which waits up to the timeout
+        # Serialize farm resets across every history requester.
-        # for a result before triggering a data feed reset.
         while not result_ready.is_set():
-
             with trio.move_on_after(feed_reset_timeout):
                 await result_ready.wait()
                 break
 
-            if _data_resetter_task:
+            async with _data_reset_lock:
-                # don't double invoke the reset hack if another
+                if result_ready.is_set():
-                # requester task already has it covered.
+                    break
-                continue
+                if pacing_reset_completed:
+                    pacing_reset_completed = False
+                    continue
 
-            else:
+                (
-                _data_resetter_task = trio.lowlevel.current_task()
+                    data_cs,
-                unset_resetter: bool = True
+                    reset_done,
-
+                    reset_result,
-            # spawn new data reset task
+                ) = await tn.start(
-            data_cs, reset_done = await tn.start(
+                    partial(
-                partial(
+                        wait_on_data_reset,
-                    wait_on_data_reset,
+                        proxy,
-                    proxy,
+                        reset_type='data',
-                    reset_type='data',
+                    )
                 )
-            )
+                await reset_done.wait()
-            # sync wait on reset to complete
-            await reset_done.wait()
 
-    _data_resetter_task = (
+                if (
-        None
+                    result_ready.is_set()
-        if unset_resetter
+                    or
-        else _data_resetter_task
+                    reset_result[0]
-    )
+                ):
-    assert result
+                    failed_resets = 0
+                elif pacing_reset_pending:
+                    # Pacing recovery deliberately cancelled this reset
+                    # and will consume the retry itself.
+                    pass
+                else:
+                    failed_resets += 1
+                    if failed_resets >= max_failed_resets:
+                        raise DataUnavailable(
+                            f'IB history reset failed '
+                            f'{failed_resets} times for {fqme}'
+                        )
     return (
         result,
         data_cs is not None,

piker/storage/_audit.py

@@ -30,6 +30,12 @@ _value_fields: tuple[str, ...] = (
     'close',
     'volume',
 )
+_price_fields: tuple[str, ...] = (
+    'open',
+    'high',
+    'low',
+    'close',
+)
 
 
 def _utc_str(timestamp: float|int|None) -> str|None:
@@ -494,6 +500,36 @@ def audit_ohlcv_frame(
         if counts['nonfinite'] != 0:
             all_values_finite = False
 
+    all_zero_price_rows: int|None = None
+    epoch_sentinel_rows: int|None = None
+    if all(
+        field in df.columns
+        and
+        df[field].dtype.is_numeric()
+        for field in _price_fields
+    ):
+        zero_prices: pl.Expr = pl.all_horizontal([
+            pl.col(field) == 0
+            for field in _price_fields
+        ]).fill_null(False)
+        all_zero_price_rows = int(
+            df.select(zero_prices).to_series().sum()
+            or 0
+        )
+        if (
+            'time' in df.columns
+            and
+            df['time'].dtype.is_numeric()
+        ):
+            epoch_sentinel_rows = int(
+                df.select(
+                    (pl.col('time') <= period_s)
+                    &
+                    zero_prices
+                ).to_series().sum()
+                or 0
+            )
+
     schema_ok: bool = bool(
         not missing
         and
@@ -516,6 +552,8 @@ def audit_ohlcv_frame(
         violations.append('canonical_dtypes')
     if not all_values_finite:
         violations.append('nonfinite_ohlcv')
+    if epoch_sentinel_rows:
+        violations.append('epoch_ohlc_sentinel')
     if timestamps.get('nonfinite') != 0:
         violations.append('nonfinite_timestamps')
     if (
@@ -540,6 +578,15 @@ def audit_ohlcv_frame(
     warnings: list[str] = []
     if gaps['count']:
         warnings.append('positive_time_gaps_unclassified')
+    unclassified_zero_rows: int|None = None
+    if all_zero_price_rows is not None:
+        unclassified_zero_rows = (
+            all_zero_price_rows
+            -
+            (epoch_sentinel_rows or 0)
+        )
+        if unclassified_zero_rows:
+            warnings.append('all_zero_ohlc_prices_unclassified')
 
     structural_ok: bool = not violations
     gap_free: bool|None = (
@@ -573,6 +620,9 @@ def audit_ohlcv_frame(
             'nan_by_column': nans,
             'infinity_by_column': infinities,
             'all_finite': all_values_finite,
+            'all_zero_price_rows': all_zero_price_rows,
+            'epoch_sentinel_rows': epoch_sentinel_rows,
+            'unclassified_zero_price_rows': unclassified_zero_rows,
         },
         'gaps': gaps,
         'result': {

piker/storage/cli.py

@@ -60,6 +60,37 @@ if TYPE_CHECKING:
 store = typer.Typer()
 
 
+def _shm_period_and_invalid_count(
+    times: np.ndarray,
+
+) -> tuple[float|None, int]:
+    '''
+    Infer cadence without treating unpublished slots as samples.
+
+    '''
+    valid: np.ndarray = (
+        np.isfinite(times)
+        &
+        (times > 0)
+    )
+    invalid_count: int = int(np.count_nonzero(~valid))
+    published: np.ndarray = times[valid]
+    if published.size < 2:
+        return None, invalid_count
+
+    steps: np.ndarray = np.diff(published)
+    positive_steps: np.ndarray = steps[steps > 0]
+    if not positive_steps.size:
+        return None, invalid_count
+
+    periods, counts = np.unique(
+        positive_steps,
+        return_counts=True,
+    )
+    period: float = float(periods[np.argmax(counts)])
+    return period, invalid_count
+
+
 def _render_audit_report(report: dict) -> None:
     '''
     Render a compact human summary and explicit gap endpoints.
@@ -506,38 +537,25 @@ def ldshm(
                 shm_df,
             ) in tsp.iter_dfs_from_shms(fqme):
 
-                times: np.ndarray = shm.array['time']
+                times: np.ndarray = shm_df['time'].to_numpy()
-                d1: float = float(times[-1] - times[-2])
+                (
-                d2: float = 0
+                    period_s,
-                # XXX, take a median sample rate if sufficient data
+                    invalid_count,
-                if times.size > 2:
+                ) = _shm_period_and_invalid_count(times)
-                    d2: float = float(times[-2] - times[-3])
+                if period_s is None:
-                    med: float = np.median(np.diff(times))
+                    log.warning(
-                    if (
+                        f'Could not infer a positive sample period '
-                        d1 < 1.
+                        f'for {shmfile.name}; skipping buffer\n'
-                        and d2 < 1.
+                    )
-                        and med < 1.
+                    continue
-                    ):
-                        raise ValueError(
-                            f'Something is wrong with time period for {shm}:\n{times}'
-                        )
-                period_s: float = float(max(d1, d2, med))
                 log.info(
                     f'Processing shm buffer:\n'
                     f'  file: {shmfile.name}\n'
                     f'  period: {period_s}s\n'
                 )
 
-                null_segs: tuple = tsp.get_null_segs(
-                    frame=shm.array,
-                    period=period_s,
-                )
-
                 # TODO: call null-seg fixer somehow?
-                if null_segs:
+                if invalid_count:
-
-                    if tractor.runtime._state.is_debug_mode():
-                        await tractor.pause()
                 #     async with (
                 #         trio.open_nursery() as tn,
                 #         mod.open_history_client(
@@ -553,6 +571,14 @@ def ldshm(
                 #             sampler_stream=sampler_stream,
                 #             mkt=mkt,
                 #         ))
+                    if tractor.runtime._state.is_debug_mode():
+                        await tractor.pause()
+                    log.warning(
+                        f'Found {invalid_count} invalid SHM row(s) '
+                        f'in {shmfile.name}; skipping persistence '
+                        f'and reload\n'
+                    )
+                    continue
 
                 # over-write back to shm?
                 wdts: pl.DataFrame  # with dts

piker/storage/nativedb.py

@@ -74,6 +74,12 @@ from . import TimeseriesNotFound
 
 
 log = get_logger('storage.nativedb')
+_price_fields: tuple[str, ...] = (
+    'open',
+    'high',
+    'low',
+    'close',
+)
 
 
 def detect_period(shm: ShmArray) -> float:
@@ -233,6 +239,8 @@ class NativeStorageClient:
             ) from fnfe
 
         times = array['time']
+        if array.size == 0:
+            return None
         return (
             array,
             from_timestamp(times[0]),
@@ -319,6 +327,41 @@ class NativeStorageClient:
             .cast(schema)
         )
 
+    def _repair_stored_ohlcv(
+        self,
+        df: pl.DataFrame,
+        timeframe: int,
+
+    ) -> pl.DataFrame:
+        '''
+        Drop known legacy sentinels before hydration or merge.
+
+        '''
+        stored: pl.DataFrame = self._canonicalize_ohlcv(df)
+        zero_prices: pl.Expr = pl.all_horizontal([
+            pl.col(field) == 0
+            for field in _price_fields
+        ]).fill_null(False)
+        invalid: pl.Expr = (
+            (pl.col('time') <= 0)
+            |
+            (
+                (pl.col('time') <= timeframe)
+                &
+                zero_prices
+            )
+        )
+        stored_len: int = stored.height
+        stored = stored.filter(~invalid)
+        dropped: int = stored_len - stored.height
+        if dropped:
+            log.warning(
+                f'Dropping {dropped} invalid persisted OHLCV '
+                f'row(s) during read repair'
+            )
+            stored = self._canonicalize_ohlcv(stored)
+        return stored
+
     async def read_ohlcv(
         self,
         fqme: str,
@@ -331,7 +374,10 @@ class NativeStorageClient:
             fqme,
             period=int(timeframe),
         )
-        df: pl.DataFrame = pl.read_parquet(path)
+        df: pl.DataFrame = self._repair_stored_ohlcv(
+            pl.read_parquet(path),
+            timeframe=int(timeframe),
+        )
 
         self._cache_df(
             fqme=fqme,
@@ -384,7 +430,7 @@ class NativeStorageClient:
         else:
             df = ohlcv
         df = self._canonicalize_ohlcv(df)
-        self._validate_ohlcv(df)
+        self._validate_ohlcv(df, timeframe)
 
         # TODO: in terms of managing the ultra long term data
         # -[ ] use a proper profiler to measure all this IO and
@@ -405,7 +451,7 @@ class NativeStorageClient:
         try:
             df.write_parquet(tmp_path)
             committed: pl.DataFrame = pl.read_parquet(tmp_path)
-            self._validate_ohlcv(committed)
+            self._validate_ohlcv(committed, timeframe)
             if not committed.equals(df):
                 raise IOError(
                     'Temporary parquet differs from input frame'
@@ -442,6 +488,7 @@ class NativeStorageClient:
     def _validate_ohlcv(
         self,
         df: pl.DataFrame,
+        timeframe: int,
 
     ) -> None:
         '''
@@ -483,6 +530,17 @@ class NativeStorageClient:
                 'OHLCV timestamps must be finite and positive'
             )
 
+        prices: np.ndarray = df.select(_price_fields).to_numpy()
+        epoch_sentinel: np.ndarray = (
+            (times <= timeframe)
+            &
+            np.all(prices == 0, axis=1)
+        )
+        if np.any(epoch_sentinel):
+            raise ValueError(
+                'OHLCV frame contains a zero-initialized epoch sentinel'
+            )
+
         if np.any(np.diff(times) <= 0):
             raise ValueError(
                 'OHLCV timestamps must be strictly increasing'
@@ -507,21 +565,14 @@ class NativeStorageClient:
         else:
             incoming = ohlcv
         incoming = self._canonicalize_ohlcv(incoming)
-        self._validate_ohlcv(incoming)
+        self._validate_ohlcv(incoming, timeframe)
 
         path: Path = self.mk_path(fqme, timeframe)
         if path.exists():
-            stored: pl.DataFrame = self._canonicalize_ohlcv(
+            stored: pl.DataFrame = self._repair_stored_ohlcv(
-                pl.read_parquet(path)
+                pl.read_parquet(path),
+                timeframe=timeframe,
             )
-            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',
@@ -537,7 +588,7 @@ class NativeStorageClient:
         else:
             merged = incoming
 
-        self._validate_ohlcv(merged)
+        self._validate_ohlcv(merged, timeframe)
         return self._write_ohlcv(
             fqme,
             merged,

piker/tsp/_history.py

@@ -206,6 +206,34 @@ async def notify_backfill(
         )
 
 
+def _synthetic_gap_times(
+    start_t: float,
+    right_t: float|None,
+    count: int,
+    timeframe: float,
+
+) -> np.ndarray|None:
+    '''
+    Build bounded synthetic timestamps without crossing valid data.
+
+    '''
+    times: np.ndarray = (
+        start_t
+        +
+        np.arange(1, count + 1)*timeframe
+    )
+    if right_t is None:
+        return None
+
+    if not np.isclose(
+        right_t,
+        start_t + (count + 1)*timeframe,
+    ):
+        return None
+
+    return times
+
+
 async def shm_push_in_between(
     shm: ShmArray,
     to_push: np.ndarray,
@@ -262,11 +290,11 @@ async def maybe_fill_null_segments(
 
     task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
 
-) -> list[Frame]:
+) -> None:
 
     task_status.started()
 
-    frame: Frame = shm.array
+    frame: Frame = shm.array.copy()
 
     # TODO, put in parent task/daemon root!
     import greenback
@@ -297,15 +325,26 @@ async def maybe_fill_null_segments(
             await tractor.pause()
             break
 
-        (
+        try:
-            array,
+            (
-            next_start_dt,
+                array,
-            next_end_dt,
+                next_start_dt,
-        ) = await get_hist(
+                next_end_dt,
-            timeframe,
+            ) = await get_hist(
-            start_dt=start_dt,
+                timeframe,
-            end_dt=end_dt,
+                start_dt=start_dt,
-        )
+                end_dt=end_dt,
+            )
+        except (
+            NoData,
+            DataUnavailable,
+        ) as exc:
+            log.warning(
+                f'Provider could not repair SHM gap:\n'
+                f'{start_dt} -> {end_dt}\n'
+                f'{exc}\n'
+            )
+            break
 
         if array.size == 0:
             log.warning(
@@ -314,7 +353,6 @@ async def maybe_fill_null_segments(
             )
             # ?TODO? do we want to remove the nulls and push
             # the close price here for the gap duration?
-            await tractor.pause()
             break
 
         if (
@@ -327,7 +365,7 @@ async def maybe_fill_null_segments(
                 f'frame_start_dt: {frame_start_dt!r}\n'
                 f'backfill_until_dt: {backfill_until_dt!r}\n'
             )
-            await tractor.pause()
+            break
 
         # XXX TODO: pretty sure if i plot tsla, btcusdt.binance
         # and mnq.cme.ib this causes a Qt crash XXDDD
@@ -349,21 +387,11 @@ async def maybe_fill_null_segments(
         # - remember that in the display side, only refersh this
         #   if the respective history is actually "in view".
         #   loop
-        try:
+        await notify_backfill(
-            await sampler_stream.send({
+            sampler_stream,
-                'broadcast_all': {
+            mkt,
-
+            timeframe,
-                    # XXX NOTE XXX: see the
+        )
-                    # `.ui._display.increment_history_view()` if block
-                    # that looks for this info to FORCE a hard viz
-                    # redraw!
-                    'backfilling': (mkt.fqme, timeframe),
-                },
-            })
-        except tractor.ContextCancelled:
-            # log.exception
-            await tractor.pause()
-            raise
 
     # RECHECK for more null-gaps
     frame: Frame = shm.array
@@ -379,7 +407,7 @@ async def maybe_fill_null_segments(
         (
             iabs_slices,
             iabs_zero_rows,
-            zero_t,
+            _zero_t,
         ) = null_segs
         log.warning(
             f'{len(iabs_slices)} NULL TIME SEGMENTS DETECTED!\n'
@@ -398,13 +426,61 @@ async def maybe_fill_null_segments(
             'close',
         ]
 
-        for istart, istop in iabs_slices:
+        zero_indexes: np.ndarray = np.asarray(iabs_zero_rows)
+        split_at: np.ndarray = (
+            np.flatnonzero(np.diff(zero_indexes) != 1)
+            +
+            1
+        )
+        zero_groups: list[np.ndarray] = np.split(
+            zero_indexes,
+            split_at,
+        )
+        repaired: int = 0
+        frame_start: int = int(frame['index'][0])
+        for zero_group in zero_groups:
+            istart: int = int(zero_group[0])
+            istop: int = int(zero_group[-1]) + 1
+            seed_index: int = istart - 1
+            if seed_index < frame_start:
+                log.warning(
+                    f'Can not forward-fill leading SHM nulls:\n'
+                    f'{istart}:{istop}\n'
+                )
+                continue
 
-            # get view into buffer for null-segment
+            seed: np.void = shm._array[seed_index]
+            start_t: float = seed['time']
+            if start_t <= 0:
+                log.warning(
+                    f'Can not forward-fill SHM nulls from '
+                    f'zero-time seed at {seed_index}\n'
+                )
+                continue
+
+            # Fill only the null rows, preserving both valid margins.
             gap: np.ndarray = shm._array[istart:istop]
+            cls: float = seed['close']
 
-            # copy the oldest OHLC samples forward
+            frame_end: int = int(frame['index'][-1])
-            cls: float = shm._array[istart]['close']
+            right_t: float|None = (
+                shm._array[istop]['time']
+                if istop <= frame_end
+                else None
+            )
+            gap_times: np.ndarray|None = _synthetic_gap_times(
+                start_t,
+                right_t,
+                len(gap),
+                timeframe,
+            )
+            if gap_times is None:
+                log.warning(
+                    f'Can not safely forward-fill SHM nulls:\n'
+                    f'left={start_t} right={right_t} '
+                    f'rows={len(gap)} period={timeframe}\n'
+                )
+                continue
 
             # TODO: how can we mark this range as being a gap tho?
             # -[ ] maybe pg finally supports nulls in ndarray to
@@ -413,30 +489,13 @@ async def maybe_fill_null_segments(
             #     another col/field to denote?
             gap[ohlc_fields] = cls
 
-            start_t: float = shm._array[istart]['time']
+            gap['time'] = gap_times
-            t_diff: float = (istop - istart)*timeframe
+            repaired += len(gap)
-
-            gap['time'] = np.arange(
-                start=start_t,
-                stop=start_t + t_diff,
-                step=timeframe,
-            )
 
             # TODO: reimpl using the new `.ui._remote_ctl` ctx
             # ideally using some kinda decent
             # tractory-reverse-lookup-connnection from some other
             # `Context` type thingy?
-            await sampler_stream.send({
-                'broadcast_all': {
-
-                    # XXX NOTE XXX: see the
-                    # `.ui._display.increment_history_view()` if block
-                    # that looks for this info to FORCE a hard viz
-                    # redraw!
-                    'backfilling': (mkt.fqme, timeframe),
-                },
-            })
-
             # TODO: interatively step through any remaining
             # time-gaps/null-segments and spawn piecewise backfiller
             # tasks in a nursery?
@@ -447,6 +506,13 @@ async def maybe_fill_null_segments(
             # -[ ] fill algo: do queries in alternating "latest, then
             #    earliest, then latest.. etc?"
 
+        if repaired:
+            await notify_backfill(
+                sampler_stream,
+                mkt,
+                timeframe,
+            )
+
 
 async def start_backfill(
     get_hist,
@@ -680,7 +746,12 @@ async def start_backfill(
                 array,
                 prepend_until_dt=backfill_until_dt,
             )
-            to_push: np.ndarray = to_store
+            # The requested end bar is already the oldest published
+            # sample. Keep it for idempotent storage merge, but do not
+            # consume another physical SHM row for the overlap.
+            to_push: np.ndarray = to_store[
+                to_store['time'] < last_start_dt.timestamp()
+            ]
             ln: int = len(to_push)
             if ln:
                 log.info(
@@ -1314,28 +1385,26 @@ async def tsdb_backfill(
 
                         log.info(f'Loaded {to_push.shape} datums from storage')
 
-                    # NOTE: ASYNC-conduct tsdb timestamp gap detection and backfill any
-                    # seemingly missing (null-time) segments..
-                    # TODO: ideally these can never exist!
-                    # -[ ] somehow it seems sometimes we're writing zero-ed
-                    #    segments to tsdbs during teardown?
-                    # -[ ] can we ensure that the backfiller tasks do this
-                    #    work PREVENTAVELY instead?
-                    # -[ ] fill in non-zero epoch time values ALWAYS!
-                    # await maybe_fill_null_segments(
-                    await tn.start(partial(
-                        maybe_fill_null_segments,
-
-                        shm=shm,
-                        timeframe=timeframe,
-                        get_hist=get_hist,
-                        sampler_stream=sampler_stream,
-                        mkt=mkt,
-                        backfill_until_dt=last_tsdb_dt,
-                    ))
-
             # 2nd nursery END
 
+            if last_tsdb_dt is not None:
+                # Repair only after reverse backfill releases providers
+                # such as IB which permit one history request at a time.
+                # TODO: ideally these null segments can never exist!
+                # -[ ] somehow it seems sometimes we're writing zero-ed
+                #    segments to tsdbs during teardown?
+                # -[ ] can we ensure that the backfiller tasks do this
+                #    work PREVENTAVELY instead?
+                # -[ ] fill in non-zero epoch time values ALWAYS!
+                await maybe_fill_null_segments(
+                    shm=shm,
+                    timeframe=timeframe,
+                    get_hist=get_hist,
+                    sampler_stream=sampler_stream,
+                    mkt=mkt,
+                    backfill_until_dt=last_tsdb_dt,
+                )
+
         # TODO: maybe start history anal and load missing "history
         # gaps" via backend..
 

tests/test_history_backfill.py

@@ -29,6 +29,8 @@ from piker.storage.nativedb import (
     ohlc_key_map,
 )
 from piker.tsp._history import (
+    _synthetic_gap_times,
+    maybe_fill_null_segments,
     notify_backfill,
     publish_latest_frame,
     start_backfill,
@@ -92,15 +94,22 @@ def test_empty_frame_completes() -> None:
 
 def test_storage_receives_full_provider_delta() -> None:
     '''
-    SHM capacity truncation does not truncate durable history.
+    SHM drops the published endpoint without truncating storage.
+
+    Reverse history queries include their requested end bar, which is
+    already the oldest sample in SHM. Publishing the overlap consumed
+    an extra physical row and made reserved null counts disagree with
+    elapsed timestamps. Return a frame with the overlap and prove
+    NativeDB receives the complete provider delta while SHM receives
+    only bars older than its published boundary.
 
     '''
     frame = np.zeros(
-        2,
+        3,
         dtype=np.dtype(def_iohlcv_fields),
     )
-    frame['index'] = [0, 1]
+    frame['index'] = [0, 1, 2]
-    frame['time'] = [60, 120]
+    frame['time'] = [60, 120, 180]
 
     events: list[str] = []
 
@@ -142,7 +151,7 @@ def test_storage_receives_full_provider_delta() -> None:
         return (
             frame,
             from_timestamp(60),
-            from_timestamp(120),
+            from_timestamp(180),
         )
 
     async def main() -> None:
@@ -154,7 +163,7 @@ def test_storage_receives_full_provider_delta() -> None:
                 mkt=mkt,
                 shm=shm,
                 timeframe=60,
-                backfill_from_shm_index=1,
+                backfill_from_shm_index=2,
                 backfill_from_dt=from_timestamp(180),
                 sampler_stream=Sampler(),
                 backfill_until_dt=from_timestamp(60),
@@ -163,8 +172,8 @@ def test_storage_receives_full_provider_delta() -> None:
             )
 
     trio.run(main)
-    assert shm.pushed[0]['time'].tolist() == [120]
+    assert shm.pushed[0]['time'].tolist() == [60, 120]
-    assert storage.frames[0]['time'].tolist() == [60, 120]
+    assert storage.frames[0]['time'].tolist() == [60, 120, 180]
     assert events == ['storage', 'shm', 'sampler']
 
 
@@ -344,3 +353,163 @@ def test_backfill_notification_timeout_is_bounded(
             )
 
     trio.run(main)
+
+
+def test_synthetic_gap_times_require_valid_right_boundary() -> None:
+    '''
+    Synthetic rows must remain strictly between valid boundary bars.
+
+    The live QQQ reservation held one more physical zero row than its
+    timestamp interval could represent. Blind fill would assign the
+    null the right boundary's timestamp, creating a duplicate. Prove
+    aligned bounds produce cadence while an overfull segment
+    is rejected instead of inventing non-monotonic timestamps.
+
+    '''
+    aligned = _synthetic_gap_times(
+        start_t=60,
+        right_t=300,
+        count=3,
+        timeframe=60,
+    )
+    overfull = _synthetic_gap_times(
+        start_t=60,
+        right_t=240,
+        count=3,
+        timeframe=60,
+    )
+    trailing = _synthetic_gap_times(
+        start_t=60,
+        right_t=None,
+        count=3,
+        timeframe=60,
+    )
+
+    assert aligned is not None
+    assert aligned.tolist() == [120, 180, 240]
+    assert overfull is None
+    assert trailing is None
+
+
+def test_null_repair_falls_back_without_debug_pause(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Empty provider repair must resolve interior zero rows locally.
+
+    QQQ startup exposed 653 zero-time rows between stored history and
+    an IB frame. The null repair queried IB while reverse backfill
+    occupied its sole history request slot, then could
+    enter an interactive debug pause before reaching forward-fill.
+    Arrange an interior null run, return an empty provider frame, and
+    wedge sampler notification. A short timeout and fail-fast pause
+    prove local fallback needs no interactive or IPC progress.
+    The assertions verify only the
+    zero rows inherit the preceding close and synthesized timestamps;
+    both valid boundary rows remain unchanged.
+
+    '''
+    monkeypatch.setattr(
+        'piker.tsp._history._notify_timeout_s',
+        0.01,
+    )
+
+    backing = np.zeros(
+        105,
+        dtype=np.dtype(def_iohlcv_fields),
+    )
+    backing['index'] = np.arange(105)
+    frame = backing[100:105]
+    frame['index'] = np.arange(100, 105)
+    frame['time'] = [60, 0, 0, 0, 300]
+    frame['open'] = [9, 0, 0, 0, 19]
+    frame['high'] = [11, 0, 0, 0, 21]
+    frame['low'] = [8, 0, 0, 0, 18]
+    frame['close'] = [10, 0, 0, 0, 20]
+    frame['volume'] = [5, 0, 0, 0, 6]
+
+    class Shm:
+        '''
+        Expose the test frame through the SHM repair interface.
+
+        '''
+        def __init__(self) -> None:
+            self._array = backing
+
+        @property
+        def array(self) -> np.ndarray:
+            '''
+            Return the readable SHM view.
+
+            '''
+            return self._array[100:105]
+
+    class Sampler:
+        '''
+        Model a backpressured UI notification stream.
+
+        '''
+        async def send(self, msg) -> None:
+            '''
+            Block until the bounded notifier cancels this send.
+
+            '''
+            await trio.sleep_forever()
+
+    async def get_hist(*args, **kwargs):
+        '''
+        Return no provider bars for the requested null segment.
+
+        '''
+        end_dt = kwargs['end_dt']
+        empty = np.empty(
+            0,
+            dtype=np.dtype(def_iohlcv_fields),
+        )
+        return empty, end_dt, end_dt
+
+    async def fail_pause() -> None:
+        '''
+        Reject interactive debugging in normal repair fallbacks.
+
+        '''
+        raise AssertionError('null repair entered tractor.pause()')
+
+    monkeypatch.setattr(tractor, 'pause', fail_pause)
+
+    async def main() -> None:
+        import greenback
+
+        async def ensure_portal() -> None:
+            '''
+            Avoid installing a greenback portal in this unit test.
+
+            '''
+
+        monkeypatch.setattr(
+            greenback,
+            'ensure_portal',
+            ensure_portal,
+        )
+        with trio.fail_after(0.1):
+            await maybe_fill_null_segments(
+                shm=Shm(),
+                timeframe=60,
+                get_hist=get_hist,
+                sampler_stream=Sampler(),
+                mkt=SimpleNamespace(fqme='qqq.nasdaq.ib'),
+                backfill_until_dt=from_timestamp(60),
+            )
+
+    trio.run(main)
+
+    assert frame['time'].tolist() == [60, 120, 180, 240, 300]
+    for field in ('open', 'high', 'low', 'close'):
+        assert frame[field].tolist() == [
+            frame[field][0],
+            10,
+            10,
+            10,
+            frame[field][-1],
+        ]
+    assert frame['volume'].tolist() == [5, 0, 0, 0, 6]

tests/test_ib_history.py

@@ -8,11 +8,19 @@ from datetime import (
 )
 from types import SimpleNamespace
 
-from pendulum import datetime as pdatetime
+from ib_async import RequestError
+import numpy as np
+from pendulum import (
+    datetime as pdatetime,
+    duration,
+)
 import pytest
 import trio
 
+from piker.brokers import DataUnavailable
+from piker.brokers.ib import feed as ib_feed
 from piker.brokers.ib.api import Client
+from piker.brokers.ib.feed import get_bars
 
 
 @pytest.mark.parametrize(
@@ -79,3 +87,502 @@ def test_history_end_datetime_uses_ib_utc_format(
 
     trio.run(main)
     assert request['endDateTime'] == expected
+
+
+def test_empty_history_frame_completes() -> None:
+    '''
+    A blank IB response must release the history request slot.
+
+    ``get_bars()`` previously returned from its query task without
+    signalling the parent, which then reset the feed forever. That
+    kept serialized SHM repair from running. Return IB's empty
+    response and prove the request completes explicitly
+    with no result instead of relying on a reset timeout.
+
+    '''
+    class Proxy:
+        '''
+        Return one empty IB history response.
+
+        '''
+        async def bars(self, **kwargs):
+            '''
+            Model ``Client.bars()`` during a venue data gap.
+
+            '''
+            return (
+                [],
+                np.empty(0),
+                duration(seconds=2000),
+            )
+
+    async def main() -> None:
+        with trio.fail_after(0.1):
+            result, timedout = await get_bars(
+                Proxy(),
+                'qqq.nasdaq.ib',
+                1,
+            )
+        assert result is None
+        assert not timedout
+
+    trio.run(main)
+
+
+def test_repeated_ib_cancellation_is_bounded() -> None:
+    '''
+    Repeated IB cancellation must not monopolize history forever.
+
+    IB can answer a historical request with error 162 cancellation.
+    The old retry loop had no limit, so reverse backfill could retain
+    the backend's sole request slot and prevent final null repair.
+    Always raise that response and prove the configured retry bound
+    terminates with ``DataUnavailable`` deterministically.
+
+    '''
+    class Proxy:
+        '''
+        Cancel every IB history request.
+
+        '''
+        async def bars(self, **kwargs):
+            '''
+            Raise IB's historical-query cancellation response.
+
+            '''
+            raise RequestError(
+                1,
+                162,
+                'API historical data query cancelled',
+            )
+
+    async def main() -> None:
+        with (
+            trio.fail_after(0.1),
+            pytest.raises(DataUnavailable),
+        ):
+            await get_bars(
+                Proxy(),
+                'qqq.nasdaq.ib',
+                1,
+                max_failed_resets=2,
+            )
+
+    trio.run(main)
+
+
+def test_query_error_cancels_stalled_reset(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Exceptional history exit must cancel its in-flight reset task.
+
+    A timed-out request acquires the reset lock before launching
+    IB's VNC reset task. If the query then raises while that task is
+    blocked, stale work must not retain the reset lock. Force that
+    ordering with explicit Trio checkpoints, then prove bounded query
+    failure cancels the reset task before propagating.
+
+    '''
+    reset_cancelled: list[bool] = []
+
+    class Proxy:
+        '''
+        Delay once, then return IB cancellation.
+
+        '''
+        async def bars(self, **kwargs):
+            '''
+            Let the parent acquire the reset lock before failing.
+
+            '''
+            await trio.sleep(0.02)
+            raise RequestError(
+                1,
+                162,
+                'API historical data query cancelled',
+            )
+
+    async def stalled_reset(
+        proxy,
+        reset_type='data',
+        task_status=trio.TASK_STATUS_IGNORED,
+    ) -> bool:
+        '''
+        Publish reset state, then block until sibling failure.
+
+        '''
+        done = trio.Event()
+        result: list[bool] = []
+        with trio.CancelScope() as cs:
+            task_status.started((cs, done, result))
+            try:
+                await trio.sleep_forever()
+            finally:
+                reset_cancelled.append(True)
+        return False
+
+    monkeypatch.setattr(
+        ib_feed,
+        'wait_on_data_reset',
+        stalled_reset,
+    )
+
+    async def main() -> None:
+        with (
+            trio.fail_after(0.1),
+            pytest.raises(DataUnavailable),
+        ):
+            await get_bars(
+                Proxy(),
+                'qqq.nasdaq.ib',
+                1,
+                feed_reset_timeout=0.001,
+                max_failed_resets=1,
+            )
+
+    trio.run(main)
+    assert reset_cancelled == [True]
+
+
+def test_failed_pacing_resets_reach_retry_limit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Failed pacing resets must increment the request's retry count.
+
+    The reset API returns an ``Event`` and result box. The old
+    code tested the always-truthy event and reset the failure counter
+    after each failure. Return immediate failures and prove two
+    pacing responses terminate instead of looping forever.
+
+    '''
+    reset_calls: list[bool] = []
+
+    class Proxy:
+        '''
+        Return a pacing violation for every request.
+
+        '''
+        _aio_ns = SimpleNamespace(
+            ib=SimpleNamespace(
+                client=SimpleNamespace(),
+            ),
+        )
+
+        async def bars(self, **kwargs):
+            '''
+            Raise IB's pacing response.
+
+            '''
+            raise RequestError(
+                1,
+                162,
+                ib_feed._pacing,
+            )
+
+    async def failed_reset(
+        proxy,
+        reset_type='data',
+        task_status=trio.TASK_STATUS_IGNORED,
+    ) -> bool:
+        '''
+        Publish one completed, unsuccessful reset.
+
+        '''
+        done = trio.Event()
+        result = [False]
+        reset_calls.append(True)
+        with trio.CancelScope() as cs:
+            task_status.started((cs, done, result))
+            done.set()
+        return False
+
+    monkeypatch.setattr(
+        ib_feed,
+        'wait_on_data_reset',
+        failed_reset,
+    )
+
+    async def main() -> None:
+        with (
+            trio.fail_after(0.1),
+            pytest.raises(DataUnavailable),
+        ):
+            await get_bars(
+                Proxy(),
+                'qqq.nasdaq.ib',
+                1,
+                max_failed_resets=2,
+            )
+
+    trio.run(main)
+    assert reset_calls == [True, True]
+
+
+def test_pacing_resets_are_serialized_across_requests(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Concurrent history requests must never reset IB farms together.
+
+    Pacing recovery bypassed reset ownership, allowing 1s and 60s
+    requests to launch the VNC hack concurrently. Start two pacing
+    failures in one nursery and hold each fake reset across a Trio
+    checkpoint. Active-count assertions prove the actor-global lock
+    serializes reset lifetimes without depending on scheduler timing.
+
+    '''
+    active: int = 0
+    max_active: int = 0
+    reset_calls: int = 0
+
+    class Proxy:
+        '''
+        Return a pacing violation for every request.
+
+        '''
+        _aio_ns = SimpleNamespace(
+            ib=SimpleNamespace(
+                client=SimpleNamespace(),
+            ),
+        )
+
+        async def bars(self, **kwargs):
+            '''
+            Raise IB's pacing response.
+
+            '''
+            raise RequestError(
+                1,
+                162,
+                ib_feed._pacing,
+            )
+
+    async def failed_reset(
+        proxy,
+        reset_type='data',
+        task_status=trio.TASK_STATUS_IGNORED,
+    ) -> bool:
+        '''
+        Hold one failed reset long enough to expose overlap.
+
+        '''
+        nonlocal active, max_active, reset_calls
+        done = trio.Event()
+        result: list[bool] = []
+        with trio.CancelScope() as cs:
+            task_status.started((cs, done, result))
+            active += 1
+            reset_calls += 1
+            max_active = max(max_active, active)
+            await trio.sleep(0.01)
+            active -= 1
+            result.append(False)
+            done.set()
+        return False
+
+    monkeypatch.setattr(
+        ib_feed,
+        'wait_on_data_reset',
+        failed_reset,
+    )
+
+    async def request() -> None:
+        '''
+        Run one independently bounded history request.
+
+        '''
+        with pytest.raises(DataUnavailable):
+            await get_bars(
+                Proxy(),
+                'qqq.nasdaq.ib',
+                1,
+                max_failed_resets=1,
+            )
+
+    async def main() -> None:
+        with trio.fail_after(0.1):
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(request)
+                nursery.start_soon(request)
+
+    trio.run(main)
+    assert reset_calls == 2
+    assert max_active == 1
+
+
+def test_pacing_supersedes_timeout_reset_once(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Pacing handoff must not count cancellation as another failure.
+
+    A slow request can trigger timeout recovery immediately before IB
+    returns a pacing violation. Query-side recovery then deliberately
+    cancels that data reset and waits for the same actor-global lock.
+    Reproduce this with blocked first and failed second resets. A
+    one-attempt limit proves cancellation releases the lock
+    without aborting before the intended connection reset runs.
+
+    '''
+    reset_types: list[str] = []
+
+    class Proxy:
+        '''
+        Return pacing only after timeout recovery starts.
+
+        '''
+        _aio_ns = SimpleNamespace(
+            ib=SimpleNamespace(
+                client=SimpleNamespace(),
+            ),
+        )
+
+        async def bars(self, **kwargs):
+            '''
+            Delay past the reset timeout, then report pacing.
+
+            '''
+            await trio.sleep(0.02)
+            raise RequestError(
+                1,
+                162,
+                ib_feed._pacing,
+            )
+
+    async def reset(
+        proxy,
+        reset_type='data',
+        task_status=trio.TASK_STATUS_IGNORED,
+    ) -> bool:
+        '''
+        Block data reset and fail connection reset immediately.
+
+        '''
+        done = trio.Event()
+        result: list[bool] = []
+        reset_types.append(reset_type)
+        with trio.CancelScope() as cs:
+            task_status.started((cs, done, result))
+            if reset_type == 'data':
+                await trio.sleep_forever()
+
+        result.append(False)
+        done.set()
+        return False
+
+    monkeypatch.setattr(
+        ib_feed,
+        'wait_on_data_reset',
+        reset,
+    )
+
+    async def main() -> None:
+        with (
+            trio.fail_after(0.1),
+            pytest.raises(DataUnavailable),
+        ):
+            await get_bars(
+                Proxy(),
+                'qqq.nasdaq.ib',
+                1,
+                feed_reset_timeout=0.001,
+                max_failed_resets=1,
+            )
+
+    trio.run(main)
+    assert reset_types == ['data', 'connection']
+
+
+def test_completed_pacing_reset_restarts_timeout_wait(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A queued timeout waiter must consume completed pacing recovery.
+
+    While query-side pacing recovery holds the reset lock, the outer
+    timeout task can already queue behind it. Without a completion
+    handoff, that waiter launches a redundant data reset immediately.
+    Return a blank frame just after a successful connection reset and
+    prove no third reset runs before the restarted timeout observes
+    completion.
+
+    '''
+    reset_types: list[str] = []
+
+    class Proxy:
+        '''
+        Pace the first query and return blank data on retry.
+
+        '''
+        _aio_ns = SimpleNamespace(
+            ib=SimpleNamespace(
+                client=SimpleNamespace(),
+            ),
+        )
+
+        def __init__(self) -> None:
+            self.calls: int = 0
+
+        async def bars(self, **kwargs):
+            '''
+            Control the pacing-to-success handoff ordering.
+
+            '''
+            self.calls += 1
+            if self.calls == 1:
+                await trio.sleep(0.02)
+                raise RequestError(
+                    1,
+                    162,
+                    ib_feed._pacing,
+                )
+
+            await trio.sleep(0.0005)
+            return (
+                [],
+                np.empty(0),
+                duration(seconds=2000),
+            )
+
+    async def reset(
+        proxy,
+        reset_type='data',
+        task_status=trio.TASK_STATUS_IGNORED,
+    ) -> bool:
+        '''
+        Cancel timeout recovery and complete pacing recovery.
+
+        '''
+        done = trio.Event()
+        result: list[bool] = []
+        reset_types.append(reset_type)
+        with trio.CancelScope() as cs:
+            task_status.started((cs, done, result))
+            if reset_type == 'data':
+                await trio.sleep_forever()
+
+        result.append(reset_type == 'connection')
+        done.set()
+        return result[0]
+
+    monkeypatch.setattr(
+        ib_feed,
+        'wait_on_data_reset',
+        reset,
+    )
+
+    async def main() -> None:
+        with trio.fail_after(0.1):
+            result, _ = await get_bars(
+                Proxy(),
+                'qqq.nasdaq.ib',
+                1,
+                feed_reset_timeout=0.001,
+                max_failed_resets=2,
+            )
+        assert result is None
+
+    trio.run(main)
+    assert reset_types == ['data', 'connection']

tests/test_ldshm.py

@@ -0,0 +1,235 @@
+'''
+Shared-memory persistence regressions.
+
+'''
+from collections.abc import (
+    AsyncIterator,
+    Iterator,
+)
+from contextlib import asynccontextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import numpy as np
+import polars as pl
+import pytest
+
+from piker import tsp
+from piker.storage import cli as storage_cli
+from piker.storage.cli import (
+    _shm_period_and_invalid_count,
+    ldshm,
+)
+
+
+def test_ldshm_detects_nulls_without_skewing_period() -> None:
+    '''
+    Null SHM slots must stop persistence without skewing cadence.
+
+    A live QQQ buffer contained an interior run of zero-time rows.
+    ``ldshm`` included them in period inference and then sent them to
+    NativeDB, which correctly rejected the replacement. Build a
+    60-second series with the same interior hole and enough
+    bars around the gap. Prove inspection counts every null while
+    inferring the published series' 60-second period; the
+    command uses that count to skip both persistence and SHM reload.
+
+    '''
+    times = np.array([
+        60,
+        120,
+        0,
+        0,
+        300,
+        360,
+        420,
+    ])
+
+    period_s, invalid_count = _shm_period_and_invalid_count(times)
+
+    assert period_s == 60
+    assert invalid_count == 2
+
+
+def test_ldshm_period_is_an_observed_step() -> None:
+    '''
+    Sparse timestamps must not invent a storage timeframe.
+
+    The arithmetic median of two steps can produce a cadence that
+    occurs nowhere in the source series. Such a value can select the
+    wrong NativeDB timeframe and hide the gap. Arrange tied 60 and
+    120-second steps and prove the deterministic tie-break chooses
+    the smaller observed period.
+
+    '''
+    times = np.array([60, 120, 240])
+
+    period_s, invalid_count = _shm_period_and_invalid_count(times)
+
+    assert period_s == 60
+    assert invalid_count == 0
+
+
+def test_ldshm_counts_every_invalid_timestamp() -> None:
+    '''
+    Corrupt timestamps stop persistence just like null SHM slots.
+
+    Exact zero denotes an unpublished slot, while negative, NaN, and
+    infinite values indicate corruption. NativeDB rejects them all,
+    but allowing a replacement attempt would still abort ``ldshm``.
+    Mix each invalid class into a regular series and prove the same
+    snapshot inspection counts all four while preserving cadence.
+
+    '''
+    times = np.array([
+        60,
+        120,
+        0,
+        -1,
+        np.nan,
+        np.inf,
+        180,
+    ])
+
+    period_s, invalid_count = _shm_period_and_invalid_count(times)
+
+    assert period_s == 60
+    assert invalid_count == 4
+
+
+def test_ldshm_invalid_snapshot_never_reaches_storage(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Invalid snapshots stop before analyzers and storage mutation.
+
+    The original QQQ failure passed zero-time rows through dedupe and
+    into NativeDB. An intermediate fix called null-segment analysis
+    before its guard, allowing corrupt indexes to abort the command.
+    Supply one immutable Polars snapshot containing that combination,
+    then replace dedupe and writes with failing spies. Completing the
+    command proves its guard runs first, preventing persistence or
+    SHM reload without relying on task timing.
+
+    '''
+    shm_df = pl.DataFrame({
+        'index': [100, 500, 104],
+        'time': [60, 0, 180],
+    })
+
+    @asynccontextmanager
+    async def open_runtime(
+        *args: Any,
+        **kwargs: Any,
+    ) -> AsyncIterator[None]:
+        '''
+        Yield a runtime-free command context.
+
+        '''
+        yield
+
+    async def fail_write(
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        '''
+        Fail if invalid SHM reaches NativeDB.
+
+        '''
+        raise AssertionError('invalid SHM reached NativeDB')
+
+    client = SimpleNamespace(write_ohlcv=fail_write)
+
+    @asynccontextmanager
+    async def open_storage(
+        *args: Any,
+        **kwargs: Any,
+    ) -> AsyncIterator[tuple[SimpleNamespace, SimpleNamespace]]:
+        '''
+        Yield storage objects with a failing writer.
+
+        '''
+        yield SimpleNamespace(), client
+
+    @asynccontextmanager
+    async def open_annotations(
+        *args: Any,
+        **kwargs: Any,
+    ) -> AsyncIterator[SimpleNamespace]:
+        '''
+        Yield an inert annotation controller.
+
+        '''
+        yield SimpleNamespace()
+
+    def iter_shms(
+        fqme: str,
+    ) -> Iterator[tuple[Path, SimpleNamespace, pl.DataFrame]]:
+        '''
+        Yield the single captured invalid snapshot.
+
+        '''
+        yield Path('qqq.rt'), SimpleNamespace(), shm_df
+
+    def dedupe(*args: Any, **kwargs: Any) -> None:
+        '''
+        Fail if invalid SHM reaches deduplication.
+
+        '''
+        raise AssertionError('invalid SHM reached dedupe')
+
+    async def pause() -> None:
+        '''
+        Replace the command's final interactive pause.
+
+        '''
+
+    monkeypatch.setattr(
+        storage_cli,
+        'open_piker_runtime',
+        open_runtime,
+    )
+    monkeypatch.setattr(
+        storage_cli,
+        'open_storage_client',
+        open_storage,
+    )
+    monkeypatch.setattr(tsp, 'iter_dfs_from_shms', iter_shms)
+    monkeypatch.setattr(tsp, 'dedupe_ohlcv_smart', dedupe)
+    monkeypatch.setattr(storage_cli.tractor, 'pause', pause)
+
+    from piker.ui import _remote_ctl
+    monkeypatch.setattr(
+        _remote_ctl,
+        'open_annot_ctl',
+        open_annotations,
+    )
+
+    ldshm('qqq.nasdaq.ib')
+
+
+@pytest.mark.parametrize(
+    'times',
+    [
+        np.array([0, 0]),
+        np.array([60]),
+        np.array([60, 60]),
+    ],
+)
+def test_ldshm_skips_frame_without_positive_cadence(
+    times: np.ndarray,
+) -> None:
+    '''
+    Degenerate buffers are skipped before indexing or persistence.
+
+    The old inference indexed final timestamps and left its median
+    undefined for short buffers. Fully unpublished, one-row, and
+    duplicate-only frames could therefore fail before the command's
+    guard. Exercise each shape and prove inspection
+    returns no cadence, which directs ``ldshm`` to skip the buffer.
+
+    '''
+    period_s, _ = _shm_period_and_invalid_count(times)
+
+    assert period_s is None

tests/test_storage_audit.py

@@ -129,6 +129,69 @@ def test_audit_preserves_raw_defect_evidence() -> None:
     assert report['values']['infinity_by_column']['volume'] == 1
 
 
+def test_audit_rejects_epoch_zero_price_sentinel() -> None:
+    '''
+    The legacy first-slot sentinel is structural corruption.
+
+    The repaired MNQ Parquet retained a row at epoch 60 whose OHLCV
+    payload was zero. Numeric and finite checks marked it valid even
+    though chart auto-ranging then included zero. Audit a canonical
+    frame with that sentinel and prove the report counts it and fails
+    structural qualification.
+
+    '''
+    frame = mk_frame((60, 120)).with_columns(
+        pl.Series('open', [0, 1], dtype=pl.Float64),
+        pl.Series('high', [0, 2], dtype=pl.Float64),
+        pl.Series('low', [0, 1], dtype=pl.Float64),
+        pl.Series('close', [0, 2], dtype=pl.Float64),
+    )
+
+    report: dict = audit_ohlcv_frame(
+        frame,
+        fqme='x.test',
+        period_s=60,
+    )
+
+    assert report['values']['all_zero_price_rows'] == 1
+    assert report['values']['epoch_sentinel_rows'] == 1
+    assert 'epoch_ohlc_sentinel' in report['result']['violations']
+    assert report['result']['structural_ok'] is False
+
+
+def test_audit_leaves_modern_zero_prices_unclassified() -> None:
+    '''
+    Audit must not call every all-zero price bar corrupt.
+
+    A spread or synthetic instrument may legitimately trade at zero.
+    The known corruption is tied to the first slot. Place an all-zero
+    bar later and prove it remains visible as a warning without
+    failing structural qualification.
+
+    '''
+    frame = mk_frame((120,)).with_columns(
+        pl.Series('open', [0], dtype=pl.Float64),
+        pl.Series('high', [0], dtype=pl.Float64),
+        pl.Series('low', [0], dtype=pl.Float64),
+        pl.Series('close', [0], dtype=pl.Float64),
+    )
+
+    report: dict = audit_ohlcv_frame(
+        frame,
+        fqme='x.test',
+        period_s=60,
+    )
+
+    assert report['values']['all_zero_price_rows'] == 1
+    assert report['values']['epoch_sentinel_rows'] == 0
+    assert report['values']['unclassified_zero_price_rows'] == 1
+    assert report['result']['structural_ok'] is True
+    assert (
+        'all_zero_ohlc_prices_unclassified'
+        in report['result']['warnings']
+    )
+
+
 def test_gap_aggregates_are_not_truncated_with_details() -> None:
     '''
     Limiting JSON detail must not undercount total missing coverage.

tests/test_storage_nativedb.py

@@ -131,47 +131,162 @@ def test_update_preserves_history_and_resolves_conflicts(
     assert stored['index'].to_list() == [0, 1, 2, 3]
 
 
-def test_update_repairs_nonpositive_persisted_timestamps(
+def test_update_repairs_invalid_persisted_sentinel_rows(
     tmp_path: Path,
 ) -> None:
     '''
-    Valid incoming history must repair a legacy epoch-zero row.
+    Valid incoming history must repair legacy sentinel rows.
 
-    The known MNQ baseline contains one zero timestamp plus extra derived
+    The known MNQ baseline contained epoch-zero and all-zero rows,
-    columns and noncanonical absolute indexes. ``update_ohlcv()`` used to
+    plus extra columns and absolute indexes. ``update_ohlcv()`` first
-    canonicalize those bytes but retain the zero row, then reject the
+    retained both, then removed only epoch zero and left chart range
-    whole merged frame before fresh IB bars could publish. Write that
+    anchored at zero. Write both sentinels, append a valid provider
-    legacy shape directly, append a valid provider frame, and prove only
+    frame, and prove only valid bars survive with canonical
-    the invalid persisted row disappears while old and new valid bars
+    indexes.
-    survive with canonical indexes.
 
     '''
     client = NativeStorageClient(tmp_path)
     legacy = (
         tsp.np2pl(mk_ohlcv(
-            (0, 60, 120),
+            (0, 60, 120, 180),
-            (0, 1, 2),
+            (0, 0, 2, 3),
         ))
         .with_columns(
-            pl.Series('index', [3137800, 3137801, 3137802]),
+            pl.Series(
-            pl.Series('time_prev', [None, 0, 60]),
+                'index',
+                [3137800, 3137801, 3137802, 3137803],
+            ),
+            pl.Series('time_prev', [None, 0, 60, 120]),
         )
     )
     path: Path = client.mk_path('x.test', 60)
     legacy.write_parquet(path)
 
+    loaded = trio.run(client.read_ohlcv, 'x.test', 60)
+    assert loaded['time'].tolist() == [120, 180]
+    assert loaded['index'].tolist() == [0, 1]
+    assert pl.read_parquet(path).height == 4
+
     run(client.update_ohlcv(
         'x.test',
-        mk_ohlcv((180,), (3,)),
+        mk_ohlcv((240,), (4,)),
         60,
     ))
     stored: pl.DataFrame = pl.read_parquet(path)
 
-    assert stored['time'].to_list() == [60, 120, 180]
+    assert stored['time'].to_list() == [120, 180, 240]
-    assert stored['close'].to_list() == [1, 2, 3]
+    assert stored['close'].to_list() == [2, 3, 4]
     assert stored['index'].to_list() == [0, 1, 2]
 
 
+def test_write_preserves_modern_all_zero_prices(
+    tmp_path: Path,
+) -> None:
+    '''
+    Zero prices outside the legacy sentinel range remain valid.
+
+    Some spreads or synthetic instruments may legitimately trade at
+    zero. Repair targets the observed first-slot sentinel, not every
+    all-zero OHLC bar. Write a modern all-zero row and
+    prove it survives normal replacement unchanged.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    zero_bar = mk_ohlcv((1_800_000_000,), (0,))
+
+    run(client.write_ohlcv('x.test', zero_bar, 60))
+    stored = pl.read_parquet(client.mk_path('x.test', 60))
+
+    assert stored['time'].to_list() == [1_800_000_000]
+    assert stored['close'].to_list() == [0]
+
+
+def test_write_rejects_first_slot_zero_sentinel(
+    tmp_path: Path,
+) -> None:
+    '''
+    New writes must not recreate the repaired legacy sentinel.
+
+    Read repair removes all-zero OHLC first-slot rows. Accepting one from
+    new input would recreate a file whose audit fails and whose read view
+    hides a row. Submit the exact 60-second
+    sentinel and prove no file publishes.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    sentinel = mk_ohlcv((60,), (0,))
+
+    with pytest.raises(ValueError, match='epoch sentinel'):
+        run(client.write_ohlcv('x.test', sentinel, 60))
+
+    assert not client.mk_path('x.test', 60).exists()
+
+
+def test_sentinel_only_storage_loads_as_no_history(
+    tmp_path: Path,
+) -> None:
+    '''
+    A fully repairable legacy file must permit fresh startup.
+
+    Filtering a file containing only sentinels yields an empty array.
+    ``load()`` previously indexed its endpoint timestamps and crashed
+    before valid data could replace it. Persist only sentinel rows,
+    prove load reports no history, then append a valid bar and
+    verify durable storage becomes canonical.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    path: Path = client.mk_path('x.test', 60)
+    tsp.np2pl(mk_ohlcv(
+        (0, 60),
+        (0, 0),
+    )).write_parquet(path)
+
+    assert trio.run(client.load, 'x.test', 60) is None
+    run(client.update_ohlcv(
+        'x.test',
+        mk_ohlcv((120,), (1,)),
+        60,
+    ))
+    stored: pl.DataFrame = pl.read_parquet(path)
+
+    assert stored['time'].to_list() == [120]
+    assert stored['index'].to_list() == [0]
+
+
+def test_merge_does_not_hide_null_price_rows(
+    tmp_path: Path,
+) -> None:
+    '''
+    Sentinel repair must not silently discard other malformed rows.
+
+    Polars boolean filters drop null predicates. Without an explicit
+    false fill, one null plus three zeros looked neither valid nor
+    all-zero but disappeared before validation. Persist that shape,
+    append valid data, and prove merge still rejects the null
+    while preserving the original evidence file.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    path: Path = client.mk_path('x.test', 60)
+    malformed = tsp.np2pl(
+        mk_ohlcv((1_800_000_000,), (0,))
+    ).with_columns(
+        pl.Series('open', [None], dtype=pl.Float64)
+    )
+    malformed.write_parquet(path)
+    before: bytes = path.read_bytes()
+
+    with pytest.raises(ValueError, match='finite numeric'):
+        run(client.update_ohlcv(
+            'x.test',
+            mk_ohlcv((1_800_000_060,), (1,)),
+            60,
+        ))
+
+    assert path.read_bytes() == before
+
+
 def test_write_ohlcv_remains_an_explicit_replacement(
     tmp_path: Path,
 ) -> None: