"""已授权的合成原文；独立PG验证先租约、完整归档与中断恢复。"""

import hashlib
from datetime import UTC, datetime, timedelta

import pytest

from muse.任务运行.接口 import (
    任务服务,
    任务请求,
    原文服务,
    原文错误,
    步骤处理器,
    步骤结果,
    步骤计划,
    证据服务,
)
from muse.共享.时间 import 假时钟
from muse.共享.调用身份 import 内容用途, 用途
from muse.基础设施.受控文件 import 受控文件, 受控文件错误
from muse.编排.接口 import 流程定义, 流程服务, 流程登记

pytestmark = pytest.mark.数据库


def test_raw_orphan_scope__a82112(原文环境, tmp_path):
    """保留LC-d6e3f58513a1；租约替身升级为真实PG，原占位孤儿改用规范UUID。"""
    import json
    import subprocess
    import sys
    import uuid

    服务, 任务, 授权, 数据, 哈希, 时钟, 库 = 原文环境
    租约 = 服务.创建租约(任务, 授权, "kept", 时钟.现在() + timedelta(hours=1), 最少剩余秒=30)
    服务.写入(任务, 租约, 哈希, 数据)
    孤儿 = str(uuid.uuid4())
    原孤儿字节 = b"secret"  # 旧例的合成字节，仅模拟缺租约残留。
    孤儿哈希 = hashlib.sha256(原孤儿字节).hexdigest()
    服务.文件.写入(孤儿, 孤儿哈希, 原孤儿字节)
    外根 = 受控文件(tmp_path / "other-namespace")
    外根.绑定命名空间(str(uuid.uuid4()), "production")
    外根.写入(孤儿, 孤儿哈希, 原孤儿字节)
    未知 = 服务.文件.根 / "合成资料.md"
    未知.write_text("不属于原文对象的合成资料")
    应用 = tmp_path / "cleanup.toml"
    应用.write_text(
        '["数据库"]\n"取值方式"="受控存储"\n"位置"='
        + json.dumps(库[用途.生产].引用.位置, ensure_ascii=False)
        + '\n["资源"]\n"发布身份"="test"\n["文件"]\n"原文暂存"='
        + json.dumps(str(服务.文件.根), ensure_ascii=False)
        + "\n"
    )
    结果 = subprocess.run(
        [sys.executable, "-I", "-m", "muse", "管理", str(应用), "清理无租约暂存"],
        cwd=tmp_path,
        capture_output=True,
        text=True,
    )
    assert 结果.returncode == 0, 结果.stderr
    报告 = json.loads(结果.stdout)
    assert 报告["cleaned_orphan_ids"] == [孤儿] and 报告["unknown_entries"] == 1
    assert not (服务.文件.根 / 孤儿).exists()
    assert 服务.读取(任务, 租约, 哈希) == 数据
    assert (
        外根.读取(孤儿, 孤儿哈希) == 原孤儿字节 and 未知.read_text() == "不属于原文对象的合成资料"
    )
    回执 = 服务.读取孤儿清理回执(孤儿)
    assert 回执["receipt_id"] in 报告["receipts"] and 回执["state"] == "completed"
    assert 服务.清理无租约原文()["cleaned_orphan_ids"] == []
    assert 服务.读取孤儿清理回执(孤儿) == 回执
    with pytest.raises(受控文件错误, match="其他数据库"):
        原文服务(库[用途.生产], 外根).清理无租约原文()
    assert 外根.读取(孤儿, 孤儿哈希) == 原孤儿字节


def test_孤儿清理先保存意图且确认失败可幂等恢复__a82111(原文环境):
    import uuid

    服务, 任务, 授权, 数据, 哈希, 时钟, 库 = 原文环境
    租约 = 服务.创建租约(
        任务, 授权, "kept-for-namespace", 时钟.现在() + timedelta(hours=1), 最少剩余秒=0
    )
    服务.写入(任务, 租约, 哈希, 数据)
    孤儿 = str(uuid.uuid4())
    服务.文件.写入(孤儿, 哈希, 数据)
    with 库[用途.维护].连接() as 连:
        连.execute("""CREATE FUNCTION reject_cleanup() RETURNS trigger LANGUAGE plpgsql AS $$
        BEGIN RAISE EXCEPTION 'synthetic cleanup confirmation failure'; END $$;
        CREATE TRIGGER reject_cleanup BEFORE UPDATE ON muse_raw_orphan_cleanup
        FOR EACH ROW EXECUTE FUNCTION reject_cleanup();""")
    首次 = 服务.清理无租约原文()
    assert 首次["needs_recovery"] == [孤儿] and not (服务.文件.根 / 孤儿).exists()
    待续 = 服务.读取孤儿清理回执(孤儿)
    assert 待续["state"] == "pending"
    with 库[用途.维护].连接() as 连:
        连.execute("DROP TRIGGER reject_cleanup ON muse_raw_orphan_cleanup")
    再次 = 服务.清理无租约原文()
    assert 再次["cleaned_orphan_ids"] == [孤儿]
    完成 = 服务.读取孤儿清理回执(孤儿)
    assert 完成["receipt_id"] == 待续["receipt_id"] and 完成["state"] == "completed"
    assert 完成["removed"] is False  # 此次确认已不存在，不补造首次删除成功记录。
    assert 服务.读取(任务, 租约, 哈希) == 数据


def 建证据(环境):
    原文, 任务, _, 数据, 哈希, _, 库 = 环境
    登记 = 流程登记()
    登记.登记处理器(步骤处理器("raw-test", "1", lambda _: 步骤结果({}), "1", "1"))
    运行 = 任务服务(库[用途.生产], 登记)
    领取 = 运行.领取步骤("evidence-worker", ["raw-test"])
    assert 领取 is not None and 领取.任务ID == 任务
    return 证据服务(库[用途.生产]), 领取


def test_失败任务仍可恢复所以保留有效原文租约__a82113(原文环境):
    from muse.任务运行.接口 import 任务状态

    服务, 任务, 授权, 数据, 哈希, 时钟, 库 = 原文环境
    租约 = 服务.创建租约(
        任务, 授权, "recoverable-failure", 时钟.现在() + timedelta(hours=1), 最少剩余秒=30
    )
    服务.写入(任务, 租约, 哈希, 数据)
    登记 = 流程登记()
    登记.登记处理器(步骤处理器("raw-test", "1", lambda _: 步骤结果({}), "1", "1"))
    运行 = 任务服务(库[用途.生产], 登记)
    领取 = 运行.领取步骤("failed-worker", ["raw-test"])
    assert 领取 is not None
    运行.失败步骤(领取, "synthetic_failure")
    assert 运行.读取任务(任务).状态 == 任务状态.已失败
    assert 服务.恢复任务原文(任务, "author") == [{"lease_id": 租约, "action": "retained"}]
    assert 服务.读取(任务, 租约, 哈希) == 数据


def test_仅哈希补交完整证据保持失败与同一身份__a82007(原文环境) -> None:
    原文, 任务, 临时授权, 数据, 哈希, 时钟, 库 = 原文环境
    证据, 领取 = 建证据(原文环境)
    元信息 = {"call_id": "call-failed", "error_code": "MODEL_INCOMPLETE", "output_tokens": None}
    回执 = 证据.保存(任务, 领取.尝试ID, "model_response", 哈希, "failed", 元信息)
    assert 回执["retention"] == "hash_only" and 回执["outcome"] == "failed"
    with pytest.raises(原文错误):
        证据.补交原文(任务, 回执["evidence_id"], 数据, 临时授权)
    授权 = 原文.批准保留(
        任务,
        "author",
        "persistent",
        来源版本="source-v1",
        哈希=(哈希,),
        内容用途="detection",
        方式="persistent",
        有效期=时钟.现在() + timedelta(days=1),
    )
    with pytest.raises(原文错误):
        证据.补交原文(任务, 回执["evidence_id"], 数据 + b"changed", 授权)
    补交 = 证据.补交原文(任务, 回执["evidence_id"], 数据, 授权)
    assert 补交 == {**回执, "retention": "full", "revision": 2}
    assert 证据.补交原文(任务, 回执["evidence_id"], 数据, 授权) == 补交
    with 库[用途.生产].连接() as 连:
        assert (
            连.execute("SELECT count(*) FROM muse_attempt WHERE task_id=%s", (任务,)).fetchone()[0]
            == 1
        )
        assert (
            连.execute(
                "SELECT content FROM muse_runtime_evidence WHERE evidence_id=%s",
                (回执["evidence_id"],),
            ).fetchone()[0]
            == 数据
        )


def test_证据不能覆盖结果或写入未登记正文元信息__a82008(原文环境) -> None:
    _, 任务, _, _, 哈希, _, _ = 原文环境
    证据, 领取 = 建证据(原文环境)
    元信息 = {"call_id": "call-partial"}
    回执 = 证据.保存(任务, 领取.尝试ID, "model_response", 哈希, "partial", 元信息)
    with pytest.raises(原文错误):
        证据.保存(任务, 领取.尝试ID, "model_response", 哈希, "completed", 元信息)
    with pytest.raises(原文错误):
        证据.保存(任务, 领取.尝试ID, "failure", 哈希, "failed", {"source_refs": [{"raw": "正文"}]})
    assert 证据.读取回执(任务, 回执["evidence_id"]) == 回执


@pytest.fixture
def 原文环境(应用测试库, tmp_path):
    登记 = 流程登记()
    登记.登记处理器(步骤处理器("raw-test", "1", lambda _: 步骤结果({}), "1", "1"))
    登记.登记类型("raw-test", 必需保护=())
    任务 = 任务服务(应用测试库[用途.生产], 登记)
    流程 = 流程服务(任务, 登记)
    流程.发布(流程定义("raw-test", "1", (步骤计划("raw", "raw-test", "1"),)))
    id_ = 流程.发起(
        任务请求(
            "raw-test",
            "cmd",
            "author",
            用途.生产,
            内容用途.检测,
            {},
            "role-1",
            "res-1",
            {
                "source_scope": {},
                "schema_versions": {},
                "authorization": "grant",
                "budget": {},
                "stop_conditions": [],
            },
        ),
        "raw-test",
        "1",
    )
    时钟 = 假时钟(datetime.now(UTC))
    文件 = 受控文件(tmp_path / "私人原文")
    服务 = 原文服务(应用测试库[用途.生产], 文件, 时钟)
    数据 = "合成原文，保留每一个字节。".encode()
    哈希 = hashlib.sha256(数据).hexdigest()
    授权 = 服务.批准保留(
        id_,
        "author",
        "approve",
        来源版本="source-v1",
        哈希=(哈希,),
        内容用途="detection",
        方式="temporary",
        有效期=时钟.现在() + timedelta(hours=2),
    )
    return 服务, id_, 授权, 数据, 哈希, 时钟, 应用测试库


def test_租约先提交才写字节且过期仍可清理__a82001(原文环境) -> None:
    服务, 任务, 授权, 数据, 哈希, 时钟, 库 = 原文环境
    with pytest.raises(原文错误):
        服务.创建租约(任务, 授权, "too-long", 时钟.现在() + timedelta(hours=25), 最少剩余秒=60)
    assert list(服务.文件.根.iterdir()) == []
    租约 = 服务.创建租约(任务, 授权, "lease", 时钟.现在() + timedelta(hours=1), 最少剩余秒=60)
    with 库[用途.生产].连接() as 连:
        assert (
            连.execute("SELECT state FROM muse_raw_lease WHERE lease_id=%s", (租约,)).fetchone()[0]
            == "open"
        )
    assert list(服务.文件.根.iterdir()) == []
    服务.写入(任务, 租约, 哈希, 数据)
    assert 服务.读取(任务, 租约, 哈希) == 数据
    时钟.推进(timedelta(hours=1))
    with pytest.raises(原文错误):
        服务.读取(任务, 租约, 哈希)
    with pytest.raises(原文错误):
        服务.写入(任务, 租约, 哈希, 数据)
    服务.清理(任务, 租约)
    assert 服务.状态(任务, 租约)["state"] == "closed"
    服务.清理(任务, 租约)


def test_完整PG归档可重复对账并清理临时副本__a82002(原文环境) -> None:
    服务, 任务, 授权, 数据, 哈希, 时钟, 库 = 原文环境
    租约 = 服务.创建租约(任务, 授权, "lease", 时钟.现在() + timedelta(hours=1), 最少剩余秒=60)
    服务.写入(任务, 租约, 哈希, 数据)
    归档授权 = 服务.批准保留(
        任务,
        "author",
        "archive",
        来源版本="source-v1",
        哈希=(哈希,),
        内容用途="detection",
        方式="archive",
        有效期=时钟.现在() + timedelta(days=1),
    )
    回执 = 服务.归档(任务, 租约, 归档授权)
    assert 回执["entry_count"] == 1 and 回执["total_bytes"] == len(数据)
    assert 服务.归档(任务, 租约, 归档授权) == 回执
    assert 服务.状态(任务, 租约) == {
        "lease_id": 租约,
        "state": "migrated",
        "archive_id": 回执["archive_id"],
        "cleanup_complete": True,
    }
    assert not (服务.文件.根 / 租约).exists()
    assert 服务.读取归档(任务, 回执["archive_id"], 哈希) == 数据
    assert 原文服务(库[用途.生产], 时钟实现=时钟).读取归档(任务, 回执["archive_id"], 哈希) == 数据
    时钟.推进(timedelta(hours=2))
    服务.清理(任务, 租约)
    assert 服务.读取归档(任务, 回执["archive_id"], 哈希) == 数据
    with 库[用途.生产].连接() as 连:
        assert (
            连.execute(
                "SELECT content FROM muse_raw_archive_item WHERE archive_id=%s",
                (回执["archive_id"],),
            ).fetchone()[0]
            == 数据
        )


def test_缺字节归档保留migrating不被过期删除__a82003(原文环境) -> None:
    服务, 任务, 授权, 数据, 哈希, 时钟, _ = 原文环境
    租约 = 服务.创建租约(任务, 授权, "lease", 时钟.现在() + timedelta(hours=1), 最少剩余秒=60)
    归档授权 = 服务.批准保留(
        任务,
        "author",
        "archive",
        来源版本="source-v1",
        哈希=(哈希,),
        内容用途="detection",
        方式="archive",
        有效期=时钟.现在() + timedelta(days=1),
    )
    with pytest.raises(受控文件错误):
        服务.归档(任务, 租约, 归档授权)
    assert 服务.状态(任务, 租约)["state"] == "migrating"
    assert 服务.恢复任务原文(任务, "author")[0]["action"] == "needs_recovery"
    时钟.推进(timedelta(hours=2))
    with pytest.raises(原文错误):
        服务.清理(任务, 租约)
    assert 服务.状态(任务, 租约)["state"] == "migrating"


def test_归档事务失败不留半归档且同ID恢复__a82004(原文环境) -> None:
    服务, 任务, 授权, 数据, 哈希, 时钟, 库 = 原文环境
    租约 = 服务.创建租约(任务, 授权, "lease", 时钟.现在() + timedelta(hours=1), 最少剩余秒=60)
    服务.写入(任务, 租约, 哈希, 数据)
    归档授权 = 服务.批准保留(
        任务,
        "author",
        "archive",
        来源版本="source-v1",
        哈希=(哈希,),
        内容用途="detection",
        方式="archive",
        有效期=时钟.现在() + timedelta(days=1),
    )
    with 库[用途.维护].连接() as 连:
        连.execute("""CREATE FUNCTION reject_raw() RETURNS trigger LANGUAGE plpgsql AS $$
        BEGIN RAISE EXCEPTION 'synthetic archive failure'; END $$;
        CREATE TRIGGER reject_raw BEFORE INSERT ON muse_raw_archive_item
        FOR EACH ROW EXECUTE FUNCTION reject_raw();""")
    with pytest.raises(原文错误):
        服务.归档(任务, 租约, 归档授权)
    原状态 = 服务.状态(任务, 租约)
    assert 原状态["state"] == "migrating"
    assert 服务.文件.读取(租约, 哈希) == 数据
    with 库[用途.维护].连接() as 连:
        assert 连.execute("SELECT count(*) FROM muse_raw_archive").fetchone()[0] == 0
        assert 连.execute("SELECT count(*) FROM muse_raw_archive_item").fetchone()[0] == 0
        连.execute("DROP TRIGGER reject_raw ON muse_raw_archive_item")
    assert 服务.归档(任务, 租约, 归档授权)["archive_id"] == 原状态["archive_id"]


def test_清理失败不关闭且错误作者不能批准__a82005(原文环境) -> None:
    服务, 任务, 授权, 数据, 哈希, 时钟, _ = 原文环境
    with pytest.raises(原文错误):
        服务.批准保留(
            任务,
            "创始人",
            "forged",
            来源版本="source-v1",
            哈希=(哈希,),
            内容用途="detection",
            方式="archive",
            有效期=时钟.现在() + timedelta(days=1),
        )
    租约 = 服务.创建租约(任务, 授权, "lease", 时钟.现在() + timedelta(hours=1), 最少剩余秒=60)
    服务.写入(任务, 租约, 哈希, 数据)
    目录 = 服务.文件.根 / 租约
    目录.chmod(0o755)
    try:
        with pytest.raises(受控文件错误):
            服务.清理(任务, 租约, 作者="author")
        assert 服务.状态(任务, 租约)["state"] == "open"
        assert not 服务.状态(任务, 租约)["cleanup_complete"]
    finally:
        目录.chmod(0o700)
    服务.清理(任务, 租约, 作者="author")
    assert 服务.状态(任务, 租约)["state"] == "closed"


def test_HTTP作者授权到原文写入归档使用真实会话__a82006(原文环境, tmp_path) -> None:
    from fastapi.testclient import TestClient

    from muse.接入.http.应用 import 创建应用
    from muse.配置 import 应用配置, 服务配置

    _, 任务, _, 数据, 哈希, _, 库 = 原文环境
    口令 = tmp_path / "口令.txt"
    口令.write_text("synthetic-raw-author")
    配置 = 应用配置(
        库[用途.生产].引用,
        "test",
        HTTP=服务配置(
            str(口令),
            作者ID="author",
            公开地址="http://testserver",
            允许来源=("http://testserver",),
        ),
        原文暂存=str(tmp_path / "HTTP原文"),
    )
    base = f"/api/v1/tasks/{任务}"
    body = {
        "command_id": "http-approve",
        "source_version": "source-v1",
        "content_hashes": [哈希],
        "content_purpose": "detection",
        "retention_mode": "temporary",
        "valid_until": (datetime.now(UTC) + timedelta(hours=2)).isoformat(),
    }
    with TestClient(创建应用(配置)) as client:
        client.headers["Origin"] = "http://testserver"
        assert client.post(base + "/raw-authorizations", json=body).status_code == 401
        assert (
            client.post("/api/v1/session", json={"password": "synthetic-raw-author"}).status_code
            == 200
        )
        assert (
            client.post(
                base + "/raw-authorizations", json={**body, "approved_by": "创始人"}
            ).status_code
            == 422
        )
        授权 = client.post(base + "/raw-authorizations", json=body)
        assert 授权.status_code == 200, 授权.text
        租约 = client.post(
            base + "/raw-leases",
            json={
                "authorization_id": 授权.json()["authorization_id"],
                "command_id": "http-lease",
                "retain_until": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
                "min_remaining": 60,
            },
        )
        assert 租约.status_code == 200, 租约.text
        lease = 租约.json()["lease_id"]
        write = client.put(
            base + f"/raw-leases/{lease}/content/{哈希}",
            content=数据,
            headers={"Content-Type": "application/octet-stream"},
        )
        assert write.status_code == 200 and write.json()["bytes"] == len(数据)
        archive = client.post(
            base + "/raw-authorizations",
            json={**body, "command_id": "http-archive", "retention_mode": "archive"},
        )
        result = client.post(
            base + f"/raw-leases/{lease}/archive",
            json={"authorization_id": archive.json()["authorization_id"]},
        )
        assert result.status_code == 200 and result.json()["entry_count"] == 1
        assert not (tmp_path / "HTTP原文" / lease).exists()


def test_历史归档完整导入PG且保留旧载体__a82009(原文环境, tmp_path):
    import json

    from muse.基础设施.受控文件 import 受控文件错误

    服务, 任务, _, _, _, 时钟, 库 = 原文环境
    目录 = tmp_path / "历史归档"
    目录.mkdir()
    字节 = {
        "input.txt": "旧输入。".encode(),
        "output.txt": "旧输出。".encode(),
        "same-output.txt": "旧输出。".encode(),
    }
    清单 = []
    for name, data in sorted(字节.items()):
        (目录 / name).write_bytes(data)
        清单.append(
            {
                "relativePath": name,
                "contentSha256": "sha256:" + hashlib.sha256(data).hexdigest(),
                "sizeBytes": len(data),
            }
        )
    tree = hashlib.sha256(
        json.dumps(清单, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()
    old = {
        "schemaVersion": "raw-vault-migration-receipt-v1",
        "archiveId": "legacy-1",
        "status": "migrated",
        "archiveSha256": "sha256:" + tree,
        "fileCount": 3,
        "totalBytes": sum(map(len, 字节.values())),
    }
    (目录 / ".migration-receipt.json").write_text(json.dumps(old, ensure_ascii=False))
    哈希 = tuple(sorted({hashlib.sha256(v).hexdigest() for v in 字节.values()}))
    授权 = 服务.批准保留(
        任务,
        "author",
        "legacy-approve",
        来源版本="legacy-source-1",
        哈希=哈希,
        内容用途="detection",
        方式="archive",
        有效期=时钟.现在() + timedelta(days=1),
    )
    回执 = 服务.导入历史归档(任务, 授权, "legacy-source-1", 目录)
    assert 回执 == 服务.导入历史归档(任务, 授权, "legacy-source-1", 目录)
    assert 回执["entry_count"] == 2
    with 库[用途.生产].连接() as 连:
        assert (
            连.execute(
                "SELECT legacy_manifest FROM muse_raw_archive WHERE archive_id=%s",
                (回执["archive_id"],),
            ).fetchone()[0]["entries"]
            == 清单
        )
    for name, data in 字节.items():
        assert (目录 / name).read_bytes() == data
        assert 服务.读取归档(任务, 回执["archive_id"], hashlib.sha256(data).hexdigest()) == data
    (目录 / "output.txt").write_bytes(b"changed")
    with pytest.raises(受控文件错误):
        服务.导入历史归档(任务, 授权, "legacy-source-1", 目录)


def test_PG已提交但响应丢失以原归档身份恢复__a82010(原文环境):
    import psycopg

    服务, 任务, 授权, 数据, 哈希, 时钟, 库 = 原文环境
    租约 = 服务.创建租约(任务, 授权, "lease", 时钟.现在() + timedelta(hours=1), 最少剩余秒=60)
    服务.写入(任务, 租约, 哈希, 数据)
    归档授权 = 服务.批准保留(
        任务,
        "author",
        "archive",
        来源版本="source-v1",
        哈希=(哈希,),
        内容用途="detection",
        方式="archive",
        有效期=时钟.现在() + timedelta(days=1),
    )
    原工厂 = 服务.数据库

    class 丢失确认工厂:
        用途 = 原工厂.用途
        次数 = 0

        def 连接(self):
            self.次数 += 1
            次数 = self.次数
            实际 = 原工厂.连接()

            class 连接上下文:
                def __enter__(self):
                    return 实际.__enter__()

                def __exit__(self, *args):
                    result = 实际.__exit__(*args)
                    if 次数 == 2 and args[0] is None:
                        raise psycopg.OperationalError(
                            "synthetic acknowledgement loss after commit"
                        )
                    return result

            return 连接上下文()

    服务.数据库 = 丢失确认工厂()
    try:
        with pytest.raises(原文错误):
            服务.归档(任务, 租约, 归档授权)
    finally:
        服务.数据库 = 原工厂
    assert 服务.文件.读取(租约, 哈希) == 数据
    旧状态 = 服务.状态(任务, 租约)
    assert 旧状态["state"] == "migrated" and not 旧状态["cleanup_complete"]
    恢复 = 服务.归档(任务, 租约, 归档授权)
    assert 恢复["archive_id"] == 旧状态["archive_id"]
    assert 服务.状态(任务, 租约)["cleanup_complete"]
    with 库[用途.生产].连接() as 连:
        assert 连.execute("SELECT count(*) FROM muse_raw_archive").fetchone()[0] == 1


@pytest.mark.parametrize("缺失", [False, True], ids=["retained-bytes", "missing-bytes"])
def test_终止任务恢复清理尚未到期原文__a82011(原文环境, 缺失):
    from muse.任务运行.接口 import 任务状态

    服务, 任务, 授权, 数据, 哈希, 时钟, 库 = 原文环境
    租约 = 服务.创建租约(任务, 授权, "lease", 时钟.现在() + timedelta(hours=1), 最少剩余秒=60)
    服务.写入(任务, 租约, 哈希, 数据)
    if 缺失:
        服务.文件.删除对象(租约)
    运行 = 任务服务(库[用途.生产], 流程登记())
    运行.控制任务(任务, "author", 任务状态.待运行, "取消", 命令ID="cancel")
    assert 服务.恢复任务原文(任务, "author") == [
        {"lease_id": 租约, "action": "missing" if 缺失 else "cleaned"}
    ]
    assert 服务.状态(任务, 租约)["state"] == "closed"
    assert not (服务.文件.根 / 租约).exists()
    with 库[用途.生产].连接() as 连:
        失败码 = 连.execute(
            "SELECT failure_code FROM muse_raw_lease WHERE lease_id=%s", (租约,)
        ).fetchone()[0]
        assert 失败码 == ("RAW_CONTENT_MISSING" if 缺失 else None)