From 775eecc6acf6021c8c8d1b66601608e3066cbca7 Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 11:13:44 -0300 Subject: [PATCH 1/9] feat(engine): give each flatten close a seq above every earlier one A second flatten on the same store reused seq 1. The manager drops a signal id it has seen, so the close never went out and the run exited 1. The counter now starts from the saga high-water fold, which moved to Cache so StrategyHost and Flatten share it. Co-Authored-By: Claude Opus 5.5 --- README.md | 3 --- docs/module-maps/safe-exit.md | 2 ++ src/tickwright/engine/cache.py | 21 ++++++++++++++++++++- src/tickwright/engine/exit_run.py | 10 +++++++--- src/tickwright/engine/strategy_host.py | 20 +------------------- tests/engine/test_flatten_e2e.py | 17 +++++++++++++++++ 6 files changed, 47 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index eb8be738..a52402f0 100644 --- a/README.md +++ b/README.md @@ -191,9 +191,6 @@ A position the feed never priced is not closed. The run exits `1`, and `exit.fin symbol and its size, for example `left=BTC=0.02`. Under `GUARD=noop` there is no kill switch to trip, so the run warns with `exit.unguarded`. -Flatten closes each symbol only once per store today. If a later run opens that symbol again, the -next flatten sends no close for it. It exits `1` and names the symbol in `left`. #474 fixes this. - ## What it does today - Venues: [Hyperliquid](https://hyperliquid.gitbook.io/hyperliquid-docs) (mainnet and testnet), diff --git a/docs/module-maps/safe-exit.md b/docs/module-maps/safe-exit.md index 7be3453f..5387df39 100644 --- a/docs/module-maps/safe-exit.md +++ b/docs/module-maps/safe-exit.md @@ -162,6 +162,8 @@ context. Starts the feed through the `FeedHandle`. - **(Deferred by #472:** every close uses seq 1. The manager drops a signal id it has seen (ADR-0006). So a second flatten on the same store sends no close for a symbol an earlier flatten closed. It exits 1 and names the symbol in `left`. #474 builds the counter.**)** +- **(Built by #474:** the fold moved to `Cache.seq_high_water()`. `StrategyHost` and `Flatten` + both read it. Each close takes the next seq.**)** **The per-symbol loop (private):** diff --git a/src/tickwright/engine/cache.py b/src/tickwright/engine/cache.py index c065d4f9..b5b2e4dd 100644 --- a/src/tickwright/engine/cache.py +++ b/src/tickwright/engine/cache.py @@ -7,7 +7,7 @@ checkpoint path. """ -from tickwright.domain import Order, Store +from tickwright.domain import Order, SignalId, Store class Cache: @@ -70,6 +70,25 @@ def all_orders(self) -> list[Order]: """ return list(self._orders.values()) + def seq_high_water(self) -> dict[str, int]: + """Max consumed seq per strategy across every saga (ADR-0016). + + Every place consumed its ``signal_id``'s seq, and every requested + cancel consumed its own (ADR-0026 — omitting cancels would let a + restart reuse a consumed id). One per-strategy counter across all + symbols, so this is a single max over the strategy's records. The + strategies and the operator's exit jobs both start their counters + here, so neither can reuse an id after a restart. + """ + high_water: dict[str, int] = {} + for order in self._orders.values(): + seqs = [SignalId.parse(order.signal_id).seq] + if order.cancel_signal_id is not None: + seqs.append(SignalId.parse(order.cancel_signal_id).seq) + current = high_water.get(order.strategy_id, 0) + high_water[order.strategy_id] = max(current, *seqs) + return high_water + def last_event_ts(self, cloid: str) -> int | None: """The ``ts_ns`` of ``cloid``'s most recent checkpoint *this session*, or ``None`` if it has not been written since the last ``rebuild`` — the diff --git a/src/tickwright/engine/exit_run.py b/src/tickwright/engine/exit_run.py index 8ab35099..39923d37 100644 --- a/src/tickwright/engine/exit_run.py +++ b/src/tickwright/engine/exit_run.py @@ -138,15 +138,19 @@ async def note_price(tick: MarketTick) -> None: # Only a replay ends, and it plays the whole file first. So each close # fills at the last row (ADR-0059). The CLI refuses a live feed until #475. await context.feed.ended() + # From the sagas, as a strategy's is. An earlier flatten's seq is + # consumed, and the manager drops a signal id it has seen (ADR-0006). + seq = context.checkpointer.cache.seq_high_water().get(OPERATOR, 0) for symbol, size in positions.items(): # No venue can send a market close without a price. The symbol # stays open, and the final read reports its size. if symbol in priced: - await _await_verdict(await _close(symbol, size, context), context) + seq += 1 + await _await_verdict(await _close(symbol, size, seq, context), context) return await _final_read(context) -async def _close(symbol: str, size: Decimal, context: ExitContext) -> str: +async def _close(symbol: str, size: Decimal, seq: int, context: ExitContext) -> str: """Send one reduce-only market IOC order for the whole of ``size`` (ADR-0054). Returns the cloid of the order's saga. @@ -157,7 +161,7 @@ async def _close(symbol: str, size: Decimal, context: ExitContext) -> str: ts_init=now, strategy_id=OPERATOR, symbol=symbol, - seq=1, + seq=seq, side=Side.SELL if size > 0 else Side.BUY, quantity=abs(size), order_type=OrderType.MARKET, diff --git a/src/tickwright/engine/strategy_host.py b/src/tickwright/engine/strategy_host.py index fecaeb32..bf6c177a 100644 --- a/src/tickwright/engine/strategy_host.py +++ b/src/tickwright/engine/strategy_host.py @@ -20,7 +20,6 @@ InvariantViolation, MarketTick, OrderEvent, - SignalId, Store, Strategy, SymbolOwnership, @@ -119,7 +118,7 @@ def start(self) -> None: ``Cache`` the runner rebuilt from that store, so the boot deserializes the history once (issue #233). """ - high_water = self._seq_high_water() + high_water = self._cache.seq_high_water() for strategy in self._strategies.values(): self._restore(strategy) # Probe once before any tick: a strategy the host cannot persist @@ -129,23 +128,6 @@ def start(self) -> None: self._subscribe(strategy) self._started = True - def _seq_high_water(self) -> dict[str, int]: - """Max consumed seq per strategy across every checkpointed saga. - - Every place consumed its ``signal_id``'s seq, and every requested - cancel consumed its own (ADR-0026 — omitting cancels would let a - restart reuse a consumed id). One per-strategy counter across all - symbols, so this is a single max over the strategy's records. - """ - high_water: dict[str, int] = {} - for order in self._cache.all_orders(): - seqs = [SignalId.parse(order.signal_id).seq] - if order.cancel_signal_id is not None: - seqs.append(SignalId.parse(order.cancel_signal_id).seq) - current = high_water.get(order.strategy_id, 0) - high_water[order.strategy_id] = max(current, *seqs) - return high_water - def _restore(self, strategy: Strategy) -> None: """Feed the persisted snapshot to ``strategy.restore()``, if one exists. diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index 072f73cd..89ad980f 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -417,6 +417,23 @@ def test_a_lost_close_is_reconciled_to_a_verdict_and_the_run_stops_with_the_size assert close.state is OrderState.FAILED +def test_a_second_flatten_closes_with_a_seq_above_the_first_ones_close(tmp_path: Path) -> None: + # The manager drops a signal id it has seen (ADR-0006). So a second flatten that + # reused seq 1 would send nothing, and the size would stay open. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + assert _flatten(tmp_path, link=_DropsEveryPlace)[0] == 1 + + exit_code, logs = _flatten(tmp_path) + + assert exit_code == 0 + assert logs[-1]["outcome"] == "done" + assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} + with SQLiteStore(tmp_path / "saga.db") as store: + close = store.get_order(derive_cloid(f"{OPERATOR}:BTC:2")) + assert close is not None + assert close.state is OrderState.FILLED + + class _LostAndUnreadable(_DropsEveryPlace): """The send is lost and every order read fails, so no pass can ever decide it.""" From e266ab61efac9778503131a7c27d8070a755b0ff Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 11:17:23 -0300 Subject: [PATCH 2/9] test(engine): prove a flatten resumes the close a crash left unbooked A flatten dies after its close reaches the venue and before the fill is booked. The next flatten's boot books that fill by the stored split. Then it closes the other symbol with a higher seq and exits 0. Co-Authored-By: Claude Opus 5.5 --- tests/engine/test_flatten_e2e.py | 126 ++++++++++++++++++++++++++++++- 1 file changed, 123 insertions(+), 3 deletions(-) diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index 89ad980f..b2ee975a 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -92,8 +92,8 @@ def _paper(bus: InMemoryBus, clock: ManualClock, store: SQLiteStore) -> PaperExc ) -def _hold(path: Path, sizes: dict[str, str]) -> None: - """Open each signed size as the strategy ``trivial``, at the symbol's first row price. +def _hold(path: Path, sizes: dict[str, str], *, strategy_id: str = "trivial") -> None: + """Open each signed size as ``strategy_id``, at the symbol's first row price. Seeded through the engine's own order path on a separate bus, as an earlier run would. """ @@ -114,7 +114,7 @@ async def go() -> None: PlaceSignal( ts_event=0, ts_init=0, - strategy_id="trivial", + strategy_id=strategy_id, symbol=symbol, seq=seq, side=Side.BUY if Decimal(size) > 0 else Side.SELL, @@ -459,3 +459,123 @@ def test_a_failed_reconcile_pass_ends_the_wait_and_the_final_read_decides( assert exit_code == 1 assert logs[-1]["outcome"] == "stopped" assert logs[-1]["left"] == "BTC=0.02" + + +class _Venue: + """Paper on a bus of its own, so it outlives the engine run it serves. + + The real venue keeps an order a crashed engine never booked. Paper keeps its + book in memory only (#484), so the test holds one instance across both runs. + """ + + def __init__(self, path: Path, clock: ManualClock) -> None: + self.bus = InMemoryBus() + self.store = SQLiteStore(path) + self._engine_bus: InMemoryBus | None = None + self.paper = PaperExchange( + bus=self.bus, + clock=clock, + fill_model=ImmediateFillModel(), + genesis_collateral=GENESIS, + instrument_specs=_SPECS, + account_net=lambda: paper_store_reads(self.store).account_net(), + applied_fills=lambda cloid: paper_store_reads(self.store).applied_fills(cloid), + ) + self.bus.subscribe(ExecutionReport, self._report) + + def serve(self, engine_bus: InMemoryBus) -> None: + """Link the venue to a new engine run: its ticks in, the venue's reports out.""" + self._engine_bus = engine_bus + + async def tick(tick: MarketTick) -> None: + await self.bus.publish(tick) + + engine_bus.subscribe(MarketTick, tick) + + def cut(self) -> None: + """The engine dies. Nothing the venue says from here reaches it.""" + self._engine_bus = None + + async def _report(self, report: ExecutionReport) -> None: + if self._engine_bus is not None: + await self._engine_bus.publish(report) + + +class _CrashesAfterTheSend(VenueLink): + """The close reaches the venue, and the engine dies before it hears the fill.""" + + def __init__(self, venue: Exchange, crash: Callable[[], None]) -> None: + super().__init__(venue) + self._crash = crash + self.crashed = asyncio.Event() + + async def place(self, order: PlaceOrder) -> None: + self._crash() + await super().place(order) + self.crashed.set() + await asyncio.Event().wait() + + +def _engine(venue: _Venue, exchange: Exchange, tmp_path: Path) -> Engine: + """One engine run of flatten, served by ``venue`` through ``exchange``. + + Each run replays the file from its first row, so each gets its own clock. + """ + bus = InMemoryBus() + clock = ManualClock() + venue.serve(bus) + return Engine( + bus=bus, + clock=clock, + store=venue.store, + exchange=exchange, + feed=ReplayFeed(path=_ticks(tmp_path / "ticks.jsonl", _ROWS), bus=bus, clock=clock), + guard=RealGuard(specs=_SPECS, store=venue.store, clock=clock), + exit_job=Flatten(), + ) + + +def test_a_flatten_that_crashed_after_its_send_is_booked_by_the_split_and_the_rest_closed( + tmp_path: Path, +) -> None: + # Two partitions share BTC, so the boot books the resumed fill by its stored + # split (ADR-0054). The other symbol is still open, and the next run closes it. + _hold(tmp_path / "saga.db", {"BTC": "0.02", "ETH": "0.5"}) + _hold(tmp_path / "saga.db", {"BTC": "0.01"}, strategy_id="other") + venue = _Venue(tmp_path / "saga.db", ManualClock()) + link = _CrashesAfterTheSend(venue.paper, venue.cut) + + async def first_life() -> None: + run = asyncio.create_task(_engine(venue, link, tmp_path).run()) + await asyncio.wait_for(link.crashed.wait(), timeout=5) + run.cancel() + await asyncio.gather(run, return_exceptions=True) + + asyncio.run(first_life()) + venue.store.close() + venue.store = SQLiteStore(tmp_path / "saga.db") + + async def second_life() -> int: + return await asyncio.wait_for(_engine(venue, venue.paper, tmp_path).run(), 5) + + with capture_events() as logs: + exit_code = asyncio.run(second_life()) + venue.store.close() + + assert exit_code == 0 + assert logs[-1]["outcome"] == "done" + # Opened at 42000 and 3000. Closed at the last rows, 44000 and 3200 (ADR-0059). + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("40")), + ("other", "BTC"): (Decimal("0"), Decimal("20")), + ("trivial", "ETH"): (Decimal("0"), Decimal("100")), + } + with SQLiteStore(tmp_path / "saga.db") as store: + closes = [order for order in store.all_orders() if order.strategy_id == OPERATOR] + # One close per symbol, so no size was closed twice. The resumed run's seq is + # above the crashed run's. + assert sorted((order.symbol, order.signal_id) for order in closes) in ( + [("BTC", f"{OPERATOR}:BTC:1"), ("ETH", f"{OPERATOR}:ETH:2")], + [("BTC", f"{OPERATOR}:BTC:2"), ("ETH", f"{OPERATOR}:ETH:1")], + ) + assert all(order.state is OrderState.FILLED for order in closes) From 6aa675f6638ea5121f95bb92e3c6b52d1cee75b0 Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 11:20:36 -0300 Subject: [PATCH 3/9] feat(engine): wait for a close the boot left open before flatten closes again A close the venue acked and then dropped can still be open after a crash. It may have filled, so a second close beside it could close the size twice. Flatten now waits for its verdict and reads the size again. If the wait ends with the close still open, it sends nothing for that symbol. Co-Authored-By: Claude Opus 5.5 --- docs/module-maps/safe-exit.md | 4 ++ src/tickwright/engine/exit_run.py | 33 +++++++++++++- tests/engine/test_flatten_e2e.py | 76 +++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+), 2 deletions(-) diff --git a/docs/module-maps/safe-exit.md b/docs/module-maps/safe-exit.md index 5387df39..42c0b4d6 100644 --- a/docs/module-maps/safe-exit.md +++ b/docs/module-maps/safe-exit.md @@ -170,6 +170,10 @@ context. Starts the feed through the `FeedHandle`. - Before the first attempt, wait for the verdict on any open `__operator__` saga for the symbol that the boot left open. It waits the same way as after an attempt, below. A symbol never has two attempts open. +- **(Built by #474, before the loops of #473:** the wait runs per symbol, after the feed has + played. Its passes move the replay clock, so an earlier wait would fault the replay. The size is + read again after the wait, because the old close may have filled. A wait that ends with the + saga still open sends no close for that symbol. The final read reports its size.**)** - Place only when the partition sizes sum to the venue size from `fetch_positions()` (ADR-0054). If they differ, call `bus.drain()` first, so every fill already published is applied. If they still differ and the venue has account truth, run `LedgerReconciliation.reconcile_account()`. diff --git a/src/tickwright/engine/exit_run.py b/src/tickwright/engine/exit_run.py index 39923d37..73faa6fe 100644 --- a/src/tickwright/engine/exit_run.py +++ b/src/tickwright/engine/exit_run.py @@ -144,12 +144,41 @@ async def note_price(tick: MarketTick) -> None: for symbol, size in positions.items(): # No venue can send a market close without a price. The symbol # stays open, and the final read reports its size. - if symbol in priced: + if symbol not in priced: + continue + resumed = await _resume_closes(symbol, size, context) + if isinstance(resumed, VenueReadFailure): + return ExitOutcome(status=ExitStatus.STOPPED, left=describe_failure(resumed)) + if resumed: seq += 1 - await _await_verdict(await _close(symbol, size, seq, context), context) + await _await_verdict(await _close(symbol, resumed, seq, context), context) return await _final_read(context) +async def _resume_closes( + symbol: str, size: Decimal, context: ExitContext +) -> Decimal | VenueReadFailure: + """Wait for any close the boot left open in ``symbol``, then say what is left to close. + + That close could still fill, so a second one sent beside it could close the + size twice. A symbol never has two closes open (ADR-0056). So a wait that ends + with one still open leaves nothing to close. The final read reports the size, + and the next boot resumes the saga. The wait runs after the feed has played, + because its passes move the replay clock. + """ + left_open = context.checkpointer.cache.open_orders(strategy_id=OPERATOR, symbol=symbol) + if not left_open: + return size + for order in left_open: + await _await_verdict(order.cloid, context) + if not all(_decided(order.cloid, context) for order in left_open): + return Decimal(0) + positions = await context.account.fetch_positions() + if isinstance(positions, VenueReadFailure): + return positions + return positions.get(symbol, Decimal(0)) + + async def _close(symbol: str, size: Decimal, seq: int, context: ExitContext) -> str: """Send one reduce-only market IOC order for the whole of ``size`` (ADR-0054). diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index b2ee975a..5d5bbcc8 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -461,6 +461,82 @@ def test_a_failed_reconcile_pass_ends_the_wait_and_the_final_read_decides( assert logs[-1]["left"] == "BTC=0.02" +def _left_open(path: Path, symbol: str) -> str: + """Leave an ``__operator__`` close ``LIVE`` in the store, one paper never lists. + + That is a close the venue acked and then dropped (#242) before a crash. The + boot can only arm its ghost window, so the saga is still open when flatten + starts. Returns its cloid. + """ + cloid = derive_cloid(f"{OPERATOR}:{symbol}:1") + clock = ManualClock() + with SQLiteStore(path) as store: + paper = _paper(InMemoryBus(), clock, store) + checks = Checkpointer(spec=paper.account_spec(), store=store, clock=clock, specs=_SPECS) + checks.recover() + checks.checkpoint( + Order( + cloid=cloid, + strategy_id=OPERATOR, + signal_id=f"{OPERATOR}:{symbol}:1", + symbol=symbol, + side=Side.SELL, + quantity=Decimal("0.02"), + order_type=OrderType.MARKET, + state=OrderState.LIVE, + created_ts_ns=0, + ) + ) + return cloid + + +def test_flatten_waits_for_the_verdict_on_a_close_the_boot_left_open_before_it_closes_again( + tmp_path: Path, +) -> None: + # The dropped close could still have filled. A second close sent beside it + # could close the size twice, so a symbol never has two closes open (ADR-0056). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + left_open = _left_open(tmp_path / "saga.db", "BTC") + + exit_code, logs = _flatten(tmp_path) + + assert exit_code == 0 + events = [(log["event"], log.get("cloid"), log.get("signal_id")) for log in logs] + verdict = events.index(("ghost.reconciled", left_open, None)) + placed = events.index(("order.placed", None, f"{OPERATOR}:BTC:2")) + assert verdict < placed + assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} + + +class _ReadsOnlyAtBoot(VenueLink): + """The boot reads every order. After that, no order read reaches the venue.""" + + def __init__(self, venue: Exchange) -> None: + super().__init__(venue) + self.reads = 0 + + async def fetch_order(self, ref: OrderRef) -> VenueOrderView | VenueReadFailure: + self.reads += 1 + if self.reads == 1: + return await super().fetch_order(ref) + return VenueReadFailure.SEND_FAILED + + +def test_a_close_left_open_that_no_pass_can_decide_gets_no_second_close_beside_it( + tmp_path: Path, +) -> None: + # The wait ends on a failed pass (ADR-0060), with the old close still open. + # The run stops with the size left, and the next boot resumes the saga. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + _left_open(tmp_path / "saga.db", "BTC") + + exit_code, logs = _flatten(tmp_path, link=_ReadsOnlyAtBoot) + + assert exit_code == 1 + assert logs[-1]["left"] == "BTC=0.02" + assert not [log for log in logs if log.get("signal_id") == f"{OPERATOR}:BTC:2"] + + class _Venue: """Paper on a bus of its own, so it outlives the engine run it serves. From d543e3a232516392bcf19355760ad0ba85c60f84 Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 11:25:27 -0300 Subject: [PATCH 4/9] test(engine): prove cancel-all books a crashed flatten's close at boot The resume belongs to the boot, not to the flatten job. So a cancel-all run after a crashed flatten books the close's fill by its split too. The crash setup moved into helpers both tests share. Co-Authored-By: Claude Opus 5.5 --- tests/engine/test_flatten_e2e.py | 66 +++++++++++++++++++++++++------- 1 file changed, 52 insertions(+), 14 deletions(-) diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index 5d5bbcc8..a3923d7d 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -47,7 +47,7 @@ ) from tickwright.engine.checkpoint import Checkpointer from tickwright.engine.execution import ExecutionManager -from tickwright.engine.exit_run import Flatten +from tickwright.engine.exit_run import ExitJob, Flatten, OperatorCancelAll from tickwright.engine.guard import NoopGuard, RealGuard from tickwright.engine.runner import Engine from tickwright.observability.testing import capture_events @@ -592,8 +592,10 @@ async def place(self, order: PlaceOrder) -> None: await asyncio.Event().wait() -def _engine(venue: _Venue, exchange: Exchange, tmp_path: Path) -> Engine: - """One engine run of flatten, served by ``venue`` through ``exchange``. +def _engine( + venue: _Venue, exchange: Exchange, tmp_path: Path, job: ExitJob | None = None +) -> Engine: + """One engine run of ``job``, flatten by default, served by ``venue`` through ``exchange``. Each run replays the file from its first row, so each gets its own clock. """ @@ -607,17 +609,15 @@ def _engine(venue: _Venue, exchange: Exchange, tmp_path: Path) -> Engine: exchange=exchange, feed=ReplayFeed(path=_ticks(tmp_path / "ticks.jsonl", _ROWS), bus=bus, clock=clock), guard=RealGuard(specs=_SPECS, store=venue.store, clock=clock), - exit_job=Flatten(), + exit_job=Flatten() if job is None else job, ) -def test_a_flatten_that_crashed_after_its_send_is_booked_by_the_split_and_the_rest_closed( - tmp_path: Path, -) -> None: - # Two partitions share BTC, so the boot books the resumed fill by its stored - # split (ADR-0054). The other symbol is still open, and the next run closes it. - _hold(tmp_path / "saga.db", {"BTC": "0.02", "ETH": "0.5"}) - _hold(tmp_path / "saga.db", {"BTC": "0.01"}, strategy_id="other") +def _crash_a_flatten(tmp_path: Path) -> _Venue: + """Run flatten until its first close reaches paper, then kill the run. + + Returns the venue that still holds the fill, with the store open again. + """ venue = _Venue(tmp_path / "saga.db", ManualClock()) link = _CrashesAfterTheSend(venue.paper, venue.cut) @@ -630,13 +630,31 @@ async def first_life() -> None: asyncio.run(first_life()) venue.store.close() venue.store = SQLiteStore(tmp_path / "saga.db") + return venue - async def second_life() -> int: - return await asyncio.wait_for(_engine(venue, venue.paper, tmp_path).run(), 5) + +def _next_run(venue: _Venue, tmp_path: Path, job: ExitJob) -> tuple[int, list[EventDict]]: + """Run ``job`` after the crash. Returns the exit code and every event it emitted.""" + + async def go() -> int: + return await asyncio.wait_for(_engine(venue, venue.paper, tmp_path, job).run(), 5) with capture_events() as logs: - exit_code = asyncio.run(second_life()) + exit_code = asyncio.run(go()) venue.store.close() + return exit_code, logs + + +def test_a_flatten_that_crashed_after_its_send_is_booked_by_the_split_and_the_rest_closed( + tmp_path: Path, +) -> None: + # Two partitions share BTC, so the boot books the resumed fill by its stored + # split (ADR-0054). The other symbol is still open, and the next run closes it. + _hold(tmp_path / "saga.db", {"BTC": "0.02", "ETH": "0.5"}) + _hold(tmp_path / "saga.db", {"BTC": "0.01"}, strategy_id="other") + venue = _crash_a_flatten(tmp_path) + + exit_code, logs = _next_run(venue, tmp_path, Flatten()) assert exit_code == 0 assert logs[-1]["outcome"] == "done" @@ -655,3 +673,23 @@ async def second_life() -> int: [("BTC", f"{OPERATOR}:BTC:2"), ("ETH", f"{OPERATOR}:ETH:1")], ) assert all(order.state is OrderState.FILLED for order in closes) + + +def test_a_cancel_all_after_a_crashed_flatten_books_the_close_at_boot(tmp_path: Path) -> None: + # The resume is the boot, not the flatten job, so any exit run does it (ADR-0056). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + _hold(tmp_path / "saga.db", {"BTC": "0.01"}, strategy_id="other") + venue = _crash_a_flatten(tmp_path) + + exit_code, logs = _next_run(venue, tmp_path, OperatorCancelAll()) + + assert exit_code == 0 + assert logs[-1]["job"] == "cancel_all" + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("40")), + ("other", "BTC"): (Decimal("0"), Decimal("20")), + } + with SQLiteStore(tmp_path / "saga.db") as store: + close = store.get_order(derive_cloid(f"{OPERATOR}:BTC:1")) + assert close is not None + assert close.state is OrderState.FILLED From a5b92b7fcd1780694f8ae54d8e26b64008bf2256 Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 11:53:27 -0300 Subject: [PATCH 5/9] docs(engine): say what #474 built and that only flatten reads the seq fold CONTEXT.md listed what flatten had built so far, but not the resume after a crash. The Cache docstring said every exit job starts a seq counter there. Cancel-all uses no seq, so only Flatten does. Co-Authored-By: Claude Opus 5.5 --- CONTEXT.md | 4 ++-- src/tickwright/engine/cache.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index 8ef6def5..073c4c38 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -477,8 +477,8 @@ partition at the strategy's entry price. A failed venue read also counts as an a nothing (ADR-0060). A read refused for a rate limit does not count. It waits like an order. Flatten is done only when the venue holds no resting order and no position. The fill split is built (#471). #472 built one attempt per symbol on paper replay. A symbol the feed never priced gets no -order, and the final read reports it. The size check before an order moved to #475. The rest is -not built yet (#408). +order, and the final read reports it. The size check before an order moved to #475. #474 built +the resume after a crash. Each close takes a new seq. The rest is not built yet (#408). _Avoid_: close out, liquidate (that is the venue's forced close), panic sell. **Reduce-only order**: diff --git a/src/tickwright/engine/cache.py b/src/tickwright/engine/cache.py index b5b2e4dd..3680eb9a 100644 --- a/src/tickwright/engine/cache.py +++ b/src/tickwright/engine/cache.py @@ -77,8 +77,8 @@ def seq_high_water(self) -> dict[str, int]: cancel consumed its own (ADR-0026 — omitting cancels would let a restart reuse a consumed id). One per-strategy counter across all symbols, so this is a single max over the strategy's records. The - strategies and the operator's exit jobs both start their counters - here, so neither can reuse an id after a restart. + strategies and ``Flatten`` both start their counters here, so neither + can reuse an id after a restart. """ high_water: dict[str, int] = {} for order in self._orders.values(): From 8fb8dd58115370886be5f9a5de28bc4707f533f3 Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 11:58:58 -0300 Subject: [PATCH 6/9] test(engine): prove a failed read after the wait stops flatten with no close Review R001 on #488. The branch had no test. A mutation that sized the close from the old read went green before this test. Co-Authored-By: Claude Opus 5.5 --- tests/engine/test_flatten_e2e.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index a3923d7d..0d874fb1 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -537,6 +537,23 @@ def test_a_close_left_open_that_no_pass_can_decide_gets_no_second_close_beside_i assert not [log for log in logs if log.get("signal_id") == f"{OPERATOR}:BTC:2"] +def test_a_failed_read_after_the_wait_stops_flatten_with_no_second_close( + tmp_path: Path, +) -> None: + # The old close may have filled during the wait. With no read of what is + # left, flatten cannot size a close, so it sends none (ADR-0060). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + _left_open(tmp_path / "saga.db", "BTC") + + # The first positions read is flatten's own. The second is the read after the wait. + exit_code, logs = _flatten(tmp_path, link=partial(_FailsOneRead, read="fetch_positions", nth=2)) + + assert exit_code == 1 + assert logs[-1]["outcome"] == "stopped" + assert logs[-1]["left"] == f"read_failed: {VenueReadFailure.SEND_FAILED.value}" + assert not [log for log in logs if log.get("signal_id") == f"{OPERATOR}:BTC:2"] + + class _Venue: """Paper on a bus of its own, so it outlives the engine run it serves. From 205c94bed2c24bd30b36980bd0032e42a62c0866 Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 12:00:14 -0300 Subject: [PATCH 7/9] test(engine): prove flatten sizes its close after a fill during the wait Review R002 on #488. No test failed if the close kept the size from the read before the wait. Now the old close fills during the wait, and the next close must ask only for what is left. Co-Authored-By: Claude Opus 5.5 --- tests/engine/test_flatten_e2e.py | 79 ++++++++++++++++++++++++++++++-- 1 file changed, 74 insertions(+), 5 deletions(-) diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index 0d874fb1..4b5a2bee 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -461,7 +461,7 @@ def test_a_failed_reconcile_pass_ends_the_wait_and_the_final_read_decides( assert logs[-1]["left"] == "BTC=0.02" -def _left_open(path: Path, symbol: str) -> str: +def _left_open(path: Path, symbol: str, *, quantity: str = "0.02") -> str: """Leave an ``__operator__`` close ``LIVE`` in the store, one paper never lists. That is a close the venue acked and then dropped (#242) before a crash. The @@ -481,10 +481,12 @@ def _left_open(path: Path, symbol: str) -> str: signal_id=f"{OPERATOR}:{symbol}:1", symbol=symbol, side=Side.SELL, - quantity=Decimal("0.02"), + quantity=Decimal(quantity), order_type=OrderType.MARKET, state=OrderState.LIVE, created_ts_ns=0, + # As the manager stores it at place, so a fill can be booked (ADR-0054). + split_basis=checks.split_basis(symbol), ) ) return cloid @@ -650,11 +652,21 @@ async def first_life() -> None: return venue -def _next_run(venue: _Venue, tmp_path: Path, job: ExitJob) -> tuple[int, list[EventDict]]: - """Run ``job`` after the crash. Returns the exit code and every event it emitted.""" +def _next_run( + venue: _Venue, + tmp_path: Path, + job: ExitJob, + *, + link: Callable[[PaperExchange], Exchange] | None = None, +) -> tuple[int, list[EventDict]]: + """Run ``job`` after the crash. Returns the exit code and every event it emitted. + + ``link`` puts a venue double in front of paper. + """ + exchange = venue.paper if link is None else link(venue.paper) async def go() -> int: - return await asyncio.wait_for(_engine(venue, venue.paper, tmp_path, job).run(), 5) + return await asyncio.wait_for(_engine(venue, exchange, tmp_path, job).run(), 5) with capture_events() as logs: exit_code = asyncio.run(go()) @@ -710,3 +722,60 @@ def test_a_cancel_all_after_a_crashed_flatten_books_the_close_at_boot(tmp_path: close = store.get_order(derive_cloid(f"{OPERATOR}:BTC:1")) assert close is not None assert close.state is OrderState.FILLED + + +class _FillsDuringTheWait(VenueLink): + """``close`` is unknown to the boot's read. It fills just before the next read. + + So it fills after flatten's first position read and before the wait ends. + """ + + def __init__(self, venue: Exchange, *, close: PlaceOrder) -> None: + super().__init__(venue) + self._close = close + self._reads = 0 + + async def fetch_order(self, ref: OrderRef) -> VenueOrderView | VenueReadFailure: + if ref.cloid == self._close.cloid: + self._reads += 1 + if self._reads == 1: + return VenueOrderView(status=None) + if self._reads == 2: + await self.place(self._close) + return await super().fetch_order(ref) + + +def test_a_close_left_open_that_fills_during_the_wait_leaves_only_the_rest_to_close( + tmp_path: Path, +) -> None: + # A close sized from the read before the wait would ask for the whole size + # again. Paper would shrink it to what is left and cancel the rest (ADR-0057). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + left_open = _left_open(tmp_path / "saga.db", "BTC", quantity="0.01") + close = PlaceOrder( + cloid=left_open, + symbol="BTC", + side=Side.SELL, + quantity=Decimal("0.01"), + order_type=OrderType.MARKET, + time_in_force=TimeInForce.IOC, + reduce_only=True, + ) + venue = _Venue(tmp_path / "saga.db", ManualClock()) + + exit_code, logs = _next_run( + venue, tmp_path, Flatten(), link=partial(_FillsDuringTheWait, close=close) + ) + + assert exit_code == 0 + assert logs[-1]["outcome"] == "done" + # Opened at 42000. Both halves close at the last row, 44000 (ADR-0059). + assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} + with SQLiteStore(tmp_path / "saga.db") as store: + old = store.get_order(left_open) + close = store.get_order(derive_cloid(f"{OPERATOR}:BTC:2")) + assert old is not None + assert old.state is OrderState.FILLED + assert close is not None + assert close.quantity == Decimal("0.01") + assert close.state is OrderState.FILLED From b961e9d850464b08fc2e4f69e4394b6bde28f98f Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 12:01:49 -0300 Subject: [PATCH 8/9] test(engine): give the late close its own name so mypy passes 205c94b used one name for the venue order and for the stored saga. Co-Authored-By: Claude Opus 5.5 --- tests/engine/test_flatten_e2e.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index 4b5a2bee..4c1a9f44 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -752,7 +752,7 @@ def test_a_close_left_open_that_fills_during_the_wait_leaves_only_the_rest_to_cl # again. Paper would shrink it to what is left and cancel the rest (ADR-0057). _hold(tmp_path / "saga.db", {"BTC": "0.02"}) left_open = _left_open(tmp_path / "saga.db", "BTC", quantity="0.01") - close = PlaceOrder( + late = PlaceOrder( cloid=left_open, symbol="BTC", side=Side.SELL, @@ -764,7 +764,7 @@ def test_a_close_left_open_that_fills_during_the_wait_leaves_only_the_rest_to_cl venue = _Venue(tmp_path / "saga.db", ManualClock()) exit_code, logs = _next_run( - venue, tmp_path, Flatten(), link=partial(_FillsDuringTheWait, close=close) + venue, tmp_path, Flatten(), link=partial(_FillsDuringTheWait, close=late) ) assert exit_code == 0 From f27904d79886f225e438c50a240cbb5555e68cd3 Mon Sep 17 00:00:00 2001 From: Marcos Achaval Date: Mon, 12 Oct 2026 12:01:49 -0300 Subject: [PATCH 9/9] refactor(engine): fold the seq high-water over Cache.all_orders again Review R003 on #488. ADR-0016, ADR-0024, and the all_orders docstring say the fold reads Cache.all_orders(). Moving the fold into the Cache made it read the dict directly, so all_orders lost its only caller. Co-Authored-By: Claude Opus 5.5 --- src/tickwright/engine/cache.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tickwright/engine/cache.py b/src/tickwright/engine/cache.py index 3680eb9a..546c2725 100644 --- a/src/tickwright/engine/cache.py +++ b/src/tickwright/engine/cache.py @@ -81,7 +81,7 @@ def seq_high_water(self) -> dict[str, int]: can reuse an id after a restart. """ high_water: dict[str, int] = {} - for order in self._orders.values(): + for order in self.all_orders(): seqs = [SignalId.parse(order.signal_id).seq] if order.cancel_signal_id is not None: seqs.append(SignalId.parse(order.cancel_signal_id).seq)