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"}]}
|