fix: align agent timeline event contract

This commit is contained in:
2026-05-26 12:20:33 +08:00
parent f58a57e5b8
commit 96562877cc
5 changed files with 55 additions and 3 deletions

View File

@ -137,6 +137,8 @@ def test_process_projection_maps_task_team_events(tmp_path: Path) -> None:
planned_event = next(event for event in projection["events"] if event["kind"] == "task_planned")
assert planned_event["metadata"]["timeline_type"] == "plan"
assert planned_event["metadata"]["plan_mode"] == "team"
assert planned_event["metadata"]["strategy"] == "sequence"
assert planned_event["metadata"]["selected_skill_names"] == ["research-workflow"]
skill_event = next(event for event in projection["events"] if event["kind"] == "skill_selected")
@ -149,6 +151,7 @@ def test_process_projection_maps_task_team_events(tmp_path: Path) -> None:
node_event = next(event for event in projection["events"] if event["kind"] == "agent_finished")
assert node_event["metadata"]["timeline_type"] == "agent_progress"
assert "node_result" not in node_event["metadata"]
evidence_event = next(event for event in projection["events"] if event["kind"] == "task_result_ready")
assert evidence_event["metadata"]["timeline_type"] == "result"
@ -209,6 +212,32 @@ def test_process_projection_maps_failed_task_team_events(tmp_path: Path) -> None
assert node_event["metadata"]["timeline_type"] == "agent_progress"
def test_process_projection_uses_normalized_plan_metadata_defaults(tmp_path: Path) -> None:
session = SessionManager(tmp_path)
run_store = RunMemoryStore(tmp_path / "memory" / "runs")
session.append_message(
"web:test",
role="system",
event_type="task_execution_planned",
event_payload={
"task_id": "task-1",
"attempt_index": 1,
"plan_mode": None,
"strategy": None,
},
context_visible=False,
)
projection = SessionProcessProjector(session, run_store).project("web:test")
root_run = next(run for run in projection["runs"] if run["run_id"] == "task:task-1:attempt:1")
assert root_run["metadata"]["plan_mode"] == "single"
assert root_run["metadata"]["strategy"] == "single"
planned_event = next(event for event in projection["events"] if event["kind"] == "task_planned")
assert planned_event["metadata"]["plan_mode"] == "single"
assert planned_event["metadata"]["strategy"] == "single"
def test_process_projection_exposes_ephemeral_guidance_artifacts(tmp_path: Path) -> None:
session = SessionManager(tmp_path)
run_store = RunMemoryStore(tmp_path / "memory" / "runs")