私人笔记
+作品探索
++ 先保留一个念头,也可以留下暂时想不明白的地方。 +
++ 还没有探索稿。从一个人物、一件物品或一句话开始。 +
+ )} +{draft.title}
+继续整理想法 →
+diff --git a/.agent/skills/操作/完善故事基础设定/SKILL.md b/.agent/skills/操作/完善故事基础设定/SKILL.md new file mode 100644 index 0000000..427e0a5 --- /dev/null +++ b/.agent/skills/操作/完善故事基础设定/SKILL.md @@ -0,0 +1,22 @@ +--- +id: develop-story-foundation +name: 完善故事基础设定 +category: operation +contract_version: 1 +commands: [list_explorations, read_exploration, save_exploration, check_exploration_handoff, select_exploration, read_exploration_selection] +description: 通过新版作者入口保存私人探索原话与分歧,整理交付时检查格式;内容和方向由作者决定。 +--- + +# 完善故事基础设定 + +先读取作者已有素材和本轮点名的探索稿。素材、联系、中心、结构可以来回发展,不把九章完整度、候选数量或字数设为开始条件。作者反馈只作用于点名的内容,不把局部采纳扩大为正式规划确认。 + +使用明确配置的私人探索目录。`.venv/bin/python -m muse 探索 <作者配置> 列表`和`查看 <稿件ID>`读取已有稿。保存请求包含draft_id、title、content、expected_hash,以`保存 <私人请求JSON文件>`交给B01。新稿expected_hash为null,后续使用读回的draft_hash;内容保留原话,稿件名称与稳定身份分开。保存冲突先比较当前稿和自己的修改,再执行作者选择。 + +日常内容可以很短、不完整或互相矛盾,不运行最终格式检查。作者要求整理交付时,才按[交付结构](references/交付结构.json)整理已有内容:唯一作品标题,根设定、九章及待决与取舍依次出现;未探索部分明确说明,可选子标题按导航顺序展开。没有固定编号段或篇幅门。 + +整理稿保存后,执行`检查 <稿件ID> --预期哈希 <稿件哈希>`。结果只说明格式,失败返回非零状态;格式通过不证明根设定已获作者认可或内容自洽。 + +作者明确选定本稿并要求交给作品时,先用`查看选定 <作品ID>`读取当前选定版本,再用`选定 <私人请求JSON>`提交command_id和selection。selection包含work_id、draft_id、expected_hash和expected_revision。只使用作者实际点名的稿件版本及作品;不填confirmedByUser、confirmedAt或作者身份。S01将所选快照和回执一起保存,后续改稿不改变已选快照。回执未返回时保持原请求重试。选定方向只授权其成为规划来源,正式规划候选还需要独立审阅与确认。 + +权威职责见[B01作品规划](../../../../docs/系统架构/新版设计/模块设计/B01-作品规划.md)。作者资料不进入共享文档,未交接草稿不进入正文生成上下文。 diff --git a/.agent/skills/操作/完善故事基础设定/references/交付结构.json b/.agent/skills/操作/完善故事基础设定/references/交付结构.json new file mode 100644 index 0000000..245624f --- /dev/null +++ b/.agent/skills/操作/完善故事基础设定/references/交付结构.json @@ -0,0 +1,391 @@ +{ + "contractVersion": "完善故事基础设定/output-v3", + "chapters": [ + { + "heading": "一、作品定位与核心命题", + "sections": [ + { + "heading": "1.1 作品形态与定位", + "units": [ + "1.1.1 题材、载体与目标读者", + "1.1.2 篇幅、连载方式与结构规模", + "1.1.3 同类作品中的差异" + ] + }, + { + "heading": "1.2 故事承诺与一句话梗概", + "units": [ + "1.2.1 一句话故事", + "1.2.2 主角起点、目标与外部期限", + "1.2.3 持续驱动力与核心看点" + ] + }, + { + "heading": "1.3 核心问题与主题命题", + "units": [ + "1.3.1 故事真正要追问的问题", + "1.3.2 人物、制度或社会提供的对立答案", + "1.3.3 结局必须回答和可以保留的问题" + ] + }, + { + "heading": "1.4 阅读体验与情绪基调", + "units": [ + "1.4.1 作品的情绪底色", + "1.4.2 压力、满足与释放的来源", + "1.4.3 作品区别于同类的质感" + ] + }, + { + "heading": "1.5 题材边界与创作禁区", + "units": [ + "1.5.1 可以使用的题材元素", + "1.5.2 不使用的套路和处理方式", + "1.5.3 不可破坏的主题、人物与规则边界" + ] + } + ] + }, + { + "heading": "二、故事世界与社会系统", + "sections": [ + { + "heading": "2.1 稳定世界规则", + "units": [ + "2.1.1 时空、基本存在与世界层级", + "2.1.2 自然、技术与超常规则", + "2.1.3 社会共同使用的尺度和分类", + "2.1.4 世界真实状态与当前未知边界" + ] + }, + { + "heading": "2.2 空间与舞台", + "units": [ + "2.2.1 主舞台卡", + "2.2.2 地点层级与人物活动范围", + "2.2.3 资源、交通与进入限制", + "2.2.4 舞台扩展和变化条件" + ] + }, + { + "heading": "2.3 社会结构、组织与制度", + "units": [ + "2.3.1 权力层级与治理方式", + "2.3.2 组织的职责、资源与约束", + "2.3.3 制度、资格、法律与责任", + "2.3.4 组织之间的关系与利益冲突", + "2.3.5 社会流动与阶层边界" + ] + }, + { + "heading": "2.4 经济、资源与职业", + "units": [ + "2.4.1 货币、价值与基本生活", + "2.4.2 资源的生产、流通与分配", + "2.4.3 职业、资格、收入与上升路径", + "2.4.4 稀缺、成本、风险与机会成本", + "2.4.5 资源量级变化的条件" + ] + }, + { + "heading": "2.5 信息传播与公共评价", + "units": [ + "2.5.1 信息渠道、传播速度与可信度", + "2.5.2 公共评价、声誉与证明材料", + "2.5.3 信息差、权限差与阶层差", + "2.5.4 公开事件引发的社会反应" + ] + } + ] + }, + { + "heading": "三、角色与关系", + "sections": [ + { + "heading": "3.1 主角", + "units": [ + "3.1.1 身份、起点与当前处境", + "3.1.2 欲望、需求与核心缺口", + "3.1.3 能力、资源与现实限制", + "3.1.4 价值观、底线与秘密", + "3.1.5 角色弧线与终局状态" + ] + }, + { + "heading": "3.2 核心角色群", + "units": [ + "3.2.1 原生关系与生活支点", + "3.2.2 同伴、盟友与协作者", + "3.2.3 竞争者与镜像角色", + "3.2.4 关键组织代理人和中介人物", + "3.2.5 角色的进入、退出与位置转移" + ] + }, + { + "heading": "3.3 关系网络与关系弧线", + "units": [ + "3.3.1 关系图与关系类型", + "3.3.2 关系的初始状态", + "3.3.3 关系变化的触发事件", + "3.3.4 关系中的利益、信息与责任", + "3.3.5 各阶段关系状态与最终走向" + ] + }, + { + "heading": "3.4 对手与阻力系统", + "units": [ + "3.4.1 对手来源与阻力类型", + "3.4.2 对手卡:目标、优势、盲区与代价", + "3.4.3 对手的手段、边界与让步条件", + "3.4.4 对手的适应、升级、转化与退出", + "3.4.5 制度、环境、资源和时间阻力" + ] + } + ] + }, + { + "heading": "四、故事机制与关键实体", + "sections": [ + { + "heading": "4.1 核心机制的不变量", + "units": [ + "4.1.1 机制的定义与存在形式", + "4.1.2 触发条件与输入", + "4.1.3 输出、反馈与可观察结果", + "4.1.4 适用范围与明确不能做什么", + "4.1.5 永久边界、异常情况与不可逆规则" + ] + }, + { + "heading": "4.2 成长、资源与代价", + "units": [ + "4.2.1 成长路径与解锁条件", + "4.2.2 资源来源、权属与转换", + "4.2.3 成本、消耗、恢复与机会成本", + "4.2.4 失败、损耗、补救与责任", + "4.2.5 不同阶段的规模与可用范围" + ] + }, + { + "heading": "4.3 能力和力量尺度", + "units": [ + "4.3.1 能力或力量的评价维度", + "4.3.2 等级、层级与认证方式", + "4.3.3 同级差异、越级边界与现场状态", + "4.3.4 人、装备、环境与对手的比较口径", + "4.3.5 能力提升的证明方式和记录" + ] + }, + { + "heading": "4.4 关键物件与资产生命周期", + "units": [ + "4.4.1 物件或资产卡片字段", + "4.4.2 来源、权属、取得与交接", + "4.4.3 当前状态、维护、使用与损耗", + "4.4.4 升级、替换、转让与损毁", + "4.4.5 生命周期对剧情和人物的作用" + ] + }, + { + "heading": "4.5 谜团、风险与长期冲突", + "units": [ + "4.5.1 核心谜团对象", + "4.5.2 异常现象与风险机制", + "4.5.3 威胁层级、表现与应对边界", + "4.5.4 已证事实、人物假设与未知部分", + "4.5.5 长期冲突与终局问题" + ] + } + ] + }, + { + "heading": "五、阶段与状态变化", + "sections": [ + { + "heading": "5.1 阶段划分与边界", + "units": [ + "5.1.1 阶段的命名与划分依据", + "5.1.2 阶段的核心问题与主要目标", + "5.1.3 阶段的进入条件", + "5.1.4 阶段的结束条件", + "5.1.5 阶段内部的卷、篇或故事单元" + ] + }, + { + "heading": "5.2 阶段状态矩阵", + "units": [ + "5.2.1 主角的身份、目标与责任", + "5.2.2 舞台、组织与社会状态", + "5.2.3 角色、关系与对手状态", + "5.2.4 资源、能力与资产状态", + "5.2.5 信息、公共评价与威胁状态", + "5.2.6 阶段入口、出口与遗留问题" + ] + }, + { + "heading": "5.3 转段事件与遗留问题", + "units": [ + "5.3.1 触发阶段变化的事件", + "5.3.2 状态变化和实际代价", + "5.3.3 带入下一阶段的资源与关系", + "5.3.4 本阶段解决和未解决的问题", + "5.3.5 转段结果的证据与核对条件" + ] + } + ] + }, + { + "heading": "六、情节架构与全书大纲", + "sections": [ + { + "heading": "6.1 主冲突与因果链", + "units": [ + "6.1.1 主线目标与终局方向", + "6.1.2 主线阻力与冲突机制", + "6.1.3 关键事件之间的因果链", + "6.1.4 转折、高潮与阶段里程碑", + "6.1.5 主线与主题命题的对应关系" + ] + }, + { + "heading": "6.2 各阶段或各卷大纲", + "units": [ + "6.2.1 阶段或各卷的目标", + "6.2.2 故事单元的分解与顺序", + "6.2.3 关键事件、结果与代价", + "6.2.4 角色、关系和设定的调用", + "6.2.5 阶段收束与下一阶段入口" + ] + }, + { + "heading": "6.3 当前阶段事件树", + "units": [ + "6.3.1 事件节点的统一字段", + "6.3.2 目标、阻力、行动与结果", + "6.3.3 分支、失败路径与替代方案", + "6.3.4 节奏、章节分配与章末状态", + "6.3.5 事件完成后的状态更新" + ] + }, + { + "heading": "6.4 并行线与回收", + "units": [ + "6.4.1 支线目标与独立价值", + "6.4.2 支线与主线的交汇点", + "6.4.3 支线角色、资源与关系", + "6.4.4 伏笔回收与线索合流", + "6.4.5 支线终止、转入主线与有意留白" + ] + } + ] + }, + { + "heading": "七、主题与读者体验弧线", + "sections": [ + { + "heading": "7.1 各阶段主题", + "units": [ + "7.1.1 阶段核心命题", + "7.1.2 主角在本阶段要回答的问题", + "7.1.3 对立价值与冲突答案", + "7.1.4 阶段结束时形成的暂时答案" + ] + }, + { + "heading": "7.2 情绪基调与阅读期待", + "units": [ + "7.2.1 阶段情绪底色", + "7.2.2 紧张、释放与安静段的安排", + "7.2.3 读者在本阶段期待看到什么", + "7.2.4 阶段切换时阅读感受如何变化" + ] + }, + { + "heading": "7.3 承诺、升级与兑现", + "units": [ + "7.3.1 本阶段需要兑现的旧承诺", + "7.3.2 本阶段新增的承诺与问题", + "7.3.3 能力、舞台与责任的升级", + "7.3.4 兑现证据与有意保留的未知" + ] + } + ] + }, + { + "heading": "八、信息、悬念与伏笔管理", + "sections": [ + { + "heading": "8.1 世界事实、角色知情、读者披露", + "units": [ + "8.1.1 世界中客观存在的事实", + "8.1.2 各角色当前知道什么", + "8.1.3 读者当前已经知道什么", + "8.1.4 三种状态允许存在的差异" + ] + }, + { + "heading": "8.2 信息披露节奏", + "units": [ + "8.2.1 第一次露头", + "8.2.2 局部确认", + "8.2.3 官方或公共定性", + "8.2.4 重大揭示与反转", + "8.2.5 揭示后改变了什么选择" + ] + }, + { + "heading": "8.3 伏笔登记与回收", + "units": [ + "8.3.1 埋设位置与当时含义", + "8.3.2 提醒方式和再次出现", + "8.3.3 回收条件", + "8.3.4 回收后改变的目标、关系或判断", + "8.3.5 故意留白与暂不回收" + ] + }, + { + "heading": "8.4 时间线与连续性核对", + "units": [ + "8.4.1 客观时间线", + "8.4.2 人物知情时间线", + "8.4.3 事件、阶段与披露的交叉核对" + ] + } + ] + }, + { + "heading": "九、叙事执行与一致性", + "sections": [ + { + "heading": "9.1 视角、语言和场景", + "units": [ + "9.1.1 视角、人称与叙述距离", + "9.1.2 场景的进入、推进与退出", + "9.1.3 设定如何通过动作进入正文", + "9.1.4 人物声音与对白边界" + ] + }, + { + "heading": "9.2 节奏、章末和对话", + "units": [ + "9.2.1 场景节奏与信息密度", + "9.2.2 对话、动作与叙述的比例", + "9.2.3 章内目标、转折与后果", + "9.2.4 章末钩子与下一章压力", + "9.2.5 安静段、恢复段与关系场景" + ] + }, + { + "heading": "9.3 设定落地与一致性检查", + "units": [ + "9.3.1 数字、单位与时间", + "9.3.2 规则、能力与代价", + "9.3.3 角色、关系与知情边界", + "9.3.4 资产、权属与损耗", + "9.3.5 阶段状态与伏笔更新" + ] + } + ] + } + ] +} diff --git a/.agent/skills/操作/完善故事基础设定/references/目录.md b/.agent/skills/操作/完善故事基础设定/references/目录.md new file mode 100644 index 0000000..a491627 --- /dev/null +++ b/.agent/skills/操作/完善故事基础设定/references/目录.md @@ -0,0 +1,3 @@ +| 名称 | 相对地址 | 内容描述 | 使用场景 | 使用要求 | +|------|----------|----------|----------|----------| +| 交付结构 | [交付结构.json](交付结构.json) | 最终整理稿的章节与展开导航 | 打包供B01格式校验 | 保留未知说明,不要求全部展开 | diff --git a/.agent/skills/操作/完善故事基础设定/目录.md b/.agent/skills/操作/完善故事基础设定/目录.md new file mode 100644 index 0000000..13846a1 --- /dev/null +++ b/.agent/skills/操作/完善故事基础设定/目录.md @@ -0,0 +1,4 @@ +| 名称 | 相对地址 | 内容描述 | 使用场景 | 使用要求 | +|------|----------|----------|----------|----------| +| 操作说明 | [SKILL.md](SKILL.md) | 私人探索保存、读回及最终格式检查 | 使用新版探索入口 | 不自动进入正式规划 | +| 交付资源 | [目录.md](references/目录.md) | 最终整理稿的格式资源 | W13探索交付检查 | 日常探索不按最终格式通关 | diff --git a/.agent/skills/操作/确认规划候选/SKILL.md b/.agent/skills/操作/确认规划候选/SKILL.md new file mode 100644 index 0000000..7768b04 --- /dev/null +++ b/.agent/skills/操作/确认规划候选/SKILL.md @@ -0,0 +1,22 @@ +--- +id: confirm-plan-candidate +name: 确认规划候选 +category: operation +contract_version: 1 +commands: [read_plan_candidate, edit_plan_candidate, open_plan_review, decide_plan_candidate, read_plan, list_plans] +description: 读取新版规划候选和固定版本,执行作者明确的采纳、拒绝或暂缓,并回查正式规划与事务回执。 +--- + +# 确认规划候选 + +先用`.venv/bin/python -m muse 规划 <作者配置> 查看候选 <候选ID>`读取内容。当前人工规划候选只做字段与版本检查,不表示模型或文学评审已经通过。 + +作者要修改候选时,使用`编辑候选 <私人请求JSON>`提交command_id、candidate(确切定位)与changes字段操作。服务端在同一身份下追加候选版本,原版保留。编辑后重新读取并打开审阅,旧审阅不能确认改后的内容。 + +向作者展示确切内容和保存后的影响。准备candidate_id、candidate_revision、candidate_hash、plan_id、expected_revision,执行`打开审阅 <私人定位JSON>`取得审阅记录。不能把打开审阅当作确认。 + +作者明确决定后,`决定 <私人请求JSON>`提交command_id、candidate(上述定位)、action、author_review_id、review_hash、approved_changes。action为adopt、reject或defer;采纳整份规划时approved_changes为`["plan"]`。使用实际审阅时的版本和哈希,不自动刷新成最新版本后重试采纳。 + +返回回执后,用`查看 <规划ID>`或`列表 <作品ID>`核对正式内容。网络结果未知时保持同一请求重试;参数改变使用新命令。陈旧候选需要重新比较,不能覆盖新的正式规划。规划确认只写B01,不代替正文或故事事实的独立确认。 + +职责和来源约束见[B01作品规划](../../../../docs/系统架构/新版设计/模块设计/B01-作品规划.md)与[命令与作者决策](../../../../docs/系统架构/新版设计/接口契约/命令与作者决策.md)。 diff --git a/.agent/skills/操作/确认规划候选/目录.md b/.agent/skills/操作/确认规划候选/目录.md new file mode 100644 index 0000000..fdaf846 --- /dev/null +++ b/.agent/skills/操作/确认规划候选/目录.md @@ -0,0 +1,3 @@ +| 名称 | 相对地址 | 内容描述 | 使用场景 | 使用要求 | +|------|----------|----------|----------|----------| +| 操作说明 | [SKILL.md](SKILL.md) | 规划候选的确切版本审阅与作者决定 | 使用新版规划入口 | 不以打开审阅或字段通过替代作者确认 | diff --git a/src/muse/作品规划/分层规划.py b/src/muse/作品规划/分层规划.py new file mode 100644 index 0000000..68549c5 --- /dev/null +++ b/src/muse/作品规划/分层规划.py @@ -0,0 +1,245 @@ +"""规划结构、目标章节及真实来源校验;动态内容不决定实例归属。""" + +from dataclasses import asdict +from uuid import NAMESPACE_URL, uuid5 + +from pydantic import TypeAdapter + +from muse.作品规划.存储 import 作品存储, 规划存储 +from muse.作品规划.模型 import 作品错误, 档案结构, 规划保存 +from muse.作品规划.目录调整 import 读取作者目录 +from muse.元数据.接口 import ( + 元数据服务, + 字段变更, + 投影值, + 投影字段, + 授权摘要, + 校验值, + 校验字段变更, + 选择实例维护者, +) +from muse.共享.错误 import Muse错误 +from muse.正式变更.接口 import 依赖引用, 固定哈希 + + +def 规划授权(作者: str, work_id: str): + return 授权摘要( + 固定哈希({"author": 作者}), 固定哈希({"work": work_id, "owner": "B01.planning"}) + ) + + +def 读取规划结构(连, 作者: str, work_id: str, 选择: 档案结构, 用途名: str = "userEditable"): + meta = 元数据服务(连) + schema = meta.读取有效结构( + 选择.schema_id, 选择.base_version, work_id=work_id, extension_version=选择.extension_version + ) + type_ = meta.读取类型(schema.基础.type_id) + family = "scene_plan" if "scene_plan" in type_.实例族 else "planning" + if 选择实例维护者(type_, family) != "B01": + raise 作品错误("SCOPE_DENIED", "该结构不属于本书规划") + projection = 投影字段( + schema, + meta.当前策略(type_.type_id), + 规划授权(作者, work_id), + 字段用途名=用途名, + 内容用途="planning", + 运行用途="production", + ) + return schema, type_, projection + + +def 规划身份(请求: 规划保存, 类型ID: str) -> str: + return str( + uuid5( + NAMESPACE_URL, + "muse:B01:plan:" + 固定哈希([请求.work_id, 请求.node_id or 请求.chapter_id, 类型ID]), + ) + ) + + +def 规划目标(连, 请求: 规划保存, 类型, directory: dict) -> dict | None: + chapter_id, kind = 请求.chapter_id, "chapter" if 请求.chapter_id else "work" + if 请求.node_id is not None: + tree = 作品存储(连).读取节点目录(请求.work_id) + node = next((n for n in tree["nodes"] if n["node_id"] == 请求.node_id), None) + if node is None: + raise 作品错误("SCOPE_DENIED", "规划节点不属于本书目录") + kind = node["kind"] + chapter_id = ( + node["node_id"] if kind == "chapter" else node["parent_id"] if kind == "scene" else None + ) + if 请求.chapter_id is not None and 请求.chapter_id != chapter_id: + raise 作品错误("SCOPE_DENIED", "规划节点与指定章节不一致") + grains = (类型.实例粒度,) if isinstance(类型.实例粒度, str) else 类型.实例粒度 + allowed = { + "scene" if grain == "block" and "scene_plan" in 类型.实例族 else grain for grain in grains + } + if kind not in allowed: + raise 作品错误("PLAN_SCOPE_INVALID", "规划结构不适用于该层目录节点") + chapter = next((c for c in directory["chapters"] if c["chapter_id"] == chapter_id), None) + if chapter_id is not None and chapter is None: + raise 作品错误("SCOPE_DENIED", "规划章节不属于该作品") + return chapter + + +def 核验规划( + 连, + 作者: str, + 请求: 规划保存, + *, + 检查提交: bool = True, + 规范内容: dict | None = None, + 祖先: frozenset[str] = frozenset(), + 检查来源: bool | None = None, + 引用解析=None, + 消费用途: str = "planning", +): + 来源须有效 = 检查提交 if 检查来源 is None else 检查来源 + directory = 读取作者目录(连, 作者, 请求.work_id) + schema, type_, projection = 读取规划结构(连, 作者, 请求.work_id, 请求.schema) + chapter = 规划目标(连, 请求, type_, directory) + id_ = 规划身份(请求, type_.type_id) + if id_ in 祖先: + raise 作品错误("PLAN_REFERENCE_INVALID", "规划依据不能形成当前版本依赖环") + current = 规划存储(连).读取(id_) + if 检查提交: + if (current["current_revision"] if current else 0) != 请求.expected_revision: + raise 作品错误("REVISION_CONFLICT", "规划正式版本已变化") + if schema.effective_schema_hash != 请求.schema.effective_schema_hash: + raise 作品错误("SCHEMA_STALE", "规划结构版本已变化") + if projection.projection_version != 请求.schema.projection_version: + raise 作品错误("PROJECTION_STALE", "规划字段用途已变化") + values = ( + 校验字段变更( + schema, + projection, + current["content"] if current else {}, + (*tuple(字段变更((k,), "set", v) for k, v in 请求.content.items()), *请求.changes), + ) + if 检查提交 + else 校验值(schema, 规范内容 if 规范内容 is not None else 请求.content) + ) + if ( + 检查提交 + and type_.type_id == "fine_outline" + and (chapter is None or values.值["targetChapter"] != chapter["position"]) + ): + raise 作品错误("PLAN_SCOPE_INVALID", "细纲展示章号必须对应固定目录中的目标章节") + deps = [依赖引用("B01.directory", "directory:" + 请求.work_id, str(directory["revision"]))] + if 请求.selection_revision is not None: + selected = 作品存储(连).读取探索选定(请求.work_id) + actual = selected["revision"] if selected else 0 + if 来源须有效 and actual != 请求.selection_revision: + raise 作品错误("SOURCE_STALE", "所选探索方向已改变") + deps.append( + 依赖引用( + "B01.exploration_selection", "exploration-selection:" + 请求.work_id, str(actual) + ) + ) + for ref in values.引用需求: + meta = 元数据服务(连) + 规划类型 = { + t + for t in ref.规则.目标类型 + if {"planning", "scene_plan"} & set(meta.读取类型(t).实例族) + } + target = ref.值.get("sourceId", ref.值.get("id")) if isinstance(ref.值, dict) else ref.值 + parent = 规划存储(连).读取(target) if 规划类型 and isinstance(target, str) else None + if parent is None and set(ref.规则.目标类型) != 规划类型: + if 引用解析 is None: + raise 作品错误("PRECONDITION_PENDING", "该规划引用还需要对应业务来源解析") + deps.extend(引用解析(连, 作者, 请求.work_id, ref, 必须有效=来源须有效)) + continue + if ( + parent is None + or parent["work_id"] != 请求.work_id + or parent["type_id"] not in 规划类型 + or parent["plan_id"] == id_ + ): + raise 作品错误("PLAN_REFERENCE_INVALID", "规划来源缺失、类型不符或不属于本书") + if ref.规则.关系用途 == "关联": + # 目录式关联核对身份与所属,不把下层计划反向变成上层生成依据。 + continue + if ( + 来源须有效 + and isinstance(ref.值, dict) + and "sourceVersion" in ref.值 + and str(parent["current_revision"]) != str(ref.值["sourceVersion"]) + ): + raise 作品错误("SOURCE_STALE", "规划引用不再是当前确认版本") + deps.extend( + 核对可用规划( + 连, + 作者, + parent, + 祖先 | {id_}, + 严格=来源须有效, + 引用解析=引用解析, + 消费用途=消费用途, + ) + ) + deps.append( + 依赖引用("B01.plan", "plan:" + parent["plan_id"], str(parent["current_revision"])) + ) + return id_, values.值, schema, type_, projection, tuple(sorted(set(deps))) + + +def 候选哈希(请求: 规划保存, 内容: dict, 来源: tuple) -> str: + return 固定哈希([asdict(请求), 内容, [asdict(s) for s in 来源]]) + + +def 核对可用规划( + 连, + 作者: str, + 记录: dict, + 祖先: frozenset[str] = frozenset(), + *, + 严格: bool = True, + 引用解析=None, + 消费用途: str = "planning", +) -> tuple: + id_ = 记录["plan_id"] + if id_ in 祖先: + raise 作品错误("SOURCE_STALE", "规划来源形成循环,需要重新确定依据") + if 固定哈希(记录["content"]) != 记录["content_hash"]: + raise 作品错误("INVALID_OUTPUT", "规划内容与保存哈希不一致") + req = TypeAdapter(规划保存).validate_python(记录["request"]) + _, _, _, _, _, sources = 核验规划( + 连, + 作者, + req, + 检查提交=False, + 规范内容=记录["content"], + 祖先=祖先, + 检查来源=严格, + 引用解析=引用解析, + 消费用途=消费用途, + ) + if 严格 and [asdict(s) for s in sources] != 记录["source_dependencies"]: + raise 作品错误("SOURCE_STALE", "规划所用的目录、选定方向或前置版本已变化") + binding = 元数据服务(连).读取绑定("plan:" + id_) + if 严格 and asdict(binding) != 记录["schema_binding"]: + raise 作品错误("SCHEMA_STALE", "规划的当前结构绑定已变化") + if 严格: + meta = 元数据服务(连) + schema = meta.读取有效结构( + req.schema.schema_id, + req.schema.base_version, + work_id=req.work_id, + extension_version=req.schema.extension_version, + ) + projection = 投影字段( + schema, + meta.当前策略(schema.基础.type_id), + 规划授权(作者, req.work_id), + 字段用途名="aiContext", + 内容用途=消费用途, + 运行用途="production", + ) + try: + 校验值(schema, 投影值(projection, 记录["content"])) + except Muse错误 as exc: + raise 作品错误( + "PRECONDITION_PENDING", "规划的必需字段已停止写作用途,需要重新准备完整前置" + ) from exc + return sources diff --git a/src/muse/作品规划/探索交接.py b/src/muse/作品规划/探索交接.py new file mode 100644 index 0000000..f0c6e5a --- /dev/null +++ b/src/muse/作品规划/探索交接.py @@ -0,0 +1,162 @@ +"""最终格式与作者选定的持久交接;不调度探索子阶段或判断创作内容。""" + +from __future__ import annotations + +import json +import re +from collections import Counter +from dataclasses import dataclass +from itertools import pairwise +from typing import TYPE_CHECKING + +from muse.作品规划.存储 import 作品存储 +from muse.作品规划.模型 import 作品错误, 探索选择 +from muse.作品规划.目录调整 import 读取作者目录 +from muse.正式变更.接口 import 依赖引用, 修改影响, 参与操作, 固定哈希, 审阅内容, 提交计划 +from muse.资源加载 import 打开资源 + +if TYPE_CHECKING: + from muse.作品规划.探索草稿 import 探索服务 + +_标题 = re.compile(r"^(#{1,6})[ \t]+(.+?)[ \t]*$", re.MULTILINE) +_条目 = re.compile(r"^\s*[-*]\s+(?:\*\*)?S(\d{3,})(?=[^\d]|$)", re.MULTILINE) +_占位 = re.compile( + r"^\s*(?:[-*>]\s*)?(?:TODO|TBD|PLACEHOLDER|\[待填\])\s*$", re.MULTILINE | re.IGNORECASE +) + + +def 校验交付格式(文本: str) -> dict: + 树 = json.loads(打开资源("规划/探索交付结构.json")) + 标题 = list(_标题.finditer(文本)) + 问题 = [] + + def 记录(code: str, message: str) -> None: + 问题.append({"code": code, "message": message}) + + if sum(len(h[1]) == 1 for h in 标题) != 1 or not 标题 or len(标题[0][1]) != 1: + 记录("CANDIDATE_STRUCTURE", "整理稿须以唯一的作品一级标题开始") + if any(len(h[1]) >= 4 for h in 标题): + 记录("CANDIDATE_STRUCTURE", "整理稿不使用四级及更深标题") + if _占位.search(文本): + 记录("CANDIDATE_PLACEHOLDER", "请将裸占位符改为具体的待定或未探索说明") + 章 = {c["heading"]: c for c in 树["chapters"]} + 二级 = [h for h in 标题 if len(h[1]) == 2] + 章名 = [h[2] for h in 二级] + if 章名 and 章名[0] == "目录": + 章名 = 章名[1:] + if 章名 != ["根设定", *章, "待决与取舍"]: + 记录("TREE_MISMATCH", "整理稿须按导航顺序包含根设定、九章和待决与取舍") + if any(len(h[1]) == 3 and (not 二级 or h.start() < 二级[0].start()) for h in 标题): + 记录("CANDIDATE_STRUCTURE", "展开标题须位于对应章节内") + for i, h in enumerate(二级): + end = 二级[i + 1].start() if i + 1 < len(二级) else len(文本) + 内容 = re.sub(r"", "", _标题.sub("", 文本[h.end() : end]), flags=re.DOTALL) + if not re.sub(r"[\s>*_`-]", "", 内容): + 记录("SECTION_EMPTY", f"“{h[2]}”需要已有内容或具体的未探索说明") + 可用 = [] + for 节 in 章.get(h[2], {}).get("sections", []): + 可用.extend([节["heading"], *节["units"]]) + 顺序 = {名: 位置 for 位置, 名 in enumerate(可用)} + 已展开 = [s[2] for s in 标题 if len(s[1]) == 3 and h.end() <= s.start() < end] + if any(s not in 顺序 for s in 已展开) or any( + 顺序[后] <= 顺序[前] for 前, 后 in pairwise(已展开) if 前 in 顺序 and 后 in 顺序 + ): + 记录("TREE_MISMATCH", f"“{h[2]}”的展开标题不属于导航或顺序有误") + 编号 = [int(v) for v in _条目.findall(文本)] + if any(数 > 1 for 数 in Counter(编号).values()): + 记录("SETTING_ID_DUPLICATE", "可选S编号须在同一稿件内唯一") + return { + "valid": not 问题, + "problems": 问题, + "setting_count": len(编号), + "contract_version": 树["contractVersion"], + } + + +@dataclass(frozen=True) +class 探索选定写入: + 作者: str + 请求: 探索选择 + 快照: dict + + +class 探索选定入口: + 身份 = "B01.select_exploration" + 请求类型 = 探索选择 + + def __init__(self, 草稿: 探索服务 | None): + self.草稿 = 草稿 + + def _取得稿(self, 身份, 请求: 探索选择) -> dict: + if self.草稿 is None: + raise 作品错误("PRECONDITION_PENDING", "探索交接需要明确的私人草稿目录") + 稿 = self.草稿.读取(身份, 请求.draft_id) + if 稿["draft_hash"] != 请求.expected_hash: + raise 作品错误("REVISION_CONFLICT", "探索稿已经变化,请重新比较要选定的版本") + if not 校验交付格式(稿["content"])["valid"]: + raise 作品错误("EXPLORATION_FORMAT_INVALID", "请先整理最终交付稿的格式") + return 稿 + + def 查看(self, 连, 身份, 命令): + 请求 = 命令.业务请求 + if 请求.target_ref != 命令.目标: + raise 作品错误("SCOPE_DENIED", "探索选定与作品目标不一致") + 读取作者目录(连, 身份.作者, 请求.work_id) + 稿 = self._取得稿(身份, 请求) + 存储 = 作品存储(连) + 当前 = 存储.读取探索选定(请求.work_id) + 作品 = 存储.读取(请求.work_id) + assert 作品 is not None + 版本 = 当前["revision"] if 当前 else 0 + return 审阅内容( + 请求.target_ref, + 版本, + None, + None, + None, + 固定哈希([当前["snapshot"] if 当前 else None, 稿]), + None, + None, + 请求.expected_hash, + ("exploration_selection",), + ( + 依赖引用("B01.work", "work:" + 请求.work_id, str(作品["current_revision"])), + 依赖引用("B01.exploration_selection", 请求.target_ref, str(版本)), + ), + ) + + def 准备(self, 连, 身份, 命令, 当前): + 请求 = 命令.业务请求 + return 提交计划( + ( + 参与操作( + "B01.write_exploration_selection", + 探索选定写入(身份.作者, 请求, self._取得稿(身份, 请求)), + ), + ), + ( + 修改影响( + 请求.target_ref, + "planning_direction_changed", + str(当前.数据版本), + str(当前.数据版本 + 1), + ), + ), + ) + + +class 探索选定参与者: + 身份 = "B01.write_exploration_selection" + 请求类型 = 探索选定写入 + + def 提交(self, 连, 操作): + return 作品存储(连).写探索选定(操作.作者, 操作.请求, 操作.快照) + + +class 探索选定依赖: + 种类 = "B01.exploration_selection" + + def 锁定当前(self, 连, 对象ID): + if not 对象ID.startswith("exploration-selection:"): + raise 作品错误("SCOPE_DENIED", "不是作品探索选定身份") + return str(作品存储(连).锁探索选定(对象ID.removeprefix("exploration-selection:"))) diff --git a/src/muse/作品规划/探索草稿.py b/src/muse/作品规划/探索草稿.py new file mode 100644 index 0000000..dbcf729 --- /dev/null +++ b/src/muse/作品规划/探索草稿.py @@ -0,0 +1,111 @@ +"""私人探索稿的原子保存与内容比较;不生成PG记录或正式上下文。""" + +import fcntl +import hashlib +import json +import os +import re +import tempfile +from pathlib import Path + +from muse.作品规划.探索交接 import 校验交付格式 +from muse.作品规划.模型 import 作品错误 + + +def _稿件哈希(title: str, content: str) -> str: + return hashlib.sha256(json.dumps([title, content], ensure_ascii=False).encode()).hexdigest() + + +class 探索服务: + def __init__(self, 根: Path): + self.根 = 根 + + def _目录(self, 身份) -> Path: + if not 身份.允许写正式内容: + raise 作品错误("SCOPE_DENIED", "私人探索需要已认证作者") + # 作者身份参与物理分区;调用方不传相对路径或作品目录名。 + return self.根 / hashlib.sha256(身份.作者.encode()).hexdigest() + + @staticmethod + def _核对ID(draft_id: str) -> None: + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,80}", draft_id): + raise 作品错误("EXPLORATION_INVALID", "探索稿需要稳定稿件身份") + + def _读取(self, 文件: Path, draft_id: str) -> dict: + try: + 稿 = json.loads(文件.read_text(encoding="utf-8")) + if 稿["draft_id"] != draft_id or 稿["draft_hash"] != _稿件哈希( + 稿["title"], 稿["content"] + ): + raise ValueError + return 稿 + except FileNotFoundError: + raise 作品错误("EXPLORATION_NOT_FOUND", "私人探索稿不存在") from None + except (OSError, ValueError, KeyError, TypeError): + raise 作品错误( + "EXPLORATION_UNREADABLE", "私人探索稿无法读回或内容与保存哈希不符" + ) from None + + def 读取(self, 身份, draft_id: str) -> dict: + self._核对ID(draft_id) + return self._读取(self._目录(身份) / (draft_id + ".json"), draft_id) + + def 列出(self, 身份) -> list[dict]: + 结果 = [] + for 文件 in sorted(self._目录(身份).glob("*.json")): + 稿 = self._读取(文件, 文件.stem) + 结果.append({key: 稿[key] for key in ("draft_id", "title", "revision", "draft_hash")}) + return 结果 + + def 保存( + self, 身份, draft_id: str, title: str, content: str, expected_hash: str | None + ) -> dict: + self._核对ID(draft_id) + if not isinstance(title, str) or not title.strip() or not isinstance(content, str): + raise 作品错误("EXPLORATION_INVALID", "探索稿需要名称和原始文本") + 目录 = self._目录(身份) + 目录.mkdir(mode=0o700, parents=True, exist_ok=True) + 文件 = 目录 / (draft_id + ".json") + 锁 = os.open(目录 / (draft_id + ".lock"), os.O_CREAT | os.O_RDWR, 0o600) + try: + fcntl.flock(锁, fcntl.LOCK_EX) + 旧 = self._读取(文件, draft_id) if 文件.exists() else None + 哈希 = _稿件哈希(title, content) + if 旧 is not None and 旧["draft_hash"] == 哈希: + return 旧 + if (旧["draft_hash"] if 旧 else None) != expected_hash: + raise 作品错误("REVISION_CONFLICT", "探索稿已经改变,请比较当前稿后再保存") + 稿 = { + "draft_id": draft_id, + "title": title, + "content": content, + "draft_hash": 哈希, + "revision": 旧["revision"] + 1 if 旧 else 1, + } + 临时 = None + try: + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=目录, prefix=".保存-", delete=False + ) as out: + 临时 = Path(out.name) + out.write(json.dumps(稿, ensure_ascii=False, indent=2) + "\n") + out.flush() + os.fsync(out.fileno()) + os.replace(临时, 文件) + 目录描述符 = os.open(目录, os.O_RDONLY) + try: + os.fsync(目录描述符) + finally: + os.close(目录描述符) + finally: + if 临时 is not None: + 临时.unlink(missing_ok=True) + return 稿 + finally: + os.close(锁) + + def 检查交付(self, 身份, draft_id: str, expected_hash: str) -> dict: + 稿 = self.读取(身份, draft_id) + if 稿["draft_hash"] != expected_hash: + raise 作品错误("REVISION_CONFLICT", "探索稿已改变,不能继续使用旧交付检查") + return {"draft_id": draft_id, "draft_hash": 稿["draft_hash"], **校验交付格式(稿["content"])} diff --git a/src/muse/作品规划/规划版本.py b/src/muse/作品规划/规划版本.py new file mode 100644 index 0000000..f8f4d9c --- /dev/null +++ b/src/muse/作品规划/规划版本.py @@ -0,0 +1,475 @@ +"""人工规划候选与明确作者决定;正式值由B01参与S01事务写入。""" + +from dataclasses import asdict, dataclass, replace +from uuid import NAMESPACE_URL, uuid5 + +from pydantic import TypeAdapter + +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 读候选(连, 作者, id_, 版本=None): + item = 规划存储(连).读取候选(id_, 版本) + if item is None: + raise 作品错误("PLAN_CANDIDATE_NOT_FOUND", "规划候选不存在") + 读取作者目录(连, 作者, item["work_id"]) + req = TypeAdapter(规划保存).validate_python(item["request"]) + deps = tuple(依赖引用(**r) for r in item["source_dependencies"]) + if 候选哈希(req, item["content"], deps) != item["candidate_hash"]: + raise 作品错误("INVALID_OUTPUT", "规划候选内容与哈希不一致") + return item, req + + +@dataclass(frozen=True) +class 候选保存: + candidate_id: str + 请求: 规划保存 | None = None + 前版: 规划定位 | None = None + 改动: tuple[字段变更, ...] = () + + +def 候选保存请求(连, 作者, 操作: 候选保存, 引用解析): + if 操作.前版 is None: + if 操作.请求 is None: + raise 作品错误("INVALID_REQUEST", "新规划候选需要完整请求") + return 操作.请求 + loc = 操作.前版 + item, req = 读候选(连, 作者, 操作.candidate_id) + if ( + loc.candidate_id, + loc.candidate_revision, + loc.candidate_hash, + loc.plan_id, + loc.expected_revision, + ) != ( + 操作.candidate_id, + item["revision"], + item["candidate_hash"], + item["plan_id"], + req.expected_revision, + ): + raise 作品错误("REVISION_CONFLICT", "待编辑的规划候选已改变") + _, original, schema, _, projection, sources = 核验规划(连, 作者, req, 引用解析=引用解析) + if 候选哈希(req, original, sources) != item["candidate_hash"]: + raise 作品错误("SOURCE_STALE", "候选来源已变化,请基于当前规划另拟候选") + values = 校验字段变更(schema, projection, item["content"], 操作.改动).值 + current = 规划存储(连).读取(loc.plan_id) + base = current["content"] if current else {} + # 保存相对正式基线的完整变更,避免后续确认丢掉未再次编辑的候选字段。 + changes = tuple( + 字段变更((key,), "set", values[key]) if key in values else 字段变更((key,), "remove") + for key in sorted(set(base) | set(values)) + if key not in base or key not in values or base[key] != values[key] + ) + return replace(req, content={}, changes=changes) + + +@dataclass(frozen=True) +class 规划写入: + 作者: str + 请求: 规划保存 + candidate_id: str | None = None + candidate_revision: int = 0 + + +class 规划候选入口: + 身份 = "B01.propose_plan" + 请求类型 = 候选保存 + + def __init__(self, 引用解析=None): + self.引用解析 = 引用解析 + + def 查看(self, 连, 身份, 命令): + input_ = 命令.业务请求 + req = 候选保存请求(连, 身份.作者, input_, self.引用解析) + id_, values, schema, _, projection, sources = 核验规划( + 连, 身份.作者, req, 引用解析=self.引用解析 + ) + current = 规划存储(连).读取候选(input_.candidate_id) + revision = current["current_revision"] if current else 0 + return 审阅内容( + input_.candidate_id, + revision, + None, + None, + None, + 候选哈希(req, values, sources), + schema.effective_schema_hash, + projection.projection_version, + 固定哈希([asdict(s) for s in sources]), + ("candidate",), + ( + *sources, + 依赖引用("B01.plan", "plan:" + id_, str(req.expected_revision)), + 依赖引用("B01.plan_candidate", input_.candidate_id, str(revision)), + ), + ) + + def 准备(self, 连, 身份, 命令, 当前): + input_ = 命令.业务请求 + req = 候选保存请求(连, 身份.作者, input_, self.引用解析) + return 提交计划( + ( + 参与操作( + "B01.write_plan_candidate", + 规划写入(身份.作者, req, input_.candidate_id, 当前.数据版本), + ), + ) + ) + + +class 规划候选参与者: + 身份 = "B01.write_plan_candidate" + 请求类型 = 规划写入 + + def __init__(self, 引用解析=None): + self.引用解析 = 引用解析 + + def 提交(self, 连, 操作): + id_, content, _, _, _, sources = 核验规划(连, 操作.作者, 操作.请求, 引用解析=self.引用解析) + return 规划存储(连).写候选( + 操作.candidate_id, + id_, + 操作.请求, + content, + sources, + 候选哈希(操作.请求, content, sources), + 操作.candidate_revision, + ) + + +class 规划决定入口: + 身份 = "B01.decide_plan" + 请求类型 = 规划定位 + + def __init__(self, 引用解析=None): + self.引用解析 = 引用解析 + + def 查看(self, 连, 身份, 命令): + loc = 命令.业务请求 + item, req = 读候选(连, 身份.作者, loc.candidate_id) + if (str(item["plan_id"]), item["revision"], item["candidate_hash"]) != ( + loc.plan_id, + loc.candidate_revision, + loc.candidate_hash, + ): + raise 作品错误("REVISION_CONFLICT", "规划候选与被审版本不一致") + id_, _, schema, _, projection, sources = 核验规划( + 连, 身份.作者, req, 检查提交=False, 规范内容=item["content"], 引用解析=self.引用解析 + ) + if id_ != item["plan_id"]: + raise 作品错误("SCOPE_DENIED", "规划候选的所属目标与请求范围不一致") + current = 规划存储(连).读取(id_) + revision = current["current_revision"] if current else 0 + return 审阅内容( + loc.target_ref, + revision, + loc.candidate_id, + item["revision"], + item["candidate_hash"], + 固定哈希([current["content"] if current else None, item["content"]]), + schema.effective_schema_hash, + projection.projection_version, + 固定哈希([asdict(s) for s in sources]), + ("plan",), + ( + *sources, + 依赖引用("B01.plan", loc.target_ref, str(revision)), + 依赖引用("B01.plan_candidate", loc.candidate_id, str(item["revision"])), + ), + ) + + def 准备(self, 连, 身份, 命令, 当前): + if 命令.动作 in {作者动作.拒绝, 作者动作.暂缓}: + return 提交计划(()) + if 命令.动作 is not 作者动作.采纳: + raise 作品错误("INVALID_REQUEST", "规划候选需要明确作者决定") + item, req = 读候选(连, 身份.作者, 命令.业务请求.candidate_id) + _, content, _, _, _, sources = 核验规划(连, 身份.作者, req, 引用解析=self.引用解析) + if 候选哈希(req, content, sources) != item["candidate_hash"]: + raise 作品错误("SOURCE_STALE", "规划候选的内容或必要来源已变化") + return 提交计划( + (参与操作("B01.write_plan", 规划写入(身份.作者, req)),), + (修改影响(当前.目标, "plan_changed", str(当前.数据版本), str(当前.数据版本 + 1)),), + ) + + +class 规划参与者: + 身份 = "B01.write_plan" + 请求类型 = 规划写入 + + def __init__(self, 引用解析=None): + self.引用解析 = 引用解析 + + def 提交(self, 连, 操作): + req = 操作.请求 + id_, content, schema, type_, _, sources = 核验规划( + 连, 操作.作者, req, 引用解析=self.引用解析 + ) + meta = 元数据服务(连) + target = "plan:" + id_ + binding = ( + meta.建立初始绑定(target, req.work_id, schema) + if 规划存储(连).读取(id_) is None + else meta.读取绑定(target) + ) + meta.保护提交( + target, + req.schema.effective_schema_hash, + req.schema.projection_version, + 规划授权(操作.作者, req.work_id), + 字段用途名="userEditable", + 内容用途="planning", + 运行用途="production", + ) + chapter = 规划目标(连, req, type_, 读取作者目录(连, 操作.作者, req.work_id)) + return 规划存储(连).写规划( + id_, + req, + type_.type_id, + content, + binding, + sources, + 操作.作者, + 固定哈希(content), + chapter["chapter_id"] if chapter else None, + ) + + +class 规划依赖: + 种类 = "B01.plan" + + def 锁定当前(self, 连, 对象ID): + if not 对象ID.startswith("plan:"): + raise 作品错误("SCOPE_DENIED", "不是规划身份") + return str(规划存储(连).锁规划(对象ID.removeprefix("plan:"))) + + +class 规划候选依赖: + 种类 = "B01.plan_candidate" + + def 锁定当前(self, 连, 对象ID): + return str(规划存储(连).锁候选(对象ID)) + + +def 登记规划参与者(目录, 引用解析=None): + 目录.登记入口(规划候选入口(引用解析)) + 目录.登记入口(规划决定入口(引用解析)) + 目录.登记参与者(规划候选参与者(引用解析)) + 目录.登记参与者(规划参与者(引用解析)) + 目录.登记依赖(规划依赖()) + 目录.登记依赖(规划候选依赖()) + + +class 规划服务: + def __init__(self, 数据库, 正式, 引用解析=None): + self.引用解析 = 引用解析 + self.数据库, self.正式 = 数据库, 正式 + + def 读取表单( + self, + 身份, + work_id: str, + schema_id: str, + 版本: int, + *, + chapter_id: str | None = None, + node_id: str | None = None, + ): + 核对作者(身份) + with self.数据库.连接(只读=True) as 连: + directory = 读取作者目录(连, 身份.作者, work_id) + choice = 档案结构(schema_id, 版本, "", "") + _, type_, _ = 读取规划结构(连, 身份.作者, work_id, choice) + target = 规划保存(work_id, 0, choice, {}, chapter_id=chapter_id, node_id=node_id) + chapter = 规划目标(连, target, type_, directory) + id_ = 规划身份(target, type_.type_id) + current = 规划存储(连).读取(id_) + if current: + saved = current["schema_binding"] + choice = 档案结构( + saved["schema_id"], + saved["base_version"], + saved["effective_schema_hash"], + "", + saved["extension_version"], + ) + schema, _, editable = 读取规划结构(连, 身份.作者, work_id, choice) + _, _, visible = 读取规划结构(连, 身份.作者, work_id, choice, "uiVisible") + return { + "work_id": work_id, + "chapter_id": chapter["chapter_id"] if chapter else None, + "node_id": node_id or chapter_id, + "plan_id": id_, + "revision": current["revision"] if current else 0, + "schema_binding": asdict( + 档案结构( + choice.schema_id, + choice.base_version, + schema.effective_schema_hash, + editable.projection_version, + choice.extension_version, + ) + ), + "content": 投影值(visible, current["content"] if current else {}), + "visible_schema": 生成结构合同(visible), + "editable_schema": 生成结构合同(editable), + } + + def 列出候选(self, 身份, work_id: str): + 核对作者(身份) + with self.数据库.连接(只读=True) as 连: + 读取作者目录(连, 身份.作者, work_id) + ids = 规划存储(连).列出候选(work_id) + return [self.读取候选(身份, id_) for id_ in ids] + + def 新规划结构(self, 身份, work_id, schema_id, 版本, *, 扩展版本=None): + 核对作者(身份) + with self.数据库.连接(只读=True) as 连: + 读取作者目录(连, 身份.作者, work_id) + schema, _, projection = 读取规划结构( + 连, 身份.作者, work_id, 档案结构(schema_id, 版本, "", "", 扩展版本) + ) + return 档案结构( + schema_id, + 版本, + schema.effective_schema_hash, + projection.projection_version, + 扩展版本, + ) + + def 保存人工候选(self, 身份, 命令ID, 请求): + id_ = str(uuid5(NAMESPACE_URL, "muse:B01:plan-candidate:" + 固定哈希([身份.作者, 命令ID]))) + return self.正式.提交变更( + 身份, + 变更命令(命令ID, "B01.propose_plan", id_, 作者动作.人工保存, 0, 候选保存(id_, 请求)), + ) + + def 编辑候选(self, 身份, 命令ID, 定位: 规划定位, 改动: tuple[字段变更, ...]): + return self.正式.提交变更( + 身份, + 变更命令( + 命令ID, + "B01.propose_plan", + 定位.candidate_id, + 作者动作.人工保存, + 定位.candidate_revision, + 候选保存(定位.candidate_id, 前版=定位, 改动=改动), + ), + ) + + def 读取候选(self, 身份, candidate_id, *, 版本: int | None = None): + 核对作者(身份) + with self.数据库.连接(只读=True) as 连: + item, req = 读候选(连, 身份.作者, candidate_id, 版本) + _, _, projection = 读取规划结构(连, 身份.作者, req.work_id, req.schema, "uiVisible") + _, _, editable = 读取规划结构(连, 身份.作者, req.work_id, req.schema) + return { + "candidate_id": candidate_id, + "plan_id": item["plan_id"], + "revision": item["revision"], + "candidate_hash": item["candidate_hash"], + "work_id": req.work_id, + "chapter_id": req.chapter_id, + "node_id": req.node_id, + "expected_revision": req.expected_revision, + "schema_binding": asdict(req.schema), + "content": 投影值(projection, item["content"]), + "decision": 读取作者决定(连, candidate_id, item["revision"]), + "visible_schema": 生成结构合同(projection), + "editable_schema": 生成结构合同(editable), + } + + def 打开审阅(self, 身份, 定位): + review = self.正式.打开审阅( + 身份, + 变更命令( + "review", + "B01.decide_plan", + 定位.target_ref, + 作者动作.采纳, + 定位.expected_revision, + 定位, + ), + ) + base = review["snapshot"]["数据版本"] + return { + **review, + "candidate": self.读取候选( + 身份, 定位.candidate_id, 版本=review["snapshot"]["候选版本"] + ), + "before": self.读取规划(身份, 定位.plan_id, 版本=base) if base else None, + } + + def 决定(self, 身份, 命令ID, 定位, 动作, 审阅ID, 审阅哈希, 批准): + return self.正式.提交变更( + 身份, + 变更命令( + 命令ID, + "B01.decide_plan", + 定位.target_ref, + 动作, + 定位.expected_revision, + 定位, + 审阅ID, + 审阅哈希, + 批准, + ), + ) + + def 读取规划(self, 身份, plan_id, *, 版本: int | None = None): + 核对作者(身份) + with self.数据库.连接(只读=True) as 连: + item = 规划存储(连).读取(plan_id, 版本) + if item is None: + raise 作品错误("PLAN_NOT_FOUND", "正式规划不存在") + 读取作者目录(连, 身份.作者, item["work_id"]) + return self._呈现规划(连, 身份, item) + + def _呈现规划(self, 连, 身份, item): + content = 解析规划载荷(item["content"]) + if 固定哈希(content) != item["content_hash"]: + raise 作品错误("INVALID_OUTPUT", "规划内容与保存哈希不一致") + req = TypeAdapter(规划保存).validate_python(item["request"]) + _, _, projection = 读取规划结构(连, 身份.作者, req.work_id, req.schema, "uiVisible") + return { + "plan_id": item["plan_id"], + "work_id": item["work_id"], + "chapter_id": item["chapter_id"], + "node_id": item["node_id"], + "type_id": item["type_id"], + "revision": item["revision"], + "content": 投影值(projection, content), + "schema_binding": item["schema_binding"], + "source_dependencies": item["source_dependencies"], + } + + def 读取可用细纲(self, 身份, work_id: str, chapter_id: str): + 核对作者(身份) + with self.数据库.连接(只读=True) as 连: + item = 读取已确认细纲(连, 身份.作者, work_id, chapter_id, self.引用解析) + return self._呈现规划(连, 身份, item) + + def 列出规划(self, 身份, work_id): + 核对作者(身份) + with self.数据库.连接(只读=True) as 连: + 读取作者目录(连, 身份.作者, work_id) + ids = 规划存储(连).列出(work_id) + return [self.读取规划(身份, id_) for id_ in ids] diff --git a/src/muse/作品规划/近期细化.py b/src/muse/作品规划/近期细化.py new file mode 100644 index 0000000..ad1e141 --- /dev/null +++ b/src/muse/作品规划/近期细化.py @@ -0,0 +1,17 @@ +"""近期细纲只复用当前有效的已确认规划;新细化由持久任务执行。""" + +from muse.作品规划.分层规划 import 核对可用规划 +from muse.作品规划.存储 import 规划存储 +from muse.作品规划.模型 import 作品错误 +from muse.作品规划.目录调整 import 读取作者目录 + + +def 读取已确认细纲(连, 作者: str, work_id: str, chapter_id: str, 引用解析=None) -> dict: + 目录 = 读取作者目录(连, 作者, work_id) + if chapter_id not in {c["chapter_id"] for c in 目录["chapters"]}: + raise 作品错误("SCOPE_DENIED", "目标章节不属于本书") + item = 规划存储(连).查章规划(work_id, chapter_id, "fine_outline") + if item is None: + raise 作品错误("PRECONDITION_PENDING", "目标章节还没有已确认细纲") + 核对可用规划(连, 作者, item, 引用解析=引用解析, 消费用途="generation") + return item diff --git a/tests/单元/test_探索交接与规划.py b/tests/单元/test_探索交接与规划.py new file mode 100644 index 0000000..786eefa --- /dev/null +++ b/tests/单元/test_探索交接与规划.py @@ -0,0 +1,516 @@ +"""私人探索与正式交接分开;局部草稿不受最终目录约束。""" + +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from muse.作品规划.接口 import 作品错误, 探索服务, 校验交付格式 +from muse.共享.调用身份 import 内容用途, 用途, 调用身份 + + +def 作者(id_="author"): + return 调用身份(id_, None, 用途.生产, 内容用途.规划) + + +def test_私人稿保存原话且并发冲突不覆盖__ad1001(tmp_path): + 服务 = 探索服务(tmp_path / "私人探索") + 首次 = 服务.保存(作者(), "umbrella", "雨伞", "哥哥没回来。\n妹妹望着门口的伞。", None) + assert 首次["revision"] == 1 + assert 服务.读取(作者(), "umbrella")["content"] == "哥哥没回来。\n妹妹望着门口的伞。" + assert 服务.列出(作者("second")) == [] + with pytest.raises(作品错误): + 服务.读取(作者("second"), "umbrella") + + def 修改(正文): + try: + return 服务.保存(作者(), "umbrella", "雨伞", 正文, 首次["draft_hash"]) + except 作品错误 as exc: + return exc.错误码 + + with ThreadPoolExecutor(2) as pool: + 结果 = list(pool.map(修改, ["她把伞撑开。", "她把伞收起。"])) + assert sum(isinstance(x, dict) for x in 结果) == 1 + assert "REVISION_CONFLICT" in 结果 + 当前 = 服务.读取(作者(), "umbrella") + assert 当前["revision"] == 2 + # 同一内容的未知返回可以重试,不额外产生版本。 + assert 服务.保存(作者(), "umbrella", 当前["title"], 当前["content"], 首次["draft_hash"]) == 当前 + + +def test_探索不先查最终格式且检查不产生正式内容__ad1002(tmp_path): + 服务 = 探索服务(tmp_path / "私人探索") + 稿 = 服务.保存(作者(), "first-fragment", "只有一个动作", "有人敲门。", None) + 检查 = 服务.检查交付(作者(), "first-fragment", 稿["draft_hash"]) + assert not 检查["valid"] and 检查["problems"] + assert 服务.读取(作者(), "first-fragment") == 稿 + assert [v["draft_id"] for v in 服务.列出(作者())] == ["first-fragment"] + + +def test_局部展开的最终稿不设编号和篇幅门__61c0a6(): + assert 校验交付格式(原探索稿)["valid"] + + +def test_不同探索稿可展开不同子集__8eed6e(): + 展开 = 原探索稿.replace( + "待定:先从人物行动中寻找中心。", + "### 1.1 作品形态与定位\n\n### 1.1.1 题材、载体与目标读者\n\n暂定现实题材。", + ) + assert all(校验交付格式(s)["valid"] for s in (原探索稿, 展开)) + + +@pytest.mark.parametrize("场景", ["unknown", "missing", "reordered"]) +def test_最终导航标题未知缺失或倒序拒绝__c6ad42(场景): + 输入 = { + "unknown": 原探索稿.replace("## 三、角色与关系", "## 三、人物"), + "missing": 原探索稿.replace("## 三、角色与关系\n", ""), + "reordered": 原探索稿.replace( + "待定:先从人物行动中寻找中心。", + "### 1.2 故事承诺与一句话梗概\n\n设想。\n\n### 1.1 作品形态与定位\n\n设想。", + ), + }[场景] + assert "TREE_MISMATCH" in {v["code"] for v in 校验交付格式(输入)["problems"]} + + +def test_空节和裸占位拒绝而明确未知允许__64d9e4(): + for 原, 替换, 错误码 in [ + ("未探索:尚未讨论社会制度。", "", "SECTION_EMPTY"), + ("待定:哥哥为何留着伞。", "TODO", "CANDIDATE_PLACEHOLDER"), + ]: + assert 错误码 in {v["code"] for v in 校验交付格式(原探索稿.replace(原, 替换))["problems"]} + assert 校验交付格式(原探索稿)["valid"] + + +def test_可选编号无冻结号段但必须唯一__ec3fc3(): + 编号稿 = 原探索稿.replace("哥哥每天", "- **S090|哥哥:** 哥哥每天").replace( + "先从动作里表现关系。", "- **S003|表达:** 先从动作里表现关系。" + ) + assert 校验交付格式(编号稿)["valid"] + assert "SETTING_ID_DUPLICATE" in { + v["code"] for v in 校验交付格式(编号稿.replace("S003", "S090"))["problems"] + } + + +@pytest.mark.parametrize("场景", ["no-title", "deep-heading"]) +def test_最终稿缺标题或标题过深拒绝__1d35b4(场景): + 内容 = ( + 原探索稿.replace("# 雨伞|作品探索\n", "") + if 场景 == "no-title" + else 原探索稿 + "\n#### 额外层级\n说明。" + ) + assert "CANDIDATE_STRUCTURE" in {v["code"] for v in 校验交付格式(内容)["problems"]} + + +def test_缺失私人稿不会得到交付成功__3a6d20(tmp_path): + with pytest.raises(作品错误) as 错: + 探索服务(tmp_path / "私人探索").读取(作者(), "missing") + assert 错.value.错误码 == "EXPLORATION_NOT_FOUND" + + +def test_HTTP与CLI保存探索不访问数据库__ad1003(tmp_path): + import json + import subprocess + import sys + + from fastapi.testclient import TestClient + + from muse.接入.http.应用 import 创建应用 + from muse.配置 import 读取配置 + + secret = tmp_path / "口令.txt" + secret.write_text("synthetic-exploration-password") + secret.chmod(0o600) + cfg = tmp_path / "作者.toml" + cfg.write_text( + '["数据库"]\n"取值方式"="受控存储"\n"位置"="missing-database.txt"\n["资源"]\n"发布身份"="synthetic"\n[HTTP]\n"口令文件"="口令.txt"\n"作者ID"="author"\n"公开地址"="http://testserver"\n"允许来源"=["http://testserver"]\n["文件"]\n"探索草稿"="私人探索"\n' + ) + app = 创建应用(读取配置(cfg)) + with TestClient(app, headers={"origin": "http://testserver"}) as client: + assert client.get("/api/v1/explorations").status_code == 401 + assert ( + client.post( + "/api/v1/session", json={"password": "synthetic-exploration-password"} + ).status_code + == 200 + ) + req = { + "draft_id": "http-draft", + "title": "原话", + "content": "只是一个念头。", + "expected_hash": None, + } + assert ( + client.post( + "/api/v1/explorations/http-draft", json={**req, "author_id": "forged"} + ).status_code + == 422 + ) + saved = client.post("/api/v1/explorations/http-draft", json=req) + assert saved.status_code == 200, saved.text + assert client.get("/api/v1/explorations").json()[0]["title"] == "原话" + checked = client.post( + "/api/v1/explorations/http-draft/check", + json={"expected_hash": saved.json()["draft_hash"]}, + ) + assert checked.status_code == 200 and not checked.json()["valid"] + result = subprocess.run( + [sys.executable, "-m", "muse", "探索", str(cfg), "查看", "http-draft"], + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout) == saved.json() + invalid = subprocess.run( + [ + sys.executable, + "-m", + "muse", + "探索", + str(cfg), + "检查", + "http-draft", + "--预期哈希", + saved.json()["draft_hash"], + ], + capture_output=True, + text=True, + ) + assert invalid.returncode == 1 and not json.loads(invalid.stdout)["valid"] + + +# 保留LC原始探索输入,不借旧测试的导入和运行顺序。 +原探索稿 = """# 雨伞|作品探索 + +## 根设定 +作者希望保留雨天的兄妹关系。 + +## 一、作品定位与核心命题 +待定:先从人物行动中寻找中心。 + +## 二、故事世界与社会系统 +未探索:尚未讨论社会制度。 + +## 三、角色与关系 +哥哥每天擦一把从不用的伞,妹妹故意把它拿走。 + +## 四、故事机制与关键实体 +未探索:尚未决定是否有超自然力量。 + +## 五、阶段与状态变化 +未探索:只有当前这一幕。 + +## 六、情节架构与全书大纲 +未探索:兄妹争执后的事情还没想好。 + +## 七、主题与读者体验弧线 +暂想保留亲近中带着疑问的感觉。 + +## 八、信息、悬念与伏笔管理 +待定:哥哥为何留着伞。 + +## 九、叙事执行与一致性 +先从动作里表现关系。 + +## 待决与取舍 +保留争执这一幕,暂缓决定雨伞的来历。 +""" + + +@pytest.mark.parametrize( + "场景,输入", + [ + ("missing", None), + ("unconfirmed", False), + ("string-false", "false"), + ("string-true", "true"), + ("user-string", "user"), + ("integer-true", 1), + ("list-true", ["true"]), + ("empty-id", ""), + ("list-id", ["x"]), + ("blank-time", " "), + ], + ids=[ + "missing", + "unconfirmed", + "string-false", + "string-true", + "user-string", + "integer-true", + "list-true", + "empty-id", + "list-id", + "blank-time", + ], +) +def test_选定输入不能自报确认时间或缺失稿件__ad1004(场景, 输入): + from pydantic import ValidationError + + from muse.接入.创作操作 import 探索选定请求 + + selection = { + "work_id": "work", + "draft_id": "x", + "expected_hash": "source-hash", + "expected_revision": 0, + } + if 场景 == "missing": + selection = {} + elif 场景 in {"empty-id", "list-id"}: + selection["draft_id"] = 输入 + elif 场景 == "blank-time": + selection["confirmedAt"] = 输入 + else: + selection["confirmedByUser"] = 输入 + with pytest.raises((ValidationError, 作品错误)): + 探索选定请求.model_validate({"command_id": "select-direction", "selection": selection}) + + +# --------------------------------------------------------------------------- +# W13 承接:作者旅程、细纲合同与失败边界(原 AuthorInstruction/FineOutlineContract) +# --------------------------------------------------------------------------- + +作者根 = Path(__file__).resolve().parents[2] / ".agent" / "作者" +仓库根 = Path(__file__).resolve().parents[2] + + +def test_自然语言旅程到达作者决定__aab5b6(): + """路由表命中六大场景;最小旅程按序推进,最终决定交给新版候选决定技能。""" + import re + + 入口 = (作者根 / "指令.md").read_text(encoding="utf-8") + 路由 = re.search( + r"^## 自然语言路由\s*$\n(?P
.*?)(?=^## )", + 入口, + flags=re.MULTILINE | re.DOTALL, + ).group("body") + for 说法, 场景 in ( + ("我只有一个点子", "定故事"), + ("帮我排一下整本书", "排故事"), + ("这个人物还站不住", "塑人物"), + ("接着写下一章", "写下一章"), + ("把这段改顺", "改正文"), + ("看看这部作品哪里有问题", "查作品"), + ): + assert re.search(rf"\|\s*{re.escape(说法)}\s*\|\s*\[{re.escape(场景)}\]", 路由) + + 旅程 = re.search( + r"^## 最小作者旅程\s*$\n(?P.*?)(?=^## |\Z)", + 入口, + flags=re.MULTILINE | re.DOTALL, + ).group("body") + 顺序 = re.findall(r"\[([^\]]+)\]\(场景/[^)]+\)", 旅程) + assert 顺序 == ["定故事", "排故事", "写下一章", "查作品", "改正文"] + 决定 = re.search(r"\[决定正文候选去留\]\(([^)]+)\)", 旅程) + assert 决定 and "skills/操作/决定正文候选去留/SKILL.md" in 决定.group(1) + for 动作 in ("采用", "丢弃", "修改"): + assert f"`{动作}`" in 旅程 + + +def _细纲结构(): + from muse.元数据.接口 import 合成结构, 读取内置结构 + + return 合成结构(读取内置结构("fine_outline")) + + +def _完整细纲值(): + return { + "targetChapter": 3, + "sourceRef": {"sourceId": "outline:v1", "sourceVersion": 1}, + "chapterGoal": "本章戏剧任务", + "hardConstraints": ["衔接上一章结尾"], + "keyEvents": [ + { + "id": "e1", + "order": 1, + "event": "事件", + "participants": ["林深"], + "trigger": "触发", + "resultDirection": "结果方向", + } + ], + "adjustableBeats": ["次序可调整"], + "declaredNewFacts": [ + { + "factId": "f1", + "text": "新事实", + "sourceRef": {"sourceId": "outline:v1", "sourceVersion": 1}, + } + ], + "mustAppearEntities": ["林深"], + "chapterEndHook": "章末钩子", + } + + +def test_细纲技能引用结构为字段权威__aaae60(): + """字段权威是S04结构:合同由结构生成并带字段身份;操作技能不再自持字段清单。""" + from muse.元数据.接口 import 生成结构合同 + + 结构 = _细纲结构() + 合同 = 生成结构合同(结构) + assert 合同["required"], "细纲合同必须声明必填字段" + for 字段 in ("targetChapter", "keyEvents", "chapterEndHook"): + 属性 = 合同["properties"][字段] + assert 属性["x-field-id"].startswith("fine_outline."), 字段 + assert 属性["title"], f"{字段} 缺中文名" + 技能 = (仓库根 / ".agent" / "skills" / "操作" / "确认规划候选" / "SKILL.md").read_text( + encoding="utf-8" + ) + for 字段名 in ("targetChapter", "chapterGoal", "hardConstraints"): + assert 字段名 not in 技能, "操作技能不得自持细纲字段清单,字段权威在结构" + + +@pytest.mark.parametrize( + "缺失字段", + [ + "targetChapter", + "sourceRef", + "chapterGoal", + "hardConstraints", + "keyEvents", + "adjustableBeats", + "declaredNewFacts", + "mustAppearEntities", + "chapterEndHook", + ], +) +def test_细纲必填字段齐全且缺一拒绝__759215(缺失字段): + from muse.元数据.接口 import 校验值, 生成结构合同 + from muse.元数据.模型 import 元数据错误 + + 结构 = _细纲结构() + 合同 = 生成结构合同(结构) + assert set(合同["required"]) == { + "targetChapter", + "sourceRef", + "chapterGoal", + "hardConstraints", + "keyEvents", + "adjustableBeats", + "declaredNewFacts", + "mustAppearEntities", + "chapterEndHook", + } + 值 = {k: v for k, v in _完整细纲值().items() if k != 缺失字段} + with pytest.raises(元数据错误) as 错: + 校验值(结构, 值) + assert 错.value.错误码 == "FIELD_INVALID" + assert 缺失字段 in 错.value.呈现()["details"]["字段"] + + +def test_细纲推荐字段缺失不拦类型错误拒__0766da(): + from muse.元数据.接口 import 校验值 + from muse.元数据.模型 import 元数据错误 + + 结构 = _细纲结构() + 结果 = 校验值(结构, _完整细纲值()) + assert 结果.值["targetChapter"] == 3 # 推荐字段全部缺失仍通过 + with pytest.raises(元数据错误) as 错: + 校验值(结构, {**_完整细纲值(), "unknowns": "不是列表"}) + assert 错.value.错误码 == "FIELD_INVALID" + + +def test_阴影边界与红线明确__a5ee39(): + """候选保持Shadow;正文与规划决定入口分离;决定必须引用被审版本。""" + 决定正文 = (仓库根 / ".agent" / "skills" / "操作" / "决定正文候选去留" / "SKILL.md").read_text( + encoding="utf-8" + ) + 确认规划 = (仓库根 / ".agent" / "skills" / "操作" / "确认规划候选" / "SKILL.md").read_text( + encoding="utf-8" + ) + assert "不能自行填补或改用最新版本" in 决定正文 + assert "必须重新展示再决定" in 决定正文 + assert "不代替正文或故事事实的独立确认" in 确认规划 + assert "不自动刷新成最新版本后重试采纳" in 确认规划 + # 作者层红线:候选未经作者明确确认不进入正式内容。 + 入口 = (作者根 / "指令.md").read_text(encoding="utf-8") + assert "不进入 Canonical" in 入口 and "主会话不代替角色" in 入口 + + +def test_冻结与来源两行规则明确__9bb829(): + """审阅绑定确切版本与哈希;检索呈现内容与来源版本两份信息。""" + from muse.元数据.接口 import 投影字段, 授权摘要, 策略快照 + + 确认规划 = (仓库根 / ".agent" / "skills" / "操作" / "确认规划候选" / "SKILL.md").read_text( + encoding="utf-8" + ) + for 锚 in ("candidate_hash", "review_hash", "expected_revision"): + assert 锚 in 确认规划, f"确认规划缺少冻结锚点 {锚}" + 结构 = _细纲结构() + 投影 = 投影字段( + 结构, + 策略快照("fine_outline", 1), + 授权摘要("a1", "s1"), + 字段用途名="aiContext", + 内容用途="generation", + 运行用途="production", + ) + # 检索规则:细纲进入模型上下文的字段带来源引用,正文不在细纲内自造。 + 键 = {f.key for f in 投影.允许字段} + assert {"targetChapter", "chapterGoal", "keyEvents", "sourceRef"} <= 键 + assert not any("正文" in k or k == "content" for k in 键), "细纲检索不得携带正文字段" + + +def test_CLI失败可识别且不转为成功__1f5c8c(tmp_path): + """版本冲突与非法输入经CLI返回非零与稳定错误码,不静默成功。""" + import json + import subprocess + import sys + + cfg = tmp_path / "作者.toml" + cfg.write_text( + '["数据库"]\n"取值方式"="受控存储"\n"位置"="missing-database.txt"\n' + '["资源"]\n"发布身份"="synthetic"\n[HTTP]\n"口令文件"="口令.txt"\n"作者ID"="author"\n' + '"公开地址"="http://localhost"\n"允许来源"=["http://localhost"]\n' + '["文件"]\n"探索草稿"="私人探索"\n' + ) + (tmp_path / "口令.txt").write_text("synthetic") + 请求 = tmp_path / "保存.json" + 首次 = { + "draft_id": "cli-draft", + "title": "伞", + "content": "第一稿。", + "expected_hash": None, + } + 请求.write_text(json.dumps(首次, ensure_ascii=False)) + 保存 = subprocess.run( + [sys.executable, "-m", "muse", "探索", str(cfg), "保存", str(请求)], + capture_output=True, + text=True, + ) + assert 保存.returncode == 0, 保存.stderr + 冲突 = {**首次, "content": "第二稿。", "expected_hash": "stale-hash"} + 请求.write_text(json.dumps(冲突, ensure_ascii=False)) + 失败 = subprocess.run( + [sys.executable, "-m", "muse", "探索", str(cfg), "保存", str(请求)], + capture_output=True, + text=True, + ) + assert 失败.returncode == 1, (失败.stdout, 失败.stderr) + 呈现 = json.loads(失败.stderr) + assert 呈现["code"] == "REVISION_CONFLICT" + 缺稿 = subprocess.run( + [sys.executable, "-m", "muse", "探索", str(cfg), "查看", "missing-draft"], + capture_output=True, + text=True, + ) + assert 缺稿.returncode == 1 + assert json.loads(缺稿.stderr)["code"] == "EXPLORATION_NOT_FOUND" + + +def test_细纲合同绑定既有规划槽位__d55c2d(): + """细纲归属B01规划族并由规划员角色承担;不再靠旧技能文档字符串绑定。""" + import json as _json + + from muse.元数据.接口 import 读取内置种子, 选择实例维护者 + + 类型 = next(s.类型 for s in 读取内置种子() if s.类型.type_id == "fine_outline") + assert "planning" in 类型.实例族 + assert 选择实例维护者(类型, "planning") == "B01" + 登记 = _json.loads((仓库根 / "工具" / "资源登记.json").read_text(encoding="utf-8")) + 角色文件 = 仓库根 / ".agent" / "角色" / "规划员.md" + assert ".agent/角色/规划员.md" in 登记["能力文件"] + 头部 = 角色文件.read_text(encoding="utf-8").split("---")[1] + assert "id: planner" in 头部 + 正文 = 角色文件.read_text(encoding="utf-8") + assert "规划候选" in 正文 and "不替正文" in 正文 diff --git a/tests/集成/test_探索选定与规划.py b/tests/集成/test_探索选定与规划.py new file mode 100644 index 0000000..254aaed --- /dev/null +++ b/tests/集成/test_探索选定与规划.py @@ -0,0 +1,176 @@ +"""探索方向的真实作者动作与PG回执;选择方向不确认正式规划。""" + +import json +from dataclasses import replace + +import pytest + +from muse.作品规划.接口 import 作品错误, 探索选择, 档案保存 +from muse.元数据.接口 import 元数据服务, 导入内置结构 +from muse.共享.调用身份 import 内容用途, 用途, 调用身份 +from muse.启动 import 构建 +from muse.资源加载 import 打开资源 +from muse.配置 import 应用配置 + +pytestmark = pytest.mark.数据库 + + +@pytest.fixture +def 交接环境(应用测试库, tmp_path): + with 应用测试库[用途.维护].连接() as 连, 连.transaction(): + 导入内置结构(元数据服务(连)) + app = 构建( + 应用配置(应用测试库[用途.生产].引用, "synthetic", 探索草稿=str(tmp_path / "私人探索")) + ) + actor = 调用身份("author", None, 用途.生产, 内容用途.规划) + profile = app.要求作品().新档案结构(actor, "selection-work", "work_core", 1) + app.要求作品().保存档案( + actor, "create-work", 档案保存("selection-work", 0, {"名称": "合成作品"}, profile) + ) + tree = json.loads(打开资源("规划/探索交付结构.json")) + draft = "# 浮岛|作品探索\n\n" + "\n\n".join( + "## " + h + "\n未探索:先保留空中城镇停电的画面。" + for h in ["根设定", *[c["heading"] for c in tree["chapters"]], "待决与取舍"] + ) + saved = app.要求探索().保存(actor, "island", "浮岛", draft, None) + return app, actor, saved, 应用测试库 + + +def test_选定稿冻结来源且重复请求回原回执__ad1201(交接环境): + app, actor, draft, _ = 交接环境 + works = app.要求作品() + assert works.读取探索选定(actor, "selection-work")["revision"] == 0 + req = 探索选择("selection-work", "island", draft["draft_hash"], 0) + receipt = works.选定探索稿(actor, "select-direction", req) + chosen = works.读取探索选定(actor, "selection-work") + assert chosen["revision"] == 1 and chosen["selection"]["snapshot"] == draft + assert chosen["selection"]["selected_by"] == "author" + app.要求探索().保存( + actor, "island", "浮岛", draft["content"] + "\n新的分歧,暂不选。", draft["draft_hash"] + ) + assert works.读取探索选定(actor, "selection-work") == chosen + assert works.选定探索稿(actor, "select-direction", req) == receipt + with pytest.raises(作品错误) as 错: + works.选定探索稿(actor, "select-stale", replace(req, expected_revision=1)) + assert 错.value.错误码 == "REVISION_CONFLICT" + + +def test_选定失败完整回滚且私人稿保留__ad1202(交接环境): + import psycopg + + app, actor, draft, db = 交接环境 + with db[用途.维护].连接() as 连, 连.transaction(): + 连.execute( + "CREATE FUNCTION reject_selection_receipt() RETURNS trigger LANGUAGE plpgsql AS $$ " + "BEGIN RAISE EXCEPTION 'synthetic receipt failure'; END $$" + ) + 连.execute( + "CREATE TRIGGER reject_selection BEFORE UPDATE OF receipt ON muse_change_command " + "FOR EACH ROW EXECUTE FUNCTION reject_selection_receipt()" + ) + request = 探索选择("selection-work", "island", draft["draft_hash"], 0) + with pytest.raises(psycopg.Error): + app.要求作品().选定探索稿(actor, "retry-selection", request) + assert app.要求作品().读取探索选定(actor, "selection-work")["revision"] == 0 + assert app.要求探索().读取(actor, "island") == draft + with db[用途.维护].连接() as 连, 连.transaction(): + 连.execute("DROP TRIGGER reject_selection ON muse_change_command") + assert ( + app.要求作品().选定探索稿(actor, "retry-selection", request)["results"][0]["revision"] == 1 + ) + + +def test_作者实际选定经HTTP与CLI回查来源__bfcbc2(交接环境, tmp_path): + import subprocess + import sys + from datetime import datetime + + from fastapi.testclient import TestClient + + from muse.接入.http.应用 import 创建应用 + from muse.配置 import 读取配置 + + app, actor, draft, db = 交接环境 + password = tmp_path / "口令.txt" + password.write_text("synthetic-selection-password") + password.chmod(0o600) + cfg = tmp_path / "选定.toml" + cfg.write_text( + '["数据库"]\n"取值方式"="受控存储"\n"位置"=' + + json.dumps(db[用途.生产].引用.位置) + + '\n["资源"]\n"发布身份"="synthetic"\n[HTTP]\n"口令文件"="口令.txt"\n"作者ID"="author"\n"公开地址"="http://testserver"\n"允许来源"=["http://testserver"]\n["文件"]\n"探索草稿"=' + + json.dumps(app.配置.探索草稿) + + "\n" + ) + path = "/api/v1/works/selection-work/exploration-selection" + with TestClient(创建应用(读取配置(cfg)), headers={"origin": "http://testserver"}) as client: + assert client.get(path).status_code == 401 + client.post("/api/v1/session", json={"password": "synthetic-selection-password"}) + request = { + "command_id": "http-select", + "selection": { + "work_id": "selection-work", + "draft_id": "island", + "expected_hash": draft["draft_hash"], + "expected_revision": 0, + }, + } + assert ( + client.post( + path, + json={ + **request, + "confirmedByUser": True, + "confirmedAt": "2026-08-20T12:00:00+08:00", + }, + ).status_code + == 422 + ) + result = client.post(path, json=request) + assert result.status_code == 200, result.text + assert client.post(path, json=request).json() == result.json() + selected = client.get(path).json() + assert selected["selection"]["snapshot"] == draft + assert selected["selection"]["selected_by"] == "author" + assert datetime.fromisoformat(selected["selection"]["selected_at"]).tzinfo is not None + probe = subprocess.run( + [sys.executable, "-m", "muse", "探索", str(cfg), "查看选定", "selection-work"], + text=True, + capture_output=True, + ) + assert probe.returncode == 0, probe.stderr + assert json.loads(probe.stdout) == selected + with pytest.raises(作品错误) as 错: + app.要求作品().读取探索选定(replace(actor, 作者="another-author"), "selection-work") + assert 错.value.错误码 == "SCOPE_DENIED" + + +def test_不同方向同时交接只推进一个选定版本__ad1203(交接环境): + from concurrent.futures import ThreadPoolExecutor + from threading import Barrier + + from muse.共享.错误 import Muse错误 + + app, actor, one, _ = 交接环境 + two = app.要求探索().保存( + actor, "village", "山村", one["content"].replace("空中城镇停电", "山村渡口停航"), None + ) + start = Barrier(2) + + def choose(draft): + start.wait(3) + try: + return app.要求作品().选定探索稿( + actor, + "choose-" + draft["draft_id"], + 探索选择("selection-work", draft["draft_id"], draft["draft_hash"], 0), + ) + except Muse错误 as exc: + return exc.错误码 + + with ThreadPoolExecutor(2) as pool: + results = list(pool.map(choose, [one, two])) + assert sum(isinstance(r, dict) for r in results) == 1 + # 预读时已变化为REVISION_CONFLICT;等待S01依赖锁后变化为SOURCE_STALE。 + assert any(r in {"REVISION_CONFLICT", "SOURCE_STALE"} for r in results if isinstance(r, str)) + assert app.要求作品().读取探索选定(actor, "selection-work")["revision"] == 1 diff --git a/tests/集成/test_规划候选与确认.py b/tests/集成/test_规划候选与确认.py new file mode 100644 index 0000000..4c51132 --- /dev/null +++ b/tests/集成/test_规划候选与确认.py @@ -0,0 +1,696 @@ +"""规划字段由S04描述;候选和正式版本通过真实S01事务区分。""" + +import pytest + +from muse.作品规划.接口 import 档案保存, 规划保存, 规划定位 +from muse.元数据.接口 import 元数据服务, 导入内置结构 +from muse.共享.调用身份 import 内容用途, 用途, 调用身份 +from muse.启动 import 构建 +from muse.正式变更.接口 import 作者动作 +from muse.配置 import 应用配置 + +pytestmark = pytest.mark.数据库 + + +@pytest.mark.parametrize("title", ["雪中的信", "浮岛航线"], ids=["letter", "floating-island"]) +def test_同一目录承接卷章场景且移动保持身份__ad1310(规划环境, title): + from dataclasses import replace + + from muse.作品规划.接口 import 目录节点, 节点目录保存 + from muse.正文写作.接口 import 文本节点, 正文草稿, 段落 + + app, actor, _ = 规划环境 + works = app.要求作品() + nodes = ( + 目录节点("volume-a", "volume", title + "·卷一"), + 目录节点("chapter-a", "chapter", "出发", "volume-a"), + 目录节点("scene-a", "scene", "有人敲门", "chapter-a"), + 目录节点("chapter-b", "chapter", "另一条路"), + ) + request = 节点目录保存("planned-work", 0, nodes) + receipt = works.保存节点目录(actor, "create-nodes", request) + assert receipt == works.保存节点目录(actor, "create-nodes", request) + current = works.读取节点目录(actor, "planned-work") + assert current["revision"] == 1 and len(current["nodes"]) == 4 + assert [c["chapter_id"] for c in works.读取目录(actor, "planned-work")["chapters"]] == [ + "chapter-a", + "chapter-b", + ] + prose = app.要求正文() + prose.保存人工( + actor, + "write-before-move", + "chapter-a", + 0, + 正文草稿((段落("paragraph-a", (文本节点(title + ",有人来了。"),)),)), + ) + before = prose.读取正文(actor, "chapter-a") + moved = (nodes[3], nodes[0], nodes[1], replace(nodes[2], parent_id="chapter-b")) + works.保存节点目录(actor, "move-nodes", replace(request, expected_revision=1, nodes=moved)) + latest = works.读取节点目录(actor, "planned-work") + assert latest["revision"] == 2 + assert {n["node_id"] for n in latest["nodes"]} == {n.node_id for n in nodes} + assert next(n for n in latest["nodes"] if n["node_id"] == "scene-a")["parent_id"] == "chapter-b" + assert [c["chapter_id"] for c in works.读取目录(actor, "planned-work")["chapters"]] == [ + "chapter-b", + "chapter-a", + ] + assert works.读取节点目录(actor, "planned-work", 版本=1) == current + assert prose.读取正文(actor, "chapter-a") == before + + +def test_规划候选编辑追加版本且旧审阅失效__ad1306(规划环境): + from muse.元数据.接口 import 字段变更 + from muse.共享.错误 import Muse错误 + + app, actor, _ = 规划环境 + service = app.要求规划() + schema = service.新规划结构(actor, "planned-work", "outline", 1) + reply = service.保存人工候选( + actor, + "editable-plan", + 规划保存("planned-work", 0, schema, {"名称": "未寄出的信", "主线一句话": "寻找收信人"}), + ) + id_ = reply["results"][0]["candidate_id"] + old = service.读取候选(actor, id_) + loc = 规划定位(id_, 1, old["candidate_hash"], old["plan_id"], 0) + review = service.打开审阅(actor, loc) + changes = (字段变更(("主线一句话",), "set", "收信人寻找寄信者"),) + edited = service.编辑候选(actor, "edit-plan-once", loc, changes) + assert edited == service.编辑候选(actor, "edit-plan-once", loc, changes) + current = service.读取候选(actor, id_) + assert current["revision"] == 2 and current["candidate_hash"] != old["candidate_hash"] + assert current["content"] == {"名称": "未寄出的信", "主线一句话": "收信人寻找寄信者"} + assert service.读取候选(actor, id_, 版本=1) == old + assert service.列出规划(actor, "planned-work") == [] + with pytest.raises(Muse错误): + service.决定( + actor, + "old-plan-review", + loc, + 作者动作.采纳, + review["review_id"], + review["review_hash"], + ("plan",), + ) + with pytest.raises(Muse错误): + service.编辑候选(actor, "stale-edit-plan", loc, changes) + new_loc = 规划定位(id_, 2, current["candidate_hash"], current["plan_id"], 0) + new_review = service.打开审阅(actor, new_loc) + service.决定( + actor, + "accept-edited-plan", + new_loc, + 作者动作.采纳, + new_review["review_id"], + new_review["review_hash"], + ("plan",), + ) + assert service.读取规划(actor, current["plan_id"])["content"] == current["content"] + + +def test_卷章场景采用动态结构且章入口共用身份__ad1311(规划环境): + from dataclasses import replace + + from muse.作品规划.接口 import 目录节点, 节点目录保存 + + app, actor, _ = 规划环境 + app.要求作品().保存节点目录( + actor, + "plan-targets", + 节点目录保存( + "planned-work", + 0, + ( + 目录节点("volume", "volume", "卷一"), + 目录节点("chapter", "chapter", "开门", "volume"), + 目录节点("scene", "scene", "门外的访客", "chapter"), + ), + ), + ) + service = app.要求规划() + outline = service.新规划结构(actor, "planned-work", "outline", 1) + volume = 保存并确认( + service, + actor, + 规划保存("planned-work", 0, outline, {"名称": "卷一大纲"}, node_id="volume"), + "volume-plan", + ) + scene = service.新规划结构(actor, "planned-work", "scene", 1) + saved_scene = 保存并确认( + service, + actor, + 规划保存( + "planned-work", + 0, + scene, + {"目标推进收束": "来客想借一盏灯,邮差让他进屋。"}, + node_id="scene", + ), + "scene-plan", + ) + assert saved_scene["chapter_id"] == "chapter" and saved_scene["node_id"] == "scene" + fine = service.新规划结构(actor, "planned-work", "fine_outline", 1) + req = 规划保存( + "planned-work", + 0, + fine, + {**细纲内容(volume["plan_id"]), "targetChapter": 1}, + node_id="chapter", + ) + first = 保存并确认(service, actor, req, "chapter-node-plan") + assert first["chapter_id"] == "chapter" + second = 保存并确认( + service, + actor, + replace(req, node_id=None, chapter_id="chapter", expected_revision=1), + "chapter-legacy-plan", + ) + assert first["plan_id"] == second["plan_id"] and second["revision"] == 2 + assert service.读取可用细纲(actor, "planned-work", "chapter")["revision"] == 2 + + +@pytest.fixture +def 规划环境(应用测试库): + with 应用测试库[用途.维护].连接() as 连, 连.transaction(): + 导入内置结构(元数据服务(连)) + app = 构建(应用配置(应用测试库[用途.生产].引用, "synthetic")) + actor = 调用身份("author", None, 用途.生产, 内容用途.规划) + schema = app.要求作品().新档案结构(actor, "planned-work", "work_core", 1) + app.要求作品().保存档案( + actor, "create-planned-work", 档案保存("planned-work", 0, {"名称": "雪中的信"}, schema) + ) + return app, actor, 应用测试库 + + +def test_人工规划候选经明确审阅后形成正式版本__ad1301(规划环境): + app, actor, _ = 规划环境 + service = app.要求规划() + schema = service.新规划结构(actor, "planned-work", "outline", 1) + request = 规划保存( + "planned-work", 0, schema, {"名称": "雪中的信", "主线一句话": "邮差寻找一封无人认领的信。"} + ) + receipt = service.保存人工候选(actor, "outline-draft", request) + candidate = service.读取候选(actor, receipt["results"][0]["candidate_id"]) + assert service.列出规划(actor, "planned-work") == [] + location = 规划定位( + candidate["candidate_id"], + candidate["revision"], + candidate["candidate_hash"], + candidate["plan_id"], + 0, + ) + review = service.打开审阅(actor, location) + result = service.决定( + actor, + "confirm-outline", + location, + 作者动作.采纳, + review["review_id"], + review["review_hash"], + ("plan",), + ) + saved = service.读取规划(actor, candidate["plan_id"]) + assert saved["revision"] == 1 and saved["content"] == request.content + assert ( + service.决定( + actor, + "confirm-outline", + location, + 作者动作.采纳, + review["review_id"], + review["review_hash"], + ("plan",), + ) + == result + ) + + +def test_规划HTTP候选与CLI决定共享版本回执__ad1302(规划环境, tmp_path): + import json + import subprocess + import sys + + from fastapi.testclient import TestClient + + from muse.接入.http.应用 import 创建应用 + from muse.配置 import 读取配置 + + _, _, db = 规划环境 + password = tmp_path / "口令.txt" + password.write_text("synthetic-plan-password") + password.chmod(0o600) + cfg = tmp_path / "规划.toml" + cfg.write_text( + '["数据库"]\n"取值方式"="受控存储"\n"位置"=' + + json.dumps(db[用途.生产].引用.位置) + + '\n["资源"]\n"发布身份"="synthetic"\n[HTTP]\n"口令文件"="口令.txt"\n"作者ID"="author"\n"公开地址"="http://testserver"\n"允许来源"=["http://testserver"]\n' + ) + with TestClient(创建应用(读取配置(cfg)), headers={"origin": "http://testserver"}) as client: + assert ( + client.post("/api/v1/session", json={"password": "synthetic-plan-password"}).status_code + == 200 + ) + nodes_request = { + "command_id": "http-plan-nodes", + "directory": { + "work_id": "planned-work", + "expected_revision": 0, + "nodes": [ + {"node_id": "vol-http", "kind": "volume", "title": "卷一", "parent_id": None} + ], + }, + } + nodes_response = client.post("/api/v1/works/planned-work/plan-nodes", json=nodes_request) + assert nodes_response.status_code == 200, nodes_response.text + node_cli = subprocess.run( + [sys.executable, "-m", "muse", "作品", str(cfg), "节点目录", "planned-work"], + text=True, + capture_output=True, + ) + assert node_cli.returncode == 0, node_cli.stderr + assert ( + json.loads(node_cli.stdout) + == client.get("/api/v1/works/planned-work/plan-nodes").json() + ) + node_path = tmp_path / "节点.json" + node_path.write_text(json.dumps(nodes_request, ensure_ascii=False)) + node_replay = subprocess.run( + [sys.executable, "-m", "muse", "作品", str(cfg), "保存节点", str(node_path)], + text=True, + capture_output=True, + ) + assert node_replay.returncode == 0, node_replay.stderr + assert json.loads(node_replay.stdout) == nodes_response.json() + schema = client.get( + "/api/v1/works/planned-work/plans/schema", + params={"schema_id": "outline", "base_version": 1}, + ).json() + request = { + "command_id": "http-plan", + "request": { + "work_id": "planned-work", + "expected_revision": 0, + "schema": schema, + "content": {"名称": "下雪以前"}, + }, + } + created = client.post("/api/v1/works/planned-work/plans/candidates", json=request) + assert created.status_code == 200, created.text + id_ = created.json()["results"][0]["candidate_id"] + candidate = client.get(f"/api/v1/plan-candidates/{id_}").json() + location = { + "candidate_id": id_, + "candidate_revision": 1, + "candidate_hash": candidate["candidate_hash"], + "plan_id": candidate["plan_id"], + "expected_revision": 0, + } + edit = { + "command_id": "edit-plan-cli", + "candidate": location, + "changes": [{"路径": ["名称"], "动作": "set", "值": "下雪后的信"}], + } + edit_path = tmp_path / "编辑.json" + edit_path.write_text(json.dumps(edit, ensure_ascii=False)) + edit_result = subprocess.run( + [sys.executable, "-m", "muse", "规划", str(cfg), "编辑候选", str(edit_path)], + text=True, + capture_output=True, + ) + assert edit_result.returncode == 0, edit_result.stderr + assert client.post( + f"/api/v1/plan-candidates/{id_}/versions", json=edit + ).json() == json.loads(edit_result.stdout) + assert client.get(f"/api/v1/plan-candidates/{id_}", params={"version": 1}).json()[ + "content" + ] == {"名称": "下雪以前"} + candidate = client.get(f"/api/v1/plan-candidates/{id_}").json() + location.update(candidate_revision=2, candidate_hash=candidate["candidate_hash"]) + review = client.post(f"/api/v1/plan-candidates/{id_}/reviews", json=location) + assert review.status_code == 200, review.text + decision = { + "command_id": "cli-accept-plan", + "candidate": location, + "action": "adopt", + "author_review_id": review.json()["review_id"], + "review_hash": review.json()["review_hash"], + "approved_changes": ["plan"], + } + path = tmp_path / "决定.json" + path.write_text(json.dumps(decision, ensure_ascii=False)) + result = subprocess.run( + [sys.executable, "-m", "muse", "规划", str(cfg), "决定", str(path)], + text=True, + capture_output=True, + ) + assert result.returncode == 0, result.stderr + readback = client.get(f"/api/v1/plans/{candidate['plan_id']}") + assert readback.status_code == 200 and readback.json()["content"] == {"名称": "下雪后的信"} + assert client.post( + f"/api/v1/plan-candidates/{id_}/decisions", json=decision + ).json() == json.loads(result.stdout) + + +def test_陈旧规划审阅不能覆盖新版本但可重新拒绝__ad1303(规划环境): + from dataclasses import replace + + from muse.共享.错误 import Muse错误 + + app, actor, _ = 规划环境 + service = app.要求规划() + schema = service.新规划结构(actor, "planned-work", "outline", 1) + locations = [] + for name in ("一封信", "一封旧信"): + reply = service.保存人工候选( + actor, name, 规划保存("planned-work", 0, schema, {"名称": name}) + ) + item = service.读取候选(actor, reply["results"][0]["candidate_id"]) + locations.append( + 规划定位(item["candidate_id"], 1, item["candidate_hash"], item["plan_id"], 0) + ) + old_review = service.打开审阅(actor, locations[0]) + new_review = service.打开审阅(actor, locations[1]) + service.决定( + actor, + "approve-new", + locations[1], + 作者动作.采纳, + new_review["review_id"], + new_review["review_hash"], + ("plan",), + ) + with pytest.raises(Muse错误): + service.决定( + actor, + "approve-stale", + locations[0], + 作者动作.采纳, + old_review["review_id"], + old_review["review_hash"], + ("plan",), + ) + current = service.读取规划(actor, locations[1].plan_id) + assert current["revision"] == 1 and current["content"]["名称"] == "一封旧信" + rejected = replace(locations[0], expected_revision=1) + review = service.打开审阅(actor, rejected) + result = service.决定( + actor, "reject-old", rejected, 作者动作.拒绝, review["review_id"], review["review_hash"], () + ) + assert result["action"] == "reject" + assert service.读取规划(actor, locations[1].plan_id) == current + + +def 保存并确认(service, actor, request, command): + reply = service.保存人工候选(actor, command + "-draft", request) + item = service.读取候选(actor, reply["results"][0]["candidate_id"]) + location = 规划定位( + item["candidate_id"], + item["revision"], + item["candidate_hash"], + item["plan_id"], + request.expected_revision, + ) + review = service.打开审阅(actor, location) + service.决定( + actor, + command + "-accept", + location, + 作者动作.采纳, + review["review_id"], + review["review_hash"], + ("plan",), + ) + return service.读取规划(actor, item["plan_id"]) + + +def test_已确认细纲返回字典并检验当前来源__e1f9cb(规划环境): + from dataclasses import replace + + from muse.共享.错误 import Muse错误 + + app, actor, _ = 规划环境 + service = app.要求规划() + for i in range(1, 4): + app.要求作品().添加章节( + actor, f"chapter-{i}", "planned-work", f"chapter-{i}", f"第{i}章", 预期目录版本=i - 1 + ) + outline_schema = service.新规划结构(actor, "planned-work", "outline", 1) + outline = 规划保存("planned-work", 0, outline_schema, {"名称": "原大纲"}) + parent = 保存并确认(service, actor, outline, "parent") + with pytest.raises(Muse错误) as missing: + service.读取可用细纲(actor, "planned-work", "chapter-3") + assert missing.value.错误码 == "PRECONDITION_PENDING" + schema = service.新规划结构(actor, "planned-work", "fine_outline", 1) + payload = 细纲内容(parent["plan_id"]) + req = 规划保存("planned-work", 0, schema, payload, chapter_id="chapter-3") + first = 保存并确认(service, actor, req, "fine") + loaded = service.读取可用细纲(actor, "planned-work", "chapter-3") + assert isinstance(loaded["content"], dict) and loaded["content"]["targetChapter"] == 3 + assert loaded["content"]["chapterGoal"] == "x" and loaded["revision"] == 1 + second = 保存并确认( + service, + actor, + replace(req, expected_revision=1, content={**payload, "chapterGoal": "x2"}), + "fine-v2", + ) + assert first["plan_id"] == second["plan_id"] + assert ( + service.读取可用细纲(actor, "planned-work", "chapter-3")["content"]["chapterGoal"] == "x2" + ) + 保存并确认( + service, + actor, + replace(outline, expected_revision=1, content={"名称": "新的大纲"}), + "parent-v2", + ) + with pytest.raises(Muse错误) as stale: + service.读取可用细纲(actor, "planned-work", "chapter-3") + assert stale.value.错误码 == "SOURCE_STALE" + + +def test_规划动态引用由故事世界核实并纳入提交依赖__ad1304(规划环境): + from muse.元数据.接口 import 作品扩展, 值规则, 字段定义, 引用规则 + from muse.共享.错误 import Muse错误 + + app, actor, db = 规划环境 + 建立人物(app, actor) + with db[用途.维护].连接() as 连, 连.transaction(): + extension = 作品扩展( + "planned-work", + "outline", + 1, + 1, + ( + 字段定义( + "outline.concern", + "牵挂的人", + "牵挂的人", + 值规则("reference", 引用=引用规则(("character",))), + ), + ), + ) + 元数据服务(连).发布扩展(extension, extension.内容哈希) + service = app.要求规划() + schema = service.新规划结构(actor, "planned-work", "outline", 1, 扩展版本=1) + bad = 规划保存("planned-work", 0, schema, {"名称": "找信", "牵挂的人": "missing-person"}) + with pytest.raises(Muse错误) as missing: + service.保存人工候选(actor, "missing-reference", bad) + assert missing.value.错误码 == "FACT_REFERENCE_INVALID" + saved = 保存并确认( + service, + actor, + 规划保存("planned-work", 0, schema, {"名称": "找信", "牵挂的人": "postman"}), + "linked-plan", + ) + assert saved["content"]["牵挂的人"] == "postman" + assert {"种类": "B02.object", "对象ID": "world:postman", "版本": "1"} in saved[ + "source_dependencies" + ] + + +def 细纲内容(parent_id): + return { + "targetChapter": 3, + "sourceRef": {"sourceId": parent_id, "sourceVersion": 1}, + "chapterGoal": "x", + "hardConstraints": [], + "keyEvents": [ + { + "id": "letter", + "order": 1, + "event": "邮差走到门口", + "participants": [], + "trigger": "收到退回的信", + "resultDirection": "确认这座房子还住着人", + } + ], + "adjustableBeats": [], + "declaredNewFacts": [], + "mustAppearEntities": [], + "chapterEndHook": "门内传来了脚步声。", + } + + +def 准备可用细纲(规划环境): + app, actor, _ = 规划环境 + service = app.要求规划() + for i in range(1, 4): + app.要求作品().添加章节( + actor, f"add-{i}", "planned-work", f"chapter-{i}", f"第{i}章", 预期目录版本=i - 1 + ) + parent_req = 规划保存( + "planned-work", + 0, + service.新规划结构(actor, "planned-work", "outline", 1), + {"名称": "寄信者"}, + ) + parent = 保存并确认(service, actor, parent_req, "parent") + child_req = 规划保存( + "planned-work", + 0, + service.新规划结构(actor, "planned-work", "fine_outline", 1), + 细纲内容(parent["plan_id"]), + chapter_id="chapter-3", + ) + child = 保存并确认(service, actor, child_req, "child") + return parent_req, parent, child_req, child + + +def test_规划关联按类型归属解析且不构成依据环__ad1307(规划环境): + from dataclasses import replace + + app, actor, _ = 规划环境 + service = app.要求规划() + parent_req, parent, child_req, child = 准备可用细纲(规划环境) + linked = 保存并确认( + service, + actor, + replace( + parent_req, + expected_revision=1, + content={"名称": "寄信者", "近三章细纲": [child["plan_id"]]}, + ), + "link-child", + ) + assert linked["content"]["近三章细纲"] == [child["plan_id"]] + assert not any(d["对象ID"] == "plan:" + child["plan_id"] for d in linked["source_dependencies"]) + refreshed = replace( + child_req, + expected_revision=1, + content={ + **child_req.content, + "sourceRef": {"sourceId": parent["plan_id"], "sourceVersion": 2}, + }, + ) + 保存并确认(service, actor, refreshed, "refresh-child") + assert service.读取可用细纲(actor, "planned-work", "chapter-3")["revision"] == 2 + + +@pytest.mark.parametrize( + "policy", ["deprecated", "writing-revoked"], ids=["deprecated", "writing-revoked"] +) +def test_细纲必需字段撤回用途后不能用于写作__ad1308(规划环境, policy): + from dataclasses import asdict + + from muse.元数据.接口 import 启用命令, 字段限制, 定义哈希, 策略快照 + from muse.共享.错误 import Muse错误 + + app, actor, db = 规划环境 + _, _, _, child = 准备可用细纲(规划环境) + with db[用途.维护].连接() as 连, 连.transaction(): + rule = ( + 字段限制("fine_outline.chapterGoal", (), True) + if policy == "deprecated" + else 字段限制("fine_outline.chapterGoal", ("aiContext:generation",)) + ) + version = 策略快照("fine_outline", 1, (rule,)) + 元数据服务(连).更新策略( + version, + 启用命令("restrict-plan", 定义哈希(asdict(version)), 0, "author", "type:fine_outline"), + ) + # 历史仍可查看;严格可用读取不把裁剪后缺失的章目标当作完整细纲。 + assert app.要求规划().读取规划(actor, child["plan_id"])["revision"] == 1 + with pytest.raises(Muse错误) as rejected: + app.要求规划().读取可用细纲(actor, "planned-work", "chapter-3") + assert rejected.value.错误码 == "PRECONDITION_PENDING" + + +def 建立人物(app, actor): + from muse.故事世界.接口 import 事实定位, 事实提案请求 + + world = app.要求故事世界() + structure = world.新事实结构(actor, "planned-work", "postman", "character", 1) + result = world.提出事实( + actor, + "postman-fact", + 事实提案请求( + "planned-work", "postman", 0, structure, {"名称": "邮差"}, (), None, independent=True + ), + ) + item = world.读取提案(actor, result["results"][0]["proposal_id"]) + location = 事实定位(item["proposal_id"], 1, item["proposal_hash"], "world:postman", 0) + review = world.打开事实审阅(actor, location) + world.决定事实( + actor, + "confirm-postman", + location, + 作者动作.采纳, + review["review_id"], + review["review_hash"], + ("object",), + ) + + +@pytest.mark.parametrize("形态", ["string", "object"]) +def test_细纲两种实体形式都核对真实本书引用__ad1305(规划环境, 形态): + from dataclasses import replace + + from muse.共享.错误 import Muse错误 + + app, actor, _ = 规划环境 + 建立人物(app, actor) + for i in range(1, 4): + app.要求作品().添加章节( + actor, + f"ref-chapter-{i}", + "planned-work", + f"chapter-{i}", + f"第{i}章", + 预期目录版本=i - 1, + ) + service = app.要求规划() + parent = 保存并确认( + service, + actor, + 规划保存( + "planned-work", + 0, + service.新规划结构(actor, "planned-work", "outline", 1), + {"名称": "找信"}, + ), + "reference-parent", + ) + content = 细纲内容(parent["plan_id"]) + value = ( + "postman" if 形态 == "string" else {"id": "postman", "type": "character", "name": "邮差"} + ) + content["mustAppearEntities"] = [value] + schema = service.新规划结构(actor, "planned-work", "fine_outline", 1) + req = 规划保存("planned-work", 0, schema, content, chapter_id="chapter-3") + missing = ( + "missing" if 形态 == "string" else {"id": "missing", "type": "character", "name": "邮差"} + ) + with pytest.raises(Muse错误) as error: + service.保存人工候选( + actor, + "invalid-entity", + replace(req, content={**content, "mustAppearEntities": [missing]}), + ) + assert error.value.错误码 == "FACT_REFERENCE_INVALID" + 保存并确认(service, actor, req, "referenced-fine") + loaded = service.读取可用细纲(actor, "planned-work", "chapter-3") + assert loaded["content"]["mustAppearEntities"] == [value] + assert {"种类": "B02.object", "对象ID": "world:postman", "版本": "1"} in loaded[ + "source_dependencies" + ] diff --git a/web/src/功能/作品规划/探索工作区.tsx b/web/src/功能/作品规划/探索工作区.tsx new file mode 100644 index 0000000..f834535 --- /dev/null +++ b/web/src/功能/作品规划/探索工作区.tsx @@ -0,0 +1,315 @@ +import { useEffect, useState } from "react"; +import { useQuery, useQueryClient } from "@tanstack/react-query"; +import { + Link, + useNavigate, + useParams, + useSearchParams, +} from "react-router-dom"; +import { useMuse作者会话 } from "../../应用/作者会话"; +import { 接口错误 } from "../../接口/传输"; +import { Ui反馈状态 } from "../../界面/反馈状态"; +import { Ui探索交接 } from "./规划工作区"; +import { + 列出探索, + 读取探索, + 保存探索, + 检查探索, + type 探索稿, + type 探索保存, +} from "./数据"; + +export function Ui探索工作区() { + const author = useMuse作者会话(); + const { draft_id } = useParams(); + if (!author.data) return私人笔记
++ 先保留一个念头,也可以留下暂时想不明白的地方。 +
++ 还没有探索稿。从一个人物、一件物品或一句话开始。 +
+ )} +继续整理想法 →
+私人笔记
++ 可以记片段、比较不同方向,也可以原样留下问题。整理交付时再检查格式。 +
+{message}
+ {buffer.pending && !busy && ( +保存结果暂未确认,重试会使用刚才的同一份内容。
+ )} + {error &&{error}
} + {current && ( + + )} + {check && ( +格式检查通过。故事方向和正式规划仍由你决定。
+ ) : ( +{query.error.message}
} + {query.data && ( ++ 目录调整保留已有章和场景身份,规划内容仍需分别审阅确认。 +
+ + {conflict && ( + + )} + + {message &&{message}
} ++ 确认采用当前整理的方向后,保存这一版本作为规划依据。后续修改私人稿不会改动已选定的版本。 +
+ {works.error &&{works.error.message}
} + {works.data?.length === 0 ? ( ++ 先到 + + 我的作品 + + 建立作品档案,再回来交接。 +
+ ) : ( + + )} + {current.data?.selection && !selected && ( ++ 作品此前采用“{current.data.selection.snapshot.title} + ”。本次选择会留下新的版本,旧记录保留。 +
+ )} + {current.error &&{current.error.message}
} + + {pending.current && !busy && ( +交接结果尚未确认,重试使用同一次请求。
+ )} + {error &&{error}
} + {notice &&{notice}
} + {selected && ( + + 继续整理作品规划 → + + )} +把故事排出来
++ 当前方向来自“{selection.selection.snapshot.title}”。 +
+ )} + {notice &&{notice}
} + {error &&{error}
} + {editing ? ( +保存为候选,审阅后再确认。
+ {conflict && ( + + )} +保存一份候选后,可以在这里比较和决定。
+ )} +尚未确认规划。
+ )} +已有更新的正式规划;这份候选可以暂缓或拒绝。
+ )} +