deer-flow/backend/tests/test_subagent_timeout_config.py

"""Tests for subagent runtime configuration.

Covers:
- SubagentsAppConfig / SubagentOverrideConfig model validation and defaults
- get_timeout_for() / get_max_turns_for() resolution logic
- AppConfig.subagents field access via AppConfig.current()
- registry.get_subagent_config() applies config overrides
- registry.list_subagents() applies overrides for all agents
- Polling timeout calculation in task_tool is consistent with config
"""

from unittest.mock import patch

import pytest

from deerflow.config.app_config import AppConfig
from deerflow.config.sandbox_config import SandboxConfig
from deerflow.config.subagents_config import (
    SubagentOverrideConfig,
    SubagentsAppConfig,
)


def _make_config(
    timeout_seconds: int = 900,
    *,
    max_turns: int | None = None,
    agents: dict | None = None,
) -> AppConfig:
    """Build an AppConfig with the given subagents settings."""
    return AppConfig(
        sandbox=SandboxConfig(use="test"),
        subagents=SubagentsAppConfig(
            timeout_seconds=timeout_seconds,
            max_turns=max_turns,
            agents={k: SubagentOverrideConfig(**v) for k, v in (agents or {}).items()},
        ),
    )


# ---------------------------------------------------------------------------
# SubagentOverrideConfig
# ---------------------------------------------------------------------------


class TestSubagentOverrideConfig:
    def test_default_is_none(self):
        override = SubagentOverrideConfig()
        assert override.timeout_seconds is None
        assert override.max_turns is None

    def test_explicit_values(self):
        override = SubagentOverrideConfig(timeout_seconds=120, max_turns=50)
        assert override.timeout_seconds == 120
        assert override.max_turns == 50

    def test_rejects_negative_timeout(self):
        with pytest.raises(Exception):
            SubagentOverrideConfig(timeout_seconds=-1)

    def test_rejects_zero_timeout(self):
        with pytest.raises(Exception):
            SubagentOverrideConfig(timeout_seconds=0)


# ---------------------------------------------------------------------------
# SubagentsAppConfig model
# ---------------------------------------------------------------------------


class TestSubagentsAppConfig:
    def test_default_timeout_is_900(self):
        config = SubagentsAppConfig()
        assert config.timeout_seconds == 900
        assert config.max_turns is None
        assert config.agents == {}

    def test_custom_defaults(self):
        config = SubagentsAppConfig(timeout_seconds=300, max_turns=50)
        assert config.timeout_seconds == 300
        assert config.max_turns == 50


# ---------------------------------------------------------------------------
# get_timeout_for / get_max_turns_for
# ---------------------------------------------------------------------------


class TestTimeoutResolution:
    def test_global_timeout_for_unknown_agent(self):
        config = SubagentsAppConfig(timeout_seconds=600)
        assert config.get_timeout_for("unknown") == 600

    def test_per_agent_timeout_overrides_global(self):
        config = SubagentsAppConfig(
            timeout_seconds=600,
            agents={"bash": SubagentOverrideConfig(timeout_seconds=120)},
        )
        assert config.get_timeout_for("bash") == 120
        assert config.get_timeout_for("general-purpose") == 600

    def test_per_agent_override_none_falls_back_to_global(self):
        config = SubagentsAppConfig(
            timeout_seconds=600,
            agents={"bash": SubagentOverrideConfig(timeout_seconds=None)},
        )
        assert config.get_timeout_for("bash") == 600


class TestMaxTurnsResolution:
    def test_builtin_default_when_no_override(self):
        config = SubagentsAppConfig()
        assert config.get_max_turns_for("bash", 60) == 60

    def test_global_max_turns_overrides_builtin(self):
        config = SubagentsAppConfig(max_turns=100)
        assert config.get_max_turns_for("bash", 60) == 100

    def test_per_agent_max_turns_overrides_global(self):
        config = SubagentsAppConfig(
            max_turns=100,
            agents={"bash": SubagentOverrideConfig(max_turns=30)},
        )
        assert config.get_max_turns_for("bash", 60) == 30
        assert config.get_max_turns_for("general-purpose", 60) == 100

    def test_per_agent_override_none_falls_back(self):
        config = SubagentsAppConfig(
            max_turns=100,
            agents={"bash": SubagentOverrideConfig(max_turns=None)},
        )
        assert config.get_max_turns_for("bash", 60) == 100


# ---------------------------------------------------------------------------
# AppConfig.subagents via AppConfig.current()
# ---------------------------------------------------------------------------


class TestAppConfigSubagents:
    def test_load_global_timeout(self):
        cfg = _make_config(timeout_seconds=300, max_turns=120)
        with patch.object(AppConfig, "current", return_value=cfg):
            sub = AppConfig.current().subagents
            assert sub.timeout_seconds == 300
            assert sub.max_turns == 120

    def test_load_with_per_agent_overrides(self):
        cfg = _make_config(
            timeout_seconds=900,
            max_turns=120,
            agents={
                "general-purpose": {"timeout_seconds": 1800, "max_turns": 200},
                "bash": {"timeout_seconds": 60, "max_turns": 80},
            },
        )
        with patch.object(AppConfig, "current", return_value=cfg):
            sub = AppConfig.current().subagents
            assert sub.get_timeout_for("general-purpose") == 1800
            assert sub.get_timeout_for("bash") == 60
            assert sub.get_max_turns_for("general-purpose", 100) == 200
            assert sub.get_max_turns_for("bash", 60) == 80

    def test_load_partial_override(self):
        cfg = _make_config(
            timeout_seconds=600,
            agents={"bash": {"timeout_seconds": 120, "max_turns": 70}},
        )
        with patch.object(AppConfig, "current", return_value=cfg):
            sub = AppConfig.current().subagents
            assert sub.get_timeout_for("general-purpose") == 600
            assert sub.get_timeout_for("bash") == 120
            assert sub.get_max_turns_for("general-purpose", 100) == 100
            assert sub.get_max_turns_for("bash", 60) == 70

    def test_load_empty_uses_defaults(self):
        cfg = _make_config()
        with patch.object(AppConfig, "current", return_value=cfg):
            sub = AppConfig.current().subagents
            assert sub.timeout_seconds == 900
            assert sub.max_turns is None
            assert sub.agents == {}