From a9b7de9e1f8a21f88857c3eba5721c02decc720a Mon Sep 17 00:00:00 2001 From: Jason Robert Date: Mon, 24 Aug 2026 12:40:44 -0400 Subject: [PATCH 1/2] fix(providers): suppress idle watchdog during in-flight Copilot tool calls The Copilot SDK emits no events between tool.execution_start and tool.execution_complete, so a stale idle clock during a long-running tool call was indistinguishable from a genuinely stuck session, triggering a spurious "please continue" recovery prompt mid tool-call that could overwrite the agent's eventual structured output. In-flight tool calls (tracked by tool_call_id) now suppress idle recovery entirely while any remain outstanding; max_session_seconds / max_agent_iterations remain the backstop for a genuinely wedged tool. last_activity_ref's tool name is cleared (or rolled to another still-in-flight tool) on tool.execution_complete instead of only ever being set. Adds configurable runtime.idle_timeout_seconds and runtime.max_idle_recovery_attempts (Copilot-only) so workflows with legitimately long tool calls can tune the watchdog. Closes #488 --- CHANGELOG.md | 24 ++ docs/configuration.md | 14 + docs/workflow-syntax.md | 8 + .../conductor/references/yaml-schema.md | 2 + src/conductor/config/schema.py | 20 ++ src/conductor/providers/copilot.py | 69 ++++- src/conductor/providers/factory.py | 35 ++- src/conductor/providers/registry.py | 2 + tests/test_config/test_schema.py | 61 ++++ tests/test_providers/test_factory.py | 56 ++++ tests/test_providers/test_idle_recovery.py | 288 ++++++++++++++++++ tests/test_providers/test_registry.py | 2 + 12 files changed, 577 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 505b7afa..c975f449 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 A custom `base_url` requires an explicit `api_key`: an ambient `OPENAI_API_KEY` is never forwarded to a non-OpenAI endpoint. +- **`runtime.idle_timeout_seconds` / `runtime.max_idle_recovery_attempts`** + (#488) — Copilot-only knobs to tune the idle watchdog for workflows with + legitimately long tool calls. `idle_timeout_seconds` sets the time without + SDK events before a session is treated as idle (default 90s); + `max_idle_recovery_attempts` caps the number of "please continue" prompts + sent before failing (default 5; `0` fails on the first genuine idle + without ever injecting a prompt). See `docs/configuration.md` and + `docs/workflow-syntax.md`. + ### Changed - The Pydantic AI dependency was narrowed from the full `pydantic-ai` package to @@ -34,6 +43,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **The Copilot idle watchdog no longer fires during long-running tool + calls** (#488). The SDK emits no events between `tool.execution_start` and + `tool.execution_complete`, so a stale idle clock while a tool was still + executing was previously indistinguishable from a genuinely stuck + session — triggering a spurious "please continue" recovery prompt mid + tool-call. That prompt's conversational reply then overwrote the agent's + eventual structured output (`response_content` is last-message-wins), + turning a healthy run into a non-retryable failure. In-flight tool calls + (tracked by `tool_call_id`) now suppress idle recovery entirely while any + remain outstanding; `max_session_seconds` / `max_agent_iterations` remain + the backstop for a genuinely wedged tool. Recovery-prompt and + stuck-session messages also no longer misattribute the failure to a tool + that has already completed — `last_activity_ref`'s tool name is now + cleared (or rolled to another still-in-flight tool) on + `tool.execution_complete` instead of only ever being set. - Retry classification now covers the `ModelHTTPError` and `ModelAPIError` types pydantic-ai actually raises, so `408`, `429` and `5xx` responses are retried on the Claude provider as well as the new OpenAI one. Previously they were treated diff --git a/docs/configuration.md b/docs/configuration.md index df5491dd..e8606a96 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -17,6 +17,8 @@ workflow: max_tokens: 4096 default_reasoning_effort: medium # low | medium | high | xhigh | max (optional) default_context_tier: default # default | long_context (optional, Copilot only) + idle_timeout_seconds: 90 # Copilot only (optional) + max_idle_recovery_attempts: 5 # Copilot only (optional) # Provider-specific settings... ``` @@ -31,6 +33,16 @@ context-window tier that every provider-backed agent inherits unless it declares its own `context_tier` override. See [Context Tier](#context-tier) for details. This is a Copilot-only capability. +The `idle_timeout_seconds` and `max_idle_recovery_attempts` fields tune the +Copilot provider's idle watchdog: when a session stops emitting SDK events +for `idle_timeout_seconds` (default 90s), Conductor sends a "please continue" +recovery prompt, up to `max_idle_recovery_attempts` times (default 5) before +failing the session. A tool call that is still executing suppresses the +watchdog entirely — the SDK emits no events between the start and completion +of a tool call, so a stale idle clock during a long-running tool means the +tool is still running, not that the session is stuck. Both fields are +Copilot-only; other providers ignore them. + ## Provider Selection ### Copilot Provider @@ -47,6 +59,8 @@ workflow: command: npx args: ["-y", "open-websearch@latest"] tools: ["*"] + idle_timeout_seconds: 90 + max_idle_recovery_attempts: 5 ``` **Features**: diff --git a/docs/workflow-syntax.md b/docs/workflow-syntax.md index b99e51ca..356f5ef0 100644 --- a/docs/workflow-syntax.md +++ b/docs/workflow-syntax.md @@ -75,6 +75,14 @@ workflow: # its own `context_tier`. # See docs/configuration.md#context-tier. + idle_timeout_seconds: 90 # Optional: seconds without SDK events before a + # Copilot session is treated as idle (Copilot only). + # Default: 90. Suppressed entirely while a tool + # call is in flight. + max_idle_recovery_attempts: 5 # Optional: "please continue" prompts sent before + # failing an idle Copilot session (Copilot only). + # Default: 5. 0 means fail on first genuine idle. + working_dir: "/path/to/cwd" # Optional: global default working directory for LLM agents # and their MCP servers. Relative paths resolve against the # parent directory of the workflow YAML file. diff --git a/plugins/conductor/skills/conductor/references/yaml-schema.md b/plugins/conductor/skills/conductor/references/yaml-schema.md index 31b43505..3da0b81f 100644 --- a/plugins/conductor/skills/conductor/references/yaml-schema.md +++ b/plugins/conductor/skills/conductor/references/yaml-schema.md @@ -35,6 +35,8 @@ workflow: timeout: float # Per-request timeout in seconds (optional, default: 600, copilot/claude only) max_agent_iterations: integer # Max tool-use roundtrips per agent (1-500, optional) max_session_seconds: float # Wall-clock timeout per agent session in seconds (optional) + idle_timeout_seconds: float # Seconds without SDK events before session is idle (optional, Copilot only, default 90) + max_idle_recovery_attempts: integer # "please continue" prompts before failing idle session (optional, Copilot only, default 5) default_reasoning_effort: string # Workflow-wide reasoning/thinking effort: low, medium, high, xhigh, max (optional) skills: [string] # Skills enabled for every provider-backed agent (default: []) # Each entry is a built-in NAME or a filesystem PATH. diff --git a/src/conductor/config/schema.py b/src/conductor/config/schema.py index c9a8a891..0ff8ad03 100644 --- a/src/conductor/config/schema.py +++ b/src/conductor/config/schema.py @@ -3419,6 +3419,26 @@ def _coerce_provider(cls, value: Any) -> Any: (Claude: 50, Copilot: unlimited). """ + idle_timeout_seconds: float | None = Field(None, ge=1.0) + """Time without SDK events before a Copilot session is treated as idle. + + Copilot provider only; other providers ignore this field. Default is + None, which uses the provider's built-in default (90s). A session is + only considered idle when no SDK events at all have arrived within the + window — an in-flight tool call (between ``tool.execution_start`` and + ``tool.execution_complete``) suppresses the check entirely, since the + SDK emits no events while a tool is running. + """ + + max_idle_recovery_attempts: int | None = Field(None, ge=0) + """Maximum number of "please continue" prompts sent to an idle Copilot session. + + Copilot provider only; other providers ignore this field. Default is + None, which uses the provider's built-in default (5). ``0`` means the + session fails on the first genuine idle timeout without ever injecting + a recovery prompt. + """ + default_reasoning_effort: ReasoningEffort | None = None """Workflow-wide default reasoning effort applied to provider-backed agents. diff --git a/src/conductor/providers/copilot.py b/src/conductor/providers/copilot.py index ca9fbbd4..3337299b 100644 --- a/src/conductor/providers/copilot.py +++ b/src/conductor/providers/copilot.py @@ -153,7 +153,14 @@ class IdleRecoveryConfig: Attributes: idle_timeout_seconds: Time without any SDK events before considering session idle. + Settable via ``runtime.idle_timeout_seconds`` in workflow YAML (Copilot only). + The clock is suppressed entirely while a tool call is in flight (#488) — the + SDK emits no events between ``tool.execution_start`` and + ``tool.execution_complete``, so a stale clock there means the tool is still + running, not that the session is stuck. max_recovery_attempts: Maximum number of "continue" messages to send before failing. + Settable via ``runtime.max_idle_recovery_attempts`` in workflow YAML + (Copilot only). max_session_seconds: Hard wall-clock limit on total session duration. Prevents sessions from hanging indefinitely even if non-idle events keep flowing. recovery_prompt: Template for the recovery message sent to stuck sessions. @@ -1573,6 +1580,13 @@ async def _send_and_wait( # Mutable container for tool iteration counting tool_iteration_ref: list[int] = [0] + # In-flight tool calls, keyed by tool_call_id (falling back to the + # tool name when no id is available). The SDK emits no events + # between tool.execution_start and tool.execution_complete, so a + # non-empty dict here means a tool is genuinely still running, not + # that the session is stuck (#488). + active_tools: dict[str, str] = {} + def on_event(event: Any) -> None: nonlocal response_content, error_message event_type = event.type.value if hasattr(event.type, "value") else str(event.type) @@ -1654,6 +1668,26 @@ def on_event(event: Any) -> None: last_activity_ref[1] = tool_name # Count tool-use iterations tool_iteration_ref[0] += 1 + # Record the in-flight call so idle detection can tell "the + # tool is still running" from "the session is stuck" (#488). + # tool_call_id is a required field on real SDK events, but + # tests exercise this with Mock objects where getattr always + # succeeds without returning a string — guard with isinstance + # rather than trusting the attribute is present and usable. + tool_call_id = getattr(event.data, "tool_call_id", None) + key = tool_call_id if isinstance(tool_call_id, str) else str(tool_name) + active_tools[key] = str(tool_name) + elif event_type == "tool.execution_complete": + tool_call_id = getattr(event.data, "tool_call_id", None) + key = tool_call_id if isinstance(tool_call_id, str) else None + if key is not None and key in active_tools: + active_tools.pop(key, None) + elif active_tools: + # Unresolvable key (mismatched/mocked id) — pop the + # oldest in-flight entry so the set can still drain. + oldest_key = next(iter(active_tools)) + active_tools.pop(oldest_key, None) + last_activity_ref[1] = next(iter(active_tools.values()), None) # Forward structured events upstream via event_callback if event_callback is not None: @@ -1686,6 +1720,7 @@ def on_event(event: Any) -> None: max_agent_iterations=max_agent_iterations, interrupt_signal=interrupt_signal, agent_name=agent_name, + active_tools_ref=active_tools, ) if was_interrupted: # Return partial content (don't check error_message for partial) @@ -2120,6 +2155,7 @@ def _build_recovery_prompt( elif last_event_type: activity_map = { "tool.execution_start": "starting a tool call", + "tool.execution_complete": "finishing a tool call", "assistant.reasoning": "reasoning about the problem", "assistant.turn_start": "beginning a response", "assistant.message": "sending a message", @@ -2202,6 +2238,7 @@ async def _wait_with_idle_detection( max_agent_iterations: int | None = None, interrupt_signal: asyncio.Event | None = None, agent_name: str | None = None, + active_tools_ref: dict[str, str] | None = None, ) -> bool: """Wait for session completion with idle detection, recovery, and optional interrupt. @@ -2209,8 +2246,12 @@ async def _wait_with_idle_detection( with interrupt support (aborting on user request). When the model is actively working (SDK events flowing), the idle timer continuously resets — so stuck-detection is suppressed while the model is actively - working. The interrupt signal, however, is always raced regardless of - activity. + working. It is also suppressed while any tool call is in flight + (``active_tools_ref`` non-empty): the SDK emits no events between + ``tool.execution_start`` and ``tool.execution_complete``, so a stale + idle clock during a long-running tool call means the tool is still + running, not that the session is stuck (#488). The interrupt signal, + however, is always raced regardless of activity. Args: done: Event that signals session completion. @@ -2230,6 +2271,11 @@ async def _wait_with_idle_detection( logging so that idle-recovery messages emitted from concurrent for-each or parallel iterations can be attributed to a specific agent. ``None`` means no attribution tag. + active_tools_ref: Mutable dict of in-flight tool calls (id -> tool + name). A non-empty dict suppresses idle recovery entirely, + since the SDK emits no events while a tool is executing. + ``None`` (the default) preserves the pre-#488 behaviour of + every existing caller. Returns: True if interrupted, False if completed normally. @@ -2330,6 +2376,25 @@ async def _wait_with_idle_detection( return False # Completed successfully except TimeoutError as e: + # Timeout fired — but check if a tool call is still in + # flight first. The SDK emits no events between + # tool.execution_start and tool.execution_complete, so a + # stale idle clock here means the tool is still running + # (#488), not that the session is stuck. max_session_seconds + # / max_agent_iterations (checked at the top of this loop) + # remain the backstop for a genuinely wedged tool. + if active_tools_ref: + logger.debug( + "Idle timeout reached while tool(s) in flight%s: %s — suppressing " + "idle recovery", + f" agent={agent_name}" if agent_name else "", + ", ".join(active_tools_ref.values()), + ) + recovery_attempts = 0 + if not done.is_set(): + done.clear() + continue + # Timeout fired — but check if events were recently received. # The agent may be actively working (tool calls, reasoning) without # having reached session.idle yet. Only consider it stuck if no diff --git a/src/conductor/providers/factory.py b/src/conductor/providers/factory.py index 96154667..ea801247 100644 --- a/src/conductor/providers/factory.py +++ b/src/conductor/providers/factory.py @@ -59,6 +59,8 @@ async def create_provider( timeout: float | None = None, max_session_seconds: float | None = None, max_agent_iterations: int | None = None, + idle_timeout_seconds: float | None = None, + max_idle_recovery_attempts: int | None = None, default_reasoning_effort: ReasoningEffort | None = None, default_context_tier: ContextTier | None = None, provider_settings: ProviderSettings | None = None, @@ -83,6 +85,12 @@ async def create_provider( timeout: Request timeout in seconds. max_session_seconds: Maximum wall-clock duration for agent sessions. max_agent_iterations: Maximum tool-use iterations per agent execution. + idle_timeout_seconds: Time without SDK events before a Copilot session + is treated as idle. Copilot only; ``None`` uses the provider's + built-in default (90s). + max_idle_recovery_attempts: Maximum number of "please continue" + prompts sent to an idle Copilot session before failing. Copilot + only; ``None`` uses the provider's built-in default (5). default_reasoning_effort: Workflow-wide default reasoning effort (``low`` / ``medium`` / ``high`` / ``xhigh`` / ``max``) applied when an agent does not specify its own ``reasoning.effort``. @@ -113,9 +121,28 @@ async def create_provider( match provider_type: case "copilot": idle_recovery_config = None - if max_session_seconds is not None: + if ( + max_session_seconds is not None + or idle_timeout_seconds is not None + or max_idle_recovery_attempts is not None + ): + idle_recovery_defaults = IdleRecoveryConfig() idle_recovery_config = IdleRecoveryConfig( - max_session_seconds=max_session_seconds, + idle_timeout_seconds=( + idle_timeout_seconds + if idle_timeout_seconds is not None + else idle_recovery_defaults.idle_timeout_seconds + ), + max_recovery_attempts=( + max_idle_recovery_attempts + if max_idle_recovery_attempts is not None + else idle_recovery_defaults.max_recovery_attempts + ), + max_session_seconds=( + max_session_seconds + if max_session_seconds is not None + else idle_recovery_defaults.max_session_seconds + ), ) provider = CopilotProvider( mcp_servers=mcp_servers, @@ -334,6 +361,8 @@ async def create_provider( timeout = getattr(runtime_config, "timeout", None) max_session_seconds = getattr(runtime_config, "max_session_seconds", None) max_agent_iterations = getattr(runtime_config, "max_agent_iterations", None) + idle_timeout_seconds = getattr(runtime_config, "idle_timeout_seconds", None) + max_idle_recovery_attempts = getattr(runtime_config, "max_idle_recovery_attempts", None) default_reasoning_effort = getattr(runtime_config, "default_reasoning_effort", None) default_context_tier = getattr(runtime_config, "default_context_tier", None) tool_output = getattr(runtime_config, "tool_output", None) @@ -349,6 +378,8 @@ async def create_provider( timeout=timeout, max_session_seconds=max_session_seconds, max_agent_iterations=max_agent_iterations, + idle_timeout_seconds=idle_timeout_seconds, + max_idle_recovery_attempts=max_idle_recovery_attempts, default_reasoning_effort=default_reasoning_effort, default_context_tier=default_context_tier, provider_settings=provider_settings, diff --git a/src/conductor/providers/registry.py b/src/conductor/providers/registry.py index cddd7a7f..1d7cbbe2 100644 --- a/src/conductor/providers/registry.py +++ b/src/conductor/providers/registry.py @@ -125,6 +125,8 @@ async def _get_or_create_provider(self, provider_type: ProviderType) -> AgentPro timeout=runtime.timeout, max_session_seconds=runtime.max_session_seconds, max_agent_iterations=runtime.max_agent_iterations, + idle_timeout_seconds=runtime.idle_timeout_seconds, + max_idle_recovery_attempts=runtime.max_idle_recovery_attempts, default_reasoning_effort=runtime.default_reasoning_effort, provider_settings=provider_settings, tool_output=runtime.tool_output, diff --git a/tests/test_config/test_schema.py b/tests/test_config/test_schema.py index fd693a5f..270515fc 100644 --- a/tests/test_config/test_schema.py +++ b/tests/test_config/test_schema.py @@ -885,6 +885,67 @@ def test_serialization_includes_when_set(self) -> None: assert serialized["max_session_seconds"] == 90.0 +class TestRuntimeConfigIdleWatchdog: + """Tests for idle_timeout_seconds / max_idle_recovery_attempts on RuntimeConfig (#488).""" + + def test_idle_timeout_seconds_defaults_to_none(self) -> None: + """Test that idle_timeout_seconds defaults to None (provider default applies).""" + config = RuntimeConfig() + assert config.idle_timeout_seconds is None + + def test_idle_timeout_seconds_accepts_valid_value(self) -> None: + """Test that idle_timeout_seconds accepts a valid float value.""" + config = RuntimeConfig(idle_timeout_seconds=45.0) + assert config.idle_timeout_seconds == 45.0 + + def test_idle_timeout_seconds_rejects_below_minimum(self) -> None: + """Test that idle_timeout_seconds rejects values below 1.0.""" + with pytest.raises(ValidationError) as exc_info: + RuntimeConfig(idle_timeout_seconds=0.5) + assert "greater than or equal to 1" in str(exc_info.value) + + def test_idle_timeout_seconds_rejects_zero(self) -> None: + """Test that idle_timeout_seconds rejects zero.""" + with pytest.raises(ValidationError) as exc_info: + RuntimeConfig(idle_timeout_seconds=0) + assert "greater than or equal to 1" in str(exc_info.value) + + def test_max_idle_recovery_attempts_defaults_to_none(self) -> None: + """Test that max_idle_recovery_attempts defaults to None (provider default applies).""" + config = RuntimeConfig() + assert config.max_idle_recovery_attempts is None + + def test_max_idle_recovery_attempts_accepts_zero(self) -> None: + """Test that max_idle_recovery_attempts accepts 0 (fail on first idle timeout).""" + config = RuntimeConfig(max_idle_recovery_attempts=0) + assert config.max_idle_recovery_attempts == 0 + + def test_max_idle_recovery_attempts_accepts_positive_value(self) -> None: + """Test that max_idle_recovery_attempts accepts a positive int value.""" + config = RuntimeConfig(max_idle_recovery_attempts=10) + assert config.max_idle_recovery_attempts == 10 + + def test_max_idle_recovery_attempts_rejects_negative(self) -> None: + """Test that max_idle_recovery_attempts rejects negative values.""" + with pytest.raises(ValidationError) as exc_info: + RuntimeConfig(max_idle_recovery_attempts=-1) + assert "greater than or equal to 0" in str(exc_info.value) + + def test_serialization_excludes_both_when_none(self) -> None: + """Test that both new fields are excluded from serialization when None.""" + config = RuntimeConfig() + serialized = config.model_dump(exclude_none=True) + assert "idle_timeout_seconds" not in serialized + assert "max_idle_recovery_attempts" not in serialized + + def test_serialization_includes_both_when_set(self) -> None: + """Test that both new fields are included in serialization when set.""" + config = RuntimeConfig(idle_timeout_seconds=30.0, max_idle_recovery_attempts=1) + serialized = config.model_dump(exclude_none=True) + assert serialized["idle_timeout_seconds"] == 30.0 + assert serialized["max_idle_recovery_attempts"] == 1 + + class TestWorkflowDef: """Tests for WorkflowDef model.""" diff --git a/tests/test_providers/test_factory.py b/tests/test_providers/test_factory.py index cdeb8a87..643044f4 100644 --- a/tests/test_providers/test_factory.py +++ b/tests/test_providers/test_factory.py @@ -426,6 +426,62 @@ async def test_max_session_seconds_preserves_other_idle_recovery_defaults(self) assert provider._idle_recovery_config.max_recovery_attempts == 5 await provider.close() + @pytest.mark.asyncio + async def test_idle_timeout_seconds_alone_preserves_other_defaults(self) -> None: + """Setting only idle_timeout_seconds leaves the other two fields default.""" + provider = await create_provider("copilot", validate=False, idle_timeout_seconds=45.0) + assert isinstance(provider, CopilotProvider) + assert provider._idle_recovery_config.idle_timeout_seconds == 45.0 + assert provider._idle_recovery_config.max_recovery_attempts == 5 + assert provider._idle_recovery_config.max_session_seconds == 1800.0 + await provider.close() + + @pytest.mark.asyncio + async def test_max_idle_recovery_attempts_alone_preserves_other_defaults(self) -> None: + """Setting only max_idle_recovery_attempts leaves the other two fields default.""" + provider = await create_provider("copilot", validate=False, max_idle_recovery_attempts=2) + assert isinstance(provider, CopilotProvider) + assert provider._idle_recovery_config.max_recovery_attempts == 2 + assert provider._idle_recovery_config.idle_timeout_seconds == 90.0 + assert provider._idle_recovery_config.max_session_seconds == 1800.0 + await provider.close() + + @pytest.mark.asyncio + async def test_max_idle_recovery_attempts_zero_is_honored(self) -> None: + """0 means fail on first genuine idle without ever sending a recovery prompt.""" + provider = await create_provider("copilot", validate=False, max_idle_recovery_attempts=0) + assert isinstance(provider, CopilotProvider) + assert provider._idle_recovery_config.max_recovery_attempts == 0 + await provider.close() + + @pytest.mark.asyncio + async def test_all_three_idle_recovery_fields_together(self) -> None: + """All three fields set together are all forwarded.""" + provider = await create_provider( + "copilot", + validate=False, + max_session_seconds=60.0, + idle_timeout_seconds=15.0, + max_idle_recovery_attempts=3, + ) + assert isinstance(provider, CopilotProvider) + assert provider._idle_recovery_config.max_session_seconds == 60.0 + assert provider._idle_recovery_config.idle_timeout_seconds == 15.0 + assert provider._idle_recovery_config.max_recovery_attempts == 3 + await provider.close() + + @pytest.mark.asyncio + async def test_none_of_the_three_leaves_idle_recovery_config_at_provider_default( + self, + ) -> None: + """None of the three set leaves idle_recovery_config unset (provider builds its own).""" + provider = await create_provider("copilot", validate=False) + assert isinstance(provider, CopilotProvider) + assert provider._idle_recovery_config.idle_timeout_seconds == 90.0 + assert provider._idle_recovery_config.max_recovery_attempts == 5 + assert provider._idle_recovery_config.max_session_seconds == 1800.0 + await provider.close() + class TestClaudeAgentSdkFactoryRejections: """Factory rejects workflow features claude-agent-sdk does not honor (#241 / A2). diff --git a/tests/test_providers/test_idle_recovery.py b/tests/test_providers/test_idle_recovery.py index 4e7087b8..01164f48 100644 --- a/tests/test_providers/test_idle_recovery.py +++ b/tests/test_providers/test_idle_recovery.py @@ -2,6 +2,7 @@ import asyncio import time +import unittest.mock from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -13,6 +14,7 @@ _IDLE_IGNORED_EVENTS, CopilotProvider, IdleRecoveryConfig, + RetryConfig, ) @@ -850,3 +852,289 @@ async def set_done_after_delay(): # Should have sent at least one recovery message via idle detection assert mock_session.send.call_count >= 1 + + +class TestToolCallSuppression: + """Tests for #488: in-flight tool calls suppress idle recovery.""" + + @pytest.mark.asyncio + async def test_no_recovery_while_tool_in_flight(self) -> None: + """A non-empty active_tools_ref suppresses idle recovery entirely. + + The SDK emits no events between tool.execution_start and + tool.execution_complete, so a stale idle clock during a long-running + tool call must not trigger a recovery prompt. + """ + config = IdleRecoveryConfig( + idle_timeout_seconds=0.05, + max_recovery_attempts=2, + ) + provider = CopilotProvider( + mock_handler=stub_handler, + idle_recovery_config=config, + ) + + done = asyncio.Event() + mock_session = MagicMock() + mock_session.send = AsyncMock() + + # Deliberately stale — no activity update at all while the tool runs. + last_activity_ref: list[Any] = ["tool.execution_start", "read_agent", time.monotonic()] + active_tools_ref: dict[str, str] = {"call-1": "read_agent"} + + async def finish_after_in_flight_window() -> None: + # Stay in-flight for several multiples of the idle timeout, then + # signal completion while still in-flight (a tool call can + # legitimately finish and emit session.idle without an explicit + # tool.execution_complete order dependency in this test). + await asyncio.sleep(0.05 * 4) + done.set() + + await asyncio.gather( + provider._wait_with_idle_detection( + done=done, + session=mock_session, + verbose_enabled=False, + full_enabled=False, + last_activity_ref=last_activity_ref, + active_tools_ref=active_tools_ref, + ), + finish_after_in_flight_window(), + ) + + assert mock_session.send.call_count == 0 + + @pytest.mark.asyncio + async def test_recovery_fires_once_tool_dict_drains(self) -> None: + """Suppression is not a permanent disable — once the dict empties and + the clock is still stale, idle recovery resumes normally.""" + config = IdleRecoveryConfig( + idle_timeout_seconds=0.05, + max_recovery_attempts=10, + ) + provider = CopilotProvider( + mock_handler=stub_handler, + idle_recovery_config=config, + ) + + done = asyncio.Event() + mock_session = MagicMock() + mock_session.send = AsyncMock() + + last_activity_ref: list[Any] = ["tool.execution_start", "read_agent", time.monotonic()] + active_tools_ref: dict[str, str] = {"call-1": "read_agent"} + + async def drain_then_wait_for_recovery() -> None: + # Let a couple of idle windows pass while suppressed. + await asyncio.sleep(0.05 * 2) + assert mock_session.send.call_count == 0 + active_tools_ref.clear() + # Wait for a recovery attempt to actually fire, then finish. + while mock_session.send.call_count == 0: + await asyncio.sleep(0.01) + done.set() + + await asyncio.gather( + provider._wait_with_idle_detection( + done=done, + session=mock_session, + verbose_enabled=False, + full_enabled=False, + last_activity_ref=last_activity_ref, + active_tools_ref=active_tools_ref, + ), + drain_then_wait_for_recovery(), + ) + + assert mock_session.send.call_count >= 1 + + @pytest.mark.asyncio + async def test_max_session_seconds_still_raises_while_tool_in_flight(self) -> None: + """max_session_seconds remains the backstop even while a tool call + is in flight — the suppression only applies to idle recovery.""" + config = IdleRecoveryConfig( + idle_timeout_seconds=0.05, + max_recovery_attempts=10, + max_session_seconds=0.1, + ) + provider = CopilotProvider( + mock_handler=stub_handler, + idle_recovery_config=config, + ) + + done = asyncio.Event() # Never set + mock_session = MagicMock() + mock_session.send = AsyncMock() + + last_activity_ref: list[Any] = ["tool.execution_start", "read_agent", time.monotonic()] + active_tools_ref: dict[str, str] = {"call-1": "read_agent"} + + with pytest.raises(ProviderError) as exc_info: + await provider._wait_with_idle_detection( + done=done, + session=mock_session, + verbose_enabled=False, + full_enabled=False, + last_activity_ref=last_activity_ref, + active_tools_ref=active_tools_ref, + ) + + assert "exceeded maximum duration" in str(exc_info.value) + # No idle-recovery prompts were sent — only the wall-clock cap fired. + assert mock_session.send.call_count == 0 + + @pytest.mark.asyncio + async def test_none_active_tools_ref_preserves_existing_behavior(self) -> None: + """active_tools_ref=None (the default) behaves exactly like before #488.""" + config = IdleRecoveryConfig( + idle_timeout_seconds=0.05, + max_recovery_attempts=10, + ) + provider = CopilotProvider( + mock_handler=stub_handler, + idle_recovery_config=config, + ) + + done = asyncio.Event() + mock_session = MagicMock() + mock_session.send = AsyncMock() + + last_activity_ref: list[Any] = ["tool.execution_start", "read_agent", time.monotonic()] + + async def finish_after_first_recovery() -> None: + while mock_session.send.call_count == 0: + await asyncio.sleep(0.01) + done.set() + + await asyncio.gather( + provider._wait_with_idle_detection( + done=done, + session=mock_session, + verbose_enabled=False, + full_enabled=False, + last_activity_ref=last_activity_ref, + active_tools_ref=None, + ), + finish_after_first_recovery(), + ) + + assert mock_session.send.call_count >= 1 + + +class TestOnEventActiveTools: + """Tests for the on_event tracking of in-flight tool calls in _send_and_wait.""" + + @pytest.mark.asyncio + async def test_matching_start_and_complete_clears_last_tool_call(self) -> None: + """A tool.execution_start followed by its matching complete leaves + last_activity_ref[1] as None (no tool in flight).""" + from unittest.mock import Mock as _Mock + + provider = CopilotProvider(retry_config=RetryConfig(max_attempts=1)) + captured_cb: list[Any] = [] + captured_active_tools: dict[str, dict[str, str]] = {} + + start_ev = _Mock() + start_ev.type.value = "tool.execution_start" + start_ev.data.tool_name = "read_agent" + start_ev.data.tool_call_id = "call-1" + + complete_ev = _Mock() + complete_ev.type.value = "tool.execution_complete" + complete_ev.data.tool_call_id = "call-1" + + idle_ev = _Mock() + idle_ev.type.value = "session.idle" + + def on_event(callback: Any) -> None: + captured_cb.append(callback) + + session = _Mock() + session.on = on_event + + async def fake_send(prompt: str) -> None: + callback = captured_cb[0] + for ev in (start_ev, complete_ev, idle_ev): + callback(ev) + + session.send = fake_send + + # Patch _wait_with_idle_detection to capture active_tools_ref before + # returning, since _send_and_wait doesn't expose it directly. + original_wait = provider._wait_with_idle_detection + + async def spy_wait(*args: Any, **kwargs: Any) -> Any: + captured_active_tools["snapshot"] = dict(kwargs.get("active_tools_ref") or {}) + return await original_wait(*args, **kwargs) + + with unittest.mock.patch.object(provider, "_wait_with_idle_detection", spy_wait): + await provider._send_and_wait( + session=session, + prompt="hello", + verbose_enabled=False, + full_enabled=False, + ) + + # By the time _wait_with_idle_detection was invoked, send() had + # already fired every event synchronously, so the dict is empty. + assert captured_active_tools["snapshot"] == {} + + @pytest.mark.asyncio + async def test_second_tool_name_survives_first_complete(self) -> None: + """With two concurrent tool calls, completing one leaves the other's + name as last_activity_ref[1].""" + from unittest.mock import Mock as _Mock + + provider = CopilotProvider(retry_config=RetryConfig(max_attempts=1)) + captured_cb: list[Any] = [] + + start_ev_1 = _Mock() + start_ev_1.type.value = "tool.execution_start" + start_ev_1.data.tool_name = "read_agent" + start_ev_1.data.tool_call_id = "call-1" + + start_ev_2 = _Mock() + start_ev_2.type.value = "tool.execution_start" + start_ev_2.data.tool_name = "bash" + start_ev_2.data.tool_call_id = "call-2" + + complete_ev_1 = _Mock() + complete_ev_1.type.value = "tool.execution_complete" + complete_ev_1.data.tool_call_id = "call-1" + + idle_ev = _Mock() + idle_ev.type.value = "session.idle" + + def on_event(callback: Any) -> None: + captured_cb.append(callback) + + session = _Mock() + session.on = on_event + + async def fake_send(prompt: str) -> None: + callback = captured_cb[0] + for ev in (start_ev_1, start_ev_2, complete_ev_1): + callback(ev) + callback(idle_ev) + + session.send = fake_send + + original_wait = provider._wait_with_idle_detection + captured_ref: dict[str, Any] = {} + + async def spy_wait(*args: Any, **kwargs: Any) -> Any: + captured_ref["last_activity_ref"] = kwargs.get("last_activity_ref") or ( + args[4] if len(args) > 4 else None + ) + return await original_wait(*args, **kwargs) + + with unittest.mock.patch.object(provider, "_wait_with_idle_detection", spy_wait): + await provider._send_and_wait( + session=session, + prompt="hello", + verbose_enabled=False, + full_enabled=False, + ) + + last_activity_ref = captured_ref["last_activity_ref"] + assert last_activity_ref[1] == "bash" diff --git a/tests/test_providers/test_registry.py b/tests/test_providers/test_registry.py index 43af9efb..ad99169b 100644 --- a/tests/test_providers/test_registry.py +++ b/tests/test_providers/test_registry.py @@ -340,6 +340,8 @@ async def test_runtime_config_passed_to_provider(self, mock_create: MagicMock) - timeout=60.0, max_session_seconds=None, max_agent_iterations=None, + idle_timeout_seconds=None, + max_idle_recovery_attempts=None, default_reasoning_effort="high", provider_settings=config.workflow.runtime.provider, tool_output=config.workflow.runtime.tool_output, From a7e6e58e09d9c51e089742031e3804a790a5a2b6 Mon Sep 17 00:00:00 2001 From: Jason Robert Date: Mon, 24 Aug 2026 13:32:17 -0400 Subject: [PATCH 2/2] fix(providers): address PR #490 review findings on idle watchdog suppression MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Blocking fixes (#488): - Remove the "pop the oldest entry" fallback in the tool.execution_complete handler that could evict a different, still-running tool's active_tools entry on a duplicate/unmatched event, re-arming the watchdog mid-tool-call and reproducing #488 while appearing fixed. Replaced with a non-mutating debug log; max_session_seconds remains the backstop for a stale entry. - Strengthen TestOnEventActiveTools assertions so both tests actually pin the fix (verified to fail against the pre-fix provider, pass post-fix). - Add an end-to-end regression test driving a real tool.execution_start -> silence -> tool.execution_complete sequence through _send_and_wait, asserting the recovery prompt never clobbers response_content. - Add a warn-once latch (mirroring _context_window_anomaly_warned) so the first occurrence of extended idle-recovery suppression during a session is logged at warning level (console + logger), instead of silently degrading a previously console-visible 90s warning into up to 31.5 minutes of total silence. Recommendations applied: - Corrected the repeated false claim that the SDK "emits no events" during a tool call (it does not guarantee any, but tool.execution_progress / tool.execution_partial_result exist and are opt-in) across copilot.py, schema.py, docs/configuration.md, CHANGELOG.md, and the PR description; consolidated the rationale into one canonical docstring. - Corrected the inaccurate claim that max_agent_iterations backstops a wedged tool call (its counter only advances on tool.execution_start, so it's frozen for the whole wedge) — max_session_seconds is the sole backstop. - Added IdleRecoveryConfig.__post_init__ validation so directly-constructed configs (bypassing the Pydantic schema bounds) can't produce an unbounded busy-wait loop. - Simplified factory.py's IdleRecoveryConfig construction to a dict-filter + single constructor call instead of a three-way ternary per field. - Added ProviderCapabilities.idle_recovery (Copilot-only) with a workflow-level validator warning (not an error, since these are tuning knobs rather than safety bounds) when idle_timeout_seconds / max_idle_recovery_attempts are set against a provider that ignores them. - Bounded two previously-unbounded busy-wait test loops with asyncio.wait_for(..., timeout=5.0). - Added an overlapping-tool-calls end-to-end test keyed on tool_call_id (not tool_name), verified to reproduce the hang if the dict were mistakenly keyed by tool name instead. - Documented the max_session_seconds backstop in docs/configuration.md so a legitimately long tool call doesn't silently exceed it unexpectedly. Skipped: ACA forwarding of the two idle-recovery fields (larger, separate scope spanning factory/aca/aca_runner) and AGENTS.md documentation update (the two runtime knobs and active_tools mechanism are already documented in the config docs and code comments; deferring to keep this diff scoped to the review findings). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 28 +- docs/configuration.md | 12 +- src/conductor/config/schema.py | 6 +- src/conductor/config/validator.py | 32 +++ src/conductor/providers/capabilities.py | 11 + src/conductor/providers/copilot.py | 161 ++++++++++-- src/conductor/providers/factory.py | 34 +-- .../test_validator_capabilities.py | 49 ++++ tests/test_providers/test_capabilities.py | 1 + tests/test_providers/test_idle_recovery.py | 245 +++++++++++++++++- 10 files changed, 507 insertions(+), 72 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c975f449..1612d99d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,20 +44,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - **The Copilot idle watchdog no longer fires during long-running tool - calls** (#488). The SDK emits no events between `tool.execution_start` and - `tool.execution_complete`, so a stale idle clock while a tool was still - executing was previously indistinguishable from a genuinely stuck - session — triggering a spurious "please continue" recovery prompt mid - tool-call. That prompt's conversational reply then overwrote the agent's - eventual structured output (`response_content` is last-message-wins), - turning a healthy run into a non-retryable failure. In-flight tool calls - (tracked by `tool_call_id`) now suppress idle recovery entirely while any - remain outstanding; `max_session_seconds` / `max_agent_iterations` remain - the backstop for a genuinely wedged tool. Recovery-prompt and + calls** (#488). The SDK does not guarantee any events during a tool + call — `tool.execution_progress` / `tool.execution_partial_result` exist + in the SDK schema but are opt-in per tool, so for most tool calls nothing + arrives between `tool.execution_start` and `tool.execution_complete` — so + a stale idle clock while a tool was still executing was previously + indistinguishable from a genuinely stuck session — triggering a spurious + "please continue" recovery prompt mid tool-call. That prompt's + conversational reply then overwrote the agent's eventual structured + output (`response_content` is last-message-wins), turning a healthy run + into a non-retryable failure. In-flight tool calls (tracked by + `tool_call_id`) now suppress idle recovery entirely while any remain + outstanding; `max_session_seconds` is the sole backstop for a genuinely + wedged tool. Recovery-prompt and stuck-session messages also no longer misattribute the failure to a tool that has already completed — `last_activity_ref`'s tool name is now cleared (or rolled to another still-in-flight tool) on - `tool.execution_complete` instead of only ever being set. + `tool.execution_complete` instead of only ever being set. The first + occurrence of extended suppression during a session is logged at + `warning` level (naming the in-flight tools and the `max_session_seconds` + backstop); further occurrences in the same session are debug-only. - Retry classification now covers the `ModelHTTPError` and `ModelAPIError` types pydantic-ai actually raises, so `408`, `429` and `5xx` responses are retried on the Claude provider as well as the new OpenAI one. Previously they were treated diff --git a/docs/configuration.md b/docs/configuration.md index e8606a96..b2b5e1cf 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -38,10 +38,14 @@ Copilot provider's idle watchdog: when a session stops emitting SDK events for `idle_timeout_seconds` (default 90s), Conductor sends a "please continue" recovery prompt, up to `max_idle_recovery_attempts` times (default 5) before failing the session. A tool call that is still executing suppresses the -watchdog entirely — the SDK emits no events between the start and completion -of a tool call, so a stale idle clock during a long-running tool means the -tool is still running, not that the session is stuck. Both fields are -Copilot-only; other providers ignore them. +watchdog — most tools emit nothing while running, so a stale idle clock +during a long-running tool call usually means the tool is still running, not +that the session is stuck. Suppression is bounded by `max_session_seconds` +(default 1800s), which is still enforced while a tool is in flight and is the +only limit that can end a genuinely hung tool call. Raise it alongside +`idle_timeout_seconds` if your workflow has tool calls that legitimately run +longer than 30 minutes. Both `idle_timeout_seconds` and +`max_idle_recovery_attempts` are Copilot-only; other providers ignore them. ## Provider Selection diff --git a/src/conductor/config/schema.py b/src/conductor/config/schema.py index 0ff8ad03..fa2bd643 100644 --- a/src/conductor/config/schema.py +++ b/src/conductor/config/schema.py @@ -3426,8 +3426,10 @@ def _coerce_provider(cls, value: Any) -> Any: None, which uses the provider's built-in default (90s). A session is only considered idle when no SDK events at all have arrived within the window — an in-flight tool call (between ``tool.execution_start`` and - ``tool.execution_complete``) suppresses the check entirely, since the - SDK emits no events while a tool is running. + ``tool.execution_complete``) suppresses the check entirely, since most + tools emit nothing during execution (see + ``IdleRecoveryConfig.idle_timeout_seconds`` in ``providers/copilot.py`` + for the full rationale). """ max_idle_recovery_attempts: int | None = Field(None, ge=0) diff --git a/src/conductor/config/validator.py b/src/conductor/config/validator.py index 9143fd29..cd55294d 100644 --- a/src/conductor/config/validator.py +++ b/src/conductor/config/validator.py @@ -1821,6 +1821,8 @@ def _validate_provider_capabilities( # no matter where future call sites are added. runtime_default_effort = config.workflow.runtime.default_reasoning_effort runtime_max_session_seconds = config.workflow.runtime.max_session_seconds + runtime_idle_timeout_seconds = config.workflow.runtime.idle_timeout_seconds + runtime_max_idle_recovery_attempts = config.workflow.runtime.max_idle_recovery_attempts runtime_working_dir = config.workflow.runtime.working_dir runtime_skills = config.workflow.runtime.skills skill_limits = config.workflow.runtime.skill_injection @@ -2466,6 +2468,36 @@ def _check_agent_capabilities( f"max_session_seconds." ) + # ----- Workflow-level: idle_recovery tuning knobs ----- + # runtime.idle_timeout_seconds / runtime.max_idle_recovery_attempts are + # Copilot-only tuning knobs (#488). Unlike max_session_seconds, these are + # not safety bounds — a provider that ignores them just runs its own + # idle-detection defaults (or none at all) rather than violating an + # operational guarantee — so a mismatch is a warning, not an error. + if runtime_idle_timeout_seconds is not None or runtime_max_idle_recovery_attempts is not None: + providers_using_idle_recovery: dict[str, list[str]] = {} + for agent in all_llm_agents: + pname = _resolved_provider_name(agent, default_provider) + providers_using_idle_recovery.setdefault(pname, []).append(agent.name) + for pname, agent_names in providers_using_idle_recovery.items(): + pcaps = _caps_for(pname) + if pcaps is not None and not pcaps.idle_recovery: + set_fields = [ + name + for name, value in ( + ("idle_timeout_seconds", runtime_idle_timeout_seconds), + ("max_idle_recovery_attempts", runtime_max_idle_recovery_attempts), + ) + if value is not None + ] + warnings.append( + f"Workflow declares 'runtime.{'/'.join(set_fields)}' but provider " + f"'{pname}' does not support idle-recovery tuning " + f"(capabilities.idle_recovery=False) and is used by agent(s): " + f"{sorted(agent_names)!r}. The setting will be silently ignored " + f"for these agents." + ) + # ----- Workflow-level: working_dir ----- # A runtime-wide working_dir is inherited by every LLM agent that does # not set its own. A provider that cannot apply it would silently run diff --git a/src/conductor/providers/capabilities.py b/src/conductor/providers/capabilities.py index fe87734b..8a59d5c1 100644 --- a/src/conductor/providers/capabilities.py +++ b/src/conductor/providers/capabilities.py @@ -192,6 +192,15 @@ class ProviderCapabilities(BaseModel): session_continuity: bool = False """``True`` when the provider supports per-agent ``session_key``.""" + idle_recovery: bool = False + """``True`` when the provider honors ``runtime.idle_timeout_seconds`` / + ``runtime.max_idle_recovery_attempts`` (#488). These are Copilot-only + tuning knobs for its SDK-event-driven idle watchdog; other providers + have no equivalent mechanism and silently ignore both fields. Unlike + ``max_session_seconds``, this is a tuning knob rather than a safety + bound, so a mismatch is a validate-time **warning**, not an error. + Defaults to ``False``.""" + max_temperature: float | None = None """Highest temperature the provider accepts. @@ -274,6 +283,8 @@ def declared_limitations(self) -> list[str]: items.append("no skills support") if not self.session_continuity: items.append("no session_key continuity") + if not self.idle_recovery: + items.append("idle_timeout_seconds/max_idle_recovery_attempts ignored") return items diff --git a/src/conductor/providers/copilot.py b/src/conductor/providers/copilot.py index 3337299b..abeb359c 100644 --- a/src/conductor/providers/copilot.py +++ b/src/conductor/providers/copilot.py @@ -154,10 +154,14 @@ class IdleRecoveryConfig: Attributes: idle_timeout_seconds: Time without any SDK events before considering session idle. Settable via ``runtime.idle_timeout_seconds`` in workflow YAML (Copilot only). - The clock is suppressed entirely while a tool call is in flight (#488) — the - SDK emits no events between ``tool.execution_start`` and - ``tool.execution_complete``, so a stale clock there means the tool is still - running, not that the session is stuck. + The clock is suppressed entirely while a tool call is in flight (#488). The + SDK does not guarantee any events during a tool call — ``tool.execution_progress`` + and ``tool.execution_partial_result`` exist in the SDK schema but are opt-in + per tool, so for most tool calls nothing arrives between + ``tool.execution_start`` and ``tool.execution_complete``. A stale clock in + that window therefore means the tool is still running, not that the session + is stuck. ``max_session_seconds`` is the sole backstop for a genuinely + wedged tool. max_recovery_attempts: Maximum number of "continue" messages to send before failing. Settable via ``runtime.max_idle_recovery_attempts`` in workflow YAML (Copilot only). @@ -176,6 +180,26 @@ class IdleRecoveryConfig: "Please continue with your task from where you left off." ) + def __post_init__(self) -> None: + """Validate fields on direct construction. + + The Pydantic bounds on the corresponding ``runtime.*`` schema fields + only guard the YAML path; ``CopilotProvider(idle_recovery_config=...)`` + is a documented, directly-constructible path (used throughout the + test suite) that bypasses them entirely. Without this, an invalid + value like ``idle_timeout_seconds=0`` turns the new + tool-in-flight-suppression branch into an unbounded busy-wait for + the full ``max_session_seconds``. + """ + if self.idle_timeout_seconds <= 0: + raise ValueError(f"idle_timeout_seconds must be > 0, got {self.idle_timeout_seconds}") + if self.max_recovery_attempts < 0: + raise ValueError( + f"max_recovery_attempts must be >= 0, got {self.max_recovery_attempts}" + ) + if self.max_session_seconds <= 0: + raise ValueError(f"max_session_seconds must be > 0, got {self.max_session_seconds}") + @dataclass class SDKResponse: @@ -278,6 +302,9 @@ class CopilotProvider(AgentProvider): # ``mcp_servers`` for MCP. The SDK's ``plugin_directories`` is # deliberately unused — see conductor.plugins for why. plugins=True, + # ``runtime.idle_timeout_seconds`` / ``runtime.max_idle_recovery_attempts`` + # tune the IdleRecoveryConfig-backed watchdog (#488). + idle_recovery=True, max_temperature=1.0, upstream_pin=None, maintainer="@microsoft/conductor", @@ -407,6 +434,12 @@ def __init__( self._provider_settings = provider_settings self._tool_output_config = tool_output or ToolOutputConfig() self._github_token = github_token + # Warn-once latch for extended idle-recovery suppression while a + # tool call is in flight (#488) — mirrors the + # `_pricing_hook_failed_warned` / `_context_window_anomaly_warned` + # pattern in engine/workflow.py: "a debug-only record would never + # reach an operator." + self._tool_suppression_warned = False self._warn_custom_routing_default_model() @staticmethod @@ -1581,10 +1614,10 @@ async def _send_and_wait( tool_iteration_ref: list[int] = [0] # In-flight tool calls, keyed by tool_call_id (falling back to the - # tool name when no id is available). The SDK emits no events - # between tool.execution_start and tool.execution_complete, so a - # non-empty dict here means a tool is genuinely still running, not - # that the session is stuck (#488). + # tool name when no id is available). A non-empty dict here means a + # tool is genuinely still running, not that the session is stuck + # (#488) — see IdleRecoveryConfig.idle_timeout_seconds docstring for + # why a stale clock isn't reliable here. active_tools: dict[str, str] = {} def on_event(event: Any) -> None: @@ -1682,11 +1715,20 @@ def on_event(event: Any) -> None: key = tool_call_id if isinstance(tool_call_id, str) else None if key is not None and key in active_tools: active_tools.pop(key, None) - elif active_tools: - # Unresolvable key (mismatched/mocked id) — pop the - # oldest in-flight entry so the set can still drain. - oldest_key = next(iter(active_tools)) - active_tools.pop(oldest_key, None) + elif isinstance(tool_call_id, str): + # Unrecognized tool_call_id (duplicate or unmatched + # completion event — this SDK re-delivers events, see + # seen_call_ids above). Do NOT guess which in-flight + # entry to evict: popping the wrong one would re-arm the + # idle watchdog while a different tool is still running, + # reproducing #488. Leaving a stale entry is safe because + # max_session_seconds backstops it. + logger.debug( + "tool.execution_complete for untracked tool_call_id %r — " + "ignoring; in-flight: %s", + tool_call_id, + sorted(active_tools), + ) last_activity_ref[1] = next(iter(active_tools.values()), None) # Forward structured events upstream via event_callback @@ -2226,6 +2268,47 @@ def _log_recovery_attempt( console.print(text) + def _log_tool_suppression_warning( + self, + in_flight_tools: str, + elapsed: float, + max_session: float, + agent_name: str | None = None, + ) -> None: + """Log the first occurrence of idle-recovery suppression to the console. + + Mirrors ``_log_recovery_attempt``'s console output, but distinct from + it: this fires once per provider lifetime (see + ``_tool_suppression_warned``) rather than per recovery attempt, since + a genuinely long tool call can suppress idle recovery for many idle + windows in a row and a repeated console line would just be noise. + + Args: + in_flight_tools: Comma-joined names of the currently in-flight tools. + elapsed: Seconds the session has been running so far. + max_session: The wall-clock session limit that still applies. + agent_name: Optional agent identifier for attribution. + """ + from rich.text import Text + + from conductor.console import make_console + + console = make_console(stderr=True, highlight=False) + + text = Text() + text.append(" ├─ ", style="dim") + if agent_name: + text.append(f"[{agent_name}] ", style="magenta") + text.append("⚠️ ", style="yellow") + text.append("Idle recovery suppressed", style="yellow bold") + text.append( + f" - tool(s) in flight: '{in_flight_tools}', running {elapsed:.0f}s, " + f"will hard-fail at max_session_seconds={max_session:.0f}s", + style="dim italic", + ) + + console.print(text) + async def _wait_with_idle_detection( self, done: asyncio.Event, @@ -2247,11 +2330,11 @@ async def _wait_with_idle_detection( actively working (SDK events flowing), the idle timer continuously resets — so stuck-detection is suppressed while the model is actively working. It is also suppressed while any tool call is in flight - (``active_tools_ref`` non-empty): the SDK emits no events between - ``tool.execution_start`` and ``tool.execution_complete``, so a stale - idle clock during a long-running tool call means the tool is still - running, not that the session is stuck (#488). The interrupt signal, - however, is always raced regardless of activity. + (``active_tools_ref`` non-empty): a stale idle clock during a + long-running tool call means the tool is still running, not that the + session is stuck (#488; see ``IdleRecoveryConfig.idle_timeout_seconds`` + docstring for why events can't be relied on during a tool call). The + interrupt signal, however, is always raced regardless of activity. Args: done: Event that signals session completion. @@ -2272,8 +2355,8 @@ async def _wait_with_idle_detection( for-each or parallel iterations can be attributed to a specific agent. ``None`` means no attribution tag. active_tools_ref: Mutable dict of in-flight tool calls (id -> tool - name). A non-empty dict suppresses idle recovery entirely, - since the SDK emits no events while a tool is executing. + name). A non-empty dict suppresses idle recovery entirely + (see ``IdleRecoveryConfig.idle_timeout_seconds`` docstring). ``None`` (the default) preserves the pre-#488 behaviour of every existing caller. @@ -2377,12 +2460,18 @@ async def _wait_with_idle_detection( except TimeoutError as e: # Timeout fired — but check if a tool call is still in - # flight first. The SDK emits no events between - # tool.execution_start and tool.execution_complete, so a - # stale idle clock here means the tool is still running - # (#488), not that the session is stuck. max_session_seconds - # / max_agent_iterations (checked at the top of this loop) - # remain the backstop for a genuinely wedged tool. + # flight first. The SDK does not guarantee any events during + # a tool call — tool.execution_progress and + # tool.execution_partial_result exist in the schema but are + # opt-in per tool, so for most tool calls nothing arrives + # between start and complete (see IdleRecoveryConfig + # docstring). A stale idle clock here therefore means the + # tool is still running (#488), not that the session is + # stuck. max_session_seconds (checked at the top of this + # loop) is the sole backstop for a genuinely wedged tool — + # max_agent_iterations cannot help, since its counter only + # advances on a new tool.execution_start and is frozen for + # the entire duration of a wedge. if active_tools_ref: logger.debug( "Idle timeout reached while tool(s) in flight%s: %s — suppressing " @@ -2390,6 +2479,26 @@ async def _wait_with_idle_detection( f" agent={agent_name}" if agent_name else "", ", ".join(active_tools_ref.values()), ) + if not self._tool_suppression_warned: + self._tool_suppression_warned = True + logger.warning( + "Idle recovery suppressed while tool(s) in flight%s: %s. " + "Suppression has been in effect for %.0fs; if the tool(s) " + "are genuinely stuck, the session will hard-fail once " + "max_session_seconds (%.0fs) is exceeded. Further " + "occurrences are logged at debug level.", + f" agent={agent_name}" if agent_name else "", + ", ".join(active_tools_ref.values()), + elapsed, + max_session, + ) + if verbose_enabled: + self._log_tool_suppression_warning( + in_flight_tools=", ".join(active_tools_ref.values()), + elapsed=elapsed, + max_session=max_session, + agent_name=agent_name, + ) recovery_attempts = 0 if not done.is_set(): done.clear() diff --git a/src/conductor/providers/factory.py b/src/conductor/providers/factory.py index ea801247..e4f2fbcb 100644 --- a/src/conductor/providers/factory.py +++ b/src/conductor/providers/factory.py @@ -120,30 +120,18 @@ async def create_provider( match provider_type: case "copilot": - idle_recovery_config = None - if ( - max_session_seconds is not None - or idle_timeout_seconds is not None - or max_idle_recovery_attempts is not None - ): - idle_recovery_defaults = IdleRecoveryConfig() - idle_recovery_config = IdleRecoveryConfig( - idle_timeout_seconds=( - idle_timeout_seconds - if idle_timeout_seconds is not None - else idle_recovery_defaults.idle_timeout_seconds - ), - max_recovery_attempts=( - max_idle_recovery_attempts - if max_idle_recovery_attempts is not None - else idle_recovery_defaults.max_recovery_attempts - ), - max_session_seconds=( - max_session_seconds - if max_session_seconds is not None - else idle_recovery_defaults.max_session_seconds - ), + idle_recovery_overrides: dict[str, Any] = { + k: v + for k, v in ( + ("idle_timeout_seconds", idle_timeout_seconds), + ("max_recovery_attempts", max_idle_recovery_attempts), + ("max_session_seconds", max_session_seconds), ) + if v is not None + } + idle_recovery_config = ( + IdleRecoveryConfig(**idle_recovery_overrides) if idle_recovery_overrides else None + ) provider = CopilotProvider( mcp_servers=mcp_servers, model=default_model, diff --git a/tests/test_config/test_validator_capabilities.py b/tests/test_config/test_validator_capabilities.py index f0ff0fd8..4794c4bd 100644 --- a/tests/test_config/test_validator_capabilities.py +++ b/tests/test_config/test_validator_capabilities.py @@ -1141,6 +1141,55 @@ def test_runtime_timeout_with_supported_provider_passes(self, patch_caps: Any) - validate_workflow_config(config) +class TestWorkflowLevelIdleRecovery: + """Workflow-level idle_timeout_seconds / max_idle_recovery_attempts (#488).""" + + def test_idle_timeout_against_unsupported_provider_warns(self, patch_caps: Any) -> None: + """Unlike max_session_seconds, this is a tuning knob — a warning, + not an error.""" + patch_caps({"copilot": _caps(idle_recovery=False)}) + config = WorkflowConfig( + workflow=WorkflowDef( + name="t", + entry_point="a", + runtime=RuntimeConfig(provider="copilot", idle_timeout_seconds=30.0), + ), + agents=[AgentDef(name="a", prompt="hi")], + ) + warnings = validate_workflow_config(config) + assert any("idle_timeout_seconds" in w and "idle_recovery=False" in w for w in warnings) + + def test_max_idle_recovery_attempts_against_unsupported_provider_warns( + self, patch_caps: Any + ) -> None: + patch_caps({"copilot": _caps(idle_recovery=False)}) + config = WorkflowConfig( + workflow=WorkflowDef( + name="t", + entry_point="a", + runtime=RuntimeConfig(provider="copilot", max_idle_recovery_attempts=3), + ), + agents=[AgentDef(name="a", prompt="hi")], + ) + warnings = validate_workflow_config(config) + assert any( + "max_idle_recovery_attempts" in w and "idle_recovery=False" in w for w in warnings + ) + + def test_idle_timeout_with_supported_provider_no_warning(self, patch_caps: Any) -> None: + patch_caps({"copilot": _caps(idle_recovery=True)}) + config = WorkflowConfig( + workflow=WorkflowDef( + name="t", + entry_point="a", + runtime=RuntimeConfig(provider="copilot", idle_timeout_seconds=30.0), + ), + agents=[AgentDef(name="a", prompt="hi")], + ) + warnings = validate_workflow_config(config) + assert not any("idle_recovery=False" in w for w in warnings) + + class TestForEachProviderRecorded: """ForEach inline agent providers appear in workflow_started.providers block (#241 gap).""" diff --git a/tests/test_providers/test_capabilities.py b/tests/test_providers/test_capabilities.py index 6037363b..28bcc2b4 100644 --- a/tests/test_providers/test_capabilities.py +++ b/tests/test_providers/test_capabilities.py @@ -30,6 +30,7 @@ def _stable_capabilities(**overrides: object) -> ProviderCapabilities: "concurrent_safe": True, "skills": True, "session_continuity": True, + "idle_recovery": True, "upstream_pin": None, "maintainer": None, } diff --git a/tests/test_providers/test_idle_recovery.py b/tests/test_providers/test_idle_recovery.py index 01164f48..0e5133a5 100644 --- a/tests/test_providers/test_idle_recovery.py +++ b/tests/test_providers/test_idle_recovery.py @@ -1,6 +1,7 @@ """Unit tests for idle detection and recovery in CopilotProvider.""" import asyncio +import logging import time import unittest.mock from typing import Any @@ -929,9 +930,13 @@ async def drain_then_wait_for_recovery() -> None: await asyncio.sleep(0.05 * 2) assert mock_session.send.call_count == 0 active_tools_ref.clear() + # Wait for a recovery attempt to actually fire, then finish. - while mock_session.send.call_count == 0: - await asyncio.sleep(0.01) + async def _poll_for_send() -> None: + while mock_session.send.call_count == 0: + await asyncio.sleep(0.01) + + await asyncio.wait_for(_poll_for_send(), timeout=5.0) done.set() await asyncio.gather( @@ -1002,8 +1007,11 @@ async def test_none_active_tools_ref_preserves_existing_behavior(self) -> None: last_activity_ref: list[Any] = ["tool.execution_start", "read_agent", time.monotonic()] async def finish_after_first_recovery() -> None: - while mock_session.send.call_count == 0: - await asyncio.sleep(0.01) + async def _poll_for_send() -> None: + while mock_session.send.call_count == 0: + await asyncio.sleep(0.01) + + await asyncio.wait_for(_poll_for_send(), timeout=5.0) done.set() await asyncio.gather( @@ -1020,6 +1028,216 @@ async def finish_after_first_recovery() -> None: assert mock_session.send.call_count >= 1 + @pytest.mark.asyncio + async def test_no_recovery_prompt_clobbers_response_end_to_end(self) -> None: + """End-to-end regression test for #488, driven through the real + ``_send_and_wait`` -> ``on_event`` -> ``_wait_with_idle_detection`` + path (not a hand-injected ``active_tools_ref``). + + A long-running tool call that emits nothing for several idle + windows must not trigger idle recovery — and, critically, must not + let a recovery prompt's conversational reply clobber + ``response_content`` via last-message-wins (copilot.py:1615), which + is the actual user-visible symptom of #488. + """ + idle_timeout = 0.02 + config = IdleRecoveryConfig( + idle_timeout_seconds=idle_timeout, + max_recovery_attempts=5, + ) + provider = CopilotProvider( + mock_handler=stub_handler, + idle_recovery_config=config, + ) + + captured_cb: list[Any] = [] + sent: list[str] = [] + + def on_event(callback: Any) -> None: + captured_cb.append(callback) + + session = MagicMock() + session.on = on_event + + async def fake_send(prompt: str) -> None: + sent.append(prompt) + + session.send = fake_send + + async def driver() -> None: + callback = captured_cb[0] + + start_ev = MagicMock() + start_ev.type.value = "tool.execution_start" + start_ev.data.tool_name = "bash" + start_ev.data.tool_call_id = "c1" + callback(start_ev) + + # Total silence for several multiples of the idle window while + # the tool is "running" — this is exactly the window that used + # to trigger a recovery prompt before #488. + await asyncio.sleep(idle_timeout * 5) + + complete_ev = MagicMock() + complete_ev.type.value = "tool.execution_complete" + complete_ev.data.tool_call_id = "c1" + callback(complete_ev) + + message_ev = MagicMock() + message_ev.type.value = "assistant.message" + message_ev.data.content = "DONE" + callback(message_ev) + + idle_ev = MagicMock() + idle_ev.type.value = "session.idle" + callback(idle_ev) + + resp, _ = await asyncio.gather( + provider._send_and_wait( + session=session, + prompt="go", + verbose_enabled=False, + full_enabled=False, + ), + driver(), + ) + + assert resp.content == "DONE" + # No idle-recovery prompt was ever sent through the session — only + # the original prompt. + assert sent == ["go"] + + @pytest.mark.asyncio + async def test_overlapping_tool_calls_keyed_by_call_id_not_name(self) -> None: + """Two concurrent tool calls, driven through real events: suppression + must persist until BOTH complete, and must key on tool_call_id — a + regression that keyed the dict by tool_name instead would collapse + two same-named calls into one entry that drains on the first + completion, which this test's same-name variant would catch.""" + idle_timeout = 0.02 + config = IdleRecoveryConfig( + idle_timeout_seconds=idle_timeout, + max_recovery_attempts=5, + ) + provider = CopilotProvider( + mock_handler=stub_handler, + idle_recovery_config=config, + ) + + captured_cb: list[Any] = [] + sent: list[str] = [] + + def on_event(callback: Any) -> None: + captured_cb.append(callback) + + session = MagicMock() + session.on = on_event + + async def fake_send(prompt: str) -> None: + sent.append(prompt) + + session.send = fake_send + + def _mock_event(event_type: str, tool_call_id: str, tool_name: str | None) -> Any: + ev = MagicMock() + ev.type.value = event_type + ev.data.tool_call_id = tool_call_id + if tool_name is not None: + ev.data.tool_name = tool_name + return ev + + async def driver() -> None: + callback = captured_cb[0] + + # Both calls share the SAME tool_name but different call ids — + # if the implementation keyed on tool_name, these would + # collapse to a single dict entry. + callback(_mock_event("tool.execution_start", "call-a", "bash")) + callback(_mock_event("tool.execution_start", "call-b", "bash")) + + # Silence for several idle windows — still suppressed because + # BOTH calls are in flight. Only the original prompt has been + # sent so far. + await asyncio.sleep(idle_timeout * 5) + assert sent == ["go"] + + # Complete only call-a — call-b is still in flight, so + # suppression must continue. + callback(_mock_event("tool.execution_complete", "call-a", None)) + await asyncio.sleep(idle_timeout * 5) + assert sent == ["go"] + + # Complete call-b — nothing left in flight, so idle recovery + # resumes and fires a recovery prompt. + callback(_mock_event("tool.execution_complete", "call-b", None)) + + while len(sent) < 2: # original prompt + recovery prompt + await asyncio.sleep(0.005) + + idle_ev = MagicMock() + idle_ev.type.value = "session.idle" + callback(idle_ev) + + await asyncio.wait_for( + asyncio.gather( + provider._send_and_wait( + session=session, + prompt="go", + verbose_enabled=False, + full_enabled=False, + ), + driver(), + ), + timeout=5.0, + ) + + assert len(sent) == 2 + assert sent[0] == "go" + + @pytest.mark.asyncio + async def test_suppression_warns_once(self, caplog: Any) -> None: + """The first suppressed idle window logs a warning naming the + in-flight tools; a second suppressed window on the same provider + instance only logs at debug (warn-once latch, mirroring + ``_context_window_anomaly_warned``).""" + config = IdleRecoveryConfig( + idle_timeout_seconds=0.02, + max_recovery_attempts=2, + ) + provider = CopilotProvider( + mock_handler=stub_handler, + idle_recovery_config=config, + ) + + done = asyncio.Event() + mock_session = MagicMock() + mock_session.send = AsyncMock() + + last_activity_ref: list[Any] = ["tool.execution_start", "read_agent", time.monotonic()] + active_tools_ref: dict[str, str] = {"call-1": "read_agent"} + + async def finish_after_a_few_windows() -> None: + await asyncio.sleep(0.02 * 4) + done.set() + + with caplog.at_level(logging.WARNING, logger="conductor.providers.copilot"): + await asyncio.gather( + provider._wait_with_idle_detection( + done=done, + session=mock_session, + verbose_enabled=False, + full_enabled=False, + last_activity_ref=last_activity_ref, + active_tools_ref=active_tools_ref, + ), + finish_after_a_few_windows(), + ) + + assert provider._tool_suppression_warned is True + warning_records = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warning_records) == 1 + assert "read_agent" in warning_records[0].getMessage() + class TestOnEventActiveTools: """Tests for the on_event tracking of in-flight tool calls in _send_and_wait.""" @@ -1033,6 +1251,7 @@ async def test_matching_start_and_complete_clears_last_tool_call(self) -> None: provider = CopilotProvider(retry_config=RetryConfig(max_attempts=1)) captured_cb: list[Any] = [] captured_active_tools: dict[str, dict[str, str]] = {} + captured_ref: dict[str, Any] = {} start_ev = _Mock() start_ev.type.value = "tool.execution_start" @@ -1059,12 +1278,16 @@ async def fake_send(prompt: str) -> None: session.send = fake_send - # Patch _wait_with_idle_detection to capture active_tools_ref before - # returning, since _send_and_wait doesn't expose it directly. + # Patch _wait_with_idle_detection to capture active_tools_ref and + # last_activity_ref before returning, since _send_and_wait doesn't + # expose them directly. original_wait = provider._wait_with_idle_detection async def spy_wait(*args: Any, **kwargs: Any) -> Any: captured_active_tools["snapshot"] = dict(kwargs.get("active_tools_ref") or {}) + captured_ref["last_activity_ref"] = kwargs.get("last_activity_ref") or ( + args[4] if len(args) > 4 else None + ) return await original_wait(*args, **kwargs) with unittest.mock.patch.object(provider, "_wait_with_idle_detection", spy_wait): @@ -1078,6 +1301,10 @@ async def spy_wait(*args: Any, **kwargs: Any) -> Any: # By the time _wait_with_idle_detection was invoked, send() had # already fired every event synchronously, so the dict is empty. assert captured_active_tools["snapshot"] == {} + # The matching complete must clear last_activity_ref[1] rather than + # leaving it attributed to the tool that just finished (#488 + # misattribution fix). + assert captured_ref["last_activity_ref"][1] is None @pytest.mark.asyncio async def test_second_tool_name_survives_first_complete(self) -> None: @@ -1126,6 +1353,7 @@ async def spy_wait(*args: Any, **kwargs: Any) -> Any: captured_ref["last_activity_ref"] = kwargs.get("last_activity_ref") or ( args[4] if len(args) > 4 else None ) + captured_ref["active_tools_ref"] = dict(kwargs.get("active_tools_ref") or {}) return await original_wait(*args, **kwargs) with unittest.mock.patch.object(provider, "_wait_with_idle_detection", spy_wait): @@ -1138,3 +1366,8 @@ async def spy_wait(*args: Any, **kwargs: Any) -> Any: last_activity_ref = captured_ref["last_activity_ref"] assert last_activity_ref[1] == "bash" + # Distinguishing assertion: call-1 must be gone from active_tools + # while call-2 (still running) remains — pre-fix there is no + # tool.execution_complete handling at all, so active_tools_ref is + # never even threaded through to this call. + assert captured_ref["active_tools_ref"] == {"call-2": "bash"}