348 lines
12 KiB
Python
348 lines
12 KiB
Python
"""Deferred background review on the managed local runtime.
|
|
|
|
Behavior contracts for agent/review_idle_queue.py and the decision
|
|
wrapper in run_agent.AIAgent._spawn_background_review:
|
|
|
|
- defer: auto + review runtime == managed local -> queued, not spawned
|
|
- defer: never, or non-managed runtime, or /refine -> immediate spawn
|
|
- queue coalesces per session (newest snapshot wins, age preserved)
|
|
- dispatch requires sustained process-quiet AND server idle
|
|
- aged-out items dispatch regardless of idleness (delay, never lose)
|
|
- preempted deferred reviews requeue with a bounded attempt cap
|
|
"""
|
|
|
|
import threading
|
|
import time
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from agent.review_idle_queue import (
|
|
ReviewIdleQueue,
|
|
_IDLE_SETTLE_S,
|
|
defer_max_age_s,
|
|
defer_mode,
|
|
)
|
|
|
|
|
|
# ── config parsing ───────────────────────────────────────────────
|
|
|
|
|
|
def test_defer_mode_values():
|
|
assert defer_mode(None) == "auto"
|
|
assert defer_mode({}) == "auto"
|
|
assert defer_mode({"defer": "auto"}) == "auto"
|
|
assert defer_mode({"defer": "never"}) == "never"
|
|
assert defer_mode({"defer": "NEVER"}) == "never"
|
|
# Unknown values fall back to auto (the safe, documented default).
|
|
assert defer_mode({"defer": "sometimes"}) == "auto"
|
|
assert defer_mode({"defer": 3}) == "auto"
|
|
|
|
|
|
def test_defer_max_age_parsing():
|
|
assert defer_max_age_s(None) == 30 * 60
|
|
assert defer_max_age_s({"defer_max_age_s": 120}) == 120.0
|
|
assert defer_max_age_s({"defer_max_age_s": "600"}) == 600.0
|
|
# Nonsense and non-positive fall back to the default.
|
|
assert defer_max_age_s({"defer_max_age_s": "soon"}) == 30 * 60
|
|
assert defer_max_age_s({"defer_max_age_s": 0}) == 30 * 60
|
|
assert defer_max_age_s({"defer_max_age_s": -5}) == 30 * 60
|
|
|
|
|
|
# ── queue harness ────────────────────────────────────────────────
|
|
|
|
|
|
class _FakeAgent:
|
|
def __init__(self):
|
|
self.spawned = []
|
|
self.session_id = "sess-x"
|
|
|
|
def _spawn_background_review_now(self, **kwargs):
|
|
self.spawned.append(kwargs)
|
|
|
|
|
|
def _make_queue(now=None, server_idle=True):
|
|
q = ReviewIdleQueue()
|
|
clock = {"t": 0.0}
|
|
if now is None:
|
|
q._now = lambda: clock["t"]
|
|
else:
|
|
q._now = now
|
|
q._server_idle = lambda: server_idle
|
|
# Never start the real dispatcher thread in unit tests.
|
|
q._ensure_thread = lambda: None
|
|
return q, clock
|
|
|
|
|
|
def test_enqueue_coalesces_per_session_newest_wins_oldest_age():
|
|
q, clock = _make_queue()
|
|
agent = _FakeAgent()
|
|
|
|
clock["t"] = 100.0
|
|
q.enqueue(agent, "s1", {"messages_snapshot": ["old"], "task_cfg": {}})
|
|
clock["t"] = 200.0
|
|
q.enqueue(agent, "s1", {"messages_snapshot": ["new"], "task_cfg": {}})
|
|
q.enqueue(agent, "s2", {"messages_snapshot": ["other"], "task_cfg": {}})
|
|
|
|
assert q.pending_count() == 2
|
|
with q._lock:
|
|
item = q._pending["s1"]
|
|
# Newest snapshot won, but the age clock kept the ORIGINAL enqueue
|
|
# time so a busy session cannot push its own age-out forever.
|
|
assert item.kwargs["messages_snapshot"] == ["new"]
|
|
assert item.enqueued_at == 100.0
|
|
|
|
|
|
def test_dispatch_waits_for_sustained_quiet():
|
|
q, clock = _make_queue()
|
|
agent = _FakeAgent()
|
|
q.enqueue(agent, "s1", {"task_cfg": {}})
|
|
|
|
# A live turn: nothing dispatches.
|
|
q.note_turn_started()
|
|
assert q._pop_dispatchable() is None
|
|
|
|
# Turn finished, but the settle window hasn't elapsed.
|
|
q.note_turn_finished()
|
|
assert q._pop_dispatchable() is None
|
|
|
|
# Quiet long enough -> dispatchable.
|
|
clock["t"] += _IDLE_SETTLE_S + 1
|
|
item = q._pop_dispatchable()
|
|
assert item is not None and item.session_key == "s1"
|
|
assert q.pending_count() == 0
|
|
|
|
|
|
def test_dispatch_blocked_by_busy_server():
|
|
q, clock = _make_queue(server_idle=False)
|
|
agent = _FakeAgent()
|
|
q.enqueue(agent, "s1", {"task_cfg": {}})
|
|
q.note_turn_started()
|
|
q.note_turn_finished()
|
|
clock["t"] += _IDLE_SETTLE_S + 1
|
|
# Process is quiet but the managed server has a processing slot
|
|
# (another profile's session, a live prefill): hold.
|
|
assert q._pop_dispatchable() is None
|
|
assert q.pending_count() == 1
|
|
|
|
|
|
def test_aged_out_item_dispatches_despite_busy_server():
|
|
q, clock = _make_queue(server_idle=False)
|
|
agent = _FakeAgent()
|
|
q.enqueue(agent, "s1", {"task_cfg": {"defer_max_age_s": 60}})
|
|
q.note_turn_started() # never goes quiet
|
|
clock["t"] += 61
|
|
item = q._pop_dispatchable()
|
|
assert item is not None
|
|
assert item.session_key == "s1"
|
|
|
|
|
|
def test_new_turn_resets_the_quiet_clock():
|
|
q, clock = _make_queue()
|
|
agent = _FakeAgent()
|
|
q.enqueue(agent, "s1", {"task_cfg": {}})
|
|
q.note_turn_started()
|
|
q.note_turn_finished()
|
|
clock["t"] += _IDLE_SETTLE_S - 2
|
|
# A new prompt arrives just before the settle window closes.
|
|
q.note_turn_started()
|
|
clock["t"] += 30
|
|
assert q._pop_dispatchable() is None # still live
|
|
q.note_turn_finished()
|
|
assert q._pop_dispatchable() is None # settle restarts
|
|
clock["t"] += _IDLE_SETTLE_S + 1
|
|
assert q._pop_dispatchable() is not None
|
|
|
|
|
|
def test_nested_turns_require_all_to_finish():
|
|
q, clock = _make_queue()
|
|
agent = _FakeAgent()
|
|
q.enqueue(agent, "s1", {"task_cfg": {}})
|
|
q.note_turn_started()
|
|
q.note_turn_started()
|
|
q.note_turn_finished()
|
|
clock["t"] += _IDLE_SETTLE_S + 1
|
|
assert q._pop_dispatchable() is None # one turn still live
|
|
q.note_turn_finished()
|
|
clock["t"] += _IDLE_SETTLE_S + 1
|
|
assert q._pop_dispatchable() is not None
|
|
|
|
|
|
# ── the decision wrapper ─────────────────────────────────────────
|
|
|
|
|
|
def _wrapper_agent(monkeypatch, defer="auto", managed=True):
|
|
"""A minimal object wearing the real _spawn_background_review."""
|
|
import run_agent
|
|
from agent import review_idle_queue as riq
|
|
|
|
agent = _FakeAgent()
|
|
agent._delegate_depth = 0
|
|
calls = {"enqueued": [], "spawned": []}
|
|
|
|
monkeypatch.setattr(
|
|
"agent.background_review.load_background_review_settings",
|
|
lambda: (True, {"defer": defer}),
|
|
)
|
|
monkeypatch.setattr(
|
|
riq, "review_targets_managed_local", lambda a, cfg: managed
|
|
)
|
|
monkeypatch.setattr(
|
|
riq.QUEUE, "enqueue",
|
|
lambda a, key, kw: calls["enqueued"].append((key, kw)),
|
|
)
|
|
agent._spawn_background_review_now = (
|
|
lambda **kw: calls["spawned"].append(kw)
|
|
)
|
|
bound = types.MethodType(run_agent.AIAgent._spawn_background_review, agent)
|
|
return bound, calls
|
|
|
|
|
|
def test_wrapper_defers_managed_local_auto(monkeypatch):
|
|
spawn, calls = _wrapper_agent(monkeypatch, defer="auto", managed=True)
|
|
spawn([{"role": "user", "content": "hi"}], review_memory=True)
|
|
assert len(calls["enqueued"]) == 1
|
|
assert calls["spawned"] == []
|
|
key, kwargs = calls["enqueued"][0]
|
|
assert key == "sess-x"
|
|
assert kwargs["review_memory"] is True
|
|
|
|
|
|
def test_wrapper_spawns_immediately_for_non_managed(monkeypatch):
|
|
spawn, calls = _wrapper_agent(monkeypatch, defer="auto", managed=False)
|
|
spawn([{"role": "user", "content": "hi"}], review_skills=True)
|
|
assert calls["enqueued"] == []
|
|
assert len(calls["spawned"]) == 1
|
|
|
|
|
|
def test_wrapper_defer_never_is_old_behavior(monkeypatch):
|
|
spawn, calls = _wrapper_agent(monkeypatch, defer="never", managed=True)
|
|
spawn([{"role": "user", "content": "hi"}], review_memory=True)
|
|
assert calls["enqueued"] == []
|
|
assert len(calls["spawned"]) == 1
|
|
|
|
|
|
def test_wrapper_refine_bypasses_queue(monkeypatch):
|
|
spawn, calls = _wrapper_agent(monkeypatch, defer="auto", managed=True)
|
|
spawn([{"role": "user", "content": "hi"}], review_memory=True,
|
|
focus="save the deploy workflow")
|
|
assert calls["enqueued"] == []
|
|
assert len(calls["spawned"]) == 1
|
|
assert calls["spawned"][0]["focus"] == "save the deploy workflow"
|
|
|
|
|
|
def test_wrapper_bare_refine_bypasses_queue(monkeypatch):
|
|
"""/refine with no focus text is still explicit: never deferred."""
|
|
spawn, calls = _wrapper_agent(monkeypatch, defer="auto", managed=True)
|
|
spawn([{"role": "user", "content": "hi"}], review_memory=True,
|
|
focus=None, explicit=True)
|
|
assert calls["enqueued"] == []
|
|
assert len(calls["spawned"]) == 1
|
|
|
|
|
|
def test_wrapper_cloud_fast_path_skips_runtime_resolution(monkeypatch):
|
|
"""No managed server on the machine -> the classifier answers from the
|
|
TTL-cached netloc probe alone, without resolving the review runtime.
|
|
Guards the cloud-only turn tail from growing new work."""
|
|
from agent import review_idle_queue as riq
|
|
|
|
resolved = {"count": 0}
|
|
|
|
def _explode(agent, cfg):
|
|
resolved["count"] += 1
|
|
raise AssertionError("runtime resolution must not run")
|
|
|
|
monkeypatch.setattr(
|
|
"agent.auxiliary_client._managed_local_netloc", lambda: "")
|
|
monkeypatch.setattr(
|
|
"agent.background_review._resolve_review_runtime", _explode)
|
|
assert riq.review_targets_managed_local(object(), {}) is False
|
|
assert resolved["count"] == 0
|
|
|
|
|
|
def test_dispatcher_rechecks_enabled_gate(monkeypatch):
|
|
"""A review disabled while queued must not be resurrected at dispatch."""
|
|
from agent import review_idle_queue as riq
|
|
|
|
q, clock = _make_queue()
|
|
agent = _FakeAgent()
|
|
q.enqueue(agent, "s1", {"task_cfg": {}})
|
|
monkeypatch.setattr(
|
|
"agent.background_review.load_background_review_settings",
|
|
lambda: (False, {}),
|
|
)
|
|
item = None
|
|
clock["t"] += _IDLE_SETTLE_S + 1
|
|
q.note_turn_started()
|
|
q.note_turn_finished()
|
|
clock["t"] += _IDLE_SETTLE_S + 1
|
|
item = q._pop_dispatchable()
|
|
assert item is not None
|
|
assert q._still_enabled(item) is False
|
|
|
|
|
|
# ── requeue on preemption ────────────────────────────────────────
|
|
|
|
|
|
class _Run:
|
|
def __init__(self, cancelled):
|
|
self.cancel_requested = threading.Event()
|
|
if cancelled:
|
|
self.cancel_requested.set()
|
|
|
|
|
|
def _requeue_agent(monkeypatch, managed=True):
|
|
import run_agent
|
|
from agent import review_idle_queue as riq
|
|
|
|
agent = _FakeAgent()
|
|
calls = {"enqueued": []}
|
|
monkeypatch.setattr(
|
|
riq, "review_targets_managed_local", lambda a, cfg: managed
|
|
)
|
|
monkeypatch.setattr(
|
|
riq.QUEUE, "enqueue",
|
|
lambda a, key, kw: calls["enqueued"].append(kw),
|
|
)
|
|
agent._REVIEW_REQUEUE_MAX_ATTEMPTS = (
|
|
run_agent.AIAgent._REVIEW_REQUEUE_MAX_ATTEMPTS
|
|
)
|
|
bound = types.MethodType(
|
|
run_agent.AIAgent._maybe_requeue_preempted_review, agent
|
|
)
|
|
return bound, calls
|
|
|
|
|
|
def test_preempted_review_requeues(monkeypatch):
|
|
requeue, calls = _requeue_agent(monkeypatch)
|
|
requeue(_Run(cancelled=True),
|
|
{"task_cfg": {"defer": "auto"}, "focus": None,
|
|
"_requeue_attempts": 1})
|
|
assert len(calls["enqueued"]) == 1
|
|
# The attempt counter rides along so the cap survives the round trip.
|
|
assert calls["enqueued"][0]["_requeue_attempts"] == 1
|
|
|
|
|
|
def test_completed_review_does_not_requeue(monkeypatch):
|
|
requeue, calls = _requeue_agent(monkeypatch)
|
|
requeue(_Run(cancelled=False),
|
|
{"task_cfg": {"defer": "auto"}, "focus": None,
|
|
"_requeue_attempts": 1})
|
|
assert calls["enqueued"] == []
|
|
|
|
|
|
def test_requeue_attempt_cap(monkeypatch):
|
|
requeue, calls = _requeue_agent(monkeypatch)
|
|
requeue(_Run(cancelled=True),
|
|
{"task_cfg": {"defer": "auto"}, "focus": None,
|
|
"_requeue_attempts": 4})
|
|
assert calls["enqueued"] == []
|
|
|
|
|
|
def test_requeue_skips_non_managed(monkeypatch):
|
|
requeue, calls = _requeue_agent(monkeypatch, managed=False)
|
|
requeue(_Run(cancelled=True),
|
|
{"task_cfg": {"defer": "auto"}, "focus": None,
|
|
"_requeue_attempts": 1})
|
|
assert calls["enqueued"] == []
|