73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
"""Pure classification logic for the agent supervisor — no rclaude/process."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from uuid import uuid4
|
|
|
|
from claire.agent.supervisor import detect_wedged_and_orphaned, should_auto_continue
|
|
from claire.rclaude import SessionRow, TmuxRow
|
|
|
|
NOW = 1_000_000.0
|
|
|
|
|
|
def _sess(uuid, *, host="local", age_s=0):
|
|
return SessionRow(host=host, uuid=uuid, snippet="", cwd="/x", mtime_epoch=int(NOW) - age_s)
|
|
|
|
|
|
def _tmux(resumed_uuid, *, host="local", name="claude-x-1"):
|
|
return TmuxRow(host=host, session_name=name, detail="1 windows", resumed_uuid=resumed_uuid)
|
|
|
|
|
|
def test_wedged_when_live_pane_and_stale_mtime():
|
|
u = uuid4()
|
|
wedged, orphaned = detect_wedged_and_orphaned(
|
|
[_sess(u, age_s=400)], [_tmux(u)], wedge_threshold_s=300, now=NOW
|
|
)
|
|
assert [s.uuid for s in wedged] == [u]
|
|
assert orphaned == []
|
|
|
|
|
|
def test_not_wedged_when_fresh():
|
|
u = uuid4()
|
|
wedged, orphaned = detect_wedged_and_orphaned(
|
|
[_sess(u, age_s=10)], [_tmux(u)], wedge_threshold_s=300, now=NOW
|
|
)
|
|
assert wedged == [] and orphaned == []
|
|
|
|
|
|
def test_orphaned_when_no_live_pane():
|
|
u = uuid4()
|
|
wedged, orphaned = detect_wedged_and_orphaned(
|
|
[_sess(u, age_s=9999)], [], wedge_threshold_s=300, now=NOW
|
|
)
|
|
assert wedged == []
|
|
assert [s.uuid for s in orphaned] == [u]
|
|
|
|
|
|
def test_remote_sessions_not_supervised():
|
|
u = uuid4() # host != "local" → another machine's session, skip it
|
|
wedged, orphaned = detect_wedged_and_orphaned(
|
|
[_sess(u, host="apricot", age_s=9999)], [], wedge_threshold_s=300, now=NOW
|
|
)
|
|
assert wedged == [] and orphaned == []
|
|
|
|
|
|
def test_auto_continue_gate():
|
|
# Continuable when not parked and under the cap.
|
|
assert should_auto_continue(None, 0, 3) is True
|
|
assert should_auto_continue("in_progress", 2, 3) is True
|
|
# Capped.
|
|
assert should_auto_continue("in_progress", 3, 3) is False
|
|
# Parked states never auto-continue.
|
|
for parked in ("blocked", "user_review", "claire_review", "done"):
|
|
assert should_auto_continue(parked, 0, 3) is False
|
|
|
|
|
|
def test_no_resumed_uuid_means_no_wedge_classification():
|
|
# Older rclaude omits resumed_uuid → can't correlate → never act blind.
|
|
u = uuid4()
|
|
wedged, orphaned = detect_wedged_and_orphaned(
|
|
[_sess(u, age_s=9999)], [_tmux(None)], wedge_threshold_s=300, now=NOW
|
|
)
|
|
assert wedged == [] # not classified wedged without correlation
|
|
assert [s.uuid for s in orphaned] == [u] # no live pane matched → orphaned
|