feat(engine): resume a crashed flatten without sending a second close

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**:

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),

docs/module-maps/safe-exit.md

@@ -162,12 +162,18 @@ 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):**
 
 - 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()`.

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 ``Flatten`` both start their counters here, so neither
+        can reuse an id after a restart.
+        """
+        high_water: dict[str, int] = {}
+        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)
+            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

src/tickwright/engine/exit_run.py

@@ -138,15 +138,48 @@ 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)
+                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, resumed, seq, context), context)
         return await _final_read(context)
 
 
-async def _close(symbol: str, size: Decimal, context: ExitContext) -> str:
+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).
 
     Returns the cloid of the order's saga.
@@ -157,7 +190,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,

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.