fix(execution): bind parent row to request identity before child evaluation; add parent mismatch and race tests

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

tests/application/test_execution_cancel.py

@@ -225,3 +225,69 @@ def test_timeout_cancel_first_attempt_with_retry_before_none():
     )
     execution.run_once()  # should trigger timeout cancel
     assert broker.cancel_calls == 1
+
+
+# ── Regression: concurrent timeout/operator cancel race — broker call count 1 ─
+
+def test_timeout_operator_race_broker_called_exactly_once():
+    """Timeout cancel loop and operator cancel_pending_slot race to the same first
+    reservation.  The repository CAS guarantees exactly one winner; broker must
+    receive cancel_order exactly once regardless of scheduling order.
+
+    Uses a threading barrier so both paths enter reserve_cancel_attempt
+    simultaneously, making the race deterministic.
+    """
+    import threading
+
+    broker = ProgrammableBroker()
+    repo, _, intents, execution, _, _ = _wire(broker)
+    sid = _buy_to_holding(intents, execution)
+    intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid))
+    broker.responder = lambda side: FillStatus(
+        state="accepted",
+        raw={"order_open_confirmed": True},
+    )
+    execution.run_once()  # dispatch sell
+
+    # Backdate so timeout_cancel fires in the next run_once()
+    s = repo.fetch_slot(sid)
+    repo._slots[sid] = replace(
+        s,
+        order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=120),
+        order_date="20260901",
+        timeout_sec=30,
+    )
+
+    barrier = threading.Barrier(2)
+    original_reserve = repo.reserve_cancel_attempt
+
+    def _barrier_reserve(slot_id, broker_order_id, *, retry_before=None):
+        barrier.wait(timeout=5)
+        return original_reserve(slot_id, broker_order_id, retry_before=retry_before)
+
+    repo.reserve_cancel_attempt = _barrier_reserve
+
+    errors = []
+
+    def _operator():
+        try:
+            execution.cancel_pending_slot(sid)
+        except Exception as exc:  # noqa: BLE001
+            errors.append(exc)
+
+    t = threading.Thread(target=_operator)
+    t.start()
+
+    # Execution loop's timeout cancel is the second concurrent caller.
+    try:
+        execution.run_once()
+    except Exception as exc:  # noqa: BLE001
+        errors.append(exc)
+
+    t.join(timeout=5)
+    assert not errors, f"thread errors: {errors}"
+
+    # CAS ensures exactly one winner — broker cancel must be called once.
+    assert broker.cancel_calls == 1, (
+        f"expected exactly 1 broker cancel call, got {broker.cancel_calls}"
+    )