Preserve IB history request bounds on no data

Complete HMDS no-data replies without silently retrying a day
older. Let `open_history_client()` classify the original request
and leave frame stepping to `start_backfill()`.

Distinguish `request_end_dt` from `oldest_shm_dt` so the first
frame after an empty interval retains its unwritten end bar.
Inline the storage time cutoff and drop the unused two-direction
`diff_history()` helper. Explain both dates and filtering rules.

Add request-boundary tests and real SHM/NativeDB regressions:
all 2,004 supplied bars survive, while the closure stays empty.

Prompt-IO: ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.md

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

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

@@ -0,0 +1,64 @@
+---
+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

@@ -0,0 +1,87 @@
+---
+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

@@ -0,0 +1,37 @@
+---
+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

@@ -0,0 +1,45 @@
+---
+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

@@ -0,0 +1,61 @@
+---
+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

@@ -0,0 +1,25 @@
+---
+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,8 +1276,11 @@ 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,
@@ -1290,6 +1293,25 @@ 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 = '',
@@ -1851,6 +1873,9 @@ 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,7 +338,8 @@ async def recv_trade_updates(
 
             case 'commissionReportEvent':
                 assert report
-                emit: CommissionReport = report
+                assert fill is not None
+                emit: tuple[Fill, CommissionReport] = (fill, report)
 
             case 'execDetailsEvent':
                 # execution details event
@@ -390,6 +391,7 @@ 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(
@@ -1189,15 +1191,10 @@ async def deliver_trade_events(
     Format and relay all trade events for a given client to emsd.
 
     '''
-    # task local msg dialog tracking
+    # Positions belong to an account/contract; commissions belong to
-    clears: dict[
+    # executions. Never recover a Fill from just its Contract.
-        Contract,
+    positions: dict[tuple[str, int], IbPosition] = {}
-        list[
+    pending_fills: dict[tuple[str, int], dict[str, Fill]] = {}
-            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
@@ -1296,91 +1293,28 @@ async def deliver_trade_events(
             # translating them to ``Status`` msgs) if we can
             # show the equivalent status events are no more latent.
             case 'execDetailsEvent':
-                # unpack attrs pep-0526 style.
+                # Accounting waits for commissionReportEvent, whose
-                trade: Trade
+                # callback supplies the exact Fill, including zero
-                fill: Fill
+                # commissions. Order-status handling relays fills
-                trade, fill = item
+                # independently above.
-                con: Contract = trade.contract
+                continue
-                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
-                cr: CommissionReport = item
                 execid: str = cr.execId
-
+                if execid != fill.execution.execId:
-                # only fire a pp msg update if,
+                    raise ValueError(
-                # - we haven't already
+                        f'IB commission/fill execution mismatch:\n'
-                # - the fill event has already arrived
+                        f'{cr!r}\n{fill!r}'
-                #   but it didn't yet have a commision report
-                #   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'
                     )
+                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(
                     ems_stream,
@@ -1388,13 +1322,9 @@ async def deliver_trade_events(
                     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
@@ -1407,9 +1337,17 @@ async def deliver_trade_events(
                 bs_mktid, ppmsg = pack_position(pos, accounts_def)
                 log.info(f'New IB position msg: {ppmsg}')
 
-                _, fill = clears.setdefault(
+                key = (pos.account, con.conId)
-                    con,
+                positions[key] = pos
-                    [pos, None],
+                for fill in pending_fills.pop(key, {}).values():
+                    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
@@ -1424,6 +1362,9 @@ 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,7 +38,6 @@ 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,
@@ -221,7 +220,11 @@ async def open_history_client(
                     f'mean: {mean}'
                 )
 
-            # could be trying to retreive bars over weekend
+            # get_bars() uses None internally for an empty reply.
+            # 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} !?!?"
@@ -456,10 +459,6 @@ 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,
@@ -479,7 +478,6 @@ 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
@@ -496,8 +494,8 @@ async def get_bars(
 
     async def query():
 
-        nonlocal result, data_cs, end_dt
+        nonlocal result, data_cs
-        nonlocal nodatas_count, cancelled_count, failed_resets
+        nonlocal cancelled_count, failed_resets
         nonlocal pacing_reset_pending, pacing_reset_completed
 
         while True:
@@ -625,27 +623,21 @@ 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
-                    if nodatas_count >= max_nodatas:
+                    # internal to the IB adapter, not its backfill
-                        raise DataUnavailable(
+                    # protocol. open_history_client() converts it
-                            f'Presuming {fqme} has no further history '
+                    # to NoData (or DataUnavailable at history start).
-                            f'after {max_nodatas} tries..'
+                    # Wake the waiting get_bars() task before exiting
-                        )
+                    # so it does not mistake this reply for a timeout.
-
+                    result = None
-                    nodatas_count += 1
+                    failed_resets = 0
-                    continue
+                    result_ready.set()
+                    if data_cs:
+                        data_cs.cancel()
+                    return None
 
                 elif (
                     'API historical data query cancelled'

piker/brokers/ib/ledger.py

@@ -429,9 +429,16 @@ def api_trades_to_ledger_entries(
         for attr_name, val in fdict.items():
             match attr_name:
                 # value is a `@dataclass` subtype
-                case 'contract' | 'execution' | 'commissionReport':
+                case 'contract' | 'execution':
                     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,29 +145,6 @@ _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.
@@ -586,16 +563,24 @@ 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?
-        last_start_dt: datetime = backfill_from_dt
+        # start_backfill() asks get_hist() for bars ending at
+        # 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 last_start_dt > backfill_until_dt:
+        while request_end_dt > backfill_until_dt:
             log.info(
                 f'Requesting {timeframe}s frame:\n'
                 f'backfill_until_dt: {backfill_until_dt}\n'
-                f'last_start_dt: {last_start_dt}\n'
+                f'request_end_dt: {request_end_dt}\n'
             )
             try:
                 (
@@ -604,22 +589,23 @@ async def start_backfill(
                     next_end_dt,
                 ) = await get_hist(
                     timeframe,
-                    end_dt=(end_dt_param := last_start_dt),
+                    end_dt=(end_dt_param := request_end_dt),
                 )
             except NoData as nodata:
                 _nodata = nodata
-                orig_last_start_dt: datetime = last_start_dt
+                previous_request_end_dt: datetime = request_end_dt
                 gap_report: str = (
-                    f'EMPTY FRAME for `end_dt: {last_start_dt}`?\n'
+                    f'EMPTY FRAME for `end_dt: {request_end_dt}`?\n'
                     f'{mod.name} -> tf@fqme: {timeframe}@{mkt.fqme}\n'
-                    f'last_start_dt: {orig_last_start_dt}\n\n'
+                    f'request_end_dt: {previous_request_end_dt}\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 `last_start_dt`
+                #    value `request_end_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
@@ -630,13 +616,13 @@ async def start_backfill(
                     # as maybe indicated by the backend to see if we
                     # can get older data before this possible
                     # "history gap".
-                    last_start_dt: datetime = last_start_dt.subtract(
+                    request_end_dt = request_end_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) last_start_dt: {last_start_dt}\n'
+                        f'(new) request_end_dt: {request_end_dt}\n'
                     )
                     log.warning(gap_report)
                     # skip writing to shm/tsdb and try the next
@@ -657,7 +643,7 @@ async def start_backfill(
 
                     f'fqme: {mkt.fqme}\n'
                     f'timeframe: {timeframe}\n'
-                    f'last_start_dt: {last_start_dt}\n'
+                    f'request_end_dt: {request_end_dt}\n'
                     f'bf_until: {backfill_until_dt}\n'
                 )
                 # UGH: what's a better way?
@@ -712,7 +698,7 @@ async def start_backfill(
                 # await tractor.pause()
 
             expected_dur: Interval = (
-                last_start_dt.subtract(
+                request_end_dt.subtract(
                     seconds=timeframe
                     # ^XXX, always "up to" the bar *before*
                 )
@@ -747,7 +733,8 @@ async def start_backfill(
                 log.warning(
                     f'{timeframe}s-series {reason} detected!\n'
                     f'fqme: {mkt.fqme}\n'
-                    f'last_start_dt: {last_start_dt}\n\n'
+                    f'request_end_dt: {request_end_dt}\n'
+                    f'\n'
                     f'recv interval: {recv_frame_dur}\n'
                     f'expected interval: {expected_dur}\n\n'
 
@@ -756,26 +743,33 @@ async def start_backfill(
                 )
                 # await tractor.pause()
 
-            to_store: np.ndarray = diff_history(
+            # Stop retrieving history at backfill_until_dt. Keep
-                array,
+            # that timestamp for the storage merge, plus newer bars;
-                prepend_until_dt=backfill_until_dt,
+            # discard any older bars included in this provider frame.
-            )
+            to_store: np.ndarray = array[
-            # The requested end bar is already the oldest published
+                array['time'] >= backfill_until_dt.timestamp()
-            # sample. Keep it for storage merge, but do not
+            ]
-            # consume another physical SHM row for the overlap.
+            # Keep only bars older than SHM's earliest written bar.
+            # 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'] < last_start_dt.timestamp()
+                to_store['time'] < oldest_shm_dt.timestamp()
             ]
             ln: int = len(to_push)
             if ln:
                 log.info(
-                    f'{ln} bars for {next_start_dt} -> {last_start_dt}'
+                    f'{ln} bars for {next_start_dt} -> {request_end_dt}'
                 )
 
             else:
                 log.warning(
                     f'0 BARS TO PUSH after diff!?\n'
-                    f'{next_start_dt} -> {last_start_dt}'
+                    f'{next_start_dt} -> {request_end_dt}'
                     f'\n'
                     f'This might mean we rxed a gap frame which starts BEFORE,\n'
                     f'backfill_until_dt: {backfill_until_dt}\n'
@@ -812,7 +806,7 @@ async def start_backfill(
             ):
                 log.info(
                     f'Writing {len(to_store)} frame to storage:\n'
-                    f'{next_start_dt} -> {last_start_dt}'
+                    f'{next_start_dt} -> {request_end_dt}'
                 )
                 await storage.update_ohlcv(
                     mk_storage_key(mkt),
@@ -835,7 +829,10 @@ async def start_backfill(
 
                 # decrement next prepend point
                 next_prepend_index = next_prepend_index - ln
-                last_start_dt = next_start_dt
+                request_end_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:
@@ -848,7 +845,8 @@ async def start_backfill(
             except ValueError as ve:
                 _ve = ve
                 log.error(
-                    f'Shm prepend OVERRUN on: {next_start_dt} -> {last_start_dt}?'
+                    f'Shm prepend OVERRUN on: {next_start_dt} -> '
+                    f'{request_end_dt}?'
                 )
 
                 if next_prepend_index < ln:
@@ -867,12 +865,15 @@ async def start_backfill(
                     update_start_on_prepend=update_start_on_prepend,
                 )
                 next_prepend_index = next_prepend_index - ln
-                last_start_dt = next_start_dt
+                request_end_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} -> {last_start_dt}'
+                f'{next_start_dt} -> {request_end_dt}'
             )
 
             await notify_backfill(

tests/test_history_backfill.py

@@ -706,3 +706,160 @@ 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,3 +586,72 @@ 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,6 +5,7 @@ IB asyncio method-proxy regressions.
 from types import SimpleNamespace
 from typing import Any
 
+from eventkit import Event
 import pytest
 import trio
 
@@ -63,6 +64,33 @@ 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

@@ -0,0 +1,160 @@
+'''
+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'),
+    ]