test(agent): synchronize lifecycle and MoA concurrency checks

tests/agent/test_moa_loop_mode.py

@@ -536,27 +536,26 @@ def test_run_reference_prepends_advisory_system_prompt(monkeypatch):
 def test_references_run_in_parallel(monkeypatch):
     """References fan out concurrently (delegate-batch semantics), not serially.
 
-    Each reference sleeps; wall-time must approximate the slowest single call,
-    not the sum. Order is preserved and a failing reference is isolated.
+    Both successful references must enter their calls before either can finish.
+    Order is preserved and a failing reference is isolated.
     """
-    import time
+    import threading
 
     from agent import moa_loop
 
     # Force _extract_text down its fallback path (no transport normalize).
     monkeypatch.setattr(moa_loop, "get_transport", lambda *_a, **_k: None)
 
-    barrier_hits = []
+    successful_calls_entered = threading.Barrier(2)
 
-    def slow_call_llm(**kwargs):
-        barrier_hits.append(time.monotonic())
+    def synchronized_call_llm(**kwargs):
         model = kwargs["model"]
         if model == "boom":
             raise RuntimeError("kaboom")
-        time.sleep(0.5)
+        successful_calls_entered.wait(timeout=5)
         return _response(f"resp-{kwargs['provider']}")
 
-    monkeypatch.setattr(moa_loop, "call_llm", slow_call_llm)
+    monkeypatch.setattr(moa_loop, "call_llm", synchronized_call_llm)
 
     refs = [
         {"provider": "p1", "model": "ok"},
@@ -565,22 +564,16 @@ def test_references_run_in_parallel(monkeypatch):
         {"provider": "p3", "model": "ok"},
     ]
 
-    start = time.monotonic()
     out = moa_loop._run_references_parallel(
         refs, [{"role": "user", "content": "hi"}], temperature=0.6, max_tokens=64
     )
-    elapsed = time.monotonic() - start
 
-    # Two 0.5s sleeps run concurrently → well under the 1.0s serial floor.
-    # Threshold sits at 0.95s (not tight against 0.5s) to tolerate CI
-    # thread-pool startup jitter while still failing hard if the two calls
-    # ran serially (which would be ≥1.0s).
-    assert elapsed < 0.95, f"references did not run in parallel (took {elapsed:.2f}s)"
     # Output order matches input order (stable Reference N labelling).
     assert [label for label, _, _ in out] == ["p1:ok", "moa:preset", "p2:boom", "p3:ok"]
     assert "recursively reference MoA" in out[1][1]
     assert out[2][1].startswith("[failed:")
     assert out[0][1] == "resp-p1"
+    assert out[3][1] == "resp-p3"
 
 
 def test_references_parallel_without_agent_is_unaffected(monkeypatch):

tests/agent/test_openai_client_lifecycle.py

@@ -1,6 +1,5 @@
 import sys
 import threading
-import time
 import types
 from types import SimpleNamespace
 
@@ -100,16 +99,32 @@ def test_retry_after_api_connection_error_recreates_request_client(monkeypatch):
 
 
 def test_stale_non_stream_close_is_single_owner(monkeypatch):
-    def slow_responder(**kwargs):
-        time.sleep(0.1)
+    from agent import chat_completion_helpers
+
+    response_started = threading.Event()
+    request_aborted = threading.Event()
+
+    def blocked_responder(**kwargs):
+        response_started.set()
+        assert request_aborted.wait(timeout=5), "stale watchdog never aborted request"
         raise _connection_error()
 
-    request_client = FakeRequestClient(slow_responder)
+    request_client = FakeRequestClient(blocked_responder)
     factory = OpenAIFactory([request_client])
     monkeypatch.setattr("agent.process_bootstrap.OpenAI", factory)
+    monkeypatch.setattr(
+        chat_completion_helpers,
+        "time",
+        SimpleNamespace(time=lambda: 1.0 if response_started.is_set() else 0.0),
+    )
 
     agent = _build_agent()
-    agent._compute_non_stream_stale_timeout = lambda api_payload: 0.01
+    agent._compute_non_stream_stale_timeout = lambda api_payload: 0.5
+    monkeypatch.setattr(
+        agent,
+        "_force_close_tcp_sockets",
+        lambda client: (request_aborted.set(), 1)[1],
+    )
 
     with pytest.raises(APIConnectionError):
         agent._interruptible_api_call({"model": agent.model, "messages": []})