Fix IB `MethodProxy` response routing

`open_client_proxy()` broadcast its one-slot linked channel to an always-running status relay and an idle method receiver. Status traffic advanced only the relay, so the first lazy contract lookup for an uncached order raised `Lagged` before receiving its response. Give `relay_client_proxy_messages()` sole receive ownership. Route tagged responses from concurrent caller tasks into their per-`mid` slots while each proxy context retains its own channel and relay. Deats, - log stale responses with their original method name; - warn when unresolved-request metadata exceeds its bounded limit; - drop late responses after caller cancellation; - preserve the reconnect design note and message-case rationale; - accept clean channel EOF during proxy teardown; - cover idle status, concurrency, cancellation and EOF paths. Prompt-IO: ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
Fix sparse IB history prepends in `ShmArray`
2026-08-11 18:46:23 -04:00 · 2026-08-05 19:40:07 -04:00 · 2026-08-01 00:34:55 -04:00 · 2026-07-31 17:26:11 -04:00
23 changed files with 2838 additions and 304 deletions
--- a/.claude/skills/piker-slang/examples.md
+++ b/.claude/skills/piker-slang/examples.md
@ -47,7 +47,7 @@ that's why it's like 1000x faster ya know?"
 - Wrap all code symbols: `function()`,
  `ClassName`, `field_name`
 - File paths: `piker/ui/_remote_ctl.py`
- Commands: `git status`, `piker store ldshm`
+- Commands: `git status`, `piker store shm`
 **Explain like you're pair programming:**
 ```
--- a/.claude/skills/run-tests/test-harness-reference.md
+++ b/.claude/skills/run-tests/test-harness-reference.md
@ -105,8 +105,10 @@ Deterministic or local first-pass targets:
 - `tests/test_watchlists.py`
 - `tests/test_storage_audit.py`
 - `tests/test_store_cli.py`
 - `tests/test_backfill_audit_snippet.py`
 - `tests/test_ib_history.py`
 - `tests/test_ib_method_proxy.py`
 - `tests/test_history_backfill.py`
 - `tests/test_ldshm.py`
 - `tests/test_accounting.py::test_account_file_default_empty`
@ -133,8 +135,8 @@ Require explicit authorization before running:
 `tests/test_cli.py` is currently hard-skipped. It is not an active CLI
 regression gate.
-Never execute `piker store anal` or `piker store ldshm` as tests. They are
+Never execute `piker store anal` or `piker store shm --write-parquet`
-mutating or interactive operational commands.
+as tests. They are mutating or interactive operational commands.
 ## Project-Specific Flags And Backend Matrix
@ -192,10 +194,12 @@ 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_ib_method_proxy.py   deterministic IB asyncio proxy routing
  test_history_backfill.py  deterministic history/SHM orchestration
-  test_ldshm.py             ldshm unpublished-slot guard
+  test_ldshm.py             SHM unpublished-slot guard
  test_questrade.py         obsolete credentialed tests; skipped
  test_services.py          pikerd/datad/feed/EMS actor lifecycle
  test_store_cli.py         storage command help and diagnostics UX
  test_storage_audit.py     read-only NativeDB audit and JSON CLI
  test_backfill_audit_snippet.py
                            disposable xonsh qualification helpers
@ -208,10 +212,12 @@ 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 |
 | `piker/storage/cli.py` command UX | `tests/test_store_cli.py` | fake SHM/runtime, no mutation |
 | `snippets/nativedb_backfill_audit.xsh` | `tests/test_backfill_audit_snippet.py` | disposable paths only |
 | `piker/brokers/ib/api.py`, `feed.py` history | `tests/test_ib_history.py` | fake client, no network |
 | `piker/brokers/ib/api.py` method proxy | `tests/test_ib_method_proxy.py` | fake channel, 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/storage/cli.py` SHM 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 |
--- a/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.md
+++ b/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.md
@ -0,0 +1,41 @@
 ---
 model: openai/gpt-5.6-sol
 service: opencode
 session: ses_0799212ebffe42arY96czXn89F
 timestamp: 2026-07-31T16:52:15Z
 git_ref: 0846cbd4
 scope: code
 substantive: true
 raw_file: 20260731T165215Z_0846cbd4_prompt_io.raw.md
 ---
 ## Prompt
 Evaluate and update `piker store` so bare groups and endpoints immediately
 show help; prepare for eventual optional fuzzy selection in `anal` and
 `ldshm`; aggregate recent time-series diagnostics into CLI endpoints. The
 user selected the safe-foundation scope.
 ## Response summary
 Added immediate help UX, exact NativeDB and SHM identities, read-only
 `series` and `shm` diagnostics, selector-ready exact SHM names, and robust
 discovery/attachment. Fixed related `ldshm` option and no-match failures.
 Deferred optional `fzf` and mutation-default changes.
 ## Files changed
 - `piker/storage/cli.py` - help UX and read-only series/SHM diagnostics.
 - `piker/storage/nativedb.py` - canonical compound series identities.
 - `piker/tsp/_history.py` - exact SHM discovery and dtype resolution.
 - `piker/data/_sharedmem.py` - attach-only existing-segment helper.
 - `piker/tsp/__init__.py` - SHM identity helper exports.
 - `tests/test_store_cli.py` - command UX and endpoint regressions.
 - `tests/test_ldshm.py` - SHM diagnostics and operational regressions.
 - `tests/test_storage_nativedb.py` - series parsing/index regressions.
 - `docs/manual_backfill_qualification.rst` - triage command documentation.
 - `.claude/skills/run-tests/test-harness-reference.md` - test mapping.
 ## Human edits
 None - generated output remains uncommitted.
--- a/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.raw.md
+++ b/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.raw.md
@ -0,0 +1,93 @@
 ---
 model: openai/gpt-5.6-sol
 service: opencode
 timestamp: 2026-07-31T16:52:15Z
 git_ref: 0846cbd4
 diff_cmd: git diff HEAD~1..HEAD
 ---
 The user requested an evaluation and possible update of `piker store`,
 including immediate help output when a group or endpoint is invoked with
 no arguments. The user also proposed eventual optional `fzf` selection for
 `anal` and `ldshm`, and authorized aggregating the recent time-series and
 SHM debugging work into CLI endpoints. After reviewing three scopes, the
 user selected the recommended safe foundation rather than implementing
 interactive fuzzy selection or changing mutation defaults immediately.
 > `git diff HEAD~1..HEAD -- piker/storage/cli.py`
 Generated no-argument help behavior for the `store` group and every
 endpoint. Added read-only `series` and `shm` commands, JSON and Rich output,
 exact `--shm-name` plumbing, SHM cadence/invalid/order/gap summaries, and
 safe corrupt-epoch formatting. Fixed existing `ldshm` no-write/no-reload
 control flow and no-match debugger behavior.
 > `git diff HEAD~1..HEAD -- piker/storage/nativedb.py`
 Generated canonical `NativeSeriesRef` parsing and deterministic listing.
 Changed the internal index to preserve `(fqme, period)` identities while
 keeping unique-FQME compatibility listing and cardinality. Hardened
 malformed, sidecar, Unicode-period, symlink, and traversal handling.
 > `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
 Generated exact `ShmBufferRef` parsing and stable discovery, replacing
 unsafe substring globbing. Added dtype resolution that mirrors history
 allocation, exact-ref attachment, disappearance handling, and optional
 exact-name iteration.
 > `git diff HEAD~1..HEAD -- piker/data/_sharedmem.py`
 Generated attach-only SHM access which never creates a segment when a
 discovered object disappears.
 > `git diff HEAD~1..HEAD -- piker/tsp/__init__.py`
 Exported selector-ready SHM identity, parsing, listing, and dtype helpers.
 > `git diff HEAD~1..HEAD -- tests/test_store_cli.py`
 Generated command-level help, exact durable-series JSON, read-only SHM JSON,
 and unknown exact-name regressions.
 > `git diff HEAD~1..HEAD -- tests/test_ldshm.py`
 Generated SHM identity, summary, ordering, dtype, forged-name, disabled
 mutation-flag, and gap-markup regressions.
 > `git diff HEAD~1..HEAD -- tests/test_storage_nativedb.py`
 Generated compound index, canonical parser, malformed filename, and dotted
 FQME regressions.
 > `git diff HEAD~1..HEAD -- docs/manual_backfill_qualification.rst`
 Documented `store series` and `store shm` as separate read-only durable and
 live-memory triage layers.
 > `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
 Registered the deterministic storage CLI UX target and mapping.
 Verification output:
 ```text
 ........................................................................ [ 71%]
 .............................                                            [100%]
 101 passed in 1.72s
 ```
 Real `piker store` and `piker store audit` invocations printed complete
 help without `--help`, exiting with Click's normal no-args help status 2.
 `git diff --check` passed. Ruff was unavailable in the existing worktree
 environment.
 Adversarial review drove fixes for provider dtype reconstruction, observed
 versus expected cadence, exact SHM grammar, discovery races, Typer direct
 defaults, malformed NativeDB periods, compound cardinality compatibility,
 undefined `ldshm` reload payloads, no-match debugger hangs, ordering
 corruption reporting, and dtype precedence. Final review found no issues.
 Optional external `fzf` or built-in numbered selection remains deferred.
 The exact series and SHM identities added here are the intended foundation
 for that later patch.
--- a/ai/prompt-io/opencode/20260801T015321Z_74ea2152_prompt_io.md
+++ b/ai/prompt-io/opencode/20260801T015321Z_74ea2152_prompt_io.md
@ -0,0 +1,39 @@
 ---
 model: openai/gpt-5.6-sol
 service: opencode
 session: ses_0799212ebffe42arY96czXn89F
 timestamp: 2026-08-01T01:53:21Z
 git_ref: 74ea2152
 scope: code
 substantive: true
 raw_file: 20260801T015321Z_74ea2152_prompt_io.raw.md
 ---
 ## Prompt
 Make `piker store ls` default to all NativeDB series as a compact counterpart
 to detailed `series`, with sensible no-flag output and a visible help
 description. Consolidate SHM inspection and repair under canonical `shm`,
 because attachment to volatile segments is implied, and make inspection the
 safe default while retaining explicit persistence and reload controls.
 ## Response summary
 Replaced backend-oriented `ls` with compact NativeDB discovery and made both
 listing commands default to all series. Consolidated SHM behavior under a
 read-only-by-default `shm` command, guarded explicit repair options, updated
 executable references and docs, and added command regressions.
 ## Files changed
 - `piker/storage/cli.py` - listing defaults and unified SHM command modes.
 - `tests/test_store_cli.py` - listing, help, and SHM mode regressions.
 - `tests/test_ldshm.py` - explicit persistence-path regressions.
 - `docs/manual_backfill_qualification.rst` - revised safe CLI workflow.
 - `snippets/claude_debug_helper.py` - explicit SHM persistence commands.
 - `.claude/skills/run-tests/test-harness-reference.md` - safety and test map.
 - `.claude/skills/piker-slang/examples.md` - canonical command example.
 ## Human edits
 None - generated output incorporates the user's design refinements.
--- a/ai/prompt-io/opencode/20260801T015321Z_74ea2152_prompt_io.raw.md
+++ b/ai/prompt-io/opencode/20260801T015321Z_74ea2152_prompt_io.raw.md
@ -0,0 +1,78 @@
 ---
 model: openai/gpt-5.6-sol
 service: opencode
 timestamp: 2026-08-01T01:53:21Z
 git_ref: 74ea2152
 diff_cmd: git diff HEAD~1..HEAD
 ---
 The user requested command-surface refinements after reviewing the new
 storage CLI. Bare `ls` should default to all series and act as a compact
 form of detailed `series`; commands should produce sensible default output
 without requiring trigger flags; and `ls` needed a parent-help description.
 The user first proposed folding read-only SHM inspection into `ldshm -r`,
 then clarified that `shm` is the better canonical command because attaching
 to volatile segments is implied by the resource name.
 The resulting design uses `piker store ls` as a compact NativeDB FQME and
 period view, while `piker store series` reports each exact sub-series with
 size and path details. Both default to all records and accept an optional
 FQME substring.
 `piker store shm FQME` is read-only by default. Persistence and reload enter
 the prior repair workflow only through explicit `--write-parquet` and
 `--reload-parquet-to-shm` flags. Reload requires persistence, and read-only
 formatting flags are rejected in repair mode instead of being ignored.
 > `git diff HEAD~1..HEAD -- piker/storage/cli.py`
 Generated shared NativeDB payload collection, compact and detailed output,
 numeric period ordering, default-all semantics, the canonical `shm` command,
 safe mode dispatch, explicit repair validation, and debugger propagation in
 read-only mode.
 > `git diff HEAD~1..HEAD -- tests/test_store_cli.py`
 Generated regressions for no-flag listing, parent-help descriptions, numeric
 period display, default read-only SHM JSON, reload dependencies, and repair
 versus read-only option conflicts.
 > `git diff HEAD~1..HEAD -- tests/test_ldshm.py`
 Adjusted persistence regressions for the renamed command and explicit repair
 entry while preserving the historical failure rationale.
 > `git diff HEAD~1..HEAD -- docs/manual_backfill_qualification.rst`
 Documented compact versus detailed NativeDB discovery, default read-only SHM
 inspection, and the explicitly mutating repair invocation.
 > `git diff HEAD~1..HEAD -- snippets/claude_debug_helper.py`
 Updated executable debug helper commands to select explicit SHM persistence.
 > `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
 Updated operational safety guidance and SHM regression mappings.
 > `git diff HEAD~1..HEAD -- .claude/skills/piker-slang/examples.md`
 Updated the command example to the canonical `shm` spelling.
 Verification output:
 ```text
 ........................................................................ [ 69%]
 ................................                                         [100%]
 104 passed in 5.77s
 ```
 Real command-help rendering was checked for the parent group, `ls`, `series`,
 and `shm`. Real read-only `ls` and `series --json` output was also exercised
 against the configured NativeDB. `git diff --check` passed.
 Adversarial review identified and drove fixes for lexicographic period
 ordering, stale runbook safety wording, ignored mode-specific options, and an
 explicit-default `--max-gaps 10` validation bypass. Final review found no
 remaining issues. Real OS-backed SHM attachment and successful repair/reload
 remain integration-level validation concerns.
--- a/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.md
+++ b/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.md
@ -0,0 +1,33 @@
 ---
 model: openai/gpt-5.6-sol
 service: opencode
 session: ses_0799212ebffe42arY96czXn89F
 timestamp: 2026-08-05T19:50:01Z
 git_ref: f0f3c806
 scope: code
 substantive: true
 raw_file: 20260805T195001Z_f0f3c806_prompt_io.raw.md
 ---
 ## Prompt
 Inspect the completed NVDA 60s SHM backfill before workspace restart, explain
 the persistent zero region, and continue fixing the indexing logic so sparse
 IB history does not leave physical null reservations.
 ## Response summary
 Captured a 42,720-row live SHM null span whose valid margins shared the same
 timestamp. Replaced wall-clock row reservation with actual provider-boundary
 placement, fixed stale SHM first-index, endpoint insertion and field-map
 semantics, published the final storage prepend, and added a full multi-frame
 `tsdb_backfill()` regression using a real `ShmArray`.
 ## Files changed
 - `piker/tsp/_history.py` - contiguous provider and NativeDB prepends.
 - `tests/test_history_backfill.py` - sparse multi-frame SHM regression.
 ## Human edits
 None - generated output incorporates the user's live observations.
--- a/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.raw.md
+++ b/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.raw.md
@ -0,0 +1,71 @@
 ---
 model: openai/gpt-5.6-sol
 service: opencode
 timestamp: 2026-08-05T19:50:01Z
 git_ref: f0f3c806
 diff_cmd: git diff HEAD~1..HEAD
 ---
 The user reported that a completed NVDA 60s backfill left a large zero
 region in live SHM and asked for immediate inspection before restarting the
 workspace. The captured buffer contained 42,720 invalid rows at absolute
 indexes `3065895..3108614`. The valid rows on both sides had the identical
 timestamp `1781271360`, proving the region was an erroneous physical-index
 reservation rather than a real temporal gap.
 The user then asked the agent to continue fixing the backfill logic.
 > `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
 Removed wall-clock-based SHM reservation from `tsdb_backfill()`. Reverse
 provider frames now extend the visible `ShmArray._first` boundary on every
 prepend, beginning exactly at the current first index after inclusive query
 endpoints are removed. Added `_prepend_tsdb_history()` to trim storage rows
 against the actual earliest provider timestamp and prepend them directly
 adjacent only after reverse provider retrieval completes.
 This preserves sparse venue history as ordinal samples rather than treating
 every elapsed calendar period as a physical SHM row. It also prevents the
 old `_first + 1` insertion point from overwriting the earliest published bar.
 > `git diff HEAD~1..HEAD -- tests/test_history_backfill.py`
 Added a full `tsdb_backfill()` regression modeled on the live NVDA failure.
 It uses a real `ShmArray`, NativeDB's field map, two inclusive reverse-provider
 frames, and overlapping stored history. Assertions prove the complete
 timestamp sequence is contiguous and lossless, contains no non-positive
 timestamps, sends complete provider frames to durable storage, and notifies
 chart/FSP consumers after the final storage prepend.
 Live evidence captured before workspace restart:
 ```text
 rows 1575296 first_index 1568233 last_index 3143528
 invalid_count 42720
 left boundary
 [(3065893, 1781271300, 204.03, 204.82)
 (3065894, 1781271360, 204.81, 204.72)
 (3065895,          0,   0.  ,   0.  )]
 right boundary
 [(3108614,          0,   0.  ,   0.  )
 (3108615, 1781271360, 204.81, 205.44)
 (3108616, 1781271420, 205.46, 205.48)]
 ```
 Verification output:
 ```text
 ........................................................................ [ 68%]
 .................................                                        [100%]
 105 passed in 4.80s
 ```
 Adversarial review caught stale `_first` semantics and an off-by-one
 insertion point in the initial patch. The first live restart then exposed an
 undefined `time_key` local at the new `tsdb_backfill()` call site, which
 crashed `datad.ib` and propagated to the chart as `RemoteActorError`. The
 helper now derives the source timestamp field from its field map, and the
 test enters the full enclosing lifecycle so the original `NameError` can not
 recur unnoticed. A final review also drove explicit publication after stored
 history is prepended and deterministic final-stage synchronization. Final
 review found no remaining issues; another live restart remains pending.
--- a/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md
+++ b/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md
@ -0,0 +1,36 @@
 ---
 model: openai/gpt-5.6-sol
 service: opencode
 session: ses_0799212ebffe42arY96czXn89F
 timestamp: 2026-08-11T20:58:26Z
 git_ref: 06d5ea5d
 scope: code
 substantive: true
 raw_file: 20260811T205826Z_06d5ea5d_prompt_io.raw.md
 ---
 ## Prompt
 Diagnose and fix an IB live-order failure where lazy NVDA contract
 qualification raised `tractor.trionics.Lagged` after the order reached
 `brokerd.ib`. Determine whether recent history changes or cancelled symbol
 searches caused the stale state.
 ## Response summary
 Traced the failure to competing consumers of a one-slot Trio broadcast, not
 the recent SHM change or stale asyncio client state. Replaced broadcast
 fan-out with one response/status demultiplexer, retained request-ID
 correlation, handled cancellation and teardown, documented the task/channel
 ownership model, warned on diagnostic metadata eviction, and added
 deterministic proxy regressions.
 ## Files changed
 - `piker/brokers/ib/api.py` - sole-reader IB proxy response routing.
 - `tests/test_ib_method_proxy.py` - status, concurrency, cancellation and EOF regressions.
 - `.claude/skills/run-tests/test-harness-reference.md` - proxy test mapping.
 ## Human edits
 None - generated output follows the user's live failure diagnosis.
--- a/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md
+++ b/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md
@ -0,0 +1,66 @@
 ---
 model: openai/gpt-5.6-sol
 service: opencode
 timestamp: 2026-08-11T20:58:26Z
 git_ref: 06d5ea5d
 diff_cmd: git diff HEAD~1..HEAD
 ---
 The user reported that certain IB order entries might have broken and
 provided a live failure for a first NVDA sell order. EMS delivered the
 request to `brokerd.ib`, but lazy `get_mkt_info('nvda.nasdaq')` contract
 qualification failed in `MethodProxy._run_method()` with:
 ```text
 tractor.trionics._broadcast.Lagged:
 Task `piker.brokers.ib.broker.handle_order_requests` overrun and
 dropped `0` values
 ```
 The user suspected a cancelled symbol-search task might have left stale
 asyncio state. Cancellation can leave an orphaned method response, but the
 observed failure was stale Trio broadcast state. `open_client_proxy()`
 broadcast a one-slot `LinkedTaskChannel` to an always-running IB status relay
 and an idle method receiver. Status traffic advanced only the relay, so the
 first method call for an uncached symbol entered an already-lagged receiver.
 Orders for eagerly cached contracts bypassed this path.
 The immediately preceding sparse-history commit did not modify IB order,
 symbol, or proxy code; it exposed no direct dependency on this failure.
 > `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py`
 Generated a sole-reader message demultiplexer for each IB client proxy.
 `relay_client_proxy_messages()` owns every asyncio-to-Trio receive, handles
 status/error traffic, and dispatches `mid`-tagged responses into per-call
 `MethodProxy` wait slots. Cancelled calls remove their pending slot, causing
 late responses to be logged and dropped without skewing later calls. Clean
 channel EOF is accepted during normal proxy teardown.
 Follow-up review restored the masked reconnect TODO and per-message case
 rationale, documented the one-reader/many-writer task hierarchy and separate
 per-proxy channels, and added a warning with request method metadata when the
 bounded unresolved-request diagnostic table evicts its oldest entry.
 > `git diff HEAD~1..HEAD -- tests/test_ib_method_proxy.py`
 Generated deterministic regressions for idle status traffic before first
 qualification, two concurrent calls with reverse-order responses, cancelled
 calls with late responses followed by a successful call, and graceful
 channel EOF. Every synchronization wait is timeout-bounded.
 > `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
 Registered the deterministic IB method-proxy target and source mapping.
 Verification output:
 ```text
 ........................                                                 [100%]
 25 passed, 4 warnings in 3.44s
 ```
 The warnings are existing Tractor/Trio deprecations from the EMS test.
 Adversarial review drove a clean-EOF guard and bounded polling. Final review
 found no actionable issues. Live IB order validation remains pending and
 requires restarting `brokerd.ib`.
--- a/docs/manual_backfill_qualification.rst
+++ b/docs/manual_backfill_qualification.rst
@ -24,8 +24,8 @@ automatically corruption or an expected venue closure.
   Do not run qualification against normal user storage. The helper below
   requires a marked disposable root before it will seed, archive, or clear
-   a Parquet. ``piker store anal`` and ``piker store ldshm`` are interactive
+   a Parquet. ``piker store anal`` and ``piker store shm --write-parquet``
-   and potentially mutating; they are not audit commands.
+   are interactive and potentially mutating; they are not audit commands.
   Keep the disposable root private and do not rename, replace, or symlink
   files under it while a helper is running. Stop the chart cleanly before
@ -193,6 +193,20 @@ visible until a venue-aware classifier proves their session alignment.
 Gap Layer Triage
 ----------------
 List exact durable series before choosing a timeframe, and inspect live
 SHM separately::
    piker store ls
    piker store series FQME_SUBSTRING
    piker store shm FQME --max-gaps 20
 All three forms are read-only. ``ls`` summarizes periods by FQME, while
 ``series`` reports each ``(FQME, period)`` file independently. Read-only
 ``shm`` reports every exact actor generation and
 ``hist``/``rt`` buffer for the FQME, including invalid rows, ordering
 defects, inferred versus expected cadence, and the largest timestamp gaps.
 Use ``--shm-name EXACT_NAME`` to inspect only one reported buffer.
 Use the persisted report and chart together:
 ===============================  =========================================
--- a/piker/brokers/ib/api.py
+++ b/piker/brokers/ib/api.py
@ -1643,6 +1643,61 @@ class MethodProxy:
        # to the next caller causing an off-by-one skew of
        # every result thereafter!
        self._mids = itertools.count()
        self._pending: dict[
            int,
            tuple[trio.Event, list[dict]],
        ] = {}
        self._request_methods: dict[int, str|None] = {}
    def _deliver_method_response(
        self,
        msg: dict,
    ) -> bool:
        '''
        Deliver one correlated response to its waiting caller.
        '''
        mid: int = msg['mid']
        meth: str|None = self._request_methods.pop(mid, None)
        pending = self._pending.get(mid)
        if pending is None:
            log.warning(
                f'Dropping stale method-resp,\n'
                f'mid: {mid}\n'
                f'meth: {meth!r}\n'
                f'(caller prolly got cancelled '
                f'before its resp?)\n'
            )
            return False
        event, slot = pending
        slot.append(msg)
        event.set()
        return True
    def _track_request_method(
        self,
        mid: int,
        meth: str|None,
    ) -> None:
        '''
        Retain bounded request metadata for stale-response logs.
        '''
        self._request_methods[mid] = meth
        if len(self._request_methods) > 256:
            oldest_mid: int = next(iter(self._request_methods))
            oldest_meth: str|None = self._request_methods.pop(
                oldest_mid
            )
            log.warning(
                f'Evicting unresolved method-request metadata,\n'
                f'mid: {oldest_mid}\n'
                f'meth: {oldest_meth!r}\n'
                f'pending metadata limit: 256\n'
            )
    async def _run_method(
        self,
@ -1656,65 +1711,30 @@ class MethodProxy:
        ``tractor.to_asyncio`` layer.
        '''
        chan = self.chan
        mid: int = next(self._mids)
-        await chan.send((meth, kwargs, mid))
+        event = trio.Event()
        slot: list[dict] = []
        self._pending[mid] = event, slot
        self._track_request_method(mid, meth)
        try:
            await self.chan.send((meth, kwargs, mid))
            await event.wait()
        finally:
            self._pending.pop(mid, None)
-        while not chan.closed():
+        msg: dict = slot[0]
-            msg = await chan.receive()
+        match msg:
            # OUR method-call response B)
            case {'result': res}:
                return res
-            # TODO: implement reconnect functionality like
+            case {'exception': err}:
-            # in our `.data._web_bs.NoBsWs`
+                raise err
            # try:
            #     msg = await chan.receive()
            # except ConnectionError:
            #     self.reset()
-            match msg:
+            case _:
-                # OUR method-call response B)
+                raise RuntimeError(
-                case {'mid': resp_mid, 'result': res} if (
+                    f'Invalid method response for mid={mid}: {msg!r}'
-                    resp_mid == mid
+                )
                ):
                    return res
                case {'mid': resp_mid, 'exception': err} if (
                    resp_mid == mid
                ):
                    raise err
                # an "orphaned" response to some prior
                # (cancelled) caller's request; drop it and
                # keep waiting for ours.
                case {'mid': resp_mid}:
                    log.warning(
                        f'Dropping stale method-resp,\n'
                        f'mid: {resp_mid} (ours: {mid})\n'
                        f'(a prior caller prolly got '
                        f'cancelled before its resp?)\n'
                    )
                    continue
                # out-of-band (inline) client error: raise
                # to the current caller as before.
                case {'exception': err}:
                    raise err
                case ('error', emsg):
                    log.warning(f'IB error relay: {emsg}')
                    continue
                # routine (api-farm conn) status events
                # relayed inline by `Client.inline_errors()`;
                # not a response to our method call so just
                # log at info and keep waiting.
                case ('event', emsg):
                    log.info(
                        f'IB status event relay: {emsg}'
                    )
                    continue
                case _:
                    log.warning(f'UNKNOWN IB MSG: {msg}')
    def status_event(
        self,
@ -1800,6 +1820,73 @@ async def open_aio_client_method_relay(
                raise ValueError(f'Unhandled msg {msg}')
 async def relay_client_proxy_messages(
    chan: tractor.to_asyncio.LinkedTaskChannel,
    proxy: MethodProxy,
 ) -> None:
    '''
    Route one proxy's asyncio-client msgs from a single reader.
    Each `open_client_proxy()` allocates its own linked channel and
    starts this task in its relay nursery. Any number of Trio caller
    tasks may concurrently write tagged requests through
    `MethodProxy._run_method()` and wait on their per-`mid` slots.
    This task alone reads the channel and dispatches each response to
    its waiting caller.
    Separate proxy contexts own separate channels and reader tasks;
    no cross-proxy response broadcast is intended. IB status events
    still reach each proxy through its own `Client.inline_errors()`
    handler.
    '''
    try:
        while not chan.closed():
            # TODO: implement reconnect functionality like
            # in our `.data._web_bs.NoBsWs`
            # try:
            msg = await chan.receive()
            # except ConnectionError:
            #     proxy.reset()
            match msg:
                # Correlated response from a method call.
                case {'mid': _}:
                    proxy._deliver_method_response(msg)
                # routine (api-farm conn) status events
                # relayed inline by `Client.inline_errors()`;
                # not a response to a method call so just
                # log at info and keep waiting.
                case ('event', status_msg):
                    reason = status_msg['reason']
                    event = proxy.event_table.pop(reason, None)
                    if (
                        event
                        and
                        event.statistics().tasks_waiting
                    ):
                        log.info(f'Relaying ib status message: {msg}')
                        event.set()
                # Inline API error with no request correlation.
                case ('error', emsg):
                    log.warning(f'IB error relay: {emsg}')
                # Out-of-band client error with no method `mid`.
                case {'exception': err}:
                    log.error(
                        f'Uncorrelated IB client error: {err!r}'
                    )
                # Preserve visibility into unexpected relay traffic.
                case _:
                    log.warning(f'UNKNOWN IB MSG: {msg}')
    except trio.EndOfChannel:
        log.info('IB client proxy relay closed')
@acm
 async def open_client_proxy(
    client: Client,
@ -1836,28 +1923,11 @@ async def open_client_proxy(
                continue
            setattr(proxy, name, partial(proxy._run_method, meth=name))
-        async def relay_events():
+        relay_tn.start_soon(
-
+            relay_client_proxy_messages,
-            async with chan.subscribe() as msg_stream:
+            chan,
-
+            proxy,
-                async for msg in msg_stream:
+        )
                    if 'event' not in msg:
                        continue
                    # if 'event' in msg:
                    # wake up any system event waiters.
                    etype, status_msg = msg
                    reason = status_msg['reason']
                    ev = proxy.event_table.pop(reason, None)
                    if ev and ev.statistics().tasks_waiting:
                        log.info(f'Relaying ib status message: {msg}')
                        ev.set()
                    continue
        relay_tn.start_soon(relay_events)
        yield proxy
--- a/piker/data/_sharedmem.py
+++ b/piker/data/_sharedmem.py
@ -63,6 +63,28 @@ def _make_token(
    )
 def attach_existing_shm_array(
    key: str,
    size: int,
    dtype: np.dtype|None = None,
    readonly: bool = True,
 ) -> ShmArray:
    '''
    Attach to an existing segment without creating one on races.
    '''
    token: NDToken = _known_tokens.get(key) or _make_token(
        key,
        size=size,
        dtype=dtype,
    )
    return attach_shm_ndarray(
        token=token,
        readonly=readonly,
    )
 def maybe_open_shm_array(
    key: str,
    size: int,
--- a/piker/storage/cli.py
+++ b/piker/storage/cli.py
@ -19,12 +19,17 @@ Storage middle-ware CLIs.
 """
 from __future__ import annotations
 from datetime import (
    UTC,
    datetime,
 )
 import json
 from pathlib import Path
 import sys
 import time
 from types import ModuleType
 from typing import (
    Annotated,
    TYPE_CHECKING,
 )
@ -46,18 +51,43 @@ from piker import tsp
 from piker import config
 from . import log
 from . import (
    __tsdbs__,
    open_storage_client,
    StorageClient,
 )
 from ._audit import audit_ohlcv_parquet
-from .nativedb import mk_ohlcv_shm_keyed_filepath
+from .nativedb import (
    iter_native_series,
    mk_ohlcv_shm_keyed_filepath,
    NativeSeriesRef,
 )
 if TYPE_CHECKING:
    from piker.ui._remote_ctl import AnnotCtl
-store = typer.Typer()
+store = typer.Typer(no_args_is_help=True)
 _shm_periods: dict[str, int] = {
    'hist': 60,
    'rt': 1,
 }
 def _format_utc(timestamp: float) -> str|None:
    '''
    Format a timestamp without aborting on corrupt epochs.
    '''
    try:
        return datetime.fromtimestamp(
            timestamp,
            tz=UTC,
        ).isoformat()
    except (
        OSError,
        OverflowError,
        ValueError,
    ):
        return None
 def _shm_period_and_invalid_count(
@ -91,6 +121,95 @@ def _shm_period_and_invalid_count(
    return period, invalid_count
 def _summarize_shm_frame(
    ref: tsp.ShmBufferRef,
    frame: np.ndarray,
    max_gaps: int,
 ) -> dict:
    '''
    Summarize one immutable OHLCV SHM snapshot.
    '''
    times: np.ndarray = frame['time']
    observed_period_s, invalid_count = (
        _shm_period_and_invalid_count(times)
    )
    period_s: int = _shm_periods[ref.kind]
    valid: np.ndarray = (
        np.isfinite(times)
        &
        (times > 0)
    )
    published: np.ndarray = times[valid]
    steps: np.ndarray = np.diff(published)
    duplicate_steps: int = int(np.count_nonzero(steps == 0))
    reversed_steps: int = int(np.count_nonzero(steps < 0))
    gaps: list[dict] = []
    if published.size > 1:
        indexes: np.ndarray = np.flatnonzero(steps > period_s)
        ranked: list[int] = sorted(
            indexes.tolist(),
            key=lambda index: steps[index],
            reverse=True,
        )
        for index in ranked[:max_gaps]:
            left: float = float(published[index])
            right: float = float(published[index + 1])
            gaps.append({
                'delta_s': float(steps[index]),
                'left_timestamp': left,
                'left_utc': _format_utc(left),
                'right_timestamp': right,
                'right_utc': _format_utc(right),
            })
    invalid: np.ndarray = np.flatnonzero(~valid)
    return {
        'shm_name': ref.path.name,
        'service': ref.service,
        'generation': ref.generation,
        'fqme': ref.fqme,
        'kind': ref.kind,
        'rows': len(frame),
        'first_index': (
            int(frame['index'][0])
            if len(frame)
            else None
        ),
        'last_index': (
            int(frame['index'][-1])
            if len(frame)
            else None
        ),
        'first_timestamp': (
            float(published[0])
            if published.size
            else None
        ),
        'last_timestamp': (
            float(published[-1])
            if published.size
            else None
        ),
        'period_s': period_s,
        'observed_period_s': observed_period_s,
        'invalid_count': invalid_count,
        'duplicate_step_count': duplicate_steps,
        'reversed_step_count': reversed_steps,
        'invalid_index_bounds': (
            [
                int(frame['index'][invalid[0]]),
                int(frame['index'][invalid[-1]]),
            ]
            if invalid.size
            else None
        ),
        'gap_count': len(np.flatnonzero(steps > period_s)),
        'largest_gaps': gaps,
    }
 def _render_audit_report(report: dict) -> None:
    '''
    Render a compact human summary and explicit gap endpoints.
@ -150,48 +269,146 @@ def _render_audit_report(report: dict) -> None:
    console.print(gap_table)
 def _native_series_payload(
    pattern: str|None,
 ) -> tuple[Path, list[dict]]:
    '''
    Collect exact NativeDB series metadata for CLI presentation.
    '''
    datadir: Path = config.get_conf_dir() / 'nativedb'
    refs: list[NativeSeriesRef] = list(iter_native_series(datadir))
    if pattern is not None:
        needle: str = pattern.casefold()
        refs = [
            ref
            for ref in refs
            if needle in ref.fqme.casefold()
        ]
    payload: list[dict] = []
    for ref in refs:
        try:
            size_bytes: int = ref.path.stat().st_size
        except FileNotFoundError:
            continue
        payload.append({
            'fqme': ref.fqme,
            'period_s': ref.period_s,
            'path': str(ref.path),
            'size_bytes': size_bytes,
        })
    payload.sort(key=lambda item: (
        item['fqme'],
        item['period_s'],
    ))
    return datadir, payload
@store.command()
 def ls(
-    backends: list[str] = typer.Argument(
+    pattern: Annotated[
-        default=None,
+        str|None,
-        help='Storage backends to query, default is all.'
+        typer.Argument(
-    ),
+            help='Optional case-insensitive FQME substring.',
-):
+        ),
-    from rich.table import Table
+    ] = None,
    all_series: Annotated[
        bool,
        typer.Option(
            '--all',
            help='List all FQMEs (the default).',
        ),
    ] = False,
 ) -> None:
    '''
    List NativeDB FQMEs and their available sample periods.
-    if not backends:
+    '''
-        backends: list[str] = __tsdbs__
+    if (
        all_series
        and
        pattern is not None
    ):
        raise typer.BadParameter('Pass a pattern or --all, not both')
-    console = Console()
+    datadir, payload = _native_series_payload(pattern)
    periods_by_fqme: dict[str, list[int]] = {}
    for item in payload:
        periods_by_fqme.setdefault(item['fqme'], []).append(
            item['period_s']
        )
-    async def query_all():
+    table = Table(title=f'NativeDB series @ {datadir}')
-        nonlocal backends
+    table.add_column('FQME')
-
+    table.add_column('Periods')
-        async with (
+    for fqme, periods in periods_by_fqme.items():
-            open_piker_runtime(
+        table.add_row(
-                'tsdb_storage',
+            fqme,
            ', '.join(
                f'{period}s'
                for period in sorted(periods)
            ),
-        ):
+        )
-            for i, backend in enumerate(backends):
+    Console().print(table)
                table = Table()
                try:
                    async with open_storage_client(backend=backend) as (
                        mod,
                        client,
                    ):
                        table.add_column(f'{mod.name}@{client.address}')
                        keys: list[str] = await client.list_keys()
                        for key in keys:
                            table.add_row(key)
                    console.print(table)
                except Exception:
                    log.error(f'Unable to connect to storage engine: `{backend}`')
    trio.run(query_all)
@store.command()
 def series(
    pattern: Annotated[
        str|None,
        typer.Argument(
            help='Optional case-insensitive FQME substring.',
        ),
    ] = None,
    all_series: Annotated[
        bool,
        typer.Option(
            '--all',
            help='List all series (the default).',
        ),
    ] = False,
    json_output: Annotated[
        bool,
        typer.Option(
            '--json',
            help='Emit machine-readable JSON.',
        ),
    ] = False,
 ) -> None:
    '''
    List detailed NativeDB sub-series without opening a runtime.
    '''
    if (
        all_series
        and
        pattern is not None
    ):
        raise typer.BadParameter('Pass a pattern or --all, not both')
    datadir, payload = _native_series_payload(pattern)
    if json_output:
        typer.echo(json.dumps(payload, indent=2, sort_keys=True))
        return
    table = Table(title=f'NativeDB series @ {datadir}')
    table.add_column('FQME')
    table.add_column('Period')
    table.add_column('Bytes', justify='right')
    table.add_column('Path')
    for item in payload:
        table.add_row(
            item['fqme'],
            f'{item["period_s"]}s',
            str(item['size_bytes']),
            item['path'],
        )
    Console().print(table)
@store.command(no_args_is_help=True)
 def audit(
    fqme: str,
    period: int = typer.Option(
@ -367,7 +584,7 @@ def audit(
 #     ...
-@store.command()
+@store.command(no_args_is_help=True)
 def delete(
    symbols: list[str],
@ -405,7 +622,7 @@ def delete(
    trio.run(main, symbols)
-@store.command()
+@store.command(no_args_is_help=True)
 def anal(
    fqme: str,
    period: int = 60,
@ -494,21 +711,234 @@ def anal(
    trio.run(main)
-@store.command()
+def _inspect_shm(
 def ldshm(
    fqme: str,
-    write_parquet: bool = True,
+    shm_name: Annotated[
-    reload_parquet_to_shm: bool = True,
+        str|None,
        typer.Option(
            '--shm-name',
            help='Inspect only this exact SHM object name.',
        ),
    ] = None,
    max_gaps: Annotated[
        int,
        typer.Option(
            '--max-gaps',
            min=0,
            help='Maximum largest gaps to show per buffer.',
        ),
    ] = 10,
    json_output: Annotated[
        bool,
        typer.Option(
            '--json',
            help='Emit machine-readable JSON.',
        ),
    ] = False,
    pdb: bool = False,
 ) -> None:
    '''
    Inspect matching OHLCV SHM buffers without mutating them.
    '''
    refs: list[tsp.ShmBufferRef] = [
        ref
        for ref in tsp.iter_shm_buffer_refs(fqme=fqme)
        if (
            shm_name is None
            or
            ref.path.name == shm_name
        )
    ]
    if not refs:
        typer.echo(
            f'No exact OHLCV SHM buffers found for {fqme!r}',
            err=True,
        )
        raise typer.Exit(code=2)
    ref_by_name: dict[str, tsp.ShmBufferRef] = {
        ref.path.name: ref
        for ref in refs
    }
    reports: list[dict] = []
    async def main() -> None:
        async with open_piker_runtime(
            'shm_inspector',
            debug_mode=pdb,
        ):
            for (
                shmfile,
                shm,
                _shm_df,
            ) in tsp.iter_dfs_from_shms(
                fqme,
                shm_name=shm_name,
                refs=refs,
            ):
                ref: tsp.ShmBufferRef = ref_by_name[shmfile.name]
                reports.append(_summarize_shm_frame(
                    ref,
                    shm.array.copy(),
                    max_gaps,
                ))
    trio.run(main)
    if not reports:
        typer.echo(
            'Selected SHM buffers disappeared before inspection',
            err=True,
        )
        raise typer.Exit(code=2)
    if json_output:
        typer.echo(json.dumps(reports, indent=2, sort_keys=True))
        return
    table = Table(title=f'OHLCV SHM buffers for {fqme}')
    table.add_column('Kind')
    table.add_column('Generation')
    table.add_column('Rows', justify='right')
    table.add_column('Period')
    table.add_column('Invalid', justify='right')
    table.add_column('Order', justify='right')
    table.add_column('Gaps', justify='right')
    table.add_column('SHM name')
    for report in reports:
        period_s: int = report['period_s']
        table.add_row(
            report['kind'],
            report['generation'],
            str(report['rows']),
            f'{period_s}s',
            str(report['invalid_count']),
            str(
                report['duplicate_step_count']
                +
                report['reversed_step_count']
            ),
            str(report['gap_count']),
            report['shm_name'],
        )
    Console().print(table)
    for report in reports:
        gaps: list[dict] = report['largest_gaps']
        if not gaps:
            continue
        gap_table = Table(
            title=f'Largest gaps: {report["shm_name"]}'
        )
        gap_table.add_column('Delta (s)')
        gap_table.add_column('Left UTC')
        gap_table.add_column('Right UTC')
        for gap in gaps:
            gap_table.add_row(
                str(gap['delta_s']),
                str(gap['left_utc']),
                str(gap['right_utc']),
            )
        Console().print(gap_table)
@store.command(no_args_is_help=True)
 def shm(
    fqme: str,
    write_parquet: Annotated[
        bool,
        typer.Option(
            '--write-parquet',
            help='Persist repaired frames to NativeDB.',
        ),
    ] = False,
    reload_parquet_to_shm: Annotated[
        bool,
        typer.Option(
            '--reload-parquet-to-shm',
            help='Reload persisted repairs into SHM.',
        ),
    ] = False,
    pdb: bool = False,  # --pdb passed?
    max_gaps: Annotated[
        int|None,
        typer.Option(
            '--max-gaps',
            min=0,
            help=(
                'Maximum largest gaps in read-only output '
                '(default: 10).'
            ),
            show_default=False,
        ),
    ] = None,
    json_output: Annotated[
        bool,
        typer.Option(
            '--json',
            help='Emit read-only output as JSON.',
        ),
    ] = False,
    shm_name: Annotated[
        str|None,
        typer.Option(
            '--shm-name',
            help='Select only this exact SHM object name.',
        ),
    ] = None,
 ) -> None:
    '''
-    Linux ONLY: load any fqme file name matching shm buffer from
+    Inspect volatile OHLCV SHM, with explicit repair options.
    /dev/shm/ into an OHLCV numpy array and polars DataFrame,
    optionally write to offline storage via `.parquet` file.
    '''
-    async def main():
+    if (
        reload_parquet_to_shm
        and
        not write_parquet
    ):
        raise typer.BadParameter(
            '--reload-parquet-to-shm requires --write-parquet'
        )
    if (
        write_parquet
        and (
            json_output
            or
            max_gaps is not None
        )
    ):
        raise typer.BadParameter(
            '--json and --max-gaps apply only to read-only output'
        )
    if not write_parquet:
        _inspect_shm(
            fqme,
            shm_name=shm_name,
            max_gaps=(
                10
                if max_gaps is None
                else max_gaps
            ),
            json_output=json_output,
            pdb=pdb,
        )
        return
    if (
        shm_name is not None
        and
        not any(
            ref.path.name == shm_name
            for ref in tsp.iter_shm_buffer_refs(fqme=fqme)
        )
    ):
        typer.echo(
            f'No exact OHLCV SHM buffer named {shm_name!r}',
            err=True,
        )
        raise typer.Exit(code=2)
    async def main() -> bool:
        from piker.ui._remote_ctl import (
            open_annot_ctl,
        )
@ -530,24 +960,35 @@ def ldshm(
            shm_df: pl.DataFrame | None = None
            tf2aids: dict[float, dict] = {}
            iter_kwargs: dict = {}
            if shm_name is not None:
                iter_kwargs['shm_name'] = shm_name
            for (
                shmfile,
                shm,
                # parquet_path,
                shm_df,
-            ) in tsp.iter_dfs_from_shms(fqme):
+            ) in tsp.iter_dfs_from_shms(fqme, **iter_kwargs):
                times: np.ndarray = shm_df['time'].to_numpy()
                (
-                    period_s,
+                    observed_period_s,
                    invalid_count,
                ) = _shm_period_and_invalid_count(times)
-                if period_s is None:
+                if observed_period_s is None:
                    log.warning(
                        f'Could not infer a positive sample period '
                        f'for {shmfile.name}; skipping buffer\n'
                    )
                    continue
                kind: str = shmfile.suffix.removeprefix('.')
                period_s: int = _shm_periods[kind]
                if observed_period_s != period_s:
                    log.warning(
                        f'Modal step {observed_period_s}s in '
                        f'{period_s}s {shmfile.name}; using kind '
                        f'for storage identity\n'
                    )
                log.info(
                    f'Processing shm buffer:\n'
                    f'  file: {shmfile.name}\n'
@ -696,12 +1137,11 @@ def ldshm(
                    do_markup_gaps: bool = True
                    if do_markup_gaps:
                        new_df: pl.DataFrame = tsp.np2pl(new)
                        aids: dict = await tsp._annotate.markup_gaps(
                            fqme,
                            period_s,
                            actl,
-                            new_df,
+                            deduped,
                            step_gaps,
                        )
                        # last chance manual overwrites in REPL
@ -724,16 +1164,19 @@ def ldshm(
                            'but no significant time gaps!\n'
                        )
            await tractor.pause()
            log.info('Exiting TSP shm anal-izer!')
            if shm_df is None:
                log.error(
                    f'No matching shm buffers for {fqme} ?'
                )
                return False
-    trio.run(main)
+            await tractor.pause()
            log.info('Exiting TSP shm anal-izer!')
            return True
    if not trio.run(main):
        raise typer.Exit(code=2)
 typer_click_object = typer.main.get_command(store)
--- a/piker/storage/nativedb.py
+++ b/piker/storage/nativedb.py
@ -51,7 +51,9 @@ YET!
 #   - https://github.com/spslater/borgapi
 #   - https://nixos.wiki/wiki/ZFS
 from __future__ import annotations
 from collections.abc import Iterator
 from contextlib import asynccontextmanager as acm
 from dataclasses import dataclass
 from datetime import datetime
 import os
 from pathlib import Path
@ -82,6 +84,17 @@ _price_fields: tuple[str, ...] = (
 )
@dataclass(frozen=True)
 class NativeSeriesRef:
    '''
    Identify one canonical NativeDB OHLCV file.
    '''
    fqme: str
    period_s: int
    path: Path
 def detect_period(shm: ShmArray) -> float:
    '''
    Attempt to detect the series time step sampling period
@ -107,8 +120,14 @@ def mk_ohlcv_shm_keyed_filepath(
 ) -> Path:
-    if period < 1.:
+    if (
-        raise ValueError('Sample period should be >= 1.!?')
+        period < 1.
        or
        int(period) != period
    ):
        raise ValueError(
            'Sample period must be positive whole seconds'
        )
    path: Path = (
        datadir
@ -118,12 +137,98 @@ def mk_ohlcv_shm_keyed_filepath(
    return path
-def unpack_fqme_from_parquet_filepath(path: Path) -> str:
+def parse_ohlcv_parquet_path(
    path: Path,
-    filename: str = str(path.name)
+) -> NativeSeriesRef|None:
-    fqme, fmt_descr, suffix = filename.split('.')
+    '''
-    assert suffix == 'parquet'
+    Parse one canonical NativeDB OHLCV filename.
-    return fqme
+
    '''
    name: str = path.name
    if not name.endswith('.parquet'):
        return None
    stem: str = name.removesuffix('.parquet')
    fqme, marker, period_text = stem.rpartition('.ohlcv')
    if (
        marker != '.ohlcv'
        or
        not fqme
        or
        not period_text.endswith('s')
    ):
        return None
    digits: str = period_text.removesuffix('s')
    if (
        not digits.isascii()
        or
        not digits.isdecimal()
        or
        digits.startswith('0')
        or
        '..' in path.parts
    ):
        return None
    period_s: int = int(digits)
    if (
        fqme in {'.', '..'}
        or
        Path(fqme).name != fqme
        or
        '/' in fqme
        or
        '\\' in fqme
    ):
        return None
    expected: Path = mk_ohlcv_shm_keyed_filepath(
        fqme,
        period_s,
        path.parent,
    )
    if expected.name != name:
        return None
    return NativeSeriesRef(
        fqme=fqme,
        period_s=period_s,
        path=path,
    )
 def iter_native_series(
    datadir: Path,
 ) -> Iterator[NativeSeriesRef]:
    '''
    Yield canonical NativeDB series in deterministic order.
    '''
    if not datadir.is_dir():
        return
    for path in sorted(
        datadir.iterdir(),
        key=lambda candidate: candidate.name,
    ):
        if (
            path.is_symlink()
            or
            not path.is_file()
        ):
            continue
        ref: NativeSeriesRef|None = parse_ohlcv_parquet_path(path)
        if ref is not None:
            yield ref
 def unpack_fqme_from_parquet_filepath(path: Path) -> str:
    ref: NativeSeriesRef|None = parse_ohlcv_parquet_path(path)
    if ref is None:
        raise ValueError(f'Invalid NativeDB OHLCV path: {path}')
    return ref.fqme
 # Only hydrate fields shared by canonical storage and provider SHM.
@ -152,7 +257,7 @@ class NativeStorageClient:
    ) -> None:
        self._datadir = datadir
-        self._index: dict[str, dict] = {}
+        self._index: dict[tuple[str, int], NativeSeriesRef] = {}
        # series' cache from tsdb reads
        self._dfs: dict[str, dict[str, pl.DataFrame]] = {}
@ -163,6 +268,10 @@ class NativeStorageClient:
    @property
    def cardinality(self) -> int:
        return len({fqme for fqme, _ in self._index})
    @property
    def series_cardinality(self) -> int:
        return len(self._index)
    # @property
@ -170,29 +279,15 @@ class NativeStorageClient:
    #     ...
    async def list_keys(self) -> list[str]:
-        return list(self._index)
+        return sorted({fqme for fqme, _ in self._index})
    async def list_series(self) -> list[NativeSeriesRef]:
        return list(self._index.values())
    def index_files(self):
-        for path in self._datadir.iterdir():
+        self._index.clear()
-            if (
+        for ref in iter_native_series(self._datadir):
-                path.is_dir()
+            self._index[(ref.fqme, ref.period_s)] = ref
                or
                path.suffix != '.parquet'
                # or
                # path.name in {'borked', 'expired',}
            ):
                continue
            key: str = path.name.removesuffix('.parquet')
            fqme, _, descr = key.rpartition('.')
            prefix, _, suffix = descr.partition('ohlcv')
            period: int = int(suffix.strip('s'))
            # cache description data
            self._index[fqme] = {
                'path': path,
                'period': period,
            }
        return self._index
@ -227,11 +322,11 @@ class NativeStorageClient:
            bs_fqme, _, *_ = fqme.rpartition('.')
            possible_matches: list[str] = []
-            for tskey in self._index:
+            for tsfqme, _period in self._index:
-                if bs_fqme in tskey:
+                if bs_fqme in tsfqme:
-                    possible_matches.append(tskey)
+                    possible_matches.append(tsfqme)
-            match_str: str = '\n'.join(sorted(possible_matches))
+            match_str: str = '\n'.join(sorted(set(possible_matches)))
            raise TimeseriesNotFound(
                f'No entry for `{fqme}`?\n'
                f'Maybe you need a more specific fqme-key like:\n\n'
--- a/piker/tsp/init.py
+++ b/piker/tsp/init.py
@ -45,8 +45,12 @@ from ._dedupe_smart import (
    dedupe_ohlcv_smart as dedupe_ohlcv_smart,
 )
 from ._history import (
    iter_shm_buffer_refs as iter_shm_buffer_refs,
    iter_dfs_from_shms as iter_dfs_from_shms,
    manage_history as manage_history,
    parse_shm_buffer_path as parse_shm_buffer_path,
    resolve_shm_buffer_dtype as resolve_shm_buffer_dtype,
    ShmBufferRef as ShmBufferRef,
 )
 from ._annotate import (
    markup_gaps as markup_gaps,
--- a/piker/tsp/_history.py
+++ b/piker/tsp/_history.py
@ -29,15 +29,18 @@ Historical TSP (time-series processing) lowlevel mgmt machinery and biz logic fo
 '''
 from __future__ import annotations
 from dataclasses import dataclass
 from datetime import datetime
 from functools import partial
 from pathlib import Path
 import platform
 from pprint import pformat
 from string import hexdigits
 from types import ModuleType
 from typing import (
    Callable,
    Generator,
    Literal,
    TYPE_CHECKING,
 )
@ -56,6 +59,7 @@ import numpy as np
 import polars as pl
 from piker.brokers import NoData
 from piker.brokers import get_brokermod
 from piker.accounting import (
    MktPair,
 )
@ -64,7 +68,10 @@ from piker.log import (
    get_console_log,
 )
 from tractor.ipc._shm import ShmArray
-from ..data._sharedmem import maybe_open_shm_array
+from ..data._sharedmem import (
    attach_existing_shm_array,
    maybe_open_shm_array,
 )
 from piker.data._source import (
    def_iohlcv_fields,
 )
@ -104,6 +111,19 @@ log = get_logger(
 )
@dataclass(frozen=True)
 class ShmBufferRef:
    '''
    Identify one Piker OHLCV shared-memory buffer.
    '''
    service: str
    generation: str
    fqme: str
    kind: Literal['hist', 'rt']
    path: Path
 # `ShmArray` buffer sizing configuration:
 _mins_in_day = int(60 * 24)
 # how much is probably dependent on lifestyle
@ -240,7 +260,7 @@ async def shm_push_in_between(
    prepend_index: int,
    backfill_until_dt: datetime,
-    update_start_on_prepend: bool = False,
+    update_start_on_prepend: bool = True,
 ) -> int:
@ -262,16 +282,12 @@ async def shm_push_in_between(
        to_push,
        prepend=True,
-        # XXX: only update the ._first index if no tsdb
+        # Keep the readable `ShmArray.array` boundary aligned with
-        # segment was previously prepended by the
+        # every completed provider prepend.
        # parent task.
        update_first=update_start_on_prepend,
-        # XXX: only prepend from a manually calculated shm
+        # An explicit index is only needed by callers retaining a
-        # index if there was already a tsdb history
+        # separately placed segment before the readable boundary.
        # segment prepended (since then the
        # ._first.value is going to be wayyy in the
        # past!)
        start=(
            prepend_index
            if not update_start_on_prepend
@ -538,7 +554,7 @@ async def start_backfill(
        task_status.started()
        # based on the sample step size, maybe load a certain amount history
-        update_start_on_prepend: bool = False
+        update_start_on_prepend: bool = True
        if (
            _until_was_none := (backfill_until_dt is None)
        ):
@ -559,8 +575,6 @@ async def start_backfill(
                60: {'years': 6},
            }
            period_duration: int = periods[timeframe]
            update_start_on_prepend: bool = True
            # NOTE: manually set the "latest" datetime which we intend to
            # backfill history "until" so as to adhere to the history
            # settings above when the tsdb is detected as being empty.
@ -747,7 +761,7 @@ async def start_backfill(
                prepend_until_dt=backfill_until_dt,
            )
            # The requested end bar is already the oldest published
-            # sample. Keep it for idempotent storage merge, but do not
+            # sample. Keep it for 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()
@ -1144,6 +1158,74 @@ async def load_tsdb_hist(
        return None
 def _prepend_tsdb_history(
    shm: ShmArray,
    tsdb_history: np.ndarray,
    field_map: bidict|None,
 ) -> np.ndarray:
    '''
    Prepend stored rows beside the actual provider boundary.
    '''
    available: int = max(0, int(shm._first.value))
    if (
        not available
        or
        not len(tsdb_history)
    ):
        return tsdb_history[:0]
    frame: np.ndarray = shm.array
    if not len(frame):
        return tsdb_history[:0]
    # XXX EDGE CASEs: the most recent provider frame can overlap
    # prior TSDB history when its start is earlier than the latest
    # stored sample.
    #
    # Alternatively, this can occur when the venue was closed (say
    # over a weekend), causing a wall-clock time-series gap while the
    # provider's query frame is larger than the current market's
    # operating session. For example, IB's 1s default returns around
    # 2k datums (~33.33m), which can include samples from before a
    # closure. The wall-clock interval between frames then differs
    # from the number of actual venue samples and may even look like
    # a negative overlap when compared only by endpoints.
    #
    # The previous placement converted that elapsed interval into a
    # physical SHM offset. Sparse venue series contain samples, not
    # one row per wall-clock period, so closures accumulated as null
    # reservations between otherwise adjacent data. Instead, wait
    # until reverse provider retrieval establishes the actual first
    # sample, retain its values for any overlap, and prepend only TSDB
    # samples older than that boundary.
    #
    # TODO: assert overlapping TSDB and provider segments contain the
    # same values before preferring the provider's copy.
    time_key: str = (
        field_map.inverse.get('time', 'time')
        if field_map
        else 'time'
    )
    earliest_t: float = frame['time'][0]
    older: np.ndarray = tsdb_history[
        tsdb_history[time_key] < earliest_t
    ]
    # Stored history may extend further into the past than remaining
    # SHM capacity. Keep only its most recent rows in that case.
    to_push: np.ndarray = older[-available:]
    if len(to_push):
        shm.push(
            to_push,
            prepend=True,
            field_map=field_map,
        )
    return to_push
 async def tsdb_backfill(
    mod: ModuleType,
    storemod: ModuleType,
@ -1229,7 +1311,7 @@ async def tsdb_backfill(
        # NOTE: iabs to start backfilling from, reverse chronological,
        # ONLY AFTER the first history frame has been pushed to
        # mem!
-        backfill_gap_from_shm_index: int = shm._first.value + 1
+        backfill_gap_from_shm_index: int = shm._first.value
        # Prepend any tsdb history into the rt-shm-buffer which
        # should NOW be getting filled with the most recent history
@ -1318,76 +1400,29 @@ async def tsdb_backfill(
                        write_tsdb=True,
                    )
                )
                if last_tsdb_dt is not None:
                    # calc the index from which the tsdb data should be
                    # prepended, presuming there is a gap between the
                    # latest frame (loaded/read above) and the latest
                    # sample loaded from the tsdb.
                    backfill_diff: Duration = mr_start_dt - last_tsdb_dt
                    offset_s: float = backfill_diff.in_seconds()
                    # XXX EDGE CASEs: the most recent frame overlaps with
                    # prior tsdb history!!
                    # - so the latest frame's start time is earlier then
                    # the tsdb's latest sample.
                    # - alternatively this may also more generally occur
                    # when the venue was closed (say over the weeknd)
                    # causing a timeseries gap, AND the query frames size
                    # (eg. for ib's 1s we rx 2k datums ~= 33.33m) IS
                    # GREATER THAN the current venue-market's operating
                    # session (time) we will receive datums from BEFORE THE
                    # CLOSURE GAP and thus the `offset_s` value will be
                    # NEGATIVE! In this case we need to ensure we don't try
                    # to push datums that have already been recorded in the
                    # tsdb. In this case we instead only retreive and push
                    # the series portion missing from the db's data set.
                    # if offset_s < 0:
                    #     non_overlap_diff: Duration = mr_end_dt - last_tsdb_dt
                    #     non_overlap_offset_s: float = backfill_diff.in_seconds()
                    offset_samples: int = round(offset_s / timeframe)
                    # TODO: see if there's faster multi-field reads:
                    # https://numpy.org/doc/stable/user/basics.rec.html#accessing-multiple-fields
                    # re-index  with a `time` and index field
                    if offset_s > 0:
                        # NOTE XXX: ONLY when there is an actual gap
                        # between the earliest sample in the latest history
                        # frame do we want to NOT stick the latest tsdb
                        # history adjacent to that latest frame!
                        prepend_start = shm._first.value - offset_samples + 1
                        to_push = tsdb_history[-prepend_start:]
                    else:
                        # when there is overlap we want to remove the
                        # overlapping samples from the tsdb portion (taking
                        # instead the latest frame's values since THEY
                        # SHOULD BE THE SAME) and prepend DIRECTLY adjacent
                        # to the latest frame!
                        # TODO: assert the overlap segment array contains
                        # the same values!?!
                        prepend_start = shm._first.value
                        to_push = tsdb_history[-(shm._first.value):offset_samples - 1]
                    # tsdb history is so far in the past we can't fit it in
                    # shm buffer space so simply don't load it!
                    if prepend_start > 0:
                        shm.push(
                            to_push,
                            # insert the history pre a "days worth" of samples
                            # to leave some real-time buffer space at the end.
                            prepend=True,
                            # update_first=False,
                            start=prepend_start,
                            field_map=storemod.ohlc_key_map,
                        )
                        log.info(f'Loaded {to_push.shape} datums from storage')
            # 2nd nursery END
            if last_tsdb_dt is not None:
                # Provider frames contain venue samples, not one row
                # per wall-clock period. Wait until reverse retrieval
                # establishes the actual earliest sample, then prepend
                # storage directly beside it without reserving closure
                # rows.
                to_push: np.ndarray = _prepend_tsdb_history(
                    shm,
                    tsdb_history,
                    field_map=storemod.ohlc_key_map,
                )
                log.info(
                    f'Loaded {to_push.shape} datums from storage'
                )
                if len(to_push):
                    await notify_backfill(
                        sampler_stream,
                        mkt,
                        timeframe,
                    )
                # 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!
@ -1681,8 +1716,182 @@ async def manage_history(
            await trio.sleep_forever()
 def parse_shm_buffer_path(
    path: Path,
 ) -> ShmBufferRef|None:
    '''
    Parse one Piker OHLCV shared-memory filename.
    '''
    kind: Literal['hist', 'rt']
    if path.name.endswith('.hist'):
        kind = 'hist'
    elif path.name.endswith('.rt'):
        kind = 'rt'
    else:
        return None
    stem: str = path.name.removesuffix(f'.{kind}')
    service_prefix, open_sep, generation_tail = stem.partition('[')
    generation, close_sep, fqme = generation_tail.partition('].')
    if (
        open_sep != '['
        or
        close_sep != '].'
        or
        not service_prefix.startswith('piker.')
        or
        not generation
        or
        not fqme
    ):
        return None
    service: str = service_prefix.removeprefix('piker.')
    generation_valid: bool = (
        len(generation) == 16
        and
        generation[8] == '-'
        and
        generation[13] == '-'
        and
        all(
            char in hexdigits
            for index, char in enumerate(generation)
            if index not in {8, 13}
        )
    )
    service_valid: bool = (
        bool(service)
        and
        service[0].isalnum()
        and
        service[-1].isalnum()
        and
        all(
            char.isalnum()
            or
            char in {'-', '_', '.'}
            for char in service
        )
    )
    fqme_valid: bool = (
        fqme not in {'.', '..'}
        and
        Path(fqme).name == fqme
        and
        '/' not in fqme
        and
        '\\' not in fqme
        and
        '[' not in fqme
        and
        ']' not in fqme
    )
    if not (
        generation_valid
        and
        service_valid
        and
        fqme_valid
    ):
        return None
    return ShmBufferRef(
        service=service,
        generation=generation,
        fqme=fqme,
        kind=kind,
        path=path,
    )
 def resolve_shm_buffer_dtype(
    ref: ShmBufferRef,
    size: int,
 ) -> np.dtype:
    '''
    Resolve a buffer dtype from its size and backend declarations.
    '''
    canonical: np.dtype = np.dtype(def_iohlcv_fields)
    provider_candidates: list[np.dtype] = []
    brokername: str = ref.fqme.rpartition('.')[2]
    try:
        brokermod: ModuleType|None = get_brokermod(brokername)
    except ModuleNotFoundError:
        brokermod = None
    if brokermod is not None:
        provider_dtype = getattr(brokermod, '_ohlc_dtype', None)
        if provider_dtype is not None:
            provider_candidates.append(np.dtype(provider_dtype))
    candidates: list[np.dtype] = (
        provider_candidates
        if provider_candidates
        else [canonical]
    )
    actual_bytes: int = ref.path.stat().st_size
    matches: dict[tuple, np.dtype] = {
        tuple(candidate.descr): candidate
        for candidate in candidates
        if candidate.itemsize * size == actual_bytes
    }
    if len(matches) != 1:
        descriptions: list[str] = [
            f'{candidate.itemsize}B:{candidate.names}'
            for candidate in candidates
        ]
        raise ValueError(
            f'Can not resolve dtype for {ref.path.name}: '
            f'{actual_bytes} bytes over {size} rows; '
            f'candidates={descriptions}'
        )
    return next(iter(matches.values()))
 def iter_shm_buffer_refs(
    fqme: str|None = None,
    shmdir: Path = Path('/dev/shm/'),
 ) -> Generator[ShmBufferRef, None, None]:
    '''
    Yield exact Piker OHLCV SHM identities in stable order.
    '''
    if not shmdir.is_dir():
        return
    for path in sorted(
        shmdir.iterdir(),
        key=lambda candidate: candidate.name,
    ):
        if (
            path.is_symlink()
            or
            not path.is_file()
        ):
            continue
        ref: ShmBufferRef|None = parse_shm_buffer_path(path)
        if (
            ref is not None
            and
            (
                fqme is None
                or
                ref.fqme == fqme
            )
        ):
            yield ref
 def iter_dfs_from_shms(
-    fqme: str
+    fqme: str,
    shm_name: str|None = None,
    shmdir: Path = Path('/dev/shm/'),
    refs: list[ShmBufferRef]|None = None,
 ) -> Generator[
    tuple[Path, ShmArray, pl.DataFrame],
    None,
@ -1694,46 +1903,48 @@ def iter_dfs_from_shms(
        'rt': _default_rt_size,
    }
-    # load all detected shm buffer files which have the
+    selected_refs: list[ShmBufferRef] = (
-    # passed FQME pattern in the file name.
+        refs
-    shmfiles: list[Path] = []
+        if refs is not None
-    shmdir = Path('/dev/shm/')
+        else [
-
+            candidate
-    for shmfile in shmdir.glob(f'*{fqme}*'):
+            for candidate in iter_shm_buffer_refs(
-        filename: str = shmfile.name
+                fqme=fqme,
-
+                shmdir=shmdir,
-        # skip index files
+            )
-        if (
+            if (
-            '_first' in filename
+                shm_name is None
-            or '_last' in filename
+                or
-        ):
+                candidate.path.name == shm_name
-            continue
+            )
-
+        ]
-        assert shmfile.is_file()
+    )
-        log.debug(f'Found matching shm buffer file: {filename}')
+    for ref in selected_refs:
-        shmfiles.append(shmfile)
+        shmfile: Path = ref.path
-
+        log.debug(f'Found matching shm buffer file: {shmfile.name}')
    for shmfile in shmfiles:
        # lookup array buffer size based on file suffix
        # being either .rt or .hist
-        key: str = shmfile.name.rsplit('.')[-1]
+        size: int = sizes[ref.kind]
        # skip FSP buffers for now..
        if key not in sizes:
            continue
        size: int = sizes[key]
        # attach to any shm buffer, load array into polars df,
        # write to local parquet file.
-        shm, opened = maybe_open_shm_array(
+        try:
-            key=shmfile.name,
+            dtype: np.dtype = resolve_shm_buffer_dtype(ref, size)
-            size=size,
+            shm: ShmArray = attach_existing_shm_array(
-            dtype=def_iohlcv_fields,
+                key=shmfile.name,
-            readonly=True,
+                size=size,
-        )
+                dtype=dtype,
-        assert not opened
+                readonly=True,
            )
        except (
            FileNotFoundError,
            ValueError,
        ) as err:
            log.warning(
                f'Could not attach SHM {shmfile.name}: {err}'
            )
            continue
        ohlcv: np.ndarray = shm.array
        df: pl.DataFrame = np2pl(ohlcv)
--- a/snippets/claude_debug_helper.py
+++ b/snippets/claude_debug_helper.py
@ -59,7 +59,10 @@ def expect(
 def run_pdb_commands(
    commands: list[str],
-    initial_cmd: str = 'piker store ldshm xmrusdt.usdtm.perp.binance',
+    initial_cmd: str = (
        'piker store shm xmrusdt.usdtm.perp.binance '
        '--write-parquet'
    ),
    timeout: int = 30,
    print_output: bool = True,
 ) -> dict[str, str]:
@ -145,7 +148,10 @@ class InteractivePdbSession:
    '''
    def __init__(
        self,
-        cmd: str = 'piker store ldshm xmrusdt.usdtm.perp.binance',
+        cmd: str = (
            'piker store shm xmrusdt.usdtm.perp.binance '
            '--write-parquet'
        ),
        timeout: int = 30,
    ):
        self.cmd: str = cmd
--- a/tests/test_history_backfill.py
+++ b/tests/test_history_backfill.py
@ -2,6 +2,7 @@
 Deterministic history-backfill regressions.
 '''
 from contextlib import asynccontextmanager
 from functools import partial
 from pathlib import Path
 from types import SimpleNamespace
@ -34,6 +35,7 @@ from piker.tsp._history import (
    notify_backfill,
    publish_latest_frame,
    start_backfill,
    tsdb_backfill,
 )
@ -216,6 +218,197 @@ def test_latest_frame_is_persisted_before_shm() -> None:
    assert events == ['storage', 'shm']
 def test_tsdb_prepend_uses_actual_provider_boundary(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    '''
    Sparse venue history must not reserve wall-clock rows in SHM.
    NVDA's 60s startup converted elapsed calendar minutes between
    NativeDB and the latest IB frame into physical SHM offsets. IB
    returned only trading bars, leaving 42,720 zero rows even though
    both valid margins ended at the same timestamp. Model completed
    reverse retrieval with a provider boundary overlapping NativeDB.
    Exercise real ``ShmArray.push()`` indexing and prove the stored,
    reverse-provider, and latest frames form one contiguous lossless
    sequence without an explicit wall-clock offset.
    '''
    stored = np.zeros(
        3,
        dtype=np.dtype(def_iohlcv_fields),
    )
    stored['time'] = [60, 120, 180]
    older_reverse = np.zeros(
        3,
        dtype=np.dtype(def_iohlcv_fields),
    )
    older_reverse['time'] = [180, 240, 300]
    newer_reverse = np.zeros(
        3,
        dtype=np.dtype(def_iohlcv_fields),
    )
    newer_reverse['time'] = [300, 360, 420]
    latest = np.zeros(
        2,
        dtype=np.dtype(def_iohlcv_fields),
    )
    latest['time'] = [420, 480]
    storage_frames: list[np.ndarray] = []
    notifications: list[dict] = []
    startup_done = trio.Event()
    class Storage:
        '''
        Capture the complete provider delta written to NativeDB.
        '''
        async def update_ohlcv(
            self,
            fqme: str,
            ohlcv: np.ndarray,
            timeframe: int,
        ) -> None:
            '''
            Record one reverse provider response.
            '''
            storage_frames.append(ohlcv.copy())
        async def load(
            self,
            fqme: str,
            timeframe: int,
        ) -> tuple:
            '''
            Return overlapping NativeDB history and its boundaries.
            '''
            return (
                stored,
                from_timestamp(60),
                from_timestamp(180),
            )
    class Sampler:
        '''
        Accept deterministic backfill notifications.
        '''
        async def send(self, msg) -> None:
            '''
            Capture notification without actor IPC.
            '''
            notifications.append(msg)
    mkt = SimpleNamespace(
        fqme='nvda.nasdaq.ib',
        dst=SimpleNamespace(atype='stock'),
        src=SimpleNamespace(atype='fiat'),
        get_fqme=lambda **kwargs: 'nvda.nasdaq.ib',
    )
    async def get_hist(*args, **kwargs):
        '''
        Return inclusive reverse frames overlapping each SHM
        boundary.
        '''
        end_dt = kwargs['end_dt']
        if end_dt is None:
            return (
                latest,
                from_timestamp(420),
                from_timestamp(480),
            )
        end_t: float = end_dt.timestamp()
        reverse: np.ndarray = (
            newer_reverse
            if end_t == 420
            else older_reverse
        )
        return (
            reverse,
            from_timestamp(reverse['time'][0]),
            from_timestamp(reverse['time'][-1]),
        )
    @asynccontextmanager
    async def open_history_client(mkt):
        '''
        Yield the deterministic provider and default frame config.
        '''
        yield get_hist, {}
    async def finish_null_repair(**kwargs) -> None:
        '''
        Signal that provider and NativeDB prepends both completed.
        '''
        startup_done.set()
    monkeypatch.setattr(
        'piker.tsp._history.maybe_fill_null_segments',
        finish_null_repair,
    )
    async def main() -> None:
        shm_key = f'test_sparse_prepend_{uuid4().hex}'
        async with tractor.open_root_actor(
            name=shm_key,
            tpt_bind_addrs=[('127.0.0.1', 0)],
        ):
            shm, opened = maybe_open_shm_array(
                key=shm_key,
                size=16,
                dtype=np.dtype(def_iohlcv_fields),
                append_start_index=12,
            )
            assert opened
            storage = Storage()
            async with trio.open_nursery() as nursery:
                nursery.start_soon(partial(
                    tsdb_backfill,
                    mod=SimpleNamespace(
                        name='fake',
                        open_history_client=open_history_client,
                    ),
                    storemod=SimpleNamespace(
                        ohlc_key_map=ohlc_key_map,
                    ),
                    storage=storage,
                    mkt=mkt,
                    shm=shm,
                    timeframe=60,
                    sampler_stream=Sampler(),
                ))
                with trio.fail_after(0.5):
                    await startup_done.wait()
                nursery.cancel_scope.cancel()
            assert shm.array['time'].tolist() == [
                60,
                120,
                180,
                240,
                300,
                360,
                420,
                480,
            ]
            assert not np.any(shm.array['time'] <= 0)
    trio.run(main)
    assert len(storage_frames) == 3
    assert storage_frames[0]['time'].tolist() == [420, 480]
    assert storage_frames[1]['time'].tolist() == [300, 360, 420]
    assert storage_frames[2]['time'].tolist() == [180, 240, 300]
    assert len(notifications) == 3
 def test_ib_latest_frame_round_trips_through_nativedb(
    tmp_path: Path,
 ) -> None:
--- a/tests/test_ib_method_proxy.py
+++ b/tests/test_ib_method_proxy.py
@ -0,0 +1,271 @@
 '''
 IB asyncio method-proxy regressions.
 '''
 from types import SimpleNamespace
 from typing import Any
 import pytest
 import trio
 from piker.brokers.ib import api as ib_api
 from piker.brokers.ib.api import (
    MethodProxy,
    relay_client_proxy_messages,
 )
 class FakeChannel:
    '''
    Model the Trio side of an asyncio linked-task channel.
    '''
    def __init__(self) -> None:
        self.requests: list[tuple] = []
        (
            self._tx,
            self._rx,
        ) = trio.open_memory_channel[Any](10)
    async def send(self, msg: tuple) -> None:
        '''
        Capture one Trio-to-asyncio method request.
        '''
        self.requests.append(msg)
    async def receive(self) -> Any:
        '''
        Receive one simulated asyncio-to-Trio response.
        '''
        return await self._rx.receive()
    def closed(self) -> bool:
        '''
        Report an open channel for the lifetime of each test.
        '''
        return False
    async def respond(self, msg: Any) -> None:
        '''
        Send one response or status event to the proxy relay.
        '''
        await self._tx.send(msg)
    async def close_responses(self) -> None:
        '''
        Close the simulated asyncio-to-Trio response channel.
        '''
        await self._tx.aclose()
 def test_method_proxy_routes_after_idle_status_events() -> None:
    '''
    Idle IB status traffic must not break first-symbol qualification.
    ``open_client_proxy()`` previously broadcast a one-slot channel
    to an always-running status relay and an idle method receiver.
    Status events advanced only the relay, so a first order for an
    uncached symbol raised ``Lagged`` before receiving
    ``get_sym_details()``. Feed several idle events through the
    sole-reader relay, then start two method calls and return their
    responses in reverse order. Both exact results prove status
    traffic cannot overrun a dormant caller and concurrent responses
    remain correlated by request ID.
    '''
    async def main() -> None:
        chan = FakeChannel()
        proxy = MethodProxy(
            chan,
            event_table={},
            asyncio_ns=SimpleNamespace(),
        )
        results: dict[str, str] = {}
        async def call(name: str) -> None:
            results[name] = await proxy._run_method(meth=name)
        async with trio.open_nursery() as nursery:
            nursery.start_soon(
                relay_client_proxy_messages,
                chan,
                proxy,
            )
            for i in range(3):
                await chan.respond((
                    'event',
                    {'reason': f'farm-status-{i}'},
                ))
            nursery.start_soon(call, 'first')
            nursery.start_soon(call, 'second')
            with trio.fail_after(0.1):
                while len(chan.requests) < 2:
                    await trio.lowlevel.checkpoint()
            mids: dict[str, int] = {
                meth: mid
                for meth, _kwargs, mid in chan.requests
            }
            await chan.respond({
                'mid': mids['second'],
                'result': 'second-result',
            })
            await chan.respond({
                'mid': mids['first'],
                'result': 'first-result',
            })
            with trio.fail_after(0.1):
                while len(results) < 2:
                    await trio.lowlevel.checkpoint()
            nursery.cancel_scope.cancel()
        assert results == {
            'first': 'first-result',
            'second': 'second-result',
        }
    trio.run(main)
 def test_method_proxy_relay_accepts_clean_eof() -> None:
    '''
    Normal asyncio relay completion must not fail proxy teardown.
    The sole-reader relay blocks in ``LinkedTaskChannel.receive()``
    until the asyncio task exits and closes its Trio send channel.
    Close the fake response sender before entering the relay and
    prove the resulting ``EndOfChannel`` is graceful completion.
    '''
    async def main() -> None:
        chan = FakeChannel()
        proxy = MethodProxy(
            chan,
            event_table={},
            asyncio_ns=SimpleNamespace(),
        )
        await chan.close_responses()
        with trio.fail_after(0.1):
            await relay_client_proxy_messages(chan, proxy)
    trio.run(main)
 def test_method_proxy_drops_cancelled_call_response() -> None:
    '''
    A cancelled symbol search must not skew the next method response.
    Search timeouts can cancel a Trio caller after its request
    reaches the asyncio client. Wait until one request is sent,
    cancel its caller, and then deliver the orphaned response. The
    sole-reader relay must drop that stale ``mid`` without failing; a
    subsequent call must still receive its own exact response.
    '''
    async def main() -> None:
        chan = FakeChannel()
        proxy = MethodProxy(
            chan,
            event_table={},
            asyncio_ns=SimpleNamespace(),
        )
        call_scope = trio.CancelScope()
        cancelled = trio.Event()
        result: list[str] = []
        async def cancelled_call() -> None:
            with call_scope:
                try:
                    await proxy._run_method(meth='cancelled')
                finally:
                    cancelled.set()
        async def next_call() -> None:
            value = await proxy._run_method(meth='next')
            result.append(value)
        async with trio.open_nursery() as nursery:
            nursery.start_soon(
                relay_client_proxy_messages,
                chan,
                proxy,
            )
            nursery.start_soon(cancelled_call)
            with trio.fail_after(0.1):
                while len(chan.requests) < 1:
                    await trio.lowlevel.checkpoint()
            cancelled_mid: int = chan.requests[0][2]
            call_scope.cancel()
            with trio.fail_after(0.1):
                await cancelled.wait()
            assert proxy._pending == {}
            assert (
                proxy._request_methods[cancelled_mid]
                ==
                'cancelled'
            )
            await chan.respond({
                'mid': cancelled_mid,
                'result': 'stale',
            })
            with trio.fail_after(0.1):
                while cancelled_mid in proxy._request_methods:
                    await trio.lowlevel.checkpoint()
            nursery.start_soon(next_call)
            with trio.fail_after(0.1):
                while len(chan.requests) < 2:
                    await trio.lowlevel.checkpoint()
            next_mid: int = chan.requests[1][2]
            await chan.respond({
                'mid': next_mid,
                'result': 'fresh',
            })
            with trio.fail_after(0.1):
                while not result:
                    await trio.lowlevel.checkpoint()
            nursery.cancel_scope.cancel()
        assert result == ['fresh']
    trio.run(main)
 def test_method_proxy_warns_when_metadata_is_evicted(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    '''
    Unresolved metadata eviction must remain operator-visible.
    Cancelled calls whose asyncio responses never arrive retain a
    bounded method name solely for later stale-response diagnostics.
    Fill that table beyond its limit and prove the oldest entry is
    evicted with its request ID, method name, and configured bound in
    the warning.
    '''
    proxy = MethodProxy(
        FakeChannel(),
        event_table={},
        asyncio_ns=SimpleNamespace(),
    )
    warnings: list[str] = []
    monkeypatch.setattr(ib_api.log, 'warning', warnings.append)
    for mid in range(257):
        proxy._track_request_method(mid, f'method-{mid}')
    assert len(proxy._request_methods) == 256
    assert 0 not in proxy._request_methods
    assert warnings == [
        'Evicting unresolved method-request metadata,\n'
        'mid: 0\n'
        "meth: 'method-0'\n"
        'pending metadata limit: 256\n'
    ]
--- a/tests/test_ldshm.py
+++ b/tests/test_ldshm.py
@ -16,19 +16,21 @@ import polars as pl
 import pytest
 from piker import tsp
 from piker.data import def_iohlcv_fields
 from piker.storage import cli as storage_cli
 from piker.storage.cli import (
    _shm_period_and_invalid_count,
-    ldshm,
+    _summarize_shm_frame,
    shm as shm_cmd,
 )
-def test_ldshm_detects_nulls_without_skewing_period() -> None:
+def test_shm_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
+    ``shm`` 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
@ -52,7 +54,7 @@ def test_ldshm_detects_nulls_without_skewing_period() -> None:
    assert invalid_count == 2
-def test_ldshm_period_is_an_observed_step() -> None:
+def test_shm_period_is_an_observed_step() -> None:
    '''
    Sparse timestamps must not invent a storage timeframe.
@ -71,13 +73,13 @@ def test_ldshm_period_is_an_observed_step() -> None:
    assert invalid_count == 0
-def test_ldshm_counts_every_invalid_timestamp() -> None:
+def test_shm_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``.
+    but allowing a replacement attempt would still abort ``shm``.
    Mix each invalid class into a regular series and prove the same
    snapshot inspection counts all four while preserving cadence.
@ -98,7 +100,7 @@ def test_ldshm_counts_every_invalid_timestamp() -> None:
    assert invalid_count == 4
-def test_ldshm_invalid_snapshot_never_reaches_storage(
+def test_shm_invalid_snapshot_never_reaches_storage(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    '''
@ -206,7 +208,156 @@ def test_ldshm_invalid_snapshot_never_reaches_storage(
        open_annotations,
    )
-    ldshm('qqq.nasdaq.ib')
+    shm_cmd(
        'qqq.nasdaq.ib',
        write_parquet=True,
    )
 def test_shm_write_without_reload_uses_deduped_markup_frame(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    '''
    A Parquet write without SHM reload must not need a reload
    payload.
    ``ldshm`` created ``new`` only in reload but used it for markup.
    Its no-reload mode crashed after a durable write. Arrange one
    gap, explicitly enable persistence with reload disabled, and
    prove markup receives the deduplicated frame without any SHM
    mutation.
    '''
    array = np.zeros(
        3,
        dtype=np.dtype(def_iohlcv_fields),
    )
    array['index'] = np.arange(3)
    array['time'] = [60, 120, 240]
    array['open'] = [10, 11, 12]
    array['high'] = [12, 13, 14]
    array['low'] = [8, 9, 10]
    array['close'] = [11, 12, 13]
    array['volume'] = [100, 101, 102]
    frame = tsp.np2pl(array)
    shm = SimpleNamespace(array=array)
    shmfile = Path('piker.datad[aaaaaaaa-aaaa-aa].x.test.hist')
    writes: list[pl.DataFrame] = []
    markups: list[pl.DataFrame] = []
    @asynccontextmanager
    async def open_runtime(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[None]:
        '''
        Yield a runtime-free command context.
        '''
        yield
    class Client:
        '''
        Record optional durable writes.
        '''
        async def write_ohlcv(
            self,
            fqme: str,
            ohlcv: pl.DataFrame,
            timeframe: int,
        ) -> Path:
            '''
            Persist the frame for the command's read-back step.
            '''
            writes.append(ohlcv)
            path = tmp_path / 'frame.parquet'
            ohlcv.write_parquet(path)
            return path
    @asynccontextmanager
    async def open_storage(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[tuple[SimpleNamespace, Client]]:
        '''
        Yield the recording storage client.
        '''
        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 one valid history snapshot.
        '''
        yield shmfile, shm, frame
    async def markup_gaps(
        fqme: str,
        period_s: int,
        actl: SimpleNamespace,
        new_df: pl.DataFrame,
        step_gaps: pl.DataFrame,
    ) -> dict:
        '''
        Capture the frame used for non-mutating markup.
        '''
        markups.append(new_df)
        return {}
    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._annotate, 'markup_gaps', markup_gaps)
    monkeypatch.setattr(storage_cli.tractor, 'pause', pause)
    from piker.ui import _remote_ctl
    monkeypatch.setattr(
        _remote_ctl,
        'open_annot_ctl',
        open_annotations,
    )
    shm_cmd(
        'x.test',
        write_parquet=True,
        reload_parquet_to_shm=False,
    )
    assert len(writes) == 1
    assert len(markups) == 1
    assert markups[0]['time'].to_list() == [60, 120, 240]
@pytest.mark.parametrize(
@ -217,7 +368,7 @@ def test_ldshm_invalid_snapshot_never_reaches_storage(
        np.array([60, 60]),
    ],
 )
-def test_ldshm_skips_frame_without_positive_cadence(
+def test_shm_skips_frame_without_positive_cadence(
    times: np.ndarray,
 ) -> None:
    '''
@ -227,9 +378,182 @@ def test_ldshm_skips_frame_without_positive_cadence(
    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.
+    returns no cadence, which directs ``shm`` to skip the buffer.
    '''
    period_s, _ = _shm_period_and_invalid_count(times)
    assert period_s is None
 def test_shm_buffer_discovery_uses_exact_identities(
    tmp_path: Path,
 ) -> None:
    '''
    SHM discovery must not mix substring matches or index companions.
    ``shm`` globbed ``*fqme*`` and could process stale and
    active objects for another instrument whose name merely contained
    requested text. Create two generations, both OHLCV kinds, index
    companions, and a substring collision. Prove discovery returns
    exact parsed identities in deterministic filename order.
    '''
    names = [
        'piker.datad[bbbbbbbb-bbbb-bb].qqq.nasdaq.ib.rt',
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist',
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist_first',
        'piker.datad[aaaaaaaa-aaaa-aa].myqqq.nasdaq.ib.rt',
        'unrelated.qqq.nasdaq.ib.rt',
    ]
    for name in names:
        (tmp_path / name).touch()
    refs = list(tsp.iter_shm_buffer_refs(
        fqme='qqq.nasdaq.ib',
        shmdir=tmp_path,
    ))
    assert [ref.path.name for ref in refs] == [
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist',
        'piker.datad[bbbbbbbb-bbbb-bb].qqq.nasdaq.ib.rt',
    ]
    assert [
        (ref.service, ref.generation, ref.kind)
        for ref in refs
    ] == [
        ('datad', 'aaaaaaaa-aaaa-aa', 'hist'),
        ('datad', 'bbbbbbbb-bbbb-bb', 'rt'),
    ]
 def test_shm_summary_reports_invalid_rows_and_ranked_gaps() -> None:
    '''
    Read-only SHM diagnostics must preserve the live triage evidence.
    The QQQ investigation needed cadence, invalid bounds, and largest
    timestamp jumps without mutation. Build one zero row and two gaps;
    prove summary excludes invalid timestamps and ranks durations.
    '''
    frame = np.zeros(
        5,
        dtype=np.dtype(def_iohlcv_fields),
    )
    frame['index'] = np.arange(100, 105)
    frame['time'] = [1, 0, 3, 4, 10]
    ref = tsp.ShmBufferRef(
        service='datad',
        generation='aaaaaaaa-aaaa-aa',
        fqme='qqq.nasdaq.ib',
        kind='rt',
        path=Path(
            'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt'
        ),
    )
    report = _summarize_shm_frame(ref, frame, max_gaps=2)
    assert report['period_s'] == 1
    assert report['observed_period_s'] == 1
    assert report['invalid_count'] == 1
    assert report['invalid_index_bounds'] == [101, 101]
    assert report['gap_count'] == 2
    assert [
        gap['delta_s']
        for gap in report['largest_gaps']
    ] == [6, 2]
 def test_shm_summary_reports_ordering_corruption() -> None:
    '''
    Duplicate and reversed timestamps remain visible in diagnostics.
    Positive-gap detection considers both defects gap-free. Build
    one exact RT snapshot with a duplicate and reversal and prove the
    read-only report distinguishes both without classifying values as
    unpublished slots.
    '''
    frame = np.zeros(
        5,
        dtype=np.dtype(def_iohlcv_fields),
    )
    frame['index'] = np.arange(5)
    frame['time'] = [1, 2, 2, 1, 3]
    ref = tsp.ShmBufferRef(
        service='datad',
        generation='aaaaaaaa-aaaa-aa',
        fqme='qqq.nasdaq.ib',
        kind='rt',
        path=Path(
            'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt'
        ),
    )
    report = _summarize_shm_frame(ref, frame, max_gaps=2)
    assert report['invalid_count'] == 0
    assert report['duplicate_step_count'] == 1
    assert report['reversed_step_count'] == 1
 def test_shm_dtype_resolution_matches_allocator_layout(
    tmp_path: Path,
 ) -> None:
    '''
    Inspector actors must mirror history allocation dtype lookup.
    Allocation reads ``_ohlc_dtype`` from the package; IB does not
    export its API-internal extended dtype there.
    Create four production-shaped canonical rows and prove inspection
    resolves the same layout without searching unrelated submodules.
    '''
    path = tmp_path / (
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt'
    )
    path.write_bytes(b'\0' * (4 * 56))
    ref = tsp.ShmBufferRef(
        service='datad',
        generation='aaaaaaaa-aaaa-aa',
        fqme='qqq.nasdaq.ib',
        kind='rt',
        path=path,
    )
    dtype = tsp.resolve_shm_buffer_dtype(ref, size=4)
    assert dtype.itemsize == 56
    assert dtype.names == (
        'index',
        'time',
        'open',
        'high',
        'low',
        'close',
        'volume',
    )
@pytest.mark.parametrize(
    'name',
    [
        'piker..datad[aaaaaaaa-aaaa-aa].x.test.rt',
        'piker.datad[short].x.test.rt',
        'piker.datad[aaaaaaaa-aaaa-aa].x\\test.rt',
        'piker.datad[aaaaaaaa-aaaa-aa].x[y].test.rt',
    ],
 )
 def test_shm_parser_rejects_forged_identities(
    name: str,
 ) -> None:
    '''
    SHM selectors must accept only generated Piker identities.
    Reject malformed service, generation, and FQME components so a
    same-user object that merely resembles OHLCV can not become an
    ``shm --shm-name`` target.
    '''
    assert tsp.parse_shm_buffer_path(Path(name)) is None
--- a/tests/test_storage_nativedb.py
+++ b/tests/test_storage_nativedb.py
@ -12,7 +12,11 @@ import trio
 from piker import tsp
 from piker.data._source import def_iohlcv_fields
-from piker.storage.nativedb import NativeStorageClient
+from piker.storage.nativedb import (
    iter_native_series,
    NativeStorageClient,
    parse_ohlcv_parquet_path,
 )
 def mk_ohlcv(
@ -540,8 +544,96 @@ def test_index_files_ignores_sidecar_files(
    index = client.index_files()
-    assert list(index) == ['x.test']
+    assert list(index) == [('x.test', 60)]
-    assert index['x.test']['period'] == 60
+    ref = index[('x.test', 60)]
    assert ref.fqme == 'x.test'
    assert ref.period_s == 60
 def test_index_files_preserves_every_series_timeframe(
    tmp_path: Path,
 ) -> None:
    '''
    NativeDB discovery must identify FQME and timeframe together.
    The old index keyed only by FQME, so filesystem iteration silently
    discarded either the 1s or 60s file. Write both periods plus another
    instrument, rebuild the index, and prove exact series remain stable
    while compatibility key listing still returns unique FQMEs.
    '''
    client = NativeStorageClient(tmp_path)
    run(client.write_ohlcv('x.test', mk_ohlcv((1, 2)), 1))
    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
    run(client.write_ohlcv('y.test', mk_ohlcv((60,)), 60))
    index = client.index_files()
    keys = trio.run(client.list_keys)
    series = trio.run(client.list_series)
    assert list(index) == [
        ('x.test', 1),
        ('x.test', 60),
        ('y.test', 60),
    ]
    assert keys == ['x.test', 'y.test']
    assert [
        (ref.fqme, ref.period_s)
        for ref in series
    ] == list(index)
@pytest.mark.parametrize(
    'name',
    [
        'x.test.ohlcv0s.parquet',
        'x.test.ohlcv01s.parquet',
        'x.test.ohlcv²s.parquet',
        'x.test.ohlcv1.parquet',
        'x.test.ohlcv1s.parquet.tmp',
        'x.test.parquet',
        '.ohlcv1s.parquet',
    ],
 )
 def test_native_series_parser_rejects_noncanonical_names(
    tmp_path: Path,
    name: str,
 ) -> None:
    '''
    Malformed files must not become selectable storage identities.
    NativeDB directories can contain crash files, legacy sidecars, and
    unrelated Parquet. Exercise ambiguous period and suffix forms and
    prove parser and directory discovery ignore them deterministically.
    '''
    path: Path = tmp_path / name
    path.touch()
    assert parse_ohlcv_parquet_path(path) is None
    assert list(iter_native_series(tmp_path)) == []
 def test_native_series_parser_round_trips_dotted_fqme(
    tmp_path: Path,
 ) -> None:
    '''
    Dotted market identities must survive canonical filename parsing.
    The legacy parser split on every dot and failed for ordinary FQMEs.
    Parse a futures-style identity and prove every selector field is
    preserved exactly.
    '''
    path = tmp_path / 'mnq.cme.20260918.ib.ohlcv1s.parquet'
    path.touch()
    ref = parse_ohlcv_parquet_path(path)
    assert ref is not None
    assert ref.fqme == 'mnq.cme.20260918.ib'
    assert ref.period_s == 1
    assert ref.path == path
 def test_writes_create_no_lock_sidecars(
--- a/tests/test_store_cli.py
+++ b/tests/test_store_cli.py
@ -0,0 +1,326 @@
 '''
 Storage command UX regressions.
 '''
 from collections.abc import (
    AsyncIterator,
    Iterator,
 )
 from contextlib import asynccontextmanager
 import json
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
 import numpy as np
 import pytest
 from typer.testing import CliRunner
 from piker import tsp
 from piker import config
 from piker.data import def_iohlcv_fields
 from piker.storage import cli as storage_cli
 from piker.storage.cli import store
 def test_store_group_shows_help_without_arguments() -> None:
    '''
    Bare ``piker store`` must present its command map immediately.
    The Typer group emitted only a missing-command error and
    required another ``--help`` invocation. Invoke the group
    without arguments and prove complete help replaces that error.
    '''
    result = CliRunner().invoke(store, [])
    assert result.exit_code == 2
    assert 'Usage:' in result.output
    assert 'Missing command' not in result.output
    for command in (
        'anal',
        'audit',
        'delete',
        'ls',
        'series',
        'shm',
    ):
        assert command in result.output
 def test_store_commands_show_help_without_arguments() -> None:
    '''
    Bare endpoints must not open runtimes or report missing args.
    Every endpoint is discoverable by typing its name once. Exercise
    required-input commands and prove help rendering exits before any
    callback can run.
    '''
    runner = CliRunner()
    for command in (
        'anal',
        'audit',
        'delete',
        'shm',
    ):
        result = runner.invoke(store, [command])
        assert result.exit_code == 2
        assert 'Usage:' in result.output
        assert 'Missing argument' not in result.output
 def test_series_lists_exact_native_periods(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    '''
    Durable listing must expose separate 1s and 60s identities.
    The old FQME-only index hid one timeframe. Create canonical files
    without storage or actor services, invoke the JSON endpoint,
    and prove both exact paths and periods are returned.
    '''
    nativedb: Path = tmp_path / 'nativedb'
    nativedb.mkdir()
    one = nativedb / 'qqq.nasdaq.ib.ohlcv1s.parquet'
    sixty = nativedb / 'qqq.nasdaq.ib.ohlcv60s.parquet'
    three_hundred = nativedb / (
        'qqq.nasdaq.ib.ohlcv300s.parquet'
    )
    one.write_bytes(b'1')
    sixty.write_bytes(b'60')
    three_hundred.write_bytes(b'300')
    monkeypatch.setattr(config, 'get_conf_dir', lambda: tmp_path)
    result = CliRunner().invoke(
        store,
        ['series', '--json'],
    )
    assert result.exit_code == 0
    payload = json.loads(result.output)
    assert [item['period_s'] for item in payload] == [1, 60, 300]
    assert [item['path'] for item in payload] == [
        str(one),
        str(sixty),
        str(three_hundred),
    ]
    compact = CliRunner().invoke(store, ['ls'])
    assert compact.exit_code == 0
    assert 'qqq.nasdaq.ib' in compact.output
    assert '1s, 60s, 300s' in compact.output
 def test_shm_reload_requires_explicit_persistence() -> None:
    '''
    SHM reload must not silently opt into durable persistence.
    Reload reads back the Parquet produced by the repair path, so the
    reload flag alone has no valid input and must not open runtime or
    storage services. Invoke that invalid combination and prove CLI
    validation rejects it before inspecting an FQME.
    '''
    result = CliRunner().invoke(
        store,
        ['shm', 'qqq.nasdaq.ib', '--reload-parquet-to-shm'],
    )
    assert result.exit_code == 2
    assert 'requires --write-parquet' in result.output
@pytest.mark.parametrize(
    'option',
    [
        ['--json'],
        ['--max-gaps', '10'],
    ],
 )
 def test_shm_repair_rejects_read_only_formatting(
    option: list[str],
 ) -> None:
    '''
    Repair mode must not silently ignore read-only output options.
    Combining the old repair workflow with the new inspector made
    ``--json`` and an explicitly default-valued ``--max-gaps``
    appeared accepted even though repair produced logs and entered an
    interactive pause. Invoke each conflicting option and prove
    validation stops before runtime or storage can be opened.
    '''
    result = CliRunner().invoke(
        store,
        [
            'shm',
            'qqq.nasdaq.ib',
            '--write-parquet',
            *option,
        ],
    )
    assert result.exit_code == 2
    assert 'apply only to read-only output' in result.output
 def test_ls_has_description_in_store_help() -> None:
    '''
    Compact NativeDB discovery must explain itself in group help.
    The backend-oriented ``ls`` callback had no docstring, leaving
    its command-map description blank. Render the parent help and
    prove the compact series purpose is visible without opening the
    command.
    '''
    result = CliRunner().invoke(store, [])
    assert result.exit_code == 2
    assert 'List NativeDB FQMEs' in result.output
 def test_shm_endpoint_reports_immutable_snapshot(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    '''
    SHM diagnostics report exact buffers without storage mutation.
    Build one RT identity and snapshot, then replace runtime
    attachment with fakes. Invoke JSON output and prove it carries
    generation, cadence, and invalid-slot
    evidence without opening a storage client or writer.
    '''
    path = Path(
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt'
    )
    ref = tsp.ShmBufferRef(
        service='datad',
        generation='aaaaaaaa-aaaa-aa',
        fqme='qqq.nasdaq.ib',
        kind='rt',
        path=path,
    )
    frame = np.zeros(
        4,
        dtype=np.dtype(def_iohlcv_fields),
    )
    frame['index'] = np.arange(4)
    frame['time'] = [1, 0, 3, 4]
    @asynccontextmanager
    async def open_runtime(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[None]:
        '''
        Yield a runtime-free diagnostic context.
        '''
        yield
    def iter_refs(
        fqme: str|None = None,
    ) -> Iterator[tsp.ShmBufferRef]:
        '''
        Yield the selected exact SHM identity.
        '''
        yield ref
    def iter_frames(
        fqme: str,
        shm_name: str|None = None,
        refs: list[tsp.ShmBufferRef]|None = None,
    ) -> Iterator[tuple[Path, SimpleNamespace, None]]:
        '''
        Yield an immutable snapshot through the attachment interface.
        '''
        shm = SimpleNamespace(array=frame)
        yield path, shm, None
    def fail_storage(*args: Any, **kwargs: Any) -> None:
        '''
        Fail if read-only SHM inspection reaches storage.
        '''
        raise AssertionError('SHM diagnostics opened storage')
    monkeypatch.setattr(
        storage_cli,
        'open_piker_runtime',
        open_runtime,
    )
    monkeypatch.setattr(
        storage_cli,
        'open_storage_client',
        fail_storage,
    )
    monkeypatch.setattr(tsp, 'iter_shm_buffer_refs', iter_refs)
    monkeypatch.setattr(tsp, 'iter_dfs_from_shms', iter_frames)
    result = CliRunner().invoke(
        store,
        ['shm', 'qqq.nasdaq.ib', '--json'],
    )
    assert result.exit_code == 0
    payload = json.loads(result.output)
    assert len(payload) == 1
    assert payload[0]['generation'] == 'aaaaaaaa-aaaa-aa'
    assert payload[0]['period_s'] == 1
    assert payload[0]['invalid_count'] == 1
 def test_shm_rejects_unknown_exact_name_before_runtime(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    '''
    A stale exact SHM selection must fail before opening services.
    The old no-match path paused after opening runtime and storage
    setup. Return no identity, install a runtime fail-spy,
    and prove exact selection exits immediately
    with a deterministic error.
    '''
    def iter_refs(
        fqme: str|None = None,
    ) -> Iterator[tsp.ShmBufferRef]:
        '''
        Yield no matching SHM identities.
        '''
        return iter(())
    def fail_runtime(*args: Any, **kwargs: Any) -> None:
        '''
        Fail if no-match handling opens a runtime.
        '''
        raise AssertionError('unknown SHM opened runtime')
    monkeypatch.setattr(tsp, 'iter_shm_buffer_refs', iter_refs)
    monkeypatch.setattr(
        storage_cli,
        'open_piker_runtime',
        fail_runtime,
    )
    result = CliRunner().invoke(
        store,
        [
            'shm',
            'qqq.nasdaq.ib',
            '--shm-name',
            'missing',
        ],
    )
    assert result.exit_code == 2
    assert 'No exact OHLCV SHM buffer' in result.output
Author	SHA1	Message	Date
Gud Boi	2aa66c2cf6	Fix IB `MethodProxy` response routing `open_client_proxy()` broadcast its one-slot linked channel to an always-running status relay and an idle method receiver. Status traffic advanced only the relay, so the first lazy contract lookup for an uncached order raised `Lagged` before receiving its response. Give `relay_client_proxy_messages()` sole receive ownership. Route tagged responses from concurrent caller tasks into their per-`mid` slots while each proxy context retains its own channel and relay. Deats, - log stale responses with their original method name; - warn when unresolved-request metadata exceeds its bounded limit; - drop late responses after caller cancellation; - preserve the reconnect design note and message-case rationale; - accept clean channel EOF during proxy teardown; - cover idle status, concurrency, cancellation and EOF paths. Prompt-IO: ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))	2026-08-11 18:46:23 -04:00
Gud Boi	06d5ea5dc0	Fix sparse IB history prepends in `ShmArray` IB returns trading samples, not one row per wall-clock period. Converting elapsed time to a physical prepend offset left 42,720 zero rows in NVDA's 60s buffer even though both valid margins shared the same timestamp. Use the actual earliest provider sample as the NativeDB boundary. After reverse retrieval, prepend older stored rows directly beside it. Deats, - start reverse insertion at the current first index; - update `ShmArray._first` after every provider prepend; - trim provider/NativeDB overlap by its timestamp; - publish the final storage prepend to chart and FSP consumers; - derive the source `time` field from `ohlc_key_map`; - cover the full lifecycle with a real multi-frame `ShmArray`. Prompt-IO: ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))	2026-08-05 19:40:07 -04:00
Gud Boi	f0f3c80657	Simplify `store` series and SHM discovery Make `store ls` show every NativeDB FQME and its available sample periods by default. Keep `store series` as the detailed view of each exact sub-series, including its period, size and path. Share the same optional FQME filter and numerically ordered payload. Use `store shm` as the canonical volatile-buffer ep and make its default operation a read-only report. Enter the prior persistence workflow only through `--write-parquet`, require that flag before `--reload-parquet-to-shm`. Reject read-only formatting options in repair mode instead of silently ignoring them. Also, - update the qualification docs, debug helper and harness refs; - cover no-flag listings, command help and SHM mode validation; - retain exact generation and object-name selection for stale SHM. Prompt-IO: ai/prompt-io/opencode/20260801T015321Z_74ea2152_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))	2026-08-01 00:34:55 -04:00
Gud Boi	74ea2152f9	Expose storage state through `piker store` Make bare storage groups and eps render help, so the CLI is discoverable without memorizing `Typer` option conventions. Add read-only `series` and `shm` eps for durable `NativeDB` series and live history buffers, with `Rich`-table or `JSON` output. Deats, - preserve exact `(fqme, period)` identities for `NativeDB` series; - parse exact `(generation, fqme, kind)` names for SHM buffers; - reconstruct provider dtypes when attaching to live buffers; - tolerate buffers disappearing between discovery and attach. Also, - report expected and observed cadence separately, plus invalid rows, ordering defects and ranked gaps; - add exact `ldshm --shm-name` selection and keep disabled write and reload paths from reading undefined state or entering `pdb`; - document the triage layers and cover help, identity parsing, read-only diagnostics, dtype selection and `ldshm` flow. Prompt-IO: ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))	2026-07-31 17:26:11 -04:00