muse-agent-example/tests/集成/test_案例采集与复核.py

"""案例的真实来源、许可与确认，不把元信息声明当原文授权。"""

import hashlib
from concurrent.futures import ThreadPoolExecutor

import pytest

from muse.共享.调用身份 import 用途
from muse.资料研究.接口 import 导入请求, 资料服务, 资料错误

pytestmark = pytest.mark.数据库


@pytest.mark.case_id(
    "NC-w24-240001",
    environment="隔离PG与合成资料",
    given="参考来源版本与作者归属",
    when="核对sample许可",
    then=["无sample不返回片段，跨作者拒绝，归属固定"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
def test_参考作品归属稳定且无sample许可只返回哈希__240001(应用测试库):
    pool = 应用测试库[用途.生产]
    svc = 资料服务()
    text = "仅作案例的合成文字。"
    request = 导入请求(
        "reference", "合成书", "synthetic:reference-a", text, ("analysis",), "synthetic:book-a"
    )
    with pool.连接() as conn, conn.transaction():
        source = svc.导入(conn, "case-author", request)
        assert svc.导入(
            conn,
            "case-author",
            导入请求("reference", "合成书", request.origin, text, ("analysis",)),
        ).重复
        evidence = svc.读取案例证据(conn, "case-author", source.source_id, 1, 0, 5)
        assert evidence["excerpt"] is None and not evidence["text_allowed"]
        assert evidence["excerpt_hash"] == hashlib.sha256(text[:5].encode()).hexdigest()
        assert evidence["work_group"] == "synthetic:book-a"
        with pytest.raises(资料错误):
            svc.读取案例证据(conn, "other", source.source_id, 1, 0, 5)
        # 用途集合必须与原来源一致：异用途重导入先被“重导入不能修改来源用途”拒绝，
        # 这里只验证归属固定，故沿用原 analysis 授权。
        with pytest.raises(资料错误, match="另一参考作品"):
            svc.导入(
                conn,
                "case-author",
                导入请求(
                    "reference",
                    "新标题",
                    request.origin,
                    text,
                    ("analysis",),
                    "synthetic:other-book",
                ),
            )
        with pytest.raises(资料错误, match="未授权"):
            svc.核对来源授权(conn, source.source_id, "sample")
    with pool.连接() as conn, conn.transaction():
        new = svc.导入(
            conn,
            "case-author",
            导入请求(
                "reference",
                "样例书",
                "synthetic:reference-sample",
                text,
                ("sample",),
                "synthetic:book-b",
            ),
        )
        evidence = svc.读取案例证据(conn, "case-author", new.source_id, 1, 0, 5)
        assert evidence["excerpt"] == text[:5] and evidence["text_allowed"]


@pytest.mark.case_id(
    "NC-w24-240002",
    environment="隔离PG与合成资料",
    given="同一原文并发导入",
    when="分别增加analysis/sample授权",
    then=["同一来源版本，授权不丢失，不能静默变更参考书归属"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
def test_导入串行保留两种授权且同内容不能静默补归属__240002(应用测试库):
    pool = 应用测试库[用途.生产]
    svc = 资料服务()
    text = "并发导入的同一原文。"

    def add(use):
        try:
            with pool.连接() as conn, conn.transaction():
                return svc.导入(
                    conn,
                    "case-author",
                    导入请求(
                        "reference",
                        "同一本书",
                        "synthetic:same-origin",
                        text,
                        (use,),
                        "synthetic:same-book",
                    ),
                )
        except 资料错误 as 错:
            return 错

    # 授权只能经具名变更命令维护：并发异用途导入只有一个成功，另一个被拒绝，
    # 不允许把两次请求的用途并集成“只增不减”的旧行为。
    with ThreadPoolExecutor(max_workers=2) as executor:
        结果 = list(executor.map(add, ["analysis", "sample"]))
    成功 = [
        (用, r)
        for 用, r in zip(("analysis", "sample"), 结果, strict=True)
        if not isinstance(r, Exception)
    ]
    失败 = [r for r in 结果 if isinstance(r, Exception)]
    assert len(成功) == 1 and len(失败) == 1
    胜出用途, a = 成功[0]
    assert a.revision == 1 and isinstance(失败[0], 资料错误)
    with pool.连接() as conn, conn.transaction():
        assert set(svc.核对来源授权(conn, a.source_id, 胜出用途)["authorized_uses"]) == {胜出用途}
        old = svc.导入(
            conn,
            "case-author",
            导入请求("reference", "旧未分类资料", "synthetic:unknown-work", text, ("analysis",)),
        )
        with pytest.raises(资料错误, match="重导入不能修改"):
            svc.导入(
                conn,
                "case-author",
                导入请求(
                    "reference",
                    "旧未分类资料",
                    "synthetic:unknown-work",
                    text,
                    ("sample",),
                    "synthetic:claimed-book",
                ),
            )
        with pytest.raises(资料错误, match="未授权"):
            svc.核对来源授权(conn, old.source_id, "sample")


@pytest.fixture
def 案例环境(章后环境):
    from muse.审校修订.接口 import 案例来源, 案例采集请求

    app, author = 章后环境["装配"], 章后环境["作者"]
    text = "屋里只有甲。他看着窗外。雨落在街上。"

    def source(origin="synthetic:quality-case", uses=("sample",), preserve=True):
        with app.要求数据库().连接() as conn, conn.transaction():
            row = app.要求资料().导入(
                conn,
                author.作者,
                导入请求("reference", "合成案例来源", origin, text, uses, "synthetic:book-a"),
            )
        loc = 案例来源(
            "source",
            row.source_id,
            row.revision,
            row.content_hash,
            0,
            6,
            hashlib.sha256(text[:6].encode()).hexdigest(),
        )
        return 案例采集请求(loc, preserve_text=preserve)

    return app, author, source, text


def _采集(app, author, command, request):
    from muse.审校修订.接口 import 案例定位

    result = app.要求审校().采集案例(author, command, request)["results"][0]
    return 案例定位(result["case_id"], result["revision"], result["case_hash"])


def _复核请求(loc, label="sf"):
    from muse.审校修订.接口 import 案例复核请求

    return 案例复核请求(
        loc, label, "retain" if label == "snf" else "repair", "人工阅读真实片段后的分类。"
    )


def _确认(app, author, loc, command="confirm-case"):
    svc = app.要求审校()
    req = _复核请求(loc)
    review = svc.打开案例审阅(author, req)
    return svc.决定案例(
        author, command, req, 审阅ID=review["review_id"], 审阅哈希=review["review_hash"]
    )


@pytest.mark.case_id(
    "NC-w24-240003",
    environment="隔离PG与合成资料",
    given="无sample或无显式保全同意",
    when="采集与尝试复核",
    then=["只保留元信息，不能确认，命令回执不含原文"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
@pytest.mark.parametrize("sample,preserve", [(False, True), (True, False)])
def test_未授权案例仅元信息且不能确认__240003(案例环境, sample, preserve):
    import json

    from muse.共享.错误 import Muse错误

    app, author, source, text = 案例环境
    req = source(uses=("sample",) if sample else ("analysis",), preserve=preserve)
    loc = _采集(app, author, "capture-metadata", req)
    svc = app.要求审校()
    row = svc.读取案例(author, loc.case_id)
    assert row["state"] == "shadow" and not row["sampleable"]
    assert row["payload"]["evidence"]["before"]["excerpt"] is None
    assert row["payload"]["evidence"]["before"]["context"] is None
    with pytest.raises(Muse错误, match="哈希元信息"):
        svc.打开案例审阅(author, _复核请求(loc))
    with app.要求数据库().连接(只读=True) as conn:
        stored = conn.execute(
            "SELECT payload FROM muse_quality_case_version WHERE case_id=%s", (loc.case_id,)
        ).fetchone()[0]
        commands = conn.execute(
            "SELECT to_jsonb(c) FROM muse_change_command c WHERE command_id='capture-metadata'"
        ).fetchone()[0]
    assert text[:6] not in json.dumps([stored, commands], ensure_ascii=False)
    assert len(svc.列出案例(author)) == 1


@pytest.mark.case_id(
    "NC-w24-240004",
    environment="隔离PG与合成资料",
    given="伪片段哈希、源哈希、越界或额外正文",
    when="尝试采集",
    then=["拒绝且无S01命令或案例残留"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
@pytest.mark.parametrize("bad", ["quote", "source", "bounds", "extra_text"])
def test_伪引文或非法范围在采集前拒绝__240004(案例环境, bad):
    from dataclasses import asdict, replace

    from pydantic import TypeAdapter, ValidationError

    from muse.共享.错误 import Muse错误
    from muse.审校修订.接口 import 案例采集请求

    app, author, source, _ = 案例环境
    req = source()
    with pytest.raises((Muse错误, ValidationError)):
        if bad == "extra_text":
            TypeAdapter(案例采集请求).validate_python(
                {**asdict(req), "before_excerpt": "不允许输入正文"}
            )
        else:
            changes = (
                {"excerpt_hash": "0" * 64}
                if bad == "quote"
                else {"source_hash": "0" * 64}
                if bad == "source"
                else {"end": 900}
            )
            app.要求审校().采集案例(
                author, "bad-case", replace(req, before=replace(req.before, **changes))
            )
    assert app.要求审校().列出案例(author) == []
    with app.要求数据库().连接(只读=True) as conn:
        assert (
            conn.execute(
                "SELECT count(*) FROM muse_change_command WHERE command_id='bad-case'"
            ).fetchone()[0]
            == 0
        )


@pytest.mark.case_id(
    "NC-w24-240005",
    environment="隔离PG与合成资料",
    given="待复核案例",
    when="伪审阅拒绝，真实确认后撤回",
    then=["幂等、当前停止复用、历史不变，版本UPDATE拒绝"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
def test_确认需真实审阅且撤回不改历史__240005(案例环境):
    import psycopg

    from muse.共享.错误 import Muse错误
    from muse.审校修订.接口 import 案例定位

    app, author, source, text = 案例环境
    req = source()
    svc = app.要求审校()
    loc = _采集(app, author, "capture-one", req)
    assert _采集(app, author, "capture-one", req) == loc
    assert not svc.读取案例(author, loc.case_id)["sampleable"]
    with pytest.raises(Muse错误):
        svc.决定案例(
            author,
            "fake-review",
            _复核请求(loc),
            审阅ID="00000000-0000-0000-0000-000000000000",
            审阅哈希="0" * 64,
        )
    receipt = _确认(app, author, loc)
    saved = svc.读取案例(author, loc.case_id)
    assert saved["state"] == "confirmed" and saved["sampleable"]
    assert saved["revision"] == 2 and saved["payload"]["evidence"]["before"]["excerpt"] == text[:6]
    assert receipt["action"] == "adopt"
    target = 案例定位(loc.case_id, saved["revision"], saved["case_hash"])
    svc.撤回案例(author, "withdraw", target)
    current = svc.读取案例(author, loc.case_id)
    assert current["state"] == "withdrawn" and not current["sampleable"]
    old = svc.读取案例(author, loc.case_id, revision=2)
    assert old["payload"] == saved["payload"] and old["case_hash"] == saved["case_hash"]
    assert not old["sampleable"]
    with app.要求数据库().连接() as conn:
        with pytest.raises(psycopg.Error), conn.transaction():
            conn.execute(
                "UPDATE muse_quality_case_version SET state='shadow' WHERE case_id=%s",
                (loc.case_id,),
            )


@pytest.mark.case_id(
    "NC-w24-240006",
    environment="隔离PG与合成资料",
    given="已打开审阅后来源追加版本",
    when="旧审阅确认与重开历史审阅",
    then=["拒绝旧批准，新审阅可确认明确历史片段"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
def test_来源推进阻止旧审阅但可重新审历史__240006(案例环境):
    from muse.共享.错误 import Muse错误

    app, author, source, text = 案例环境
    req = source()
    svc = app.要求审校()
    loc = _采集(app, author, "capture-old", req)
    review = svc.打开案例审阅(author, _复核请求(loc))
    with app.要求数据库().连接() as conn, conn.transaction():
        app.要求资料().导入(
            conn,
            author.作者,
            导入请求(
                "reference",
                "合成案例来源",
                "synthetic:quality-case",
                text + "门开了。",
                ("sample",),
            ),
        )
    with pytest.raises(Muse错误):
        svc.决定案例(
            author,
            "stale-review",
            _复核请求(loc),
            审阅ID=review["review_id"],
            审阅哈希=review["review_hash"],
        )
    assert svc.读取案例(author, loc.case_id)["state"] == "shadow"
    _确认(app, author, loc)
    assert svc.读取案例(author, loc.case_id)["sampleable"]


@pytest.mark.case_id(
    "NC-w24-240007",
    environment="隔离PG与合成资料",
    given="已确认与待复核案例",
    when="撤回来源sample许可",
    then=["停止确认和例证投影，隐藏原文与说明，仍可撤回"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
def test_许可撤回即时隐藏原文且阻止确认__240007(案例环境):
    from muse.共享.错误 import Muse错误
    from muse.审校修订.接口 import 案例定位

    app, author, source, _ = 案例环境
    req = source()
    svc = app.要求审校()
    first = _采集(app, author, "confirmed-before-revoke", req)
    _确认(app, author, first)
    second = _采集(app, author, "pending-before-revoke", req)
    review = svc.打开案例审阅(author, _复核请求(second))
    with app.要求数据库().连接() as conn, conn.transaction():
        conn.execute(
            "UPDATE muse_source SET authorized_uses=%s::jsonb WHERE source_id=%s",
            ('["analysis"]', req.before.ref_id),
        )
    with pytest.raises(Muse错误):
        svc.决定案例(
            author,
            "revoked-case",
            _复核请求(second),
            审阅ID=review["review_id"],
            审阅哈希=review["review_hash"],
        )
    row = svc.读取案例(author, first.case_id)
    assert row["state"] == "confirmed" and not row["sampleable"]
    assert row["payload"]["evidence"]["before"]["excerpt"] is None
    assert row["payload"]["review_note"] is None
    svc.撤回案例(
        author, "withdraw-unlicensed", 案例定位(first.case_id, row["revision"], row["case_hash"])
    )
    assert svc.读取案例(author, first.case_id)["state"] == "withdrawn"


@pytest.mark.case_id(
    "NC-w24-240008",
    environment="隔离PG与合成资料",
    given="真实B05前后正文或被拒候选",
    when="核验S01回执后记录反馈",
    then=["同章同分支同基线，假回执拒绝，不再写正文"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
@pytest.mark.parametrize("after_kind", ["candidate", "document"])
def test_真实反馈绑定确切正文候选与S01回执__240008(案例环境, after_kind):
    from dataclasses import replace

    from muse.作品规划.接口 import 档案保存
    from muse.共享.错误 import Muse错误
    from muse.审校修订.接口 import 案例来源, 案例采集请求
    from muse.正式变更.接口 import 作者动作
    from muse.正文写作.接口 import 候选定位, 文本节点, 正文草稿, 段落

    app, author, _, _ = 案例环境
    works, body, cases = app.要求作品(), app.要求正文(), app.要求审校()
    schema = works.新档案结构(author, "case-work", "work_core", 1)
    works.保存档案(
        author, "create-case-work", 档案保存("case-work", 0, {"名称": "案例作品"}, schema)
    )
    works.添加章节(author, "create-case-ch", "case-work", "case-ch", "第一章", 预期目录版本=0)
    original, changed = "甲走进了门里。", "甲推开门走进去。"

    def draft(text):
        return 正文草稿((段落("p1", (文本节点(text),)),))

    body.保存人工(author, "before-text", "case-ch", 0, draft(original))
    before = body.读取正文(author, "case-ch")
    if after_kind == "candidate":
        candidate = body.创建人工候选(author, "after-candidate", "case-ch", 1, draft(changed))[
            "results"
        ][0]
        row = body.读取候选(author, candidate["candidate_id"])
        loc = 候选定位(
            row["candidate_id"], row["revision"], row["candidate_hash"], before["document_id"], 1
        )
        review = body.打开候选审阅(author, loc)
        receipt = body.决定候选(
            author,
            "choose-original",
            loc,
            作者动作.拒绝,
            review["review_id"],
            review["review_hash"],
            (),
        )
        from muse.正文写作.接口 import 正文结构哈希

        after = 案例来源(
            "candidate",
            row["candidate_id"],
            row["revision"],
            正文结构哈希(draft(changed)),
            0,
            len(changed),
            hashlib.sha256(changed.encode()).hexdigest(),
            "main",
        )
    else:
        receipt = body.保存人工(author, "after-text", "case-ch", 1, draft(changed))
        row = body.读取正文(author, "case-ch")
        after = 案例来源(
            "document",
            "case-ch",
            2,
            row["document_hash"],
            0,
            len(changed),
            hashlib.sha256(changed.encode()).hexdigest(),
            "main",
        )
    request = 案例采集请求(
        案例来源(
            "document",
            "case-ch",
            1,
            before["document_hash"],
            0,
            len(original),
            hashlib.sha256(original.encode()).hexdigest(),
            "main",
        ),
        after,
        mode="live_feedback",
        preserve_text=True,
        feedback_command_id=receipt["command_id"],
    )
    with pytest.raises(Muse错误):
        cases.采集案例(author, "wrong-receipt", replace(request, feedback_command_id="before-text"))
    target = _采集(app, author, "actual-feedback", request)
    saved = cases.读取案例(author, target.case_id)
    assert saved["payload"]["feedback"] == {
        k: receipt[k] for k in ("command_id", "receipt_id", "action")
    }
    assert saved["payload"]["evidence"]["before"]["excerpt"] == original
    assert saved["payload"]["evidence"]["after"]["excerpt"] == changed
    assert saved["payload"]["evidence"]["before"]["work_group"] == "work:case-work"
    assert not saved["sampleable"]
    _确认(app, author, target)
    assert cases.读取案例(author, target.case_id)["sampleable"]
    assert body.读取正文(author, "case-ch")["revision"] == (1 if after_kind == "candidate" else 2)


@pytest.fixture
def 四类案例(案例环境):
    from muse.审校修订.接口 import 案例定位, 案例来源, 案例采集请求

    app, author, _, _ = 案例环境
    svc = app.要求审校()

    def group(works=("book-a", "book-b"), labels=("sf", "snf", "boundary", "regression")):
        refs = []
        for i, label in enumerate(labels):
            text = f"第{i}份例证。前半句作为证据，后半句保留语境。"
            with app.要求数据库().连接() as conn, conn.transaction():
                row = app.要求资料().导入(
                    conn,
                    author.作者,
                    导入请求(
                        "reference",
                        "已声明书目",
                        f"synthetic:rule-source:{i}",
                        text,
                        ("sample",),
                        works[i % len(works)],
                    ),
                )
            source = 案例来源(
                "source",
                row.source_id,
                1,
                row.content_hash,
                0,
                7,
                hashlib.sha256(text[:7].encode()).hexdigest(),
            )
            c = _采集(app, author, f"sample-case-{i}", 案例采集请求(source, preserve_text=True))
            r = _复核请求(c, label)
            review = svc.打开案例审阅(author, r)
            svc.决定案例(
                author,
                f"confirm-{i}",
                r,
                审阅ID=review["review_id"],
                审阅哈希=review["review_hash"],
            )
            case = svc.读取案例(author, c.case_id)
            refs.append(案例定位(c.case_id, case["revision"], case["case_hash"]))
        return tuple(refs)

    return app, author, group


def _规则草案():
    return {
        "id": "case-rule",
        "name": "待评的案例规则",
        "layer": "semantic",
        "carrier_scope": "all",
        "trigger": {"type": "model_judgment", "criteria": "根据片段前后语境核对功能是否重复。"},
        "default_disposition": "advisory",
        "fix_hint": "先判断功能，保留有叙事作用的重复。",
        "version": 1,
    }


@pytest.mark.case_id(
    "NC-w24-240009",
    environment="隔离PG与合成资料",
    given="四类跨书确认案例",
    when="提出规则、颠倒输入重放及撤回案例",
    then=["同一候选幂等，例证真实，撤回停止消费，未启用"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
def test_四类跨书确认案例才能提出可追溯候选__240009(四类案例):
    from dataclasses import replace

    from muse.共享.错误 import Muse错误
    from muse.审校修订.接口 import 案例规则请求

    app, author, group = 四类案例
    refs = group()
    svc = app.要求审校()
    req = 案例规则请求(_规则草案(), refs)
    result = svc.提出案例规则(author, req)
    assert result["status"] == "candidate" and not result["duplicate"]
    assert svc.提出案例规则(author, replace(req, cases=tuple(reversed(refs))))["duplicate"]
    library = svc.读取规则库(author, 预览候选=True)
    rule = library["rules"]["case-rule"]
    assert rule["status"] == "candidate" and set(rule["samples"]) == {
        "sf",
        "snf",
        "boundary",
        "regression",
    }
    assert set(rule["case_card_ids"]) == {r.case_id for r in refs}
    assert svc.读取规则库(author)["rules"] == {}
    for sample in library["samples"].values():
        case = svc.读取案例(author, sample["case_card_id"])
        assert sample["text"] == case["payload"]["evidence"]["before"]["excerpt"]
        assert case["case_hash"] in sample["source_ref"]
    svc.撤回案例(author, "withdraw-rule-evidence", refs[0])
    with pytest.raises(Muse错误, match="当前已确认"):
        svc.读取规则库(author, 预览候选=True)
    with pytest.raises(Muse错误, match="当前已确认"):
        svc.提出案例规则(author, replace(req, definition={**req.definition, "version": 2}))
    with app.要求数据库().连接(只读=True) as conn:
        assert (
            conn.execute(
                "SELECT count(*) FROM muse_rule_version WHERE rule_id='case-rule'"
            ).fetchone()[0]
            == 1
        )
        assert conn.execute("SELECT count(*) FROM muse_rule_activation").fetchone()[0] == 0


@pytest.mark.case_id(
    "NC-w24-24000a",
    environment="隔离PG与合成资料",
    given="单书、未知书目或单边正例",
    when="提出规则候选",
    then=["拒绝且无规则版本"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
@pytest.mark.parametrize("bad", ["one_book", "unknown_book", "only_positive"])
def test_单书未知归属或单边案例不能形成规则__24000a(四类案例, bad):
    from muse.共享.错误 import Muse错误
    from muse.审校修订.接口 import 案例规则请求

    app, author, group = 四类案例
    refs = (
        group(works=("one-book",))
        if bad == "one_book"
        else group(works=(None,))
        if bad == "unknown_book"
        else group(labels=("sf",) * 4)
    )
    with pytest.raises(Muse错误, match="作品|四类"):
        app.要求审校().提出案例规则(author, 案例规则请求(_规则草案(), refs))
    with app.要求数据库().连接(只读=True) as conn:
        assert (
            conn.execute(
                "SELECT count(*) FROM muse_rule_version WHERE rule_id='case-rule'"
            ).fetchone()[0]
            == 0
        )


@pytest.mark.case_id(
    "NC-w24-24000b",
    environment="隔离PG与合成资料",
    given="真实HTTP会话与隔离PG",
    when="HTTP采集、CLI确认及交叉重放",
    then=["同一回执和落库案例；参考作品归属经导入端点保存"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
def test_HTTP与CLI采集确认重放共用真实回执__24000b(案例环境, tmp_path):
    import json
    from dataclasses import asdict, replace

    from fastapi.testclient import TestClient

    from muse.审校修订.接口 import 案例定位
    from muse.接入.cli.审校命令 import 运行审校命令
    from muse.接入.http.应用 import 创建应用
    from muse.配置 import 服务配置

    app, author, source, text = 案例环境
    request = {"command_id": "http-case", "request": asdict(source())}
    secret = tmp_path / "password"
    secret.write_text("synthetic-case-password")
    secret.chmod(0o600)
    app.配置 = replace(
        app.配置,
        HTTP=服务配置(
            str(secret),
            作者ID=author.作者,
            公开地址="http://testserver",
            允许来源=("http://testserver",),
        ),
    )
    http = 创建应用(app.配置)
    path = tmp_path / "case-request.json"
    with TestClient(http, headers={"origin": "http://testserver"}) as client:
        http.state.装配 = app
        assert client.post("/api/v1/quality-cases", json=request).status_code == 401
        assert (
            client.post("/api/v1/session", json={"password": secret.read_text()}).status_code == 200
        )
        response = client.post("/api/v1/quality-cases", json=request)
        assert response.status_code == 201, response.text
        path.write_text(json.dumps(request))
        assert 运行审校命令(app, "案例采集", str(path)) == response.json()
        result = response.json()["results"][0]
        loc = 案例定位(result["case_id"], 1, result["case_hash"])
        req = asdict(_复核请求(loc))
        review = client.post("/api/v1/quality-case-reviews", json=req)
        assert review.status_code == 200, review.text
        decision = {
            "command_id": "http-case-confirm",
            "request": req,
            "action": "adopt",
            "review_id": review.json()["review_id"],
            "review_hash": review.json()["review_hash"],
        }
        path.write_text(json.dumps(decision))
        receipt = 运行审校命令(app, "案例决定", str(path))
        assert client.post("/api/v1/quality-case-decisions", json=decision).json() == receipt
        current = client.get(f"/api/v1/quality-cases/{loc.case_id}").json()
        assert current["state"] == "confirmed" and current["sampleable"]
        assert current == 运行审校命令(app, "案例", loc.case_id)
        assert len(运行审校命令(app, "案例清单", None)) == 1
        source_response = client.post(
            "/api/v1/sources/import?work_id=synthetic:case",
            json={
                "kind": "reference",
                "title": "合成来源",
                "origin": "synthetic:http-group",
                "content": text,
                "authorized_uses": ["sample", "reference_reading"],
                "reference_work_ref": "synthetic:declared-book",
            },
        )
        assert source_response.status_code == 201, source_response.text
        source_id = source_response.json()["source_id"]
        source_row = client.get(f"/api/v1/sources/{source_id}/versions/1").json()
        assert source_row["import_result"]["参考作品归属"] == "synthetic:declared-book"


@pytest.mark.case_id(
    "NC-w24-24000c",
    environment="隔离PG与合成资料",
    given="同组四类跨书确认案例",
    when="创建两条不同规则并生成诊断",
    then=["两条规则及8个投影例证可共存，4个源案例身份不变，诊断可读"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
def test_同组案例支持两条规则且整库和诊断可读__24000c(四类案例):
    from muse.审校修订.接口 import 案例规则请求
    from muse.正文写作.接口 import 文本节点, 正文草稿, 段落

    app, author, group = 四类案例
    refs = group()
    svc, body = app.要求审校(), app.要求正文()
    first = _规则草案()
    svc.提出案例规则(author, 案例规则请求(first, refs))
    before = svc.读取规则库(author, 预览候选=True)
    second = {**first, "id": "case-rule-second"}
    svc.提出案例规则(author, 案例规则请求(second, refs))
    after = svc.读取规则库(author, 预览候选=True)
    assert set(after["rules"]) == {first["id"], second["id"]}
    assert len(after["samples"]) == 8
    assert after["rules"][first["id"]] == before["rules"][first["id"]]
    assert all(after["samples"][key] == value for key, value in before["samples"].items())
    assert {s["case_card_id"] for s in after["samples"].values()} == {r.case_id for r in refs}
    assert len({s["source_ref"] for s in after["samples"].values()}) == 4
    body.保存人工(
        author,
        "diagnose-shared-cases",
        "ch-1",
        0,
        正文草稿((段落("p1", (文本节点("诊断仍可组合多个案例规则。"),)),)),
    )
    doc = body.读取正文(author, "ch-1")
    target = {
        "work_id": "post-work",
        "chapter_id": "ch-1",
        "branch_id": "main",
        "revision": doc["revision"],
        "document_hash": doc["document_hash"],
    }
    result = svc.生成诊断(author, target, 预览候选=True)
    assert result["report"]["rule_library_version"] == after["fingerprint"]
    assert svc.读取诊断(author, result["diagnosis_id"])["report"] == result["report"]
    assert body.读取正文(author, "ch-1") == doc


@pytest.mark.case_id(
    "NC-w24-24000d",
    environment="隔离PG与合成资料",
    given="已保存的候选预览诊断",
    when="案例撤回或sample许可变化",
    then=["新消费拒绝，历史报告原样读回并标记当前规则依据失效"],
    contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md",
)
@pytest.mark.parametrize("change", ["withdraw", "permission"])
def test_案例依据失效后旧诊断仍可读取并标过期__24000d(四类案例, change):
    import json

    from muse.共享.错误 import Muse错误
    from muse.审校修订.接口 import 案例规则请求
    from muse.正文写作.接口 import 文本节点, 正文草稿, 段落

    app, author, group = 四类案例
    refs = group()
    svc, body = app.要求审校(), app.要求正文()
    svc.提出案例规则(author, 案例规则请求(_规则草案(), refs))
    case = svc.读取案例(author, refs[0].case_id)
    excerpt = case["payload"]["evidence"]["before"]["excerpt"]
    body.保存人工(
        author,
        "diagnose-before-revocation",
        "ch-1",
        0,
        正文草稿((段落("p1", (文本节点("我的正文只保留自己的历史报告。"),)),)),
    )
    doc = body.读取正文(author, "ch-1")
    target = {
        "work_id": "post-work",
        "chapter_id": "ch-1",
        "branch_id": "main",
        "revision": doc["revision"],
        "document_hash": doc["document_hash"],
    }
    result = svc.生成诊断(author, target, 预览候选=True)
    assert not svc.读取诊断(author, result["diagnosis_id"])["stale"]
    if change == "withdraw":
        svc.撤回案例(author, "withdraw-diagnosis-case", refs[0])
    else:
        source_id = case["payload"]["request"]["before"]["ref_id"]
        with app.要求数据库().连接() as conn, conn.transaction():
            conn.execute(
                "UPDATE muse_source SET authorized_uses=%s::jsonb WHERE source_id=%s",
                ('["analysis"]', source_id),
            )
    readback = svc.读取诊断(author, result["diagnosis_id"])
    assert readback["stale"] and any("规则依据失效" in r for r in readback["stale_reasons"])
    assert readback["report"] == result["report"]
    assert excerpt not in json.dumps(readback, ensure_ascii=False)
    with pytest.raises(Muse错误):
        svc.生成诊断(author, target, 预览候选=True)
    assert body.读取正文(author, "ch-1") == doc
    with app.要求数据库().连接(只读=True) as conn:
        assert conn.execute("SELECT count(*) FROM muse_diagnosis").fetchone()[0] == 1