"""合成流程驱动真实 S02、B01、B05 和隔离 PG；不调用模型、不证明完整创作链。""" from dataclasses import asdict, replace from uuid import uuid4 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 应用配置 pytestmark = pytest.mark.数据库 原文 = "　雨夜🌧，e\u0301。\n门还开着。" 前段, 后段 = 原文.split("\n") def 原文草稿(章: str) -> 正文草稿: """段内换行走显式换行节点；可见文本仍与原文逐字一致。""" return 正文草稿((段落("p-" + 章, (文本节点(前段), 换行节点(), 文本节点(后段))),)) @pytest.fixture def 历史环境(内置结构测试库): 应用测试库 = 内置结构测试库 登记 = 流程登记() 装配 = 构建(应用配置(应用测试库[用途.生产].引用, "test"), 流程=登记) with 装配.生命周期(): 作者 = 调用身份("history-author", None, 用途.生产, 内容用途.生成) 作品, 正文 = 装配.要求作品(), 装配.要求正文() for work, chapters in [ ("history-work", ("past", "other", "future")), ("another-work", ("foreign",)), ]: 结构 = 作品.新档案结构(作者, work, "work_core", 1) 作品.保存档案( 作者, "create-" + work, 档案保存(work, 0, {"名称": work, "题材定位": "现实", "主题立意": "选择"}, 结构), ) for i, chapter in enumerate(chapters): 作品.添加章节(作者, "add-" + chapter, work, chapter, chapter, 预期目录版本=i) 正文.保存人工(作者, "save-" + chapter, chapter, 0, 原文草稿(chapter)) def 执行(上下文): 工具 = 装配.要求只读工具(上下文.领取) 结果 = 工具.调用( 工具请求( "history-read", "read_history_document", {"chapter_id": "past", "revision": 1} ) ) 材料 = {结果.来源[0].来源ID: 结果.内容["text"]} 上限 = 上下文.任务.冻结输入["冻结上下文"]["budget"]["context_bytes"] return 步骤结果(asdict(装配材料(材料, tuple(材料), 字节上限=上限))) 登记.登记处理器( 步骤处理器( "synthetic-history", "1", 执行, "history/v1", "history/v1", 保护职责="范围核对", 允许工具=("read_history_document",), ) ) 登记.登记类型("synthetic-history", 必需保护=("范围核对",)) 登记.登记恢复检查( "synthetic-history", "1", lambda 任务: 装配.要求上下文().读取范围(任务.任务ID) ) 运行 = 装配.任务运行 assert 运行 is not None 流程服务(运行, 登记).发布( 流程定义( "synthetic-history", "1", (步骤计划("read", "synthetic-history", "1", 工具=("read_history_document",)),), ) ) def 开始(范围=None): if 范围 is None: 范围 = 装配.要求上下文().授权历史范围( 作者, "history-work", 2, ({"chapter_id": "past", "revision": 1},) ) id_ = 运行.创建任务( 任务请求( "synthetic-history", str(uuid4()), 作者.作者, 用途.生产, 内容用途.生成, {}, "test", "test", { "source_scope": 范围, "schema_versions": {}, "authorization": "synthetic-author-choice", "budget": {"context_bytes": 1024}, "stop_conditions": {"max_reads": 1}, }, 作品ID="history-work", ), "synthetic-history", "1", ) 领取 = 运行.领取步骤("history-test", ["synthetic-history"], 租期秒=120) assert 领取 is not None and 领取.任务ID == id_ return id_, 领取, 装配.要求只读工具(领取) yield 装配, 作者, 开始 @pytest.mark.case_id( "NC-ctx-history-task-read", environment="真实隔离PostgreSQL", given="真实隔离 PG 中已保存的过去章正文与冻结了历史范围的任务", when="合成处理器经 S02 只读工具集读取并按字节预算装配", then=[ "正文逐字进入材料预算且字节数按 UTF-8 核算", "工具读取事件带来源 revision、结构哈希与投影版本，可追溯", ], contract="docs/系统架构/新版设计/模块设计/B09-上下文.md", ) def test_历史正文经真实任务读取留痕并进入材料预算__b09101(历史环境): 装配, _, 开始 = 历史环境 task_id, 领取, _ = 开始() 结果 = 装配.任务运行.执行一步(领取) assert 结果.输出["文本"] == 原文 assert 结果.输出["字节数"] == len(原文.encode("utf-8")) 事件 = 装配.任务运行.续接事件(task_id).事件 读取 = [e for e in 事件 if e.类型 == 事件类型.工具读取] assert len(读取) == 1 引用 = 读取[0].载荷["source_refs"][0] assert 引用["revision"] == "1" assert 引用["schema_hash"] and 引用["projection_version"] assert 读取[0].载荷["tool"] == "read_history_document" @pytest.mark.case_id( "NC-ctx-history-codepoint-slice", environment="真实隔离PostgreSQL", given="含 emoji 与组合字符的已保存正文", when="按码点左闭右开 [3:8) 请求片段", then=[ "返回原文码点片段，start/end 原样回显", "run_purpose 与 content_purpose 来自任务冻结用途", ], contract="docs/系统架构/新版设计/模块设计/B09-上下文.md", ) def test_正文工具按原文码点返回片段与用途__b09102(历史环境): _, _, 开始 = 历史环境 _, _, 工具 = 开始() 结果 = 工具.调用( 工具请求( "slice", "read_history_document", {"chapter_id": "past", "revision": 1, "start": 3, "end": 8}, ) ) assert 结果.内容["text"] == "🌧，e\u0301。" assert (结果.内容["start"], 结果.内容["end"]) == (3, 8) assert 结果.内容["run_purpose"] == "production" assert 结果.内容["content_purpose"] == "generation" @pytest.mark.case_id( "NC-ctx-history-scope-denied", environment="真实隔离PostgreSQL", given="跨作品章、叙事截止之后的章与同作品未授权章", when="经工具请求读取", then=["一律 SCOPE_DENIED；拒绝路径不留下工具读取事件"], contract="docs/系统架构/新版设计/模块设计/B09-上下文.md", ) @pytest.mark.parametrize( "章节", ["foreign", "future", "other"], ids=["cross-work", "future", "unauthorized"] ) def test_正文工具拒绝冻结范围以外的文稿__b09103(历史环境, 章节): 装配, _, 开始 = 历史环境 task_id, _, 工具 = 开始() with pytest.raises(上下文错误) as 错误: 工具.调用(工具请求("denied", "read_history_document", {"chapter_id": 章节, "revision": 1})) assert 错误.value.错误码 == "SCOPE_DENIED" assert not [e for e in 装配.任务运行.续接事件(task_id).事件 if e.类型 == 事件类型.工具读取] @pytest.mark.case_id( "NC-ctx-history-source-stale", environment="真实隔离PostgreSQL", given="任务冻结后正文保存新版本或目录重排", when="继续用原任务工具读取", then=["返回 SOURCE_STALE，不静默改读最新版本或新目录"], contract="docs/系统架构/新版设计/模块设计/B09-上下文.md", ) @pytest.mark.parametrize( "变化", ["document", "directory"], ids=["new-document", "reordered-directory"] ) def test_正文或目录变化拒绝沿用原任务__b09104(历史环境, 变化): 装配, 作者, 开始 = 历史环境 _, _, 工具 = 开始() if 变化 == "document": 装配.要求正文().保存人工( 作者, "revision-2", "past", 1, 正文草稿((段落("p-past", (文本节点("改过的正文"),)),)) ) else: 装配.要求作品().重排章节( 作者, "reorder", "history-work", ("future", "other", "past"), 预期目录版本=3 ) with pytest.raises(上下文错误) as 错误: 工具.调用(工具请求("stale", "read_history_document", {"chapter_id": "past", "revision": 1})) assert 错误.value.错误码 == "SOURCE_STALE" @pytest.mark.case_id( "NC-ctx-history-version-span", environment="真实隔离PostgreSQL", given="请求未授权修订号或超出文本长度的码点片段", when="经工具请求读取", then=["分别以 SCOPE_DENIED 或 CONTEXT_INPUT_INVALID 拒绝，不返回正文"], contract="docs/系统架构/新版设计/模块设计/B09-上下文.md", ) @pytest.mark.parametrize( "参数", [{"chapter_id": "past", "revision": 2}, {"chapter_id": "past", "revision": 1, "end": 1000}], ids=["unapproved-version", "invalid-span"], ) def test_未批准版本与越界片段不返回正文__b09105(历史环境, 参数): _, _, 开始 = 历史环境 _, _, 工具 = 开始() with pytest.raises(上下文错误) as 错误: 工具.调用(工具请求("invalid", "read_history_document", 参数)) assert 错误.value.错误码 in {"SCOPE_DENIED", "CONTEXT_INPUT_INVALID"} @pytest.mark.case_id( "NC-ctx-history-fail-closed", environment="真实隔离PostgreSQL", given="陌生工具名、携带 work_id 扩权参数、删除 as_of 的范围、他人作者身份或不存在的修订号", when="授权或调用工具", then=[ "扩权参数与陌生工具由 S02 工具集拒绝；缺失截止返回 CONTEXT_SCOPE_INVALID", "他人作者 SCOPE_DENIED，缺失版本 SOURCE_NOT_FOUND", ], contract="docs/系统架构/新版设计/模块设计/B09-上下文.md", ) @pytest.mark.parametrize( "场景", ["unknown-tool", "work-id", "missing-cutoff", "wrong-author", "missing-version"], ids=str, ) def test_未登记工具扩权参数及不完整授权失败关闭__b09106(历史环境, 场景): 装配, 作者, 开始 = 历史环境 if 场景 == "wrong-author": with pytest.raises(上下文错误) as 错误: 装配.要求上下文().授权历史范围( replace(作者, 作者="another-author"), "history-work", 2, ({"chapter_id": "past", "revision": 1},), ) assert 错误.value.错误码 == "SCOPE_DENIED" return if 场景 == "missing-version": with pytest.raises(上下文错误) as 错误: 装配.要求上下文().授权历史范围( 作者, "history-work", 2, ({"chapter_id": "past", "revision": 99},) ) assert 错误.value.错误码 == "SOURCE_NOT_FOUND" return 范围 = 装配.要求上下文().授权历史范围( 作者, "history-work", 2, ({"chapter_id": "past", "revision": 1},) ) if 场景 == "missing-cutoff": del 范围["as_of"] with pytest.raises(上下文错误) as 错误: 开始(范围) assert 错误.value.错误码 == "CONTEXT_SCOPE_INVALID" return _, _, 工具 = 开始(范围) 参数 = {"chapter_id": "past", "revision": 1} if 场景 == "work-id": 参数["work_id"] = "another-work" with pytest.raises(模型协议错误): 工具.调用( 工具请求( "no-tool", "read_unknown" if 场景 == "unknown-tool" else "read_history_document", 参数, ) )