Plan known-FQME backfill qualification

- preserve dirty main-tree work while transferring branch ownership
- hold provider, FQME, timeframe, and symptom fixed while debugging
- capture read-only baselines before disposable live runs
- turn each reproduced failure into deterministic coverage
- qualify interruption and restart behavior after the normal lifecycle

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

.claude/skills/commit-msg/SKILL.md

@@ -1,119 +0,0 @@
----
-name: commit-msg
-description: >
-  Generate piker-style git commit messages from staged
-  changes or prompt input. Use when the user asks for a
-  commit message or invokes a commit-message command.
-compatibility: >
-  Requires git and a coding harness able to read and
-  write files in the active checkout.
-metadata:
-  author: goodboy
-  version: "1.0"
----
-
-# Piker Git Commit Message Generator
-
-Generate a commit message from the staged diff following
-the piker project's conventions, learned from 500 repo
-commits.
-
-For the full style guide with verb frequencies,
-section markers, abbreviations, piker-specific terms,
-and examples, see
-[style-guide-reference.md](./style-guide-reference.md).
-
-## Safety
-
-- Never run `git commit`, amend, rebase, merge, or push
-  unless the user explicitly requests that operation.
-- Do not stage or unstage files for a single-message run.
-- Never change issue, plan, or checklist state merely
-  because the related implementation is complete.
-- Preserve unrelated worktree changes.
-
-## Working Context
-
-1. Run `git rev-parse --show-toplevel`,
-   `git rev-parse --git-dir`, and
-   `git rev-parse --git-common-dir`.
-2. If the git dir differs from the common dir, report the
-   active worktree before generating the message.
-3. Inspect `git status --short`, the complete staged diff,
-   `git diff --staged --check`, and recent commit history.
-4. If nothing is staged, stop and tell the user. Prompt
-   text may clarify intent but does not replace a staged
-   patch unless the user explicitly asks for a draft with
-   no staged changes.
-
-## Message Style
-
-- **Subject**: ~50 chars, present tense verb, use
-  backticks for code refs; hard maximum 67 columns
-- **Body**: only for complex/multi-file changes,
-  filled close to a 67-column maximum
-- **Section markers**: Also, / Deats, / Other,
-- **Bullets**: use `-` style
-- **Tone**: technical but casual (piker style)
-
-Use any user-supplied scope or description as context,
-but verify it against the staged patch. The message must
-describe the complete staged boundary, not only the most
-obvious file.
-
-## Provenance
-
-If staged paths include prompt-I/O log entries under
-`ai/prompt-io/`, add one trailer per non-raw log file:
-
-```
-Prompt-IO: ai/prompt-io/<service>/<entry>.md
-```
-
-When the patch has substantive AI-authored changes but no
-prompt-I/O entry is staged, remind the user that repository
-policy may require one. Do not block message generation.
-
-Always identify the active harness and model from runtime
-context rather than hardcoding Claude Code or OpenCode.
-
-When the harness assisted with the patch, append:
-
-```
-(this patch was generated in some part by `<harness>` using `<model>` (`<provider>`))
-```
-
-When it generated only the message, append:
-
-```
-(this commit msg was generated in some part by `<harness>` using `<model>` (`<provider>`))
-```
-
-## Output Files
-
-Write the final message to both of these paths relative to
-the active checkout returned by `git rev-parse
---show-toplevel`:
-
-- `.claude/skills/commit-msg/msgs/<timestamp>_<hash>_commit_msg.md`
-- `.claude/git_commit_msg_LATEST.md`
-
-Use UTC `<timestamp>` format `YYYYMMDDTHHMMSSZ` and the
-first seven characters of the current `HEAD` hash. Create
-the archive directory when needed. The `.claude` path is a
-legacy repository artifact location shared by all coding
-harnesses; it does not imply which harness generated the
-message.
-
-Finish by reporting both paths and this review-first
-command without running it:
-
-```
-git commit --edit --file .claude/git_commit_msg_LATEST.md
-```
-
----
-
-**Analysis date:** 2026-01-27
-**Commits analyzed:** 500 from piker repository
-**Maintained by:** Tyler Goodlet

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

@@ -0,0 +1,264 @@
+# Test Harness Reference: piker
+
+This repository-local file supplements the canonical `/run-tests` skill.
+Keep shared execution, worktree, failure-inspection, and cleanup policy in the
+canonical `SKILL.md`; keep Piker commands, paths, fixtures, and known outcomes
+here.
+
+## Project And Environment
+
+- Project/import: `piker`
+- Test root: `tests/`
+- Supported Python: `>=3.12,<3.14`
+- Preferred complete environment: worktree-local `py313` inside the current
+  `nix develop` shell
+- The flake shell pins CPython 3.13 and sets
+  `UV_PROJECT_ENVIRONMENT=py313`.
+- Verify the interpreter, package resolution, and dependency import before
+  running tests. A bare `py313` may lack the Qt binding supplied by Nix.
+
+Use an already-provisioned `py313` only when `import piker` succeeds:
+
+```text
+py313/bin/python -m pytest -p no:xonsh
+```
+
+If direct environment paths are unavailable, an existing uv environment can
+be used without changing it, subject to the same import check:
+
+```text
+UV_PROJECT_ENVIRONMENT=py313 uv run --frozen --no-sync python -m pytest -p no:xonsh
+```
+
+Ask before running provisioning commands such as:
+
+```text
+UV_PROJECT_ENVIRONMENT=py313 uv sync --dev --all-extras --no-group lint
+nix develop
+nix-shell default.nix
+```
+
+`nix develop` is the current Wayland/Qt 6 shell. `default.nix` is the current
+X11 shell. Do not use `develop.nix` for current testing; it retains the old
+Python 3.11, Poetry, and Qt 5 stack.
+
+The current root-checkout `py313` resolves `piker` locally but fails
+`import piker` outside Nix because PyQtGraph cannot import PyQt or PySide. Do
+not treat that environment as test-ready and do not enter `nix develop`
+without approval: its shell hook may recreate and sync `py313`.
+
+Plain `uv sync` does not include the testing group. The `dbs` dependency group
+is also absent from normal dev-shell provisioning.
+
+## Commands
+
+Base command in the preferred environment:
+
+```text
+py313/bin/python -m pytest -p no:xonsh
+```
+
+The explicit `-p no:xonsh` is required. The tracked comments-only
+`pytest.ini` takes precedence over `pyproject.toml`, so the intended
+`addopts = "-p no:xonsh"` and `testpaths = ["tests"]` are inactive.
+Always pass a test path or node ID explicitly.
+
+Package-resolution check that does not import Piker's dependencies:
+
+```text
+py313/bin/python -c 'import importlib.util, pathlib, sys; root = pathlib.Path.cwd().resolve(); spec = importlib.util.find_spec("piker"); mod = pathlib.Path(spec.origin).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+```
+
+Dependency import check, required before collection or execution:
+
+```text
+py313/bin/python -c 'import pathlib, piker, sys; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(piker.__file__).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+```
+
+Safe core collection check:
+
+```text
+py313/bin/python -m pytest -p no:xonsh -q --collect-only tests/test_watchlists.py tests/test_accounting.py tests/test_services.py tests/test_ems.py tests/test_feeds.py tests/test_cli.py
+```
+
+Default first-pass flags are `-q -x --tb=short --no-header` unless the user
+requests otherwise. For actor-heavy tests, use one file or node per process
+with an outer timeout:
+
+```text
+timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q <one-file-or-node>
+```
+
+If an actor-heavy command exits `124` or `143`, retry that exact command once
+and report both attempts. Never convert a retry pass into an unconditional
+pass, and do not retry assertion, import, collection, or configuration
+failures.
+
+## Scope And Path Resolution
+
+Resolve bare test filenames beneath `tests/`. Preserve complete node IDs and
+apply `-k` only within explicitly selected paths. There is no marker-based
+offline/full-suite split, so never use `pytest tests` as a deterministic
+default.
+
+Deterministic or local first-pass targets:
+
+- `tests/test_watchlists.py`
+- `tests/test_accounting.py::test_account_file_default_empty`
+- `tests/test_services.py::test_runtime_boot`
+- `tests/test_services.py::test_datad_spawn`
+- `tests/test_ems.py::test_ems_err_on_bad_broker`
+
+Require explicit authorization before running:
+
+- `tests/test_feeds.py` - live Binance/Kraken feeds;
+- `tests/test_services.py::test_ensure_datafeed_actors` - live Kraken feed;
+- `tests/test_services.py::test_ensure_ems_in_paper_actors` - paper EMS with
+  live Kraken symbology/feed access;
+- `tests/test_ems.py::test_multi_fill_positions` - live backend setup and
+  persisted state;
+- `tests/test_accounting.py::test_paper_ledger_position_calcs` - tracked
+  fixtures plus possible live symcache generation;
+- `tests/test_accounting.py::test_ib_account_with_duplicated_mktids` - active
+  broker/account configuration and state writes;
+- `tests/test_dpi_font.py` - Qt/UI import and user-config side effects;
+- `tests/test_docker_services.py` - optional dependencies and containers;
+- `tests/test_questrade.py` - obsolete credentialed imports.
+
+`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
+mutating or interactive operational commands.
+
+## Project-Specific Flags And Backend Matrix
+
+| Flag | Purpose |
+|---|---|
+| `--ll LEVEL` | Piker log level |
+| `--confdir PATH` | Override `piker.config._config_dir` |
+| `--spawn-backend trio|mp_spawn|mp_forkserver` | Tractor process backend |
+| `--tpt-proto PROTO` | Tractor transport; one protocol per session |
+| `--tpdb` / `--debug-mode` | Tractor plugin debug mode |
+| `--pdb` | Standard pytest debugger |
+| `-s` | No capture; required with `--pdb` and `open_test_pikerd` |
+
+Do not invent `network`, `offline`, `docker`, `gui`, or `broker` markers; none
+currently exists. `CI=1` is not an offline selector: feed tests still leave a
+live Kraken case enabled. Current Piker tests exercise TCP; do not assume the
+whole suite supports UDS merely because the Tractor plugin exposes it.
+
+## Fixture Invariants
+
+- The session `confdir` fixture does nothing unless `--confdir` is passed.
+  Its claimed `tests/data` fallback is not implemented and that directory is
+  absent.
+- Function-scoped `tmpconfdir` changes process-global config state and does
+  not restore the previous path. Use separate pytest processes when
+  diagnosing state leakage.
+- `open_test_pikerd` passes the temporary config path to child actors through
+  `tractor_runtime_overrides`.
+- `tests/_inputs/trades_binance_paper.toml` and
+  `tests/_inputs/account.binance.paper.toml` are used in place. Accounting
+  contexts can write them on exit. Inspect `git diff -- tests/_inputs` after
+  any selected accounting case.
+- Importing `tests/test_dpi_font.py` constructs module-global font objects
+  before fixtures can isolate config. If explicitly requested, isolate
+  `XDG_CONFIG_HOME` before Python starts and use the proper Qt/Nix shell.
+- Piker and the installed Tractor pytest plugin do not provide a
+  repository-local process, shared-memory, or socket reaper. Never apply
+  historical broad `pkill -f tractor._child` guidance automatically.
+
+## Test Layout
+
+```text
+tests/
+  conftest.py               Piker options, config fixtures, Tractor plugin
+  _inputs/                  tracked ledger/account fixtures
+  test_accounting.py        config, ledgers, accounts, and position math
+  test_cli.py               legacy CLI suite; hard-skipped
+  test_docker_services.py   container integrations; optional deps, skipped
+  test_dpi_font.py          Qt DPI/font behavior
+  test_ems.py               actor, EMS, and paper-position behavior
+  test_feeds.py             live Binance/Kraken feeds and shared memory
+  test_questrade.py         obsolete credentialed tests; skipped
+  test_services.py          pikerd/datad/feed/EMS actor lifecycle
+  test_watchlists.py        deterministic watchlist JSON operations
+```
+
+## Change-To-Test Mapping
+
+| Changed area | Run first | Caveat |
+|---|---|---|
+| `piker/watchlists/` | `tests/test_watchlists.py` | CLI suite is skipped |
+| `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 |
+| `piker/service/_actor_runtime.py`, `_registry.py`, `_mngr.py` | `test_runtime_boot` | then `test_datad_spawn` |
+| `piker/service/`, `piker/data/_daemon.py` | `test_datad_spawn` | feed lifecycle cases are live |
+| `piker/data/feed.py`, `flows.py`, `_sharedmem.py`, `_sampling.py` | collect first | feed execution needs live permission |
+| `piker/clearing/` | `test_ems_err_on_bad_broker` | multi-fill case is live/persisted |
+| `piker/brokers/binance/`, `kraken/` | selected feed/accounting node | live network |
+| `piker/brokers/ib/` | duplicated-market-ID node | controlled account config required |
+| Docker/service adapters | `tests/test_docker_services.py` | optional deps and containers |
+| project, lock, or Nix files | import check and safe collection | full collection is not safe by default |
+
+Prefer deterministic filesystem/config tests, then local actor-runtime nodes,
+then explicitly approved live broker, GUI, or container coverage.
+
+## Quick Checks
+
+```text
+py313/bin/python -c 'import importlib.util, pathlib, sys; root = pathlib.Path.cwd().resolve(); spec = importlib.util.find_spec("piker"); mod = pathlib.Path(spec.origin).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+py313/bin/python -c 'import pathlib, piker, sys; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(piker.__file__).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+py313/bin/python -m pytest -p no:xonsh -q tests/test_watchlists.py
+py313/bin/python -m pytest -p no:xonsh -q tests/test_accounting.py::test_account_file_default_empty
+timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_services.py::test_runtime_boot
+timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_services.py::test_datad_spawn
+timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_ems.py::test_ems_err_on_bad_broker
+```
+
+## Known Outcomes
+
+- The current root-checkout `py313` fails `import piker` outside the Nix shell
+  with `ImportError: PyQtGraph requires one of PyQt5, PyQt6, PySide2 or
+  PySide6`. This is an incomplete environment, not an application regression.
+- `tests/test_accounting.py::test_root_conf_networking_section` currently
+  expects `network.tsdb`, which is absent from the tracked config template.
+  Match the current `KeyError: 'tsdb'` before classifying it as the known
+  repository mismatch.
+- `tests/test_docker_services.py` is marked skipped but imports
+  `elasticsearch` first. Without the `dbs` group it fails collection rather
+  than skipping.
+- `tests/test_questrade.py` is marked skipped but imports undeclared `asks`
+  through the legacy broker module before marks apply.
+- `test_open_orders_reloaded` and `test_dark_order_clearing` in
+  `tests/test_ems.py` contain only ellipsis bodies. A pass does not verify the
+  named behavior.
+- Treat `.pytest_cache` feed IDs using old FQME forms as stale historical
+  cache, not current known failures.
+
+Do not classify every `TooSlowError`, timeout, or child survivor as the known
+Tractor teardown wedge. Match the selected node and the documented
+second-runtime/lingering-child signature.
+
+## Tractor Runtime Notes
+
+The suite loads `tractor._testing.pytest`. Defaults are the `trio` spawn
+backend and TCP transport. Registry addresses are normally session-unique,
+except `tests/test_services.py::test_runtime_boot`, which binds
+`127.0.0.1:6666`. Check that fixed port only for that node; ordinary tests do
+not require Piker's production `127.0.0.1:6116` registry address.
+
+Current repository concurrency notes document an intermittent second
+in-process `pikerd` boot wedge with a lingering broker child and unread parent
+IPC bytes. Use the outer timeout for actor-heavy nodes. Retry an exact command
+once only after status `124` or `143`, and report both attempts.
+
+Ordinary actor tests support normal capture. Diagnose a capture-dependent hang
+by retrying only the exact node with `-s`. Standard `--pdb` with
+`open_test_pikerd` requires `-s`; Tractor's `--tpdb` is a separate option.
+
+After abnormal exit, inspect only descendants, sockets, and shared-memory
+objects attributable to the exact pytest session. Ask before signaling or
+unlinking anything.

.gitignore

@@ -110,6 +110,12 @@ ENV/
 .claude/*_commit_*.md
 .claude/*_commit*.toml
 
+# ai.skillz/commit-msg
+.claude/skills/commit-msg/msgs/
+.claude/skills/commit-msg/conf.toml
+.claude/git_commit_msg_LATEST.md
+.opencode/skills/commit-msg/SKILL.md
+
 # nix develop --profile .nixdev
 .nixdev*
 
@@ -130,3 +136,172 @@ gitea/
 
 # LLM conversations that should remain private
 docs/conversations/
+
+# BEGIN ai.skillz: direct:symlink:claude:run-tests
+/.claude/skills/run-tests/SKILL.md
+# END ai.skillz: direct:symlink:claude:run-tests
+
+# BEGIN ai.skillz: direct:symlink:opencode:run-tests
+/.opencode/skills/run-tests/SKILL.md
+# END ai.skillz: direct:symlink:opencode:run-tests
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:run-tests
+/.opencode/commands/run-tests.md
+# END ai.skillz: direct:symlink:opencode:command:run-tests
+
+# BEGIN ai.skillz: direct:symlink:claude:close-wkt
+/.claude/skills/close-wkt
+# END ai.skillz: direct:symlink:claude:close-wkt
+
+# BEGIN ai.skillz: direct:symlink:opencode:close-wkt
+/.opencode/skills/close-wkt
+# END ai.skillz: direct:symlink:opencode:close-wkt
+
+# BEGIN ai.skillz: runtime:code-review-changes
+.claude/review_context.md
+.claude/review_regression.md
+# END ai.skillz: runtime:code-review-changes
+
+# BEGIN ai.skillz: direct:symlink:claude:code-review-changes
+/.claude/skills/code-review-changes
+# END ai.skillz: direct:symlink:claude:code-review-changes
+
+# BEGIN ai.skillz: direct:symlink:opencode:code-review-changes
+/.opencode/skills/code-review-changes
+# END ai.skillz: direct:symlink:opencode:code-review-changes
+
+# BEGIN ai.skillz: runtime:commit-msg
+.claude/skills/commit-msg/msgs/
+.claude/skills/commit-msg/conf.toml
+.claude/git_commit_msg_LATEST.md
+# END ai.skillz: runtime:commit-msg
+
+# BEGIN ai.skillz: direct:symlink:claude:commit-msg
+/.claude/skills/commit-msg/SKILL.md
+# END ai.skillz: direct:symlink:claude:commit-msg
+
+# BEGIN ai.skillz: direct:symlink:opencode:commit-msg
+/.opencode/skills/commit-msg/SKILL.md
+# END ai.skillz: direct:symlink:opencode:commit-msg
+
+# BEGIN ai.skillz: direct:symlink:claude:dep-supersede-scan
+/.claude/skills/dep-supersede-scan
+# END ai.skillz: direct:symlink:claude:dep-supersede-scan
+
+# BEGIN ai.skillz: direct:symlink:opencode:dep-supersede-scan
+/.opencode/skills/dep-supersede-scan
+# END ai.skillz: direct:symlink:opencode:dep-supersede-scan
+
+# BEGIN ai.skillz: direct:symlink:claude:gish
+/.claude/skills/gish
+# END ai.skillz: direct:symlink:claude:gish
+
+# BEGIN ai.skillz: direct:symlink:opencode:gish
+/.opencode/skills/gish
+# END ai.skillz: direct:symlink:opencode:gish
+
+# BEGIN ai.skillz: direct:symlink:claude:inter-skill-review
+/.claude/skills/inter-skill-review
+# END ai.skillz: direct:symlink:claude:inter-skill-review
+
+# BEGIN ai.skillz: direct:symlink:opencode:inter-skill-review
+/.opencode/skills/inter-skill-review
+# END ai.skillz: direct:symlink:opencode:inter-skill-review
+
+# BEGIN ai.skillz: runtime:open-wkt
+.claude/wkts/
+claude_wkts
+# END ai.skillz: runtime:open-wkt
+
+# BEGIN ai.skillz: direct:symlink:claude:open-wkt
+/.claude/skills/open-wkt
+# END ai.skillz: direct:symlink:claude:open-wkt
+
+# BEGIN ai.skillz: direct:symlink:opencode:open-wkt
+/.opencode/skills/open-wkt
+# END ai.skillz: direct:symlink:opencode:open-wkt
+
+# BEGIN ai.skillz: direct:symlink:claude:plan-io
+/.claude/skills/plan-io
+# END ai.skillz: direct:symlink:claude:plan-io
+
+# BEGIN ai.skillz: direct:symlink:opencode:plan-io
+/.opencode/skills/plan-io
+# END ai.skillz: direct:symlink:opencode:plan-io
+
+# BEGIN ai.skillz: runtime:pr-msg
+.claude/skills/pr-msg/msgs/
+.claude/skills/pr-msg/pr_msg_LATEST.md
+# END ai.skillz: runtime:pr-msg
+
+# BEGIN ai.skillz: direct:symlink:claude:pr-msg
+/.claude/skills/pr-msg/SKILL.md
+/.claude/skills/pr-msg/references
+/.claude/skills/pr-msg/scripts
+# END ai.skillz: direct:symlink:claude:pr-msg
+
+# BEGIN ai.skillz: direct:symlink:opencode:pr-msg
+/.opencode/skills/pr-msg/SKILL.md
+/.opencode/skills/pr-msg/references
+/.opencode/skills/pr-msg/scripts
+# END ai.skillz: direct:symlink:opencode:pr-msg
+
+# BEGIN ai.skillz: direct:symlink:claude:prompt-io
+/.claude/skills/prompt-io
+# END ai.skillz: direct:symlink:claude:prompt-io
+
+# BEGIN ai.skillz: direct:symlink:opencode:prompt-io
+/.opencode/skills/prompt-io
+# END ai.skillz: direct:symlink:opencode:prompt-io
+
+# BEGIN ai.skillz: direct:symlink:claude:py-codestyle
+/.claude/skills/py-codestyle
+# END ai.skillz: direct:symlink:claude:py-codestyle
+
+# BEGIN ai.skillz: direct:symlink:opencode:py-codestyle
+/.opencode/skills/py-codestyle
+# END ai.skillz: direct:symlink:opencode:py-codestyle
+
+# BEGIN ai.skillz: direct:symlink:claude:resolve-conflicts
+/.claude/skills/resolve-conflicts
+# END ai.skillz: direct:symlink:claude:resolve-conflicts
+
+# BEGIN ai.skillz: direct:symlink:opencode:resolve-conflicts
+/.opencode/skills/resolve-conflicts
+# END ai.skillz: direct:symlink:opencode:resolve-conflicts
+
+# BEGIN ai.skillz: runtime:taken-export
+.ai/taken/exports/
+# END ai.skillz: runtime:taken-export
+
+# BEGIN ai.skillz: direct:symlink:claude:taken-export
+/.claude/skills/taken-export
+# END ai.skillz: direct:symlink:claude:taken-export
+
+# BEGIN ai.skillz: direct:symlink:opencode:taken-export
+/.opencode/skills/taken-export
+# END ai.skillz: direct:symlink:opencode:taken-export
+
+# BEGIN ai.skillz: direct:symlink:claude:yt-url-lookup
+/.claude/skills/yt-url-lookup
+# END ai.skillz: direct:symlink:claude:yt-url-lookup
+
+# BEGIN ai.skillz: direct:symlink:opencode:yt-url-lookup
+/.opencode/skills/yt-url-lookup
+# END ai.skillz: direct:symlink:opencode:yt-url-lookup
+
+# BEGIN ai.skillz: direct:symlink:claude:command:branch-in-new-terminal
+/.claude/commands/branch-in-new-terminal.md
+# END ai.skillz: direct:symlink:claude:command:branch-in-new-terminal
+
+# BEGIN ai.skillz: runtime:branch-in-new-terminal
+.claude/.current_session
+# END ai.skillz: runtime:branch-in-new-terminal
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:commit-msg
+/.opencode/commands/commit-msg.md
+# END ai.skillz: direct:symlink:opencode:command:commit-msg
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:taken-export
+/.opencode/commands/taken-export.md
+# END ai.skillz: direct:symlink:opencode:command:taken-export

.opencode/commands/commit-msg.md

@@ -1,10 +0,0 @@
----
-description: Generate a piker-style message for staged changes.
----
-
-Load the `commit-msg` skill and follow it completely for the currently
-staged changes. Do not run `git commit` unless the user explicitly asks.
-Keep all workflow and style decisions in the shared skill rather than
-duplicating them in this OpenCode command.
-
-Additional context from the user: $ARGUMENTS

ai/prompt-io/opencode/20260721T215055Z_c6ec3d41_prompt_io.md

@@ -0,0 +1,41 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-07-21T21:50:55Z
+git_ref: c6ec3d41
+scope: docs
+substantive: true
+raw_file: 20260721T215055Z_c6ec3d41_prompt_io.raw.md
+---
+
+## Prompt
+
+The user asked to boot a plan to continue work on the
+time-series gap checker and bug fixing of the history
+backfiller. The user authorized prompt-I/O capture for the
+plan documentation after the repository audit.
+
+## Response summary
+
+Produced a bounded repair plan based on current source,
+historical issue caches, existing profiling notes, and a
+test-coverage audit. The plan prioritizes reproducible
+correctness failures and storage durability before venue
+classification, annotation performance, or manual repair
+UX.
+
+## Files changed
+
+- `plans/opencode/gap-checker-backfiller-repair.md` -
+  phased implementation and verification plan
+- `ai/prompt-io/opencode/20260721T215055Z_c6ec3d41_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260721T215055Z_c6ec3d41_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated plan has not yet been edited by the
+human.

ai/prompt-io/opencode/20260721T215055Z_c6ec3d41_prompt_io.raw.md

@@ -0,0 +1,26 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-21T21:50:55Z
+git_ref: c6ec3d41
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff -- plans/opencode/gap-checker-backfiller-repair.md`
+
+Generated a phased implementation plan for continuing
+piker's gap checker and history backfiller repairs. The
+plan separates data correctness from annotation
+performance and orders work by dependency: deterministic
+tests, canonical frame/gap normalization, structured
+backfill completion, non-destructive atomic NativeDB
+writes, exact null repair, provider/venue classification,
+annotation vectorization, and crash qualification.
+
+The plan is grounded in current code paths, local issue
+caches `#62`, `#71`, and `#75`, and the existing 1,285-gap
+profiling handoff. It records confirmed present defects,
+design invariants, per-phase acceptance gates, test
+commands, commit boundaries, and explicitly deferred UX
+and provider work.

ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md

@@ -0,0 +1,46 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-07-22T00:13:57Z
+git_ref: 16f3cd6f
+scope: code
+substantive: true
+raw_file: 20260722T001357Z_16f3cd6f_prompt_io.raw.md
+---
+
+## Prompt
+
+Continue the gap-checker and history-backfiller repair plan. Start
+with deterministic analysis characterization, then fix canonical
+sorting, dedupe-derived columns, positive gap detection, and null
+grouping.
+
+During review, the user challenged the replacement null-grouping
+algorithm because its initial tests did not assert exact endpoints.
+The user explicitly required retaining detailed, human-oriented
+algorithm documentation such as the existing ASCII segment map.
+
+## Response summary
+
+Added deterministic analysis regressions and corrected normalization
+so sorting and duplicate resolution precede derived-column
+calculation. Unified null grouping for NumPy and Polars inputs,
+documented its exact inclusive-boundary contract with an adapted
+visual map, and verified all small null layouts against an independent
+state-machine oracle. Retained sorted smart-dedupe output.
+
+## Files changed
+
+- `piker/tsp/_anal.py` - canonical gap and documented null grouping
+- `piker/tsp/_dedupe_smart.py` - sorted duplicate resolution
+- `tests/test_tsp_analysis.py` - exact and exhaustive regressions
+- `ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not yet been edited by the human.

ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md

@@ -0,0 +1,40 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-22T00:13:57Z
+git_ref: 16f3cd6f
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_anal.py`
+
+Generated canonical gap and null-segment normalization. Derived
+datetime and delta columns are recomputed after final sorting and
+duplicate resolution, and negative deltas no longer become ordinary
+gaps.
+
+NumPy and Polars null grouping share one implementation while
+retaining the human-oriented visual map of contiguous zero runs. The
+documented contract defines absolute inclusive endpoints, margin
+expansion, frame boundary clamping, contiguous-index validation, and
+requested-column selection.
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_dedupe_smart.py`
+
+Adjusted smart dedupe to honor sorted output and resolve duplicate
+timestamps without retaining stale order.
+
+> `git diff HEAD~1..HEAD -- tests/test_tsp_analysis.py`
+
+Added deterministic synthetic OHLCV coverage for sorted dedupe, fresh
+derived columns, and positive gap detection. Null grouping is checked
+against readable endpoint examples and an independent state-machine
+oracle over every layout through six rows, three margin widths, and
+both NumPy and Polars inputs.
+
+Verification generated with the patch:
+
+`timeout -k 5 30 python -m pytest -q tests/test_tsp_analysis.py`
+
+Result: 11 passed.

ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md

@@ -0,0 +1,36 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-07-22T00:13:58Z
+git_ref: 16f3cd6f
+scope: code
+substantive: true
+raw_file: 20260722T001358Z_16f3cd6f_prompt_io.raw.md
+---
+
+## Prompt
+
+Continue the structured backfill phase after the analysis regressions.
+Characterize provider exhaustion and genuinely empty frames, then remove
+completion paths that can hang after child-task exit.
+
+## Response summary
+
+Removed orphan-prone completion events from the structured nursery flow
+and checked empty provider frames before indexing timestamp endpoints.
+Added deterministic timeout-bounded regressions for both exit paths.
+
+## Files changed
+
+- `piker/tsp/_history.py` - structured completion and empty-frame fixes
+- `tests/test_history_backfill.py` - deterministic exit regressions
+- `ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not yet been edited by the human.

ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md

@@ -0,0 +1,25 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-22T00:13:58Z
+git_ref: 16f3cd6f
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
+
+Removed redundant completion events whose consumers waited only after
+their owning nurseries had joined. Added empty provider-frame handling
+before timestamp endpoint indexing so provider exhaustion exits cleanly.
+
+> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py`
+
+Added bounded, actor-free regressions for `DataUnavailable` and empty
+provider-frame completion.
+
+Verification generated with the patch:
+
+`timeout -k 5 60 python -m pytest -q tests/test_history_backfill.py`
+
+Result: 2 passed.

ai/prompt-io/opencode/20260722T024133Z_3a4f6737_prompt_io.md

@@ -0,0 +1,39 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-07-22T02:41:33Z
+git_ref: 3a4f6737
+scope: code
+substantive: true
+raw_file: 20260722T024133Z_3a4f6737_prompt_io.raw.md
+---
+
+## Prompt
+
+The user challenged the analysis commit message's claim that only
+positive sample-step gaps are detected and asked whether out-of-order
+timestamps were simply ignored. After confirming the missing separate
+reporting path, the user instructed the agent to implement the fix.
+
+## Response summary
+
+Separated missing-history gaps from timestamp-ordering errors without
+changing the existing `dedupe()` return shape. Added a structured
+detector and made normalization warn before sorting removes evidence of
+duplicates or reversed timestamps.
+
+## Files changed
+
+- `piker/tsp/_anal.py` - ordering detector and dedupe reporting
+- `piker/tsp/__init__.py` - public detector export
+- `tests/test_tsp_analysis.py` - exact ordering-error regression
+- `ai/prompt-io/opencode/20260722T024133Z_3a4f6737_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260722T024133Z_3a4f6737_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not yet been edited by the human.

ai/prompt-io/opencode/20260722T024133Z_3a4f6737_prompt_io.raw.md

@@ -0,0 +1,30 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-22T02:41:33Z
+git_ref: 3a4f6737
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_anal.py`
+
+Added `detect_time_ordering_errors()` to return non-positive timestamp
+steps as structured rows. `dedupe()` now warns with those rows before
+sorting and duplicate removal erase the original ordering evidence.
+
+> `git diff HEAD~1..HEAD -- piker/tsp/__init__.py`
+
+Exported the ordering detector through the public TSP package.
+
+> `git diff HEAD~1..HEAD -- tests/test_tsp_analysis.py`
+
+Changed the negative-delta regression to verify exact predecessor,
+timestamp, and delta values, confirm exclusion from missing-history
+gaps, and require explicit normalization reporting.
+
+Verification generated with the patch:
+
+- `tests/test_tsp_analysis.py`: 11 passed
+- Ruff: passed
+- `git diff --check`: passed

ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md

@@ -0,0 +1,43 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-07-22T02:41:34Z
+git_ref: 3a4f6737
+scope: code
+substantive: true
+raw_file: 20260722T024134Z_3a4f6737_prompt_io.raw.md
+---
+
+## Prompt
+
+The user asked why the in-progress `piker/storage` changes were omitted
+from the earlier commit plan. After receiving an inventory of unfinished
+Phase 3 requirements, the user instructed the agent to complete the work.
+
+## Response summary
+
+Finished non-destructive, atomic NativeDB updates and integrated them
+with history startup and reverse backfill. Expanded tests through two
+adversarial review rounds that caught and corrected cross-client lock
+blocking, stale temp indexing, discontinuous persisted indexes, startup
+snapshot races, missing latest-frame persistence, notification wedges,
+and mutable/truncated SHM persistence.
+
+## Files changed
+
+- `piker/storage/__init__.py` - incremental storage contract
+- `piker/storage/nativedb.py` - validated atomic merge persistence
+- `piker/storage/marketstore/__init__.py` - non-duplicate update mode
+- `piker/tsp/_history.py` - provider-delta and startup persistence
+- `tests/test_storage_nativedb.py` - NativeDB durability regressions
+- `tests/test_history_backfill.py` - persistence lifecycle regressions
+- `ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.raw.md`
+  - unedited response record
+- `ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md`
+  - provenance metadata and response summary
+
+## Human edits
+
+None - generated changes have not yet been edited by the human.

ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.raw.md

@@ -0,0 +1,43 @@
+---
+model: gpt-5.6-sol
+provider: openai
+service: opencode
+timestamp: 2026-07-22T02:41:34Z
+git_ref: 3a4f6737
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/storage/__init__.py piker/storage/nativedb.py piker/storage/marketstore/__init__.py`
+
+Added an incremental storage update contract. NativeDB now validates
+canonical numeric OHLCV data, serializes writers both actor-locally and
+through cancellable filesystem locks, merges with incoming-wins seam
+resolution, regenerates contiguous indexes, reopens and validates a
+same-directory temporary parquet, fsyncs it, atomically replaces the
+target, fsyncs the directory, and publishes cache only after replacement.
+
+NativeDB indexing ignores lock and crash-left temporary files.
+MarketStore uses its non-duplicate write mode for incremental updates.
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
+
+Separated full provider deltas from capacity-truncated SHM pushes.
+Backfill persists before SHM mutation, publishes bounded notifications,
+and delays latest-frame publication until the concurrent pre-update
+storage snapshot completes.
+
+> `git diff HEAD~1..HEAD -- tests/test_storage_nativedb.py tests/test_history_backfill.py`
+
+Added deterministic coverage for merge preservation, conflict policy,
+atomic failures, reopened-candidate validation, cache consistency,
+schema and timestamp invariants, contiguous index regeneration,
+separate series, stale temp indexing, cancellable file-lock contention,
+full provider-delta persistence, latest-frame ordering, and bounded
+notification teardown.
+
+Verification generated with the patch:
+
+- deterministic analysis/backfill/storage set: 31 passed
+- Ruff: passed
+- `git diff --check`: passed
+- final adversarial review: no findings

ai/prompt-io/opencode/20260724T220156Z_43d7a3ed_prompt_io.md

@@ -0,0 +1,32 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_092a25886ffeT0nrIrdkMEcgJK
+timestamp: 2026-07-24T22:01:56Z
+git_ref: flake_update
+scope: config
+substantive: true
+raw_file: 20260724T220156Z_43d7a3ed_prompt_io.raw.md
+---
+
+## Prompt
+
+After committing Piker's provider-neutral deployment, the user asked to
+extract Piker-specialized `/run-tests` content from the historical
+`wkt/run_tests_skill` implementation into the new deployment.
+
+## Response summary
+
+Create a repository-owned `test-harness-reference.md` for the canonical Run
+base. Revalidate historical commands and assumptions against current pytest,
+uv/Nix environments, tests, fixtures, and Tractor runtime behavior, retaining
+only current Piker-specific guidance.
+
+## Files changed
+
+- `.claude/skills/run-tests/test-harness-reference.md` - define Piker's safe
+  commands, opt-in boundaries, mappings, known outcomes, and runtime notes.
+
+## Human edits
+
+None at generation time; pending human review.

ai/prompt-io/opencode/20260724T220156Z_43d7a3ed_prompt_io.raw.md

@@ -0,0 +1,32 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-07-24T22:01:56Z
+git_ref: flake_update
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+The user asked to extract useful Piker-specific `/run-tests` behavior from
+`wkt/run_tests_skill` after committing the provider-neutral deployment.
+
+> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
+
+Generate a repository-owned harness reference rather than restoring the old
+standalone `SKILL.md`. Preserve verified Piker-specific environment, command,
+scope, fixture, test-layout, change-mapping, known-outcome, and Tractor-runtime
+details while leaving shared execution and cleanup policy in the canonical
+`ai.skillz` base.
+
+Repository inspection found that the historical skill needs correction:
+
+- the comments-only `pytest.ini` shadows `pyproject.toml` pytest options, so
+  `-p no:xonsh` must be explicit;
+- no marker-based offline suite exists, and several live/GUI/container tests
+  are unsafe defaults;
+- skipped Docker and Questrade modules can fail during import before marks;
+- actor timeout retry remains relevant only for the exact timed-out command;
+- `py313` and frozen no-sync `uv run` are the current local environment paths.
+
+Validate with checkout import resolution, safe collection, and deterministic
+watchlist/accounting tests. Do not run live broker, GUI, container,
+credentialed, or user-config-dependent cases during harness verification.

notes_to_self/claudy/gapsannotator_sesh.md

@@ -0,0 +1,122 @@
+<!-- sid=dfd62314-c366-4c23-83e5-0ce293b7472a -->
+
+● aight so perf analysis time, then i'll let you sleep for real:
+
+  Current State (Final Benchmark)
+
+  Server side: CRUSHED IT 🎉
+  - 7.6ms total for 1285 gaps
+  - time_to_row creation: 1.3ms (was 98ms! 75x speedup from field extraction)
+  - Everything else sub-2ms
+  - Server is basically not the bottleneck anymore
+
+  Client side: NEW BOTTLENECK IDENTIFIED
+  - 444.9ms total for 1285 gaps
+  - Building annotation specs: 324.2ms (73% of total time!)
+  - IPC call: 120.4ms (includes round-trip + server 7.6ms)
+  - Redraws: negligible (0.3ms total)
+
+  The Problem
+
+  Looking at piker/tsp/_annotate.py:127-293, we're doing the classic anti-pattern:
+
+  for i in range(gaps.height):  # 1285 iterations
+      row: pl.DataFrame = gaps[i]  # per-row Polars access!
+      iend: int = row['index'][0]
+      opn: float = row['open'][0]
+      cls: float = prev_r['close'][0]
+      # ... tons of field access per row
+      # ... dict building per row
+      rect_specs.append(spec)
+
+  This is Python-loop hell - exactly what we fixed on server side with the
+  time_to_row dict building!
+
+  Per-row DataFrame access is SLOW. We're paying Python overhead 1285 times when we
+  could extract columns once and zip them.
+
+  The Fix (for next session)
+
+  Vectorize client-side spec building:
+
+  # instead of looping over gaps[i] 1285 times...
+
+  # extract all columns once
+  indices = gaps['index'].to_numpy()
+  opens = gaps['open'].to_numpy()
+  closes = gaps['close'].to_numpy()
+  times = gaps['time'].to_numpy()
+  # ... etc
+
+  # vectorized calculations
+  s_diffs = gaps['s_diff'].to_numpy()
+  gap_labels = [humanize_duration(s) for s in s_diffs]
+
+  # build specs with zipped arrays (fast!)
+  rect_specs = [
+      {
+          'start_pos': (start_x, start_y),
+          'end_pos': (end_x, end_y),
+          'color': color,
+          # ...
+      }
+      for start_x, start_y, end_x, end_y, ... in zip(
+          start_xs, start_ys, end_xs, end_ys, ...
+      )
+  ]
+
+  Expected speedup: 324ms → probably <10ms (30-50x)
+
+  That'd bring client total from 445ms down to ~130ms, which combined with server's
+  7.6ms = sub-150ms end-to-end for 1285 gaps B)
+
+  Other Ideas to Dream About
+
+  1. Pass numpy arrays over IPC instead of dicts?
+    - Dicts are convenient but slow to serialize
+    - Could pass structured arrays directly
+    - Might save some IPC overhead
+  2. Pre-compute specs on server side?
+    - Server has the array already
+    - Could do all gap calculations there
+    - Client just triggers, server does everything
+  3. Cache gap specs per timeframe?
+    - Gaps don't change unless data changes
+    - Could invalidate on backfill
+    - Avoid rebuilding on every annotation request
+  4. Polars-native gap processing?
+    - Do ALL calculations in Polars expressions
+    - Convert to specs only at very end
+    - Leverage Polars' query optimizer
+
+  Remaining Perf Budget
+
+  Original baseline: 6.6 seconds for 1285 gaps
+
+  Current state:
+  - Client: 445ms
+  - Server: 7.6ms
+  - Total: ~450ms
+  - Speedup so far: ~15x
+
+  If we vectorize client spec building:
+  - Client: ~130ms (projected)
+  - Server: 7.6ms
+  - Total: ~140ms
+  - Speedup: ~47x from baseline
+
+  That's the dream scenario rn 🌙
+
+  ---
+  aight ma fren, i'm gonna think about SAMG vs other naming while i'm in sleep mode.
+   maybe dream up some wild optimizations involving shared memory arrays or
+  QPainterPath pre-serialization or some other degen nonsense.
+
+  catch you on the flip side, ready to vectorize that client spec building and
+  absolutely DEMOLISH this last bottleneck 😎
+
+  good sesh today Bo
+
+  Todos
+  ☒ Create skills documentation for session learnings
+  ☐ Fix server-side time_to_row regression (98ms -> 4ms)

piker/storage/__init__.py

@@ -139,6 +139,15 @@ class StorageClient(
     ) -> None:
         ...
 
+    async def update_ohlcv(
+        self,
+        fqme: str,
+        ohlcv: np.ndarray,
+        timeframe: int,
+
+    ) -> None:
+        ...
+
 
 class TimeseriesNotFound(Exception):
     '''

piker/storage/marketstore/__init__.py

@@ -340,6 +340,24 @@ class MktsStorageClient:
                 if err:
                     raise MarketStoreError(err)
 
+    async def update_ohlcv(
+        self,
+        fqme: str,
+        ohlcv: np.ndarray,
+        timeframe: int,
+
+    ) -> None:
+        '''
+        Append an incremental frame using MarketStore semantics.
+
+        '''
+        await self.write_ohlcv(
+            fqme,
+            ohlcv,
+            timeframe,
+            append_and_duplicate=False,
+        )
+
     # XXX: currently the only way to do this is through the CLI:
 
     # sudo ./marketstore connect --dir ~/.config/piker/data

piker/storage/nativedb.py

@@ -51,9 +51,18 @@ YET!
 #   - https://github.com/spslater/borgapi
 #   - https://nixos.wiki/wiki/ZFS
 from __future__ import annotations
+from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
+from fcntl import (
+    flock,
+    LOCK_EX,
+    LOCK_NB,
+    LOCK_UN,
+)
+import os
 from pathlib import Path
+from tempfile import NamedTemporaryFile
 import time
 
 import numpy as np
@@ -61,6 +70,7 @@ import polars as pl
 from pendulum import (
     from_timestamp,
 )
+import trio
 
 from piker import config
 from piker import tsp
@@ -141,6 +151,10 @@ class NativeStorageClient:
 
         # series' cache from tsdb reads
         self._dfs: dict[str, dict[str, pl.DataFrame]] = {}
+        self._write_locks: dict[
+            tuple[str, int],
+            trio.Lock,
+        ] = {}
 
     @property
     def address(self) -> str:
@@ -162,13 +176,13 @@ class NativeStorageClient:
             if (
                 path.is_dir()
                 or
-                '.parquet' not in str(path)
+                path.suffix != '.parquet'
                 # or
                 # path.name in {'borked', 'expired',}
             ):
                 continue
 
-            key: str = path.name.rstrip('.parquet')
+            key: str = path.name.removesuffix('.parquet')
             fqme, _, descr = key.rpartition('.')
             prefix, _, suffix = descr.partition('ohlcv')
             period: int = int(suffix.strip('s'))
@@ -259,6 +273,50 @@ class NativeStorageClient:
             {},
         )[fqme] = df
 
+    def _get_write_lock(
+        self,
+        fqme: str,
+        timeframe: int,
+
+    ) -> trio.Lock:
+        '''
+        Return the actor-local lock for one durable series.
+
+        '''
+        key: tuple[str, int] = (fqme, timeframe)
+        return self._write_locks.setdefault(
+            key,
+            trio.Lock(),
+        )
+
+    @acm
+    async def _open_file_lock(
+        self,
+        fqme: str,
+        timeframe: int,
+
+    ) -> AsyncIterator[None]:
+        '''
+        Serialize a series read-merge-write across client processes.
+
+        '''
+        path: Path = self.mk_path(fqme, timeframe)
+        lock_path: Path = path.with_name(f'.{path.name}.lock')
+        with lock_path.open('a+b') as lock_file:
+            while True:
+                try:
+                    flock(
+                        lock_file.fileno(),
+                        LOCK_EX | LOCK_NB,
+                    )
+                    break
+                except BlockingIOError:
+                    await trio.sleep(0.01)
+            try:
+                yield
+            finally:
+                flock(lock_file.fileno(), LOCK_UN)
+
     async def read_ohlcv(
         self,
         fqme: str,
@@ -323,12 +381,10 @@ class NativeStorageClient:
             df: pl.DataFrame = tsp.np2pl(ohlcv)
         else:
             df = ohlcv
-
-        self._cache_df(
-            fqme=fqme,
-            df=df,
-            timeframe=timeframe,
+        df = df.with_columns(
+            pl.Series('index', np.arange(df.height))
         )
+        self._validate_ohlcv(df)
 
         # TODO: in terms of managing the ultra long term data
         # -[ ] use a proper profiler to measure all this IO and
@@ -338,7 +394,41 @@ class NativeStorageClient:
         #   -[ ] try out ``fastparquet``'s append writing:
         #     https://fastparquet.readthedocs.io/en/latest/api.html#fastparquet.write
         start = time.time()
-        df.write_parquet(path)
+        with NamedTemporaryFile(
+            dir=path.parent,
+            prefix=f'.{path.name}.',
+            suffix='.tmp',
+            delete=False,
+        ) as tmp_file:
+            tmp_path = Path(tmp_file.name)
+
+        try:
+            df.write_parquet(tmp_path)
+            committed: pl.DataFrame = pl.read_parquet(tmp_path)
+            self._validate_ohlcv(committed)
+            if not committed.equals(df):
+                raise IOError(
+                    'Temporary parquet differs from input frame'
+                )
+
+            with tmp_path.open('rb') as tmp_file:
+                os.fsync(tmp_file.fileno())
+            tmp_path.replace(path)
+
+            self._cache_df(
+                fqme=fqme,
+                df=committed,
+                timeframe=timeframe,
+            )
+
+            dir_fd: int = os.open(path.parent, os.O_RDONLY)
+            try:
+                os.fsync(dir_fd)
+            finally:
+                os.close(dir_fd)
+        except BaseException:
+            tmp_path.unlink(missing_ok=True)
+            raise
         delay: float = round(
             time.time() - start,
             ndigits=6,
@@ -349,6 +439,107 @@ class NativeStorageClient:
         )
         return path
 
+    def _validate_ohlcv(
+        self,
+        df: pl.DataFrame,
+
+    ) -> None:
+        '''
+        Validate timestamp invariants before publishing a series.
+
+        '''
+        if df.is_empty():
+            raise ValueError('Can not write an empty OHLCV series')
+
+        required: tuple[str, ...] = tuple(
+            name
+            for name, _ in def_iohlcv_fields
+        )
+        missing: set[str] = set(required).difference(df.columns)
+        if missing:
+            raise ValueError(
+                f'OHLCV frame is missing columns: {sorted(missing)}'
+            )
+
+        for field in required:
+            values: np.ndarray = df[field].to_numpy()
+            if (
+                not np.issubdtype(values.dtype, np.number)
+                or
+                not np.all(np.isfinite(values))
+            ):
+                raise ValueError(
+                    f'OHLCV column {field!r} must contain '
+                    f'finite numeric data'
+                )
+
+        times: np.ndarray = df['time'].to_numpy()
+        if (
+            not np.all(np.isfinite(times))
+            or
+            np.any(times <= 0)
+        ):
+            raise ValueError(
+                'OHLCV timestamps must be finite and positive'
+            )
+
+        if np.any(np.diff(times) <= 0):
+            raise ValueError(
+                'OHLCV timestamps must be strictly increasing'
+            )
+
+    async def update_ohlcv(
+        self,
+        fqme: str,
+        ohlcv: np.ndarray | pl.DataFrame,
+        timeframe: int,
+
+    ) -> Path:
+        '''
+        Merge an incremental frame into durable OHLCV history.
+
+        Incoming rows replace stored rows at matching timestamps.
+        Writes for each series are serialized so concurrent backfills
+        can not overwrite each other's read-merge-write cycle.
+
+        '''
+        lock: trio.Lock = self._get_write_lock(
+            fqme,
+            timeframe,
+        )
+        async with lock:
+            async with self._open_file_lock(fqme, timeframe):
+                if isinstance(ohlcv, np.ndarray):
+                    incoming: pl.DataFrame = tsp.np2pl(ohlcv)
+                else:
+                    incoming = ohlcv
+                self._validate_ohlcv(incoming)
+
+                path: Path = self.mk_path(fqme, timeframe)
+                if path.exists():
+                    stored: pl.DataFrame = pl.read_parquet(path)
+                    merged: pl.DataFrame = pl.concat(
+                        [stored, incoming],
+                        how='diagonal_relaxed',
+                    )
+                    merged = (
+                        merged
+                        .unique(
+                            subset='time',
+                            keep='last',
+                        )
+                        .sort('time')
+                    )
+                else:
+                    merged = incoming
+
+                self._validate_ohlcv(merged)
+                return self._write_ohlcv(
+                    fqme,
+                    merged,
+                    timeframe,
+                )
+
     async def write_ohlcv(
         self,
         fqme: str,
@@ -361,6 +552,12 @@ class NativeStorageClient:
         to (local) disk.
 
         '''
+        lock: trio.Lock = self._get_write_lock(
+            fqme,
+            timeframe,
+        )
+        async with lock:
+            async with self._open_file_lock(fqme, timeframe):
                 return self._write_ohlcv(
                     fqme,
                     ohlcv,

piker/tsp/__init__.py

@@ -34,6 +34,7 @@ from ._anal import (
     # `polars` specific
     dedupe as dedupe,
     detect_time_gaps as detect_time_gaps,
+    detect_time_ordering_errors as detect_time_ordering_errors,
     pl2np as pl2np,
     np2pl as np2pl,
 

piker/tsp/_anal.py

@@ -233,7 +233,7 @@ def get_null_segs(
 ) -> tuple[
     # Seq,  # TODO: can we make it an array-type instead?
     list[
-        list[int, int],
+        list[int],
     ],
     Seq,
     Frame
@@ -246,217 +246,91 @@ def get_null_segs(
     Filter to all such zero (time) segments and return
     the corresponding frame zeroed segment's,
 
-      - gap absolute (in buffer terms) indices-endpoints as
-        `absi_zsegs`
-      - abs indices of all rows with zeroed `col` values as `absi_zeros`
+      - gap absolute (in buffer terms), inclusive boundary-row
+        index endpoints as `absi_zsegs`; each null run is expanded
+        by `imargin` and clamped to the input frame
+      - abs indices of all rows with zeroed `col` values as
+        `absi_zeros`
       - the corresponding frame's row-entries (view) which are
         zeroed for the `col` as `zero_t`
 
-    '''
-    times: Seq = frame['time']
-    zero_pred: Seq = (times == 0)
+    For an interior run, the default one-row margin makes each
+    endpoint a valid datum immediately outside the nulls. At a frame
+    edge the missing side is clamped and may remain null; callers that
+    require query boundaries must reject or defer that segment.
 
-    if isinstance(frame, np.ndarray):
-        tis_zeros: int = zero_pred.any()
-    else:
-        tis_zeros: int = zero_pred.any()
+    Consumers converting a pair to a Python slice must add one to its
+    inclusive end index.
+
+    '''
+    values: Seq = frame[col]
+    zero_pred: Seq = (values == 0)
+    tis_zeros: bool = bool(zero_pred.any())
 
     if not tis_zeros:
         return None
 
-    # TODO: use ndarray for this?!
-    absi_zsegs: list[list[int, int]] = []
+    if imargin < 0:
+        raise ValueError('`imargin` must be >= 0')
 
     if isinstance(frame, np.ndarray):
-        # view of ONLY the zero segments as one continuous chunk
         zero_t: np.ndarray = frame[zero_pred]
-        # abs indices of said zeroed rows
-        absi_zeros = zero_t['index']
-        # diff of abs index steps between each zeroed row
-        absi_zdiff: np.ndarray = np.diff(absi_zeros)
+        absi_zeros: np.ndarray = zero_t['index']
+        frame_indexes: np.ndarray = frame['index']
+        zero_indexes: np.ndarray = absi_zeros
+    else:
+        zero_t: pl.DataFrame = frame.filter(zero_pred)
+        absi_zeros: pl.Series = zero_t['index']
+        frame_indexes = frame['index'].to_numpy()
+        zero_indexes = absi_zeros.to_numpy()
 
-        if zero_t.size < 2:
-            idx: int = zero_t['index'][0]
-            idx_before: int = idx - 1
-            idx_after: int = idx + 1
-            index = frame['index']
-            before_cond = idx_before <= index
-            after_cond = index <= idx_after
-            bars: np.ndarray = frame[
-                before_cond
-                &
-                after_cond
-            ]
-            time: np.ndarray = bars['time']
-            from pendulum import (
-                from_timestamp,
-                Interval,
-            )
-            gap: Interval = (
-                from_timestamp(time[-1])
-                -
-                from_timestamp(time[0])
-            )
-            log.warning(
-                f'Single OHLCV-bar null-segment detected??\n'
-                f'gap -> {gap}\n'
+    # Null-run detection depends on absolute indexes stepping by
+    # exactly one per frame row. Without this invariant an
+    # index jump between adjacent null rows is indistinguishable from
+    # valid data rows separating two null segments.
+    if np.any(np.diff(frame_indexes) != 1):
+        raise ValueError(
+            'OHLCV frame indexes must be contiguous'
         )
 
-            # ^^XXX, if you want to debug the above bar-gap^^
-            # try:
-            #     breakpoint()
-            # except RuntimeError:
-            #     # XXX, if greenback not active from
-            #     # piker store ldshm cmd..
-            #     log.exception(
-            #         "Can't debug single-sample null!\n"
-            #     )
-
-            return None
-
-        # scan for all frame-indices where the
-        # zeroed-row-abs-index-step-diff is greater then the
-        # expected increment of 1.
-        # data  1st zero seg  data  zeros
-        # ----  ------------  ----  -----  ------  ----
-        # ||||..000000000000..||||..00000..||||||..0000
-        # ----  ------------  ----  -----  ------  ----
-        #       ^zero_t[0]                            ^zero_t[-1]
-        #                           ^fi_zgaps[0]   ^fi_zgaps[1]
-        #       ^absi_zsegs[0][0]   ^---^ => absi_zsegs[1]: tuple
-        #  absi_zsegs[0][1]^
+    # Scan zero-row absolute indexes for steps larger than one. Each
+    # such step begins another contiguous null segment:
     #
-        # NOTE: the first entry in `fi_zgaps` is where
-        # the first (absolute) index step diff is > 1.
-        # and it is a frame-relative index into `zero_t`.
-        fi_zgaps = np.argwhere(
-            absi_zdiff > 1
-            # NOTE: +1 here is ensure we index to the "start" of each
-            # segment (if we didn't the below loop needs to be
-            # re-written to expect `fi_end_rows`!
-        ) + 1
-        # the rows from the contiguous zeroed segments which have
-        # abs-index steps >1 compared to the previous zero row
-        # (indicating an end of zeroed segment).
-        fi_zseg_start_rows = zero_t[fi_zgaps]
-
-    # TODO: equiv for pl.DataFrame case!
-    else:
-        izeros: pl.Series = zero_pred.arg_true()
-        zero_t: pl.DataFrame = frame[izeros]
-
-        absi_zeros = zero_t['index']
-        absi_zdiff: pl.Series = absi_zeros.diff()
-        fi_zgaps = (absi_zdiff > 1).arg_true()
-
-    # XXX: our goal (in this func) is to select out slice index
-    # pairs (zseg0_start, zseg_end) in abs index units for each
-    # null-segment portion detected throughout entire input frame.
-
-    # only up to one null-segment in entire frame?
-    num_gaps: int = fi_zgaps.size + 1
-    if num_gaps < 1:
-        if absi_zeros.size > 1:
-            absi_zsegs = [[
-                # TODO: maybe mk these max()/min() limits func
-                # consts instead of called more then once?
-                max(
-                    absi_zeros[0] - 1,
-                    0,
-                ),
-                # NOTE: need the + 1 to guarantee we index "up to"
-                # the next non-null row-datum.
-                min(
-                    absi_zeros[-1] + 1,
-                    frame['index'][-1],
-                ),
-            ]]
-        else:
-            # XXX EDGE CASE: only one null-datum found so
-            # mark the start abs index as None to trigger
-            # a full frame-len query to the respective backend?
-            absi_zsegs = [[
-                # see `get_hist()` in backend, should ALWAYS be
-                # able to handle a `start_dt=None`!
-                # None,
-                None,
-                absi_zeros[0] + 1,
-            ]]
-
-    # XXX NOTE XXX: if >= 2 zeroed segments are found, there should
-    # ALWAYS be more then one zero-segment-abs-index-step-diff row
-    # in `absi_zdiff`, so loop through all such
-    # abs-index-step-diffs >1 (i.e. the entries of `absi_zdiff`)
-    # and add them as the "end index" entries for each segment.
-    # Then, iif NOT iterating the first such segment end, look back
-    # for the prior segments zero-segment start indext by relative
-    # indexing the `zero_t` frame by -1 and grabbing the abs index
-    # of what should be the prior zero-segment abs start index.
-    else:
-        # NOTE: since `absi_zdiff` will never have a row
-        # corresponding to the first zero-segment's row, we add it
-        # manually here.
-        absi_zsegs.append([
-            max(
-                absi_zeros[0] - 1,
-                0,
-            ),
-            None,
-        ])
-
-        # TODO: can we do it with vec ops?
-        for i, (
-            fi,  # frame index of zero-seg start
-            zseg_start_row,  # full row for ^
-        ) in enumerate(zip(
-            fi_zgaps,
-            fi_zseg_start_rows,
-        )):
-            assert (zseg_start_row == zero_t[fi]).all()
-            iabs: int = zseg_start_row['index'][0]
-            absi_zsegs.append([
-                iabs - 1,
-                None,  # backfilled on next iter
-            ])
-
-            # final iter case, backfill FINAL end iabs!
-            if (i + 1) == fi_zgaps.size:
-                absi_zsegs[-1][1] = absi_zeros[-1] + 1
-
-            # NOTE: only after the first segment (due to `.diff()`
-            # usage above) can we do a lookback to the prior
-            # segment's end row and determine it's abs index to
-            # retroactively insert to the prior
-            # `absi_zsegs[i-1][1]` entry Bo
-            last_end: int = absi_zsegs[i][1]
-            if last_end is None:
-                prev_zseg_row = zero_t[fi - 1]
-                absi_post_zseg = prev_zseg_row['index'][0] + 1
-                # XXX: MUST BACKFILL previous end iabs!
-                absi_zsegs[i][1] = absi_post_zseg
-
-        else:
-            if 0 < num_gaps < 2:
-                absi_zsegs[-1][1] = min(
-                    absi_zeros[-1] + 1,
-                    frame['index'][-1],
+    # data  1st zero seg  data  zeros   data    zeros
+    # ----  ------------  ----  -----   ------  -----
+    # ||||..000000000000..||||..00000...||||||..0000
+    # ----  ------------  ----  -----   ------  -----
+    #       ^zero_indexes[0]                  ^zero_indexes[-1]
+    #                           ^split_at[0]          ^split_at[1]
+    #       ^zero_groups[0]     ^zero_groups[1]       ^zero_groups[2]
+    #
+    # `split_at` entries are frame-relative positions in
+    # `zero_indexes`, not absolute buffer indexes. `np.split()` keeps
+    # each run together so its first and last absolute indexes can be
+    # expanded into the surrounding query boundary rows.
+    #
+    # Plain arrays also keep the NumPy and Polars paths on
+    # exactly the same segment-grouping implementation.
+    split_at: np.ndarray = (
+        np.flatnonzero(np.diff(zero_indexes) != 1)
+        +
+        1
+    )
+    zero_groups: list[np.ndarray] = np.split(
+        zero_indexes,
+        split_at,
     )
 
-            iabs_first: int = frame['index'][0]
-            for start, end in absi_zsegs:
-
-                ts_start: float = times[start - iabs_first]
-                ts_end: float = times[end - iabs_first]
-                if (
-                    (ts_start == 0 and not start == 0)
-                    or
-                    ts_end == 0
-                ):
-                    import pdbp
-                    pdbp.set_trace()
-
-                assert end
-                assert start < end
+    frame_start: int = int(frame_indexes[0])
+    frame_end: int = int(frame_indexes[-1])
+    absi_zsegs: list[list[int]] = []
+    for group in zero_groups:
+        zero_start: int = int(group[0])
+        zero_end: int = int(group[-1])
+        absi_zsegs.append([
+            max(frame_start, zero_start - imargin),
+            min(frame_end, zero_end + imargin),
+        ])
 
     log.warning(
         f'Frame has {len(absi_zsegs)} NULL GAPS!?\n'
@@ -492,12 +366,12 @@ def iter_null_segs(
     ],
     None,
 ]:
-    if not (
-        null_segs := get_null_segs(
+    if null_segs is None:
+        null_segs = get_null_segs(
             frame,
             period=timeframe,
         )
-    ):
+    if not null_segs:
         return
 
     absi_pairs_zsegs: list[list[float, float]]
@@ -609,7 +483,7 @@ def detect_time_gaps(
     # first select by any sample-period (in seconds unit) step size
     # greater then expected.
     step_gaps: pl.DataFrame = w_dts.filter(
-        pl.col('s_diff').abs() > expect_period
+        pl.col('s_diff') > expect_period
     )
 
     if gap_dt_unit is None:
@@ -638,6 +512,22 @@ def detect_time_gaps(
     )
 
 
+def detect_time_ordering_errors(
+    w_dts: pl.DataFrame,
+
+) -> pl.DataFrame:
+    '''
+    Return rows whose timestamp does not follow its predecessor.
+
+    Zero deltas identify duplicate timestamps and negative deltas
+    identify out-of-order rows. Neither is a missing-history gap.
+
+    '''
+    return w_dts.filter(
+        pl.col('s_diff') <= 0
+    )
+
+
 def detect_price_gaps(
     df: pl.DataFrame,
     gt_multiplier: float = 2.,
@@ -700,11 +590,16 @@ def dedupe(
 
     '''
     wdts: pl.DataFrame = with_dts(src_df)
-
-    deduped = wdts
+    ordering_errors = detect_time_ordering_errors(wdts)
+    if not ordering_errors.is_empty():
+        log.warning(
+            f'Found {ordering_errors.height} non-positive '
+            f'timestamp step(s) before normalization:\n'
+            f'{ordering_errors}'
+        )
 
     # remove duplicated datetime samples/sections
-    deduped: pl.DataFrame = wdts.unique(
+    deduped: pl.DataFrame = src_df.unique(
         # subset=['dt'],
         subset=['time'],
         maintain_order=True,
@@ -713,9 +608,8 @@ def dedupe(
     # maybe sort on any time field
     if sort:
         deduped = deduped.sort(by='time')
-        # TODO: detect out-of-order segments which were corrected!
-        # -[ ] report in log msg
-        # -[ ] possibly return segment sections which were moved?
+
+    deduped = with_dts(deduped)
 
     diff: int = (
         wdts.height

piker/tsp/_dedupe_smart.py

@@ -50,8 +50,18 @@ def dedupe_ohlcv_smart(
     )
 
     if dupes.is_empty():
-        # No duplicates, return as-is
-        return (wdts, wdts, 0, None, None)
+        deduped: pl.DataFrame = (
+            src_df.sort(time_col)
+            if sort
+            else src_df
+        )
+        return (
+            wdts,
+            with_dts(deduped),
+            0,
+            None,
+            None,
+        )
 
     # Analyze duplicate groups for validation
     dupe_analysis: pl.DataFrame = (
@@ -195,6 +205,8 @@ def dedupe_ohlcv_smart(
     if sort:
         deduped = deduped.sort(by=time_col)
 
+    deduped = with_dts(deduped.select(src_df.columns))
+
     diff: int = wdts.height - deduped.height
 
     return (

piker/tsp/_history.py

@@ -78,7 +78,6 @@ from piker.brokers._util import (
 )
 from piker.storage import TimeseriesNotFound
 from ._anal import (
-    dedupe,
     get_null_segs,
     iter_null_segs,
     Frame,
@@ -123,6 +122,7 @@ _default_rt_size: int = _days_worth * _secs_in_day
 # NOTE: start the append index in rt buffer such that 1 day's worth
 # can be appenened before overrun.
 _rt_buffer_start = int((_days_worth - 1) * _secs_in_day)
+_notify_timeout_s: float = 1
 
 
 def diff_history(
@@ -148,6 +148,64 @@ def diff_history(
         return array[times >= prepend_until_dt.timestamp()]
 
 
+def mk_storage_key(mkt: MktPair) -> str:
+    '''
+    Build the historical storage key for a market.
+
+    Always drop the source-asset token for non-currency-pair market
+    types. For historical reasons the table-key schema stores
+    `tsla.nasdaq.ib`, not `tsla/usd.nasdaq.ib`, while currency pairs
+    and crypto-settled futures retain both assets.
+
+    '''
+    if (
+        mkt.dst.atype not in {
+            'crypto',
+            'crypto_currency',
+            'fiat',  # a "forex pair"
+            'perpetual_future',  # stupid "perps" from cex land
+        }
+        and not (
+            mkt.src.atype == 'crypto_currency'
+            and
+            mkt.dst.atype == 'future'
+        )
+    ):
+        return mkt.get_fqme(
+            delim_char='',
+            without_src=True,
+        )
+
+    return mkt.get_fqme(delim_char='')
+
+
+async def notify_backfill(
+    sampler_stream: tractor.MsgStream,
+    mkt: MktPair,
+    timeframe: float,
+
+) -> None:
+    '''
+    Broadcast an SHM update without making teardown unbounded.
+
+    '''
+    with trio.move_on_after(
+        _notify_timeout_s,
+        shield=True,
+    ) as cs:
+        await sampler_stream.send({
+            'broadcast_all': {
+                'backfilling': (mkt.fqme, timeframe),
+            },
+        })
+
+    if cs.cancelled_caught:
+        log.warning(
+            f'Timed out broadcasting backfill for '
+            f'{timeframe}@{mkt.fqme}'
+        )
+
+
 async def shm_push_in_between(
     shm: ShmArray,
     to_push: np.ndarray,
@@ -202,12 +260,11 @@ async def maybe_fill_null_segments(
     mkt: MktPair,
     backfill_until_dt: datetime,
 
-    task_status: TaskStatus[trio.Event] = trio.TASK_STATUS_IGNORED,
+    task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
 
 ) -> list[Frame]:
 
-    null_segs_detected = trio.Event()
-    task_status.started(null_segs_detected)
+    task_status.started()
 
     frame: Frame = shm.array
 
@@ -308,7 +365,6 @@ async def maybe_fill_null_segments(
             await tractor.pause()
             raise
 
-    null_segs_detected.set()
     # RECHECK for more null-gaps
     frame: Frame = shm.array
     null_segs: tuple | None = get_null_segs(
@@ -407,15 +463,13 @@ async def start_backfill(
     storage: StorageClient|None = None,
     write_tsdb: bool = True,
 
-    task_status: TaskStatus[tuple] = trio.TASK_STATUS_IGNORED,
+    task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
 
 ) -> int:
 
-        # let caller unblock and deliver latest history frame
-        # and use to signal that backfilling the shm gap until
-        # the tsdb end is complete!
-        bf_done = trio.Event()
-        task_status.started(bf_done)
+        # Let the caller place storage history while this task
+        # continues reverse retrieval in the owning nursery.
+        task_status.started()
 
         # based on the sample step size, maybe load a certain amount history
         update_start_on_prepend: bool = False
@@ -535,6 +589,15 @@ async def start_backfill(
                 #     charge of solving such faults? yes, right?
                 return
 
+            if array.size == 0:
+                log.warning(
+                    f'Provider {mod.name!r} returned an empty frame\n'
+                    f'fqme: {mkt.fqme}\n'
+                    f'timeframe: {timeframe}\n'
+                    f'end_dt: {end_dt_param}\n'
+                )
+                break
+
             time: np.ndarray = array['time']
             assert (
                 time[0]
@@ -613,10 +676,11 @@ async def start_backfill(
                 )
                 # await tractor.pause()
 
-            to_push = diff_history(
+            to_store: np.ndarray = diff_history(
                 array,
                 prepend_until_dt=backfill_until_dt,
             )
+            to_push: np.ndarray = to_store
             ln: int = len(to_push)
             if ln:
                 log.info(
@@ -656,6 +720,23 @@ async def start_backfill(
                     )
                     break
 
+            if (
+                storage is not None
+                and
+                write_tsdb
+            ):
+                log.info(
+                    f'Writing {len(to_store)} frame to storage:\n'
+                    f'{next_start_dt} -> {last_start_dt}'
+                )
+                await storage.update_ohlcv(
+                    mk_storage_key(mkt),
+                    to_store,
+                    timeframe,
+                )
+
+            shm_full: bool = False
+
             # bail gracefully on shm allocation overrun/full
             # condition
             try:
@@ -666,11 +747,6 @@ async def start_backfill(
                     backfill_until_dt=backfill_until_dt,
                     update_start_on_prepend=update_start_on_prepend,
                 )
-                await sampler_stream.send({
-                    'broadcast_all': {
-                        'backfilling': (mkt.fqme, timeframe),
-                    },
-                })
 
                 # decrement next prepend point
                 next_prepend_index = next_prepend_index - ln
@@ -682,7 +758,7 @@ async def start_backfill(
                         f'Reached buffer start (index={next_prepend_index}), '
                         f'stopping backfill'
                     )
-                    break
+                    shm_full = True
 
             except ValueError as ve:
                 _ve = ve
@@ -697,6 +773,7 @@ async def start_backfill(
                     )
 
                 to_push = to_push[-next_prepend_index + 1:]
+                ln = len(to_push)
                 await shm_push_in_between(
                     shm,
                     to_push,
@@ -704,90 +781,24 @@ async def start_backfill(
                     backfill_until_dt=backfill_until_dt,
                     update_start_on_prepend=update_start_on_prepend,
                 )
-                await sampler_stream.send({
-                    'broadcast_all': {
-                        'backfilling': (mkt.fqme, timeframe),
-                    },
-                })
-
-                # XXX, can't push the entire frame? so
-                # push only the amount that can fit..
-                break
+                next_prepend_index = next_prepend_index - ln
+                last_start_dt = next_start_dt
+                shm_full = True
 
             log.info(
                 f'Shm pushed {ln} frame:\n'
                 f'{next_start_dt} -> {last_start_dt}'
             )
 
-            # FINALLY, maybe write immediately to the tsdb backend for
-            # long-term storage.
-            if (
-                storage is not None
-                and
-                write_tsdb
-            ):
-                log.info(
-                    f'Writing {ln} frame to storage:\n'
-                    f'{next_start_dt} -> {last_start_dt}'
-                )
-
-                # NOTE, always drop the src asset token for
-                # non-currency-pair like market types (for now)
-                #
-                # THAT IS, for now our table key schema is NOT
-                # including the dst[/src] source asset token. SO,
-                # 'tsla.nasdaq.ib' over 'tsla/usd.nasdaq.ib' for
-                # historical reasons ONLY.
-                if (
-                    mkt.dst.atype not in {
-                        'crypto',
-                        'crypto_currency',
-                        'fiat',  # a "forex pair"
-                        'perpetual_future',  # stupid "perps" from cex land
-                    }
-                    and not (
-                        mkt.src.atype == 'crypto_currency'
-                        and
-                        mkt.dst.atype in {
-                            'future',
-                        }
-                    )
-                ):
-                    col_sym_key: str = mkt.get_fqme(
-                        delim_char='',
-                        without_src=True,
-                    )
-                else:
-                    col_sym_key: str = mkt.get_fqme(
-                        delim_char='',
-                    )
-
-                await storage.write_ohlcv(
-                    col_sym_key,
-                    shm.array,
-                    timeframe,
-                )
-                df: pl.DataFrame = await storage.as_df(
-                    fqme=mkt.fqme,
-                    period=timeframe,
-                    load_from_offline=False,
-                )
-                (
-                    wdts,
-                    deduped,
-                    diff,
-                ) = dedupe(df)
-                if diff:
-                    log.warning(
-                        f'Found {diff!r} duplicates in tsdb! '
-                        f'=> Overwriting with `deduped` data !! <=\n'
-                    )
-                    await storage.write_ohlcv(
-                        col_sym_key,
-                        deduped,
+            await notify_backfill(
+                sampler_stream,
+                mkt,
                 timeframe,
             )
 
+            if shm_full:
+                break
+
         else:
             # finally filled gap
             log.info(
@@ -800,10 +811,6 @@ async def start_backfill(
         # memory...
         # await sampler_stream.send('broadcast_all')
 
-        # short-circuit (for now)
-        bf_done.set()
-
-
 # NOTE: originally this was used to cope with a tsdb (marketstore)
 # which could not delivery very large frames of history over gRPC
 # (thanks goolag) due to corruption issues.
@@ -949,11 +956,11 @@ async def back_load_from_tsdb(
         # await sampler_stream.send('broadcast_all')
 
 
-async def push_latest_frame(
+async def query_latest_frame(
     # box-type only that should get packed with the datetime
     # objects received for the latest history frame
     dt_eps: list[DateTime, DateTime],
-    shm: ShmArray,
+    frames: list[np.ndarray],
     get_hist: Callable[
         [int, datetime, datetime],
         tuple[np.ndarray, str]
@@ -982,7 +989,7 @@ async def push_latest_frame(
             mr_start_dt,
             mr_end_dt,
         ])
-        task_status.started(dt_eps)
+        frames.append(array)
 
     # XXX: timeframe not supported for backend (since
     # above exception type), terminate immediately since
@@ -996,18 +1003,38 @@ async def push_latest_frame(
         # prolly tf not supported
         return None
 
-    # NOTE: on the first history, most recent history
-    # frame we PREPEND from the current shm ._last index
-    # and thus a gap between the earliest datum loaded here
-    # and the latest loaded from the tsdb may exist!
+    task_status.started(dt_eps)
+
+    return dt_eps
+
+
+async def publish_latest_frame(
+    storage: StorageClient,
+    mkt: MktPair,
+    shm: ShmArray,
+    array: np.ndarray,
+    timeframe: float,
+
+) -> None:
+    '''
+    Persist then publish the most-recent provider frame.
+
+    '''
+    await storage.update_ohlcv(
+        mk_storage_key(mkt),
+        array,
+        timeframe,
+    )
+
+    # NOTE: on the first history, most recent history frame we
+    # PREPEND from the current shm ._last index. A gap may exist
+    # between its earliest datum and the latest loaded from the tsdb.
     log.info(f'Pushing {array.size} to shm!')
     shm.push(
         array,
         prepend=True,  # append on first frame
     )
 
-    return dt_eps
-
 
 async def load_tsdb_hist(
     storage: StorageClient,
@@ -1093,14 +1120,15 @@ async def tsdb_backfill(
         # concurrently load the provider's most-recent-frame AND any
         # pre-existing tsdb history already saved in `piker` storage.
         dt_eps: list[DateTime, DateTime] = []
+        latest_frames: list[np.ndarray] = []
         async with (
             tractor.trionics.collapse_eg(),
             trio.open_nursery() as tn
         ):
             tn.start_soon(
-                push_latest_frame,
+                query_latest_frame,
                 dt_eps,
-                shm,
+                latest_frames,
                 get_hist,
                 timeframe,
                 config,
@@ -1111,6 +1139,15 @@ async def tsdb_backfill(
                 timeframe,
             )
 
+        if latest_frames:
+            await publish_latest_frame(
+                storage,
+                mkt,
+                shm,
+                latest_frames[0],
+                timeframe,
+            )
+
         # tell parent task to continue
         # TODO: really we'd want this the other way with the
         # tsdb load happening asap and the since the latest
@@ -1190,7 +1227,7 @@ async def tsdb_backfill(
                 trio.open_nursery() as tn,
             ):
 
-                bf_done: trio.Event = await tn.start(
+                await tn.start(
                     partial(
                         start_backfill,
                         get_hist=get_hist,
@@ -1210,8 +1247,6 @@ async def tsdb_backfill(
                         write_tsdb=True,
                     )
                 )
-                nulls_detected: trio.Event|None = None
-
                 if last_tsdb_dt is not None:
 
                     # calc the index from which the tsdb data should be
@@ -1288,7 +1323,7 @@ async def tsdb_backfill(
                     #    work PREVENTAVELY instead?
                     # -[ ] fill in non-zero epoch time values ALWAYS!
                     # await maybe_fill_null_segments(
-                    nulls_detected: trio.Event = await tn.start(partial(
+                    await tn.start(partial(
                         maybe_fill_null_segments,
 
                         shm=shm,
@@ -1301,11 +1336,6 @@ async def tsdb_backfill(
 
             # 2nd nursery END
 
-            # TODO: who would want to?
-            if nulls_detected:
-                await nulls_detected.wait()
-
-            await bf_done.wait()
         # TODO: maybe start history anal and load missing "history
         # gaps" via backend..
 

plans/opencode/gap-checker-backfiller-repair.md

@@ -0,0 +1,436 @@
+# Gap Checker and Backfiller Repair
+
+## Purpose
+
+Continue the time-series gap checker and backfiller work
+from PRs/issues `#62`, `#71`, and `#75`, prioritizing data
+correctness and crash safety before annotation UX or
+rendering performance.
+
+Source baseline:
+
+- branch: `flake_update`
+- commit: `c6ec3d41`
+- gap handoff:
+  `notes_to_self/claudy/gapsannotator_sesh.md`
+- primary code:
+  `piker/tsp/_anal.py`, `piker/tsp/_history.py`,
+  `piker/storage/nativedb.py`, and
+  `piker/tsp/_annotate.py`
+
+## Goals
+
+1. Make gap detection operate on sorted, unique,
+   validated OHLCV frames with fresh derived columns.
+2. Make backfill completion, frame boundaries, and null
+   repair deterministic under exhaustion and cancellation.
+3. Prevent a bounded or transient SHM view from
+   destructively replacing older persisted history.
+4. Persist normalized history through crash-safe atomic
+   file replacement.
+5. Distinguish expected venue closures from suspicious
+   missing data without production breakpoints.
+6. Generate correct annotation specs before optimizing
+   the remaining client-side Polars bottleneck.
+
+## Confirmed Baseline Defects
+
+These are current-code findings to characterize in tests,
+not assumptions inferred from historical task state.
+
+### Gap normalization
+
+- `dedupe()` computes `time_prev`, `s_diff`, and
+  `dt_diff` before sorting/deduplication, then returns
+  stale derived columns in the normalized frame
+  (`piker/tsp/_anal.py:683-729`).
+- Negative/out-of-order deltas are treated as ordinary
+  gaps because `detect_time_gaps()` filters on
+  `abs(s_diff)` (`piker/tsp/_anal.py:582-638`).
+- `get_null_segs()` drops single zero rows and its Polars
+  path does not define all state used by the shared path
+  (`piker/tsp/_anal.py:227-471`).
+- `iter_null_segs()` ignores a supplied `null_segs`
+  result and rescans the frame (`piker/tsp/_anal.py:474-546`).
+
+### Backfill control and boundaries
+
+- `DataUnavailable` returns before `bf_done.set()`, while
+  `tsdb_backfill()` later waits forever on that event
+  (`piker/tsp/_history.py:516-536,1193-1212,1302-1308`).
+- Empty arrays are indexed before the nominal null-frame
+  branch can handle them (`piker/tsp/_history.py:538-601`).
+- Provider frame seams have no explicit inclusive/exclusive
+  contract, allowing duplicate adjacent timestamps in SHM.
+- Cancellation may occur after SHM mutation but before
+  cursor and persistence bookkeeping
+  (`piker/tsp/_history.py:661-789`).
+
+### Null repair
+
+- Null query length is derived from an absolute endpoint,
+  not the null-span width (`piker/tsp/_history.py:278-287`).
+- Fallback repair includes and mutates the valid leading
+  boundary row (`piker/tsp/_history.py:345-367`).
+- Null repair and reverse backfill share one provider
+  client concurrently without enforcing the backend's
+  declared request limits.
+
+### Persistence
+
+- Every iteration passes bounded `shm.array` to
+  `write_ohlcv()`, while NativeDB replaces the complete
+  parquet file. Older rows that do not fit in SHM can be
+  lost (`piker/tsp/_history.py:722-789`,
+  `piker/storage/nativedb.py:305-368`).
+- NativeDB writes directly to the final path and updates
+  its cache before the write succeeds; interruption can
+  leave disk and cache partial or inconsistent.
+- Intentional zero-filled SHM space may be persisted while
+  null repair and reverse backfill run concurrently.
+
+### Annotation contract
+
+- `markup_gaps()` performs a full Polars filter per gap and
+  remains the measured client bottleneck
+  (`piker/tsp/_annotate.py:139-217`).
+- Its missing-prior-row fallback selects the current row,
+  and `copysign(1, 0)` makes the flat-price branch
+  unreachable (`piker/tsp/_annotate.py:166-239`).
+- Remote/reposition timestamp lookups index
+  `searchsorted()` results before excluding values equal
+  to `len(array)`.
+
+## Design Invariants
+
+All implementation phases preserve these rules:
+
+1. Persisted provider rows have nonzero, finite timestamps.
+2. Normalized timestamps are strictly increasing and
+   unique.
+3. Time-derived columns are computed only after final
+   sorting and duplicate resolution.
+4. A detected gap explicitly carries its left and right
+   endpoints; consumers do not infer the left row from
+   `index - 1`.
+5. Provider frame metadata equals the normalized array's
+   actual first and last timestamps.
+6. Internal middleware uses one documented boundary
+   convention; provider adapters translate into it.
+7. Expected sparse venue closures are represented by
+   timestamp distance, not persisted zero or synthetic
+   bars.
+8. Storage merges preserve all unaffected older rows.
+9. A storage commit becomes visible atomically, and cache
+   state changes only after disk commit succeeds.
+10. Rendering or annotation failure cannot change the
+    checker or persistence result.
+
+## Phase 0: Deterministic Characterization
+
+Add synthetic OHLCV fixtures based on
+`piker.data._source.def_iohlcv_fields`; avoid brokers,
+networking, real actor trees, and Qt in the core suite.
+
+### Analysis tests
+
+Create `tests/test_tsp_analysis.py` covering:
+
+- uniform 1-second and 60-second series;
+- one and multiple positive gaps;
+- duplicate plus out-of-order timestamps;
+- stale derived columns after sort/dedupe;
+- zero, one, and multiple null segments;
+- nulls at frame boundaries and nonzero absolute indexes;
+- NumPy and Polars inputs;
+- smart-dedupe conflict cases, including differing closes.
+
+### Backfill tests
+
+Create `tests/test_history_backfill.py` with scripted
+`get_hist`, fake SHM, fake sampler stream, and recording
+storage implementations. Characterize:
+
+- `DataUnavailable` completion without a hang;
+- a genuinely empty frame;
+- exact adjacency, overlap, and a positive storage gap;
+- inclusive provider seam duplication;
+- one-row and two-row startup history;
+- cancellation after query, after SHM mutation, and
+  before persistence;
+- one and multiple null rows with boundary preservation.
+
+### Storage tests
+
+Create `tests/test_storage_nativedb.py` using `tmp_path`:
+
+- NumPy and Polars round trips;
+- preservation of history older than the SHM window;
+- duplicate conflict policy;
+- write failure leaves the prior parquet readable;
+- cache changes only after a successful commit;
+- separate FQME/timeframe series do not interfere.
+
+### Annotation tests
+
+Create `tests/test_gap_annotations.py` around a fake
+`AnnotCtl`, without Qt initially:
+
+- left-close/right-open rectangle endpoints;
+- up, down, and flat price direction;
+- first-row and missing-prior-row behavior;
+- empty gaps and out-of-range timestamps;
+- deterministic batched rect/arrow/text payloads.
+
+Gate: every confirmed defect above has a focused failing
+test or a written reason it requires a later integration
+test.
+
+## Phase 1: Canonical Frame and Gap Model
+
+Refactor `piker/tsp/_anal.py` around one normalization
+order:
+
+1. validate required columns and finite timestamps;
+2. sort by `time`;
+3. resolve duplicate timestamps under an explicit policy;
+4. recompute all datetime/delta columns with `with_dts()`;
+5. reject/report non-positive ordering deltas;
+6. detect positive sample-step gaps.
+
+Keep the first implementation small: a normalized Polars
+frame may remain the contract, but each gap row must carry
+explicit `time_prev`, `index_prev`, `close_prev`, current
+`time/index/open`, duration, and missing-sample count.
+
+Rewrite null grouping from extracted index/time arrays so
+the NumPy and Polars paths share one implementation. Treat
+a single null row as a normal one-row segment and honor a
+precomputed `null_segs` argument without rescanning.
+
+Gate: analysis tests pass and no deduped/sorted frame
+retains stale delta columns.
+
+## Phase 2: Structured Backfill Completion
+
+Refactor `start_backfill()` and `tsdb_backfill()` before
+changing persistence semantics:
+
+- remove completion events that are awaited only after the
+  owning nursery has already joined its children, or make
+  completion a guaranteed `finally` outcome;
+- represent `completed`, `provider_exhausted`,
+  `cancelled`, and `failed` as explicit results;
+- check empty arrays before indexing endpoints;
+- normalize each provider frame and recompute endpoint
+  metadata from the normalized data;
+- define and enforce a disjoint seam between each new
+  reverse frame and the already-published frame;
+- update cursor state immediately with the SHM mutation,
+  before any optional notification checkpoint;
+- serialize reverse backfill and null repair unless a
+  backend capability explicitly permits concurrent
+  history requests;
+- enforce declared backend request/rate limits.
+
+Prefer structured task return values over actor-local
+events whose only consumer is the parent task.
+
+Gate: exhaustion and cancellation tests terminate within
+their timeout, and SHM timestamps remain strictly
+increasing across all frame seams.
+
+## Phase 3: Non-Destructive Atomic NativeDB Writes
+
+Stop treating the bounded SHM view as the full database.
+
+Add a NativeDB merge/update operation that:
+
+- accepts only the normalized provider frame being
+  committed;
+- loads or uses the cached persisted frame;
+- merges by timestamp while preserving unaffected rows;
+- uses the historical provider bar as authoritative at a
+  closed-bar seam;
+- validates ordering, uniqueness, schema, and nonzero
+  timestamps before writing;
+- writes to a same-directory temporary parquet;
+- reopens and validates the temporary file;
+- atomically replaces the final path;
+- updates `_dfs` only after replacement succeeds;
+- serializes writers per `(fqme, timeframe)`.
+
+Backfill commits the validated frame/delta, never a live
+`shm.array` snapshot containing mutable current bars or
+intentional zero space.
+
+Gate: a persisted series can only grow backward or replace
+explicitly overlapping timestamps; its earliest timestamp
+never moves forward during ordinary backfill.
+
+## Phase 4: Exact Null Repair
+
+Use the canonical null spans from Phase 1:
+
+- query using explicit boundary timestamps;
+- trim provider responses to the actual missing interval;
+- write only null rows, never either valid boundary row;
+- validate repaired timestamps and OHLC fields before
+  publication or persistence;
+- leave expected venue closures sparse;
+- do not persist synthetic flat bars without a durable
+  quality marker and an explicit schema decision.
+
+If provider data is unavailable, retain and report the
+missing interval instead of silently converting it into
+apparently valid history.
+
+Gate: single and multi-row repairs preserve both boundary
+bars and either eliminate exactly the target nulls or
+return an unresolved-gap result.
+
+## Phase 5: Venue and Provider Classification
+
+Apply gap classification after canonical detection rather
+than embedding it in individual warning paths.
+
+For IB:
+
+- correct expiry comparison and weekday constants;
+- classify internal positive gaps as well as trailing
+  request-coverage gaps;
+- require closure interval alignment, not merely the
+  presence of a weekend/holiday somewhere in the gap;
+- cover overnight sessions, weekends, holidays, DST,
+  active/expired futures, and unknown calendars;
+- replace production `breakpoint()` calls with structured
+  data-quality results/errors.
+
+Provider adapters must return normalized endpoint metadata
+and translate their native inclusive/exclusive APIs into
+the middleware boundary contract.
+
+Gate: every gap is classified as expected closure,
+suspicious missing data, provider exhaustion, or invalid
+ordering; no production path pauses interactively.
+
+## Phase 6: Annotation Correctness and Performance
+
+Extract a synchronous, Qt-free spec builder from
+`markup_gaps()`:
+
+```python
+build_gap_annotation_specs(frame, gaps)
+```
+
+It consumes the explicit gap endpoints from Phase 1 and:
+
+- extracts needed columns once;
+- replaces per-gap `wdts.filter()` with a vectorized join
+  or indexed lookup;
+- handles equal prices explicitly;
+- preserves previous-close/current-open semantics;
+- styles expected closures separately from suspicious
+  gaps;
+- emits deterministic rect/arrow/text batches.
+
+Then fix remote lookup/reposition bounds by masking
+`searchsorted()` results before indexing and enforce both
+FQME and timeframe ownership during redraw.
+
+Benchmark the same 1,285-gap workload from the Claude
+handoff. Treat these as local benchmark goals, not portable
+CI limits:
+
+- client spec build under 20 ms;
+- server preparation under 10 ms;
+- one composite gap graphics item;
+- no full-frame Polars filter per gap;
+- total local creation near or below 150 ms.
+
+Gate: annotation contract tests pass before performance
+numbers are accepted.
+
+## Phase 7: Crash and Provider Qualification
+
+Run deterministic tests first, then the historical manual
+matrix from issue `#62` on disposable storage:
+
+- fresh full backfill;
+- append to existing history;
+- graceful Ctrl-C;
+- network interruption;
+- SIGTERM and SIGKILL during query and parquet commit;
+- restart and revalidate after every interruption.
+
+Start with Binance/Kraken/Kucoin, then run IB only with an
+available account and gateway. Record before/after earliest
+timestamp, latest timestamp, row count, unique count, null
+count, non-positive deltas, and parquet readability.
+
+## Verification Commands
+
+Use an outer process timeout because actor teardown has a
+known second-runtime wedge:
+
+```bash
+timeout -k 5 30 python -m pytest -q \
+  tests/test_tsp_analysis.py \
+  tests/test_storage_nativedb.py
+
+timeout -k 5 60 python -m pytest -q \
+  tests/test_history_backfill.py
+
+timeout -k 5 60 python -m pytest -q \
+  tests/test_gap_annotations.py
+
+CI=1 timeout -k 20 120 python -m pytest -q -rs \
+  tests/test_feeds.py
+```
+
+Run live/network tests separately from deterministic tests.
+Do not automate `piker store anal` or `piker store ldshm`:
+both may pause interactively or mutate attached data.
+
+## Commit Boundaries
+
+Keep implementation reviewable in this order:
+
+1. synthetic fixtures and analysis regressions;
+2. canonical normalization and null grouping;
+3. backfill completion and frame-boundary fixes;
+4. NativeDB merge and atomic persistence;
+5. exact null repair;
+6. IB/provider gap classification;
+7. annotation contract fixes and vectorization;
+8. qualification notes and documentation.
+
+Each boundary must leave deterministic tests green. Do not
+mix renderer optimization into storage/backfill commits.
+
+## Deferred
+
+The following remain follow-ups until the correctness
+gates above pass:
+
+- chart context-menu/manual piecewise refill UX;
+- automatic checker integration into every live chart;
+- compact columnar annotation IPC;
+- persistent synthetic-bar quality metadata/schema;
+- full Deribit history repair and broad provider cleanup;
+- Marketstore behavior changes;
+- long-term incremental parquet partitioning.
+
+## Source References
+
+- `gitea/62.md`: workaround-era backfill and manual test
+  history
+- `gitea/71.md`: IB venue closure behavior and pending
+  account verification
+- `gitea/75.md`: gap annotation, storage races, and
+  backfiller follow-ups
+- `notes_to_self/claudy/gapsannotator_sesh.md`: current
+  annotation benchmark and client bottleneck
+- https://github.com/pikers/piker/pull/486
+- https://github.com/pikers/piker/pull/446
+- https://github.com/pikers/piker/issues/536

plans/opencode/known-fqme-backfill-qualification.md

@@ -0,0 +1,221 @@
+# Known-FQME Backfill Qualification
+
+## Purpose
+
+Move `backfiller_deep_fixes` into the main checkout, reproduce the
+historical backfill failures against one known-problem market, and turn
+every confirmed failure into deterministic regression coverage before
+expanding live qualification.
+
+This is a correctness-debugging pass, not a broad provider smoke test.
+One FQME and one timeframe remain fixed until their original failure is
+understood, covered, repaired, and requalified.
+
+## Current Repository State
+
+The branch is currently checked out at:
+
+`/home/goodboy/repos/piker/.claude/wkts/backfiller_deep_fixes`
+
+Its head is `10e0ced7`. The linked worktree has only generated,
+untracked commit-message metadata.
+
+The main checkout is on `flake_update` at `c6ec3d41` and has staged and
+untracked user work. Do not switch branches, stash, remove files, or
+close the linked worktree until that state has been explicitly
+preserved by the user.
+
+## Phase 1: Transfer Branch Ownership Safely
+
+1. Review the main checkout's staged and untracked state.
+2. Preserve it using the user's chosen mechanism: finish and commit it,
+   move it to another branch/worktree, or explicitly create a stash that
+   includes the required untracked files.
+3. Review the linked worktree's generated `.claude/skills/commit-msg/`
+   artifacts. Archive or discard them explicitly; they are not branch
+   source.
+4. Remove the linked `backfiller_deep_fixes` worktree while keeping its
+   exact branch.
+5. Switch the main checkout to `backfiller_deep_fixes`.
+6. Verify all entry points import from the main checkout and resolve to
+   head `10e0ced7` or its descendant.
+7. Run the 31-test deterministic analysis/backfill/storage set before
+   booting any daemon.
+
+Gate: the main checkout owns `backfiller_deep_fixes`, tracked status is
+clean, user work from `flake_update` is preserved, and deterministic
+tests pass.
+
+## Phase 2: Record The Known Failure
+
+Capture these details for the selected case:
+
+- provider and exact FQME;
+- timeframe, initially `60` or `1` seconds;
+- known-bad date/time range and timezone;
+- original symptom: destructive truncation, duplicate seam, null rows,
+  out-of-order timestamps, hang, or another observed failure;
+- command or chart workflow that exposed it;
+- whether a known-bad parquet already exists;
+- expected venue closure or continuous-session behavior;
+- credentials, gateway, and network requirements;
+- smallest reproducible provider response or persisted frame.
+
+Do not start with multiple symbols, providers, or timeframes. Preserve a
+checksum of any known-bad parquet before copying it into disposable
+storage.
+
+Gate: one case has an explicit expected result and enough source data to
+distinguish a real provider/venue gap from a piker persistence defect.
+
+## Phase 3: Build A Disposable Runtime
+
+Use the main checkout's Python environment and a dedicated XDG config
+root. The `pikerd` process and every client process must inherit the same
+config root and import the main checkout.
+
+The disposable directory contains only:
+
+- a minimal `conf.toml`;
+- credentials required for the selected provider;
+- a dedicated `nativedb/` directory;
+- an optional copy of the single known-bad parquet;
+- captured logs and before/after audit output.
+
+Ensure no older `pikerd` is listening on the selected registry address.
+Verify `piker.__file__`, the branch, and the commit before starting the
+daemon.
+
+Gate: deleting the disposable directory cannot affect normal user
+configuration, accounting data, or production history.
+
+## Phase 4: Establish A Read-Only Baseline
+
+Before backfill, record:
+
+- parquet checksum and readability;
+- earliest and latest timestamp;
+- total and unique timestamp counts;
+- duplicate count;
+- zero and non-finite timestamp counts;
+- non-positive timestamp deltas;
+- null OHLCV counts;
+- contiguous persisted index status;
+- positive sample gaps with explicit endpoints.
+
+Use a read-only audit helper or focused Python test. Do not automate
+`piker store anal` or `piker store ldshm`; both can pause or mutate
+attached data.
+
+Gate: the original symptom is visible in saved evidence, or the plan
+records why a fresh provider fetch is required to reproduce it.
+
+## Phase 5: Reproduce One Lifecycle At A Time
+
+Run in this order:
+
+1. Load the known-bad persisted history without a provider write.
+2. Start `pikerd` and request the chosen FQME/timeframe only.
+3. Observe startup latest-frame query, pre-update TSDB snapshot, reverse
+   backfill, provider-delta commit, SHM publication, and notification.
+4. Stop cleanly and rerun the read-only audit.
+5. Restart and append to the same disposable history.
+6. Compare earliest/latest timestamps, row counts, conflicts, and gaps
+   against the baseline.
+
+Capture logs around every `update_ohlcv()` call, provider frame boundary,
+SHM truncation, lock acquisition, temporary parquet, replacement, and
+cache publication.
+
+Gate: the run either preserves every unaffected row and normalizes the
+known defect, or yields a bounded failure with enough data for a test.
+
+## Phase 6: Convert Failures Into Tests
+
+For each reproduced defect:
+
+1. Save the smallest provider frames and persisted rows needed to
+   reproduce it. Prefer synthetic arrays; use a trimmed fixture only
+   when provider-specific boundaries matter.
+2. Write a failing deterministic test before changing production code.
+3. Assert exact timestamps, endpoint policy, persisted rows, and state
+   transition ordering rather than only asserting completion.
+4. Keep actor-free tests in the existing focused modules where possible.
+5. Add subprocess integration tests only for process death, advisory
+   locks, or atomic filesystem behavior that cannot be represented in
+   one Trio runtime.
+6. Apply the smallest production fix and rerun both the focused test and
+   the complete deterministic set.
+7. Rerun the exact manual FQME lifecycle that found the failure.
+
+Potential deterministic modules:
+
+- `tests/test_tsp_analysis.py` for normalization and classification;
+- `tests/test_history_backfill.py` for provider/SHM/storage ordering;
+- `tests/test_storage_nativedb.py` for merge and atomic durability;
+- a new subprocess fault module for kill/restart and cross-process locks;
+- provider-specific fixtures only when generic frames cannot reproduce
+  the behavior.
+
+Gate: no production fix lands without a failing-before, passing-after
+test or a written explanation for an unavoidable live-only condition.
+
+## Phase 7: Fault Qualification
+
+After the normal lifecycle is stable, exercise:
+
+1. graceful Ctrl-C during provider query and storage update;
+2. SIGTERM at deterministic query/write barriers;
+3. SIGKILL after temp write, before replace, and after replace;
+4. provider disconnect and recovery;
+5. concurrent NativeDB writers for the same series;
+6. restart after every interruption;
+7. actual sampler backpressure and actor teardown;
+8. MarketStore separately from NativeDB.
+
+Use explicit barriers or failpoints rather than timing sleeps whenever
+the failure position matters. Repeat the read-only audit after every
+restart.
+
+Gate: the final parquet is always old-valid or new-valid, unaffected
+history remains present, indexes are contiguous, cache agrees with the
+visible file, and shutdown remains bounded.
+
+## Expansion Order
+
+Only after the chosen FQME passes:
+
+1. qualify its second timeframe;
+2. qualify another continuous-session provider;
+3. run Binance, Kraken, and Kucoin cases;
+4. test IB with an available account and gateway;
+5. add expected venue-closure classification;
+6. widen to the remaining manual matrix.
+
+## Commit Boundaries
+
+Keep follow-up commits narrow:
+
+1. captured fixture and failing regression;
+2. minimal analysis/backfill/storage correction;
+3. subprocess fault-injection harness;
+4. provider-specific adapter correction;
+5. qualification notes and reproducible audit output.
+
+Every boundary must leave deterministic tests green. Do not combine
+annotation performance, unrelated providers, or chart UX with the known
+FQME debugging loop.
+
+## Exit Criteria
+
+The selected FQME/timeframe is complete when:
+
+- the original bug reproduces from preserved evidence;
+- a deterministic regression covers it where technically possible;
+- fresh and append backfills preserve unaffected history;
+- provider seam conflicts follow the documented policy;
+- no zero, duplicate, non-finite, or non-positive timestamps persist;
+- indexes remain contiguous;
+- clean shutdown and restart preserve readable parquet;
+- failure injection leaves an old-valid or new-valid file;
+- remaining live-only risks are explicitly recorded.

tests/test_history_backfill.py

@@ -0,0 +1,221 @@
+'''
+Deterministic history-backfill regressions.
+
+'''
+from functools import partial
+from types import SimpleNamespace
+
+import numpy as np
+from pendulum import (
+    datetime,
+    from_timestamp,
+)
+import pytest
+import trio
+
+from piker.brokers import DataUnavailable
+from piker.data._source import def_iohlcv_fields
+from piker.tsp._history import (
+    notify_backfill,
+    publish_latest_frame,
+    start_backfill,
+)
+
+
+def run_exhausted_backfill(get_hist) -> None:
+    '''
+    Run a provider-exhaustion case without actor services.
+
+    '''
+    async def main() -> None:
+        with trio.fail_after(0.5):
+            async with trio.open_nursery() as nursery:
+                await nursery.start(partial(
+                    start_backfill,
+                    get_hist=get_hist,
+                    def_frame_duration=None,
+                    mod=SimpleNamespace(name='fake'),
+                    mkt=SimpleNamespace(fqme='x.fake'),
+                    shm=object(),
+                    timeframe=60,
+                    backfill_from_shm_index=100,
+                    backfill_from_dt=datetime(2026, 1, 2),
+                    sampler_stream=object(),
+                    backfill_until_dt=datetime(2026, 1, 1),
+                    storage=None,
+                    write_tsdb=False,
+                ))
+
+    trio.run(main)
+
+
+def test_data_unavailable_completes() -> None:
+    '''
+    Provider exhaustion exits without orphaned completion waits.
+
+    '''
+    async def get_hist(*args, **kwargs):
+        raise DataUnavailable('history exhausted')
+
+    run_exhausted_backfill(get_hist)
+
+
+def test_empty_frame_completes() -> None:
+    '''
+    An empty provider frame is checked before endpoint indexing.
+
+    '''
+    frame = np.empty(
+        0,
+        dtype=np.dtype(def_iohlcv_fields),
+    )
+
+    async def get_hist(*args, **kwargs):
+        end_dt = kwargs['end_dt']
+        return frame, end_dt, end_dt
+
+    run_exhausted_backfill(get_hist)
+
+
+def test_storage_receives_full_provider_delta() -> None:
+    '''
+    SHM capacity truncation does not truncate durable history.
+
+    '''
+    frame = np.zeros(
+        2,
+        dtype=np.dtype(def_iohlcv_fields),
+    )
+    frame['index'] = [0, 1]
+    frame['time'] = [60, 120]
+
+    events: list[str] = []
+
+    class Shm:
+        def __init__(self) -> None:
+            self.pushed: list[np.ndarray] = []
+
+        def push(self, array, **kwargs) -> None:
+            self.pushed.append(array.copy())
+            events.append('shm')
+
+    class Sampler:
+        async def send(self, msg) -> None:
+            events.append('sampler')
+
+    class Storage:
+        def __init__(self) -> None:
+            self.frames: list[np.ndarray] = []
+
+        async def update_ohlcv(
+            self,
+            fqme: str,
+            ohlcv: np.ndarray,
+            timeframe: int,
+        ) -> None:
+            self.frames.append(ohlcv.copy())
+            events.append('storage')
+
+    shm = Shm()
+    storage = Storage()
+    mkt = SimpleNamespace(
+        fqme='x.test',
+        dst=SimpleNamespace(atype='crypto'),
+        src=SimpleNamespace(atype='crypto_currency'),
+        get_fqme=lambda **kwargs: 'x.test',
+    )
+
+    async def get_hist(*args, **kwargs):
+        return (
+            frame,
+            from_timestamp(60),
+            from_timestamp(120),
+        )
+
+    async def main() -> None:
+        with trio.fail_after(0.5):
+            await start_backfill(
+                get_hist=get_hist,
+                def_frame_duration=None,
+                mod=SimpleNamespace(name='fake'),
+                mkt=mkt,
+                shm=shm,
+                timeframe=60,
+                backfill_from_shm_index=1,
+                backfill_from_dt=from_timestamp(180),
+                sampler_stream=Sampler(),
+                backfill_until_dt=from_timestamp(60),
+                storage=storage,
+                write_tsdb=True,
+            )
+
+    trio.run(main)
+    assert shm.pushed[0]['time'].tolist() == [120]
+    assert storage.frames[0]['time'].tolist() == [60, 120]
+    assert events == ['storage', 'shm', 'sampler']
+
+
+def test_latest_frame_is_persisted_before_shm() -> None:
+    '''
+    Startup persists the most-recent provider frame before publication.
+
+    '''
+    frame = np.zeros(
+        2,
+        dtype=np.dtype(def_iohlcv_fields),
+    )
+    frame['time'] = [60, 120]
+    events: list[str] = []
+
+    class Storage:
+        async def update_ohlcv(self, *args) -> None:
+            events.append('storage')
+
+    class Shm:
+        def push(self, array, **kwargs) -> None:
+            events.append('shm')
+
+    async def main() -> None:
+        with trio.fail_after(0.5):
+            await publish_latest_frame(
+                storage=Storage(),
+                mkt=SimpleNamespace(
+                    fqme='x.test',
+                    dst=SimpleNamespace(atype='crypto'),
+                    src=SimpleNamespace(atype='crypto_currency'),
+                    get_fqme=lambda **kwargs: 'x.test',
+                ),
+                shm=Shm(),
+                array=frame,
+                timeframe=60,
+            )
+
+    trio.run(main)
+    assert events == ['storage', 'shm']
+
+
+def test_backfill_notification_timeout_is_bounded(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A wedged sampler can not indefinitely shield actor teardown.
+
+    '''
+    monkeypatch.setattr(
+        'piker.tsp._history._notify_timeout_s',
+        0.01,
+    )
+
+    class Sampler:
+        async def send(self, msg) -> None:
+            await trio.sleep_forever()
+
+    async def main() -> None:
+        with trio.fail_after(0.1):
+            await notify_backfill(
+                Sampler(),
+                SimpleNamespace(fqme='x.test'),
+                60,
+            )
+
+    trio.run(main)

tests/test_storage_nativedb.py

@@ -0,0 +1,364 @@
+'''
+NativeDB durability and history-preservation regressions.
+
+'''
+from fcntl import (
+    flock,
+    LOCK_EX,
+    LOCK_UN,
+)
+import os
+from pathlib import Path
+
+import numpy as np
+import polars as pl
+import pytest
+import trio
+
+from piker import tsp
+from piker.data._source import def_iohlcv_fields
+from piker.storage.nativedb import NativeStorageClient
+
+
+def mk_ohlcv(
+    times: tuple[float, ...],
+    closes: tuple[float, ...] | None = None,
+
+) -> np.ndarray:
+    '''
+    Build a minimal structured OHLCV frame.
+
+    '''
+    array = np.zeros(
+        len(times),
+        dtype=np.dtype(def_iohlcv_fields),
+    )
+    array['index'] = np.arange(len(times))
+    array['time'] = times
+    array['close'] = closes or times
+    return array
+
+
+def run(coro) -> None:
+    '''
+    Run a NativeDB operation with a bounded Trio clock.
+
+    '''
+    async def main() -> None:
+        with trio.fail_after(1):
+            await coro
+
+    trio.run(main)
+
+
+@pytest.mark.parametrize('use_polars', [False, True])
+def test_numpy_and_polars_round_trip(
+    tmp_path: Path,
+    use_polars: bool,
+) -> None:
+    '''
+    Both supported frame types survive durable serialization.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    array = mk_ohlcv(
+        (60, 120, 180),
+        (1, 2, 3),
+    )
+    payload = tsp.np2pl(array) if use_polars else array
+
+    run(client.write_ohlcv('x.test', payload, 60))
+    loaded = trio.run(client.read_ohlcv, 'x.test', 60)
+
+    assert loaded['time'].tolist() == [60, 120, 180]
+    assert loaded['close'].tolist() == [1, 2, 3]
+
+
+def test_update_preserves_history_and_resolves_conflicts(
+    tmp_path: Path,
+) -> None:
+    '''
+    Incremental writes retain old rows and prefer incoming bars.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    old = mk_ohlcv(
+        (60, 120, 180),
+        (1, 2, 3),
+    )
+    incoming = mk_ohlcv(
+        (180, 240),
+        (30, 4),
+    )
+
+    run(client.write_ohlcv('x.test', old, 60))
+    run(client.update_ohlcv('x.test', incoming, 60))
+    stored = pl.read_parquet(client.mk_path('x.test', 60))
+
+    assert stored['time'].to_list() == [60, 120, 180, 240]
+    assert stored['close'].to_list() == [1, 2, 30, 4]
+    assert stored['index'].to_list() == [0, 1, 2, 3]
+
+
+def test_write_ohlcv_remains_an_explicit_replacement(
+    tmp_path: Path,
+) -> None:
+    '''
+    Full repair writes can intentionally remove persisted rows.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    old = mk_ohlcv((60, 120, 180))
+    replacement = mk_ohlcv((120, 180))
+
+    run(client.write_ohlcv('x.test', old, 60))
+    run(client.write_ohlcv('x.test', replacement, 60))
+    stored = pl.read_parquet(client.mk_path('x.test', 60))
+
+    assert stored['time'].to_list() == [120, 180]
+
+
+def test_failed_write_preserves_file_and_cache(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A failed parquet write leaves durable and cached data intact.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    old = mk_ohlcv((60, 120))
+    incoming = mk_ohlcv((180,))
+    run(client.write_ohlcv('x.test', old, 60))
+
+    def fail_write(
+        df: pl.DataFrame,
+        file: Path,
+        *args,
+        **kwargs,
+    ) -> None:
+        Path(file).write_bytes(b'partial parquet')
+        raise OSError('simulated write failure')
+
+    monkeypatch.setattr(
+        pl.DataFrame,
+        'write_parquet',
+        fail_write,
+    )
+    with pytest.raises(OSError, match='simulated write failure'):
+        run(client.update_ohlcv('x.test', incoming, 60))
+
+    stored = pl.read_parquet(client.mk_path('x.test', 60))
+    cached = trio.run(client.as_df, 'x.test', 60, False)
+    assert stored['time'].to_list() == [60, 120]
+    assert cached['time'].to_list() == [60, 120]
+    assert not list(tmp_path.glob('*.tmp'))
+
+
+def test_invalid_temporary_parquet_is_not_committed(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    The reopened candidate must validate before atomic replacement.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    old = mk_ohlcv((60, 120))
+    run(client.write_ohlcv('x.test', old, 60))
+    real_read = pl.read_parquet
+
+    def corrupt_candidate(source, *args, **kwargs):
+        if Path(source).suffix == '.tmp':
+            return pl.DataFrame({'time': [180]})
+        return real_read(source, *args, **kwargs)
+
+    monkeypatch.setattr(pl, 'read_parquet', corrupt_candidate)
+    with pytest.raises(ValueError, match='missing columns'):
+        run(client.update_ohlcv(
+            'x.test',
+            mk_ohlcv((180,)),
+            60,
+        ))
+
+    path = client.mk_path('x.test', 60)
+    stored = real_read(path)
+    cached = trio.run(client.as_df, 'x.test', 60, False)
+    assert stored['time'].to_list() == [60, 120]
+    assert cached['time'].to_list() == [60, 120]
+    assert not list(tmp_path.glob('*.tmp'))
+
+
+def test_directory_sync_failure_keeps_visible_cache_consistent(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A post-replace durability error leaves cache matching the file.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
+    real_fsync = os.fsync
+    calls: int = 0
+
+    def fail_directory_sync(fd: int) -> None:
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            raise OSError('simulated directory sync failure')
+        real_fsync(fd)
+
+    monkeypatch.setattr(os, 'fsync', fail_directory_sync)
+    with pytest.raises(OSError, match='directory sync failure'):
+        run(client.update_ohlcv(
+            'x.test',
+            mk_ohlcv((120,)),
+            60,
+        ))
+
+    stored = pl.read_parquet(client.mk_path('x.test', 60))
+    cached = trio.run(client.as_df, 'x.test', 60, False)
+    assert stored['time'].to_list() == [60, 120]
+    assert cached['time'].to_list() == [60, 120]
+
+
+@pytest.mark.parametrize(
+    'times',
+    [
+        (0, 60),
+        (60, 60),
+        (120, 60),
+    ],
+)
+def test_update_rejects_invalid_timestamps(
+    tmp_path: Path,
+    times: tuple[float, ...],
+) -> None:
+    '''
+    Incremental input must have positive, increasing timestamps.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    incoming = mk_ohlcv(times)
+
+    with pytest.raises(ValueError):
+        run(client.update_ohlcv('x.test', incoming, 60))
+
+    assert not client.mk_path('x.test', 60).exists()
+
+
+def test_replacement_rejects_invalid_timestamps(
+    tmp_path: Path,
+) -> None:
+    '''
+    Explicit replacement enforces the same durable invariants.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+
+    with pytest.raises(ValueError):
+        run(client.write_ohlcv(
+            'x.test',
+            mk_ohlcv((60, 60)),
+            60,
+        ))
+
+    assert not client.mk_path('x.test', 60).exists()
+
+
+def test_write_rejects_invalid_schema_and_values(
+    tmp_path: Path,
+) -> None:
+    '''
+    Durable OHLCV columns must exist and contain finite numbers.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    with pytest.raises(ValueError, match='missing columns'):
+        run(client.write_ohlcv(
+            'x.test',
+            pl.DataFrame({'time': [60]}),
+            60,
+        ))
+
+    invalid = mk_ohlcv((60,))
+    invalid['close'] = np.nan
+    with pytest.raises(ValueError, match='finite numeric'):
+        run(client.write_ohlcv('x.test', invalid, 60))
+
+
+def test_series_do_not_interfere(
+    tmp_path: Path,
+) -> None:
+    '''
+    FQME and timeframe keys isolate merge and cache state.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
+    run(client.write_ohlcv('x.test', mk_ohlcv((1, 2)), 1))
+    run(client.write_ohlcv('y.test', mk_ohlcv((60, 120)), 60))
+    run(client.update_ohlcv('x.test', mk_ohlcv((120,)), 60))
+
+    x_60 = pl.read_parquet(client.mk_path('x.test', 60))
+    x_1 = pl.read_parquet(client.mk_path('x.test', 1))
+    y_60 = pl.read_parquet(client.mk_path('y.test', 60))
+    assert x_60['time'].to_list() == [60, 120]
+    assert x_1['time'].to_list() == [1, 2]
+    assert y_60['time'].to_list() == [60, 120]
+
+
+def test_index_files_ignores_lock_and_stale_temp_files(
+    tmp_path: Path,
+) -> None:
+    '''
+    Crash leftovers and writer locks are not durable series entries.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
+    (tmp_path / 'x.test.ohlcv60s.parquet.crash.tmp').touch()
+
+    index = client.index_files()
+
+    assert list(index) == ['x.test']
+    assert index['x.test']['period'] == 60
+
+
+def test_contended_file_lock_yields_to_trio(
+    tmp_path: Path,
+) -> None:
+    '''
+    Cross-client lock contention does not block the actor loop.
+
+    '''
+    client = NativeStorageClient(tmp_path)
+    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
+    path = client.mk_path('x.test', 60)
+    lock_path = path.with_name(f'.{path.name}.lock')
+
+    async def main() -> None:
+        done = trio.Event()
+
+        async def update() -> None:
+            await client.update_ohlcv(
+                'x.test',
+                mk_ohlcv((120,)),
+                60,
+            )
+            done.set()
+
+        with lock_path.open('a+b') as lock_file:
+            flock(lock_file.fileno(), LOCK_EX)
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(update)
+                await trio.sleep(0.03)
+                assert not done.is_set()
+                flock(lock_file.fileno(), LOCK_UN)
+                with trio.fail_after(0.5):
+                    await done.wait()
+
+    trio.run(main)
+    stored = pl.read_parquet(path)
+    assert stored['time'].to_list() == [60, 120]

tests/test_tsp_analysis.py

@@ -0,0 +1,330 @@
+'''
+Deterministic time-series analysis regressions.
+
+'''
+import numpy as np
+import polars as pl
+import pytest
+
+from piker.data._source import def_iohlcv_fields
+from piker.tsp import _anal
+from piker.tsp._anal import (
+    dedupe,
+    detect_time_gaps,
+    detect_time_ordering_errors,
+    get_null_segs,
+    iter_null_segs,
+    np2pl,
+    with_dts,
+)
+from piker.tsp._dedupe_smart import dedupe_ohlcv_smart
+
+
+def mk_ohlcv(
+    times: list[int],
+    indexes: list[int]|None = None,
+) -> np.ndarray:
+    '''
+    Build a synthetic OHLCV frame from timestamps.
+
+    '''
+    size: int = len(times)
+    frame = np.zeros(
+        size,
+        dtype=np.dtype(def_iohlcv_fields),
+    )
+    frame['index'] = (
+        np.arange(size)
+        if indexes is None
+        else indexes
+    )
+    frame['time'] = times
+    frame['open'] = np.arange(size) + 10
+    frame['high'] = frame['open'] + 2
+    frame['low'] = frame['open'] - 2
+    frame['close'] = frame['open'] + 1
+    frame['volume'] = np.arange(size) + 100
+    return frame
+
+
+def expected_null_segments(
+    null_positions: list[int],
+    size: int,
+    offset: int,
+    margin: int,
+
+) -> list[list[int]]:
+    '''
+    Group relative null positions with a simple state machine.
+
+    '''
+    runs: list[list[int]] = []
+    for pos in null_positions:
+        if (
+            not runs
+            or
+            pos != runs[-1][1] + 1
+        ):
+            runs.append([pos, pos])
+        else:
+            runs[-1][1] = pos
+
+    frame_end: int = offset + size - 1
+    return [
+        [
+            max(offset, offset + start - margin),
+            min(frame_end, offset + end + margin),
+        ]
+        for start, end in runs
+    ]
+
+
+def test_dedupe_recomputes_dts_after_sort() -> None:
+    '''
+    Derived columns describe final sorted, unique rows.
+
+    '''
+    src: pl.DataFrame = np2pl(mk_ohlcv([
+        60,
+        180,
+        120,
+        180,
+    ]))
+
+    _, deduped, diff = dedupe(src)
+
+    assert diff == 1
+    assert deduped['time'].to_list() == [60, 120, 180]
+    assert deduped['time_prev'].to_list() == [None, 60, 120]
+    assert deduped['s_diff'].to_list() == [None, 60, 60]
+    assert detect_time_gaps(
+        deduped,
+        expect_period=60,
+    ).is_empty()
+
+
+def test_negative_delta_is_an_ordering_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Reversed timestamps are reported as ordering faults, not gaps.
+
+    '''
+    src = np2pl(mk_ohlcv([120, 60]))
+    wdts: pl.DataFrame = with_dts(src)
+    errors = detect_time_ordering_errors(wdts)
+
+    assert detect_time_gaps(
+        wdts,
+        expect_period=30,
+    ).is_empty()
+    assert errors['time_prev'].to_list() == [120]
+    assert errors['time'].to_list() == [60]
+    assert errors['s_diff'].to_list() == [-60]
+
+    warnings: list[str] = []
+    monkeypatch.setattr(_anal.log, 'warning', warnings.append)
+    dedupe(src)
+    assert len(warnings) == 1
+    assert 'non-positive timestamp step' in warnings[0]
+
+
+def test_single_null_row_is_a_segment() -> None:
+    '''
+    A one-row null segment remains repairable.
+
+    '''
+    frame: np.ndarray = mk_ohlcv([
+        60,
+        120,
+        0,
+        240,
+        300,
+    ])
+
+    nulls = get_null_segs(frame, period=60)
+
+    assert nulls is not None
+    segments, zero_indexes, zero_rows = nulls
+    assert segments == [[1, 3]]
+    assert zero_indexes.tolist() == [2]
+    assert zero_rows['index'].tolist() == [2]
+
+
+@pytest.mark.parametrize(
+    ('times', 'expected'),
+    [
+        ([0, 0, 180, 240, 300, 360], [[10, 12]]),
+        ([60, 0, 0, 240, 0, 360], [[10, 13], [13, 15]]),
+        ([60, 120, 180, 240, 0, 0], [[13, 15]]),
+    ],
+)
+def test_null_segment_boundaries_match_numpy_and_polars(
+    times: list[int],
+    expected: list[list[int]],
+) -> None:
+    '''
+    Segments have exact, inclusive, margin-expanded endpoints.
+
+    '''
+    frame: np.ndarray = mk_ohlcv(
+        times,
+        indexes=list(range(10, 16)),
+    )
+
+    np_nulls = get_null_segs(frame, period=60)
+    pl_nulls = get_null_segs(np2pl(frame), period=60)
+
+    assert np_nulls is not None
+    assert pl_nulls is not None
+    assert np_nulls[0] == expected
+    assert pl_nulls[0] == expected
+    assert np_nulls[1].tolist() == pl_nulls[1].to_list()
+
+
+def test_null_segment_grouping_exhaustive() -> None:
+    '''
+    Every null layout through six rows matches an independent oracle.
+
+    '''
+    offset: int = 10
+    for size in range(1, 7):
+        for mask in range(1, 1 << size):
+            null_positions: list[int] = [
+                pos
+                for pos in range(size)
+                if mask & (1 << pos)
+            ]
+            frame = mk_ohlcv(
+                [
+                    0
+                    if pos in null_positions
+                    else (pos + 1) * 60
+                    for pos in range(size)
+                ],
+                indexes=list(range(offset, offset + size)),
+            )
+
+            for margin in range(3):
+                expected = expected_null_segments(
+                    null_positions,
+                    size,
+                    offset,
+                    margin,
+                )
+                for candidate in (frame, np2pl(frame)):
+                    nulls = get_null_segs(
+                        candidate,
+                        period=60,
+                        imargin=margin,
+                    )
+                    assert nulls is not None
+                    assert nulls[0] == expected
+
+
+def test_null_segment_margin_and_column() -> None:
+    '''
+    The requested column and index margin define each segment.
+
+    '''
+    frame: np.ndarray = mk_ohlcv(
+        [60, 120, 180, 240, 300],
+        indexes=[10, 11, 12, 13, 14],
+    )
+    frame['volume'][1:3] = 0
+
+    for candidate in (frame, np2pl(frame)):
+        no_margin = get_null_segs(
+            candidate,
+            period=60,
+            imargin=0,
+            col='volume',
+        )
+        wide_margin = get_null_segs(
+            candidate,
+            period=60,
+            imargin=2,
+            col='volume',
+        )
+
+        assert no_margin is not None
+        assert wide_margin is not None
+        assert no_margin[0] == [[11, 12]]
+        assert wide_margin[0] == [[10, 14]]
+        assert list(no_margin[1]) == [11, 12]
+
+
+def test_null_segment_input_invariants() -> None:
+    '''
+    Empty results and invalid index/margin inputs are explicit.
+
+    '''
+    frame = mk_ohlcv([60, 120, 180])
+    assert get_null_segs(frame, period=60) is None
+
+    with pytest.raises(ValueError, match='imargin'):
+        get_null_segs(
+            mk_ohlcv([60, 0, 180]),
+            period=60,
+            imargin=-1,
+        )
+
+    with pytest.raises(ValueError, match='contiguous'):
+        get_null_segs(
+            mk_ohlcv(
+                [60, 0, 180],
+                indexes=[10, 12, 13],
+            ),
+            period=60,
+        )
+
+
+def test_iter_null_segs_uses_precomputed_segments(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Iteration preserves inclusive boundary rows without rescanning.
+
+    '''
+    frame = mk_ohlcv([60, 120, 0, 240, 300])
+    nulls = get_null_segs(frame, period=60)
+    assert nulls is not None
+
+    def fail_rescan(*args, **kwargs):
+        raise AssertionError('precomputed segments were ignored')
+
+    monkeypatch.setattr(_anal, 'get_null_segs', fail_rescan)
+    segments = list(iter_null_segs(
+        timeframe=60,
+        frame=frame,
+        null_segs=nulls,
+    ))
+
+    assert len(segments) == 1
+    assert segments[0][:6] == (
+        1,
+        3,
+        1,
+        3,
+        120,
+        240,
+    )
+
+
+def test_smart_dedupe_honors_sort_without_dupes() -> None:
+    '''
+    The sort option applies even when no duplicates exist.
+
+    '''
+    src: pl.DataFrame = np2pl(mk_ohlcv([
+        180,
+        60,
+        120,
+    ]))
+
+    _, deduped, diff, _, issues = dedupe_ohlcv_smart(src)
+
+    assert diff == 0
+    assert issues is None
+    assert deduped['time'].to_list() == [60, 120, 180]
+    assert deduped['s_diff'].to_list() == [None, 60, 60]