"""应用读取前拒绝、阶段用量恢复和版本化回放的合成行为检查。"""

from contextlib import nullcontext
from dataclasses import replace
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

from muse.上下文.任务范围 import 上下文服务
from muse.上下文.分页 import 解码游标
from muse.上下文.存储 import 上下文存储
from muse.上下文.来源读取 import _片段游标, _片段起点, 读取历史正文
from muse.上下文.模型 import 上下文错误, 依赖状态
from muse.任务运行.上下文预算 import (
    上下文预算错误,
    核对请求预算,
    读取用量,
    默认读取预算,
)
from muse.任务运行.工具调用 import 只读工具集, 工具定义, 工具来源, 工具结果, 工具范围
from muse.任务运行.执行合同 import 工具请求
from muse.共享.调用身份 import 用途
from muse.正文写作.存储 import 正文存储


def 策略(**变化):
    return {"context_policy_version": 2, "context_read_budget": {**默认读取预算, **变化}}


def 来源范围(**变化):
    return {
        **策略(),
        "authorized_by": "author",
        "work_id": "work",
        "directory_revision": 1,
        "run_purpose": "production",
        "content_purpose": "generation",
        "source_ids": ["doc"],
        "documents": [
            {
                "chapter_id": "ch",
                "branch_id": "main",
                "revision": 1,
                "position": 1,
                "document_id": "doc",
                "document_hash": "hash",
            }
        ],
        **变化,
    }


def 工具域(用量=None):
    return 工具范围("task", "work", ("doc",), 1, "production", "generation", 用量)


@pytest.mark.case_id("NC-O07-001")
def test_超长来源在取得正文前拒绝():
    连 = MagicMock()
    连.transaction.return_value = nullcontext()
    数据库 = SimpleNamespace(连接=lambda **_: nullcontext(连))
    with (
        patch(
            "muse.上下文.来源读取._核对目录",
            return_value={"chapters": [{"chapter_id": "ch", "position": 1}]},
        ),
        patch(
            "muse.上下文.来源读取.读取正文元数据",
            return_value={"current_revision": 1, "document_hash": "hash", "payload_bytes": 262145},
        ),
        patch("muse.上下文.来源读取.读取正文依据") as 正文,
    ):
        with pytest.raises(上下文预算错误) as 错:
            读取历史正文(数据库, 来源范围(), 工具域(), {"chapter_id": "ch", "revision": 1})
        assert 错.value.上下文["requested"] == 262145
        正文.assert_not_called()


@pytest.mark.case_id("NC-O07-002")
def test_重启从持久消费量继续而非清零():
    事件 = []
    一 = 读取用量(策略(source_reads=1), 记录=事件.append)
    一.预留来源(25)
    二 = 读取用量(策略(source_reads=1), 事件[-1])
    with pytest.raises(上下文预算错误) as 错:
        二.预留来源(25)
    assert 错.value.上下文["consumed"]["source_bytes"] == 25
    assert 错.value.上下文["dimension"] == "source_reads"


@pytest.mark.case_id("NC-O07-003")
def test_预算预留未落盘时禁止来源读取():
    def 拒写(_):
        raise RuntimeError("event unavailable")

    量 = 读取用量(策略(), 记录=拒写)
    with pytest.raises(RuntimeError):
        量.预留来源(32)
    assert 量.已用["source_reads"] == 0


@pytest.mark.case_id("NC-O07-004")
def test_单工具和累计历史在返回前分别拒绝():
    单 = 读取用量(策略(single_tool_bytes=10))
    with pytest.raises(上下文预算错误):
        单.接收工具(11)
    累 = 读取用量(策略(tool_bytes=30))
    for _ in range(3):
        累.接收工具(10)
    with pytest.raises(上下文预算错误) as 错:
        累.接收工具(1)
    assert 错.value.上下文["consumed"]["tool_bytes"] == 30


@pytest.mark.case_id("NC-O07-005")
def test_单次与累计耗时独立限制():
    量 = 读取用量(策略(single_read_milliseconds=20, read_milliseconds=25))
    with patch("muse.任务运行.上下文预算.monotonic", return_value=0.015):
        量.完成读取(0)
    with patch("muse.任务运行.上下文预算.monotonic", return_value=0.015):
        with pytest.raises(上下文预算错误) as 错:
            量.完成读取(0)
    assert 错.value.上下文["dimension"] == "read_milliseconds"
    with patch("muse.任务运行.上下文预算.monotonic", return_value=0.03):
        with pytest.raises(上下文预算错误) as 错:
            读取用量(策略(single_read_milliseconds=20)).完成读取(0)
    assert 错.value.上下文["dimension"] == "single_read_milliseconds"


@pytest.mark.case_id("NC-O07-006")
def test_旧任务超部署上界要求新授权且不改原冻结输入():
    旧 = {"legacy": True}
    with pytest.raises(上下文预算错误) as 错:
        核对请求预算(旧, b"x" * 262145)
    assert 错.value.错误码 == "CONTEXT_POLICY_REAUTH_REQUIRED"
    assert 旧 == {"legacy": True}
    with pytest.raises(上下文预算错误) as 错:
        核对请求预算(策略(request_bytes=32), b"x" * 33)
    assert 错.value.错误码 == "CONTEXT_BUDGET_EXCEEDED"


@pytest.mark.case_id("NC-O07-007")
def test_正文游标不可跨任务版本或投影():
    冻结 = 来源范围()
    键 = ("ch", "main", 1)
    域 = 工具域()
    游标 = _片段游标(域, 冻结, 键, 2048)
    assert _片段起点(域, 冻结, 键, {"cursor": 游标}) == 2048
    for 域2, 冻结2, 键2 in [
        (replace(域, 任务ID="other"), 冻结, 键),
        (域, {**冻结, "directory_revision": 2}, 键),
        (域, 冻结, ("ch", "main", 2)),
        (replace(域, 内容用途="planning"), 冻结, 键),
    ]:
        with pytest.raises(上下文错误):
            _片段起点(域2, 冻结2, 键2, {"cursor": 游标})


@pytest.mark.case_id("NC-O07-008")
def test_窄依据只查版本哈希与数据库侧大小():
    连 = MagicMock()
    正文存储(连).读取元数据("doc", 1)
    sql, params = 连.cursor.return_value.execute.call_args.args
    assert "v.document," not in sql and "v.document " not in sql
    assert "octet_length(v.document::text) AS payload_bytes" in sql
    assert params == (1, "doc")


@pytest.mark.case_id("NC-O07-009")
def test_快照摘要页不载入装配正文且请求有上界():
    存储 = 上下文存储(object(), 用途.生产)
    with patch.object(存储, "查询") as 查询:
        存储.按任务读取摘要页("task", 20, None)
    sql, params = 查询.call_args.args
    assert "SELECT *" not in sql and "assembly->'文本'" not in sql
    assert params[-1] == 21


@pytest.mark.case_id("NC-O07-010")
def test_快照页同范围只做一遍依赖核验并保留下一页位置():
    行 = {
        "snapshot_id": "00000000-0000-0000-0000-000000000001",
        "task_id": "task",
        "work_id": "work",
        "input_hash": "hash",
        "frozen_scope": 来源范围(),
        "source_bindings": [],
        "assembly": {},
        "created_at": datetime(2026, 1, 1, tzinfo=UTC),
    }
    连 = MagicMock()
    连.transaction.return_value = nullcontext()
    服务 = 上下文服务(SimpleNamespace(用途=用途.生产, 连接=lambda **_: nullcontext(连)), None)
    with (
        patch("muse.上下文.存储.上下文存储.按任务读取摘要页", return_value=[行] * 21),
        patch("muse.上下文.依赖校验.核对依赖", return_value=依赖状态(True, (), ())) as 核对,
    ):
        页 = 服务.读取任务快照页("task")
    assert len(页["items"]) == 20 and 页["has_more"]
    核对.assert_called_once()
    assert (
        解码游标(页["next_cursor"], {"task_id": "task", "purpose": "production"})[1]
        == 行["snapshot_id"]
    )
    with pytest.raises(上下文错误):
        解码游标(页["next_cursor"], {"task_id": "other", "purpose": "production"})


@pytest.mark.case_id("NC-O07-011")
def test_工具结果超限不进入历史也不产生成功读取记录():
    记录 = []
    量 = 读取用量(策略(single_tool_bytes=20))
    定义 = 工具定义(
        "read",
        "读",
        {"type": "object"},
        lambda *_: 工具结果("x" * 50, (工具来源("doc", "1", "hash", "v1"),)),
    )
    集 = 只读工具集((定义,), ("read",), 工具域(量), 记录.append)
    with pytest.raises(上下文预算错误):
        集.调用(工具请求("call", "read", {}))
    assert len(记录) == 1 and 记录[0]["outcome"] == "blocked"
    assert 记录[0]["reason_code"] == "CONTEXT_BUDGET_EXCEEDED"


@pytest.mark.case_id("NC-O07-012")
def test_多文稿依赖复检只读元数据并保留版本失效():
    from muse.上下文.依赖校验 import 核对依赖

    文稿 = [
        {
            "chapter_id": f"ch{i}",
            "document_id": f"d{i}",
            "revision": 1,
            "position": i + 1,
            "document_hash": "hash",
            "branch_id": "main",
        }
        for i in range(16)
    ]
    冻结 = 来源范围(documents=文稿)
    目录 = {
        "revision": 1,
        "chapters": [{"chapter_id": d["chapter_id"], "position": d["position"]} for d in 文稿],
    }

    def 元数据(_连, _作者, 章, _版, **_):
        return {
            "document_id": "d" + 章[2:],
            "document_hash": "hash",
            "current_revision": 2 if 章 == "ch15" else 1,
        }

    with (
        patch("muse.上下文.依赖校验.读取作品范围", return_value=目录),
        patch("muse.上下文.依赖校验.读取正文元数据", side_effect=元数据) as 元,
        patch("muse.正文写作.存储.正文存储.恢复草稿", side_effect=AssertionError("no payload")),
    ):
        状态 = 核对依赖(object(), 冻结)
    assert 元.call_count == 16 and not 状态.有效
    assert len(状态.锁定依据) == 16  # 目录和十五个未漂移正文
    assert "ch15" in 状态.变化项[0]


@pytest.mark.case_id("NC-O07-013")
def test_v2默认正文片段明确续页而v1回放内容不变():
    连 = MagicMock()
    连.transaction.return_value = nullcontext()
    数据库 = SimpleNamespace(连接=lambda **_: nullcontext(连))
    依据 = {"document": object(), "document_id": "doc", "document_hash": "hash"}
    with (
        patch("muse.上下文.来源读取._核对并读取", return_value=依据),
        patch("muse.上下文.来源读取.可见文本", return_value="a" * 3000),
        patch("muse.上下文.来源读取.可见文本哈希", return_value="hash"),
    ):
        新 = 读取历史正文(数据库, 来源范围(), 工具域(), {"chapter_id": "ch", "revision": 1})
        旧 = 来源范围()
        del 旧["context_policy_version"], 旧["context_read_budget"]
        老 = 读取历史正文(数据库, 旧, 工具域(), {"chapter_id": "ch", "revision": 1})
    assert len(新.内容["text"]) == 2048 and 新.内容["has_more"]
    assert 新.内容["next_cursor"] and 新.内容["omitted_reason"] == "bounded_fragment"
    assert len(老.内容["text"]) == 3000 and "has_more" not in 老.内容


@pytest.mark.case_id("NC-O07-014")
def test_事实全量超限在时点投影前拒绝():
    from muse.上下文.来源读取 import 读取时点事实

    连 = MagicMock()
    连.transaction.return_value = nullcontext()
    数据库 = SimpleNamespace(连接=lambda **_: nullcontext(连))
    with (
        patch("muse.上下文.来源读取._核对目录"),
        patch("muse.上下文.来源读取.读取当前事实系统版本", return_value=1),
        patch(
            "muse.上下文.来源读取.读取事实预算元数据",
            return_value={"source_headers": 5, "payload_bytes": 262145},
        ),
        patch("muse.上下文.来源读取.读取时点事实投影") as 投影,
    ):
        with pytest.raises(上下文预算错误):
            读取时点事实(数据库, 来源范围(facts={"system_revision": 1}), 工具域(), {})
    投影.assert_not_called()


@pytest.mark.case_id("NC-O07-015")
def test_细纲超限在正文投影前拒绝():
    from muse.上下文.来源读取 import 读取目标细纲

    连 = MagicMock()
    连.transaction.return_value = nullcontext()
    数据库 = SimpleNamespace(连接=lambda **_: nullcontext(连))
    with (
        patch("muse.上下文.来源读取._核对目录"),
        patch("muse.上下文.来源读取.读取细纲元数据", return_value={"payload_bytes": 262145}),
        patch("muse.上下文.来源读取.读取细纲投影") as 投影,
    ):
        with pytest.raises(上下文预算错误):
            读取目标细纲(数据库, 来源范围(target_plan={"chapter_id": "ch"}), 工具域(), {})
    投影.assert_not_called()


@pytest.mark.case_id("NC-O07-016")
def test_事实点名集合越出冻结范围建联前拒绝():
    from muse.上下文.来源读取 import 读取时点事实

    数据库 = MagicMock()
    with pytest.raises(上下文错误) as 错:
        读取时点事实(
            数据库, 来源范围(facts={"system_revision": 1}), 工具域(), {"object_ids": ["secret"]}
        )
    assert 错.value.错误码 == "SCOPE_DENIED"
    数据库.连接.assert_not_called()


@pytest.mark.case_id("NC-O07-017")
def test_篡改v2预算不能放大部署上界():
    for 冻结 in [
        {"context_policy_version": 2},
        策略(request_bytes=262145),
        策略(source_reads=True),
    ]:
        with pytest.raises(上下文预算错误):
            读取用量(冻结)


@pytest.mark.case_id("NC-O07-018")
def test_单阶段失败预检耗时也保存消费量():
    from muse.上下文.来源读取 import _读取阶段

    事件 = []
    量 = 读取用量(策略(), 记录=事件.append)
    数据库 = MagicMock()
    数据库.连接.side_effect = RuntimeError("connection failed")
    with (
        pytest.raises(RuntimeError),
        patch("muse.上下文.来源读取.monotonic", return_value=1),
        patch("muse.任务运行.上下文预算.monotonic", return_value=1.01),
    ):
        with _读取阶段(数据库, 来源范围(), 工具域(量)):
            raise AssertionError("unreachable")
    assert 事件[-1]["read_milliseconds"] >= 9


@pytest.mark.case_id("NC-O07-019")
def test_快照部分失效不将声音偏好方法误报为事实状态():
    from muse.上下文.模型 import 上下文快照, 依赖锁定, 来源绑定
    from muse.接入.http.路由.上下文 import 呈现快照

    冻结 = 来源范围(
        source_ids=["doc", "voice", "pref", "method", "fact"],
        voice={"version_id": "voice"},
        preferences=[{"preference_id": "pref"}],
        methods=[{"binding_id": "method"}],
        facts={"system_revision": 1},
    )
    快照 = 上下文快照(
        "snap",
        "task",
        "work",
        "hash",
        冻结,
        {},
        tuple(来源绑定(k, "1", "hash", "v1") for k in ["voice", "pref", "method", "fact"]),
        "SOURCE_STALE",
        "voice changed",
    )
    状态 = 依赖状态(
        False,
        (依赖锁定("work:事实系统", "facts", "1", ""), 依赖锁定("method", "method", "1", "")),
        ("voice changed",),
    )
    视图 = 呈现快照(快照, 状态)
    assert {r.source_id: r.current_state for r in 视图.sources} == {
        "voice": "SOURCE_STALE",
        "pref": "SOURCE_STALE",
        "method": "有效",
        "fact": "有效",
    }
    状态 = 依赖状态(False, (依赖锁定("voice", "voice", "1", ""),), ("fact changed",))
    assert 呈现快照(快照, 状态).sources[0].current_state == "有效"


@pytest.mark.case_id("NC-O07-020")
def test_HTTP快照页先核任务作者再取摘要():
    from muse.接入.http.路由.上下文 import 列出任务快照页

    请求 = MagicMock()
    with patch(
        "muse.接入.http.路由.上下文.读取作者任务",
        side_effect=上下文错误("SCOPE_DENIED", "wrong author"),
    ):
        with pytest.raises(上下文错误):
            列出任务快照页("task", 请求, "author", limit=20, cursor=None)
    请求.app.state.装配.要求上下文.assert_not_called()


@pytest.mark.case_id("NC-O07-021")
def test_实验臂纯组装不改变原文码点且必读超限失败():
    from muse.上下文.回放材料 import _组装实验臂, _选取足额补充

    原 = [{"text": "甲🙂乙", "document_id": "doc"}]
    片段 = _选取足额补充(原, 2)
    assert 片段[0]["text"] == "甲🙂" and 原[0]["text"] == "甲🙂乙"
    with pytest.raises(上下文错误):
        _选取足额补充(原, 4)
    with pytest.raises(上下文错误):
        _组装实验臂({"content": {"必读": "保留"}}, 原, [], [], [], 1)


@pytest.mark.case_id("NC-O07-022")
def test_并发工具实例不能覆盖已消费累计量():
    from muse.任务运行.接口 import 任务服务
    from muse.任务运行.模型 import 状态冲突

    最新 = {}
    存储 = MagicMock()
    存储.读取上下文用量.side_effect = lambda *_: dict(最新)
    存储.追加事件.side_effect = lambda _任务, _类型, payload, **_: 最新.update(
        payload["context_usage"]
    )
    服务 = 任务服务.__new__(任务服务)
    服务._存储 = lambda: nullcontext(存储)
    领取 = SimpleNamespace(任务ID="task", 步骤ID="step", 尝试ID="try")
    一 = 读取用量(策略())
    二 = 读取用量(策略())
    一.记录 = lambda 新: 服务.登记上下文用量(领取, 一.已用, 新, "a")
    二.记录 = lambda 新: 服务.登记上下文用量(领取, 二.已用, 新, "b")
    with patch("muse.任务运行.接口.校验领取") as 锁:
        一.预留来源(10)
        with pytest.raises(状态冲突):
            二.预留来源(11)
    assert 最新["source_bytes"] == 10 and 存储.追加事件.call_count == 1
    assert 锁.call_count == 2 and 二.已用["source_reads"] == 0


@pytest.mark.case_id("NC-O07-023")
def test_旧任务工具描述不被v2参数扩展改写():
    from muse.上下文.来源读取 import 事实工具, 历史正文工具

    新 = 来源范围(facts={"system_revision": 1})
    旧 = {k: v for k, v in 新.items() if k not in {"context_policy_version", "context_read_budget"}}
    assert "cursor" in 历史正文工具(None, 新).参数合同["properties"]
    assert "cursor" not in 历史正文工具(None, 旧).参数合同["properties"]
    assert "object_ids" in 事实工具(None, 新).参数合同["properties"]
    assert "object_ids" not in 事实工具(None, 旧).参数合同["properties"]
    assert "context_policy_version" not in 旧


@pytest.mark.case_id("NC-O07-024")
def test_循环读取在下一载荷前检查剩余耗时():
    量 = 读取用量(策略(single_read_milliseconds=10))
    量.当前开始 = 0
    with patch("muse.任务运行.上下文预算.monotonic", return_value=0.011):
        with pytest.raises(上下文预算错误) as 错:
            量.预留来源(8)
    assert 错.value.上下文["dimension"] == "single_read_milliseconds"
    assert 量.已用["source_reads"] == 0