Threaded fast trigger + incremental ingest + naked-stop guard

Three fixes from the first live session:

1) Naked-stop bug (critical): the entry-timeout guard cancelled an entry it
   thought was unfilled AFTER it had really filled (our order_filled events lag
   the 60s sync). Cancelling a bracket parent cancels its stop/TP legs, so the
   position went NAKED and bled 22% with no floor. Guard now checks the broker's
   LIVE positions (get_positions) before cancelling and never cancels a filled
   name. Regression test added.

2) Speed: ingest re-fetched a 240-min window for ~80 symbols every minute (~46s
   measured), and the single-threaded loop froze the 4s trigger that whole time
   -> breakouts caught once a minute, not in real time. Ingest is now
   incremental (heavy first pass covers the session so the opening range is
   present; then ~20-min refreshes, closed bars never change). Loop passes
   dropped from ~70s to ~2-5s.

3) Threaded trigger: the fast breakout detector/fire loop now runs on its own
   thread, fully decoupled from the slow scheduler. EventStore, the execution
   service, and the trigger book are made thread-safe (locks) so the thread and
   main loop can't race the shared psycopg2 connection or in-memory state.
   Verified live: fired EZGO/NEOV, all positions retain active stops.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

runtime/triggers.py

@@ -23,6 +23,8 @@ State machine per symbol:
 
 from __future__ import annotations
 
+import functools
+import threading
 from dataclasses import dataclass, field
 
 WAITING = "waiting"
@@ -36,6 +38,17 @@ FILLED = "filled"
 _COMMITTED = (FIRED, FILLED)
 
 
+def _synced(method):
+    """Serialize a book method behind self._lock so the main loop's arm() and
+    the trigger thread's poll/fire never iterate ``triggers`` while it's being
+    rebuilt."""
+    @functools.wraps(method)
+    def wrapper(self, *args, **kwargs):
+        with self._lock:
+            return method(self, *args, **kwargs)
+    return wrapper
+
+
 @dataclass
 class ArmedTrigger:
     symbol: str
@@ -83,6 +96,7 @@ class ArmedTriggerBook:
     rvol_min: float = 2.0       # min relative volume to be eligible to fire
     min_range_pct: float = 0.004  # opening range must be at least this wide
     triggers: dict[str, ArmedTrigger] = field(default_factory=dict)
+    _lock: object = field(default_factory=threading.RLock, repr=False, compare=False)
 
     def _eligible(self, t: ArmedTrigger) -> bool:
         return (
@@ -94,6 +108,7 @@ class ArmedTriggerBook:
             and t.range_pct >= self.min_range_pct
         )
 
+    @_synced
     def arm(self, candidates: list[dict]) -> None:
         """Refresh the book from a ranked candidate list.
 
@@ -131,11 +146,13 @@ class ArmedTriggerBook:
             new[sym] = t
         self.triggers = new
 
+    @_synced
     def update_price(self, symbol: str, price: float | None) -> None:
         t = self.triggers.get(symbol)
         if t is not None and price is not None:
             t.price = float(price)
 
+    @_synced
     def fires(self) -> list[ArmedTrigger]:
         """Armed triggers whose live price has reached or crossed the trigger."""
         return [
@@ -147,11 +164,13 @@ class ArmedTriggerBook:
             and t.price >= t.trigger
         ]
 
+    @_synced
     def mark_fired(self, symbol: str) -> None:
         t = self.triggers.get(symbol)
         if t is not None:
             t.state = FIRED
 
+    @_synced
     def mark_filled(self, held_symbols) -> None:
         """Promote tracked symbols that are now open broker positions to filled."""
         held = set(held_symbols or ())
@@ -159,6 +178,7 @@ class ArmedTriggerBook:
             if sym in held:
                 t.state = FILLED
 
+    @_synced
     def snapshot(self) -> list[dict]:
         """Board-ready rows, ordered armed -> fired -> filled -> waiting -> weak."""
         order = {ARMED: 0, FIRED: 1, FILLED: 2, WAITING: 3, WEAK: 4}