From f7a57f68e68412b489b4ebcae254e977fc6d3e7e Mon Sep 17 00:00:00 2001 From: zizi Date: Thu, 10 Sep 2026 19:25:41 +0800 Subject: [PATCH] =?UTF-8?q?W14=20=E5=8F=97=E9=99=90=E4=B8=8A=E4=B8=8B?= =?UTF-8?q?=E6=96=87=E3=80=81=E5=B7=A5=E5=85=B7=E4=B8=8E=E6=8E=A2=E7=B4=A2?= =?UTF-8?q?=E5=9B=9E=E6=94=BE=EF=BC=9A=E9=A6=96=E5=88=87=E7=89=87=E3=80=8C?= =?UTF-8?q?=E5=8F=97=E9=99=90=E5=8E=86=E5=8F=B2=E6=AD=A3=E6=96=87=E8=AF=BB?= =?UTF-8?q?=E5=8F=96=E4=B8=8E=E6=9D=90=E6=96=99=E9=A2=84=E7=AE=97=E3=80=8D?= =?UTF-8?q?=EF=BC=9A=E6=9D=90=E6=96=99=E9=A2=84=E7=AE=97=E7=BA=AF=E5=87=BD?= =?UTF-8?q?=E6=95=B0=E3=80=81=E6=8E=88=E6=9D=83=E5=8E=86=E5=8F=B2=E8=8C=83?= =?UTF-8?q?=E5=9B=B4=E3=80=81read=5Fhistory=5Fdocument=20=E5=85=B7?= =?UTF-8?q?=E5=90=8D=E5=8F=AA=E8=AF=BB=E5=B7=A5=E5=85=B7=E6=8E=A5=E5=85=A5?= =?UTF-8?q?=20S02=20=E5=B9=B6=E7=95=99=E7=97=95=EF=BC=9B13=20=E4=B8=AA?= =?UTF-8?q?=E6=96=B0=20NC=20=E9=80=9A=E8=BF=87=EF=BC=88=E7=A6=BB=E7=BA=BF?= =?UTF-8?q?=20237=E3=80=81=E9=9D=9E=E5=AE=BF=E4=B8=BB=20PG=20173=EF=BC=89?= =?UTF-8?q?=E3=80=82W14=20=E6=95=B4=E4=BD=93=20in=5Fprogress=EF=BC=8C?= =?UTF-8?q?=E5=B0=9A=E6=9C=AA=E9=AA=8C=E6=94=B6=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按 R2 串行阶段整理提交；包内文件为该阶段交付（含后续小增量），状态以工作包清单为准。 --- src/muse/上下文/__init__.py | 1 + src/muse/上下文/任务范围.py | 189 +++++++++++++++++++++ src/muse/上下文/接口.py | 25 +++ src/muse/上下文/材料装配.py | 71 ++++++++ src/muse/上下文/来源读取.py | 136 +++++++++++++++ src/muse/上下文/模型.py | 32 ++++ src/muse/接入/只读工具.py | 41 +++++ tests/单元/test_上下文范围与预算.py | 89 ++++++++++ tests/集成/test_上下文历史正文.py | 245 ++++++++++++++++++++++++++++ 9 files changed, 829 insertions(+) create mode 100644 src/muse/上下文/__init__.py create mode 100644 src/muse/上下文/任务范围.py create mode 100644 src/muse/上下文/接口.py create mode 100644 src/muse/上下文/材料装配.py create mode 100644 src/muse/上下文/来源读取.py create mode 100644 src/muse/上下文/模型.py create mode 100644 src/muse/接入/只读工具.py create mode 100644 tests/单元/test_上下文范围与预算.py create mode 100644 tests/集成/test_上下文历史正文.py diff --git a/src/muse/上下文/__init__.py b/src/muse/上下文/__init__.py new file mode 100644 index 0000000..36f1aa9 --- /dev/null +++ b/src/muse/上下文/__init__.py @@ -0,0 +1 @@ +"""上下文模块；公开用例与合同统一经接口.py，材料预算见材料装配.py。""" diff --git a/src/muse/上下文/任务范围.py b/src/muse/上下文/任务范围.py new file mode 100644 index 0000000..ac91f15 --- /dev/null +++ b/src/muse/上下文/任务范围.py @@ -0,0 +1,189 @@ +"""服务端授权并恢复历史正文读取范围；执行阶段只从已保存任务取回。 + +合同依据：模块设计/B09-上下文.md、接口契约/任务工具与事件.md。 + +- 授权只在服务端发生：作者、作品、叙事截止与文稿清单由作者动作传入， + 模型侧工具不接受其中任何一项。 +- 范围随任务冻结；恢复时校验形状与任务身份一致，缺失截止不是无限， + 也不静默改读最新版本。 +- 正文与目录只经 B01/B05 公开接口读取；本文件不写跨模块 SQL。 +""" + +from __future__ import annotations + +from typing import Any + +from psycopg import Connection + +from muse.上下文.模型 import 上下文错误 +from muse.任务运行.接口 import 任务快照, 任务服务 +from muse.作品规划.接口 import 作品错误, 读取作品范围, 读取章节归属 +from muse.共享.调用身份 import 调用身份 +from muse.基础设施.数据库.连接 import 数据库工厂 +from muse.正文写作.接口 import 读取正文依据 +from muse.正文写作.模型 import 正文错误 + +_文稿必需键 = frozenset( + {"chapter_id", "branch_id", "revision", "document_id", "document_hash", "position"} +) + + +def 校验截止(值: Any) -> int: + """叙事截止必须是明确的非负整数；bool 与缺失都不是截止。""" + if isinstance(值, bool) or not isinstance(值, int) or 值 < 0: + raise 上下文错误("CONTEXT_SCOPE_INVALID", f"叙事截止必须是非负整数，收到 {值!r}") + return 值 + + +def _拒绝(错误: 作品错误, 前因: str) -> 上下文错误: + 代码 = "SOURCE_NOT_FOUND" if 错误.错误码 == "CHAPTER_NOT_FOUND" else 错误.错误码 + return 上下文错误(代码, f"{前因}：{错误.说明}") + + +class 上下文服务: + """授权历史范围并从已保存任务恢复冻结范围；读取范围失败即关闭。""" + + def __init__(self, 数据库: 数据库工厂, 任务: 任务服务) -> None: + self.数据库 = 数据库 + self.任务 = 任务 + + def 授权历史范围( + self, 身份: 调用身份, 作品ID: str, 截止: int, 文稿查询: tuple[dict, ...] + ) -> dict: + """按作者、作品与叙事截止固定可读文稿、目录版本与正文哈希；只读不写入。""" + if not 作品ID: + raise 上下文错误("CONTEXT_SCOPE_INVALID", "历史范围必须绑定明确作品") + if 身份.作者 is None or 身份.用途 is not self.数据库.用途: + raise 上下文错误("SCOPE_DENIED", "历史范围授权需要与装配用途一致的作者身份") + 校验截止(截止) + if not isinstance(文稿查询, tuple) or not 文稿查询: + raise 上下文错误("CONTEXT_SCOPE_INVALID", "历史范围至少声明一个文稿来源") + with self.数据库.连接(只读=True) as 连: + 目录 = self._读取目录(连, 身份.作者, 作品ID) + 章节 = {c["chapter_id"]: c for c in 目录["chapters"]} + 文稿 = [self._授权文稿(连, 身份.作者, 作品ID, 截止, 章节, 查询) for 查询 in 文稿查询] + if len({d["document_id"] for d in 文稿}) != len(文稿): + raise 上下文错误("CONTEXT_SCOPE_INVALID", "同一文稿在历史范围内重复声明") + return { + "work_id": 作品ID, + "as_of": 截止, + "source_ids": [d["document_id"] for d in 文稿], + "documents": 文稿, + "directory_revision": 目录["revision"], + "authorized_by": 身份.作者, + } + + def 读取范围(self, 任务ID: str) -> dict: + """从已保存任务恢复冻结范围；形状或身份不一致即失败关闭。""" + return self.恢复冻结范围(self.任务.读取任务(任务ID)) + + def 恢复冻结范围(self, 任务: 任务快照) -> dict: + """校验任务内冻结范围完整且与任务、装配用途一致，然后原样返回。""" + 固定 = 任务.冻结输入 + 范围 = (固定.get("冻结上下文") or {}).get("source_scope") + if not isinstance(范围, dict): + raise 上下文错误("CONTEXT_SCOPE_INVALID", "任务缺少冻结来源范围") + if 范围.get("work_id") != 固定.get("作品ID"): + raise 上下文错误("CONTEXT_SCOPE_INVALID", "冻结范围与任务作品不一致") + if 范围.get("authorized_by") != 任务.作者: + raise 上下文错误("CONTEXT_SCOPE_INVALID", "冻结范围与任务作者不一致") + if str(固定.get("执行用途")) != self.数据库.用途.value: + raise 上下文错误("CONTEXT_SCOPE_INVALID", "任务执行用途与装配用途不一致") + 校验截止(范围.get("as_of")) + 目录版本 = 范围.get("directory_revision") + if isinstance(目录版本, bool) or not isinstance(目录版本, int) or 目录版本 < 0: + raise 上下文错误( + "CONTEXT_SCOPE_INVALID", f"冻结目录版本必须是非负整数，收到 {目录版本!r}" + ) + 文稿 = 范围.get("documents") + if not isinstance(文稿, list) or not 文稿: + raise 上下文错误("CONTEXT_SCOPE_INVALID", "冻结范围没有已授权文稿") + for 项 in 文稿: + if not isinstance(项, dict) or not _文稿必需键 <= 项.keys(): + raise 上下文错误("CONTEXT_SCOPE_INVALID", "冻结文稿缺少身份、版本或哈希绑定") + 版本 = 项["revision"] + if isinstance(版本, bool) or not isinstance(版本, int) or 版本 < 1: + raise 上下文错误( + "CONTEXT_SCOPE_INVALID", f"冻结文稿版本必须是正整数，收到 {版本!r}" + ) + 位置 = 项["position"] + if isinstance(位置, bool) or not isinstance(位置, int) or 位置 < 1: + raise 上下文错误( + "CONTEXT_SCOPE_INVALID", f"冻结文稿目录位置必须是正整数，收到 {位置!r}" + ) + for 键 in ("chapter_id", "branch_id", "document_id", "document_hash"): + if not isinstance(项[键], str) or not 项[键]: + raise 上下文错误("CONTEXT_SCOPE_INVALID", f"冻结文稿的 {键} 必须是非空字符串") + 身份集 = 范围.get("source_ids") + if ( + not isinstance(身份集, list) + or not 身份集 + or not all(isinstance(i, str) and i for i in 身份集) + or len(set(身份集)) != len(身份集) + or set(身份集) != {项["document_id"] for 项 in 文稿} + ): + raise 上下文错误("CONTEXT_SCOPE_INVALID", "冻结来源身份与文稿清单不一致") + return 范围 + + @staticmethod + def _读取目录(连: Connection, 作者: str, 作品ID: str) -> dict: + try: + return 读取作品范围(连, 作者, 作品ID) + except 作品错误 as 错误: + raise _拒绝(错误, f"作品 {作品ID} 的历史范围授权被拒绝") from None + + @staticmethod + def _授权文稿( + 连: Connection, 作者: str, 作品ID: str, 截止: int, 章节: dict, 查询: dict + ) -> dict: + if not isinstance(查询, dict): + raise 上下文错误("CONTEXT_SCOPE_INVALID", f"文稿查询必须显式声明：{查询!r}") + chapter_id = 查询.get("chapter_id") + revision = 查询.get("revision") + 分支 = 查询.get("branch_id", "main") + if not isinstance(chapter_id, str) or not chapter_id: + raise 上下文错误("CONTEXT_SCOPE_INVALID", "文稿查询缺少 chapter_id") + if isinstance(revision, bool) or not isinstance(revision, int) or revision < 1: + raise 上下文错误("CONTEXT_SCOPE_INVALID", f"文稿版本必须是正整数，收到 {revision!r}") + if not isinstance(分支, str) or not 分支: + raise 上下文错误("CONTEXT_SCOPE_INVALID", "文稿分支必须是非空字符串") + 行 = 章节.get(chapter_id) + if 行 is None: + 归属 = None + try: + 归属 = 读取章节归属(连, 作者, chapter_id) + except 作品错误 as 错误: + raise _拒绝(错误, f"章节 {chapter_id} 无法授权") from None + # 章节真实存在但不在目标作品：明确拒绝，不猜测替代作品。 + raise 上下文错误( + "SCOPE_DENIED", + f"章节 {chapter_id} 不属于作品 {作品ID}（实际属于 {归属['work_id']}）", + ) + if 行["position"] > 截止: + raise 上下文错误( + "SCOPE_DENIED", + f"章节 {chapter_id} 位于叙事截止 {截止} 之后（目录位置 {行['position']}）", + ) + try: + 依据 = 读取正文依据(连, 作者, chapter_id, revision, 分支=分支) + except 正文错误 as 错误: + raise 上下文错误( + "SOURCE_NOT_FOUND", f"章节 {chapter_id} 的正文版本 {revision} 不存在：{错误.说明}" + ) from None + if 依据["current_revision"] != revision: + raise 上下文错误( + "SOURCE_STALE", + f"章节 {chapter_id} 当前版本为 {依据['current_revision']}，" + f"与指定版本 {revision} 不一致；历史版本不再用于新任务", + ) + return { + "chapter_id": chapter_id, + "branch_id": 分支, + "revision": revision, + "document_id": 依据["document_id"], + "document_hash": 依据["document_hash"], + "position": 行["position"], + } + + +__all__ = ["上下文服务", "校验截止"] diff --git a/src/muse/上下文/接口.py b/src/muse/上下文/接口.py new file mode 100644 index 0000000..0bb94b1 --- /dev/null +++ b/src/muse/上下文/接口.py @@ -0,0 +1,25 @@ +"""上下文模块唯一公开入口；跨模块只经本文件导入。""" + +from muse.上下文.任务范围 import 上下文服务, 校验截止 +from muse.上下文.材料装配 import 装配材料 +from muse.上下文.来源读取 import ( + 历史正文参数合同, + 历史正文工具, + 投影合同, + 投影版本, + 读取历史正文, +) +from muse.上下文.模型 import 上下文错误, 材料装配结果 + +__all__ = [ + "上下文错误", + "上下文服务", + "材料装配结果", + "装配材料", + "历史正文参数合同", + "历史正文工具", + "投影合同", + "投影版本", + "读取历史正文", + "校验截止", +] diff --git a/src/muse/上下文/材料装配.py b/src/muse/上下文/材料装配.py new file mode 100644 index 0000000..94537dc --- /dev/null +++ b/src/muse/上下文/材料装配.py @@ -0,0 +1,71 @@ +"""材料按必读优先与 UTF-8 字节预算取舍；省略有原因，不补造材料。 + +边界：纯函数，不访问数据库、模型、文件或网络，也不承担来源授权。 +字节数为材料渲染后的 UTF-8 字节（含 ``\\n\\n`` 分隔符），不是 tokenizer 或计费估算。 +""" + +from __future__ import annotations + +from collections.abc import Mapping + +from muse.上下文.模型 import 上下文错误, 材料装配结果 + +分隔符 = "\n\n" + + +def 装配材料( + 材料: Mapping[str, str], + 必读: tuple[str, ...], + *, + 字节上限: int, +) -> 材料装配结果: + """按必读优先、输入顺序整项加入可选，返回实际入选与省略条目。 + + 必读按声明顺序先放入；缺失或空白必读拒绝，任一必读放不下即整体拒绝，不返回部分材料。 + 可选材料按输入顺序整项加入，空白与超预算分别记入省略原因；不截断中文字节，保留原文空白。 + 字节上限必须是真正的 int：True 是 bool 不是预算，与零、负数、浮点一并拒绝。 + """ + if isinstance(字节上限, bool) or not isinstance(字节上限, int) or 字节上限 <= 0: + raise 上下文错误("CONTEXT_INPUT_INVALID", f"字节上限必须是正整数，收到 {字节上限!r}") + if len(set(必读)) != len(必读): + raise 上下文错误("CONTEXT_INPUT_INVALID", f"必读来源身份重复：{tuple(必读)}") + + 分隔字节 = len(分隔符.encode("utf-8")) + 片段: list[str] = [] + 入选: list[str] = [] + 字节数 = 0 + for 来源 in 必读: + 文本 = 材料.get(来源) + if 文本 is None or not 文本.strip(): + raise 上下文错误("CONTEXT_REQUIRED_MISSING", f"必读来源缺失或为空：{来源}") + 必读总字节 = 字节数 + len(文本.encode("utf-8")) + (分隔字节 if 片段 else 0) + if 必读总字节 > 字节上限: + raise 上下文错误( + "CONTEXT_BUDGET_EXCEEDED", + f"必读材料需要 {必读总字节} 字节，超出上限 {字节上限}", + ) + 片段.append(文本) + 入选.append(来源) + 字节数 = 必读总字节 + + 省略: list[tuple[str, str]] = [] + for 来源, 文本 in 材料.items(): + if 来源 in 入选: + continue + if not 文本.strip(): + 省略.append((来源, "EMPTY")) + continue + 增量 = len(文本.encode("utf-8")) + (分隔字节 if 片段 else 0) + if 字节数 + 增量 > 字节上限: + 省略.append((来源, "BUDGET_EXCEEDED")) + continue + 片段.append(文本) + 入选.append(来源) + 字节数 += 增量 + + return 材料装配结果( + 入选来源=tuple(入选), 文本=分隔符.join(片段), 字节数=字节数, 省略=tuple(省略) + ) + + +__all__ = ["装配材料"] diff --git a/src/muse/上下文/来源读取.py b/src/muse/上下文/来源读取.py new file mode 100644 index 0000000..993edcd --- /dev/null +++ b/src/muse/上下文/来源读取.py @@ -0,0 +1,136 @@ +"""按 B05/B01 公开接口读取冻结范围内的历史正文；码点切片不做规范化。 + +合同依据：模块设计/B09-上下文.md、接口契约/正文结构与中文选区.md。 + +- 只返回冻结文稿、目录版本、正文版本与哈希全部匹配的可见文本； + 任一漂移返回 SOURCE_STALE，不静默改读最新版。 +- 投影版本由固定 visible-text-codepoints/v1 合同与两种用途确定性派生， + 不伪造 S04 动态结构绑定。 +""" + +from __future__ import annotations + +from psycopg import Connection + +from muse.上下文.模型 import 上下文错误 +from muse.任务运行.接口 import 工具定义, 工具来源, 工具结果, 工具范围 +from muse.作品规划.接口 import 作品错误, 读取作品范围 +from muse.基础设施.数据库.连接 import 数据库工厂 +from muse.正文写作.接口 import 可见文本, 可见文本哈希, 读取正文依据 +from muse.正文写作.模型 import 正文错误 + +投影合同 = "visible-text-codepoints/v1" + +历史正文参数合同 = { + "type": "object", + "additionalProperties": False, + "required": ["chapter_id", "revision"], + "properties": { + "chapter_id": {"type": "string", "minLength": 1}, + "revision": {"type": "integer", "minimum": 1}, + "branch_id": {"type": "string", "minLength": 1}, + "start": {"type": "integer", "minimum": 0}, + "end": {"type": "integer", "minimum": 0}, + }, +} + + +def 投影版本(运行用途: str, 内容用途: str) -> str: + """由固定投影合同与两种用途确定性派生；同一输入永远同一版本。""" + return f"{投影合同}:{运行用途}:{内容用途}" + + +def 读取历史正文(数据库: 数据库工厂, 冻结范围: dict, 范围: 工具范围, 参数: dict) -> 工具结果: + """只读冻结文稿匹配且目录、版本与哈希一致的可见文本码点片段。""" + 键 = (参数["chapter_id"], 参数.get("branch_id", "main"), 参数["revision"]) + 目标 = next( + ( + d + for d in 冻结范围["documents"] + if (d["chapter_id"], d["branch_id"], d["revision"]) == 键 + ), + None, + ) + if 目标 is None: + raise 上下文错误( + "SCOPE_DENIED", f"文稿（章 {键[0]}、分支 {键[1]}、版本 {键[2]}）不在冻结范围内" + ) + 作者 = 冻结范围["authorized_by"] + with 数据库.连接(只读=True) as 连: + 依据 = _核对并读取(连, 作者, 冻结范围, 键, 目标) + 文本 = 可见文本(依据["document"]) + start = 参数.get("start", 0) + end = 参数.get("end", len(文本)) + if not 0 <= start <= end <= len(文本): + raise 上下文错误( + "CONTEXT_INPUT_INVALID", + f"码点片段 [{start}:{end}) 超出可见文本长度 {len(文本)}", + ) + return 工具结果( + 内容={ + "chapter_id": 键[0], + "document_id": 依据["document_id"], + "revision": 键[2], + "text": 文本[start:end], + "start": start, + "end": end, + "total_codepoints": len(文本), + "visible_text_hash": 可见文本哈希(依据["document"]), + "run_purpose": 范围.运行用途, + "content_purpose": 范围.内容用途, + }, + 来源=( + 工具来源( + 来源ID=依据["document_id"], + 数据版本=str(键[2]), + 结构哈希=依据["document_hash"], + 投影版本=投影版本(范围.运行用途, 范围.内容用途), + ), + ), + ) + + +def _核对并读取( + 连: Connection, 作者: str, 冻结范围: dict, 键: tuple[str, str, int], 目标: dict +) -> dict: + try: + 目录 = 读取作品范围(连, 作者, 冻结范围["work_id"]) + except 作品错误 as 错误: + raise 上下文错误("SOURCE_STALE", f"作品目录已不可用：{错误.说明}") from None + 当前 = next((c for c in 目录["chapters"] if c["chapter_id"] == 键[0]), None) + if ( + 目录["revision"] != 冻结范围["directory_revision"] + or 当前 is None + or 当前["position"] != 目标["position"] + ): + raise 上下文错误( + "SOURCE_STALE", + f"作品目录已从版本 {冻结范围['directory_revision']} 漂移；任务范围失效，需重新授权", + ) + try: + 依据 = 读取正文依据(连, 作者, 键[0], 键[2], 分支=键[1]) + except 正文错误 as 错误: + raise 上下文错误("SOURCE_NOT_FOUND", f"冻结正文版本已不存在：{错误.说明}") from None + if 依据["current_revision"] != 键[2] or 依据["document_hash"] != 目标["document_hash"]: + raise 上下文错误( + "SOURCE_STALE", + f"文稿 {键[0]} 已漂移（当前版本 {依据['current_revision']}）；任务范围失效", + ) + return 依据 + + +def 历史正文工具(数据库: 数据库工厂, 冻结范围: dict) -> 工具定义: + """为已冻结范围构建本轮唯一具名只读工具；参数只含查询条件。""" + + def 读取(范围: 工具范围, 参数: dict) -> 工具结果: + return 读取历史正文(数据库, 冻结范围, 范围, 参数) + + return 工具定义( + "read_history_document", + "读取任务冻结范围内的历史正文可见文本；start/end 为 Unicode 码点左闭右开片段", + 历史正文参数合同, + 读取, + ) + + +__all__ = ["历史正文参数合同", "历史正文工具", "投影合同", "投影版本", "读取历史正文"] diff --git a/src/muse/上下文/模型.py b/src/muse/上下文/模型.py new file mode 100644 index 0000000..e74f906 --- /dev/null +++ b/src/muse/上下文/模型.py @@ -0,0 +1,32 @@ +"""上下文材料的稳定对象；错误带实例错误码与中文原因。""" + +from __future__ import annotations + +from dataclasses import dataclass + +from muse.共享.错误 import Muse错误 + + +class 上下文错误(Muse错误): + """上下文输入非法、必读缺项或必读超预算的稳定错误。 + + :param 错误码: CONTEXT_INPUT_INVALID、CONTEXT_REQUIRED_MISSING 或 CONTEXT_BUDGET_EXCEEDED。 + :param 原因: 面向作者的中文说明，直陈缺项、越界值或超出量。 + """ + + def __init__(self, 错误码: str, 原因: str) -> None: + self.错误码 = 错误码 + super().__init__(原因) + + +@dataclass(frozen=True, slots=True) +class 材料装配结果: + """一次材料装配的实际入选与省略；不补造未入选材料。""" + + 入选来源: tuple[str, ...] + 文本: str + 字节数: int + 省略: tuple[tuple[str, str], ...] + + +__all__ = ["上下文错误", "材料装配结果"] diff --git a/src/muse/接入/只读工具.py b/src/muse/接入/只读工具.py new file mode 100644 index 0000000..ec1576b --- /dev/null +++ b/src/muse/接入/只读工具.py @@ -0,0 +1,41 @@ +"""从已领取步骤装配 S02 只读工具集；接入层只转交，不写来源规则。 + +合同依据：文件设计/后端-装配与接入.md、接口契约/任务工具与事件.md。 +- 工具范围完全取自任务冻结输入；本文件不重新解释作者、作品或截止。 +- 只安装已实现的历史正文工具；未实现的方法、资料等工具不伪返回成功。 +""" + +from __future__ import annotations + +from uuid import uuid4 + +from muse.上下文.接口 import 上下文服务, 历史正文工具 +from muse.任务运行.接口 import 事件类型, 任务服务, 只读工具集, 工具范围, 领取凭证 + + +def 领取只读工具(上下文: 上下文服务, 运行: 任务服务, 领取: 领取凭证) -> 只读工具集: + """按任务冻结范围安装已实现工具；成功读取复用 S02 工具读取事件留痕。""" + 任务 = 运行.读取任务(领取.任务ID) + 冻结 = 上下文.恢复冻结范围(任务) + 步骤 = next(s for s in 任务.流程.步骤 if s.步骤ID == 领取.步骤ID) + 固定 = 任务.冻结输入 + + def 记录(载荷: dict) -> None: + 运行.追加事件(领取, 事件类型.工具读取, 载荷, 事件ID=str(uuid4())) + + return 只读工具集( + (历史正文工具(上下文.数据库, 冻结),), + 步骤.工具, + 工具范围( + 任务.任务ID, + 冻结["work_id"], + tuple(冻结["source_ids"]), + 冻结["as_of"], + 固定["执行用途"], + 固定["内容用途"], + ), + 记录, + ) + + +__all__ = ["领取只读工具"] diff --git a/tests/单元/test_上下文范围与预算.py b/tests/单元/test_上下文范围与预算.py new file mode 100644 index 0000000..5b8b265 --- /dev/null +++ b/tests/单元/test_上下文范围与预算.py @@ -0,0 +1,89 @@ +"""材料装配的必读优先、UTF-8 字节预算与省略原因；离线纯函数合同。 + +字节期望值按 UTF-8 逐字写死，不由被测函数自身推算，避免同源计算自证。 +""" + +import dataclasses + +import pytest + +from muse.上下文.接口 import 上下文错误, 装配材料 + + +def test_必读优先且可选按原顺序整项加入__1c4a9e() -> None: + 材料 = {"可选甲": "甲文", "必读": "必读文", "可选乙": "乙文"} + 结果 = 装配材料(材料, ("必读",), 字节上限=64) + assert 结果.入选来源 == ("必读", "可选甲", "可选乙") + assert 结果.文本 == "必读文\n\n甲文\n\n乙文" + assert 结果.字节数 == 25 # 三个片段 9+6+6 字节加两处分隔符 + assert 结果.省略 == () + with pytest.raises(dataclasses.FrozenInstanceError): + 结果.文本 = "改写" + + +def test_中文与emoji按utf8字节计数含分隔符__5d1c47() -> None: + 原文 = "第1章：雨。🌧" + 仅必读 = 装配材料({"历史": 原文}, ("历史",), 字节上限=20) + assert 仅必读.文本 == 原文 + assert 仅必读.字节数 == 20 # 含四字节 emoji + + 材料 = {"历史": 原文, "旁观": "他站着"} + 齐 = 装配材料(材料, ("历史",), 字节上限=31) + assert 齐.入选来源 == ("历史", "旁观") + assert 齐.字节数 == 31 # 20 + 分隔符 2 + 9 + assert 齐.文本 == f"{原文}\n\n他站着" + + 缺一字节 = 装配材料(材料, ("历史",), 字节上限=30) + assert 缺一字节.入选来源 == ("历史",) + assert 缺一字节.文本 == 原文 + assert 缺一字节.省略 == (("旁观", "BUDGET_EXCEEDED"),) + + +@pytest.mark.parametrize( + ("材料", "必读"), + [({}, ("孤本第三章",)), ({"孤本第三章": " \n "}, ("孤本第三章",))], + ids=["missing", "blank"], +) +def test_必读缺失或空白拒绝__a7e903(材料: dict[str, str], 必读: tuple[str, ...]) -> None: + with pytest.raises(上下文错误) as 异常: + 装配材料(材料, 必读, 字节上限=64) + assert 异常.value.错误码 == "CONTEXT_REQUIRED_MISSING" + assert "孤本第三章" in 异常.value.说明 + + +def test_必读整体超预算拒绝__c9e51f() -> None: + 材料 = {"甲": "必读甲", "乙": "必读乙", "可选": "可选丙"} + with pytest.raises(上下文错误) as 异常: + 装配材料(材料, ("甲", "乙"), 字节上限=19) + assert 异常.value.错误码 == "CONTEXT_BUDGET_EXCEEDED" + assert "20" in 异常.value.说明 + + 恰好 = 装配材料(材料, ("甲", "乙"), 字节上限=20) # 9 + 分隔符 2 + 9 + assert 恰好.入选来源 == ("甲", "乙") + assert 恰好.字节数 == 20 + assert 恰好.省略 == (("可选", "BUDGET_EXCEEDED"),) + + +def test_可选空白与超预算分别给出省略原因__4f8ab2() -> None: + 材料 = {"必读": "必读文", "空白": " ", "太长": "长" * 40, "入选": "短文"} + 结果 = 装配材料(材料, ("必读",), 字节上限=17) # 9 + 分隔符 2 + 6 + assert 结果.入选来源 == ("必读", "入选") + assert 结果.省略 == (("空白", "EMPTY"), ("太长", "BUDGET_EXCEEDED")) + + +@pytest.mark.parametrize( + ("字节上限", "必读"), + [(0, ("必读",)), (-3, ("必读",)), (True, ("必读",)), (64, ("必读", "必读"))], + ids=["zero", "negative", "boolean", "duplicate"], +) +def test_非法预算或重复必读身份拒绝__e2168c(字节上限: int, 必读: tuple[str, ...]) -> None: + with pytest.raises(上下文错误) as 异常: + 装配材料({"必读": "必读文"}, 必读, 字节上限=字节上限) + assert 异常.value.错误码 == "CONTEXT_INPUT_INVALID" + + +def test_原文不裁剪空白且按原字节保留__b4d7a1() -> None: + 原文 = " 第1章 \n雨还在下。\t\n" + 结果 = 装配材料({"历史": 原文}, ("历史",), 字节上限=29) + assert 结果.文本 == 原文 + assert 结果.字节数 == 29 diff --git a/tests/集成/test_上下文历史正文.py b/tests/集成/test_上下文历史正文.py new file mode 100644 index 0000000..5ddf615 --- /dev/null +++ b/tests/集成/test_上下文历史正文.py @@ -0,0 +1,245 @@ +"""合成流程驱动真实 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 流程定义, 流程服务, 流程登记 +from muse.配置 import 应用配置 + +pytestmark = pytest.mark.数据库 +原文 = "　雨夜🌧，e\u0301。\n门还开着。" +前段, 后段 = 原文.split("\n") + + +def 原文草稿(章: str) -> 正文草稿: + """段内换行走显式换行节点；可见文本仍与原文逐字一致。""" + return 正文草稿((段落("p-" + 章, (文本节点(前段), 换行节点(), 文本节点(后段))),)) + + +@pytest.fixture +def 历史环境(应用测试库): + with 应用测试库[用途.维护].连接() as 连, 连.transaction(): + 导入内置结构(元数据服务(连)) + 登记 = 流程登记() + 装配 = 构建(应用配置(应用测试库[用途.生产].引用, "test"), 流程=登记) + 作者 = 调用身份("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_, 领取, 装配.要求只读工具(领取) + + return 装配, 作者, 开始 + + +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" + + +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.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.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.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.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", + 参数, + ) + )