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

from types import SimpleNamespace

import pytest

from agent.core.runner import AgentRunner, RunConfig
from agent.orchestration.models import CompletionPolicy
from agent.tools.builtin.knowledge import KnowledgeConfig
from agent.tools.builtin.subagent import agent as legacy_agent
from agent.tools.builtin.subagent import evaluate as legacy_evaluate
from agent.tools.registry import ToolRegistry
from agent.trace.goal_models import GoalTree
from agent.trace.models import Trace
from agent.trace.store import FileSystemTraceStore


def test_run_config_defaults_to_legacy_auto():
    assert RunConfig().completion_policy == CompletionPolicy.LEGACY_AUTO


def test_old_trace_without_agent_role_reads_as_legacy():
    original = Trace(trace_id="old", mode="agent").to_dict()
    original.pop("agent_role")
    trace = Trace.from_dict(original)
    assert trace.agent_role == "legacy"


def test_legacy_schema_keeps_agent_evaluate_goal_but_hides_explicit_tools():
    runner = AgentRunner(llm_call=lambda **_: None)
    schemas = runner._get_run_tool_schemas(RunConfig(tool_groups=["core"]))
    names = {schema["function"]["name"] for schema in schemas}
    assert {"agent", "evaluate", "goal"}.issubset(names)
    assert names.isdisjoint({
        "task_plan", "dispatch_tasks", "task_decide", "validate_attempt",
        "submit_attempt", "submit_validation",
    })


def test_legacy_tools_and_groups_keep_union_semantics():
    runner = AgentRunner(llm_call=lambda **_: None)
    schemas = runner._get_tool_schemas(tools=["read_file"], tool_groups=["core"])
    names = {schema["function"]["name"] for schema in schemas}
    assert "read_file" in names
    assert "agent" in names


class FakeLegacyRunner:
    def __init__(self, store):
        self.trace_store = store
        self.tools = ToolRegistry()
        self.config = SimpleNamespace(parallel_tool_execution=True)
        self.debug = False
        self.calls = []

    async def run_result(self, messages, config, on_event=None):
        self.calls.append((messages, config))
        await self.trace_store.update_trace(
            config.trace_id,
            status="completed",
            result_summary=f"result-{len(self.calls)}",
        )
        return {
            "status": "completed",
            "summary": f"result-{len(self.calls)}",
            "trace_id": config.trace_id,
            "stats": {"total_messages": 1, "total_tokens": 2, "total_cost": 0.0},
            "saved_knowledge_ids": [],
        }


async def _legacy_context(tmp_path):
    store = FileSystemTraceStore(str(tmp_path))
    await store.create_trace(Trace(trace_id="root", mode="agent", task="legacy mission"))
    tree = GoalTree(mission="legacy mission")
    goal = tree.add_goals(["legacy goal"])[0]
    tree.focus(goal.id)
    await store.update_goal_tree("root", tree)
    runner = FakeLegacyRunner(store)
    context = {
        "store": store,
        "trace_id": "root",
        "goal_id": goal.id,
        "runner": runner,
        "knowledge_config": _disabled_knowledge(),
    }
    return store, runner, context


@pytest.mark.asyncio
async def test_legacy_agent_single_and_continue_from_reuse_trace(tmp_path, monkeypatch):
    import agent.tools.builtin.subagent as subagent_module

    async def noop(*args, **kwargs):
        return None

    monkeypatch.setattr(subagent_module, "broadcast_sub_trace_started", noop)
    monkeypatch.setattr(subagent_module, "broadcast_sub_trace_completed", noop)
    store, runner, context = await _legacy_context(tmp_path)

    first = await legacy_agent(task="first legacy task", context=context)
    assert first["mode"] == "delegate"
    assert first["status"] == "completed"
    sub_trace_id = first["sub_trace_id"]
    sub_trace = await store.get_trace(sub_trace_id)
    assert sub_trace.parent_trace_id == "root"
    assert runner.calls[-1][1].completion_policy == CompletionPolicy.LEGACY_AUTO

    second = await legacy_agent(
        task="repair legacy task",
        continue_from=sub_trace_id,
        context=context,
    )
    assert second["sub_trace_id"] == sub_trace_id
    assert second["continue_from"] is True
    assert runner.calls[-1][1].trace_id == sub_trace_id


@pytest.mark.asyncio
async def test_legacy_multi_agent_keeps_parallel_result_shape(tmp_path, monkeypatch):
    import agent.tools.builtin.subagent as subagent_module

    async def noop(*args, **kwargs):
        return None

    monkeypatch.setattr(subagent_module, "broadcast_sub_trace_started", noop)
    monkeypatch.setattr(subagent_module, "broadcast_sub_trace_completed", noop)
    store, runner, context = await _legacy_context(tmp_path)
    result = await legacy_agent(task=["branch a", "branch b"], context=context)
    assert result["mode"] == "explore"
    assert result["status"] == "completed"
    assert len(result["sub_trace_ids"]) == 2
    assert len({item["trace_id"] for item in result["sub_trace_ids"]}) == 2
    assert all(call[1].agent_type == "explore" for call in runner.calls)
    traces = [await store.get_trace(item["trace_id"]) for item in result["sub_trace_ids"]]
    assert all(trace.parent_trace_id == "root" for trace in traces)


@pytest.mark.asyncio
async def test_legacy_evaluate_runs_real_tool_and_preserves_result_format(tmp_path, monkeypatch):
    import agent.tools.builtin.subagent as subagent_module

    async def noop(*args, **kwargs):
        return None

    monkeypatch.setattr(subagent_module, "broadcast_sub_trace_started", noop)
    monkeypatch.setattr(subagent_module, "broadcast_sub_trace_completed", noop)
    store, runner, context = await _legacy_context(tmp_path)
    result = await legacy_evaluate(
        messages=[{"role": "user", "content": "legacy output"}],
        context=context,
    )
    assert result["mode"] == "evaluate"
    assert result["status"] == "completed"
    assert result["summary"].startswith("result-")
    assert runner.calls[-1][1].agent_type == "evaluate"
    assert (await store.get_trace(result["sub_trace_id"])).parent_trace_id == "root"


@pytest.mark.asyncio
async def test_old_trace_can_continue_with_real_runner(tmp_path):
    calls = 0

    async def fake_llm(**kwargs):
        nonlocal calls
        calls += 1
        return {"content": f"legacy answer {calls}", "tool_calls": [], "finish_reason": "stop"}

    store = FileSystemTraceStore(str(tmp_path))
    runner = AgentRunner(trace_store=store, llm_call=fake_llm)
    first = await runner.run_result(
        [{"role": "user", "content": "first"}],
        RunConfig(name="legacy", knowledge=_disabled_knowledge()),
    )
    calls_after_first = calls
    second = await runner.run_result(
        [{"role": "user", "content": "continue"}],
        RunConfig(trace_id=first["trace_id"], knowledge=_disabled_knowledge()),
    )
    assert second["trace_id"] == first["trace_id"]
    assert second["status"] == "completed"
    assert calls > calls_after_first
    assert (await store.get_trace(first["trace_id"])).agent_role == "legacy"


def _disabled_knowledge():
    return KnowledgeConfig(
        enable_extraction=False,
        enable_completion_extraction=False,
        enable_injection=False,
    )