ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.md

@@ -1,64 +0,0 @@
----
-model: gpt-6 (exact variant not exposed)
-service: codex
-timestamp: 2026-09-23T16:46:56.830897+00:00
-git_ref: efa72980
-session: 01a0a5b9-f872-7ca3-ab8a-d1cffc14e895
-scope: code
-substantive: true
-raw_file: 20260923T164656Z_efa72980_prompt_io.raw.md
----
-
-## Prompt
-
-Investigate existing branch fixes before repairing IB day-skipping.
-Create an isolated worktree on ib_connectivity_harden and add a test
-that checks actual persisted data, leaving live gap repair for later.
-
-## Response summary
-
-Preserve IB request boundaries on HMDS no-data responses. Retain the
-existing earliest-history classification and bounded reset handling.
-Separate requested and published boundaries during backfill so the
-first response after an empty interval does not lose its end bar.
-Prove supplied bars survive both SHM publication and a fresh parquet
-load while a simulated venue closure remains empty.
-
-## Files changed
-
-- `piker/brokers/ib/feed.py`: complete HMDS no-data requests unchanged.
-- `piker/tsp/_history.py`: filter overlap against published history.
-- `tests/test_ib_history.py`: six empty-response boundary cases.
-- `tests/test_history_backfill.py`: two SHM/parquet integration cases.
-
-## Human edits
-
-The human identified the live gap, required prior-branch discovery,
-selected ib_connectivity_harden as the base, and requested stronger
-end-to-end evidence. They corrected overstated reproduction claims,
-insisted on plain explanations, and pointed out that valid venue
-closures must remain gaps. Existing data repair was explicitly deferred.
-No direct human source edits have been observed; changes await review.
-
-## Review follow-up
-
-The human's tuicr review asked whether None was part of the backfill
-protocol and requested plain explanations of the two dates and overlap
-filter. Clarified the IB-internal empty-response sentinel and its
-translation into NoData/DataUnavailable before reaching the backfiller.
-Renamed last_start_dt to request_end_dt and published_start_dt to
-oldest_shm_dt. Added a concrete empty-interval example at the filter and
-explained that only successful SHM writes move oldest_shm_dt.
-These changes clarify naming and documentation without changing the
-previously tested request or publication behavior.
-
-## History-filter review follow-up
-
-The human flagged the unused append_until_dt argument and confusing
-append/prepend terminology in diff_history(). The helper had one
-caller, using only prepend_until_dt. Removed the helper and inlined
-its inclusive lower-time cutoff beside the SHM duplicate filter.
-The human's nearby TODO comment is resolved by this removal; the
-replacement comment describes the retained timestamps directly.
-The storage comparison remains >= backfill_until_dt; the SHM
-comparison remains < oldest_shm_dt.

ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.raw.md

@@ -1,87 +0,0 @@
----
-model: gpt-6 (exact variant not exposed)
-service: codex
-timestamp: 2026-09-23T16:46:56.830897+00:00
-git_ref: efa72980
-diff_cmd: git diff efa72980 -- piker/brokers/ib/feed.py piker/tsp/_history.py tests/test_ib_history.py tests/test_history_backfill.py
----
-
-## Prompt
-
-The human requested the IB request-boundary repair in a new worktree
-based on ib_connectivity_harden and a test exercising real persistence.
-The existing MNQ parquet gap is outside this repair scope.
-
-## Code output
-
-> `git diff efa72980 -- piker/brokers/ib/feed.py`
-
-> `git diff efa72980 -- piker/tsp/_history.py`
-
-> `git diff efa72980 -- tests/test_ib_history.py`
-
-> `git diff efa72980 -- tests/test_history_backfill.py`
-
-get_bars() now completes HMDS no-data responses without changing the
-requested date. open_history_client() retains its NoData and earliest
-history classification. start_backfill() tracks the published boundary
-separately from the request cursor after empty intervals.
-
-## Unedited progress output
-
-The new test failed on the original code as intended: the
-temporary parquet contained only 4 bars instead of 2,004.
-The day-skip caused 2,000 available bars to be omitted.
-
-The test also includes a simulated venue closure that must remain
-a gap. It checks that we fetch available bars without inventing
-data for an empty interval.
-
-The IB fix now preserves all 2,004 bars in parquet. The SHM
-assertion exposed a second bug: after stepping over an empty
-interval, the backfiller drops the last bar of the next response
-as though it were already published.
-
-I’ll separate the request date from the published-data boundary
-so the same test can verify that both memory and disk retain
-every available bar.
-
-## Validation
-
-Using the existing fsp_backfill_sync/py313 interpreter, with cwd in
-ib_history_request_bounds and the Piker import checked against that cwd:
-
-- tests/test_history_backfill.py: 11 passed in 1.92s.
-- tests/test_ib_history.py: 17 passed in 0.36s.
-- Ruff on all four changed Python files: all checks passed.
-- git diff --check: passed.
-
-The integration test uses simulated IB bars responses, real IB history
-adapter and backfill logic, real SHM, and temporary NativeDB parquet.
-It checks a fresh disk load. It does not test a live gateway, actor RPC,
-or Qt rendering, and does not prove the origin of the user's live gap.
-
-## Unedited review response
-
-`None` is internal to the IB adapter. The backfiller receives
-`NoData`, or `DataUnavailable` when the requested date has reached
-the known beginning of history. I’ll make that distinction explicit.
-
-The two dates answer different questions: “where should the next
-history query end?” and “what is the oldest bar already in SHM?”
-An empty reply changes only the first. I’ll rename them to
-`request_end_dt` and `oldest_shm_dt` and explain the filtering with
-a concrete example.
-
-> `git diff efa72980 -- piker/brokers/ib/feed.py piker/tsp/_history.py`
-
-## History-filter review follow-up
-
-The human flagged the unused append_until_dt argument and confusing
-append/prepend terminology in diff_history(). The helper had one
-caller, using only prepend_until_dt. Removed the helper and inlined
-its inclusive lower-time cutoff beside the SHM duplicate filter.
-The human's nearby TODO comment is resolved by this removal; the
-replacement comment describes the retained timestamps directly.
-The storage comparison remains >= backfill_until_dt; the SHM
-comparison remains < oldest_shm_dt.

ai/prompt-io/opencode/20260916T194052Z_749ca1f0_prompt_io.md

@@ -1,37 +0,0 @@
----
-model: openai/gpt-6-astra
-service: opencode
-session: unavailable
-timestamp: 2026-09-16T19:40:52Z
-git_ref: 749ca1f0
-scope: code
-substantive: true
-raw_file: 20260916T194052Z_749ca1f0_prompt_io.raw.md
----
-
-## Prompt
-
-Fix wrong IB fill/commission pairing and empty ledger normalization;
-retain and investigate silent socket loss during network roaming.
-
-## Response summary
-
-Preserve execution identity from callbacks into accounting, prevent
-empty commission IDs from erasing ledger identity, and propagate actual
-socket disconnection as a transport exception. The existing data-farm
-reset mechanism is not a socket-session reconnect implementation.
-
-## Files changed
-
-- `piker/brokers/ib/broker.py`: fill correlation and socket termination.
-- `piker/brokers/ib/ledger.py`: authoritative execution ID.
-- `piker/brokers/ib/api.py`: disconnect event propagation.
-- `tests/test_ib_trade_events.py`: accounting event regressions.
-- `tests/test_ib_method_proxy.py`: socket-disconnect regression.
-
-## Human edits
-
-The human supplied live traceback and debugger values proving the
-execution mismatch, requested a regression rather than a KeyError
-guard, and required continued attention to roaming recovery. No human
-source edits to this patch have been observed; acceptance is pending.

ai/prompt-io/opencode/20260916T194052Z_749ca1f0_prompt_io.raw.md

@@ -1,45 +0,0 @@
----
-model: openai/gpt-6-astra
-service: opencode
-timestamp: 2026-09-16T19:40:52Z
-git_ref: 749ca1f0
-diff_cmd: git diff HEAD -- piker/brokers/ib tests/test_ib_trade_events.py tests/test_ib_method_proxy.py
----
-
-## Human instructions and evidence
-
-Fix silent IB socket loss after network roaming, reported as
-`[Errno 104] Connection reset by peer`. Also investigate a live
-MNQ accounting crash in `emit_pp_update()` with empty normalized
-ledger entries. The human supplied debugger evidence that a commission
-for execution 4261 was paired with execution 4262 on the same contract
-and account, and explicitly requested implementation and regression
-tests. They asked to retain the roaming task and continue working
-through follow-up messages.
-
-## Generated changes
-
-> `git diff HEAD -- piker/brokers/ib/broker.py`
-
-Preserve the callback-supplied fill with its commission. Track positions
-by account and contract ID and pending commissioned fills by execution.
-Reject mismatched commission/execution IDs. Propagate socket closure
-instead of returning normally from the trade-event bridge.
-
-> `git diff HEAD -- piker/brokers/ib/ledger.py`
-
-Keep Execution.execId authoritative when flattening commission fields.
-
-> `git diff HEAD -- piker/brokers/ib/api.py`
-
-Relay the authoritative disconnected event as ConnectionError, retaining
-the most recent API error text, and raise it from the proxy reader.
-Automatic session reconnect/resynchronization remains unimplemented.
-
-> `git diff HEAD -- tests/test_ib_trade_events.py tests/test_ib_method_proxy.py`
-
-Cover overlapping executions, zero commissions, position-before/after
-commission ordering, empty commission IDs, and eventkit socket loss.
-Nine focused tests passed, including the actual eventkit callback
-bridge and its disconnect exception. Focused Ruff checks and
-`git diff --check` also passed.

ai/prompt-io/opencode/20260916_ib_connectivity_integration.md

@@ -1,61 +0,0 @@
----
-model: openai/gpt-6-astra
-service: opencode
-scope: code
-substantive: true
-git_ref: 5eb4fc9d
-raw_file: 20260916_ib_connectivity_integration.raw.md
----
-
-## Prompt
-
-Investigate IB Gateway loss, silent chart survival, infected-actor
-shutdown errors and leaked shared memory. Integrate pending tagged
-address and broadcast-consumer compatibility work. The human required
-isolated worktrees and preservation of the live setup, and selected
-tractor piker_pin rather than main as the runtime fix base.
-
-## Response summary
-
-Created isolated piker and tractor worktrees and initially integrated
-c07e3cbd and 4af885d9 without committing. That pending merge was later
-aborted. The current piker scope is the original IB fail-loud patch
-and its regression tests on 5eb4fc9d; neither upstream history is
-merged. The live interpreter installation was not changed.
-
-Integration-era logging and Deribit compatibility fixes were archived
-outside this patch and are absent from the current worktree. The raw
-companion preserves that earlier response as historical output.
-
-## Generated files
-
-- piker/brokers/ib/api.py: propagate actual socket disconnection with
-  the last API error through the method-proxy relay.
-- piker/brokers/ib/broker.py: raise on trade-event socket loss and
-  propagate the disconnect event to the trade consumer.
-- tests/test_ib_method_proxy.py: exercise API error plus disconnect.
-- tests/test_ib_trade_events.py: preserve callback fill identity and
-  raise when the callback bridge disconnects.
-
-Existing IB fail-loud edits and their tests were copied from the
-earlier session, whose provenance remains in the fill commit.
-The commit plan includes these four files and both provenance logs.
-
-## Verification
-
-Historically, twenty selected piker regression cases passed on the
-integrated tree against isolated, patched tractor sources. Those
-results do not validate the current post-abort tree or installed
-pinned tractor. The new exact-tree commit-plan checks remain pending;
-no project tests were run during this planning pass. No live
-service/container tests or installed dependency changes were performed.
-Original live-worktree diff matched SHA256
-cbdfa439e656ef3a5effaa0f0e1a7706f1070de863fc0e85222d36294eed8524.
-
-## Human contributions
-
-The human identified the dependency branches, corrected the proposed
-tractor main base to piker_pin, and required a new worktree after the
-agent initially started an uncommitted merge in the live checkout.
-That merge was aborted and preservation was verified before isolated
-work continued. The human supplied all live incident evidence.

ai/prompt-io/opencode/20260916_ib_connectivity_integration.raw.md

@@ -1,25 +0,0 @@
----
-model: openai/gpt-6-astra
-service: opencode
-git_ref: 5eb4fc9d
-diff_cmd: git diff HEAD
----
-
-> git diff HEAD -- piker/data/feed.py piker/brokers/deribit/api.py piker/brokers/deribit/feed.py
-
-Use the standard logging predicate for INFO and repair import-time
-Deribit message struct/annotation compatibility encountered while
-validating the merged broadcast-consumer baseline.
-
-The isolated checks pass, including real shm cleanup:
-
-- All 20 selected piker integration cases pass after the narrow
-  compatibility fixes.
-- Tractor's EOF tests pass for TCP and UDS, including an idle
-  owner and an unaffected peer.
-- All four late-cancellation variants unlink their test-owned
-  data/first/last shm segments before parent teardown.
-
-The original live-worktree diff still matches its saved hash.
-We're ready for a Gateway-container test of the full actor tree;
-automatic reconnect is still a separate, unfinished step.

piker/brokers/ib/api.py

@@ -1276,11 +1276,8 @@ class Client:
         self.ib.errorEvent.connect(push_err)
 
         api_err = self.ib.client.apiError
-        last_api_error: str = ''
 
         def report_api_err(msg: str) -> None:
-            nonlocal last_api_error
-            last_api_error = msg
             with remove_handler_on_err(
                 api_err,
                 report_api_err,
@@ -1293,25 +1290,6 @@ class Client:
 
         api_err.connect(report_api_err)
 
-        disconnected = self.ib.disconnectedEvent
-
-        def report_disconnect() -> None:
-            # IB.disconnectedEvent is authoritative for socket loss;
-            # numeric IB farm/order errors are a different namespace
-            # from OS errno values reported through Client.apiError.
-            disconnected.disconnect(report_disconnect)
-            with remove_handler_on_err(
-                disconnected,
-                report_disconnect,
-            ):
-                reason = last_api_error or 'IB API socket disconnected'
-                to_trio.send_nowait((
-                    'disconnected',
-                    ConnectionError(reason),
-                ))
-
-        disconnected.connect(report_disconnect)
-
     def positions(
         self,
         account: str = '',
@@ -1873,9 +1851,6 @@ async def relay_client_proxy_messages(
             #     proxy.reset()
 
             match msg:
-                case ('disconnected', ConnectionError() as err):
-                    raise err
-
                 # Correlated response from a method call.
                 case {'mid': _}:
                     proxy._deliver_method_response(msg)

piker/brokers/ib/broker.py

@@ -338,8 +338,7 @@ async def recv_trade_updates(
 
             case 'commissionReportEvent':
                 assert report
-                assert fill is not None
+                emit: CommissionReport = report
-                emit: tuple[Fill, CommissionReport] = (fill, report)
 
             case 'execDetailsEvent':
                 # execution details event
@@ -391,7 +390,6 @@ async def recv_trade_updates(
 
     # let the engine run and stream
     await client.ib.disconnectedEvent
-    raise ConnectionError('IB trade-event socket disconnected')
 
 
 async def update_and_audit_pos_msg(
@@ -1191,10 +1189,15 @@ async def deliver_trade_events(
     Format and relay all trade events for a given client to emsd.
 
     '''
-    # Positions belong to an account/contract; commissions belong to
+    # task local msg dialog tracking
-    # executions. Never recover a Fill from just its Contract.
+    clears: dict[
-    positions: dict[tuple[str, int], IbPosition] = {}
+        Contract,
-    pending_fills: dict[tuple[str, int], dict[str, Fill]] = {}
+        list[
+            IbPosition | None,  # filled by positionEvent
+            Fill | None,  # filled by order status and exec details
+        ]
+    ] = {}
+    execid2con: dict[str, Contract] = {}
 
     # TODO: for some reason we can receive a ``None`` here when the
     # ib-gw goes down? Not sure exactly how that's happening looking
@@ -1293,38 +1296,105 @@ async def deliver_trade_events(
             # translating them to ``Status`` msgs) if we can
             # show the equivalent status events are no more latent.
             case 'execDetailsEvent':
-                # Accounting waits for commissionReportEvent, whose
+                # unpack attrs pep-0526 style.
-                # callback supplies the exact Fill, including zero
+                trade: Trade
-                # commissions. Order-status handling relays fills
+                fill: Fill
-                # independently above.
+                trade, fill = item
-                continue
+                con: Contract = trade.contract
+                execu: Execution = fill.execution
+                execid: str = execu.execId
+                report: CommissionReport = fill.commissionReport
+
+                # always fill in id to con map so when commissions
+                # arrive we can maybe fire the pos update..
+                execid2con[execid] = con
+
+                # TODO:
+                # - normalize out commissions details?
+                # - this is the same as the unpacking loop above in
+                # ``trades_to_ledger_entries()`` no?
+
+                # 2 cases:
+                # - fill comes first or
+                # - commission report comes first
+                clear: tuple = clears.setdefault(
+                    con,
+                    [None, fill],
+                )
+                pos, _fill = clear
+
+                # NOTE: we have to handle the case where a pos msg
+                # has already been set (bc we already relayed rxed
+                # one before both the exec-deats AND the
+                # comms-report?) but the comms-report hasn't yet
+                # arrived, so we fill in the fill (XD) and wait for
+                # the cost to show up before relaying the pos msg
+                # to the EMS..
+                if _fill is None:
+                    clear[1] = fill
+
+                cost: float = report.commission
+                if (
+                    pos
+                    and fill
+                    and cost
+                ):
+                    await emit_pp_update(
+                        ems_stream,
+                        accounts_def,
+                        proxies,
+                        ledgers,
+                        tables,
+
+                        ibpos=pos,
+                        fill=fill,
+                    )
+                    clears.pop(con)
 
             case 'commissionReportEvent':
-                fill, cr = item
-                execid: str = cr.execId
-                if execid != fill.execution.execId:
-                    raise ValueError(
-                        f'IB commission/fill execution mismatch:\n'
-                        f'{cr!r}\n{fill!r}'
-                    )
-                key = (
-                    fill.execution.acctNumber,
-                    fill.contract.conId,
-                )
-                fill = fill._replace(commissionReport=cr)
-                if (pos := positions.get(key)) is None:
-                    pending_fills.setdefault(key, {})[execid] = fill
-                    continue
 
-                await emit_pp_update(
+                cr: CommissionReport = item
-                    ems_stream,
+                execid: str = cr.execId
-                    accounts_def,
+
-                    proxies,
+                # only fire a pp msg update if,
-                    ledgers,
+                # - we haven't already
-                    tables,
+                # - the fill event has already arrived
-                    ibpos=pos,
+                #   but it didn't yet have a commision report
-                    fill=fill,
+                #   which we fill in now.
-                )
+
+                # placehold i guess until someone who know wtf
+                # contract this is from can fill it in...
+                con: Contract | None = execid2con.setdefault(execid, None)
+                if (
+                    con
+                    and (clear := clears.get(con))
+                ):
+                    pos, fill = clear
+                    if (
+                        pos
+                        and fill
+                    ):
+                        now_cr: CommissionReport = fill.commissionReport
+                        if (now_cr != cr):
+                            log.warning(
+                                'UhhHh ib updated the commission report mid-fill..?\n'
+                                f'was: {pformat(cr)}\n'
+                                f'now: {pformat(now_cr)}\n'
+                            )
+
+                        await emit_pp_update(
+                            ems_stream,
+                            accounts_def,
+                            proxies,
+                            ledgers,
+                            tables,
+
+                            ibpos=pos,
+                            fill=fill,
+                        )
+                        clears.pop(con)
+                        # TODO: should we clean this?
+                        # execid2con.pop(execid)
 
             # always update with latest ib pos msg info since
             # we generally audit against it for sanity and
@@ -1337,18 +1407,10 @@ async def deliver_trade_events(
                 bs_mktid, ppmsg = pack_position(pos, accounts_def)
                 log.info(f'New IB position msg: {ppmsg}')
 
-                key = (pos.account, con.conId)
+                _, fill = clears.setdefault(
-                positions[key] = pos
+                    con,
-                for fill in pending_fills.pop(key, {}).values():
+                    [pos, None],
-                    await emit_pp_update(
+                )
-                        ems_stream,
-                        accounts_def,
-                        proxies,
-                        ledgers,
-                        tables,
-                        ibpos=pos,
-                        fill=fill,
-                    )
                 # only send a pos update once we've actually rxed
                 # the msg from IB since generally speaking we use
                 # their 'cumsize' as gospel.
@@ -1362,9 +1424,6 @@ async def deliver_trade_events(
                     ibpos=pos,
                 )
 
-            case 'disconnected':
-                raise item
-
             case 'error':
                 # NOTE: see impl deats in
                 # `Client.inline_errors()::push_err()`

piker/brokers/ib/feed.py

@@ -38,6 +38,7 @@ from async_generator import aclosing
 import ib_async as ibis
 import numpy as np
 from pendulum import (
+    now,
     from_timestamp,
     Duration,
     duration as mk_duration,
@@ -220,11 +221,7 @@ async def open_history_client(
                     f'mean: {mean}'
                 )
 
-            # get_bars() uses None internally for an empty reply.
+            # could be trying to retreive bars over weekend
-            # The backfiller never receives that sentinel: translate
-            # it into NoData, or DataUnavailable at the known start
-            # of the instrument's history. A venue closure can also
-            # yield an empty reply without exhausting all history.
             if out is None:
                 log.error(
                     f"No bars starting at {end_dt!r} !?!?"
@@ -459,6 +456,10 @@ async def get_bars(
     # how long before we trigger a feed reset (seconds)
     feed_reset_timeout: float = 3,
 
+    # how many days to subtract before giving up on further
+    # history queries for instrument, presuming that most don't
+    # not trade for a week XD
+    max_nodatas: int = 6,
     max_failed_resets: int = 6,
 
     task_status: TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED,
@@ -478,6 +479,7 @@ async def get_bars(
     `.ib.api.Client` methods.
 
     '''
+    nodatas_count: int = 0
     cancelled_count: int = 0
     failed_resets: int = 0
     pacing_reset_pending: bool = False
@@ -494,8 +496,8 @@ async def get_bars(
 
     async def query():
 
-        nonlocal result, data_cs
+        nonlocal result, data_cs, end_dt
-        nonlocal cancelled_count, failed_resets
+        nonlocal nodatas_count, cancelled_count, failed_resets
         nonlocal pacing_reset_pending, pacing_reset_completed
 
         while True:
@@ -623,21 +625,27 @@ async def get_bars(
                     # the frame dt index since the upper layer may
                     # be doing so concurrently and we don't want to
                     # be delivering frames that weren't asked for.
+                    # try to decrement start point and look further back
+                    # end_dt = end_dt.subtract(seconds=2000)
+                    logmsg = "SUBTRACTING DAY from DT index"
+                    if end_dt is not None:
+                        end_dt = end_dt.subtract(days=1)
+                    elif end_dt is None:
+                        end_dt = now().subtract(days=1)
+
                     log.warning(
                         f'NO DATA found ending @ {end_dt}\n'
+                        + logmsg
                     )
-                    # Match get_bars()'s empty-bars return: None is
+
-                    # internal to the IB adapter, not its backfill
+                    if nodatas_count >= max_nodatas:
-                    # protocol. open_history_client() converts it
+                        raise DataUnavailable(
-                    # to NoData (or DataUnavailable at history start).
+                            f'Presuming {fqme} has no further history '
-                    # Wake the waiting get_bars() task before exiting
+                            f'after {max_nodatas} tries..'
-                    # so it does not mistake this reply for a timeout.
+                        )
-                    result = None
+
-                    failed_resets = 0
+                    nodatas_count += 1
-                    result_ready.set()
+                    continue
-                    if data_cs:
-                        data_cs.cancel()
-                    return None
 
                 elif (
                     'API historical data query cancelled'

piker/brokers/ib/ledger.py

@@ -429,16 +429,9 @@ def api_trades_to_ledger_entries(
         for attr_name, val in fdict.items():
             match attr_name:
                 # value is a `@dataclass` subtype
-                case 'contract' | 'execution':
+                case 'contract' | 'execution' | 'commissionReport':
                     txn_dict.update(asdict(val))
 
-                case 'commissionReport':
-                    report_fields = asdict(val)
-                    # Execution.execId owns ledger identity. An empty
-                    # CommissionReport.execId must not erase it.
-                    report_fields.pop('execId')
-                    txn_dict.update(report_fields)
-
                 case 'time':
                     # ib has wack ns timestamps, or is that us?
                     continue

piker/tsp/_history.py

@@ -145,6 +145,29 @@ _rt_buffer_start = int((_days_worth - 1) * _secs_in_day)
 _notify_timeout_s: float = 1
 
 
+def diff_history(
+    array: np.ndarray,
+    append_until_dt: datetime|None = None,
+    prepend_until_dt: datetime|None = None,
+
+) -> np.ndarray:
+
+    # no diffing with tsdb dt index possible..
+    if (
+        prepend_until_dt is None
+        and
+        append_until_dt is None
+    ):
+        return array
+
+    times = array['time']
+
+    if append_until_dt:
+        return array[times < append_until_dt.timestamp()]
+    else:
+        return array[times >= prepend_until_dt.timestamp()]
+
+
 def mk_storage_key(mkt: MktPair) -> str:
     '''
     Build the historical storage key for a market.
@@ -563,24 +586,16 @@ async def start_backfill(
         # - after this loop continue to check for other gaps in the
         #   (tsdb) history and (at least report) maybe fill them
         #   from new frame queries to the backend?
-        # start_backfill() asks get_hist() for bars ending at
+        last_start_dt: datetime = backfill_from_dt
-        # request_end_dt. On NoData we move that date backward
-        # and try an earlier interval, even though no bars arrived.
-        request_end_dt: datetime = backfill_from_dt
-
-        # oldest_shm_dt is the earliest bar already written to SHM
-        # by this backfill. It changes only after a successful push.
-        # NoData changes request_end_dt but leaves this date alone.
-        oldest_shm_dt: datetime = backfill_from_dt
         next_prepend_index: int = backfill_from_shm_index
 
         est = timezone('EST')
 
-        while request_end_dt > backfill_until_dt:
+        while last_start_dt > backfill_until_dt:
             log.info(
                 f'Requesting {timeframe}s frame:\n'
                 f'backfill_until_dt: {backfill_until_dt}\n'
-                f'request_end_dt: {request_end_dt}\n'
+                f'last_start_dt: {last_start_dt}\n'
             )
             try:
                 (
@@ -589,23 +604,22 @@ async def start_backfill(
                     next_end_dt,
                 ) = await get_hist(
                     timeframe,
-                    end_dt=(end_dt_param := request_end_dt),
+                    end_dt=(end_dt_param := last_start_dt),
                 )
             except NoData as nodata:
                 _nodata = nodata
-                previous_request_end_dt: datetime = request_end_dt
+                orig_last_start_dt: datetime = last_start_dt
                 gap_report: str = (
-                    f'EMPTY FRAME for `end_dt: {request_end_dt}`?\n'
+                    f'EMPTY FRAME for `end_dt: {last_start_dt}`?\n'
                     f'{mod.name} -> tf@fqme: {timeframe}@{mkt.fqme}\n'
-                    f'request_end_dt: {previous_request_end_dt}\n'
+                    f'last_start_dt: {orig_last_start_dt}\n\n'
-                    f'\n'
                     f'bf_until: {backfill_until_dt}\n'
                 )
                 # EMPTY FRAME signal with 3 (likely) causes:
                 #
                 # 1. range contains legit gap in venue history
                 # 2. history actually (edge case) **began** at the
-                #    value `request_end_dt`
+                #    value `last_start_dt`
                 # 3. some other unknown error (ib blocking the
                 #    history-query bc they don't want you seeing how
                 #    they cucked all the tinas.. like with options
@@ -616,13 +630,13 @@ async def start_backfill(
                     # as maybe indicated by the backend to see if we
                     # can get older data before this possible
                     # "history gap".
-                    request_end_dt = request_end_dt.subtract(
+                    last_start_dt: datetime = last_start_dt.subtract(
                         seconds=def_frame_duration.total_seconds()
                     )
                     gap_report += (
                         f'Decrementing `end_dt` and retrying with,\n'
                         f'def_frame_duration: {def_frame_duration}\n'
-                        f'(new) request_end_dt: {request_end_dt}\n'
+                        f'(new) last_start_dt: {last_start_dt}\n'
                     )
                     log.warning(gap_report)
                     # skip writing to shm/tsdb and try the next
@@ -643,7 +657,7 @@ async def start_backfill(
 
                     f'fqme: {mkt.fqme}\n'
                     f'timeframe: {timeframe}\n'
-                    f'request_end_dt: {request_end_dt}\n'
+                    f'last_start_dt: {last_start_dt}\n'
                     f'bf_until: {backfill_until_dt}\n'
                 )
                 # UGH: what's a better way?
@@ -698,7 +712,7 @@ async def start_backfill(
                 # await tractor.pause()
 
             expected_dur: Interval = (
-                request_end_dt.subtract(
+                last_start_dt.subtract(
                     seconds=timeframe
                     # ^XXX, always "up to" the bar *before*
                 )
@@ -733,8 +747,7 @@ async def start_backfill(
                 log.warning(
                     f'{timeframe}s-series {reason} detected!\n'
                     f'fqme: {mkt.fqme}\n'
-                    f'request_end_dt: {request_end_dt}\n'
+                    f'last_start_dt: {last_start_dt}\n\n'
-                    f'\n'
                     f'recv interval: {recv_frame_dur}\n'
                     f'expected interval: {expected_dur}\n\n'
 
@@ -743,33 +756,26 @@ async def start_backfill(
                 )
                 # await tractor.pause()
 
-            # Stop retrieving history at backfill_until_dt. Keep
+            to_store: np.ndarray = diff_history(
-            # that timestamp for the storage merge, plus newer bars;
+                array,
-            # discard any older bars included in this provider frame.
+                prepend_until_dt=backfill_until_dt,
-            to_store: np.ndarray = array[
+            )
-                array['time'] >= backfill_until_dt.timestamp()
+            # The requested end bar is already the oldest published
-            ]
+            # sample. Keep it for storage merge, but do not
-            # Keep only bars older than SHM's earliest written bar.
+            # consume another physical SHM row for the overlap.
-            # A bar with the same timestamp already exists in SHM;
-            # writing it again would duplicate that bar.
-            #
-            # Example: SHM starts at 12:00. An empty response moves
-            # request_end_dt to 11:26:40, but SHM still starts at
-            # 12:00. If the next reply includes an 11:26:40 bar, we
-            # must keep it: asking for that time did not write it.
             to_push: np.ndarray = to_store[
-                to_store['time'] < oldest_shm_dt.timestamp()
+                to_store['time'] < last_start_dt.timestamp()
             ]
             ln: int = len(to_push)
             if ln:
                 log.info(
-                    f'{ln} bars for {next_start_dt} -> {request_end_dt}'
+                    f'{ln} bars for {next_start_dt} -> {last_start_dt}'
                 )
 
             else:
                 log.warning(
                     f'0 BARS TO PUSH after diff!?\n'
-                    f'{next_start_dt} -> {request_end_dt}'
+                    f'{next_start_dt} -> {last_start_dt}'
                     f'\n'
                     f'This might mean we rxed a gap frame which starts BEFORE,\n'
                     f'backfill_until_dt: {backfill_until_dt}\n'
@@ -806,7 +812,7 @@ async def start_backfill(
             ):
                 log.info(
                     f'Writing {len(to_store)} frame to storage:\n'
-                    f'{next_start_dt} -> {request_end_dt}'
+                    f'{next_start_dt} -> {last_start_dt}'
                 )
                 await storage.update_ohlcv(
                     mk_storage_key(mkt),
@@ -829,10 +835,7 @@ async def start_backfill(
 
                 # decrement next prepend point
                 next_prepend_index = next_prepend_index - ln
-                request_end_dt = next_start_dt
+                last_start_dt = next_start_dt
-                # The push succeeded: these bars now exist in SHM.
-                # Use their start to exclude duplicates next time.
-                oldest_shm_dt = next_start_dt
 
                 # Stop if we've hit buffer start
                 if next_prepend_index <= 0:
@@ -845,8 +848,7 @@ async def start_backfill(
             except ValueError as ve:
                 _ve = ve
                 log.error(
-                    f'Shm prepend OVERRUN on: {next_start_dt} -> '
+                    f'Shm prepend OVERRUN on: {next_start_dt} -> {last_start_dt}?'
-                    f'{request_end_dt}?'
                 )
 
                 if next_prepend_index < ln:
@@ -865,15 +867,12 @@ async def start_backfill(
                     update_start_on_prepend=update_start_on_prepend,
                 )
                 next_prepend_index = next_prepend_index - ln
-                request_end_dt = next_start_dt
+                last_start_dt = next_start_dt
-                # The push succeeded: these bars now exist in SHM.
-                # Use their start to exclude duplicates next time.
-                oldest_shm_dt = next_start_dt
                 shm_full = True
 
             log.info(
                 f'Shm pushed {ln} frame:\n'
-                f'{next_start_dt} -> {request_end_dt}'
+                f'{next_start_dt} -> {last_start_dt}'
             )
 
             await notify_backfill(

tests/test_history_backfill.py

@@ -706,160 +706,3 @@ def test_null_repair_falls_back_without_debug_pause(
             frame[field][-1],
         ]
     assert frame['volume'].tolist() == [5, 0, 0, 0, 6]
-
-
-@pytest.mark.parametrize('empty_reply', ['hmds_error', 'empty_bars'])
-def test_ib_empty_interval_preserves_older_bars_in_parquet(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-    empty_reply: str,
-) -> None:
-    '''
-    An empty IB interval must not skip a day of available history.
-
-    Run the real IB adapter, startup backfiller, SHM publication,
-    and NativeDB writes against a simulated IB bars endpoint. A
-    known empty interval separates recent bars from older trades.
-    The old HMDS handler jumped a day back, so those older trades
-    never reached SHM or parquet. Keep the legitimate empty interval
-    and assert every supplied trade survives a fresh storage load.
-    This covers data ingestion and persistence, not Qt rendering.
-
-    '''
-    from ib_async import RequestError
-    from pendulum import duration
-    from piker.brokers.ib import feed as ib_feed
-
-    epoch: int = 1790065963
-    requests: list[float|None] = []
-    finished = trio.Event()
-    mkt = SimpleNamespace(
-        fqme='mnq.cme.20261218.ib',
-        dst=SimpleNamespace(atype='future'),
-        src=SimpleNamespace(atype='fiat'),
-        get_fqme=lambda **kwargs: 'mnq.cme.20261218.ib',
-        get_bs_fqme=lambda **kwargs: 'mnq.cme.20261218',
-    )
-
-    def bars(offsets) -> np.ndarray:
-        frame = np.zeros(len(offsets), dtype=_bar_load_dtype)
-        frame['time'] = np.asarray(offsets) + epoch
-        for field in ('open', 'high', 'low', 'close'):
-            frame[field] = 30000 + np.asarray(offsets) / 4
-        frame['volume'] = 1
-        frame['count'] = 1
-        return frame
-
-    class Proxy:
-        async def maybe_get_head_time(self, **kwargs):
-            return from_timestamp(epoch - 864000)
-
-        async def bars(self, **kwargs):
-            end = kwargs['end_dt']
-            end_t = None if end is None else end.timestamp()
-            requests.append(end_t)
-            if end_t is None:
-                frame = bars([4001, 4002])
-            elif end_t == epoch + 4001:
-                if empty_reply == 'hmds_error':
-                    raise RequestError(
-                        1, 162, 'HMDS query returned no data',
-                    )
-                frame = bars([])
-            elif end_t == epoch + 2001:
-                frame = bars(list(range(1, 2002)))
-            else:
-                # Model the old adapter's day-skipped response.
-                # It falls before the stored endpoint and is dropped.
-                assert end_t == epoch + 4001 - 86400
-                frame = bars([-82400, -82399])
-            native = [
-                SimpleNamespace(date=from_timestamp(timestamp))
-                for timestamp in frame['time']
-            ]
-            return native, frame, duration(seconds=2000)
-
-    @asynccontextmanager
-    async def fake_data_client():
-        yield Proxy()
-
-    class Sampler:
-        async def send(self, msg) -> None:
-            pass
-
-    # Preserve the real null-repair pass; signal when startup ends.
-    from piker.tsp import _history
-    repair = _history.maybe_fill_null_segments
-
-    async def finish_repair(**kwargs) -> None:
-        await repair(**kwargs)
-        finished.set()
-
-    monkeypatch.setattr(
-        ib_feed, 'open_data_client', fake_data_client,
-    )
-    monkeypatch.setattr(
-        _history, 'maybe_fill_null_segments', finish_repair,
-    )
-    expected = bars(list(range(2002)) + [4001, 4002])
-
-    async def main() -> None:
-        storage = NativeStorageClient(tmp_path)
-        await storage.update_ohlcv(mkt.fqme, bars([0, 1]), 1)
-        key = f'test_ib_request_bounds_{uuid4().hex}'
-        with trio.fail_after(10):
-            async with tractor.open_root_actor(
-                name=key,
-                tpt_bind_addrs=[('127.0.0.1', 0)],
-            ):
-                shm, opened = maybe_open_shm_array(
-                    key=key,
-                    size=5000,
-                    dtype=np.dtype(_ohlc_dtype),
-                    append_start_index=4500,
-                )
-                assert opened
-                async with trio.open_nursery() as nursery:
-                    nursery.start_soon(partial(
-                        tsdb_backfill,
-                        mod=SimpleNamespace(
-                            name='ib',
-                            open_history_client=(
-                                ib_feed.open_history_client
-                            ),
-                        ),
-                        storemod=SimpleNamespace(
-                            ohlc_key_map=ohlc_key_map,
-                        ),
-                        storage=storage,
-                        mkt=mkt,
-                        shm=shm,
-                        timeframe=1,
-                        sampler_stream=Sampler(),
-                    ))
-                    await finished.wait()
-                    nursery.cancel_scope.cancel()
-                # Read actual parquet through a new client to avoid
-                # accidentally validating only NativeDB's cache.
-                fresh = NativeStorageClient(tmp_path)
-                loaded, _, _ = await fresh.load(mkt.fqme, 1)
-                for field in (
-                    'time', 'open', 'high', 'low', 'close', 'volume',
-                ):
-                    np.testing.assert_array_equal(
-                        loaded[field], expected[field],
-                        err_msg=f'parquet field: {field}',
-                    )
-                    np.testing.assert_array_equal(
-                        shm.array[field], expected[field],
-                        err_msg=f'SHM field: {field}',
-                    )
-                assert requests == [
-                    None, epoch + 4001, epoch + 2001,
-                ]
-                # Exactly the simulated closure remains. Do not
-                # manufacture bars to force timestamp continuity.
-                gaps = np.diff(loaded['time'])
-                assert gaps[gaps > 1].tolist() == [2000]
-
-    trio.run(main)

tests/test_ib_history.py

@@ -586,72 +586,3 @@ def test_completed_pacing_reset_restarts_timeout_wait(
 
     trio.run(main)
     assert reset_types == ['data', 'connection']
-
-
-@pytest.mark.parametrize('empty_reply', ['hmds_error', 'empty_bars'])
-@pytest.mark.parametrize(
-    'boundary', ['latest', 'after_head', 'head'],
-)
-def test_empty_history_preserves_request_boundary(
-    monkeypatch: pytest.MonkeyPatch,
-    empty_reply: str,
-    boundary: str,
-) -> None:
-    '''
-    Empty IB responses retain their original request metadata.
-
-    Exercise open_history_client() and get_bars() together. The
-    latest request must not fall back to yesterday; a dated request
-    must not issue another date or become historical exhaustion
-    unless it has actually reached the known first timestamp.
-
-    '''
-    from contextlib import asynccontextmanager
-    from piker.brokers import NoData
-
-    head = pdatetime(2026, 9, 18, tz='UTC')
-    end = {
-        'latest': None,
-        'after_head': head.add(days=4),
-        'head': head,
-    }[boundary]
-    requests: list = []
-
-    class Proxy:
-        async def maybe_get_head_time(self, **kwargs):
-            return head
-
-        async def bars(self, **kwargs):
-            requests.append(kwargs['end_dt'])
-            if empty_reply == 'hmds_error':
-                raise RequestError(
-                    1, 162, 'HMDS query returned no data',
-                )
-            return [], np.empty(0), duration(seconds=2000)
-
-    @asynccontextmanager
-    async def fake_data_client():
-        yield Proxy()
-
-    monkeypatch.setattr(
-        ib_feed, 'open_data_client', fake_data_client,
-    )
-    mkt = SimpleNamespace(
-        dst=SimpleNamespace(atype='future'),
-        get_bs_fqme=lambda **kwargs: 'mnq.cme.20261218',
-    )
-
-    async def main() -> None:
-        with trio.fail_after(1):
-            async with ib_feed.open_history_client(mkt) as pair:
-                get_hist, _ = pair
-                expected = (
-                    DataUnavailable if boundary == 'head' else NoData
-                )
-                with pytest.raises(expected) as exc:
-                    await get_hist(1, end_dt=end)
-                if expected is NoData:
-                    assert exc.value.info['end_dt'] == end
-                assert requests == [end]
-
-    trio.run(main)

tests/test_ib_method_proxy.py

@@ -5,7 +5,6 @@ IB asyncio method-proxy regressions.
 from types import SimpleNamespace
 from typing import Any
 
-from eventkit import Event
 import pytest
 import trio
 
@@ -64,33 +63,6 @@ class FakeChannel:
         await self._tx.aclose()
 
 
-def test_socket_disconnect_raises_from_proxy() -> None:
-    '''
-    Roaming reset the IB socket, but apiError was merely logged and
-    proxy callers could wait indefinitely. Emit the actual eventkit
-    apiError/disconnected sequence through Client.inline_errors().
-    The single-reader proxy must raise ConnectionError carrying the
-    reset reason, while the preceding error alone remains nonfatal.
-
-    '''
-    async def main() -> None:
-        chan = FakeChannel()
-        ib = SimpleNamespace(
-            errorEvent=Event('errorEvent'),
-            disconnectedEvent=Event('disconnectedEvent'),
-            client=SimpleNamespace(apiError=Event('apiError')),
-        )
-        client = ib_api.Client(ib=ib, config={})
-        client.inline_errors(chan._tx)
-        proxy = MethodProxy(chan, {}, asyncio_ns=client)
-        ib.client.apiError.emit('[Errno 104] Connection reset by peer')
-        ib.disconnectedEvent.emit()
-        with pytest.raises(ConnectionError, match='Errno 104'):
-            await relay_client_proxy_messages(chan, proxy)
-
-    trio.run(main)
-
-
 def test_method_proxy_routes_after_idle_status_events() -> None:
     '''
     Idle IB status traffic must not break first-symbol qualification.

tests/test_ib_trade_events.py

@@ -1,160 +0,0 @@
-'''
-Execution identity across IB fill and commission callbacks.
-
-'''
-import asyncio
-from types import SimpleNamespace
-
-from bidict import bidict
-from eventkit import Event
-from ib_async import (
-    CommissionReport,
-    Contract,
-    Execution,
-    Fill,
-    Position,
-)
-import pytest
-import trio
-
-from piker.brokers.ib import broker
-from piker.brokers.ib.ledger import api_trades_to_ledger_entries
-
-
-def make_fill(tid: str) -> Fill:
-    return Fill(
-        Contract(conId=815824267, secType='FUT', symbol='MNQ'),
-        Execution(
-            execId=tid,
-            acctNumber='DU_TEST',
-            time=1789586554.0,
-        ),
-        CommissionReport(),
-        1789586554.0,
-    )
-
-
-def test_execution_id_survives_empty_commission() -> None:
-    '''
-    Flattening a default CommissionReport erased Execution.execId,
-    causing a real fill to be skipped as an ID-less adjustment. Pass
-    a fill with a valid execution and empty commission through ledger
-    conversion and assert that its account and execution survive.
-
-    '''
-    entries = api_trades_to_ledger_entries(
-        bidict({'DU_TEST': 'ib.algopaper'}),
-        [make_fill('execution-1')],
-    )
-    assert entries['ib.algopaper']['execution-1']['execId'] == (
-        'execution-1'
-    )
-
-
-def test_callback_preserves_fill_and_disconnect() -> None:
-    '''
-    The asyncio callback dropped commissionReportEvent's exact Fill,
-    forcing the Trio side to guess by contract. Exercise the actual
-    eventkit relay and assert it preserves the supplied object. After
-    that callback, emit disconnectedEvent: the bridge must raise
-    rather than return normally and leave its Trio owner asleep.
-    One asyncio checkpoint lets recv_trade_updates install handlers
-    before the test producer emits either event.
-
-    '''
-    async def main():
-        ib = SimpleNamespace(**{
-            name: Event(name)
-            for name in (
-                'orderStatusEvent', 'commissionReportEvent',
-                'execDetailsEvent', 'positionEvent',
-                'disconnectedEvent',
-            )
-        })
-        emitted = []
-        chan = SimpleNamespace(
-            started_nowait=lambda _: None,
-            send_nowait=emitted.append,
-        )
-        client = SimpleNamespace(ib=ib, inline_errors=lambda _: None)
-        fill = make_fill('A')
-        report = CommissionReport(execId='A', commission=.61)
-
-        async def produce():
-            await asyncio.sleep(0)
-            ib.commissionReportEvent.emit(None, fill, report)
-            ib.disconnectedEvent.emit()
-
-        producer = asyncio.create_task(produce())
-        with pytest.raises(ConnectionError, match='socket'):
-            await broker.recv_trade_updates(chan, client)
-        await producer
-        assert emitted == [('commissionReportEvent', (fill, report))]
-
-    asyncio.run(main())
-
-
-@pytest.mark.parametrize('position_first', [True, False])
-def test_overlapping_execution_commissions(
-    monkeypatch: pytest.MonkeyPatch,
-    position_first: bool,
-) -> None:
-    '''
-    Two executions on one contract shared a single pending Fill.
-    A commission for execution A could therefore account execution B.
-    Queue both executions before either commission, delivering fees
-    in reverse order, and exercise positions before and after fees.
-    The deterministic channel order verifies every execution reaches
-    accounting once with its own fee, including a zero commission.
-
-    '''
-    fills = [make_fill('A'), make_fill('B')]
-    pos = Position('DU_TEST', fills[0].contract, 2, 58000)
-    recorded = []
-
-    async def capture(*args, ibpos, fill=None):
-        if fill is not None:
-            recorded.append((
-                fill.execution.execId,
-                fill.commissionReport.execId,
-                fill.commissionReport.commission,
-                ibpos.account,
-            ))
-
-    monkeypatch.setattr(broker, 'emit_pp_update', capture)
-    monkeypatch.setattr(
-        broker, 'pack_position', lambda *args: ('815824267', {})
-    )
-
-    async def main():
-        send, recv = trio.open_memory_channel(10)
-        events = [
-            ('execDetailsEvent', (
-                SimpleNamespace(contract=fill.contract), fill,
-            ))
-            for fill in fills
-        ]
-        events.extend([
-            ('commissionReportEvent', (
-                fills[1], CommissionReport(execId='B', commission=0),
-            )),
-            ('commissionReportEvent', (
-                fills[0], CommissionReport(execId='A', commission=.61),
-            )),
-        ])
-        events.insert(0 if position_first else len(events), (
-            'positionEvent', pos,
-        ))
-        async with send, recv:
-            for event in events:
-                send.send_nowait(event)
-            await send.aclose()
-            await broker.deliver_trade_events(
-                recv, None, {}, {}, {}, {}, None,
-            )
-
-    trio.run(main)
-    assert recorded == [
-        ('B', 'B', 0, 'DU_TEST'),
-        ('A', 'A', .61, 'DU_TEST'),
-    ]