script-build-and-framework-tmp/agent/tests/test_no_progress_policy.py

import pytest

from agent.core.presets import AgentPreset, register_preset
from agent.core.runner import AgentRunner, NoProgressPolicy, RunConfig
from agent.failures import FailureDetail, FailureDisposition, ToolExecutionError
from agent.orchestration.models import AgentRole, CompletionPolicy
from agent.trace.store import FileSystemTraceStore
from agent.tools.models import ToolCapability
from agent.tools.registry import ToolRegistry

from test_coordinator_integration import FakeExecutor, create_task, make_coordinator


class StaticProgressCoordinator:
    def __init__(self):
        self.ensure_calls = 0

    async def ensure_ledger(self, *_args, **_kwargs):
        self.ensure_calls += 1

    async def root_completion(self, _root_trace_id):
        return {
            "status": "needs_replan",
            "root_task_id": "ignored-id",
            "pending_tasks": [{"objective": "same task", "status": "needs_replan"}],
        }

    async def progress_snapshot(self, _root_trace_id):
        return {
            "root": {"objective": "same task", "status": "needs_replan"},
            "active_tasks": [],
            "accepted": [],
        }


def _knowledge_off():
    from agent.tools.builtin.knowledge import KnowledgeConfig

    return KnowledgeConfig(
        enable_extraction=False,
        enable_completion_extraction=False,
        enable_injection=False,
    )


def _explicit_config(preset, *, max_iterations=8, tools=None):
    return RunConfig(
        agent_type=preset,
        completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
        max_iterations=max_iterations,
        tools=tools or [],
        tool_groups=[],
        root_task_spec={
            "objective": "same task",
            "acceptance_criteria": [{"description": "finish it"}],
        },
        knowledge=_knowledge_off(),
    )


@pytest.mark.asyncio
async def test_planner_stops_after_three_unchanged_semantic_cycles(tmp_path):
    preset = "test_no_progress_planner_stall"
    register_preset(
        preset,
        AgentPreset(role=AgentRole.PLANNER, allowed_tools=[], max_iterations=8, skills=[]),
    )
    calls = 0

    async def llm_call(**_kwargs):
        nonlocal calls
        calls += 1
        return {"content": "still thinking", "tool_calls": None, "finish_reason": "stop"}

    store = FileSystemTraceStore(str(tmp_path))
    result = await AgentRunner(
        trace_store=store,
        llm_call=llm_call,
        task_coordinator=StaticProgressCoordinator(),
    ).run_result(
        [{"role": "user", "content": "plan"}],
        _explicit_config(preset),
    )

    assert calls == 3
    assert result["status"] == "incomplete"
    assert result["failure"]["code"] == "NO_PROGRESS"
    events = await store.get_events(result["trace_id"])
    assert [event["event"] for event in events].count("no_progress_detected") == 1


@pytest.mark.asyncio
async def test_second_identical_tool_failure_stops_worker(tmp_path):
    preset = "test_no_progress_worker_failure"
    register_preset(
        preset,
        AgentPreset(
            role=AgentRole.WORKER,
            allowed_tools=["flaky_write"],
            max_iterations=8,
            skills=[],
        ),
    )
    registry = ToolRegistry()

    async def flaky_write():
        raise ToolExecutionError(
            FailureDetail(
                code="WORKSPACE_NOT_READY",
                message="create the missing workspace file",
                disposition=FailureDisposition.REPAIR_ATTEMPT,
                source_tool="flaky_write",
            )
        )

    registry.register(flaky_write, capabilities=[ToolCapability.WRITE])
    calls = 0

    async def llm_call(**_kwargs):
        nonlocal calls
        calls += 1
        return {
            "content": "",
            "tool_calls": [{
                "id": f"call-{calls}",
                "type": "function",
                "function": {"name": "flaky_write", "arguments": "{}"},
            }],
            "finish_reason": "tool_calls",
        }

    result = await AgentRunner(
        trace_store=FileSystemTraceStore(str(tmp_path)),
        tool_registry=registry,
        llm_call=llm_call,
        task_coordinator=StaticProgressCoordinator(),
    ).run_result(
        [{"role": "user", "content": "work"}],
        _explicit_config(preset, tools=["flaky_write"]),
    )

    assert calls == 2
    assert result["status"] == "failed"
    assert result["failure"]["code"] == "NO_PROGRESS"
    assert result["failure"]["details"]["last_failure"]["code"] == "WORKSPACE_NOT_READY"


@pytest.mark.asyncio
async def test_planner_stall_counter_survives_resume(tmp_path):
    preset = "test_no_progress_resume"
    register_preset(
        preset,
        AgentPreset(role=AgentRole.PLANNER, allowed_tools=[], max_iterations=4, skills=[]),
    )
    calls = 0

    async def llm_call(**_kwargs):
        nonlocal calls
        calls += 1
        return {"content": "", "tool_calls": None, "finish_reason": "stop"}

    store = FileSystemTraceStore(str(tmp_path))
    runner = AgentRunner(
        trace_store=store,
        llm_call=llm_call,
        task_coordinator=StaticProgressCoordinator(),
    )
    first_config = _explicit_config(preset, max_iterations=2)
    first_config._role_run_config_override_fields = frozenset({"max_iterations"})
    first = await runner.run_result([{"role": "user", "content": "plan"}], first_config)
    assert first["status"] == "incomplete"
    assert calls == 2

    resumed = await runner.run_result(
        [],
        RunConfig(trace_id=first["trace_id"], knowledge=_knowledge_off()),
    )

    assert calls == 3
    assert resumed["status"] == "incomplete"
    assert resumed["failure"]["code"] == "NO_PROGRESS"


@pytest.mark.asyncio
async def test_semantically_equal_task_ledgers_ignore_generated_ids(tmp_path):
    first, _, _ = await make_coordinator(tmp_path / "first", FakeExecutor([]))
    second, _, _ = await make_coordinator(tmp_path / "second", FakeExecutor([]))
    first_task_id = await create_task(first, objective="equivalent child")
    second_task_id = await create_task(second, objective="equivalent child")

    assert first_task_id != second_task_id
    assert await first.progress_snapshot("root") == await second.progress_snapshot("root")


def test_no_progress_policy_rejects_invalid_thresholds():
    with pytest.raises(ValueError, match="same_failure_limit"):
        NoProgressPolicy(same_failure_limit=0)
    with pytest.raises(ValueError, match="planner_stall_limit"):
        NoProgressPolicy(planner_stall_limit=0)