Compare commits

...
Author SHA1 Message Date
Devin AI a79489829b fix(core): scope stall reaping to the caller's subtree and commit failures atomically
Reap only descendants of the waiting agent, re-check status and silence
under the coordinator lock at the moment the failure is written, and stop
claiming the terminal notice is already queued in the wait result.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-08 18:59:12 +00:00
Devin AI c8a472a6a6 fix(core): bound silent agent turns and fail stalled children from wait_for_agents
The model-stream idle guard only covers the model call. A turn that wedged
anywhere else the run loop awaits between events (a re-issued request that
never opens, a hung stream close, tool transport) had no bound, so the agent
stayed running forever and a parent parked in wait_for_agents never returned.

- run cycle: abandon a turn that emits no run event for
  STRIX_AGENT_STALL_TIMEOUT (default 30m) and route it through the
  transient-retry path
- coordinator: track per-agent last_activity; reap_stalled fails and cancels
  agents silent past a threshold
- wait_for_agents: reap stalled children before parking and on timeout,
  reporting them as stalled_agents

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-08 18:52:10 +00:00
devin-ai-integration[bot]andAhmed Allam 52b1923347 fix(models): frontier model check matches the model name only, never the provider route (#1280)
Co-authored-by: Ahmed Allam <[email protected]>
2026-09-06 11:09:24 -07:00
10 changed files with 430 additions and 65 deletions

No files matched your search

+23 -48
View File
@@ -593,17 +593,27 @@ RECOMMENDED_MODEL_NAMES = (
_RECOMMENDED_MODEL_NAME_SET = frozenset(name.lower() for name in RECOMMENDED_MODEL_NAMES)
FRONTIER_MODEL_FAMILIES = (
(("azure", "azure_ai", "bedrock_mantle", "chatgpt", "openai"), ("gpt-5",)),
(
("anthropic", "azure_ai", "bedrock", "claude", "databricks", "snowflake", "vertex_ai"),
("claude-fable-5", "claude-opus-5", "claude-opus-4", "claude-sonnet-5", "claude-sonnet-4"),
),
(("google", "gemini", "vertex_ai"), ("gemini-3",)),
(("deepseek",), ("deepseek-v4", "deepseek-r1", "deepseek-reasoner")),
(("alibaba", "dashscope", "qwen"), ("qwen3.8", "qwen3.7", "qwen3-max")),
(("moonshot", "moonshotai", "kimi"), ("kimi-k3", "kimi-k2.7", "kimi-k2.6")),
(("zai", "z-ai", "zai-org", "zhipuai"), ("glm-5.3", "glm-5.2")),
# Matched against the bare model name only: the route (``openai/``, ``openrouter/``,
# a local gateway, ...) says nothing about the model's quality.
FRONTIER_MODEL_PREFIXES = (
"gpt-5",
"claude-fable-5",
"claude-opus-5",
"claude-opus-4",
"claude-sonnet-5",
"claude-sonnet-4",
"gemini-3",
"deepseek-v4",
"deepseek-r1",
"deepseek-reasoner",
"qwen3.8",
"qwen3.7",
"qwen3-max",
"kimi-k3",
"kimi-k2.7",
"kimi-k2.6",
"glm-5.3",
"glm-5.2",
)
@@ -837,11 +847,8 @@ def is_recommended_or_frontier_model(model_name: str) -> bool:
return False
if name in _RECOMMENDED_MODEL_NAME_SET:
return True
provider_name, bare_model_name = _split_model_provider(name)
return any(
_matches_frontier_family(provider_name, bare_model_name, provider_markers, prefixes)
for provider_markers, prefixes in FRONTIER_MODEL_FAMILIES
)
bare_model_name = name.rsplit("/", 1)[-1]
return _matches_model_prefix(bare_model_name, FRONTIER_MODEL_PREFIXES)
def _normalized_model_name(model_name: str) -> str:
@@ -853,28 +860,6 @@ def _normalized_model_name(model_name: str) -> str:
return name
def _split_model_provider(model_name: str) -> tuple[str | None, str]:
if "/" not in model_name:
return None, model_name
provider_name, bare_model_name = model_name.rsplit("/", 1)
return provider_name, bare_model_name
def _matches_frontier_family(
provider_name: str | None,
model_name: str,
provider_markers: tuple[str, ...],
model_prefixes: tuple[str, ...],
) -> bool:
if not _matches_model_prefix(model_name, model_prefixes):
return False
if provider_name is None:
return True
return _contains_provider_marker(
provider_name, provider_markers, split_compound_names=True
) or _contains_provider_marker(model_name, provider_markers)
def _matches_model_prefix(model_name: str, model_prefixes: tuple[str, ...]) -> bool:
return any(
candidate.startswith(prefix)
@@ -892,16 +877,6 @@ def _model_name_candidates(model_name: str) -> tuple[str, ...]:
return (model_name, *suffixes)
def _contains_provider_marker(
value: str, provider_markers: tuple[str, ...], *, split_compound_names: bool = False
) -> bool:
parts = set(value.replace(".", "/").split("/"))
if split_compound_names:
for separator in ("_", "-"):
parts.update(piece for part in tuple(parts) for piece in part.split(separator))
return any(marker in parts for marker in provider_markers)
def is_known_openai_bare_model(model_name: str) -> bool:
import litellm
+4
View File
@@ -112,6 +112,10 @@ class RuntimeSettings(BaseSettings):
backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
# Max screenshot/image tool outputs kept live per agent context (0 = none).
max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
# Seconds a running agent may go without emitting a single run event before
# its turn is abandoned and, if it still does not recover, a waiting parent
# marks it failed (0 = never).
agent_stall_timeout: int = Field(default=1800, ge=0, alias="STRIX_AGENT_STALL_TIMEOUT")
class TelemetrySettings(BaseSettings):
+52 -2
View File
@@ -6,6 +6,7 @@ import asyncio
import json
import logging
import tempfile
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast
@@ -45,6 +46,8 @@ class AgentRuntime:
wake: asyncio.Event = field(default_factory=asyncio.Event)
mailbox: list[dict[str, Any]] = field(default_factory=list)
user_wake_required: bool = False
# Monotonic time of the agent's last sign of life: a run event or a status change.
last_activity: float = field(default_factory=time.monotonic)
class AgentCoordinator:
@@ -170,7 +173,7 @@ class AgentCoordinator:
"task": task or "",
"skills": list(skills or []),
}
self.runtimes.setdefault(agent_id, AgentRuntime())
self.runtimes.setdefault(agent_id, AgentRuntime()).last_activity = time.monotonic()
logger.info("agent.register %s (%s) parent=%s", agent_id, name, parent_id or "-")
await self._maybe_snapshot()
@@ -200,10 +203,56 @@ class AgentCoordinator:
self.statuses[agent_id] = "running"
self.errors.pop(agent_id, None)
self.wait_kinds.pop(agent_id, None)
self.runtimes.setdefault(agent_id, AgentRuntime()).user_wake_required = False
runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
runtime.user_wake_required = False
runtime.last_activity = time.monotonic()
self._parent_notified.discard(agent_id)
await self._maybe_snapshot()
def touch(self, agent_id: str) -> None:
"""Record a sign of life; called on every run event, so it takes no lock."""
runtime = self.runtimes.get(agent_id)
if runtime is not None:
runtime.last_activity = time.monotonic()
async def reap_stalled(self, max_silence: float, *, under: str) -> list[dict[str, Any]]:
"""Fail the descendants of ``under`` that were silent for ``max_silence`` seconds.
A turn wedged somewhere no timeout covers never reaches a terminal status,
so whoever waits on it would wait forever. Cancelling the task ends the
wedged turn; the agent's own loop then delivers the terminal notice.
"""
if max_silence <= 0:
return []
now = time.monotonic()
reaped: list[dict[str, Any]] = []
tasks: list[asyncio.Task[Any]] = []
async with self._lock:
for aid in self._subtree_order_locked(under):
runtime = self.runtimes.get(aid)
if aid == under or runtime is None or self.statuses.get(aid) != "running":
continue
silence = now - runtime.last_activity
if silence < max_silence:
continue
error = f"agent produced no event for {silence:.0f}s; marked failed as stalled"
self.statuses[aid] = "failed"
self.errors[aid] = error
runtime.user_wake_required = True
runtime.last_activity = now
runtime.wake.set()
if runtime.task is not None and not runtime.task.done():
tasks.append(runtime.task)
reaped.append({"agent_id": aid, "name": self.names.get(aid, aid), "error": error})
for entry in reaped:
logger.warning("agent %s stalled: %s", entry["agent_id"], entry["error"])
logger.info("agent.status %s=failed", entry["agent_id"])
for task in tasks:
task.cancel()
if reaped:
await self._maybe_snapshot()
return reaped
async def park_waiting(self, agent_id: str, *, wait_kind: WaitKind) -> None:
"""Park an agent, recording what it is waiting on so the driver can time it."""
async with self._lock:
@@ -268,6 +317,7 @@ class AgentCoordinator:
self._parent_notified.discard(agent_id)
runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
runtime.user_wake_required = status in {"failed", "crashed"}
runtime.last_activity = time.monotonic()
runtime.wake.set()
logger.info("agent.status %s=%s", agent_id, status)
await self._maybe_snapshot()
+37 -3
View File
@@ -6,7 +6,7 @@ import asyncio
import contextlib
import logging
import uuid
from collections.abc import Callable
from collections.abc import AsyncIterator, Callable
from functools import cache
from typing import TYPE_CHECKING, Any, cast
@@ -19,7 +19,7 @@ from openai import (
APITimeoutError,
)
from strix.config import codex
from strix.config import codex, load_settings
from strix.core.hooks import (
BudgetExceededError,
BudgetPausedError,
@@ -149,6 +149,38 @@ def _transient_model_retry_delay(attempt: int) -> float:
return min(delay, _TRANSIENT_MODEL_RETRY_MAX_DELAY_S)
def _agent_stall_timeout() -> float:
return float(load_settings().runtime.agent_stall_timeout)
async def _with_stall_guard(stream: Any, timeout: float, agent_id: str) -> AsyncIterator[Any]:
"""Yield run events, abandoning the turn if none arrives for ``timeout`` seconds.
The model-stream idle guard only covers the model call itself. Everything
else the run loop awaits between events (tool transport, session writes,
a re-issued request that never opens) has no bound of its own, so a hang
there would keep the agent "running" forever.
"""
events = stream.stream_events()
if timeout <= 0:
async for event in events:
yield event
return
iterator = events.__aiter__()
while True:
try:
event = await asyncio.wait_for(iterator.__anext__(), timeout)
except StopAsyncIteration:
return
except TimeoutError:
message = f"agent turn produced no event for {timeout:.0f}s"
logger.warning("%s for %s; abandoning the turn", message, agent_id)
with contextlib.suppress(Exception):
stream.cancel(mode="immediate")
raise TimeoutError(message) from None
yield event
async def _salvage_stream_to_session(
session: Session,
pre_run_items: list[Any],
@@ -658,6 +690,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
image_strips = 0
compactions = 0
model_retries = 0
stall_timeout = _agent_stall_timeout()
while True:
stream: Any = None
pre_run_items: list[Any] = []
@@ -688,7 +721,8 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
await coordinator.attach_stream(agent_id, stream)
try:
try:
async for event in stream.stream_events():
async for event in _with_stall_guard(stream, stall_timeout, agent_id):
coordinator.touch(agent_id)
if event_sink is not None:
try:
event_sink(agent_id, event)
+1 -1
View File
@@ -192,7 +192,7 @@ class TuiController:
model_warning = ""
if model and not is_recommended_or_frontier_model(model):
model_warning = (
f"{model} is not a recommended frontier model; pentest quality could be degraded"
f"{model} is not a recommended frontier model. Pentest quality could be degraded."
)
state = {
"setup_mode": self.setup_mode,
@@ -370,6 +370,17 @@ func TestStartedSnapshotTransitionsToLiveView(t *testing.T) {
}
}
func TestSplashModelWarningRendersTheBackendSentenceOnce(t *testing.T) {
warning := "openai/glm-5.3 is not a recommended frontier model. Pentest quality could be degraded."
got := ansi.Strip(splashModelWarning("openai/glm-5.3", warning))
if got != "⚠ "+warning {
t.Fatalf("splash warning = %q, want %q", got, "⚠ "+warning)
}
if got := ansi.Strip(splashModelWarning("other/model", warning)); got != "⚠ "+warning {
t.Fatalf("splash warning with unrelated model = %q", got)
}
}
func TestSetupStartScreenFitsNarrowTerminal(t *testing.T) {
model := New(nil)
model.width, model.height = 40, 18
+10 -6
View File
@@ -411,7 +411,7 @@ func (m Model) splashView() string {
welcome + "\n" + version + "\n" + tagline + "\n\n" +
start.String() + "\n\n" + url
if warn := m.snapshot.ModelWarning; warn != "" {
content += "\n\n" + splashModelWarning(warn)
content += "\n\n" + splashModelWarning(m.snapshot.Model, warn)
}
panel := lipgloss.NewStyle().Border(lipgloss.RoundedBorder()).BorderForeground(green).Padding(1, 6).Align(lipgloss.Center).Render(content)
// #splash_screen background is solid black.
@@ -419,12 +419,16 @@ func (m Model) splashView() string {
lipgloss.WithWhitespaceBackground(black))
}
// splashModelWarning ports SplashScreen._build_model_warning_text.
func splashModelWarning(model string) string {
// splashModelWarning renders the backend's full warning sentence, with the
// model name highlighted when the sentence leads with it.
func splashModelWarning(model, warning string) string {
yellow := lipgloss.Color("#eab308")
return lipgloss.NewStyle().Bold(true).Foreground(yellow).Render("⚠ ") +
lipgloss.NewStyle().Bold(true).Foreground(render.Cyan).Render(model) +
lipgloss.NewStyle().Foreground(yellow).Render(" is not a recommended frontier model - pentest quality could be degraded")
out := lipgloss.NewStyle().Bold(true).Foreground(yellow).Render("⚠ ")
if model != "" && strings.HasPrefix(warning, model) {
out += lipgloss.NewStyle().Bold(true).Foreground(render.Cyan).Render(model)
warning = strings.TrimPrefix(warning, model)
}
return out + lipgloss.NewStyle().Foreground(yellow).Render(warning)
}
// chatPaneKey identifies everything the bordered trace depends on.
+25 -2
View File
@@ -12,7 +12,8 @@ from typing import Any, Literal, get_args
from agents import RunContextWrapper, function_tool
from strix.core.agents import Status, coordinator_from_context
from strix.config import load_settings
from strix.core.agents import AgentCoordinator, Status, coordinator_from_context
from strix.core.execution import notify_parent_on_terminal
from strix.core.hooks import LLM_TURN_KEY
from strix.report.state import get_global_report_state
@@ -291,6 +292,16 @@ _WAIT_DEFAULT_TIMEOUT_S = 300
# tool's own timeout fire first and return a clean result.
_WAIT_HARD_CEILING_S = _WAIT_DEFAULT_TIMEOUT_S + 1
_WAITED_TURN_KEY = "waited_llm_turn"
# Headroom past the per-turn stall guard so a wedged agent gets to abandon and
# replay its own turn before a waiting parent gives up on it.
_STALL_REAP_GRACE_S = 300.0
async def _reap_stalled_agents(coordinator: AgentCoordinator, me: str) -> list[dict[str, Any]]:
stall_timeout = float(load_settings().runtime.agent_stall_timeout)
if stall_timeout <= 0:
return []
return await coordinator.reap_stalled(stall_timeout + _STALL_REAP_GRACE_S, under=me)
@function_tool(timeout=_WAIT_HARD_CEILING_S)
@@ -422,6 +433,8 @@ async def wait_for_agents( # noqa: PLR0911
default=str,
)
stalled = await _reap_stalled_agents(coordinator, me)
# Non-interactive agents cannot be woken once terminal, so with nobody
# running or waiting there is no message left to wait for.
if not await coordinator.active_agents_except(me):
@@ -431,6 +444,7 @@ async def wait_for_agents( # noqa: PLR0911
"success": True,
"wait_outcome": "no_active_agents",
"reason": reason,
"stalled_agents": stalled,
"agents": [
{"agent_id": aid, "name": names.get(aid, aid), "status": status}
for aid, status in statuses.items()
@@ -452,13 +466,22 @@ async def wait_for_agents( # noqa: PLR0911
await asyncio.wait_for(coordinator.wait_for_message(me), timeout_seconds)
except TimeoutError:
await coordinator.mark_running(me)
stalled = await _reap_stalled_agents(coordinator, me)
return json.dumps(
{
"success": True,
"wait_outcome": "timeout",
"timeout_seconds": timeout_seconds,
"reason": reason,
"note": "No messages within timeout — continue work or call agent_finish.",
"stalled_agents": stalled,
"note": (
"No messages within timeout — continue work or call agent_finish."
if not stalled
else "No messages within timeout. The agents in stalled_agents produced "
"no output for too long and were marked failed; treat this list as "
"their failure notice. Do not wait on them again — continue work, "
"respawn what is still needed, or call agent_finish."
),
},
ensure_ascii=False,
default=str,
+258
View File
@@ -0,0 +1,258 @@
"""Tests for the per-agent stall guard.
The model-stream idle watchdog only covers the model call. A turn can also
wedge in what the run loop awaits *between* events — a re-issued request that
never opens, a tool transport that never answers — and nothing bounded that,
so the agent stayed ``running`` forever and its parent waited on it forever.
Two layers now cover it: the run cycle abandons a turn that emits no event for
the stall timeout, and a parent's ``wait_for_agents`` fails any agent that has
been silent for longer still.
"""
from __future__ import annotations
import asyncio
import json
import time
from typing import TYPE_CHECKING, Any, cast
import pytest
from agents import RunConfig, Runner
from agents.tool_context import ToolContext
from strix.core import execution
from strix.core.agents import AgentCoordinator
from strix.tools.agents_graph import tools as graph_tools
from strix.tools.agents_graph.tools import wait_for_agents
if TYPE_CHECKING:
from collections.abc import Iterator
class _HangingStream:
"""Emits one event, then never produces another."""
def __init__(self) -> None:
self.run_loop_exception: BaseException | None = None
self.cancelled = False
async def stream_events(self) -> Any:
yield "first"
await asyncio.Event().wait()
def cancel(self, mode: str = "immediate") -> None: # noqa: ARG002
self.cancelled = True
class _HealthyStream:
def __init__(self) -> None:
self.run_loop_exception: BaseException | None = None
async def stream_events(self) -> Any:
for i in range(3):
await asyncio.sleep(0.01)
yield f"event-{i}"
@pytest.fixture
def _fast_stall(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.setattr(execution, "_agent_stall_timeout", lambda: 0.2)
monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0)
monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0)
yield
def _serve(monkeypatch: pytest.MonkeyPatch, streams: list[Any]) -> dict[str, int]:
calls = {"n": 0}
def _fake_run_streamed(*_args: Any, **_kwargs: Any) -> Any:
stream = streams[calls["n"]]
calls["n"] += 1
return stream
monkeypatch.setattr(Runner, "run_streamed", _fake_run_streamed)
return calls
async def _run_cycle(coordinator: AgentCoordinator) -> Any:
return await execution._run_cycle(
object(),
coordinator,
"root",
input_data="task",
run_config=cast("RunConfig", object()),
context={},
max_turns=5,
session=None,
interactive=False,
event_sink=None,
hooks=None,
)
@pytest.mark.asyncio
@pytest.mark.usefixtures("_fast_stall")
async def test_hung_turn_is_abandoned_and_replayed(monkeypatch: pytest.MonkeyPatch) -> None:
hung = _HangingStream()
healthy = _HealthyStream()
calls = _serve(monkeypatch, [hung, healthy])
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
started = time.monotonic()
result = await _run_cycle(coordinator)
assert result is healthy
assert calls["n"] == 2
assert hung.cancelled is True
assert time.monotonic() - started < 2
@pytest.mark.asyncio
@pytest.mark.usefixtures("_fast_stall")
async def test_agent_that_never_recovers_is_marked_failed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_serve(
monkeypatch,
[_HangingStream() for _ in range(execution._MAX_TRANSIENT_MODEL_RETRIES + 1)],
)
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
with pytest.raises(TimeoutError, match="produced no event"):
await _run_cycle(coordinator)
assert coordinator.statuses["root"] == "crashed"
@pytest.mark.asyncio
async def test_stall_guard_is_off_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(execution, "_agent_stall_timeout", lambda: 0.0)
hung = _HangingStream()
_serve(monkeypatch, [hung])
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
with pytest.raises(TimeoutError):
await asyncio.wait_for(_run_cycle(coordinator), timeout=0.5)
assert hung.cancelled is False
# --- wait_for_agents reaps silent children ------------------------------------------
async def _call_wait(coordinator: AgentCoordinator, args: dict[str, Any]) -> dict[str, Any]:
ctx = ToolContext(
context={"coordinator": coordinator, "agent_id": "root"},
tool_name=wait_for_agents.name,
tool_call_id="call-1",
tool_arguments="{}",
)
raw: str = await wait_for_agents.on_invoke_tool(ctx, json.dumps(args))
return cast("dict[str, Any]", json.loads(raw))
@pytest.fixture
def _reap_after(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
async def _reap(coordinator: AgentCoordinator, me: str) -> list[dict[str, Any]]:
return await coordinator.reap_stalled(0.3, under=me)
monkeypatch.setattr(graph_tools, "_reap_stalled_agents", _reap)
yield
@pytest.mark.asyncio
@pytest.mark.usefixtures("_reap_after")
async def test_waiting_parent_fails_a_child_silent_too_long() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "ATO-Chaining", parent_id="root")
await coordinator.attach_runtime("root", resumable=False)
child_task = asyncio.create_task(asyncio.Event().wait())
await coordinator.attach_runtime("child", task=child_task, resumable=False)
result = await _call_wait(coordinator, {"timeout_seconds": 1})
assert result["wait_outcome"] == "timeout"
assert [a["agent_id"] for a in result["stalled_agents"]] == ["child"]
assert coordinator.statuses["child"] == "failed"
assert "stalled" in coordinator.errors["child"]
await asyncio.sleep(0)
assert child_task.cancelled()
# Nothing is left to wait on, so the next wait returns at once.
result = await _call_wait(coordinator, {"timeout_seconds": 60})
assert result["wait_outcome"] == "no_active_agents"
@pytest.mark.asyncio
@pytest.mark.usefixtures("_reap_after")
async def test_child_that_keeps_emitting_events_is_left_alone() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "DeepFuzz", parent_id="root")
await coordinator.attach_runtime("root", resumable=False)
await coordinator.attach_runtime("child", resumable=False)
async def _heartbeat() -> None:
for _ in range(12):
await asyncio.sleep(0.1)
coordinator.touch("child")
beat = asyncio.create_task(_heartbeat())
result = await _call_wait(coordinator, {"timeout_seconds": 1})
await beat
assert result["wait_outcome"] == "timeout"
assert result["stalled_agents"] == []
assert coordinator.statuses["child"] == "running"
@pytest.mark.asyncio
@pytest.mark.usefixtures("_reap_after")
async def test_waiting_children_are_not_reaped() -> None:
# A child parked in its own wait emits no events by design.
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "Coordinator", parent_id="root")
await coordinator.attach_runtime("root", resumable=False)
await coordinator.attach_runtime("child", resumable=False)
await coordinator.park_waiting("child", wait_kind="agents")
result = await _call_wait(coordinator, {"timeout_seconds": 1})
assert result["wait_outcome"] == "timeout"
assert result["stalled_agents"] == []
assert coordinator.statuses["child"] == "waiting"
@pytest.mark.asyncio
async def test_reaping_is_off_when_disabled() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "X", parent_id="root")
coordinator.runtimes["child"].last_activity -= 10_000
assert await coordinator.reap_stalled(0, under="root") == []
assert coordinator.statuses["child"] == "running"
@pytest.mark.asyncio
async def test_reaping_stays_inside_the_callers_subtree() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("a", "A", parent_id="root")
await coordinator.register("a1", "A1", parent_id="a")
await coordinator.register("b", "B", parent_id="root")
for aid in ("root", "a1", "b"):
coordinator.runtimes[aid].last_activity -= 10_000
reaped = await coordinator.reap_stalled(1, under="a")
assert [r["agent_id"] for r in reaped] == ["a1"]
assert coordinator.statuses["a1"] == "failed"
assert coordinator.statuses["b"] == "running"
assert coordinator.statuses["root"] == "running"
+9 -3
View File
@@ -79,6 +79,14 @@ def test_recommended_models_are_matched_case_insensitively() -> None:
"zai/glm-5.3-flash",
"openrouter/z-ai/glm-5.3",
"novita/zai-org/glm-5.2",
"openai/glm-5.3",
"openai/zai-org/glm-5.3",
"hosted_vllm/glm-5.3",
"openai/claude-opus-4-8",
"openai/deepseek-v4-pro",
"custom-ollama/gpt-5-mini-local",
"custom-provider/claude-opus-4-local",
"custom-provider/glm-5.3-local",
],
)
def test_frontier_model_families_are_accepted(model_name: str) -> None:
@@ -93,15 +101,13 @@ def test_frontier_model_families_are_accepted(model_name: str) -> None:
"anthropic/claude-3-5-sonnet-latest",
"ollama/llama3.1",
"deepseek/deepseek-chat",
"custom-ollama/gpt-5-mini-local",
"custom-provider/claude-opus-4-local",
"xai/grok-4.5",
"openrouter/x-ai/grok-4",
"mistral/mistral-medium-3-5",
"mistral/magistral-medium-latest",
"zai/glm-4.7",
"openai/glm-4.7",
"openrouter/z-ai/glm-5",
"custom-provider/glm-5.3-local",
],
)
def test_non_frontier_models_are_rejected(model_name: str) -> None: