fix(executor): write the crash ledger before cancelling, so a dead roll is loud (#519)

keel/execution/executor.py

@@ -1658,6 +1658,13 @@ def _roll_stop(
     last-recorded `open_stop:<product_id>` -- ratchet-only, mirroring rail 9's invariant for
     entries. The replacement order still runs through `guards.check` (allowlist/caps/kill-switch/
     etc. -- every order, no exceptions) before it is placed.
+
+    **The cancel-then-place window is not atomic and cannot be made so** (see the comment at the
+    cancel below). What it CAN be is recoverable: an `unbracketed:<product_id>` record is written
+    before the venue is touched and cleared only once the replacement rests, so a process that
+    dies anywhere in between leaves levels the next cycle's sweep re-places from. Before #519 that
+    record did not exist here and one crash window was not merely unprotected but SILENT --
+    indistinguishable from a DCA tranche, which legitimately carries no stop.
     """
     prior_stop = repo.get_state(f"open_stop:{product_id}")
     if prior_stop is not None and new_stop < prior_stop:
@@ -1682,6 +1689,29 @@ def _roll_stop(
         )
         return None
 
+    # THE CRASH LEDGER, written BEFORE the venue is touched (#519).
+    #
+    # Everything below this line can die mid-flight, and until this record existed one of those
+    # deaths was SILENT. `_run_order` marks the old bracket `canceled` locally and then places the
+    # replacement; a process that dies in between left no resting bracket, no `unbracketed:`
+    # record, and therefore nothing for `reconcile_unbracketed_positions` to act on -- it took the
+    # skip branch that exists for DCA and the position stayed naked with no CRITICAL, looking
+    # exactly like a holding that carries no stop by design.
+    #
+    # Writing the intent first turns every one of those deaths into a state the existing sweep
+    # already converges: next cycle it finds no resting bracket, finds these levels, and re-places
+    # from them. The record is deliberately the SAME key `place_bracket` uses rather than a new
+    # `roll_intent:` one -- the sweep, its ledger-sized qty, its escalation and its
+    # clear-on-success semantics are written and tested once, and a second key would mean a second
+    # healer to keep correct.
+    #
+    # A record left behind by an aborted roll is harmless: the sweep skips any product whose
+    # bracket is still resting, and the levels it holds are ratchet-consistent either way.
+    repo.set_state(
+        f"{UNBRACKETED_PREFIX}{product_id}",
+        {"stop": new_stop, "target": target, "qty": qty},
+    )
+
     # CANCEL FIRST, then place. This inverts the old two-leg order of operations, and it has to:
     # the resting native bracket already commits the whole position, so placing a replacement
     # first would be rejected for insufficient funds. `edit_order` cannot avoid the inversion
@@ -1724,8 +1754,13 @@ def _roll_stop(
         ),
     )
     if not result.placed:
-        # The old bracket is already cancelled, so the position is NAKED right now. There is no
-        # silent recovery: the caller must retry or close the position. Never downgrade this.
+        # The old bracket is already cancelled, so the position is NAKED right now. The
+        # `unbracketed:` record written before the cancel is DELIBERATELY LEFT STANDING: it is
+        # what `reconcile_unbracketed_positions` re-places from on the next cycle (#519).
+        #
+        # The CRITICAL stays regardless. Automatic recovery next cycle is not a reason to
+        # downgrade an alert about a position that is unprotected RIGHT NOW -- the deployment
+        # cycles once per UTC day, so "next cycle" can be up to a day away. Never downgrade this.
         log_event(
             logger,
             logging.CRITICAL,
@@ -1736,13 +1771,18 @@ def _roll_stop(
             cancelled_order_id=old_stop_order_id,
             detail=(
                 "the previous bracket was cancelled and its replacement was REJECTED -- this "
-                "position currently has no protective stop at the exchange"
+                "position currently has no protective stop at the exchange. The unbracketed "
+                "record is retained so the next cycle's sweep re-places it."
             ),
         )
         return None
 
     repo.set_state(f"open_stop:{product_id}", new_stop)
     repo.set_state(f"open_target:{product_id}", target)
+    # The replacement is resting, so the crash ledger has served its purpose. Clearing it matches
+    # `place_bracket`'s own success path; leaving it would have the sweep re-place a bracket that
+    # already exists on the next cycle.
+    repo.set_state(f"{UNBRACKETED_PREFIX}{product_id}", None)
     log_event(
         logger,
         logging.INFO,

tests/execution/test_executor.py

@@ -30,7 +30,7 @@
 )
 from keel.data.db import connect, migrate
 from keel.data.repository import Repository
-from keel.execution import guards, sizing
+from keel.execution import executor, guards, sizing
 from keel.execution.executor import (
     CancelPending,
     CancelUnavailable,
@@ -2917,3 +2917,135 @@ def test_unserialisable_size_refuses_the_order_without_raising(repo, monkeypatch
     assert broker.place_calls == []
     assert broker.preview_calls == []
     assert repo.get_orders() == []
+
+
+def test_a_roll_writes_its_crash_ledger_before_touching_the_venue(repo, monkeypatch):
+    """#519: the window between cancel and replace must never be SILENT.
+
+    Before this, `_roll_stop` cancelled the old bracket and then placed the replacement without
+    ever writing an `unbracketed:` record -- `place_bracket` writes one only on a refused
+    PLACEMENT. A process dying in between left no resting bracket and no intent, so
+    `reconcile_unbracketed_positions` took the branch that exists for DCA and skipped the position
+    SILENTLY. Naked, and indistinguishable from a tranche that carries no stop by design.
+
+    Asserted by observing the state at the moment of the cancel, which is the crash point.
+    """
+    broker = FakeBroker()
+    stop_id = place_bracket(
+        broker,
+        repo,
+        _config(),
+        product_id="BTC-USD",
+        qty=Decimal("0.01"),
+        stop=Decimal("49000"),
+        target=Decimal("53000"),
+        rule_name="pullback_continuation",
+        now_ts=NOW_TS,
+    )
+    # place_bracket clears its own record on success -- so we start from nothing to find.
+    assert not repo.get_state(f"{executor.UNBRACKETED_PREFIX}BTC-USD")
+
+    seen: dict = {}
+    real_cancel = executor._cancel_at_exchange
+
+    def _spy(*args, **kwargs):
+        # The crash point: the old bracket is about to stop resting.
+        seen["intent"] = repo.get_state(f"{executor.UNBRACKETED_PREFIX}BTC-USD")
+        return real_cancel(*args, **kwargs)
+
+    monkeypatch.setattr(executor, "_cancel_at_exchange", _spy)
+
+    roll_to_break_even(
+        broker,
+        repo,
+        _config(),
+        product_id="BTC-USD",
+        old_stop_order_id=stop_id,
+        entry_price=Decimal("50000"),
+        qty=Decimal("0.01"),
+        rule_name="pullback_continuation",
+        now_ts=NOW_TS + 100,
+    )
+
+    assert seen["intent"], "no crash ledger existed at the cancel -- the sweep would skip silently"
+    assert seen["intent"]["stop"] == Decimal("50000")
+    assert seen["intent"]["target"] == Decimal("53000")
+
+
+def test_a_successful_roll_clears_its_crash_ledger(repo):
+    """Left standing, the sweep would re-place a bracket that already rests."""
+    broker = FakeBroker()
+    stop_id = place_bracket(
+        broker,
+        repo,
+        _config(),
+        product_id="BTC-USD",
+        qty=Decimal("0.01"),
+        stop=Decimal("49000"),
+        target=Decimal("53000"),
+        rule_name="pullback_continuation",
+        now_ts=NOW_TS,
+    )
+    new_id = roll_to_break_even(
+        broker,
+        repo,
+        _config(),
+        product_id="BTC-USD",
+        old_stop_order_id=stop_id,
+        entry_price=Decimal("50000"),
+        qty=Decimal("0.01"),
+        rule_name="pullback_continuation",
+        now_ts=NOW_TS + 100,
+    )
+
+    assert new_id is not None
+    assert not repo.get_state(f"{executor.UNBRACKETED_PREFIX}BTC-USD")
+
+
+def test_a_failed_roll_RETAINS_its_crash_ledger_for_the_sweep(repo, caplog):
+    """The naked case. The record is what the next cycle re-places from, so it must survive --
+    and the CRITICAL must survive with it: the deployment cycles once per UTC day, so "the sweep
+    will fix it" can be a day away. Recovery is not a reason to downgrade the alert."""
+
+    class _RejectingBroker(FakeBroker):
+        def __init__(self, **kw):
+            super().__init__(**kw)
+            self.calls = 0
+
+        def place_order(self, product_id, side, order_configuration):
+            self.calls += 1
+            if self.calls > 1:  # the original bracket places; the replacement fails
+                return {"success": False, "error": "INSUFFICIENT_FUND"}
+            return super().place_order(product_id, side, order_configuration)
+
+    broker = _RejectingBroker()
+    stop_id = place_bracket(
+        broker,
+        repo,
+        _config(),
+        product_id="BTC-USD",
+        qty=Decimal("0.01"),
+        stop=Decimal("49000"),
+        target=Decimal("53000"),
+        rule_name="pullback_continuation",
+        now_ts=NOW_TS,
+    )
+
+    with caplog.at_level(logging.CRITICAL):
+        result = roll_to_break_even(
+            broker,
+            repo,
+            _config(),
+            product_id="BTC-USD",
+            old_stop_order_id=stop_id,
+            entry_price=Decimal("50000"),
+            qty=Decimal("0.01"),
+            rule_name="pullback_continuation",
+            now_ts=NOW_TS + 100,
+        )
+
+    assert result is None
+    intent = repo.get_state(f"{executor.UNBRACKETED_PREFIX}BTC-USD")
+    assert intent, "a naked position with no ledger is the exact #519 hole"
+    assert intent["stop"] == Decimal("50000")
+    assert any("position_unprotected" in r.message for r in caplog.records)