diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index fee06152..1e54b734 100644 --- a/backend/apps/workflows/executor.py +++ b/backend/apps/workflows/executor.py @@ -469,7 +469,7 @@ async def execute( return run -async def _await_session_idle(session_id: str, run_id: Optional[str] = None, timeout_s: float = 600.0) -> str: +async def _await_session_idle(session_id: str, run_id: Optional[str] = None, idle_timeout_s: float = 1200.0) -> str: """Wait out the current step's agent turn. Returns a disposition: 'idle' turn finished, advance to the next step 'error' the agent session errored @@ -477,17 +477,24 @@ async def _await_session_idle(session_id: str, run_id: Optional[str] = None, tim For a real run (run_id given) a user PAUSE shows up as the session going 'stopped' WITHOUT a stop signal; that is not terminal, so we hold here - until Resume or Stop, keeping the step deadline fresh so a long pause + until Resume or Stop, keeping the idle deadline fresh so a long pause doesn't fail the step. The attended test-run driver passes no run_id and treats 'stopped' as terminal (no pause/resume there). Polls cheaply since agent_manager doesn't expose a per-session completion - future. Bounded by timeout_s so a stuck step can't hang the runner forever. + future. The deadline is idle-based, not a wall-clock cap on the step: any + agent activity (a committed message or a streamed text chunk growing + live_partial) resets it, so a legitimately long busy step never trips it + while a hung session still dies after idle_timeout_s of silence. A single + in-flight tool call commits nothing until its result lands, so + idle_timeout_s must stay >= the longest single-tool runtime (Bash caps at + 600s); don't lower it without special-casing in-flight tool calls. """ from backend.apps.agents.agent_manager import agent_manager hold_on_pause = run_id is not None - deadline = asyncio.get_event_loop().time() + timeout_s + last_activity: Optional[tuple] = None + deadline = asyncio.get_event_loop().time() + idle_timeout_s while True: if run_id is not None and _run_control.get(run_id) == "stop": return "stopped" @@ -498,8 +505,8 @@ async def _await_session_idle(session_id: str, run_id: Optional[str] = None, tim if status == "stopped": if not hold_on_pause: return "stopped" - # Paused. Hold, and reset the deadline so paused wall-time doesn't count against the step timeout. - deadline = asyncio.get_event_loop().time() + timeout_s + # Paused. Hold, and reset the deadline so paused wall-time doesn't count as idle. + deadline = asyncio.get_event_loop().time() + idle_timeout_s await asyncio.sleep(0.1) continue if status == "error": @@ -509,6 +516,22 @@ async def _await_session_idle(session_id: str, run_id: Optional[str] = None, tim task = agent_manager.tasks.get(session_id) if task is not None and task.done() and status not in ("running", "waiting_approval"): return "idle" + if status == "waiting_approval": + # A pending permission prompt is bounded by the approval flow's own ask_timeout; waiting on the user isn't idleness. + deadline = asyncio.get_event_loop().time() + idle_timeout_s + else: + msgs = getattr(sess, "messages", []) or [] + partial = agent_manager.live_partial.get(session_id) + # Timestamp catches in-place upserts (same id, fresh Message object); partial length catches mid-stream text between commits. + activity = ( + len(msgs), + msgs[-1].id if msgs else None, + msgs[-1].timestamp if msgs else None, + len(getattr(partial, "text", "") or ""), + ) + if activity != last_activity: + last_activity = activity + deadline = asyncio.get_event_loop().time() + idle_timeout_s if asyncio.get_event_loop().time() > deadline: - raise TimeoutError(f"Step exceeded {timeout_s}s on session {session_id}") + raise TimeoutError(f"No agent activity for {idle_timeout_s}s on session {session_id}") await asyncio.sleep(0.05) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index c5952c28..03dbaff1 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -126,6 +126,7 @@ class FakeAgentManager: def __init__(self): self.sessions: dict[str, SimpleNamespace] = {} self.tasks: dict[str, object] = {} + self.live_partial: dict[str, object] = {} self.launched_configs: list[object] = [] self.sent_messages: list[str] = [] # statuses[i] is the session status after the i-th send_message; absent entries default to 'completed'. cost_usd lands on the run. @@ -162,6 +163,7 @@ def fake_agent_manager(monkeypatch): fake = FakeAgentManager() monkeypatch.setattr(p_am.agent_manager, "sessions", fake.sessions) monkeypatch.setattr(p_am.agent_manager, "tasks", fake.tasks) + monkeypatch.setattr(p_am.agent_manager, "live_partial", fake.live_partial) monkeypatch.setattr(p_am.agent_manager, "launch_agent", fake.launch_agent) monkeypatch.setattr(p_am.agent_manager, "send_message", fake.send_message) monkeypatch.setattr(p_am.agent_manager, "close_session", fake.close_session) diff --git a/backend/tests/test_executor_pipeline.py b/backend/tests/test_executor_pipeline.py index 99869716..9c344609 100644 --- a/backend/tests/test_executor_pipeline.py +++ b/backend/tests/test_executor_pipeline.py @@ -15,6 +15,7 @@ from __future__ import annotations import asyncio from datetime import datetime, timedelta, timezone +from types import SimpleNamespace import pytest @@ -176,3 +177,61 @@ def test_resolved_config_uses_workflow_model_and_tools(make_wf, fake_agent_manag config = fake_agent_manager.launched_configs[0] assert config.model == "opus" assert config.allowed_tools == ["Read", "Grep"] + + +# --- idle-based step timeout --------------------------------------------------- + +def test_silent_running_session_times_out(fake_agent_manager): + """A session stuck in 'running' with no message or stream activity must + still die after idle_timeout_s, keeping the hung-step protection.""" + from backend.apps.workflows import executor + fake_agent_manager.sessions["s1"] = SimpleNamespace(id="s1", status="running", messages=[]) + with pytest.raises(TimeoutError, match="No agent activity"): + _run(executor._await_session_idle("s1", idle_timeout_s=0.3)) + + +def test_message_activity_defers_idle_timeout(fake_agent_manager): + """A busy step outliving idle_timeout_s must NOT be killed as long as new + messages keep landing; the deadline is idle-based, not a wall-clock cap.""" + from backend.apps.workflows import executor + sess = SimpleNamespace(id="s1", status="running", messages=[]) + fake_agent_manager.sessions["s1"] = sess + + async def p_drive(): + async def p_appender(): + # 10 ticks x 0.1s = 1.0s of activity, >3x the 0.3s idle window. + for i in range(10): + await asyncio.sleep(0.1) + sess.messages.append(SimpleNamespace(id=f"m{i}", timestamp=datetime.now())) + sess.status = "completed" + task = asyncio.ensure_future(p_appender()) + try: + return await executor._await_session_idle("s1", idle_timeout_s=0.3) + finally: + task.cancel() + + assert _run(p_drive()) == "idle" + + +def test_stream_partial_activity_defers_idle_timeout(fake_agent_manager): + """Mid-stream text growth (live_partial) counts as activity even when no + message has committed yet, so a long single response doesn't trip it.""" + from backend.apps.workflows import executor + sess = SimpleNamespace(id="s1", status="running", messages=[]) + fake_agent_manager.sessions["s1"] = sess + + async def p_drive(): + async def p_streamer(): + text = "" + for _ in range(10): + await asyncio.sleep(0.1) + text += "chunk " + fake_agent_manager.live_partial["s1"] = SimpleNamespace(text=text) + sess.status = "completed" + task = asyncio.ensure_future(p_streamer()) + try: + return await executor._await_session_idle("s1", idle_timeout_s=0.3) + finally: + task.cancel() + + assert _run(p_drive()) == "idle"