muse-agent-example/tests/集成/test_规划候选与确认.py
zizi 224df00706 W13 渐进探索、分层规划与近期细化:私人探索、明确选定、分层规划与近期细化;含竞态修复新用例。
按 R2 串行阶段整理提交;包内文件为该阶段交付(含后续小增量),状态以工作包清单为准。
2026-09-10 19:25:41 +08:00

697 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""规划字段由S04描述;候选和正式版本通过真实S01事务区分。"""
import pytest
from muse.作品规划.接口 import 档案保存, 规划保存, 规划定位
from muse.元数据.接口 import 元数据服务, 导入内置结构
from muse.共享.调用身份 import 内容用途, 用途, 调用身份
from muse.启动 import 构建
from muse.正式变更.接口 import 作者动作
from muse.配置 import 应用配置
pytestmark = pytest.mark.数据库
@pytest.mark.parametrize("title", ["雪中的信", "浮岛航线"], ids=["letter", "floating-island"])
def test_同一目录承接卷章场景且移动保持身份__ad1310(规划环境, title):
from dataclasses import replace
from muse.作品规划.接口 import 目录节点, 节点目录保存
from muse.正文写作.接口 import 文本节点, 正文草稿, 段落
app, actor, _ = 规划环境
works = app.要求作品()
nodes = (
目录节点("volume-a", "volume", title + "·卷一"),
目录节点("chapter-a", "chapter", "出发", "volume-a"),
目录节点("scene-a", "scene", "有人敲门", "chapter-a"),
目录节点("chapter-b", "chapter", "另一条路"),
)
request = 节点目录保存("planned-work", 0, nodes)
receipt = works.保存节点目录(actor, "create-nodes", request)
assert receipt == works.保存节点目录(actor, "create-nodes", request)
current = works.读取节点目录(actor, "planned-work")
assert current["revision"] == 1 and len(current["nodes"]) == 4
assert [c["chapter_id"] for c in works.读取目录(actor, "planned-work")["chapters"]] == [
"chapter-a",
"chapter-b",
]
prose = app.要求正文()
prose.保存人工(
actor,
"write-before-move",
"chapter-a",
0,
正文草稿((段落("paragraph-a", (文本节点(title + ",有人来了。"),)),)),
)
before = prose.读取正文(actor, "chapter-a")
moved = (nodes[3], nodes[0], nodes[1], replace(nodes[2], parent_id="chapter-b"))
works.保存节点目录(actor, "move-nodes", replace(request, expected_revision=1, nodes=moved))
latest = works.读取节点目录(actor, "planned-work")
assert latest["revision"] == 2
assert {n["node_id"] for n in latest["nodes"]} == {n.node_id for n in nodes}
assert next(n for n in latest["nodes"] if n["node_id"] == "scene-a")["parent_id"] == "chapter-b"
assert [c["chapter_id"] for c in works.读取目录(actor, "planned-work")["chapters"]] == [
"chapter-b",
"chapter-a",
]
assert works.读取节点目录(actor, "planned-work", 版本=1) == current
assert prose.读取正文(actor, "chapter-a") == before
def test_规划候选编辑追加版本且旧审阅失效__ad1306(规划环境):
from muse.元数据.接口 import 字段变更
from muse.共享.错误 import Muse错误
app, actor, _ = 规划环境
service = app.要求规划()
schema = service.新规划结构(actor, "planned-work", "outline", 1)
reply = service.保存人工候选(
actor,
"editable-plan",
规划保存("planned-work", 0, schema, {"名称": "未寄出的信", "主线一句话": "寻找收信人"}),
)
id_ = reply["results"][0]["candidate_id"]
old = service.读取候选(actor, id_)
loc = 规划定位(id_, 1, old["candidate_hash"], old["plan_id"], 0)
review = service.打开审阅(actor, loc)
changes = (字段变更(("主线一句话",), "set", "收信人寻找寄信者"),)
edited = service.编辑候选(actor, "edit-plan-once", loc, changes)
assert edited == service.编辑候选(actor, "edit-plan-once", loc, changes)
current = service.读取候选(actor, id_)
assert current["revision"] == 2 and current["candidate_hash"] != old["candidate_hash"]
assert current["content"] == {"名称": "未寄出的信", "主线一句话": "收信人寻找寄信者"}
assert service.读取候选(actor, id_, 版本=1) == old
assert service.列出规划(actor, "planned-work") == []
with pytest.raises(Muse错误):
service.决定(
actor,
"old-plan-review",
loc,
作者动作.采纳,
review["review_id"],
review["review_hash"],
("plan",),
)
with pytest.raises(Muse错误):
service.编辑候选(actor, "stale-edit-plan", loc, changes)
new_loc = 规划定位(id_, 2, current["candidate_hash"], current["plan_id"], 0)
new_review = service.打开审阅(actor, new_loc)
service.决定(
actor,
"accept-edited-plan",
new_loc,
作者动作.采纳,
new_review["review_id"],
new_review["review_hash"],
("plan",),
)
assert service.读取规划(actor, current["plan_id"])["content"] == current["content"]
def test_卷章场景采用动态结构且章入口共用身份__ad1311(规划环境):
from dataclasses import replace
from muse.作品规划.接口 import 目录节点, 节点目录保存
app, actor, _ = 规划环境
app.要求作品().保存节点目录(
actor,
"plan-targets",
节点目录保存(
"planned-work",
0,
(
目录节点("volume", "volume", "卷一"),
目录节点("chapter", "chapter", "开门", "volume"),
目录节点("scene", "scene", "门外的访客", "chapter"),
),
),
)
service = app.要求规划()
outline = service.新规划结构(actor, "planned-work", "outline", 1)
volume = 保存并确认(
service,
actor,
规划保存("planned-work", 0, outline, {"名称": "卷一大纲"}, node_id="volume"),
"volume-plan",
)
scene = service.新规划结构(actor, "planned-work", "scene", 1)
saved_scene = 保存并确认(
service,
actor,
规划保存(
"planned-work",
0,
scene,
{"目标推进收束": "来客想借一盏灯,邮差让他进屋。"},
node_id="scene",
),
"scene-plan",
)
assert saved_scene["chapter_id"] == "chapter" and saved_scene["node_id"] == "scene"
fine = service.新规划结构(actor, "planned-work", "fine_outline", 1)
req = 规划保存(
"planned-work",
0,
fine,
{**细纲内容(volume["plan_id"]), "targetChapter": 1},
node_id="chapter",
)
first = 保存并确认(service, actor, req, "chapter-node-plan")
assert first["chapter_id"] == "chapter"
second = 保存并确认(
service,
actor,
replace(req, node_id=None, chapter_id="chapter", expected_revision=1),
"chapter-legacy-plan",
)
assert first["plan_id"] == second["plan_id"] and second["revision"] == 2
assert service.读取可用细纲(actor, "planned-work", "chapter")["revision"] == 2
@pytest.fixture
def 规划环境(应用测试库):
with 应用测试库[用途.维护].连接() as 连, 连.transaction():
导入内置结构(元数据服务(连))
app = 构建(应用配置(应用测试库[用途.生产].引用, "synthetic"))
actor = 调用身份("author", None, 用途.生产, 内容用途.规划)
schema = app.要求作品().新档案结构(actor, "planned-work", "work_core", 1)
app.要求作品().保存档案(
actor, "create-planned-work", 档案保存("planned-work", 0, {"名称": "雪中的信"}, schema)
)
return app, actor, 应用测试库
def test_人工规划候选经明确审阅后形成正式版本__ad1301(规划环境):
app, actor, _ = 规划环境
service = app.要求规划()
schema = service.新规划结构(actor, "planned-work", "outline", 1)
request = 规划保存(
"planned-work", 0, schema, {"名称": "雪中的信", "主线一句话": "邮差寻找一封无人认领的信。"}
)
receipt = service.保存人工候选(actor, "outline-draft", request)
candidate = service.读取候选(actor, receipt["results"][0]["candidate_id"])
assert service.列出规划(actor, "planned-work") == []
location = 规划定位(
candidate["candidate_id"],
candidate["revision"],
candidate["candidate_hash"],
candidate["plan_id"],
0,
)
review = service.打开审阅(actor, location)
result = service.决定(
actor,
"confirm-outline",
location,
作者动作.采纳,
review["review_id"],
review["review_hash"],
("plan",),
)
saved = service.读取规划(actor, candidate["plan_id"])
assert saved["revision"] == 1 and saved["content"] == request.content
assert (
service.决定(
actor,
"confirm-outline",
location,
作者动作.采纳,
review["review_id"],
review["review_hash"],
("plan",),
)
== result
)
def test_规划HTTP候选与CLI决定共享版本回执__ad1302(规划环境, tmp_path):
import json
import subprocess
import sys
from fastapi.testclient import TestClient
from muse.接入.http.应用 import 创建应用
from muse.配置 import 读取配置
_, _, db = 规划环境
password = tmp_path / "口令.txt"
password.write_text("synthetic-plan-password")
password.chmod(0o600)
cfg = tmp_path / "规划.toml"
cfg.write_text(
'["数据库"]\n"取值方式"="受控存储"\n"位置"='
+ json.dumps(db[用途.生产].引用.位置)
+ '\n["资源"]\n"发布身份"="synthetic"\n[HTTP]\n"口令文件"="口令.txt"\n"作者ID"="author"\n"公开地址"="http://testserver"\n"允许来源"=["http://testserver"]\n'
)
with TestClient(创建应用(读取配置(cfg)), headers={"origin": "http://testserver"}) as client:
assert (
client.post("/api/v1/session", json={"password": "synthetic-plan-password"}).status_code
== 200
)
nodes_request = {
"command_id": "http-plan-nodes",
"directory": {
"work_id": "planned-work",
"expected_revision": 0,
"nodes": [
{"node_id": "vol-http", "kind": "volume", "title": "卷一", "parent_id": None}
],
},
}
nodes_response = client.post("/api/v1/works/planned-work/plan-nodes", json=nodes_request)
assert nodes_response.status_code == 200, nodes_response.text
node_cli = subprocess.run(
[sys.executable, "-m", "muse", "作品", str(cfg), "节点目录", "planned-work"],
text=True,
capture_output=True,
)
assert node_cli.returncode == 0, node_cli.stderr
assert (
json.loads(node_cli.stdout)
== client.get("/api/v1/works/planned-work/plan-nodes").json()
)
node_path = tmp_path / "节点.json"
node_path.write_text(json.dumps(nodes_request, ensure_ascii=False))
node_replay = subprocess.run(
[sys.executable, "-m", "muse", "作品", str(cfg), "保存节点", str(node_path)],
text=True,
capture_output=True,
)
assert node_replay.returncode == 0, node_replay.stderr
assert json.loads(node_replay.stdout) == nodes_response.json()
schema = client.get(
"/api/v1/works/planned-work/plans/schema",
params={"schema_id": "outline", "base_version": 1},
).json()
request = {
"command_id": "http-plan",
"request": {
"work_id": "planned-work",
"expected_revision": 0,
"schema": schema,
"content": {"名称": "下雪以前"},
},
}
created = client.post("/api/v1/works/planned-work/plans/candidates", json=request)
assert created.status_code == 200, created.text
id_ = created.json()["results"][0]["candidate_id"]
candidate = client.get(f"/api/v1/plan-candidates/{id_}").json()
location = {
"candidate_id": id_,
"candidate_revision": 1,
"candidate_hash": candidate["candidate_hash"],
"plan_id": candidate["plan_id"],
"expected_revision": 0,
}
edit = {
"command_id": "edit-plan-cli",
"candidate": location,
"changes": [{"路径": ["名称"], "动作": "set", "值": "下雪后的信"}],
}
edit_path = tmp_path / "编辑.json"
edit_path.write_text(json.dumps(edit, ensure_ascii=False))
edit_result = subprocess.run(
[sys.executable, "-m", "muse", "规划", str(cfg), "编辑候选", str(edit_path)],
text=True,
capture_output=True,
)
assert edit_result.returncode == 0, edit_result.stderr
assert client.post(
f"/api/v1/plan-candidates/{id_}/versions", json=edit
).json() == json.loads(edit_result.stdout)
assert client.get(f"/api/v1/plan-candidates/{id_}", params={"version": 1}).json()[
"content"
] == {"名称": "下雪以前"}
candidate = client.get(f"/api/v1/plan-candidates/{id_}").json()
location.update(candidate_revision=2, candidate_hash=candidate["candidate_hash"])
review = client.post(f"/api/v1/plan-candidates/{id_}/reviews", json=location)
assert review.status_code == 200, review.text
decision = {
"command_id": "cli-accept-plan",
"candidate": location,
"action": "adopt",
"author_review_id": review.json()["review_id"],
"review_hash": review.json()["review_hash"],
"approved_changes": ["plan"],
}
path = tmp_path / "决定.json"
path.write_text(json.dumps(decision, ensure_ascii=False))
result = subprocess.run(
[sys.executable, "-m", "muse", "规划", str(cfg), "决定", str(path)],
text=True,
capture_output=True,
)
assert result.returncode == 0, result.stderr
readback = client.get(f"/api/v1/plans/{candidate['plan_id']}")
assert readback.status_code == 200 and readback.json()["content"] == {"名称": "下雪后的信"}
assert client.post(
f"/api/v1/plan-candidates/{id_}/decisions", json=decision
).json() == json.loads(result.stdout)
def test_陈旧规划审阅不能覆盖新版本但可重新拒绝__ad1303(规划环境):
from dataclasses import replace
from muse.共享.错误 import Muse错误
app, actor, _ = 规划环境
service = app.要求规划()
schema = service.新规划结构(actor, "planned-work", "outline", 1)
locations = []
for name in ("一封信", "一封旧信"):
reply = service.保存人工候选(
actor, name, 规划保存("planned-work", 0, schema, {"名称": name})
)
item = service.读取候选(actor, reply["results"][0]["candidate_id"])
locations.append(
规划定位(item["candidate_id"], 1, item["candidate_hash"], item["plan_id"], 0)
)
old_review = service.打开审阅(actor, locations[0])
new_review = service.打开审阅(actor, locations[1])
service.决定(
actor,
"approve-new",
locations[1],
作者动作.采纳,
new_review["review_id"],
new_review["review_hash"],
("plan",),
)
with pytest.raises(Muse错误):
service.决定(
actor,
"approve-stale",
locations[0],
作者动作.采纳,
old_review["review_id"],
old_review["review_hash"],
("plan",),
)
current = service.读取规划(actor, locations[1].plan_id)
assert current["revision"] == 1 and current["content"]["名称"] == "一封旧信"
rejected = replace(locations[0], expected_revision=1)
review = service.打开审阅(actor, rejected)
result = service.决定(
actor, "reject-old", rejected, 作者动作.拒绝, review["review_id"], review["review_hash"], ()
)
assert result["action"] == "reject"
assert service.读取规划(actor, locations[1].plan_id) == current
def 保存并确认(service, actor, request, command):
reply = service.保存人工候选(actor, command + "-draft", request)
item = service.读取候选(actor, reply["results"][0]["candidate_id"])
location = 规划定位(
item["candidate_id"],
item["revision"],
item["candidate_hash"],
item["plan_id"],
request.expected_revision,
)
review = service.打开审阅(actor, location)
service.决定(
actor,
command + "-accept",
location,
作者动作.采纳,
review["review_id"],
review["review_hash"],
("plan",),
)
return service.读取规划(actor, item["plan_id"])
def test_已确认细纲返回字典并检验当前来源__e1f9cb(规划环境):
from dataclasses import replace
from muse.共享.错误 import Muse错误
app, actor, _ = 规划环境
service = app.要求规划()
for i in range(1, 4):
app.要求作品().添加章节(
actor, f"chapter-{i}", "planned-work", f"chapter-{i}", f"第{i}章", 预期目录版本=i - 1
)
outline_schema = service.新规划结构(actor, "planned-work", "outline", 1)
outline = 规划保存("planned-work", 0, outline_schema, {"名称": "原大纲"})
parent = 保存并确认(service, actor, outline, "parent")
with pytest.raises(Muse错误) as missing:
service.读取可用细纲(actor, "planned-work", "chapter-3")
assert missing.value.错误码 == "PRECONDITION_PENDING"
schema = service.新规划结构(actor, "planned-work", "fine_outline", 1)
payload = 细纲内容(parent["plan_id"])
req = 规划保存("planned-work", 0, schema, payload, chapter_id="chapter-3")
first = 保存并确认(service, actor, req, "fine")
loaded = service.读取可用细纲(actor, "planned-work", "chapter-3")
assert isinstance(loaded["content"], dict) and loaded["content"]["targetChapter"] == 3
assert loaded["content"]["chapterGoal"] == "x" and loaded["revision"] == 1
second = 保存并确认(
service,
actor,
replace(req, expected_revision=1, content={**payload, "chapterGoal": "x2"}),
"fine-v2",
)
assert first["plan_id"] == second["plan_id"]
assert (
service.读取可用细纲(actor, "planned-work", "chapter-3")["content"]["chapterGoal"] == "x2"
)
保存并确认(
service,
actor,
replace(outline, expected_revision=1, content={"名称": "新的大纲"}),
"parent-v2",
)
with pytest.raises(Muse错误) as stale:
service.读取可用细纲(actor, "planned-work", "chapter-3")
assert stale.value.错误码 == "SOURCE_STALE"
def test_规划动态引用由故事世界核实并纳入提交依赖__ad1304(规划环境):
from muse.元数据.接口 import 作品扩展, 值规则, 字段定义, 引用规则
from muse.共享.错误 import Muse错误
app, actor, db = 规划环境
建立人物(app, actor)
with db[用途.维护].连接() as 连, 连.transaction():
extension = 作品扩展(
"planned-work",
"outline",
1,
1,
(
字段定义(
"outline.concern",
"牵挂的人",
"牵挂的人",
值规则("reference", 引用=引用规则(("character",))),
),
),
)
元数据服务(连).发布扩展(extension, extension.内容哈希)
service = app.要求规划()
schema = service.新规划结构(actor, "planned-work", "outline", 1, 扩展版本=1)
bad = 规划保存("planned-work", 0, schema, {"名称": "找信", "牵挂的人": "missing-person"})
with pytest.raises(Muse错误) as missing:
service.保存人工候选(actor, "missing-reference", bad)
assert missing.value.错误码 == "FACT_REFERENCE_INVALID"
saved = 保存并确认(
service,
actor,
规划保存("planned-work", 0, schema, {"名称": "找信", "牵挂的人": "postman"}),
"linked-plan",
)
assert saved["content"]["牵挂的人"] == "postman"
assert {"种类": "B02.object", "对象ID": "world:postman", "版本": "1"} in saved[
"source_dependencies"
]
def 细纲内容(parent_id):
return {
"targetChapter": 3,
"sourceRef": {"sourceId": parent_id, "sourceVersion": 1},
"chapterGoal": "x",
"hardConstraints": [],
"keyEvents": [
{
"id": "letter",
"order": 1,
"event": "邮差走到门口",
"participants": [],
"trigger": "收到退回的信",
"resultDirection": "确认这座房子还住着人",
}
],
"adjustableBeats": [],
"declaredNewFacts": [],
"mustAppearEntities": [],
"chapterEndHook": "门内传来了脚步声。",
}
def 准备可用细纲(规划环境):
app, actor, _ = 规划环境
service = app.要求规划()
for i in range(1, 4):
app.要求作品().添加章节(
actor, f"add-{i}", "planned-work", f"chapter-{i}", f"第{i}章", 预期目录版本=i - 1
)
parent_req = 规划保存(
"planned-work",
0,
service.新规划结构(actor, "planned-work", "outline", 1),
{"名称": "寄信者"},
)
parent = 保存并确认(service, actor, parent_req, "parent")
child_req = 规划保存(
"planned-work",
0,
service.新规划结构(actor, "planned-work", "fine_outline", 1),
细纲内容(parent["plan_id"]),
chapter_id="chapter-3",
)
child = 保存并确认(service, actor, child_req, "child")
return parent_req, parent, child_req, child
def test_规划关联按类型归属解析且不构成依据环__ad1307(规划环境):
from dataclasses import replace
app, actor, _ = 规划环境
service = app.要求规划()
parent_req, parent, child_req, child = 准备可用细纲(规划环境)
linked = 保存并确认(
service,
actor,
replace(
parent_req,
expected_revision=1,
content={"名称": "寄信者", "近三章细纲": [child["plan_id"]]},
),
"link-child",
)
assert linked["content"]["近三章细纲"] == [child["plan_id"]]
assert not any(d["对象ID"] == "plan:" + child["plan_id"] for d in linked["source_dependencies"])
refreshed = replace(
child_req,
expected_revision=1,
content={
**child_req.content,
"sourceRef": {"sourceId": parent["plan_id"], "sourceVersion": 2},
},
)
保存并确认(service, actor, refreshed, "refresh-child")
assert service.读取可用细纲(actor, "planned-work", "chapter-3")["revision"] == 2
@pytest.mark.parametrize(
"policy", ["deprecated", "writing-revoked"], ids=["deprecated", "writing-revoked"]
)
def test_细纲必需字段撤回用途后不能用于写作__ad1308(规划环境, policy):
from dataclasses import asdict
from muse.元数据.接口 import 启用命令, 字段限制, 定义哈希, 策略快照
from muse.共享.错误 import Muse错误
app, actor, db = 规划环境
_, _, _, child = 准备可用细纲(规划环境)
with db[用途.维护].连接() as 连, 连.transaction():
rule = (
字段限制("fine_outline.chapterGoal", (), True)
if policy == "deprecated"
else 字段限制("fine_outline.chapterGoal", ("aiContext:generation",))
)
version = 策略快照("fine_outline", 1, (rule,))
元数据服务(连).更新策略(
version,
启用命令("restrict-plan", 定义哈希(asdict(version)), 0, "author", "type:fine_outline"),
)
# 历史仍可查看;严格可用读取不把裁剪后缺失的章目标当作完整细纲。
assert app.要求规划().读取规划(actor, child["plan_id"])["revision"] == 1
with pytest.raises(Muse错误) as rejected:
app.要求规划().读取可用细纲(actor, "planned-work", "chapter-3")
assert rejected.value.错误码 == "PRECONDITION_PENDING"
def 建立人物(app, actor):
from muse.故事世界.接口 import 事实定位, 事实提案请求
world = app.要求故事世界()
structure = world.新事实结构(actor, "planned-work", "postman", "character", 1)
result = world.提出事实(
actor,
"postman-fact",
事实提案请求(
"planned-work", "postman", 0, structure, {"名称": "邮差"}, (), None, independent=True
),
)
item = world.读取提案(actor, result["results"][0]["proposal_id"])
location = 事实定位(item["proposal_id"], 1, item["proposal_hash"], "world:postman", 0)
review = world.打开事实审阅(actor, location)
world.决定事实(
actor,
"confirm-postman",
location,
作者动作.采纳,
review["review_id"],
review["review_hash"],
("object",),
)
@pytest.mark.parametrize("形态", ["string", "object"])
def test_细纲两种实体形式都核对真实本书引用__ad1305(规划环境, 形态):
from dataclasses import replace
from muse.共享.错误 import Muse错误
app, actor, _ = 规划环境
建立人物(app, actor)
for i in range(1, 4):
app.要求作品().添加章节(
actor,
f"ref-chapter-{i}",
"planned-work",
f"chapter-{i}",
f"第{i}章",
预期目录版本=i - 1,
)
service = app.要求规划()
parent = 保存并确认(
service,
actor,
规划保存(
"planned-work",
0,
service.新规划结构(actor, "planned-work", "outline", 1),
{"名称": "找信"},
),
"reference-parent",
)
content = 细纲内容(parent["plan_id"])
value = (
"postman" if 形态 == "string" else {"id": "postman", "type": "character", "name": "邮差"}
)
content["mustAppearEntities"] = [value]
schema = service.新规划结构(actor, "planned-work", "fine_outline", 1)
req = 规划保存("planned-work", 0, schema, content, chapter_id="chapter-3")
missing = (
"missing" if 形态 == "string" else {"id": "missing", "type": "character", "name": "邮差"}
)
with pytest.raises(Muse错误) as error:
service.保存人工候选(
actor,
"invalid-entity",
replace(req, content={**content, "mustAppearEntities": [missing]}),
)
assert error.value.错误码 == "FACT_REFERENCE_INVALID"
保存并确认(service, actor, req, "referenced-fine")
loaded = service.读取可用细纲(actor, "planned-work", "chapter-3")
assert loaded["content"]["mustAppearEntities"] == [value]
assert {"种类": "B02.object", "对象ID": "world:postman", "版本": "1"} in loaded[
"source_dependencies"
]