测试：覆盖 Ledger 存储、快照和 Goal 投影语义

验证文件 Ledger 的 revision 冲突与序列化、Artifact Snapshot 哈希稳定性和不可变引用约束，检查状态机禁止直接完成 Task，并同时固化 legacy Goal 父节点级联与显式模式关闭级联的差异。

agent/tests/test_goal_compatibility.py

@@ -0,0 +1,34 @@
+import pytest
+
+from agent.trace.goal_models import GoalTree
+from agent.trace.models import Trace
+from agent.trace.store import FileSystemTraceStore
+
+
+def test_goal_tree_legacy_cascade_and_explicit_no_cascade():
+    legacy = GoalTree(mission="m")
+    parent = legacy.add_goals(["parent"])[0]
+    children = legacy.add_goals(["a", "b"], parent_id=parent.id)
+    legacy.complete(children[0].id, "a")
+    legacy.complete(children[1].id, "b")
+    assert legacy.find(parent.id).status == "completed"
+
+    explicit = GoalTree(mission="m")
+    parent = explicit.add_goals(["parent"])[0]
+    children = explicit.add_goals(["a", "b"], parent_id=parent.id)
+    explicit.complete(children[0].id, "a", cascade_completion=False)
+    explicit.complete(children[1].id, "b", cascade_completion=False)
+    assert explicit.find(parent.id).status == "pending"
+
+
+@pytest.mark.asyncio
+async def test_trace_store_can_disable_goal_cascade(tmp_path):
+    store = FileSystemTraceStore(str(tmp_path))
+    await store.create_trace(Trace(trace_id="root", mode="agent"))
+    tree = GoalTree(mission="m")
+    parent = tree.add_goals(["parent"])[0]
+    children = tree.add_goals(["a", "b"], parent_id=parent.id)
+    await store.update_goal_tree("root", tree)
+    await store.update_goal("root", children[0].id, cascade_completion=False, status="completed")
+    await store.update_goal("root", children[1].id, cascade_completion=False, status="completed")
+    assert (await store.get_goal_tree("root")).find(parent.id).status == "pending"

agent/tests/test_orchestration_store.py

@@ -0,0 +1,57 @@
+import json
+
+import pytest
+
+from agent.orchestration.models import (
+    ArtifactRef,
+    AttemptSubmission,
+    TaskLedger,
+    TaskStatus,
+)
+from agent.orchestration.state_machine import InvalidTaskTransition, transition
+from agent.orchestration.store import (
+    FileSystemArtifactStore,
+    FileSystemTaskStore,
+    RevisionConflict,
+)
+
+
+@pytest.mark.asyncio
+async def test_ledger_atomic_revision_and_roundtrip(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    ledger = TaskLedger(root_trace_id="root", mission="mission")
+    committed = await store.commit(ledger, expected_revision=-1)
+    assert committed.revision == 0
+    loaded = await store.load("root")
+    assert loaded.mission == "mission"
+    assert loaded.revision == 0
+
+    stale = TaskLedger(root_trace_id="root", mission="stale", revision=-1)
+    with pytest.raises(RevisionConflict):
+        await store.commit(stale, expected_revision=-1)
+
+    on_disk = json.loads((tmp_path / "root" / "orchestration" / "ledger.json").read_text())
+    assert on_disk["revision"] == 0
+
+
+@pytest.mark.asyncio
+async def test_snapshot_is_immutable_and_hash_is_stable(tmp_path):
+    store = FileSystemArtifactStore(str(tmp_path)).for_root("root")
+    submission = AttemptSubmission(
+        summary=" delivered ",
+        artifact_refs=[ArtifactRef(uri="file:///result", version="v1")],
+        evidence_refs=[ArtifactRef(uri="db://row/1", digest="abc")],
+    )
+    first = await store.freeze("attempt-1", submission)
+    second = await store.freeze("attempt-2", submission)
+    assert first.sha256 == second.sha256
+    assert (await store.get(first.snapshot_id)).normalized_content["summary"] == "delivered"
+
+    with pytest.raises(ValueError, match="version or digest"):
+        ArtifactRef(uri="file:///mutable")
+
+
+def test_state_machine_rejects_shortcut_to_completed():
+    assert transition(TaskStatus.PENDING, TaskStatus.RUNNING) == TaskStatus.RUNNING
+    with pytest.raises(InvalidTaskTransition):
+        transition(TaskStatus.RUNNING, TaskStatus.COMPLETED)