momentum devandClaude Fable 5
5896d291d8
Exits move off the single-threaded scheduler onto their own daemon thread at 1s (mirrors the fast trigger thread). Dedicated research connection (shared facade isn't thread-safe); scheduler exits step auto-disabled when the thread runs (single-caller invariant — no racing instance state, no double-submit); manage() already idempotent (stop ratchets >1c up only). 3 driver tests + 539 suite green. On fast small-caps this is the 'get out fast' the owner asked for. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
61 lines
1.6 KiB
Python
61 lines
1.6 KiB
Python
"""Plan 018 — sub-second exit management on its own thread.
|
|
|
|
The driver (run_exit_loop) must: call manage() every interval, survive a
|
|
manage() exception without dying (a live exit loop that crashes leaves
|
|
positions unmanaged), and stop promptly when the stop_event is set."""
|
|
|
|
import threading
|
|
import time
|
|
|
|
from runtime.exit_thread import run_exit_loop
|
|
|
|
|
|
def test_loop_calls_manage_repeatedly_then_stops():
|
|
calls = {"n": 0}
|
|
stop = threading.Event()
|
|
|
|
class Mgr:
|
|
def manage(self):
|
|
calls["n"] += 1
|
|
if calls["n"] >= 3:
|
|
stop.set() # end after 3 passes
|
|
return []
|
|
|
|
run_exit_loop(Mgr(), stop, interval=0.01)
|
|
assert calls["n"] == 3
|
|
|
|
|
|
def test_loop_survives_a_manage_exception():
|
|
calls = {"n": 0}
|
|
stop = threading.Event()
|
|
errs = []
|
|
|
|
class Mgr:
|
|
def manage(self):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise RuntimeError("broker blip") # must NOT kill the loop
|
|
if calls["n"] >= 3:
|
|
stop.set()
|
|
return []
|
|
|
|
run_exit_loop(Mgr(), stop, interval=0.01, on_error=errs.append)
|
|
assert calls["n"] == 3 # kept going after the exception
|
|
assert len(errs) == 1
|
|
|
|
|
|
def test_stop_event_ends_loop_without_calling_manage():
|
|
calls = {"n": 0}
|
|
stop = threading.Event()
|
|
stop.set() # already stopped
|
|
|
|
class Mgr:
|
|
def manage(self):
|
|
calls["n"] += 1
|
|
return []
|
|
|
|
t0 = time.monotonic()
|
|
run_exit_loop(Mgr(), stop, interval=5.0) # would hang 5s if it waited first
|
|
assert calls["n"] == 0
|
|
assert time.monotonic() - t0 < 1.0
|