Use IB schedules to reserve missing SHM history

Interrupted backfill can leave stored observations separated by
missing trading time. Query dated IB sessions before prepending
stored history and reserve synthetic samples only during covered
open sessions. Keep venue closures compressed and preserve the
indexes of already published rows.

Deats,
- Add `Client.history_schedule()` with UTC coverage, extended hours,
  bounded duration, timeout, and rejection of ambiguous DST times.
- Keep schedule normalization in `venues` and allocation in
  `.tsp._reserve`, with the project AGPL headers.
- Expose schedule and synthetic-range queries through `datad.ib`.
- Bound reservations by available SHM capacity and inspect only the
  newest 32 gaps by default via `newest_gaps`. Zero disables
  queries; leave unknown and oversized ranges unresolved.
- Record synthetic ranges separately from OHLCV. Automatic storage
  writes retain provider observations; placeholders copy close.
- Group the new tests under `tests/providers`.
- Cover request formatting, DST, failure and capacity cases, plus
  real SHM/NativeDB startup placement with a fake provider.
- Document a common CLI/chart segment-repair interface and defer
  coordinated insertion, repair execution, and reader integration.

Validation: 55 targeted tests passed in the working tree; live IB
schedule coverage remains unqualified. After the module moves,
33 affected tests passed again; the local review follow-up passed
18 affected cases, including four new request-budget cases.
The commit runner repeats
lint and isolated regression checks against this exact boundary.

Prompt-IO: ai/prompt-io/codex/20260924T185456Z_1da9b760_schedule_prompt_io.md

(this patch was generated in some part by `codex` using
`gpt-6` (`openai`))

ai/prompt-io/codex/20260924T165757Z_1da9b760_prompt_io.md

@@ -0,0 +1,51 @@
+---
+model: gpt-6 (exact variant not exposed)
+service: codex
+timestamp: 2026-09-24T16:57:57.342784+00:00
+git_ref: 1da9b760
+scope: code
+substantive: true
+raw_file: 20260924T165757Z_1da9b760_prompt_io.raw.md
+---
+
+## Prompt
+
+Document and test the existing null-segment repair, including its
+flat-bar fallback, while distinguishing packed timestamp gaps.
+
+## Response summary
+
+Documented SHM-only repair, provider placement assumptions, fallback
+fields, notifications, and concurrency limits. Added boundary and
+no-history coverage and fixed epoch-dependent comparison tolerance
+which could synthesize a duplicate timestamp at the right margin.
+
+## Files changed
+
+- piker/tsp/_history.py: helper contract and absolute tolerance.
+- tests/test_history_backfill.py: four additional parameter/case runs.
+
+## Human edits
+
+The human identified cancellation as a source of stored gaps,
+correctly described packed indexes and insertion costs, pointed to
+maybe_fill_null_segments(), and requested immediate documentation
+and tests. No direct source edits observed during this change;
+the patch remains uncommitted for review.
+
+## Close-copy factoring follow-up
+
+The human requested stronger internal types, a reusable close-copy
+helper, searchable comments for copying close and synthesizing
+stamps, and research into historical venue closure queries.
+Extracted _copy_close_to_null_rows(), preserving non-OHLC fields and
+rejecting occupied slots before mutation. Added explicit callback,
+array, segment-index and date annotations. Preserved the existing
+provenance TODO and suggested avoiding volume sentinels.
+Four helper cases cover successful fill and unchanged rejected
+inputs. Backfill suite: 19 passed in 2.01s; Ruff and diff check pass.
+
+IB documents historical SCHEDULE requests separately from contract
+hours. The installed ib_async exposes reqHistoricalScheduleAsync.
+No live schedule request or venue behavior change was performed.
+Source: https://interactivebrokers.github.io/tws-api/historical_bars.html

ai/prompt-io/codex/20260924T165757Z_1da9b760_prompt_io.raw.md

@@ -0,0 +1,48 @@
+---
+model: gpt-6 (exact variant not exposed)
+service: codex
+timestamp: 2026-09-24T16:57:57.342784+00:00
+git_ref: 1da9b760
+diff_cmd: git diff 1da9b760 -- piker/tsp/_history.py tests/test_history_backfill.py
+---
+
+## Prompt
+
+The human requested documentation and tests for
+maybe_fill_null_segments(), identifying its flat OHLC fallback
+and asking how packed SHM could accept later recovered bars.
+
+## Code output
+
+> `git diff 1da9b760 -- piker/tsp/_history.py`
+
+> `git diff 1da9b760 -- tests/test_history_backfill.py`
+
+Document the existing slot-based repair and its limitations.
+Extend fallback coverage to NoData and DataUnavailable; verify
+the observed packed MNQ gap remains unchanged. Reproduce the
+relative-tolerance failure at a modern epoch, then disable relative
+tolerance in the synthetic timestamp boundary check.
+
+## Validation
+
+Before the fix: modern-epoch boundary case failed, zero-epoch
+case passed. The overfull synthetic run duplicated the right bar.
+After the fix: 15 backfill tests passed in 1.99s.
+
+## Close-copy factoring follow-up
+
+The human requested stronger internal types, a reusable close-copy
+helper, searchable comments for copying close and synthesizing
+stamps, and research into historical venue closure queries.
+Extracted _copy_close_to_null_rows(), preserving non-OHLC fields and
+rejecting occupied slots before mutation. Added explicit callback,
+array, segment-index and date annotations. Preserved the existing
+provenance TODO and suggested avoiding volume sentinels.
+Four helper cases cover successful fill and unchanged rejected
+inputs. Backfill suite: 19 passed in 2.01s; Ruff and diff check pass.
+
+IB documents historical SCHEDULE requests separately from contract
+hours. The installed ib_async exposes reqHistoricalScheduleAsync.
+No live schedule request or venue behavior change was performed.
+Source: https://interactivebrokers.github.io/tws-api/historical_bars.html

ai/prompt-io/codex/20260924T185456Z_1da9b760_schedule_prompt_io.md

@@ -0,0 +1,91 @@
+---
+model: gpt-6 (exact variant not exposed)
+service: codex
+timestamp: 2026-09-24T18:54:56.893490+00:00
+git_ref: 1da9b760
+scope: code
+substantive: true
+raw_file: 20260924T185456Z_1da9b760_schedule_prompt_io.raw.md
+---
+
+## Prompt
+
+The human requested: first freeze a commit-plan for the existing
+close-copy/null-repair changes, then implement ib_async historical
+schedule support with tests, use it for datad.ib history-buffer gap
+spacing, and plan a common on-demand segment interface for CLI and
+chart selection. Avoid reindexing in this first pass and record
+fallback insertion only as a TODO.
+
+## Response summary
+
+Implemented historical schedule normalization, bounded startup
+reservations, explicit actor-local synthetic ranges, and two read
+RPC functions. Added deterministic adapter tests and real SHM/parquet
+startup coverage. Wrote the segment-backfill interface plan.
+
+This entry covers the new schedule work only. The earlier frozen
+null-repair plan has its own provenance. The shared _history.py file
+contains changes from both scopes until the earlier commit is made.
+
+## Human edits
+
+The human identified cancellation during backfill as the reproducible
+trigger, explained the packed-index insertion tradeoff, requested
+reuse of venue closure knowledge, and chose placeholders to avoid
+expensive rewrites. They required explicit placeholder annotation,
+proposed later graphics compression, and asked to defer insertion
+fallbacks. They also requested the existing null-repair commit plan
+before the new schedule work. These are human-directed architecture
+and scope choices; no acceptance or task completion state is inferred.
+
+## Validation and limits
+
+55 targeted tests passed; Ruff and diff checks passed. No live IB
+qualification was performed. Reservations are synthetic, not recovered
+observations. Repair execution, persistence of unfinished requests,
+chart compression, FSP handling, export filtering, and request-level
+IB cancellation remain follow-up work documented in the plan.
+
+## Files changed
+
+- `piker/brokers/ib/venues.py`
+- `piker/brokers/ib/api.py`
+- `piker/brokers/ib/feed.py`
+- `piker/tsp/_reserve.py`
+- `piker/tsp/_history.py`
+- `tests/providers/test_ib_schedule.py`
+- `tests/providers/test_history_reservations.py`
+- `docs/history_segment_backfill.rst`
+
+## Pre-commit module review
+
+The human requested placing historical schedule normalization in the
+existing venues.py module, naming the allocation module _reserve.py,
+and adding the missing project license headers. Applied those moves,
+updated imports and test patch targets, and added the standard AGPL
+headers to the new allocation module and both new test modules.
+
+## Local tuicr review follow-up
+
+The human supplied eleven comments requesting a configurable newest
+gap count, caller and metadata references, explanations of guards
+and provider-specific limits, multiline boolean style, capacity
+fallback rationale, an explanation of numpy split boundaries, and
+clearer test cases, fixture types, and module docstrings.
+
+Added newest_gaps=32, with zero disabling queries and negative values
+rejected. Documented tsdb_backfill ownership and the local/RPC range
+accessors. Explained recoverable overflow and half-open synthetic
+range grouping. Typed tmp_path as pathlib.Path and monkeypatch as
+pytest.MonkeyPatch. Added four request-budget cases.
+
+Verification: reservation tests 11 passed; schedule tests 7 passed;
+Ruff and git diff --check passed. Changes remain unstaged for review.
+No forge review or CI state was queried; this was local feedback.
+
+## Provider-test layout review
+
+The human requested grouping both new history/schedule test modules
+under tests/providers. Moved both files without changing test logic;
+they retain the shared tests/conftest.py fixtures.

ai/prompt-io/codex/20260924T185456Z_1da9b760_schedule_prompt_io.raw.md

@@ -0,0 +1,100 @@
+---
+model: gpt-6 (exact variant not exposed)
+service: codex
+timestamp: 2026-09-24T18:54:56.893490+00:00
+git_ref: 1da9b760
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+> `git diff HEAD~1..HEAD -- piker/brokers/ib/_schedule.py`
+
+> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py`
+
+> `git diff HEAD~1..HEAD -- piker/brokers/ib/feed.py`
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_reservations.py`
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
+
+> `git diff HEAD~1..HEAD -- tests/test_ib_schedule.py`
+
+> `git diff HEAD~1..HEAD -- tests/test_history_reservations.py`
+
+> `git diff HEAD~1..HEAD -- docs/history_segment_backfill.rst`
+
+## Generated output
+
+Added Client.history_schedule() over the installed ib_async API,
+normalizing dated sessions to UTC with explicit response coverage.
+Reject ambiguous DST times, reversed sessions, and overlapping
+sessions. Match the existing extended-hours bars policy.
+
+Added startup-only schedule reservations to tsdb_backfill before
+prepending older stored history. Reserve open-session sample slots
+only where the reply covers the full candidate gap. Return synthetic
+UTC ranges separately from OHLCV; copy preceding close into flat
+placeholders without writing those placeholders to parquet.
+Unknown coverage and capacity failures retain packed time gaps.
+
+Added datad.ib schedule and reservation metadata RPC functions and
+a proposed shared CLI/chart segment request protocol. The repair
+queue, durable unresolved intervals, chart compression, FSP handling,
+and coordinated insertion remain deferred. No live data was repaired.
+
+Verification is complete: 55 targeted tests pass, including the
+new schedule and real SHM/parquet tests. I also added checks for
+reversed session bounds and the 25-hour day at the autumn DST
+change.
+
+The segment API plan uses UTC ranges for both CLI and chart
+requests, a provider-owned queue, durable unfinished intervals,
+and replacement of reserved slots without reindexing. Live IB
+schedule coverage still needs qualification.
+
+## Limits
+
+Only stored-history gaps are reserved before prepend; published rows
+are never shifted. Newest 32 gaps, at most 31 days each, are queried
+serially with a 10-second schedule timeout. This can delay history
+publication. Underlying IB request cancellation is still a TODO.
+Manual whole-SHM export requires synthetic-range exclusion before
+it can preserve provider-only storage. The chart currently displays
+flat placeholders, and synthetic-aware FSP treatment is deferred.
+
+Tests: history backfill 19, IB history 17, method proxy 5, schedule 7,
+reservations 7. Sandbox local-socket failures were rerun with approval.
+Ruff and git diff --check passed. No live broker requests were made.
+
+## Module-layout review response
+
+Schedule normalization now lives in piker.brokers.ib.venues beside
+the existing session helpers. The allocation module is tsp._reserve.
+Added the project AGPL headers to the new module and test files.
+Earlier diff pointers record the original generated paths; use
+the following paths for the reviewed layout:
+
+> `git diff HEAD~1..HEAD -- piker/brokers/ib/venues.py`
+
+> `git diff HEAD~1..HEAD -- piker/tsp/_reserve.py`
+
+## Local review response
+
+The fixes pass all 18 tests in the two affected files.
+The new limit tests cover zero, negative, one-gap, and two-gap
+budgets, including newest-first selection.
+
+Capacity overflow remains a warning: no partial reservation is
+published for that gap, and the original observations remain
+available. The comments now explain that fallback explicitly.
+
+Diff pointers above include the review edits to tsp._reserve, both
+test modules, and the interface document. This review did not query
+a forge or publish replies.
+
+## Provider-test paths
+
+Moved both new test modules under tests/providers as requested.
+
+> `git diff HEAD~1..HEAD -- tests/providers/test_ib_schedule.py`
+
+> `git diff HEAD~1..HEAD -- tests/providers/test_history_reservations.py`

docs/history_segment_backfill.rst

@@ -0,0 +1,113 @@
+History segment backfill: first-pass interface plan
+=================================================
+
+Status and scope
+----------------
+
+The first implementation queries dated IB schedules and reserves
+synthetic samples while preparing older stored history for SHM.
+It does not move published rows, implement automatic segment repair,
+or change chart rendering. Synthetic ranges are actor-local metadata;
+parquet remains provider observations and is rescanned on restart.
+
+The existing null-row repair is not the segment-repair implementation.
+It assumes allocated zero-time slots. Schedule reservations instead
+have valid synthetic timestamps and explicit external provenance.
+
+Schedule queries
+----------------
+
+``datad.ib`` exposes ``feed.get_history_schedule(fqme, start, end)``
+for UTC epoch bounds. The response includes actual coverage, timezone,
+regular-hours policy, and half-open trading sessions. This wraps
+``Client.history_schedule()`` through the existing asyncio method proxy.
+The first pass limits a query to 31 days and times out after 10 seconds.
+Historical queries must match bars' ``useRTH=False`` selection.
+
+Do not extrapolate sessions outside response coverage. Empty schedules,
+ambiguous DST times, failures, and partial coverage are unknown for
+allocation purposes. Missing contract/session support is not a closure.
+A successful schedule establishes possible trading slots, not actual
+bar counts or proof of trades. Live MNQ coverage remains to be verified.
+
+Startup reservations
+--------------------
+
+After reverse retrieval and before prepending stored history, inspect
+its newest 32 timestamp gaps within the retained SHM capacity by
+default. ``reserve_history_gaps(newest_gaps=32)`` makes that request
+budget configurable; zero disables reservations. Query schedules
+serially through the history client's proxy. Reserve only
+covered open-session slots, preserving venue closures as time gaps.
+Large or unsupported ranges remain unresolved. Capacity bounds include
+synthetic samples; oldest samples may be clipped as with ordinary SHM
+prepending. No published provider or realtime row is shifted.
+
+``feed.get_history_reservations(fqme, timeframe)`` returns half-open UTC
+ranges of synthetic rows for this datad generation. Consumers must use
+this provenance rather than zero volume or flat prices. Index positions
+are not durable identity. Metadata is published with the synchronous SHM
+prepend and clipped to its visible timestamps.
+
+The default chart still displays flat placeholders. Compression and
+synthetic-aware FSP treatment are follow-up consumers of this metadata.
+Manual whole-SHM export must exclude synthetic ranges before writing
+provider history; automatic startup writes do not include reservations.
+
+On-demand request contract (proposed)
+------------------------------------
+
+A datad context should accept ``fqme``, ``timeframe``, ``start``, ``end``,
+``request_id``, and ``reason``. Bounds are half-open UTC timestamps;
+reasons include interrupted startup, detected gap, and operator selection.
+CLI and chart selection submit the same request, with no client-local
+array indexes. The actor validates market identity and sampling grid.
+
+Proposed events are accepted, progress, completed, partial, failed, and
+cancelled. Progress carries requested/received bounds, rows persisted,
+slots replaced, remaining synthetic ranges, and failure details.
+``completed`` means every requested interval received a classified
+provider outcome, not that every scheduled slot contained a trade.
+
+A provider-owned queue deduplicates overlapping requests and serializes
+IB history calls. Persist unresolved requests and per-interval outcomes
+before fetching so actor/client cancellation cannot erase pending work.
+Distinguish genuine empty replies from timeouts and permission failures.
+Closing a progress subscription must not implicitly discard queued work;
+explicit cancellation policy belongs in the request protocol.
+
+Merge real rows into NativeDB first, then replace matching synthetic
+SHM slots by timestamp without changing indexes. Remove provenance only
+after successful publication. Revalidate bounds and generation before
+writes, coordinate sampler/FSP readers, and broadcast a repair interval
+for downstream cache invalidation. Never use legacy null repair to place
+an arbitrary provider frame backward over already valid observations.
+
+CLI proposal: ``piker store backfill FQME --timeframe N --start UTC
+--end UTC`` with a preview showing schedules, unknown ranges, and slots.
+Chart proposal: translate selected x coordinates through the current
+view's timestamp mapping, preview the interval, and subscribe to the
+same progress context. Compression must retain inverse timestamp/index
+mapping for cursor, selection, and annotation behavior.
+
+Deferred work
+-------------
+
+TODO: replace synthetic rows using the common segment request context.
+TODO: durable unresolved/checked interval tracking and bounded retries.
+TODO: automatic scan worker separate from samplerd's live sampling loop.
+TODO: chart compression and explicit FSP handling of placeholders.
+TODO: prevent operational whole-SHM exports from persisting placeholders.
+TODO: cancel the underlying IB schedule request on timeout/disconnect.
+TODO: batch/cache session queries with coverage and contract identity.
+TODO: coordinated insertion only if schedule-driven reservation proves
+insufficient; first pass reports unknown/capacity cases without reindexing.
+
+Validation
+----------
+
+Use fake schedule responses for dated timezone conversion, DST ambiguity,
+partial coverage, closures, empty/error results, and capacity limits.
+An isolated real-SHM/NativeDB restart test must show that reserved rows
+are marked synthetic, closures stay compressed, and parquet never gains
+invented trades. Live gateway qualification is separate from these tests.

piker/brokers/ib/api.py

@@ -350,6 +350,48 @@ class Client:
             apiOnly=False,
         )
 
+    async def history_schedule(
+        self,
+        fqme: str,
+        start_dt: datetime,
+        end_dt: datetime,
+        use_rth: bool = False,
+        timeout_s: float = 10,
+    ) -> dict:
+        '''
+        Query dated trading sessions for a bounded historical range.
+
+        Return UTC coverage separately from sessions. Missing coverage
+        is unknown, never evidence that the venue was closed.
+        Match bars()'s extended-hours policy by default.
+
+        '''
+        from math import ceil
+        from .venues import normalize_schedule
+
+        if start_dt.tzinfo is None or end_dt.tzinfo is None:
+            raise ValueError('Schedule bounds must be timezone-aware')
+        seconds: float = (
+            end_dt.astimezone(UTC) - start_dt.astimezone(UTC)
+        ).total_seconds()
+        if not 0 < seconds <= 31 * 86400:
+            raise ValueError('Schedule range must be within 31 days')
+        contract: Contract = (await self.find_contracts(fqme))[0]
+        schedule = await asyncio.wait_for(
+            self.ib.reqHistoricalScheduleAsync(
+                contract,
+                numDays=max(1, ceil(seconds / 86400)),
+                endDateTime=end_dt.astimezone(UTC).strftime(
+                    '%Y%m%d-%H:%M:%S'
+                ),
+                useRTH=use_rth,
+            ),
+            timeout=timeout_s,
+        )
+        # TODO: explicit cancellation of the underlying IB request
+        # requires a request-id exposing ib_async schedule wrapper.
+        return normalize_schedule(schedule, use_rth)
+
     async def bars(
         self,
         fqme: str,

piker/brokers/ib/feed.py

@@ -317,6 +317,18 @@ async def open_history_client(
                 last_dt,
             )
 
+        async def query_schedule(left: float, right: float) -> dict:
+            '''
+            Request dated sessions through the existing IB proxy.
+
+            '''
+            return await proxy.history_schedule(
+                fqme=fqme,
+                start_dt=from_timestamp(left),
+                end_dt=from_timestamp(right),
+                use_rth=False,
+            )
+
         # TODO: it seems like we can do async queries for ohlc
         # but getting the order right still isn't working and I'm not
         # quite sure why.. needs some tinkering and probably
@@ -325,6 +337,7 @@ async def open_history_client(
         yield (
             get_hist,
             {
+                'query_schedule': query_schedule,
                 'erlangs': 1,  # max conc reqs
                 'rate': 3,  # max req rate
                 'frame_types': {  # expected frame sizes
@@ -1334,3 +1347,33 @@ async def stream_quotes(
                             # ugh, clear ticks since we've consumed them
                             # ticker.ticks = []
                             # last = time.time()
+
+
+async def get_history_schedule(
+    fqme: str,
+    start: float,
+    end: float,
+) -> dict:
+    '''
+    Query historical sessions on datad.ib using UTC epoch bounds.
+
+    '''
+    async with open_data_client() as proxy:
+        return await proxy.history_schedule(
+            fqme=fqme.removesuffix('.ib'),
+            start_dt=from_timestamp(start),
+            end_dt=from_timestamp(end),
+            use_rth=False,
+        )
+
+
+async def get_history_reservations(
+    fqme: str,
+    timeframe: float,
+) -> list[tuple[float, float]]:
+    '''
+    Report synthetic history ranges for this datad generation.
+
+    '''
+    from piker.tsp._reserve import reservation_ranges
+    return reservation_ranges(fqme, timeframe)

piker/brokers/ib/venues.py

@@ -30,8 +30,11 @@ from datetime import (  # noqa
 from typing import (
     Iterator,
     TYPE_CHECKING,
+    TypedDict,
 )
 
+from zoneinfo import ZoneInfo
+
 import exchange_calendars as xcals
 from exchange_calendars.errors import (
     InvalidCalendarName,
@@ -51,6 +54,7 @@ log = get_logger(__name__)
 if TYPE_CHECKING:
     from ib_async import (
         TradingSession,
+        HistoricalSchedule,
         Contract,
         ContractDetails,
     )
@@ -64,6 +68,72 @@ if TYPE_CHECKING:
     )
 
 
+class HistorySchedule(TypedDict):
+    '''
+    IPC-safe UTC coverage and half-open session intervals.
+
+    Coverage describes the server's response, not the requested
+    interval. Consumers must reject uncovered or ambiguous dates.
+
+    '''
+    start: float
+    end: float
+    sessions: list[tuple[float, float]]
+    timezone: str
+    use_rth: bool
+
+
+def normalize_schedule(
+    schedule: HistoricalSchedule,
+    use_rth: bool,
+) -> HistorySchedule:
+    '''
+    Parse dated session boundaries and reject DST ambiguity.
+
+    Never repeat today's trading hours across historical dates.
+    Ambiguous/nonexistent local times need provider clarification.
+
+    '''
+    zone: ZoneInfo = ZoneInfo(schedule.timeZone)
+
+    def stamp(value: str) -> float:
+        '''
+        Convert one unambiguous IB wall time into epoch seconds.
+
+        '''
+        local: datetime = datetime.strptime(value, '%Y%m%d-%H:%M:%S')
+        first: datetime = local.replace(tzinfo=zone, fold=0)
+        second: datetime = local.replace(tzinfo=zone, fold=1)
+        if first.utcoffset() != second.utcoffset():
+            raise ValueError(f'Ambiguous schedule time: {value}')
+        return first.timestamp()
+
+    start: float = stamp(schedule.startDateTime)
+    end: float = stamp(schedule.endDateTime)
+    if start >= end:
+        raise ValueError('Invalid schedule coverage')
+    sessions: list[tuple[float, float]] = []
+    for session in schedule.sessions:
+        left: float = stamp(session.startDateTime)
+        right: float = stamp(session.endDateTime)
+        if left >= right:
+            raise ValueError('Invalid schedule session')
+        left = max(start, left)
+        right = min(end, right)
+        if left < right:
+            sessions.append((left, right))
+    sessions.sort()
+    if any(a[1] > b[0] for a, b in zip(sessions, sessions[1:])):
+        raise ValueError('Overlapping schedule sessions')
+    return HistorySchedule(
+        start=start,
+        end=end,
+        sessions=sessions,
+        timezone=schedule.timeZone,
+        use_rth=use_rth,
+    )
+
+
 def is_expired(
     con_deats: ContractDetails,
 ) -> bool:

piker/tsp/_history.py

@@ -38,6 +38,7 @@ from pprint import pformat
 from string import hexdigits
 from types import ModuleType
 from typing import (
+    Awaitable,
     Callable,
     Generator,
     Literal,
@@ -87,7 +88,6 @@ from piker.storage import TimeseriesNotFound
 from ._anal import (
     get_null_segs,
     iter_null_segs,
-    Frame,
 
     # codec-ish
     np2pl as np2pl,
@@ -225,12 +225,63 @@ def _synthetic_gap_times(
     if not np.isclose(
         right_t,
         start_t + (count + 1)*timeframe,
+        # Epoch magnitude must not widen allowed boundary error.
+        rtol=0,
+        atol=1e-6,
     ):
         return None
 
     return times
 
 
+def _copy_close_to_null_rows(
+    gap: np.ndarray,
+    seed: np.void,
+    right_t: float|None,
+    timeframe: float,
+
+) -> int:
+    '''
+    Copy preceding CLOSE into flat OHLC and SYNTHESIZE TIMESTAMPS.
+
+    Mutate only an entirely zero-time `gap` view, bounded by `seed`
+    and `right_t`. Return the number of synthetic rows, or zero
+    without mutation when the bounds cannot accommodate the slots.
+    Preserve indexes and non-OHLC fields, including volume/count.
+    This synchronous operation performs no I/O or SHM publication.
+
+    These are synthetic placeholders, not provider observations.
+    Callers own provenance tracking and subscriber notification.
+
+    '''
+    start_t: float = float(seed['time'])
+    if (
+        not len(gap)
+        or
+        start_t <= 0
+        or
+        not np.all(gap['time'] == 0)
+    ):
+        return 0
+
+    gap_times: np.ndarray|None = _synthetic_gap_times(
+        start_t,
+        right_t,
+        len(gap),
+        timeframe,
+    )
+    if gap_times is None:
+        return 0
+
+    # SYNTHETIC FLAT BARS: COPY PRECEDING CLOSE into O/H/L/C.
+    # SYNTHESIZE TIMESTAMPS only for the existing zero-time slots.
+    # No volume/count fabrication and no physical row insertion.
+    close: float = float(seed['close'])
+    gap[['open', 'high', 'low', 'close']] = close
+    gap['time'] = gap_times
+    return len(gap)
+
+
 async def shm_push_in_between(
     shm: ShmArray,
     to_push: np.ndarray,
@@ -276,7 +327,9 @@ async def shm_push_in_between(
 async def maybe_fill_null_segments(
     shm: ShmArray,
     timeframe: float,
-    get_hist: Callable,
+    get_hist: Callable[..., Awaitable[
+        tuple[np.ndarray, datetime, datetime]
+    ]],
     sampler_stream: tractor.MsgStream,
     mkt: MktPair,
     backfill_until_dt: datetime,
@@ -284,19 +337,67 @@ async def maybe_fill_null_segments(
     task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
 
 ) -> None:
+    '''
+    Repair existing zero-time SHM rows, then try flat-bar fallback.
+
+    This operates on physical slots already inside `shm.array`.
+    `get_null_segs()` finds rows whose time is zero; a timestamp
+    jump between two nonzero rows does not trigger a query. No rows
+    are inserted, shifted, or reindexed, and no storage is updated.
+
+    Signal task startup, snapshot the readable frame, and request
+    provider history using each null segment's surrounding dates.
+    Provider replies are written backward from the segment's right
+    boundary through `shm_push_in_between()`, without moving the
+    readable first index. This legacy placement assumes the reply
+    fits the existing slot layout; it is not a general timestamp
+    merge and does not limit writes to only the null rows.
+
+    An empty reply, NoData, DataUnavailable, or a reply beginning
+    before `backfill_until_dt` stops further provider queries for
+    this pass. Cancellation and other provider errors propagate.
+
+    Rescan the live frame for remaining zero-time runs. For each
+    bounded run accepted by `_synthetic_gap_times()`, copy the
+    preceding valid bar's close into every OHLC field and generate
+    timestamps at `timeframe` spacing. Preserve both boundary bars,
+    physical indexes, and all other fields (including volume).
+    Leading/trailing runs and incompatible slot counts stay null.
+
+    Flat bars are synthetic display fallback, not recovered trades
+    or proof that the venue was closed. They currently have no
+    explicit provenance flag. Notify subscribers after provider
+    writes and after synthetic repair via `notify_backfill()`.
+    Callers must coordinate SHM writers and provider access; this
+    helper does not lock either across its await points.
+
+    '''
 
     task_status.started()
 
-    frame: Frame = shm.array.copy()
+    frame: np.ndarray = shm.array.copy()
 
     # TODO, put in parent task/daemon root!
     import greenback
     await greenback.ensure_portal()
 
-    null_segs: tuple|None = get_null_segs(
+    null_segs: tuple[
+        list[list[int]], np.ndarray, np.ndarray
+    ]|None = get_null_segs(
         frame,
         period=timeframe,
     )
+    absi_start: int
+    absi_end: int
+    fi_start: int|None
+    fi_end: int
+    start_t: float|None
+    end_t: float
+    start_dt: datetime|None
+    end_dt: datetime
+    array: np.ndarray
+    next_start_dt: datetime
+    next_end_dt: datetime
     for (
         absi_start, absi_end,
         fi_start, fi_end,
@@ -387,8 +488,8 @@ async def maybe_fill_null_segments(
         )
 
     # RECHECK for more null-gaps
-    frame: Frame = shm.array
+    frame = shm.array
-    null_segs: tuple | None = get_null_segs(
+    null_segs = get_null_segs(
         frame,
         period=timeframe,
     )
@@ -397,6 +498,9 @@ async def maybe_fill_null_segments(
         and
         len(null_segs[-1])
     ):
+        iabs_slices: list[list[int]]
+        iabs_zero_rows: np.ndarray
+        _zero_t: np.ndarray
         (
             iabs_slices,
             iabs_zero_rows,
@@ -412,12 +516,6 @@ async def maybe_fill_null_segments(
         # stretching the y-axis..
         # array: np.ndarray = shm.array
         # zeros = array[array['low'] == 0]
-        ohlc_fields: list[str] = [
-            'open',
-            'high',
-            'low',
-            'close',
-        ]
 
         zero_indexes: np.ndarray = np.asarray(iabs_zero_rows)
         split_at: np.ndarray = (
@@ -431,6 +529,7 @@ async def maybe_fill_null_segments(
         )
         repaired: int = 0
         frame_start: int = int(frame['index'][0])
+        zero_group: np.ndarray
         for zero_group in zero_groups:
             istart: int = int(zero_group[0])
             istop: int = int(zero_group[-1]) + 1
@@ -453,7 +552,6 @@ async def maybe_fill_null_segments(
 
             # Fill only the null rows, preserving both valid margins.
             gap: np.ndarray = shm._array[istart:istop]
-            cls: float = seed['close']
 
             frame_end: int = int(frame['index'][-1])
             right_t: float|None = (
@@ -461,29 +559,28 @@ async def maybe_fill_null_segments(
                 if istop <= frame_end
                 else None
             )
-            gap_times: np.ndarray|None = _synthetic_gap_times(
+            # TODO: how can we mark this range as being a gap tho?
-                start_t,
+            # -[ ] maybe pg finally supports nulls in ndarray to
+            #      show empty space somehow?
+            # -[ ] we could put a special value in the vlm or
+            #     another col/field to denote?
+            # Prefer explicit synthetic-row provenance over a volume
+            # sentinel: zero volume alone does not identify a repair.
+            # COPY CLOSE / SYNTHESIZE TIMESTAMPS fallback lives here.
+            copied: int = _copy_close_to_null_rows(
+                gap,
+                seed,
                 right_t,
-                len(gap),
                 timeframe,
             )
-            if gap_times is None:
+            if not copied:
                 log.warning(
                     f'Can not safely forward-fill SHM nulls:\n'
                     f'left={start_t} right={right_t} '
                     f'rows={len(gap)} period={timeframe}\n'
                 )
                 continue
-
+            repaired += copied
-            # TODO: how can we mark this range as being a gap tho?
-            # -[ ] maybe pg finally supports nulls in ndarray to
-            #      show empty space somehow?
-            # -[ ] we could put a special value in the vlm or
-            #     another col/field to denote?
-            gap[ohlc_fields] = cls
-
-            gap['time'] = gap_times
-            repaired += len(gap)
 
             # TODO: reimpl using the new `.ui._remote_ctl` ctx
             # ideally using some kinda decent
@@ -1409,11 +1506,42 @@ async def tsdb_backfill(
                 # establishes the actual earliest sample, then prepend
                 # storage directly beside it without reserving closure
                 # rows.
+                from ._reserve import (
+                    _reservations,
+                    reserve_history_gaps,
+                )
+                synthetic: list[tuple[float, float]] = []
+                query_schedule = config.get('query_schedule')
+                if (
+                    query_schedule is not None
+                    and
+                    'time' in tsdb_history.dtype.names
+                ):
+                    tsdb_history, synthetic = await reserve_history_gaps(
+                        tsdb_history,
+                        timeframe,
+                        query_schedule,
+                        max_rows=max(0, int(shm._first.value)),
+                    )
+                # Startup-only placement: the provider prefix is
+                # already published, but these older rows are not.
+                # Preserve synthetic provenance outside OHLC fields.
                 to_push: np.ndarray = _prepend_tsdb_history(
                     shm,
                     tsdb_history,
                     field_map=storemod.ohlc_key_map,
                 )
+                if synthetic and len(to_push):
+                    low: float = float(to_push['time'][0])
+                    high: float = float(to_push['time'][-1]) + timeframe
+                    synthetic = [
+                        (max(a, low), min(b, high))
+                        for a, b in synthetic
+                        if a < high and b > low
+                    ]
+                else:
+                    synthetic = []
+                _reservations[(mkt.fqme, timeframe)] = synthetic
                 log.info(
                     f'Loaded {to_push.shape} datums from storage'
                 )

piker/tsp/_reserve.py

@@ -0,0 +1,192 @@
+# piker: trading gear for hackers
+# Copyright (C) 2018-present  Tyler Goodlet (in stewardship of pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Startup-only schedule reservations and explicit synthetic provenance.
+
+No published rows are moved. Synthetic rows never enter storage here.
+
+'''
+from collections.abc import Awaitable, Callable
+import math
+
+import numpy as np
+
+from piker.log import get_logger
+
+log = get_logger(__name__)
+# Actor-local metadata, scoped by market and sampling period.
+# Each range is half-open UTC and identifies synthetic SHM samples.
+type SyntheticRanges = list[tuple[float, float]]
+_reservations: dict[tuple[str, float], SyntheticRanges] = {}
+
+
+def reservation_ranges(
+    fqme: str,
+    timeframe: float,
+) -> list[tuple[float, float]]:
+    '''
+    Return a copy for RPC readers; do not expose mutable ownership.
+
+    '''
+    return list(_reservations.get((fqme, timeframe), []))
+
+
+async def reserve_history_gaps(
+    frame: np.ndarray,
+    timeframe: float,
+    query_schedule: Callable[..., Awaitable[dict]],
+    max_rows: int,
+    newest_gaps: int = 32,
+) -> tuple[np.ndarray, list[tuple[float, float]]]:
+    '''
+    Reserve missing scheduled samples before publishing history.
+
+    Keep venue closures compressed. Fully covered schedule replies
+    permit flat OHLC placeholders at the expected sampling cadence.
+    Unknown coverage and oversized reservations remain time gaps.
+    Inspect at most `newest_gaps` gaps in retained history, newest
+    first; zero disables reservations. Return the expanded frame
+    and half-open UTC synthetic ranges separately.
+
+    `_history.tsdb_backfill()` calls this before prepending stored
+    history. It publishes the returned ranges in `_reservations`.
+    Local readers use `reservation_ranges()`; remote readers use
+    `piker.brokers.ib.feed.get_history_reservations()` on datad.ib.
+    Future segment repair will replace these slots by timestamp;
+    chart/FSP consumers still need synthetic-aware handling.
+    No synthetic rows are written to parquet here.
+
+    TODO: coordinated insertion when coverage or capacity cannot
+    support a reservation; first pass never rewrites published rows.
+
+    '''
+    if newest_gaps < 0:
+        raise ValueError('newest_gaps must be nonnegative')
+    # A gap needs two observed margins and room for both. A zero
+    # request budget disables work without slicing `gaps[-0:]`.
+    if (
+        max_rows < 2
+        or
+        len(frame) < 2
+        or
+        newest_gaps == 0
+    ):
+        return frame, []
+    frame = frame[-max_rows:]
+    times: np.ndarray = frame['time']
+    # A nonpositive cadence cannot define sample positions. Duplicate
+    # or decreasing timestamps make the left/right margins ambiguous.
+    # Leave the retained data unchanged; this helper does not sort or
+    # deduplicate provider observations to guess a valid sample grid.
+    if (
+        timeframe <= 0
+        or
+        np.any(np.diff(times) <= 0)
+    ):
+        return frame, []
+    gaps: np.ndarray = np.flatnonzero(np.diff(times) > timeframe)
+    inserts: dict[int, np.ndarray] = {}
+    ranges: list[tuple[float, float]] = []
+    budget: int = max_rows
+    # Match IB Client.history_schedule()'s first-pass 31-day bound;
+    # 86400 converts elapsed days to seconds, not local venue days.
+    # TODO: advertise query limits through backend history config.
+    max_schedule_span_s: int = 31 * 86400
+    for offset in reversed(gaps[-newest_gaps:]):
+        left: float = float(times[offset])
+        right: float = float(times[offset + 1])
+        if right - left > max_schedule_span_s:
+            continue
+        try:
+            # Backend-owned callback supplied in history config.
+            # IB's open_history_client() routes it through the proxy
+            # to Client.history_schedule(), returning UTC intervals.
+            schedule: dict = await query_schedule(left, right)
+        except Exception as exc:
+            log.warning(f'Schedule unavailable for {left}:{right}: '
+                        f'{exc!r}')
+            continue
+        if (
+            schedule['start'] > left
+            or
+            schedule['end'] < right
+            or
+            schedule['use_rth']
+            or
+            not schedule['sessions']
+        ):
+            continue
+        stamps: list[np.ndarray] = []
+        count: int = 0
+        for start, end in schedule['sessions']:
+            first: float = max(left + timeframe, start)
+            # Preserve the existing sample grid across closures.
+            first = left + math.ceil(
+                (first - left) / timeframe
+            ) * timeframe
+            stop: float = min(right, end)
+            size: int = max(0, math.ceil((stop - first) / timeframe))
+            count += size
+            if count > budget:
+                break
+            if size:
+                stamps.append(first + np.arange(size) * timeframe)
+        if not count:
+            continue
+        if count > budget:
+            # Recoverable: leave this entire gap packed/unresolved.
+            # No rows or provenance have been published for it, so
+            # later gaps can still use the remaining reservation
+            # budget. Do not partially reserve or reindex this gap.
+            log.warning(f'Reservation exceeds SHM capacity: '
+                        f'{left}:{right}')
+            continue
+        missing: np.ndarray = np.concatenate(stamps)
+        reserved: np.ndarray = np.zeros(count, dtype=frame.dtype)
+        reserved['time'] = missing
+        # COPY CLOSE / SYNTHESIZE TIMESTAMPS for scheduled slots.
+        # Return explicit provenance; zero volume is NOT a flag.
+        reserved[['open', 'high', 'low', 'close']] = (
+            frame['close'][offset]
+        )
+        inserts[int(offset)] = reserved
+        budget -= count
+        # `np.diff` locates breaks in the synthetic sample cadence.
+        # Add 1 to each break's left index to select the first row
+        # of the next run. `np.split` returns one array per run;
+        # encode each as [first timestamp, last + timeframe).
+        # E.g. [101, 102, 108] -> [101, 103), [108, 109) at 1s.
+        # Separate runs keep venue closures OUT of synthetic ranges.
+        for group in np.split(
+            missing,
+            np.flatnonzero(np.diff(missing) != timeframe) + 1,
+        ):
+            ranges.append((
+                float(group[0]), float(group[-1] + timeframe),
+            ))
+    if not inserts:
+        return frame, []
+    pieces: list[np.ndarray] = []
+    previous: int = 0
+    for offset, reserved in sorted(inserts.items()):
+        pieces.extend([frame[previous:offset + 1], reserved])
+        previous = offset + 1
+    pieces.append(frame[previous:])
+    result: np.ndarray = np.concatenate(pieces)[-max_rows:]
+    first_t: float = float(result['time'][0])
+    ranges = [(max(a, first_t), b) for a, b in ranges if b > first_t]
+    return result, sorted(ranges)

tests/providers/test_history_reservations.py

@@ -0,0 +1,248 @@
+# piker: trading gear for hackers
+# Copyright (C) 2018-present  Tyler Goodlet (in stewardship of pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Schedule-driven startup allocation without published-row movement.
+
+'''
+from functools import partial
+from pathlib import Path
+
+import numpy as np
+import pytest
+import trio
+
+from piker.data._source import def_iohlcv_fields
+from piker.tsp._reserve import reserve_history_gaps
+
+
+@pytest.mark.parametrize('case', [
+    'covered',  # Reserve open-session slots; exclude the closure.
+    'partial',  # Coverage misses the gap's left margin.
+    'timeout',  # Provider failure is not evidence of a closure.
+    'rth',  # Regular-hours sessions omit extended trading hours.
+    'capacity',  # Too many placeholders: leave the gap unresolved.
+    'empty',  # No sessions: do not assume the venue was closed.
+])
+def test_reservation_requires_covered_extended_sessions(case):
+    '''
+    Interrupted backfill leaves packed rows around missing time.
+
+    Model a closure inside that interval: reserve only open-session
+    samples and retain explicit synthetic ranges. Partial coverage,
+    provider failure, RTH-only replies, and insufficient capacity
+    must never invent a schedule. Input rows remain immutable and
+    output placeholders carry zero activity, not invented trades.
+
+    '''
+    frame = np.zeros(2, dtype=np.dtype(def_iohlcv_fields))
+    frame['time'] = [100, 110]
+    frame['close'] = [7, 9]
+    frame['volume'] = [4, 5]
+    before = frame.copy()
+    async def query(left, right):
+        assert (left, right) == (100, 110)
+        if case == 'timeout':
+            raise TimeoutError
+        return {
+            'start': 102 if case == 'partial' else 100,
+            'end': 111,
+            'sessions': [] if case == 'empty' else [
+                (100, 104), (108, 111),
+            ],
+            'use_rth': case == 'rth',
+        }
+    result, ranges = trio.run(partial(
+        reserve_history_gaps, frame, 1, query,
+        max_rows=3 if case == 'capacity' else 20,
+    ))
+    np.testing.assert_array_equal(frame, before)
+    if case == 'covered':
+        assert result['time'].tolist() == [100, 101, 102, 103,
+                                           108, 109, 110]
+        assert ranges == [(101, 104), (108, 110)]
+        assert result['volume'].tolist() == [4, 0, 0, 0, 0, 0, 5]
+        assert result['close'].tolist() == [7, 7, 7, 7, 7, 7, 9]
+    else:
+        np.testing.assert_array_equal(result, before)
+        assert ranges == []
+
+
+@pytest.mark.parametrize('newest_gaps', [-1, 0, 1, 2])
+def test_reservation_limits_newest_gap_requests(
+    newest_gaps: int,
+) -> None:
+    '''
+    A configurable request budget must bound provider calls.
+
+    Two gaps distinguish newest-first selection from array order.
+    Zero must disable queries, not select all gaps via `[-0:]`;
+    negative budgets are rejected before any provider call. Check
+    both fetched intervals and resulting synthetic timestamps.
+
+    '''
+    frame: np.ndarray = np.zeros(3, dtype=def_iohlcv_fields)
+    frame['time'] = [100, 103, 106]
+    calls: list[tuple[float, float]] = []
+
+    async def query(left: float, right: float) -> dict:
+        calls.append((left, right))
+        return {
+            'start': left,
+            'end': right,
+            'use_rth': False,
+            'sessions': [(left, right)],
+        }
+
+    run = partial(
+        reserve_history_gaps,
+        frame, 1, query, max_rows=20, newest_gaps=newest_gaps,
+    )
+    if newest_gaps < 0:
+        with pytest.raises(ValueError, match='nonnegative'):
+            trio.run(run)
+        assert calls == []
+        return
+
+    result, ranges = trio.run(run)
+    assert calls == [(103, 106), (100, 103)][:newest_gaps]
+    expected: dict[int, list[int]] = {
+        0: [100, 103, 106],
+        1: [100, 103, 104, 105, 106],
+        2: [100, 101, 102, 103, 104, 105, 106],
+    }
+    assert result['time'].tolist() == expected[newest_gaps]
+    assert len(ranges) == newest_gaps
+    np.testing.assert_array_equal(frame['time'], [100, 103, 106])
+
+
+def test_startup_reserves_shm_without_persisting_synthetic_rows(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Restart after interrupted backfill must retain gap provenance.
+
+    Seed parquet with two separated observations, then run the real
+    startup backfiller and SHM prepend. A fake schedule supplies a
+    maintenance closure inside the hole. Wait for the actual tail
+    repair to complete before cancelling the service task. Assert
+    reserved open-session rows exist in SHM with queryable metadata,
+    while a fresh parquet load contains only provider observations.
+
+    '''
+    from contextlib import asynccontextmanager
+    from types import SimpleNamespace
+    from uuid import uuid4
+
+    from pendulum import duration, from_timestamp
+    import tractor
+
+    from piker.data._sharedmem import maybe_open_shm_array
+    from piker.storage.nativedb import (
+        NativeStorageClient, ohlc_key_map,
+    )
+    from piker.tsp import _history
+    from piker.tsp._reserve import (
+        reservation_ranges,
+    )
+
+    epoch = 1790193489
+    fqme = 'schedule.test'
+    finished = trio.Event()
+    def bars(offsets):
+        frame = np.zeros(len(offsets), dtype=def_iohlcv_fields)
+        frame['time'] = np.asarray(offsets) + epoch
+        for field in ('open', 'high', 'low', 'close'):
+            frame[field] = 100
+        frame['volume'] = 1
+        return frame
+    async def history(tf, end_dt=None, **kw):
+        frame = bars([11, 12] if end_dt is None else [10, 11])
+        return (frame, from_timestamp(frame['time'][0]),
+                from_timestamp(frame['time'][-1]))
+    async def schedule(left, right):
+        return {
+            'start': epoch, 'end': epoch + 13, 'use_rth': False,
+            'sessions': [
+                (epoch, epoch + 4), (epoch + 8, epoch + 13),
+            ],
+        }
+    @asynccontextmanager
+    async def history_client(*args, **kwargs):
+        yield history, {
+            'frame_types': {1: duration(seconds=2)},
+            'query_schedule': schedule,
+        }
+    real_repair = _history.maybe_fill_null_segments
+    async def repair(**kwargs):
+        await real_repair(**kwargs)
+        finished.set()
+    monkeypatch.setattr(_history, 'maybe_fill_null_segments', repair)
+    # Isolate actor-local provenance as well as SHM and parquet.
+    monkeypatch.setattr(
+        'piker.tsp._reserve._reservations', {},
+    )
+    class Sampler:
+        async def send(self, msg):
+            pass
+    async def main():
+        store = NativeStorageClient(tmp_path)
+        await store.update_ohlcv(fqme, bars([0, 10]), 1)
+        with trio.fail_after(5):
+            async with tractor.open_root_actor(
+                name='schedule-test',
+                tpt_bind_addrs=[('127.0.0.1', 0)],
+            ):
+                shm, opened = maybe_open_shm_array(
+                    key=f'schedule_{uuid4().hex}', size=100,
+                    dtype=np.dtype(def_iohlcv_fields),
+                    append_start_index=80,
+                )
+                assert opened
+                async with trio.open_nursery() as nursery:
+                    nursery.start_soon(partial(
+                        _history.tsdb_backfill,
+                        mod=SimpleNamespace(
+                            name='fake',
+                            open_history_client=history_client,
+                        ),
+                        storemod=SimpleNamespace(
+                            ohlc_key_map=ohlc_key_map,
+                        ),
+                        storage=store,
+                        mkt=SimpleNamespace(
+                            fqme=fqme, get_fqme=lambda **kw: fqme,
+                            dst=SimpleNamespace(atype='future'),
+                            src=SimpleNamespace(atype='fiat'),
+                        ),
+                        shm=shm, timeframe=1,
+                        sampler_stream=Sampler(),
+                    ))
+                    await finished.wait()
+                    nursery.cancel_scope.cancel()
+                assert (shm.array['time'] - epoch).tolist() == [
+                    0, 1, 2, 3, 8, 9, 10, 11, 12,
+                ]
+                assert reservation_ranges(fqme, 1) == [
+                    (epoch + 1, epoch + 4), (epoch + 8, epoch + 10),
+                ]
+                fresh_store = NativeStorageClient(tmp_path)
+                disk, _, _ = await fresh_store.load(fqme, 1)
+                assert (disk['time'] - epoch).tolist() == [
+                    0, 10, 11, 12,
+                ]
+    trio.run(main)

tests/providers/test_ib_schedule.py

@@ -0,0 +1,175 @@
+# piker: trading gear for hackers
+# Copyright (C) 2018-present  Tyler Goodlet (in stewardship of pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Historical schedule normalization and request-boundary regressions.
+
+'''
+import asyncio
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from piker.brokers.ib.venues import normalize_schedule
+from piker.brokers.ib.api import Client
+
+
+def response(zone='America/New_York'):
+    '''
+    Describe two sessions separated by a maintenance closure.
+
+    '''
+    return SimpleNamespace(
+        timeZone=zone,
+        startDateTime='20260923-00:00:00',
+        endDateTime='20260925-00:00:00',
+        sessions=[SimpleNamespace(
+            startDateTime='20260923-18:00:00',
+            endDateTime='20260924-17:00:00',
+        )],
+    )
+
+
+def test_schedule_uses_dated_timezone_and_coverage():
+    '''
+    Historical allocation must use dated sessions, not today's hours.
+
+    A New York September session must convert using EDT and preserve
+    the response's wider coverage separately from its open interval.
+
+    '''
+    out = normalize_schedule(response(), False)
+    expected = datetime(2026, 9, 23, 22, tzinfo=UTC).timestamp()
+    assert out['sessions'][0][0] == expected
+    assert out['start'] < expected < out['end']
+    assert out['use_rth'] is False
+
+
+@pytest.mark.parametrize('stamp', [
+    '20261101-01:30:00', '20260308-02:30:00',
+])
+def test_schedule_rejects_ambiguous_or_nonexistent_time(stamp):
+    '''
+    A guessed DST fold can shift reservations by an hour.
+
+    Reject both the repeated autumn hour and nonexistent spring
+    hour instead of turning unknown coverage into a false closure.
+
+    '''
+    schedule = response()
+    schedule.startDateTime = stamp
+    with pytest.raises(ValueError, match='Ambiguous'):
+        normalize_schedule(schedule, False)
+
+
+def test_schedule_request_uses_extended_hours_and_utc():
+    '''
+    Schedule requests must match the bars endpoint's extended hours.
+
+    Exercise the real Client method with a fake IB transport. Assert
+    exact UTC request bounds and day rounding, and ensure response
+    coverage is not replaced by the caller's requested interval.
+
+    '''
+    calls = []
+    async def find_contracts(fqme):
+        assert fqme == 'mnq.cme.20261218'
+        return ['contract']
+    async def schedule(contract, **kwargs):
+        calls.append((contract, kwargs))
+        return response()
+    client = SimpleNamespace(
+        find_contracts=find_contracts,
+        ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
+    )
+    out = asyncio.run(Client.history_schedule(
+        client, 'mnq.cme.20261218',
+        datetime(2026, 9, 23, 0, tzinfo=UTC),
+        datetime(2026, 9, 24, 1, tzinfo=UTC),
+    ))
+    assert calls == [('contract', {
+        'numDays': 2, 'endDateTime': '20260924-01:00:00',
+        'useRTH': False,
+    })]
+    assert out['timezone'] == 'America/New_York'
+
+
+def test_schedule_timeout_is_not_an_empty_session_list():
+    '''
+    A failed schedule request must not classify the range as closed.
+
+    Block the simulated provider and apply an immediate timeout;
+    the adapter must propagate failure instead of returning coverage.
+
+    '''
+    async def find_contracts(fqme):
+        return ['contract']
+    async def schedule(*args, **kwargs):
+        await asyncio.sleep(100)
+    client = SimpleNamespace(
+        find_contracts=find_contracts,
+        ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
+    )
+    with pytest.raises(TimeoutError):
+        asyncio.run(Client.history_schedule(
+            client, 'mnq.cme',
+            datetime(2026, 9, 23, tzinfo=UTC),
+            datetime(2026, 9, 24, tzinfo=UTC),
+            timeout_s=0,
+        ))
+
+
+def test_schedule_rejects_reversed_session():
+    '''
+    A malformed session must not silently become a venue closure.
+
+    Reject the response instead of dropping its reversed interval
+    and reserving slots using the remaining incomplete session list.
+
+    '''
+    schedule = response()
+    schedule.sessions[0].endDateTime = '20260923-17:00:00'
+    with pytest.raises(ValueError, match='Invalid schedule session'):
+        normalize_schedule(schedule, False)
+
+
+def test_schedule_duration_counts_elapsed_time_across_dst():
+    '''
+    The autumn clock change makes this local day last 25 hours.
+
+    Request two elapsed days so duration rounding cannot leave an
+    hour uncovered by subtracting same-zone wall-clock datetimes.
+
+    '''
+    async def find_contracts(fqme):
+        return ['contract']
+
+    async def schedule(contract, **kwargs):
+        assert kwargs['numDays'] == 2
+        return response()
+
+    client = SimpleNamespace(
+        find_contracts=find_contracts,
+        ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
+    )
+    zone = ZoneInfo('America/New_York')
+    asyncio.run(Client.history_schedule(
+        client, 'mnq.cme',
+        datetime(2026, 11, 1, tzinfo=zone),
+        datetime(2026, 11, 2, tzinfo=zone),
+    ))

tests/test_history_backfill.py

@@ -548,7 +548,10 @@ def test_backfill_notification_timeout_is_bounded(
     trio.run(main)
 
 
-def test_synthetic_gap_times_require_valid_right_boundary() -> None:
+@pytest.mark.parametrize('epoch', [0, 1790193489])
+def test_synthetic_gap_times_require_valid_right_boundary(
+    epoch: int,
+) -> None:
     '''
     Synthetic rows must remain strictly between valid boundary bars.
 
@@ -557,35 +560,39 @@ def test_synthetic_gap_times_require_valid_right_boundary() -> None:
     null the right boundary's timestamp, creating a duplicate. Prove
     aligned bounds produce cadence while an overfull segment
     is rejected instead of inventing non-monotonic timestamps.
+    Repeat at a modern epoch: relative timestamp tolerance must not
+    allow extra slots merely because epoch seconds are large.
 
     '''
     aligned = _synthetic_gap_times(
-        start_t=60,
+        start_t=epoch + 60,
-        right_t=300,
+        right_t=epoch + 300,
         count=3,
         timeframe=60,
     )
     overfull = _synthetic_gap_times(
-        start_t=60,
+        start_t=epoch + 60,
-        right_t=240,
+        right_t=epoch + 240,
         count=3,
         timeframe=60,
     )
     trailing = _synthetic_gap_times(
-        start_t=60,
+        start_t=epoch + 60,
         right_t=None,
         count=3,
         timeframe=60,
     )
 
     assert aligned is not None
-    assert aligned.tolist() == [120, 180, 240]
+    assert aligned.tolist() == [epoch + t for t in (120, 180, 240)]
     assert overfull is None
     assert trailing is None
 
 
+@pytest.mark.parametrize('reply', ['empty', 'nodata', 'unavailable'])
 def test_null_repair_falls_back_without_debug_pause(
     monkeypatch: pytest.MonkeyPatch,
+    reply: str,
 ) -> None:
     '''
     Empty provider repair must resolve interior zero rows locally.
@@ -594,12 +601,14 @@ def test_null_repair_falls_back_without_debug_pause(
     an IB frame. The null repair queried IB while reverse backfill
     occupied its sole history request slot, then could
     enter an interactive debug pause before reaching forward-fill.
-    Arrange an interior null run, return an empty provider frame, and
+    Arrange an interior null run, return no provider history, and
     wedge sampler notification. A short timeout and fail-fast pause
     prove local fallback needs no interactive or IPC progress.
     The assertions verify only the
     zero rows inherit the preceding close and synthesized timestamps;
     both valid boundary rows remain unchanged.
+    Exercise empty arrays and both typed no-history outcomes; each
+    must reach the same fallback without requiring a debugger.
 
     '''
     monkeypatch.setattr(
@@ -654,6 +663,12 @@ def test_null_repair_falls_back_without_debug_pause(
         Return no provider bars for the requested null segment.
 
         '''
+        from piker.brokers import NoData
+
+        if reply == 'nodata':
+            raise NoData(info=kwargs)
+        if reply == 'unavailable':
+            raise DataUnavailable('no history')
         end_dt = kwargs['end_dt']
         empty = np.empty(
             0,
@@ -708,6 +723,57 @@ def test_null_repair_falls_back_without_debug_pause(
     assert frame['volume'].tolist() == [5, 0, 0, 0, 6]
 
 
+def test_null_repair_leaves_packed_time_gap_unchanged(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A missing time interval is not an allocated zero-time segment.
+
+    Cancellation during reverse backfill can leave two valid stored
+    frames separated in time. Reload packs their rows into adjacent
+    SHM slots. The null repair must not overwrite these valid bars
+    with synthetic values or pretend it recovered missing trades.
+    Use the observed MNQ gap with adjacent physical indexes; fail
+    on any provider request and compare every field after the call.
+    This documents the separate timestamp-gap repair still needed.
+
+    '''
+    import greenback
+
+    frame = np.zeros(2, dtype=np.dtype(def_iohlcv_fields))
+    frame['index'] = [100, 101]
+    frame['time'] = [1790193489, 1790248914]
+    for field in ('open', 'high', 'low', 'close'):
+        frame[field] = [100, 110]
+    frame['volume'] = [3, 7]
+    before = frame.copy()
+
+    async def ensure_portal() -> None:
+        '''
+        Avoid installing a debugger portal for this unit test.
+
+        '''
+
+    async def get_hist(*args, **kwargs):
+        '''
+        Reject treating two valid rows as a zero-time reservation.
+
+        '''
+        raise AssertionError('packed gap reached null repair query')
+
+    monkeypatch.setattr(greenback, 'ensure_portal', ensure_portal)
+    trio.run(partial(
+        maybe_fill_null_segments,
+        shm=SimpleNamespace(array=frame),
+        timeframe=1,
+        get_hist=get_hist,
+        sampler_stream=object(),
+        mkt=SimpleNamespace(fqme='mnq.cme.20261218.ib'),
+        backfill_until_dt=from_timestamp(frame['time'][0]),
+    ))
+    np.testing.assert_array_equal(frame, before)
+
+
 @pytest.mark.parametrize('empty_reply', ['hmds_error', 'empty_bars'])
 def test_ib_empty_interval_preserves_older_bars_in_parquet(
     tmp_path: Path,
@@ -863,3 +929,52 @@ def test_ib_empty_interval_preserves_older_bars_in_parquet(
                 assert gaps[gaps > 1].tolist() == [2000]
 
     trio.run(main)
+
+
+@pytest.mark.parametrize('case', [
+    'aligned', 'overfull', 'unbounded', 'occupied',
+])
+def test_copy_close_to_null_rows_preserves_other_fields(
+    case: str,
+) -> None:
+    '''
+    Synthetic repair must be isolated from SHM and provider I/O.
+
+    Reserve two rows between valid bars at a modern epoch. Only a
+    fully null, correctly bounded run may receive the preceding
+    close and generated timestamps. Reject an occupied row, a
+    duplicate right timestamp, or a missing right margin without
+    any partial mutation. Distinct volume/count values prove this
+    helper copies neither the seed's activity nor entire records.
+
+    '''
+    from piker.tsp._history import _copy_close_to_null_rows
+
+    frame: np.ndarray = np.zeros(4, dtype=np.dtype(_ohlc_dtype))
+    epoch: int = 1790193489
+    frame['index'] = np.arange(100, 104)
+    frame['time'] = [epoch, 0, 0, epoch + 3]
+    frame['close'] = [123, 0, 0, 456]
+    frame['volume'] = [8, 2, 3, 9]
+    frame['count'] = [4, 5, 6, 7]
+    right_t: float|None = float(frame['time'][-1])
+    if case == 'overfull':
+        right_t -= 1
+    elif case == 'unbounded':
+        right_t = None
+    elif case == 'occupied':
+        frame['time'][1] = epoch + 1
+    before: np.ndarray = frame.copy()
+
+    copied: int = _copy_close_to_null_rows(
+        frame[1:3], frame[0], right_t, 1,
+    )
+    expected: np.ndarray = before.copy()
+    if case == 'aligned':
+        expected['time'][1:3] = [epoch + 1, epoch + 2]
+        for field in ('open', 'high', 'low', 'close'):
+            expected[field][1:3] = 123
+        assert copied == 2
+    else:
+        assert copied == 0
+    np.testing.assert_array_equal(frame, expected)