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, )