257 lines
8.3 KiB
Python
257 lines
8.3 KiB
Python
# ============================================================
|
||
# PlanNode 可重生运行时黄金测试（plan.md §3.1 / §4.3）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
from pydantic import ValidationError
|
||
|
||
from server.agent_core.plan_runtime import (
|
||
PlanInputMismatchError,
|
||
PlanParentError,
|
||
PlanRegenerationFusedError,
|
||
PlanStore,
|
||
canonical_inputs_hash,
|
||
)
|
||
from server.gateway.plan_api import resolve_plan_store, router
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
|
||
|
||
def test_inputs_hash_is_canonical_for_json_objects() -> None:
|
||
left = {"orderIds": [3, 1], "settings": {"weight": 2, "enabled": True}}
|
||
right = {"settings": {"enabled": True, "weight": 2}, "orderIds": [3, 1]}
|
||
|
||
assert canonical_inputs_hash(left) == canonical_inputs_hash(right)
|
||
|
||
|
||
def test_plan_hierarchy_and_schema_contract(tmp_path: Path) -> None:
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
intent = store.create(
|
||
plan_id="intent-1",
|
||
layer="L0",
|
||
parent_id=None,
|
||
inputs={"prompt": "本周按交期优先排产"},
|
||
payload={"goal": "按交期优先"},
|
||
created_by="LLM",
|
||
)
|
||
strategy = store.create(
|
||
plan_id="strategy-1",
|
||
layer="L1",
|
||
parent_id=intent.planId,
|
||
inputs={"goal": intent.payload},
|
||
payload={"engine": "HYBRID"},
|
||
evidence_refs=["evidence-1"],
|
||
created_by="LLM",
|
||
)
|
||
|
||
assert strategy.parentId == intent.planId
|
||
assert strategy.version == 1
|
||
assert strategy.regenCount == 0
|
||
assert strategy.evidenceRefs == ("evidence-1",)
|
||
|
||
with pytest.raises(PlanParentError):
|
||
store.create(
|
||
layer="L2",
|
||
parent_id=intent.planId,
|
||
inputs={},
|
||
payload={},
|
||
)
|
||
|
||
schema = json.loads((ROOT / "shared/schemas/plan_node.schema.json").read_text(encoding="utf-8"))
|
||
assert set(schema["required"]) == set(type(intent).model_fields)
|
||
assert schema["additionalProperties"] is False
|
||
|
||
|
||
def test_regeneration_appends_and_preserves_old_versions(tmp_path: Path) -> None:
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
inputs = {"orders": ["SO-1"], "objective": "due-date"}
|
||
first = store.create(
|
||
plan_id="plan-immutable",
|
||
layer="L0",
|
||
parent_id=None,
|
||
inputs=inputs,
|
||
payload={"acceptance": ["无逾期"]},
|
||
status="APPROVED",
|
||
evidence_refs=["audit-1"],
|
||
created_by="USER",
|
||
)
|
||
second = store.regenerate(
|
||
first.planId,
|
||
inputs=inputs,
|
||
expected_inputs_hash=first.inputsHash,
|
||
payload={"acceptance": ["无逾期", "不超产能"]},
|
||
status="DRAFT",
|
||
created_by="LLM",
|
||
)
|
||
|
||
assert second.planId == first.planId
|
||
assert second.version == 2
|
||
assert second.regenCount == 1
|
||
assert second.inputsHash == first.inputsHash
|
||
|
||
versions = store.versions(first.planId)
|
||
assert [node.version for node in versions] == [1, 2]
|
||
assert versions[0].status == "APPROVED"
|
||
assert versions[0].payload == {"acceptance": ["无逾期"]}
|
||
assert versions[1].evidenceRefs == ("audit-1",)
|
||
|
||
with pytest.raises(ValidationError):
|
||
first.version = 99
|
||
|
||
assert isinstance(versions[0].payload, dict)
|
||
versions[0].payload["acceptance"].append("本地篡改")
|
||
assert store.versions(first.planId)[0].payload == {"acceptance": ["无逾期"]}
|
||
|
||
reloaded = PlanStore(str(tmp_path / "plans.json"))
|
||
assert [node.model_dump() for node in reloaded.versions(first.planId)] == [
|
||
node.model_dump() for node in store.versions(first.planId)
|
||
]
|
||
|
||
|
||
def test_regeneration_rejects_hash_drift_without_appending(tmp_path: Path) -> None:
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
first = store.create(layer="L0", parent_id=None, inputs={"scope": "A"}, payload={})
|
||
|
||
with pytest.raises(PlanInputMismatchError):
|
||
store.regenerate(
|
||
first.planId,
|
||
inputs={"scope": "B"},
|
||
expected_inputs_hash=canonical_inputs_hash({"scope": "B"}),
|
||
payload={"changed": True},
|
||
)
|
||
with pytest.raises(PlanInputMismatchError):
|
||
store.regenerate(
|
||
first.planId,
|
||
inputs={"scope": "A"},
|
||
expected_inputs_hash="0" * 64,
|
||
payload={"changed": True},
|
||
)
|
||
|
||
assert len(store.versions(first.planId)) == 1
|
||
|
||
|
||
def test_regeneration_fuses_after_default_threshold(tmp_path: Path) -> None:
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
inputs = {"scope": "stable"}
|
||
node = store.create(layer="L0", parent_id=None, inputs=inputs, payload={"attempt": 0})
|
||
|
||
for attempt in range(1, 4):
|
||
node = store.regenerate(
|
||
node.planId,
|
||
inputs=inputs,
|
||
expected_inputs_hash=node.inputsHash,
|
||
payload={"attempt": attempt},
|
||
)
|
||
|
||
assert node.regenCount == 3
|
||
with pytest.raises(PlanRegenerationFusedError):
|
||
store.regenerate(
|
||
node.planId,
|
||
inputs=inputs,
|
||
expected_inputs_hash=node.inputsHash,
|
||
payload={"attempt": 4},
|
||
)
|
||
assert [item.regenCount for item in store.versions(node.planId)] == [0, 1, 2, 3]
|
||
|
||
|
||
def test_failed_atomic_write_does_not_publish_in_memory_version(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
inputs = {"scope": "stable"}
|
||
first = store.create(layer="L0", parent_id=None, inputs=inputs, payload={"attempt": 0})
|
||
|
||
def fail_write() -> None:
|
||
raise OSError("disk full")
|
||
|
||
monkeypatch.setattr(store, "_write", fail_write)
|
||
with pytest.raises(OSError, match="disk full"):
|
||
store.regenerate(
|
||
first.planId,
|
||
inputs=inputs,
|
||
expected_inputs_hash=first.inputsHash,
|
||
payload={"attempt": 1},
|
||
)
|
||
|
||
assert [node.version for node in store.versions(first.planId)] == [1]
|
||
|
||
|
||
def test_minimal_plan_api_create_read_versions_and_regenerate(tmp_path: Path) -> None:
|
||
store = PlanStore(str(tmp_path / "plans.json"), max_regen=1)
|
||
app = FastAPI()
|
||
app.include_router(router)
|
||
app.dependency_overrides[resolve_plan_store] = lambda: store
|
||
client = TestClient(app)
|
||
inputs = {"prompt": "试排一版"}
|
||
|
||
created = client.post(
|
||
"/api/plans",
|
||
json={
|
||
"planId": "api-plan",
|
||
"layer": "L0",
|
||
"inputs": inputs,
|
||
"payload": {"goal": "试排"},
|
||
},
|
||
)
|
||
assert created.status_code == 201
|
||
first = created.json()["plan"]
|
||
assert first["createdBy"] == "USER"
|
||
|
||
spoofed = client.post(
|
||
"/api/plans",
|
||
json={
|
||
"planId": "spoofed-plan",
|
||
"layer": "L0",
|
||
"inputs": inputs,
|
||
"payload": {"goal": "伪造系统来源"},
|
||
"createdBy": "SYSTEM",
|
||
},
|
||
)
|
||
assert spoofed.status_code == 422
|
||
|
||
regenerated = client.post(
|
||
"/api/plans/api-plan/regenerate",
|
||
json={
|
||
"inputs": inputs,
|
||
"expectedInputsHash": first["inputsHash"],
|
||
"payload": {"goal": "试排并评估"},
|
||
},
|
||
)
|
||
assert regenerated.status_code == 201
|
||
assert regenerated.json()["plan"]["version"] == 2
|
||
assert regenerated.json()["plan"]["createdBy"] == "USER"
|
||
|
||
latest = client.get("/api/plans/api-plan")
|
||
history = client.get("/api/plans/api-plan/versions")
|
||
assert latest.json()["plan"]["version"] == 2
|
||
assert [item["version"] for item in history.json()["versions"]] == [1, 2]
|
||
|
||
fused = client.post(
|
||
"/api/plans/api-plan/regenerate",
|
||
json={
|
||
"inputs": inputs,
|
||
"expectedInputsHash": first["inputsHash"],
|
||
"payload": {"goal": "第三版"},
|
||
},
|
||
)
|
||
assert fused.status_code == 409
|
||
assert fused.json()["detail"]["code"] == "PLAN_REGEN_FUSED"
|
||
|
||
|
||
def test_gateway_registers_plan_routes() -> None:
|
||
from server.gateway.app import create_app
|
||
|
||
app = create_app()
|
||
assert str(app.url_path_for("create_plan")) == "/api/plans"
|
||
assert str(app.url_path_for("get_plan", plan_id="p1")) == "/api/plans/p1"
|
||
assert str(app.url_path_for("get_plan_versions", plan_id="p1")) == "/api/plans/p1/versions"
|
||
assert str(app.url_path_for("regenerate_plan", plan_id="p1")) == "/api/plans/p1/regenerate"
|