test(agent): verify generic mission observation e2e

agent/agent/core/runner.py

@@ -20,7 +20,10 @@ import logging
 import uuid
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import AsyncIterator, Optional, Dict, Any, List, Callable, Literal, Tuple, Union
+from typing import AsyncIterator, Optional, Dict, Any, List, Callable, Literal, Tuple, Union, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from agent.core.dream import DreamReport
 
 from agent.trace.models import Trace, Message
 from agent.trace.protocols import TraceStore
@@ -1401,7 +1404,7 @@ class AgentRunner:
                         await self.trace_store.update_goal_tree(trace_id, goal_tree)
                     self.log.info(f"自动创建 root goal: {goal_tree.goals[0].id}（未自动 focus，等待模型决定）")
                 else:
-                    self.log.debug(f"[Auto Root Goal] 检测到 goal 工具调用，跳过自动创建")
+                    self.log.debug("[Auto Root Goal] 检测到 goal 工具调用，跳过自动创建")
 
             # 获取当前 goal_id
             current_goal_id = goal_tree.current_id if (goal_tree and goal_tree.current_id) else None
@@ -1763,7 +1766,7 @@ class AgentRunner:
                         if tool_name == "goal" and goal_tree:
                             self.log.debug(f"[Goal Tool] After execution: goal_tree.goals={len(goal_tree.goals)}, current_id={goal_tree.current_id}")
                         if tool_name == "upload_knowledge" and isinstance(tool_result, dict):
-                            self.log.info(f"[Knowledge Tracking] 知识已上传")
+                            self.log.info("[Knowledge Tracking] 知识已上传")
                         if isinstance(tool_result, str):
                             tool_result = {"text": tool_result}
                         elif not isinstance(tool_result, dict):
@@ -1895,7 +1898,7 @@ class AgentRunner:
                         if tool_name == "upload_knowledge" and isinstance(tool_result, dict):
                             # upload_knowledge 返回的是统计信息，不是单个 knowledge_id
                             # 这里只记录上传动作，不跟踪具体 ID
-                            self.log.info(f"[Knowledge Tracking] 知识已上传到 Knowledge Manager")
+                            self.log.info("[Knowledge Tracking] 知识已上传到 Knowledge Manager")
     
                         # --- 支持多模态工具反馈 ---
                         # execute() 返回 dict{"text","images","tool_usage"} 或 str

agent/agent/docs/orchestration-v1.md

@@ -83,7 +83,7 @@ Worker 只能使用 `submit_attempt` 正式交付。Validator 只能使用 `subm
 - Planner 是唯一可以改变任务树和作出完成决策的角色。
 - Ledger 中只有一个特殊 Root Task；调用方提供它的 objective 和验收标准，框架不生成业务验收语义。
 - 普通顶层 Task 默认挂在 Root 下；所有直接子 Task 终态后，父 Task 回到 `needs_replan`，不会自动完成。
-- 父 Worker 会收到直接、已接受子 Task 的不可变结果引用，用于产生父 Task 的新 Attempt。
+- 父 Attempt 创建时会按直接子 Task 的 `display_path` 冻结 ACCEPT Decision ID；父 Worker 只解析这组不可变绑定，不扫描后续“最新状态”。旧 RUNNING Attempt 没有该绑定时按协议失败并重新规划。
 - Worker 看不到且不能执行 `agent`、`evaluate`、`goal` 或 Planner/Validator 工具。
 - Validator 使用独立 Trace，只能访问 preset 明确允许的只读工具。
 - explicit 模式的工具必须声明 capability；未分类工具默认拒绝。Validator 仅允许 `read + validation_submit`，Worker 不允许 `agent_spawn` 和 `task_control`。
@@ -104,11 +104,20 @@ Worker 只能使用 `submit_attempt` 正式交付。Validator 只能使用 `subm
 ```text
 .trace/{root_trace_id}/orchestration/
 ├── ledger.json
-├── artifacts/{snapshot_id}.json
-└── events.jsonl
+└── artifacts/{snapshot_id}.json
 ```
 
-`TaskStore`、`ArtifactStore`、`AgentExecutor`、`ToolPolicy`、`EventSink` 都是端口。V2 可增加远程执行器和公开 API，V3 可增加队列与分布式 Store，而不改变 Task 状态机和 Coordinator 决策语义。
+Orchestration Event v2 的 durable journal 位于 `ledger.json`，是默认唯一真源；只保存变化通知与实体 ID。`wire_orchestration(..., event_sink=...)` 可以显式注入外部 Sink；`TraceEventSink` 仅作为兼容镜像适配器，不再默认创建 `orchestration/events.jsonl`。Trace 自身的消息事件文件不受影响。
+
+## 通用观测 API
+
+- `GET /api/v2/roots/{root_trace_id}/snapshot`：单次 Ledger load 得到同一 revision 的 Mission 原子视图。
+- `GET /api/v2/roots/{root_trace_id}/completion`：Root objective、状态、阻塞原因与结果摘要。
+- `GET /api/v2/roots/{root_trace_id}/snapshots/{snapshot_id}`：Artifact 摘要哈希与不可变引用，不重复返回正文。
+- `GET /api/v2/roots/{root_trace_id}/events`：支持 v1/v2 混读的稳定 cursor。
+- `GET /api/v2/capabilities`：API、Mission Snapshot、Event schema 版本与能力发现。
+
+`OrchestrationClient` 提供对应的 Mission、Completion、Artifact、Event 读取方法，并可按 role、Root、Task、Attempt、Validation、Operation 或父 Trace 查询关联 Trace。`TaskStore`、`ArtifactStore`、`AgentExecutor`、`ToolPolicy`、`EventSink` 都是端口，可增加远程执行器或分布式 Store，而不改变 Task 状态机和 Coordinator 决策语义。
 
 ## V1 限制
 

agent/agent/orchestration/client_v2.py

@@ -220,6 +220,40 @@ class OrchestrationClient:
             )
         )
 
+    async def list_correlated_traces(
+        self,
+        *,
+        agent_role: Optional[str] = None,
+        parent_trace_id: Optional[str] = None,
+        root_trace_id: Optional[str] = None,
+        task_id: Optional[str] = None,
+        attempt_id: Optional[str] = None,
+        validation_id: Optional[str] = None,
+        operation_id: Optional[str] = None,
+        limit: int = 100,
+    ) -> List[Dict[str, Any]]:
+        params = {
+            key: value
+            for key, value in {
+                "agent_role": agent_role,
+                "parent_trace_id": parent_trace_id,
+                "root_trace_id": root_trace_id,
+                "task_id": task_id,
+                "attempt_id": attempt_id,
+                "validation_id": validation_id,
+                "operation_id": operation_id,
+                "limit": limit,
+            }.items()
+            if value is not None
+        }
+        value = await self._request("GET", "/api/traces", params=params)
+        traces = value.get("traces") if isinstance(value, dict) else None
+        if not isinstance(traces, list) or any(
+            not isinstance(item, dict) for item in traces
+        ):
+            raise ValueError("Trace API returned an invalid response")
+        return traces
+
     async def watch_events(
         self,
         root_trace_id: str,

agent/agent/trace/api.py

@@ -11,7 +11,6 @@ from datetime import datetime, timezone
 from typing import List, Optional, Dict, Any
 from fastapi import APIRouter, HTTPException, Query
 from pydantic import BaseModel
-from agent.llm.openrouter import openrouter_llm_call
 
 from .protocols import TraceStore
 
@@ -304,6 +303,8 @@ async def extract_comment_proxy(req: Dict[str, Any]):
 }}"""
 
     try:
+        from agent.llm.openrouter import openrouter_llm_call
+
         response = await openrouter_llm_call(
             messages=[{"role": "user", "content": prompt}],
             model="google/gemini-2.5-flash-lite",

agent/tests/test_attempt_input_binding.py

@@ -46,7 +46,9 @@ async def test_new_attempt_distinguishes_empty_and_ordered_child_bindings(tmp_pa
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("corruption", ["duplicate", "reverse", "cross", "non_accept"])
+@pytest.mark.parametrize(
+    "corruption", ["duplicate", "reverse", "missing", "cross", "non_accept"]
+)
 async def test_frozen_child_decision_binding_rejects_corruption(tmp_path, corruption):
     executor = FakeExecutor([ValidationVerdict.PASSED, ValidationVerdict.PASSED])
     coordinator, store, _ = await make_coordinator(tmp_path, executor)
@@ -64,8 +66,11 @@ async def test_frozen_child_decision_binding_rejects_corruption(tmp_path, corrup
         attempt.accepted_child_decision_ids = (ids[0], ids[0])
     elif corruption == "reverse":
         attempt.accepted_child_decision_ids = tuple(reversed(ids))
-    elif corruption == "cross":
+    elif corruption == "missing":
         attempt.accepted_child_decision_ids = ("missing-decision",)
+    elif corruption == "cross":
+        decision = ledger.decisions[ids[0]]
+        object.__setattr__(decision, "task_id", root.task_id)
     else:
         decision = ledger.decisions[ids[0]]
         object.__setattr__(decision, "action", DecisionAction.RETRY)
@@ -73,6 +78,57 @@ async def test_frozen_child_decision_binding_rejects_corruption(tmp_path, corrup
         coordinator._accepted_child_results(ledger, root, attempt)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("missing", ["attempt", "validation", "snapshot"])
+async def test_child_accept_binding_requires_complete_references(tmp_path, missing):
+    executor = FakeExecutor([ValidationVerdict.PASSED])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    child = await create_task(coordinator, "child")
+    await _accept_child(coordinator, child, "accept-child")
+    ledger = await store.load("root")
+    root = ledger.tasks[ledger.root_task_id]
+    reserved = await coordinator._create_attempt("root", root.task_id, "worker", None)
+    ledger = await store.load("root")
+    parent_attempt = ledger.attempts[reserved["attempt_id"]]
+    decision = ledger.decisions[parent_attempt.accepted_child_decision_ids[0]]
+    child_attempt = ledger.attempts[decision.attempt_id]
+    if missing == "attempt":
+        del ledger.attempts[decision.attempt_id]
+    elif missing == "validation":
+        del ledger.validations[decision.validation_id]
+    else:
+        child_attempt.snapshot_id = None
+    with pytest.raises(OrchestrationError):
+        coordinator._accepted_child_results(ledger, root, parent_attempt)
+
+
+@pytest.mark.asyncio
+async def test_missing_artifact_snapshot_fails_parent_as_protocol_violation(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.PASSED])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    child = await create_task(coordinator, "child")
+    await _accept_child(coordinator, child, "accept-child")
+    ledger = await store.load("root")
+    root = ledger.tasks[ledger.root_task_id]
+    reserved = await coordinator._create_attempt("root", root.task_id, "worker", None)
+    decision = ledger.decisions[ledger.tasks[child].decision_ids[-1]]
+    child_attempt = ledger.attempts[decision.attempt_id]
+    artifact_path = (
+        tmp_path
+        / "root"
+        / "orchestration"
+        / "artifacts"
+        / f"{child_attempt.snapshot_id}.json"
+    )
+    artifact_path.unlink()
+
+    result = await coordinator.advance_cycle("root", root.task_id, reserved["attempt_id"])
+    failed = (await store.load("root")).attempts[reserved["attempt_id"]]
+    assert "Invalid frozen child input binding" in result.error
+    assert failed.execution_stats.failure_code == FailureCode.PROTOCOL_VIOLATION
+    assert executor.worker_calls == 1  # child only; the parent Worker was never invoked
+
+
 @pytest.mark.asyncio
 async def test_legacy_running_attempt_never_guesses_child_inputs(tmp_path):
     executor = FakeExecutor([])

agent/tests/test_framework_observation_e2e.py

@@ -0,0 +1,140 @@
+import httpx
+import pytest
+
+import agent as framework
+from agent.orchestration.api_v2 import create_orchestration_router
+from agent.tools.builtin.knowledge import KnowledgeConfig
+from agent.trace.api import router as trace_router
+from agent.trace.api import set_trace_store
+
+from test_mission_loop import MissionLoopLLM, ROOT_OBJECTIVE, ROOT_OBJECTIVE_V2
+
+
+FastAPI = pytest.importorskip("fastapi").FastAPI
+
+
+def _knowledge_off():
+    return KnowledgeConfig(
+        enable_extraction=False,
+        enable_completion_extraction=False,
+        enable_injection=False,
+    )
+
+
+def _all_keys(value):
+    if isinstance(value, dict):
+        return set(value).union(*(_all_keys(item) for item in value.values()))
+    if isinstance(value, list):
+        return set().union(*(_all_keys(item) for item in value))
+    return set()
+
+
+@pytest.mark.asyncio
+async def test_public_framework_mission_is_fully_observable_without_ledger_access(
+    tmp_path,
+):
+    root_trace_id = "framework-blackbox-root"
+    trace_store = framework.FileSystemTraceStore(str(tmp_path))
+    task_store = framework.FileSystemTaskStore(str(tmp_path))
+    fake_llm = MissionLoopLLM(task_store)
+    fake_llm.root_trace_id = root_trace_id
+    runner = framework.AgentRunner(trace_store=trace_store, llm_call=fake_llm)
+    coordinator = framework.wire_orchestration(
+        runner,
+        task_store,
+        framework.FileSystemArtifactStore(str(tmp_path)),
+    )
+
+    result = await runner.run_result(
+        [{"role": "user", "content": "run a generic observable mission"}],
+        framework.RunConfig(
+            agent_type="planner",
+            completion_policy=framework.CompletionPolicy.EXPLICIT_VALIDATION,
+            max_iterations=30,
+            new_trace_id=root_trace_id,
+            tools=["task_plan", "dispatch_tasks", "task_decide"],
+            tool_groups=[],
+            enable_memory=False,
+            enable_research_flow=False,
+            knowledge=_knowledge_off(),
+            root_task_spec={
+                "objective": ROOT_OBJECTIVE,
+                "acceptance_criteria": [{
+                    "criterion_id": "root-ready",
+                    "description": "the root result is ready",
+                    "hard": True,
+                }],
+                "context_refs": ["memory://input/root"],
+            },
+        ),
+    )
+    assert result["status"] == "completed"
+
+    app = FastAPI()
+    app.include_router(create_orchestration_router(coordinator))
+    set_trace_store(trace_store)
+    app.include_router(trace_router)
+    http = httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="http://framework"
+    )
+    client = framework.OrchestrationClient("http://framework", client=http)
+
+    mission = await client.get_mission_snapshot(root_trace_id)
+    completion = await client.get_root_completion(root_trace_id)
+    events = await client.list_events(root_trace_id, limit=1000)
+    traces = await client.list_correlated_traces(
+        root_trace_id=root_trace_id, limit=100
+    )
+
+    assert mission.root_objective == ROOT_OBJECTIVE_V2
+    assert completion.root_objective == ROOT_OBJECTIVE_V2
+    assert completion.status == framework.TaskStatus.COMPLETED
+    assert completion.result_summary == f"submitted: {ROOT_OBJECTIVE_V2}"
+    assert any(
+        decision.action.value == "revise"
+        and decision.task_id == mission.root_task_id
+        for decision in mission.decisions
+    )
+    assert {event.schema_version for event in events.events} == {2}
+    assert {trace["agent_role"] for trace in traces} == {
+        "planner",
+        "worker",
+        "validator",
+    }
+
+    attempts = {item.attempt_id: item for item in mission.attempts}
+    validations = {item.validation_id: item for item in mission.validations}
+    trace_ids = {trace["trace_id"] for trace in traces}
+    accepted = [
+        decision for decision in mission.decisions
+        if decision.action.value == "accept"
+    ]
+    assert len(accepted) == 2
+    for decision in accepted:
+        attempt = attempts[decision.attempt_id]
+        validation = validations[decision.validation_id]
+        artifact = await client.get_artifact_snapshot(
+            root_trace_id, attempt.snapshot_id
+        )
+        assert attempt.task_id == decision.task_id
+        assert validation.task_id == decision.task_id
+        assert validation.attempt_id == attempt.attempt_id
+        assert validation.snapshot_id == artifact.snapshot_id
+        assert artifact.attempt_id == attempt.attempt_id
+        assert attempt.worker_trace_id in trace_ids
+        assert validation.validator_trace_id in trace_ids
+
+    root_attempt = attempts[next(
+        decision.attempt_id
+        for decision in accepted
+        if decision.task_id == mission.root_task_id
+    )]
+    child_accept = next(
+        decision for decision in accepted
+        if decision.task_id != mission.root_task_id
+    )
+    assert root_attempt.accepted_child_decision_ids == [child_accept.decision_id]
+
+    forbidden = {"script", "workflow", "round", "branch", "page_nodes"}
+    assert forbidden.isdisjoint(_all_keys(mission.model_dump(mode="json")))
+    await http.aclose()

agent/tests/test_mission_loop.py

@@ -18,6 +18,7 @@ from agent.trace.store import FileSystemTraceStore
 
 
 ROOT_OBJECTIVE = "complete the generic root"
+ROOT_OBJECTIVE_V2 = "complete the revised generic root"
 CHILD_OBJECTIVE_V1 = "produce the generic unit"
 CHILD_OBJECTIVE_V2 = "produce the revised generic unit"
 
@@ -86,7 +87,7 @@ class MissionLoopLLM:
     def _worker(self, messages):
         prompt = _latest_json_user_prompt(messages)
         spec = prompt["task_spec"]
-        if spec["objective"] == ROOT_OBJECTIVE:
+        if spec["objective"] in (ROOT_OBJECTIVE, ROOT_OBJECTIVE_V2):
             self.root_worker_prompt = prompt
         return self._call("submit_attempt", {
             "summary": f"submitted: {spec['objective']}",
@@ -184,7 +185,22 @@ class MissionLoopLLM:
             })
 
         assert child.status == TaskStatus.COMPLETED
-        if root.status == TaskStatus.NEEDS_REPLAN:
+        if root.status in (TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN):
+            if root.current_spec_version == 1:
+                return self._call("task_decide", {
+                    "task_id": root.task_id,
+                    "action": "revise",
+                    "reason": "clarify the root objective after child acceptance",
+                    "payload": {
+                        "objective": ROOT_OBJECTIVE_V2,
+                        "acceptance_criteria": [{
+                            "criterion_id": "root-ready",
+                            "description": "the revised root result is ready",
+                            "hard": True,
+                        }],
+                        "context_refs": ["memory://input/root"],
+                    },
+                })
             return self._call("dispatch_tasks", {"task_ids": [root.task_id]})
         if root.status == TaskStatus.AWAITING_DECISION:
             validation = ledger.validations[root.validation_ids[-1]]
@@ -262,6 +278,8 @@ async def test_real_runner_completes_replanned_root_mission(tmp_path):
     assert child.current_spec_version == 2
     assert child.status == TaskStatus.COMPLETED
     assert root.status == TaskStatus.COMPLETED
+    assert root.current_spec_version == 2
+    assert root.current_spec.objective == ROOT_OBJECTIVE_V2
     assert len(child.attempt_ids) == 2
     assert len(root.attempt_ids) == 1
 

agent/tests/test_observation_events_and_traces.py

@@ -1,4 +1,5 @@
 from datetime import datetime, timedelta
+import json
 from types import SimpleNamespace
 
 import pytest
@@ -35,6 +36,14 @@ async def test_event_v2_payloads_are_refs_and_noop_stage_start_is_silent(tmp_pat
         event for event in page_after.events if event.event_type == "stage_started"
     )
     assert stage_event.payload == {"task_id": root_id, "attempt_id": attempt_id}
+    mirrored = [
+        json.loads(line)
+        for line in (
+            tmp_path / "root" / "orchestration" / "events.jsonl"
+        ).read_text(encoding="utf-8").splitlines()
+    ]
+    assert mirrored[-1]["event_id"] == stage_event.event_id
+    assert mirrored[-1]["schema_version"] == 2
 
 
 @pytest.mark.asyncio

agent/tests/test_runner_completion_gate.py

@@ -23,6 +23,16 @@ class StubCoordinator:
         return self.snapshots[0]
 
 
+class RootFirstCoordinator(StubCoordinator):
+    def __init__(self):
+        super().__init__([{"status": "legacy"}])
+        self.root_calls = 0
+
+    async def root_completion(self, root_trace_id):
+        self.root_calls += 1
+        return {"status": "completed", "root_task_id": "root-task"}
+
+
 def _knowledge_off():
     from agent.tools.builtin.knowledge import KnowledgeConfig
 
@@ -33,6 +43,24 @@ def _knowledge_off():
     )
 
 
+@pytest.mark.asyncio
+async def test_runner_prefers_root_completion_and_keeps_legacy_fallback(tmp_path):
+    root_first = RootFirstCoordinator()
+    runner = AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path / "root-first")),
+        llm_call=None,
+        task_coordinator=root_first,
+    )
+    assert (await runner._get_mission_completion("root"))["status"] == "completed"
+    assert root_first.root_calls == 1
+    assert root_first.completion_calls == 0
+
+    legacy = StubCoordinator([{"status": "blocked"}])
+    runner.task_coordinator = legacy
+    assert (await runner._get_mission_completion("root"))["status"] == "blocked"
+    assert legacy.completion_calls == 1
+
+
 def _register_role_preset(name, role, max_iterations):
     register_preset(
         name,