muse-agent-example/tests/集成/test_章后失败与重建.py

"""章后派生失败与重建：两作品连续两章、抽取失败恢复、旧版本派生拒绝与重建幂等。"""

import pytest
from pydantic import TypeAdapter

from muse.共享.错误 import Muse错误
from muse.故事世界.接口 import 事实定位
from muse.正式变更.接口 import 作者动作
from muse.正文写作.接口 import 正文草稿

pytestmark = pytest.mark.数据库


def _事实(引文: str, object_id: str = "lin-shen") -> list[dict]:
    return [
        {
            "type": "character",
            "object_id": object_id,
            "quote": 引文,
            "content": {"名称": "林深"},
        }
    ]


def _确认提案(环境, 提案ID: str, object_id: str) -> None:
    世界 = 环境["装配"].要求故事世界()
    作者 = 环境["作者"]
    提案 = 世界.读取提案(作者, 提案ID)
    定位 = 事实定位(
        提案["proposal_id"], 提案["revision"], 提案["proposal_hash"], "world:" + object_id, 0
    )
    审阅 = 世界.打开事实审阅(作者, 定位)
    世界.决定事实(
        作者,
        f"confirm-{object_id}-{提案ID[:8]}",
        定位,
        作者动作.采纳,
        审阅["review_id"],
        审阅["review_hash"],
        ("object",),
    )


@pytest.mark.case_id(
    "NC-w16-w16c01",
    environment="真实隔离 PG + 合成抽取宿主",
    given="隔离数据库中的作品与已采纳正文",
    when="两部作品各完成保存→章后派生→确认事实；确认事实进入正确时点且不跨作品",
    then=["派生任务步骤完成", "确认事实可查询", "跨作品不可见"],
    contract="docs/系统架构/新版设计/功能规格/P05-章后处理与事实复核.md",
)
def test_两作品各连续两章派生确认事实并按时点消费__16c01a(章后环境, 章后工具):
    环境 = 章后环境
    环境["新作品"]("post-work-b", 2, "b-ch-")

    环境["剧本"].append(章后工具.抽取输出("第一章：林深滞留码头。", _事实("林深把伞收在了门边")))
    任务1, _ = 章后工具.保存一章("ch-1", "林深把伞收在了门边，雨还没停。")
    快照1 = 章后工具.跑任务(任务1)
    assert 快照1.状态.value == "completed"
    检查点1 = {s["step_id"]: (s.get("checkpoint") or {}) for s in 快照1.步骤}
    索引1 = 检查点1["登记摘要与索引"]["索引"]
    assert 索引1["revision"] == 1 and 索引1["document_hash"]
    提案1 = 检查点1["登记事实提案"]["提案"]
    assert len(提案1) == 1 and not 检查点1["登记事实提案"]["跳过"]
    _确认提案(环境, 提案1[0]["proposal_id"], "lin-shen")

    环境["剧本"].append(章后工具.抽取输出("第二章：林深离开码头。", []))
    任务2, _ = 章后工具.保存一章("ch-2", "林深沿着码头走了出去。")
    assert 章后工具.跑任务(任务2).状态.value == "completed"

    # 连续写作按正确时点消费确认事实：截至第一章的世界状态包含已确认事实。
    世界 = 环境["装配"].要求故事世界()
    时点 = 世界.查询世界(环境["作者"], "post-work", "ch-1")
    对象 = {o["object_id"]: o for o in 时点["facts"]}
    assert 对象["lin-shen"]["content"]["名称"] == "林深"

    # 第二部作品同旅程；确认事实不跨作品。
    环境["剧本"].append(章后工具.抽取输出("B 第一章。", []))
    任务B, _ = 章后工具.保存一章("b-ch-1", "B 章文本。")
    assert 章后工具.跑任务(任务B).状态.value == "completed"
    时点B = 世界.查询世界(环境["作者"], "post-work-b", "b-ch-1")
    assert "lin-shen" not in {o["object_id"] for o in 时点B["facts"]}


@pytest.mark.case_id(
    "NC-w16-w16c02",
    environment="真实隔离 PG + 合成抽取宿主",
    given="隔离数据库中的作品与已采纳正文",
    when="抽取回合失败时任务失败而正文保留；恢复重验通过后完成",
    then=["正文版本保留", "步骤失败可见", "恢复后完成"],
    contract="docs/系统架构/新版设计/功能规格/P05-章后处理与事实复核.md",
)
def test_抽取失败后正文可用且恢复重试完成__16c02b(章后环境, 章后工具):
    环境 = 章后环境
    from muse.任务运行.接口 import 任务状态

    # 输出合同不满足：调用已结算但终态无效 → 步骤失败、任务失败（无未知调用）。
    环境["剧本"].append({"类型": "文本", "文本": {"summary": "只有摘要"}})
    任务1, 回执 = 章后工具.保存一章("ch-1", "林深把伞收在了门边。")
    with pytest.raises((Muse错误, RuntimeError, OSError)):
        章后工具.跑任务(任务1)
    终态 = 环境["装配"].任务运行.读取任务(任务1)
    assert 终态.状态 == 任务状态.已失败
    抽取步 = next(s for s in 终态.步骤 if s["step_id"] == "抽取提案")
    assert 抽取步["state"] == "failed"
    # 正文成功保存：派生失败不回滚作者正文。
    正文 = 环境["装配"].要求正文().读取正文(环境["作者"], "ch-1")
    assert 正文["revision"] == 1
    assert TypeAdapter(正文草稿).validate_python(正文["document"])

    环境["剧本"].append(章后工具.抽取输出("恢复后的摘要。", []))
    环境["装配"].任务运行.控制任务(
        任务1, 环境["作者"].作者, 任务状态.已失败, "恢复", 命令ID=f"resume-{任务1[:8]}"
    )
    快照 = 章后工具.跑任务(任务1)
    assert 快照.状态.value == "completed"
    assert 回执["receipt_id"]


@pytest.mark.case_id(
    "NC-w16-w16c03",
    environment="真实隔离 PG",
    given="隔离数据库中的作品与已采纳正文",
    when="正文前进后旧派生任务被来源复检拒绝，新版本任务正常完成",
    then=["旧任务失败于来源复检", "新任务绑定新版本"],
    contract="docs/系统架构/新版设计/功能规格/P05-章后处理与事实复核.md",
)
def test_编辑后旧版本派生被拒绝且按新版本重建__16c03c(章后环境, 章后工具):
    环境 = 章后环境
    # 保存 r1 不跑派生；编辑出 r2 后再跑 r1 任务，必须被来源复检拒绝。
    任务1, _ = 章后工具.保存一章("ch-1", "第一版正文，林深在码头。")
    # r1 在来源复检即失败，不消费抽取剧本；r2 消费队头的新版本摘要。
    环境["剧本"].append(章后工具.抽取输出("新版本摘要。", []))
    任务2, _ = 章后工具.保存一章("ch-1", "第二版正文，林深在码头整理行装。", 预期版本=1)
    with pytest.raises((Muse错误, RuntimeError, OSError)):
        章后工具.跑任务(任务1)
    快照1 = 环境["装配"].任务运行.读取任务(任务1)
    assert 快照1.状态.value == "failed"
    assert any(s["step_id"] == "来源复检" and s["state"] == "failed" for s in 快照1.步骤), (
        快照1.步骤
    )

    快照2 = 章后工具.跑任务(任务2)
    assert 快照2.状态.value == "completed"
    from muse.编排.章后处理 import 查询章后状态

    状态 = 查询章后状态(环境["装配"], 环境["作者"].作者, "ch-1")
    assert 状态["revision"] == 2 and 状态["摘要"] == "新版本摘要。"
    # 旧任务的提案登记不得覆盖新版本投影：旧任务未产出提案。
    旧步骤 = {s["step_id"]: s for s in 快照1.步骤}
    assert not (旧步骤["登记事实提案"].get("checkpoint") or {}).get("提案")


@pytest.mark.case_id(
    "NC-w16-w16c04",
    environment="真实隔离 PG",
    given="隔离数据库中的作品与已采纳正文",
    when="同来源版本重建派生任务不产生重复提案",
    then=["提案唯一", "摘要可回查"],
    contract="docs/系统架构/新版设计/功能规格/P05-章后处理与事实复核.md",
)
def test_重建幂等不重复事实提案__16c04d(章后环境, 章后工具):
    环境 = 章后环境
    事实 = _事实("林深在码头")
    环境["剧本"].append(章后工具.抽取输出("摘要一。", 事实))
    任务1, _ = 章后工具.保存一章("ch-1", "林深在码头。雨停了。")
    章后工具.跑任务(任务1)

    from muse.编排.章后处理 import 发起章后处理, 查询章后状态

    重建 = 发起章后处理(
        环境["装配"],
        环境["作者"],
        f"rebuild-{任务1[:8]}",
        work_id="post-work",
        chapter_id="ch-1",
    )
    环境["剧本"].append(章后工具.抽取输出("摘要一。", 事实))
    章后工具.跑任务(重建["task_id"])
    状态 = 查询章后状态(环境["装配"], 环境["作者"].作者, "ch-1")
    assert 状态["摘要"] == "摘要一。"
    with 环境["库"].连接(只读=True) as 连:
        数 = 连.execute(
            "SELECT count(DISTINCT p.proposal_id) FROM muse_fact_proposal p "
            "WHERE p.work_id=%s AND p.object_id=%s",
            ("post-work", "lin-shen"),
        ).fetchone()[0]
        assert 数 == 1, "重建不得重复事实提案"