69 lines
2.9 KiB
Python
69 lines
2.9 KiB
Python
# ============================================================
|
||
# L2/L3 Plan 重放黄金测试（plan.md §3.1 / 矩阵 110 行剩余项）
|
||
# 覆盖：时间线按版本取节点（node_at）、输入一致可复算重放（replay_verify）、
|
||
# 输入变更拒绝、版本越界拒绝、L3 动作级重放。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from server.agent_core.plan_runtime import (
|
||
PlanInputMismatchError,
|
||
PlanNotFoundError,
|
||
PlanStore,
|
||
canonical_inputs_hash,
|
||
)
|
||
|
||
|
||
def _chain(store: PlanStore):
|
||
l0 = store.create(plan_id="l0-r", layer="L0", parent_id=None,
|
||
inputs={"prompt": "排产"}, payload={"goal": "保交期"})
|
||
l1 = store.create(plan_id="l1-r", layer="L1", parent_id=l0.planId,
|
||
inputs={"goal": l0.payload}, payload={"engine": "HYBRID"})
|
||
l2_inputs = {"orders": ["SO-1"], "objective": "due-date"}
|
||
l2 = store.create(plan_id="l2-r", layer="L2", parent_id=l1.planId,
|
||
inputs=l2_inputs, payload={"action": "RESCHEDULE"})
|
||
l3 = store.create(plan_id="l3-r", layer="L3", parent_id=l2.planId,
|
||
inputs={"action": l2.payload}, payload={"steps": [{"op": "run"}]})
|
||
return l0, l1, l2, l3, l2_inputs
|
||
|
||
|
||
def test_node_at_returns_immutable_version(tmp_path: Path):
|
||
"""时间线回放：node_at 按版本返回不可变节点，版本号落在历史范围内。"""
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
_l0, _l1, l2, _l3, l2_inputs = _chain(store)
|
||
n = store.node_at("l2-r", 1)
|
||
assert n.planId == l2.planId and n.version == 1
|
||
assert n.inputsHash == canonical_inputs_hash(l2_inputs)
|
||
with pytest.raises(PlanNotFoundError):
|
||
store.node_at("l2-r", 99)
|
||
with pytest.raises(PlanNotFoundError):
|
||
store.node_at("l2-r", 0)
|
||
|
||
|
||
def test_replay_verify_same_inputs_is_replayable(tmp_path: Path):
|
||
"""L2 单节点重放：相同输入可复算（replay_verify 通过）。"""
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
_l0, _l1, l2, _l3, l2_inputs = _chain(store)
|
||
node = store.replay_verify("l2-r", 1, l2_inputs)
|
||
assert node.inputsHash == l2.inputsHash
|
||
assert node.payload == {"action": "RESCHEDULE"}
|
||
|
||
|
||
def test_replay_verify_changed_inputs_rejected(tmp_path: Path):
|
||
"""输入变更（重放时参数漂移）必须拒绝，防止动作级重放误导。"""
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
_chain(store)
|
||
with pytest.raises(PlanInputMismatchError):
|
||
store.replay_verify("l2-r", 1, {"orders": ["SO-2"], "objective": "due-date"})
|
||
|
||
|
||
def test_l3_action_level_replay(tmp_path: Path):
|
||
"""L3 动作级重放：动作输入一致时可复算重放。"""
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
_l0, _l1, l2, _l3, _l2_inputs = _chain(store)
|
||
node = store.replay_verify("l3-r", 1, {"action": l2.payload})
|
||
assert node.payload == {"steps": [{"op": "run"}]}
|