Guard `ldshm` from invalid SHM snapshots

Do not pass unpublished or corrupt timestamp rows into dedupe,
`NativeDB` replacement writes, or live SHM reload.

Deats,
- infer cadence from finite positive observed steps
- choose the smaller observed step when frequency ties
- skip invalid or degenerate snapshots as a whole
- cover the QQQ zero-slot failure with mutation-failing spies

Prompt-IO: ai/prompt-io/opencode/20260729T185406Z_d359b1cb_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,7 @@ Deterministic or local first-pass targets:
 - `tests/test_storage_audit.py`
 - `tests/test_backfill_audit_snippet.py`
 - `tests/test_ib_history.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 +191,7 @@ 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_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
@@ -206,6 +208,7 @@ tests/
 | `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/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/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.

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.
-                    ):
-                        raise ValueError(
-                            f'Something is wrong with time period for {shm}:\n{times}'
                     )
-                period_s: float = float(max(d1, d2, med))
+                    continue
                 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

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