[eric] merge #120: idle-based workflow step timeout (pierre)

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)

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)

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"