feat(agent): harden event and trace observation

agent/agent/orchestration/__init__.py

@@ -69,6 +69,7 @@ from .validation_policy import (
 )
 from .wire import (
     ArtifactSnapshotView,
+    CapabilitiesView,
     MissionSnapshotView,
     OperationView,
     PlannerDecisionView,
@@ -88,6 +89,7 @@ __all__ = [
     "AttemptStatus",
     "AttemptSubmission",
     "BackgroundOperation",
+    "CapabilitiesView",
     "CommandRecord",
     "CompletionPolicy",
     "CriterionResult",

agent/agent/orchestration/api_v2.py

@@ -22,6 +22,7 @@ from .wire import (
     ArtifactSnapshotView,
     AttemptList,
     AttemptView,
+    CapabilitiesView,
     DispatchOperationRequest,
     EventPageView,
     EventView,
@@ -86,6 +87,18 @@ def create_orchestration_router(coordinator: TaskCoordinator) -> Any:
         prefix="/api/v2", tags=["orchestration-v2"], route_class=ProblemRoute
     )
 
+    @router.get("/capabilities", response_model=CapabilitiesView)
+    async def get_capabilities() -> CapabilitiesView:
+        return CapabilitiesView(
+            capabilities=[
+                "mission_snapshot",
+                "root_completion",
+                "artifact_snapshot",
+                "orchestration_event_cursor",
+                "trace_correlation_filters",
+            ]
+        )
+
     @router.get(
         "/roots/{root_trace_id}/snapshot",
         response_model=MissionSnapshotView,

agent/agent/orchestration/coordinator.py

@@ -204,7 +204,6 @@ class TaskCoordinator:
                             event=EventDraft(
                                 "ledger_created",
                                 {
-                                    "mission": spec.objective,
                                     "root_task_id": root_task_id,
                                 },
                             ),
@@ -214,11 +213,7 @@ class TaskCoordinator:
                             raise
                         await asyncio.sleep(_cas_backoff(attempt_number))
                         continue
-                    await self._emit(
-                        root_trace_id,
-                        "ledger_created",
-                        {"mission": spec.objective, "root_task_id": root_task_id},
-                    )
+                    await self._emit(committed.event)
                     return committed.ledger
             raise AssertionError("unreachable")
 
@@ -298,6 +293,7 @@ class TaskCoordinator:
         )
         event_payload: Dict[str, Any] = {}
         committed_fresh = False
+        committed_event = None
         async with self._lock(root_trace_id):
             for attempt_number in range(8):
                 ledger = await self.task_store.load(root_trace_id)
@@ -312,27 +308,35 @@ class TaskCoordinator:
                 if command_id and command_id in ledger.idempotency_results:
                     return dict(ledger.idempotency_results[command_id])
 
+                before = ledger.to_dict()
                 result = mutator(ledger)
-                event_payload = _plain(result)
+                domain_changed = ledger.to_dict() != before
+                event_payload = _event_entity_refs(ledger, event_type, result)
                 if command_id:
                     ledger.command_records[command_id] = CommandRecord(
                         command_id=command_id,
                         operation=event_type,
                         fingerprint=fingerprint,
-                        result_ref=event_payload,
+                        result_ref=_plain(result),
                         committed_revision=ledger.revision + 1,
                         operation_id=operation_id,
                     )
+                if not domain_changed and not command_id:
+                    return result
                 try:
-                    await self.task_store.commit(
+                    committed = await self.task_store.commit(
                         ledger,
                         expected_revision=ledger.revision,
                         idempotency_key=command_id,
-                        event=EventDraft(
-                            event_type,
-                            event_payload,
-                            command_id=command_id,
-                            operation_id=operation_id,
+                        event=(
+                            EventDraft(
+                                event_type,
+                                event_payload,
+                                command_id=command_id,
+                                operation_id=operation_id,
+                            )
+                            if domain_changed
+                            else None
                         ),
                     )
                 except RevisionConflict:
@@ -340,18 +344,19 @@ class TaskCoordinator:
                         raise
                     await asyncio.sleep(_cas_backoff(attempt_number))
                     continue
-                committed_fresh = True
+                committed_fresh = domain_changed
+                committed_event = committed.event
                 break
             else:
                 raise AssertionError("unreachable")
         if committed_fresh:
-            await self._emit(root_trace_id, event_type, event_payload)
+            await self._emit(committed_event)
         return result
 
-    async def _emit(self, root_trace_id: str, event_type: str, payload: Dict[str, Any]) -> None:
-        if self.event_sink:
+    async def _emit(self, event: Any) -> None:
+        if self.event_sink and event is not None:
             try:
-                await self.event_sink.emit(root_trace_id, event_type, payload)
+                await self.event_sink.emit(event)
             except Exception as exc:
                 # Events are an observational projection.  Once the Ledger
                 # commit succeeds, an EventSink outage must not make callers
@@ -359,8 +364,8 @@ class TaskCoordinator:
                 logger.warning(
                     "Orchestration event emission failed after Ledger commit "
                     "(%s, %s): %s",
-                    root_trace_id,
-                    event_type,
+                    event.root_trace_id,
+                    event.event_type,
                     exc,
                 )
 
@@ -1575,7 +1580,6 @@ class TaskCoordinator:
                 stage.started_at = utc_now()
                 stage.updated_at = stage.started_at
             return {
-                "operation_id": operation_id,
                 "attempt_id": attempt_id,
                 "validation_id": validation_id,
             }
@@ -1729,7 +1733,11 @@ class TaskCoordinator:
                 operation = ledger.operations[operation_id]
                 operation.validation_ids.append(validation.validation_id)
                 operation.updated_at = utc_now()
-            return {"validation_id": validation.validation_id, "task_id": task_id}
+            return {
+                "validation_id": validation.validation_id,
+                "task_id": task_id,
+                "attempt_id": attempt_id,
+            }
 
         result = await self._mutate(
             root_trace_id,
@@ -2499,7 +2507,11 @@ class TaskCoordinator:
                 operation.attempt_ids = [attempt_id]
                 operation.validation_ids.append(report.validation_id)
                 operation.updated_at = utc_now()
-            return {"validation_id": report.validation_id}
+            return {
+                "validation_id": report.validation_id,
+                "task_id": task_id,
+                "attempt_id": attempt_id,
+            }
 
         result = await self._mutate(
             root_trace_id,
@@ -2625,14 +2637,26 @@ class TaskCoordinator:
                     current, current_stage, operation_id, execution_epoch
                 )
             ):
-                return {"recorded": False}
+                return {
+                    "recorded": False,
+                    "attempt_id": attempt_id,
+                    "validation_id": validation_id,
+                }
             if current_stage.execution_stats is not None:
                 if current_stage.execution_stats != stats:
                     raise TaskConflict("Execution stats were already recorded")
-                return {"recorded": True}
+                return {
+                    "recorded": True,
+                    "attempt_id": attempt_id,
+                    "validation_id": validation_id,
+                }
             current_stage.execution_stats = stats
             current_stage.updated_at = utc_now()
-            return {"recorded": True}
+            return {
+                "recorded": True,
+                "attempt_id": attempt_id,
+                "validation_id": validation_id,
+            }
 
         result = await self._mutate(
             root_trace_id,
@@ -3161,6 +3185,39 @@ def _plain(value: Any) -> Any:
     return json_values(value)
 
 
+def _event_entity_refs(
+    ledger: TaskLedger,
+    event_type: str,
+    result: Dict[str, Any],
+) -> Dict[str, Any]:
+    """Project a command result to schema-v2 entity references only."""
+
+    if event_type.startswith("operation_"):
+        return {}
+    if event_type == "tasks_created":
+        return {
+            "parent_task_id": result.get("parent_task_id"),
+            "task_ids": list(result.get("task_ids", [])),
+        }
+    if event_type == "ledger_created":
+        return {"root_task_id": result.get("root_task_id")}
+
+    refs: Dict[str, Any] = {}
+    for key in ("task_id", "attempt_id", "validation_id", "decision_id"):
+        value = result.get(key)
+        if value is not None:
+            refs[key] = value
+    validation_id = refs.get("validation_id")
+    if validation_id in ledger.validations:
+        validation = ledger.validations[validation_id]
+        refs.setdefault("task_id", validation.task_id)
+        refs.setdefault("attempt_id", validation.attempt_id)
+    attempt_id = refs.get("attempt_id")
+    if attempt_id in ledger.attempts:
+        refs.setdefault("task_id", ledger.attempts[attempt_id].task_id)
+    return refs
+
+
 def _request_fingerprint(operation: str, payload: Any) -> str:
     canonical = json.dumps(
         {"operation": operation, "payload": _canonical_value(payload)},

agent/agent/orchestration/protocols.py

@@ -12,6 +12,7 @@ from .models import (
     EventDraft,
     EventPage,
     ExecutionStats,
+    OrchestrationEvent,
     TaskLedger,
 )
 
@@ -20,6 +21,7 @@ from .models import (
 class CommitResult:
     revision: int
     ledger: TaskLedger
+    event: Optional[OrchestrationEvent] = None
 
 
 @dataclass(frozen=True)
@@ -91,7 +93,7 @@ class ToolPolicy(Protocol):
 
 
 class EventSink(Protocol):
-    async def emit(self, root_trace_id: str, event_type: str, payload: Dict[str, Any]) -> None: ...
+    async def emit(self, event: OrchestrationEvent) -> None: ...
 
 
 __all__ = [

agent/agent/orchestration/store.py

@@ -128,9 +128,10 @@ class FileSystemTaskStore:
             data["updated_at"] = updated_at
             events = self._read_events(raw)
             next_sequence = self._next_sequence(raw, events)
+            recorded: Optional[OrchestrationEvent] = None
             if event and not self._event_already_recorded(events, event):
                 recorded = OrchestrationEvent(
-                    schema_version=1,
+                    schema_version=2,
                     event_id=_event_id(root_trace_id, next_sequence),
                     sequence=next_sequence,
                     root_trace_id=root_trace_id,
@@ -152,7 +153,11 @@ class FileSystemTaskStore:
             self._atomic_json_write(path, data)
             ledger.revision = new_revision
             ledger.updated_at = updated_at
-            return CommitResult(revision=ledger.revision, ledger=ledger)
+            return CommitResult(
+                revision=ledger.revision,
+                ledger=ledger,
+                event=recorded,
+            )
 
     async def list_events(
         self,
@@ -385,23 +390,17 @@ class TraceEventSink:
     def __init__(self, base_path: str = ".trace") -> None:
         self.base_path = Path(base_path).resolve()
 
-    async def emit(self, root_trace_id: str, event_type: str, payload: Dict[str, Any]) -> None:
-        await asyncio.to_thread(self._emit_locked, root_trace_id, event_type, payload)
+    async def emit(self, event: OrchestrationEvent) -> None:
+        await asyncio.to_thread(self._emit_locked, event)
 
-    def _emit_locked(self, root_trace_id: str, event_type: str, payload: Dict[str, Any]) -> None:
-        directory = _safe_child(self.base_path, root_trace_id, "orchestration")
+    def _emit_locked(self, event: OrchestrationEvent) -> None:
+        directory = _safe_child(self.base_path, event.root_trace_id, "orchestration")
         directory.mkdir(parents=True, exist_ok=True)
         path = directory / "events.jsonl"
-        event = {
-            "event_id": _event_id(root_trace_id, os.urandom(8).hex()),
-            "event_type": event_type,
-            "root_trace_id": root_trace_id,
-            "created_at": utc_now(),
-            "payload": json_values(payload),
-        }
+        wire_event = json_values(asdict(event))
         with FileLock(str(directory / ".events.lock")):
             with path.open("a", encoding="utf-8") as handle:
-                handle.write(json.dumps(event, ensure_ascii=False, sort_keys=True) + "\n")
+                handle.write(json.dumps(wire_event, ensure_ascii=False, sort_keys=True) + "\n")
                 handle.flush()
 
 

agent/agent/orchestration/wire.py

@@ -315,6 +315,13 @@ class EventPageView(WireModel):
     has_more: bool = False
 
 
+class CapabilitiesView(WireModel):
+    api_version: Literal["v2"] = "v2"
+    mission_snapshot_schema_version: Literal[1] = 1
+    event_schema_version: Literal[2] = 2
+    capabilities: List[str]
+
+
 class ProblemDetails(WireModel):
     type: str = "about:blank"
     title: str
@@ -328,6 +335,7 @@ __all__ = [
     "ArtifactSnapshotView",
     "AttemptList",
     "AttemptView",
+    "CapabilitiesView",
     "DispatchOperationRequest",
     "EventPageView",
     "EventView",

agent/agent/orchestration/wiring.py

@@ -7,8 +7,7 @@ from typing import Optional
 from .config import OrchestrationConfig
 from .coordinator import TaskCoordinator
 from .executor import LocalAgentExecutor
-from .protocols import ArtifactStore, TaskStore
-from .store import TraceEventSink
+from .protocols import ArtifactStore, EventSink, TaskStore
 
 
 def wire_orchestration(
@@ -16,15 +15,15 @@ def wire_orchestration(
     task_store: TaskStore,
     artifact_store: ArtifactStore,
     config: Optional[OrchestrationConfig] = None,
+    event_sink: Optional[EventSink] = None,
 ) -> TaskCoordinator:
     """Wire ports and adapters without importing any business project."""
-    base_path = getattr(task_store, "base_path", ".trace")
     coordinator = TaskCoordinator(
         task_store=task_store,
         artifact_store=artifact_store,
         trace_store=runner.trace_store,
         config=config or OrchestrationConfig(),
-        event_sink=TraceEventSink(str(base_path)),
+        event_sink=event_sink,
     )
     executor = LocalAgentExecutor(runner)
     coordinator.set_executor(executor)

agent/agent/trace/api.py

@@ -67,6 +67,13 @@ async def list_traces(
     agent_type: Optional[str] = None,
     uid: Optional[str] = None,
     status: Optional[str] = None,
+    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 = Query(20, le=100)
 ):
     """
@@ -85,6 +92,13 @@ async def list_traces(
         agent_type=agent_type,
         uid=uid,
         status=status,
+        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
     )
     return TraceListResponse(
@@ -114,10 +128,9 @@ async def get_trace(trace_id: str):
 
     # 获取所有 Sub-Traces（通过 parent_trace_id 查询）
     sub_traces = {}
-    all_traces = await store.list_traces(limit=1000)  # 获取所有 traces
-    for t in all_traces:
-        if t.parent_trace_id == trace_id:
-            sub_traces[t.trace_id] = t.to_dict()
+    children = await store.list_traces(parent_trace_id=trace_id, limit=1000)
+    for t in children:
+        sub_traces[t.trace_id] = t.to_dict()
 
     return TraceDetailResponse(
         trace=trace.to_dict(),

agent/agent/trace/protocols.py

@@ -48,6 +48,13 @@ class TraceStore(Protocol):
         agent_type: Optional[str] = None,
         uid: Optional[str] = None,
         status: Optional[str] = None,
+        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 = 50
     ) -> List[Trace]:
         """列出 Traces"""

agent/agent/trace/store.py

@@ -125,6 +125,13 @@ class FileSystemTraceStore:
         agent_type: Optional[str] = None,
         uid: Optional[str] = None,
         status: Optional[str] = None,
+        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 = 50
     ) -> List[Trace]:
         """列出 Traces"""
@@ -153,6 +160,26 @@ class FileSystemTraceStore:
                     continue
                 if status and data.get("status") != status:
                     continue
+                if agent_role and data.get("agent_role", "legacy") != agent_role:
+                    continue
+                if parent_trace_id and data.get("parent_trace_id") != parent_trace_id:
+                    continue
+                context = data.get("context") or {}
+                if not isinstance(context, dict):
+                    continue
+                if root_trace_id and not (
+                    data.get("trace_id") == root_trace_id
+                    or context.get("root_trace_id") == root_trace_id
+                ):
+                    continue
+                if task_id and context.get("task_id") != task_id:
+                    continue
+                if attempt_id and context.get("attempt_id") != attempt_id:
+                    continue
+                if validation_id and context.get("validation_id") != validation_id:
+                    continue
+                if operation_id and context.get("operation_id") != operation_id:
+                    continue
 
                 # 解析 datetime
                 if data.get("created_at"):

agent/agent/trace/websocket.py

@@ -86,10 +86,9 @@ async def watch_trace(
 
         # 获取所有 Sub-Traces（通过 parent_trace_id 查询）
         sub_traces = {}
-        all_traces = await store.list_traces(limit=1000)
-        for t in all_traces:
-            if t.parent_trace_id == trace_id:
-                sub_traces[t.trace_id] = t.to_dict()
+        children = await store.list_traces(parent_trace_id=trace_id, limit=1000)
+        for t in children:
+            sub_traces[t.trace_id] = t.to_dict()
 
         # 发送连接成功消息 + 完整状态（含 trace 当前执行状态）
         from .run_api import _running_tasks  # 避免循环导入，在函数内 import

agent/tests/test_observation_events_and_traces.py

@@ -0,0 +1,103 @@
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+
+from agent.orchestration.store import FileSystemArtifactStore, FileSystemTaskStore
+from agent.orchestration.wiring import wire_orchestration
+from agent.trace.models import Trace
+from agent.trace.store import FileSystemTraceStore
+
+from test_coordinator_integration import FakeExecutor, ROOT_TASK_SPEC, make_coordinator
+
+
+@pytest.mark.asyncio
+async def test_event_v2_payloads_are_refs_and_noop_stage_start_is_silent(tmp_path):
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    root_id = (await store.load("root")).root_task_id
+    reserved = await coordinator._create_attempt("root", root_id, "worker", None)
+    attempt_id = reserved["attempt_id"]
+    await coordinator._mark_stage_started("root", None, 0, attempt_id=attempt_id)
+    before = await store.load("root")
+    page_before = await store.list_events("root", limit=1000)
+    await coordinator._mark_stage_started("root", None, 0, attempt_id=attempt_id)
+    after = await store.load("root")
+    page_after = await store.list_events("root", limit=1000)
+
+    assert after.revision == before.revision
+    assert len(page_after.events) == len(page_before.events)
+    attempt_event = next(
+        event for event in page_after.events if event.event_type == "attempt_created"
+    )
+    assert attempt_event.schema_version == 2
+    assert attempt_event.payload == {"task_id": root_id, "attempt_id": attempt_id}
+    stage_event = next(
+        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}
+
+
+@pytest.mark.asyncio
+async def test_default_wiring_keeps_durable_journal_without_legacy_mirror(tmp_path):
+    trace_store = FileSystemTraceStore(str(tmp_path))
+    runner = SimpleNamespace(trace_store=trace_store, task_coordinator=None)
+    task_store = FileSystemTaskStore(str(tmp_path))
+    coordinator = wire_orchestration(
+        runner,
+        task_store,
+        FileSystemArtifactStore(str(tmp_path)),
+    )
+    await coordinator.ensure_ledger("root", ROOT_TASK_SPEC)
+    assert task_store.ledger_path("root").exists()
+    assert not (tmp_path / "root" / "orchestration" / "events.jsonl").exists()
+
+
+@pytest.mark.asyncio
+async def test_trace_correlation_filters_apply_before_limit(tmp_path):
+    store = FileSystemTraceStore(str(tmp_path))
+    target_time = datetime(2020, 1, 1)
+    await store.create_trace(
+        Trace(
+            trace_id="target",
+            mode="agent",
+            agent_role="validator",
+            parent_trace_id="parent",
+            created_at=target_time,
+            context={
+                "root_trace_id": "root",
+                "task_id": "task",
+                "attempt_id": "attempt",
+                "validation_id": "validation",
+                "operation_id": "operation",
+            },
+        )
+    )
+    for index in range(1001):
+        await store.create_trace(
+            Trace(
+                trace_id=f"unrelated-{index:04d}",
+                mode="agent",
+                created_at=target_time + timedelta(days=1, seconds=index),
+            )
+        )
+
+    filters = {
+        "agent_role": "validator",
+        "parent_trace_id": "parent",
+        "root_trace_id": "root",
+        "task_id": "task",
+        "attempt_id": "attempt",
+        "validation_id": "validation",
+        "operation_id": "operation",
+    }
+    for key, value in filters.items():
+        traces = await store.list_traces(limit=1, **{key: value})
+        assert [trace.trace_id for trace in traces] == ["target"]
+
+    await store.create_trace(
+        Trace(trace_id="root", mode="agent", agent_role="planner")
+    )
+    assert {
+        trace.trace_id
+        for trace in await store.list_traces(root_trace_id="root", limit=10)
+    } == {"root", "target"}

agent/tests/test_orchestration_store.py

@@ -191,7 +191,7 @@ async def test_events_commit_atomically_and_page_with_opaque_cursor(tmp_path):
 
     first_page = await store.list_events("root", limit=1)
     assert [event.sequence for event in first_page.events] == [1]
-    assert first_page.events[0].schema_version == 1
+    assert first_page.events[0].schema_version == 2
     assert first_page.has_more is True
     assert first_page.next_cursor and "command" not in first_page.next_cursor
     second_page = await store.list_events("root", cursor=first_page.next_cursor)
@@ -205,6 +205,32 @@ async def test_events_commit_atomically_and_page_with_opaque_cursor(tmp_path):
         await store.list_events("other-root", cursor=first_page.next_cursor)
 
 
+@pytest.mark.asyncio
+async def test_event_cursor_reads_mixed_v1_v2_journal(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    ledger = TaskLedger(root_trace_id="mixed", mission="mixed")
+    await store.commit(
+        ledger,
+        expected_revision=-1,
+        event=EventDraft("legacy_event", {"legacy": True}),
+    )
+    path = store.ledger_path("mixed")
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    raw["_events"][0]["schema_version"] = 1
+    path.write_text(json.dumps(raw), encoding="utf-8")
+
+    current = await store.load("mixed")
+    await store.commit(
+        current,
+        expected_revision=current.revision,
+        event=EventDraft("current_event", {"task_id": "task"}),
+    )
+    first = await store.list_events("mixed", limit=1)
+    second = await store.list_events("mixed", cursor=first.next_cursor, limit=1)
+    assert [first.events[0].schema_version, second.events[0].schema_version] == [1, 2]
+    assert [first.events[0].sequence, second.events[0].sequence] == [1, 2]
+
+
 @pytest.mark.asyncio
 async def test_event_command_replay_does_not_duplicate_event(tmp_path):
     store = FileSystemTaskStore(str(tmp_path))

agent/tests/test_orchestration_v2_api.py

@@ -374,6 +374,18 @@ async def test_atomic_mission_snapshot_completion_and_artifact_schema(api_fixtur
     missing = await http.get("/api/v2/roots/root/snapshots/missing")
     assert missing.status_code == 404
     assert missing.json()["code"] == "not_found"
+    assert (await http.get("/api/v2/capabilities")).json() == {
+        "api_version": "v2",
+        "mission_snapshot_schema_version": 1,
+        "event_schema_version": 2,
+        "capabilities": [
+            "mission_snapshot",
+            "root_completion",
+            "artifact_snapshot",
+            "orchestration_event_cursor",
+            "trace_correlation_filters",
+        ],
+    }
     await http.aclose()