248 lines
11 KiB
Python
248 lines
11 KiB
Python
# ============================================================
|
||
# PlanStore 分支回退黄金测试（矩阵 52 行：plan 版本随分支切换回退）
|
||
# 覆盖：plans_snapshot / restore_for_branch 回退与恢复、分支间隔离、
|
||
# 无锚定兼容（round-30/31 行为不变）、持久化、gateway switch 联动。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.agent_core.plan_runtime import PlanNotFoundError, PlanStore
|
||
from tests.auth_provider import install_test_auth
|
||
|
||
|
||
@pytest.fixture()
|
||
def secure_app(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "tenant.db"))
|
||
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
|
||
monkeypatch.setenv("APS_BRANCH_DIR", str(tmp_path / "branches"))
|
||
monkeypatch.setenv("APS_CHECKPOINT_PATH", str(tmp_path / "ckpt.json"))
|
||
import server.state.checkpoints as ck_mod
|
||
from server.agent_core.plan_runtime import reset_plan_stores
|
||
from server.db.database import reset_engine
|
||
from server.state import store as world_store
|
||
from server.state.branches import reset_branch_stores
|
||
from tests.pi_primary_adapter import install_fake_pi_tool_adapter
|
||
|
||
install_test_auth(monkeypatch, "tenant-a-000000000000000000000001")
|
||
install_fake_pi_tool_adapter(monkeypatch)
|
||
world_store._stores.clear()
|
||
ck_mod._checkpoints.clear()
|
||
reset_branch_stores()
|
||
reset_plan_stores()
|
||
reset_engine()
|
||
from server.gateway.app import create_app
|
||
app = create_app()
|
||
yield app
|
||
reset_engine()
|
||
world_store._stores.clear()
|
||
reset_plan_stores()
|
||
|
||
|
||
def _login(client: TestClient, username: str = "planner") -> None:
|
||
r = client.post("/api/auth/login", json={
|
||
"method": "password", "username": username, "password": "test",
|
||
})
|
||
assert r.status_code == 200
|
||
|
||
|
||
def _tree(client, sid):
|
||
return client.get(f"/api/sessions/{sid}/branches").json()["tree"]
|
||
|
||
|
||
def _checkpoint(client, sid) -> str:
|
||
"""建档并把 checkpointId 锚定到当前活动分支。"""
|
||
r = client.post("/api/chat", json={"sessionId": sid, "text": "建一个检查点"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
tree = _tree(client, sid)
|
||
active = next(b for b in tree["branches"] if b["id"] == tree["activeNode"])
|
||
cpid = active.get("checkpointId")
|
||
assert cpid, "建档后活动分支应锚定检查点"
|
||
return cpid
|
||
|
||
|
||
# ---------------- 存储层：分支视图回退 / 恢复 / 隔离 ----------------
|
||
def test_plan_views_roll_back_restore_and_isolate(tmp_path):
|
||
"""plans_snapshot/restore_for_branch：版本回退、恢复、分支间 plan 隔离。"""
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
store.create(plan_id="plan-a", layer="L0", parent_id=None,
|
||
inputs={"x": 1}, payload={"goal": "v1"}, created_by="USER")
|
||
store.create(plan_id="plan-b", layer="L0", parent_id=None,
|
||
inputs={"x": 2}, payload={"goal": "b1"}, created_by="USER")
|
||
|
||
# B1 锚定于 v1 状态
|
||
snap = store.plans_snapshot("branch-B1", checkpoint_id="ckpt-1")
|
||
assert snap["plans"] == {"plan-a": 1, "plan-b": 1}
|
||
assert store.restore_for_branch("branch-B1", checkpoint_id="ckpt-1")["restored"] is True
|
||
|
||
# B1 上流转 + 新建 plan（记录进 B1 视图）
|
||
store.transition_status("plan-a", new_status="APPROVED", created_by="USER")
|
||
store.create(plan_id="plan-c", layer="L0", parent_id=None, inputs={"x": 3}, payload={})
|
||
assert store.latest("plan-a").version == 2
|
||
|
||
# 从 B1 派生 B2（快照当前可见状态）
|
||
store.plans_snapshot("branch-B2", checkpoint_id="ckpt-2")
|
||
assert store.restore_for_branch("branch-B2", checkpoint_id="ckpt-2")["restored"] is True
|
||
assert store.latest("plan-a").version == 2
|
||
|
||
# 只在 B2 上创建 plan-d
|
||
store.create(plan_id="plan-d", layer="L0", parent_id=None, inputs={"x": 4}, payload={})
|
||
assert store.latest("plan-d").version == 1
|
||
|
||
# 回退到 B1：plan-d 不可见；plan-a 回到 B1 头（v2）；B2 的新版本不可见
|
||
assert store.restore_for_branch("branch-B1", checkpoint_id="ckpt-1")["restored"] is True
|
||
assert store.latest("plan-a").version == 2
|
||
assert [n.version for n in store.versions("plan-a")] == [1, 2]
|
||
with pytest.raises(PlanNotFoundError):
|
||
store.latest("plan-d")
|
||
with pytest.raises(PlanNotFoundError):
|
||
store.node_at("plan-a", 3)
|
||
|
||
# B1 上继续流转 → 追加 v3（与 B2 的 v2 视图分叉）
|
||
store.transition_status("plan-a", new_status="DONE", created_by="USER")
|
||
assert store.latest("plan-a").version == 3
|
||
|
||
# 恢复 B2：plan-a 回到 v2，plan-d 可见，v3 不可见
|
||
assert store.restore_for_branch("branch-B2", checkpoint_id="ckpt-2")["restored"] is True
|
||
assert store.latest("plan-a").version == 2
|
||
assert store.latest("plan-d").version == 1
|
||
assert [n.version for n in store.versions("plan-a")] == [1, 2]
|
||
with pytest.raises(PlanNotFoundError):
|
||
store.node_at("plan-a", 3)
|
||
|
||
# 审计/诊断视图
|
||
views = store.plan_views()
|
||
assert set(views) == {"branch-B1", "branch-B2"}
|
||
assert store.active_view()["branchId"] == "branch-B2"
|
||
|
||
|
||
def test_plan_views_persist_across_reload(tmp_path):
|
||
"""分支视图随 plans.json 持久化，重载后可继续回退。"""
|
||
path = str(tmp_path / "plans.json")
|
||
store = PlanStore(path)
|
||
store.create(plan_id="plan-p", layer="L0", parent_id=None,
|
||
inputs={"x": 1}, payload={"goal": "v1"})
|
||
store.plans_snapshot("branch-X", checkpoint_id="ckpt-x")
|
||
store.restore_for_branch("branch-X", checkpoint_id="ckpt-x")
|
||
store.transition_status("plan-p", new_status="APPROVED", created_by="USER")
|
||
assert store.latest("plan-p").version == 2
|
||
|
||
reloaded = PlanStore(path)
|
||
assert "branch-X" in reloaded.plan_views()
|
||
assert reloaded.restore_for_branch("branch-X", checkpoint_id="ckpt-x")["restored"] is True
|
||
assert reloaded.latest("plan-p").version == 2
|
||
|
||
|
||
def test_no_anchor_keeps_legacy_behavior(tmp_path):
|
||
"""无锚定：restore 返回 restored=False，latest()/versions()/创建流转与既有行为一致。"""
|
||
store = PlanStore(str(tmp_path / "plans.json"))
|
||
p = store.create(plan_id="plan-legacy", layer="L0", parent_id=None,
|
||
inputs={"x": 1}, payload={"goal": "v1"}, created_by="USER")
|
||
|
||
r = store.restore_for_branch("branch-none", checkpoint_id="ckpt-x", archived=None)
|
||
assert r["restored"] is False
|
||
assert store.active_view() is None
|
||
|
||
# 无锚定下行为不变（round-30/31 语义）
|
||
store.transition_status(p.planId, new_status="APPROVED", created_by="USER")
|
||
assert store.latest(p.planId).version == 2
|
||
assert [n.version for n in store.versions(p.planId)] == [1, 2]
|
||
|
||
# 仅快照不改变全局视图
|
||
store.plans_snapshot("branch-y", checkpoint_id="ckpt-y")
|
||
assert store.active_view() is None
|
||
assert store.latest(p.planId).version == 2
|
||
|
||
|
||
# ---------------- 网关层：分支切换联动回退 ----------------
|
||
def test_switch_restores_plan_view_via_gateway(secure_app):
|
||
"""gateway 分支切换：plan 版本随锚定 checkpoint 回退/恢复（矩阵 52 行）。"""
|
||
client = TestClient(secure_app)
|
||
_login(client)
|
||
created = client.post("/api/projects", json={"id": "proj-pv", "name": "plan回退项目"})
|
||
assert created.status_code == 200
|
||
sid = created.json()["session"]["id"]
|
||
tree = _tree(client, sid)
|
||
root = tree["treeRoot"]
|
||
|
||
# 主干创建 plan（v1）+ 审批（v2）
|
||
r = client.post("/api/plans", json={
|
||
"planId": "plan-rt", "layer": "L0",
|
||
"inputs": {"prompt": "交付优先"}, "payload": {"goal": "v1"},
|
||
})
|
||
assert r.status_code == 201, r.text[:300]
|
||
v2 = client.post("/api/plans/plan-rt/approve", json={"action": "approve"})
|
||
assert v2.status_code == 200, v2.text[:300]
|
||
assert v2.json()["plan"]["version"] == 2
|
||
|
||
# 排产 + 建档 → 主干锚定 C1（对话侧归档 plan v1/v2）
|
||
r = client.post("/api/chat", json={"sessionId": sid, "text": "试排一版交付优先"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
ck_root = _checkpoint(client, sid)
|
||
|
||
# fork 分支（继承 C1），分支上再生 plan → v3
|
||
forked = client.post(f"/api/sessions/{sid}/branches",
|
||
json={"name": "plan分支", "checkpointId": ck_root}).json()["branch"]
|
||
regen = client.post("/api/plans/plan-rt/regenerate", json={
|
||
"inputs": {"prompt": "交付优先"}, "expectedInputsHash": v2.json()["plan"]["inputsHash"],
|
||
"payload": {"goal": "v3"},
|
||
})
|
||
assert regen.status_code == 201, regen.text[:300]
|
||
assert regen.json()["plan"]["version"] == 3
|
||
# 分支再建档 → C2 归档 v1/v2/v3
|
||
r = client.post("/api/chat", json={"sessionId": sid, "text": "试排一版产能均衡"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
_checkpoint(client, sid)
|
||
|
||
# 切回主干：plan 视图回退到锚定版本（v2）
|
||
switched = client.post(f"/api/sessions/{sid}/branches/{root}/switch").json()
|
||
assert switched.get("restored") is True
|
||
assert switched.get("planView", {}).get("restored") is True
|
||
got = client.get("/api/plans/plan-rt")
|
||
assert got.status_code == 200
|
||
assert got.json()["plan"]["version"] == 2 # 版本回退
|
||
|
||
# 切到分支：plan 视图恢复分支状态（v3）
|
||
switched2 = client.post(f"/api/sessions/{sid}/branches/{forked['id']}/switch").json()
|
||
assert switched2.get("restored") is True
|
||
got2 = client.get("/api/plans/plan-rt")
|
||
assert got2.status_code == 200
|
||
assert got2.json()["plan"]["version"] == 3 # 版本恢复
|
||
|
||
# 分支上新建 plan → 切回主干后不可见（只见该分支 plan）
|
||
only_b = client.post("/api/plans", json={
|
||
"planId": "plan-only-b", "layer": "L0",
|
||
"inputs": {"prompt": "分支专属"}, "payload": {"goal": "b"},
|
||
})
|
||
assert only_b.status_code == 201
|
||
assert client.get("/api/plans/plan-only-b").json()["plan"]["version"] == 1
|
||
client.post(f"/api/sessions/{sid}/branches/{root}/switch")
|
||
hidden = client.get("/api/plans/plan-only-b")
|
||
assert hidden.status_code == 404 # 主干视图不可见
|
||
assert client.get("/api/plans/plan-rt").json()["plan"]["version"] == 2
|
||
|
||
|
||
def test_switch_without_anchor_keeps_plan_global(secure_app):
|
||
"""无锚定分支切换：plan 视图不激活，latest() 仍为全局（兼容既有行为）。"""
|
||
client = TestClient(secure_app)
|
||
_login(client)
|
||
created = client.post("/api/projects", json={"id": "proj-na2", "name": "无锚点plan"})
|
||
assert created.status_code == 200
|
||
sid = created.json()["session"]["id"]
|
||
|
||
r = client.post("/api/plans", json={
|
||
"planId": "plan-global", "layer": "L0",
|
||
"inputs": {"prompt": "全局"}, "payload": {"goal": "g"},
|
||
})
|
||
assert r.status_code == 201
|
||
|
||
forked = client.post(f"/api/sessions/{sid}/branches", json={"name": "无锚分支"}).json()["branch"]
|
||
assert forked["checkpointId"] is None
|
||
switched = client.post(f"/api/sessions/{sid}/branches/{forked['id']}/switch").json()
|
||
assert switched.get("restored") is False
|
||
assert switched.get("planView") is None
|
||
got = client.get("/api/plans/plan-global")
|
||
assert got.status_code == 200
|
||
assert got.json()["plan"]["version"] == 1
|