测试：补齐 legacy_auto 真实执行回归

使用真实旧 agent、并行多 Agent、evaluate 和 AgentRunner 路径验证结果格式、Sub-Trace 关系、旧 Trace 续跑及默认 completion policy，避免 V1 新闭环破坏历史业务。

agent/tests/test_legacy_compatibility.py

@@ -1,6 +1,16 @@
+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():
@@ -31,3 +41,149 @@ def test_legacy_tools_and_groups_keep_union_semantics():
     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,
+    )