diff --git a/src/muse/接入/http/路由/故事世界.py b/src/muse/接入/http/路由/故事世界.py new file mode 100644 index 0000000..d9e40f0 --- /dev/null +++ b/src/muse/接入/http/路由/故事世界.py @@ -0,0 +1,120 @@ +"""本书事实的实际查询、提案和作者决定；不接受客户端自报作者。""" + +from typing import Literal + +from fastapi import APIRouter, Query, Request + +from muse.接入.http.作者会话 import 作者依赖 +from muse.接入.创作操作 import ( + 世界查询结果, + 事实决定请求, + 事实创建请求, + 事实审阅结果, + 事实审阅请求, + 事实提案结果, + 创作身份, + 核对路径, + 正式回执结果, +) +from muse.故事世界.接口 import 事实结构 +from muse.正式变更.接口 import 作者动作 + +路由 = APIRouter(prefix="/api/v1", tags=["故事世界"]) + + +@路由.get("/works/{work_id}/facts/schema", response_model=事实结构, operation_id="read_fact_schema") +def 读取事实结构( + work_id: str, + object_id: str, + schema_id: str, + request: Request, + 作者: 作者依赖, + base_version: int = Query(default=1, ge=1), + instance_family: Literal["entity", "relation"] = "entity", + extension_version: int | None = Query(default=None, ge=1), +) -> 事实结构: + 装配 = request.app.state.装配 + return 装配.要求故事世界().新事实结构( + 创作身份(作者, 装配.配置), + work_id, + object_id, + schema_id, + base_version, + 家族=instance_family, + 扩展版本=extension_version, + ) + + +@路由.get( + "/works/{work_id}/facts", + response_model=世界查询结果, + response_model_exclude_unset=True, + operation_id="query_world_state", +) +def 查询事实( + work_id: str, + request: Request, + 作者: 作者依赖, + as_of: str | None = None, + system_revision: int | None = Query(default=None, ge=0), + view: Literal["author", "reader", "character"] = "author", + character_id: str | None = None, +) -> dict: + 装配 = request.app.state.装配 + return 装配.要求故事世界().查询世界( + 创作身份(作者, 装配.配置), + work_id, + as_of, + 系统版本=system_revision, + 视角=view, + 知情主体ID=character_id, + ) + + +@路由.post( + "/works/{work_id}/fact-proposals", + response_model=正式回执结果, + operation_id="propose_world_fact", +) +def 提出事实(work_id: str, 请求: 事实创建请求, request: Request, 作者: 作者依赖) -> dict: + 核对路径(work_id, 请求.proposal.work_id) + 装配 = request.app.state.装配 + return 装配.要求故事世界().提出事实(创作身份(作者, 装配.配置), 请求.command_id, 请求.proposal) + + +@路由.get( + "/fact-proposals/{proposal_id}", response_model=事实提案结果, operation_id="read_fact_proposal" +) +def 读取提案(proposal_id: str, request: Request, 作者: 作者依赖) -> dict: + 装配 = request.app.state.装配 + return 装配.要求故事世界().读取提案(创作身份(作者, 装配.配置), proposal_id) + + +@路由.post( + "/fact-proposals/{proposal_id}/reviews", + response_model=事实审阅结果, + operation_id="open_fact_review", +) +def 打开事实审阅(proposal_id: str, 请求: 事实审阅请求, request: Request, 作者: 作者依赖) -> dict: + 核对路径(proposal_id, 请求.proposal_id) + 装配 = request.app.state.装配 + return 装配.要求故事世界().打开事实审阅(创作身份(作者, 装配.配置), 请求.定位()) + + +@路由.post( + "/fact-proposals/{proposal_id}/decisions", + response_model=正式回执结果, + operation_id="decide_world_fact", +) +def 决定事实(proposal_id: str, 请求: 事实决定请求, request: Request, 作者: 作者依赖) -> dict: + 核对路径(proposal_id, 请求.proposal_id) + 装配 = request.app.state.装配 + return 装配.要求故事世界().决定事实( + 创作身份(作者, 装配.配置), + 请求.command_id, + 请求.定位(), + 作者动作(请求.action), + 请求.author_review_id, + 请求.review_hash, + tuple(请求.approved_changes), + ) diff --git a/src/muse/故事世界/__init__.py b/src/muse/故事世界/__init__.py new file mode 100644 index 0000000..c9c575d --- /dev/null +++ b/src/muse/故事世界/__init__.py @@ -0,0 +1 @@ +"""本书实体、关系、时点与事实；业务入口见接口.py。""" diff --git a/src/muse/故事世界/事实提案.py b/src/muse/故事世界/事实提案.py new file mode 100644 index 0000000..c949156 --- /dev/null +++ b/src/muse/故事世界/事实提案.py @@ -0,0 +1,128 @@ +"""把事实提案的字面依据绑定到固定正文；不改字节，不因引文存在而自动确认。""" + +from dataclasses import replace + +from muse.元数据.接口 import 有效结构, 校验值, 类型定义, 选择实例维护者 +from muse.故事世界.模型 import 事实条目, 事实错误, 引文锚点, 引用对象 +from muse.故事世界.连续性 import 核对叙事窗口 +from muse.正文写作.接口 import 可见文本, 正文草稿 + + +def 定位字面引文(正文: 正文草稿, 引文: str) -> 引文锚点: + """定位一个实际出现位置，仅用于来源引用；不用于自动修改正文。""" + if 引文: + for 段 in 正文.paragraphs: + 内容 = 可见文本(正文草稿((段,))) + 开始 = 内容.find(引文) + if 开始 >= 0: + return 引文锚点(段.paragraph_id, 开始, 开始 + len(引文), 引文) + raise 事实错误("FACT_QUOTE_NOT_FOUND", "证据不在正文的所声明段落中") + + +def 核对原文引文(正文: 正文草稿, 引用: 引文锚点) -> str: + 段 = next((p for p in 正文.paragraphs if p.paragraph_id == 引用.paragraph_id), None) + if 段 is None or 可见文本(正文草稿((段,)))[引用.start : 引用.end] != 引用.quote: + raise 事实错误("FACT_QUOTE_NOT_FOUND", "引文位置与固定正文不一致") + return 引用.quote + + +def 解析类型选择(选择: str, 类型目录: tuple[类型定义, ...]) -> 类型定义: + 按ID = next((t for t in 类型目录 if t.type_id == 选择), None) + if 按ID is not None: + return 按ID + 中文命中 = [t for t in 类型目录 if t.名称 == 选择] + if len(中文命中) != 1: + raise 事实错误("FACT_TYPE_UNKNOWN", "提案类型未登记或名称不能唯一定位") + return 中文命中[0] + + +def 校验事实条目( + 条目: 事实条目, + *, + 作品ID: str, + 类型: 类型定义, + 结构: 有效结构, + 正文: 正文草稿, + 可用对象: dict[str, 引用对象], + 章节顺序: dict[str, int] | None = None, +) -> 事实条目: + """类型/字段来自元数据；实际对象及当前可用性由B02在同一核验范围中提供。""" + if 条目.type_id != 类型.type_id or 结构.基础.type_id != 类型.type_id: + raise 事实错误("FACT_TYPE_UNKNOWN", "事实类型与固定结构不一致") + if 选择实例维护者(类型, 条目.instance_family) != "B02": + raise 事实错误("SCOPE_DENIED", "本书事实不能写入参考分析或公共方法") + 结果 = 校验值(结构, 条目.content) + if 条目.planned_window is not None: + 核对叙事窗口(条目.planned_window, 章节顺序 or {}) + for 引文 in 条目.evidence: + 核对原文引文(正文, 引文) + 引用数 = 0 + for 引用 in 结果.引用需求: + 值列表 = 引用.值 if isinstance(引用.值, list) else [引用.值] + for id_ in 值列表: + 对象 = 可用对象.get(id_) if isinstance(id_, str) else None + if 对象 is None or 对象.work_id != 作品ID or 对象.type_id not in 引用.规则.目标类型: + raise 事实错误("FACT_REFERENCE_INVALID", "引用对象缺失、类型不符或不属于本书") + if 对象.state not in {"confirmed", "approved_in_batch"} or not 对象.available: + raise 事实错误("FACT_REFERENCE_INVALID", "引用对象尚未批准或在当前时点不可用") + 引用数 += 1 + if 条目.instance_family == "relation" and 引用数 < 2: + raise 事实错误("FACT_REFERENCE_INVALID", "关系需要由字段合同声明并解析的端点") + return replace(条目, content=结果.值) + + +def 核对提案身份(条目: tuple[事实条目, ...]) -> None: + if len({p.proposal_id for p in 条目}) != len(条目): + raise 事实错误("FACT_DUPLICATE_ID", "同一提案批次不能包含重复条目身份") + + +def 校验事实批次( + 条目: tuple[事实条目, ...], + *, + 作品ID: str, + 结构目录: dict[str, tuple[类型定义, 有效结构]], + 正文: 正文草稿, + 可用对象: dict[str, 引用对象], + 章节顺序: dict[str, int] | None = None, +) -> tuple[事实条目, ...]: + 核对提案身份(条目) + 结果 = [] + for 项 in 条目: + if 项.type_id not in 结构目录: + raise 事实错误("FACT_TYPE_UNKNOWN", "提案类型没有可用的已登记结构") + 类型, 结构 = 结构目录[项.type_id] + 结果.append( + 校验事实条目( + 项, + 作品ID=作品ID, + 类型=类型, + 结构=结构, + 正文=正文, + 可用对象=可用对象, + 章节顺序=章节顺序, + ) + ) + return tuple(结果) + + +def 保留名称提及(原输出: dict, 正文: 正文草稿, 类型目录: tuple[类型定义, ...]) -> dict: + """失败产物只保留可定位的名称提及；不保留无依据属性，不当成完整抽取成功。""" + 提及 = [] + 舍弃 = {"entities": 0, "relations": len(原输出.get("relations", []))} + for 实体 in 原输出.get("entities", []): + try: + 类型 = 解析类型选择(实体["type"], 类型目录) + if 选择实例维护者(类型, "entity") != "B02": + raise 事实错误("SCOPE_DENIED", "不是本书实体类型") + 名称 = 实体["name"] + 引文 = 定位字面引文(正文, 名称) + except (KeyError, TypeError, 事实错误): + 舍弃["entities"] += 1 + continue + 提及.append({"type_id": 类型.type_id, "name": 名称, "evidence": 引文}) + return { + "mentions": 提及, + "relations": [], + "mechanical_drops": 舍弃, + "state": "needs_confirmation", + } diff --git a/src/muse/故事世界/存储.py b/src/muse/故事世界/存储.py new file mode 100644 index 0000000..e7c4019 --- /dev/null +++ b/src/muse/故事世界/存储.py @@ -0,0 +1,160 @@ +"""B02所属提案、实例和事实版本SQL；不跨owner写入。""" + +import hashlib +from dataclasses import asdict + +from psycopg.rows import dict_row +from psycopg.types.json import Jsonb + +from muse.故事世界.模型 import 事实错误 +from muse.故事世界.连续性 import 选择时点记录 + + +class 故事存储: + def __init__(self, 连): + self.连 = 连 + + def 读取对象(self, object_id: str): + return ( + self.连.cursor(row_factory=dict_row) + .execute( + "SELECT o.*,v.revision,v.system_revision,v.content,v.content_hash,v.schema_binding," + "v.sources,v.independent,v.effective_from,v.knowledge_mode," + "v.known_to,v.reader_visible,v.planned_window " + "FROM muse_world_object o JOIN muse_world_version v ON v.object_id=o.object_id " + "AND v.revision=o.current_revision WHERE o.object_id=%s", + (object_id,), + ) + .fetchone() + ) + + def 读取提案(self, proposal_id: str): + return ( + self.连.cursor(row_factory=dict_row) + .execute( + "SELECT p.*,v.revision,v.request,v.proposed_content,v.proposal_hash " + "FROM muse_fact_proposal p JOIN muse_fact_proposal_version v " + "ON v.proposal_id=p.proposal_id AND v.revision=p.current_revision " + "WHERE p.proposal_id=%s", + (proposal_id,), + ) + .fetchone() + ) + + def 写提案(self, proposal_id: str, 请求, 内容: dict, 哈希: str) -> dict: + self.连.execute( + "INSERT INTO muse_fact_proposal(proposal_id,work_id,object_id,current_revision) " + "VALUES (%s,%s,%s,1)", + (proposal_id, 请求.work_id, 请求.object_id), + ) + self.连.execute( + "INSERT INTO muse_fact_proposal_version" + "(proposal_id,revision,request,proposed_content,proposal_hash) " + "VALUES (%s,1,%s,%s,%s)", + (proposal_id, Jsonb(asdict(请求)), Jsonb(内容), 哈希), + ) + return {"proposal_id": proposal_id, "revision": 1, "proposal_hash": 哈希} + + def 锁对象(self, object_id: str) -> int: + key = int.from_bytes( + hashlib.sha256(("B02:object:" + object_id).encode()).digest()[:8], signed=True + ) + self.连.execute("SELECT pg_advisory_xact_lock(%s)", (key,)) + row = self.连.execute( + "SELECT current_revision FROM muse_world_object WHERE object_id=%s FOR UPDATE", + (object_id,), + ).fetchone() + return row[0] if row else 0 + + def 锁提案(self, proposal_id: str) -> int: + key = int.from_bytes( + hashlib.sha256(("B02:proposal:" + proposal_id).encode()).digest()[:8], signed=True + ) + self.连.execute("SELECT pg_advisory_xact_lock(%s)", (key,)) + row = self.连.execute( + "SELECT current_revision FROM muse_fact_proposal WHERE proposal_id=%s FOR UPDATE", + (proposal_id,), + ).fetchone() + return row[0] if row else 0 + + def 当前系统版本(self, work_id: str) -> int: + row = self.连.execute( + "SELECT system_revision FROM muse_world_clock WHERE work_id=%s", (work_id,) + ).fetchone() + return row[0] if row else 0 + + def 占系统版本(self, work_id: str) -> int: + return self.连.execute( + "INSERT INTO muse_world_clock(work_id,system_revision) VALUES (%s,1) " + "ON CONFLICT(work_id) DO UPDATE SET system_revision=muse_world_clock.system_revision+1 " + "RETURNING system_revision", + (work_id,), + ).fetchone()[0] + + def 写对象( + self, 作者: str, 请求, 类型ID: str, 内容: dict, 绑定, 系统版本: int, 内容哈希: str + ) -> dict: + version = 请求.expected_revision + 1 + if 请求.expected_revision == 0: + self.连.execute( + "INSERT INTO muse_world_object" + "(object_id,work_id,type_id,instance_family,current_revision) " + "VALUES (%s,%s,%s,%s,%s)", + (请求.object_id, 请求.work_id, 类型ID, 请求.instance_family, version), + ) + else: + row = self.连.execute( + "UPDATE muse_world_object SET current_revision=%s WHERE object_id=%s " + "AND current_revision=%s RETURNING object_id", + (version, 请求.object_id, 请求.expected_revision), + ).fetchone() + if row is None: + raise 事实错误("REVISION_CONFLICT", "事实对象版本已改变") + self.连.execute( + "INSERT INTO muse_world_version" + "(object_id,revision,system_revision,content,content_hash," + "schema_binding,sources,independent,effective_from,knowledge_mode,known_to," + "reader_visible,author_id,planned_window) " + "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", + ( + 请求.object_id, + version, + 系统版本, + Jsonb(内容), + 内容哈希, + Jsonb(asdict(绑定)), + Jsonb([asdict(s) for s in 请求.sources]), + 请求.independent, + 请求.effective_from, + 请求.knowledge_mode, + Jsonb([asdict(k) for k in 请求.known_to]), + 请求.reader_visible, + 作者, + Jsonb(asdict(请求.planned_window)) if 请求.planned_window is not None else None, + ), + ) + return { + "object_id": 请求.object_id, + "revision": version, + "system_revision": 系统版本, + "content_hash": 内容哈希, + } + + def 读取时点对象(self, work_id: str, object_id: str, 章序: dict[str, int], 截止: int): + return 选择时点记录( + self.读取历史(work_id, self.当前系统版本(work_id), object_id), 章序, 截止 + ).get(object_id) + + def 读取历史(self, work_id: str, 系统版本: int, object_id: str | None = None) -> list[dict]: + return ( + self.连.cursor(row_factory=dict_row) + .execute( + "SELECT o.object_id,o.work_id,o.type_id,o.instance_family,o.current_revision,v.* " + "FROM muse_world_object o JOIN muse_world_version v ON v.object_id=o.object_id " + "WHERE o.work_id=%s AND v.system_revision<=%s " + "AND (%s::text IS NULL OR o.object_id=%s) " + "ORDER BY o.object_id,v.system_revision DESC", + (work_id, 系统版本, object_id, object_id), + ) + .fetchall() + ) diff --git a/src/muse/故事世界/接口.py b/src/muse/故事世界/接口.py new file mode 100644 index 0000000..b96f66f --- /dev/null +++ b/src/muse/故事世界/接口.py @@ -0,0 +1,247 @@ +"""故事世界唯一公开表面；当前提案依据校验，后续同表面接入确认和查询。""" + +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 故事存储 +from muse.故事世界.时点查询 import 查询世界状态 +from muse.故事世界.来源有效性 import 当前依据可用, 读取依据状态 +from muse.故事世界.模型 import ( + 事实定位, + 事实提案请求, + 事实条目, + 事实结构, + 事实错误, + 叙事窗口, + 引文锚点, + 引用对象, + 正文依据, + 知情记录, +) +from muse.故事世界.设定修订 import ( + 世界参与者, + 世界对象依赖, + 世界提案依赖, + 事实候选入口, + 事实决定入口, + 事实授权, + 提案保存, + 提案参与者, + 核对提案归属, + 读取事实结构, +) +from muse.正式变更.接口 import ( + 作者动作, + 依赖引用, + 参与者目录, + 变更命令, + 固定哈希, + 正式变更服务, + 读取作者决定, +) + + +def 登记故事参与者(目录: 参与者目录) -> None: + 目录.登记入口(事实候选入口()) + 目录.登记入口(事实决定入口()) + 目录.登记参与者(提案参与者()) + 目录.登记参与者(世界参与者()) + 目录.登记依赖(世界对象依赖()) + 目录.登记依赖(世界提案依赖()) + + +class 故事服务: + def __init__(self, 数据库: 数据库工厂, 正式: 正式变更服务): + self.数据库, self.正式 = 数据库, 正式 + + def _核对作者(self, 身份): + if not 身份.允许写正式内容 or 身份.用途 is not self.数据库.用途: + raise 事实错误("SCOPE_DENIED", "作者事实入口需要本用途的已认证作者") + + def 新事实结构( + self, + 身份, + work_id: str, + object_id: str, + schema_id: str, + 版本: int, + *, + 家族: str = "entity", + 扩展版本: int | None = None, + ) -> 事实结构: + self._核对作者(身份) + with self.数据库.连接(只读=True) as 连: + 读取作品范围(连, 身份.作者, work_id) + 结构, _, 投影 = 读取事实结构( + 连, 身份.作者, work_id, 事实结构(schema_id, 版本, "", "", 扩展版本), 家族 + ) + return 事实结构( + schema_id, 版本, 结构.effective_schema_hash, 投影.projection_version, 扩展版本 + ) + + def 提出事实(self, 身份, 命令ID: str, 请求: 事实提案请求) -> dict: + self._核对作者(身份) + id_ = str(uuid5(NAMESPACE_URL, "muse:B02:proposal:" + 固定哈希([身份.作者, 命令ID]))) + return self.正式.提交变更( + 身份, + 变更命令(命令ID, "B02.propose_fact", id_, 作者动作.人工保存, 0, 提案保存(id_, 请求)), + ) + + def 读取提案(self, 身份, proposal_id: str) -> dict: + self._核对作者(身份) + with self.数据库.连接(只读=True) as 连: + 行, 请求 = 核对提案归属(连, 身份.作者, proposal_id) + 结构, 类型, _ = 读取事实结构( + 连, 身份.作者, 请求.work_id, 请求.schema, 请求.instance_family + ) + 投影 = 投影字段( + 结构, + 元数据服务(连).当前策略(类型.type_id), + 事实授权(身份.作者, 请求.work_id), + 字段用途名="uiVisible", + 内容用途="planning", + 运行用途="production", + ) + return { + "proposal_id": str(行["proposal_id"]), + "revision": 行["revision"], + "proposal_hash": 行["proposal_hash"], + "work_id": 请求.work_id, + "object_id": 请求.object_id, + "expected_revision": 请求.expected_revision, + "schema": asdict(请求.schema), + "content": 投影值(投影, 行["proposed_content"]), + "independent": 请求.independent, + "sources": [asdict(s) for s in 请求.sources], + "effective_from": 请求.effective_from, + "knowledge_mode": 请求.knowledge_mode, + "known_to": [asdict(k) for k in 请求.known_to], + "planned_window": asdict(请求.planned_window) if 请求.planned_window else None, + "decision": 读取作者决定(连, proposal_id, 行["revision"]), + } + + def 打开事实审阅(self, 身份, 定位: 事实定位) -> dict: + self._核对作者(身份) + 审阅 = self.正式.打开审阅( + 身份, + 变更命令( + "review", + "B02.decide_fact", + 定位.target_ref, + 作者动作.采纳, + 定位.expected_revision, + 定位, + ), + ) + return {**审阅, "proposal": self.读取提案(身份, 定位.proposal_id)} + + def 决定事实( + self, + 身份, + 命令ID: str, + 定位: 事实定位, + 动作: 作者动作, + 审阅ID: str, + 审阅哈希: str, + 批准: tuple[str, ...], + ) -> dict: + self._核对作者(身份) + return self.正式.提交变更( + 身份, + 变更命令( + 命令ID, + "B02.decide_fact", + 定位.target_ref, + 动作, + 定位.expected_revision, + 定位, + 审阅ID, + 审阅哈希, + 批准, + ), + ) + + def 查询世界( + self, + 身份, + work_id: str, + 截止章节: str | None, + *, + 系统版本: int | None = None, + 视角: str = "author", + 知情主体ID: str | None = None, + ) -> dict: + self._核对作者(身份) + with self.数据库.连接(只读=True) as 连: + return 查询世界状态( + 连, + 身份.作者, + work_id, + 截止章节, + 系统版本=系统版本, + 视角=视角, + 知情主体ID=知情主体ID, + ) + + +__all__ = [ + "事实条目", + "事实错误", + "叙事窗口", + "引用对象", + "引文锚点", + "定位字面引文", + "保留名称提及", + "核对原文引文", + "解析类型选择", + "校验事实条目", + "校验事实批次", + "核对提案身份", + "事实定位", + "事实提案请求", + "事实结构", + "正文依据", + "知情记录", + "故事服务", + "登记故事参与者", + "解析本书规划引用", +] + + +def 解析本书规划引用(连, 作者: str, work_id: str, 需求, *, 必须有效: bool) -> tuple: + """规划只取得本书对象身份和必要依赖，不越权读取参考实例或自行确认实体。""" + 读取作品范围(连, 作者, work_id) + deps = [] + for value in 需求.值 if isinstance(需求.值, list) else [需求.值]: + id_ = value.get("id") if isinstance(value, dict) else value + obj = 故事存储(连).读取对象(id_) if isinstance(id_, str) else None + if obj is None or obj["work_id"] != work_id or obj["type_id"] not in 需求.规则.目标类型: + raise 事实错误("FACT_REFERENCE_INVALID", "规划引用对象缺失、类型不符或超出本书") + if isinstance(value, dict) and value.get("type") != obj["type_id"]: + raise 事实错误("FACT_REFERENCE_INVALID", "规划引用自报类型与真实本书对象不符") + if 必须有效 and not 当前依据可用(连, 作者, obj): + raise 事实错误("SOURCE_STALE", "规划引用的本书对象已失去当前依据") + deps.append( + 依赖引用("B02.object", "world:" + obj["object_id"], str(obj["current_revision"])) + ) + if not obj["independent"]: + for raw in obj["sources"]: + source = 读取依据状态(连, 作者, work_id, TypeAdapter(正文依据).validate_python(raw)) + deps.append( + 依赖引用("B05.document", source["document_id"], str(source["current_revision"])) + ) + return tuple(sorted(set(deps))) diff --git a/src/muse/故事世界/时点查询.py b/src/muse/故事世界/时点查询.py new file mode 100644 index 0000000..a0229f0 --- /dev/null +++ b/src/muse/故事世界/时点查询.py @@ -0,0 +1,117 @@ +"""事件生效、披露时点与确认版本分开；不改变事实的保存状态。""" + +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 查询世界状态( + 连, + 作者: str, + work_id: str, + 截止章节: str | None, + *, + 系统版本: int | None = None, + 视角: str = "author", + 知情主体ID: str | None = None, +) -> dict: + if 视角 not in {"author", "reader", "character"} or ((视角 == "character") != bool(知情主体ID)): + raise 事实错误("FACT_SCOPE_INVALID", "查询需要明确的作者、读者或角色视角") + 目录 = 读取作品范围(连, 作者, work_id) + 章序 = {c["chapter_id"]: c["position"] for c in 目录["chapters"]} + if 截止章节 is not None and 截止章节 not in 章序: + raise 事实错误("FACT_TIME_INVALID", "查询截止位置不属于本书") + 截止 = 章序[截止章节] if 截止章节 is not None else max(章序.values(), default=0) + 存储 = 故事存储(连) + 当前系统版 = 存储.当前系统版本(work_id) + 系统版 = 当前系统版 if 系统版本 is None else 系统版本 + if 系统版 < 0 or 系统版 > 当前系统版: + raise 事实错误("FACT_TIME_INVALID", "请求的事实系统版本不存在") + if 知情主体ID: + 主体 = 选择时点记录(存储.读取历史(work_id, 系统版, 知情主体ID), 章序, 截止).get(知情主体ID) + if ( + 主体 is None + or 主体["instance_family"] != "entity" + or 主体["knowledge_mode"] not in {"fact", "setting"} + or not 当前依据可用(连, 作者, 主体) + ): + raise 事实错误("SCOPE_DENIED", "知情主体在查询时点不可用") + 结果 = { + "work_id": work_id, + "directory_revision": 目录["revision"], + "system_revision": 系统版, + "facts": [], + "beliefs": [], + "plans": [], + "omitted": [], + } + 元数据 = 元数据服务(连) + for 行 in 选择时点记录(存储.读取历史(work_id, 系统版), 章序, 截止).values(): + id_ = 行["object_id"] + 原因 = None + if 固定哈希(行["content"]) != 行["content_hash"]: + raise 事实错误("FACT_INVALID", "已保存事实与内容哈希不一致") + if not 当前依据可用(连, 作者, 行): + 原因 = "SOURCE_STALE" + elif 视角 == "reader" and not 行["reader_visible"]: + 原因 = "KNOWLEDGE_SCOPE" + elif 视角 == "character" and not any( + k["actor_id"] == 知情主体ID and 章序.get(k["from_chapter_id"], 截止 + 1) <= 截止 + for k in 行["known_to"] + ): + 原因 = "KNOWLEDGE_SCOPE" + elif 视角 == "reader" and not 行["independent"]: + 已披露 = False + for 源值 in 行["sources"]: + 源 = TypeAdapter(正文依据).validate_python(源值) + if 章序[源.chapter_id] <= 截止 and 读取依据状态(连, 作者, work_id, 源)["current"]: + 已披露 = True + if not 已披露: + 原因 = "FUTURE_EVIDENCE" + if 原因: + 结果["omitted"].append({"object_id": id_, "reason": 原因}) + continue + binding = 结构绑定(**行["schema_binding"]) + 结构 = 元数据.读取有效结构( + binding.schema_id, + binding.base_version, + work_id=work_id, + extension_version=binding.extension_version, + ) + 投影 = 投影字段( + 结构, + 元数据.当前策略(行["type_id"]), + 事实授权(作者, work_id), + 字段用途名="uiVisible", + 内容用途="planning", + 运行用途="production", + ) + item = { + "object_id": id_, + "type_id": 行["type_id"], + "revision": 行["revision"], + "system_revision": 行["system_revision"], + "content": 投影值(投影, 行["content"]), + "effective_from": 行["effective_from"], + "knowledge_mode": 行["knowledge_mode"], + "independent": 行["independent"], + "content_hash": 行["content_hash"], + "projection_version": 投影.projection_version, + "schema_binding": 行["schema_binding"], + "sources": [ + {k: s[k] for k in ("chapter_id", "branch_id", "revision", "document_hash")} + for s in 行["sources"] + ], + } + if 视角 == "author": + item["planned_window"] = 行["planned_window"] + key = {"belief": "beliefs", "plan": "plans"}.get(行["knowledge_mode"], "facts") + 结果[key].append(item) + return 结果 diff --git a/src/muse/故事世界/来源有效性.py b/src/muse/故事世界/来源有效性.py new file mode 100644 index 0000000..eab87da --- /dev/null +++ b/src/muse/故事世界/来源有效性.py @@ -0,0 +1,34 @@ +"""当前正文依据即时核对；独立作者依据与其他充分依据分开计算。""" + +from pydantic import TypeAdapter + +from muse.故事世界.事实提案 import 核对原文引文 +from muse.故事世界.模型 import 事实错误, 正文依据 +from muse.正文写作.接口 import 读取正文依据 + + +def 读取依据状态(连, 作者: str, work_id: str, 来源: 正文依据) -> dict: + 原 = 读取正文依据(连, 作者, 来源.chapter_id, 来源.revision, 分支=来源.branch_id) + if 原["work_id"] != work_id: + raise 事实错误("SCOPE_DENIED", "事实依据不属于当前作品") + if 原["document_hash"] != 来源.document_hash: + raise 事实错误("SOURCE_STALE", "正文依据哈希与固定版本不一致") + for 引文 in 来源.evidence: + 核对原文引文(原["document"], 引文) + return { + "document_id": 原["document_id"], + "chapter_id": 来源.chapter_id, + "current_revision": 原["current_revision"], + "source_revision": 来源.revision, + "current": 原["current_revision"] == 来源.revision and 来源.branch_id == "main", + } + + +def 当前依据可用(连, 作者: str, 记录: dict) -> bool: + if 记录["independent"]: + return True + for 原 in 记录["sources"]: + 来源 = TypeAdapter(正文依据).validate_python(原) + if 读取依据状态(连, 作者, 记录["work_id"], 来源)["current"]: + return True + return False diff --git a/src/muse/故事世界/模型.py b/src/muse/故事世界/模型.py new file mode 100644 index 0000000..eb658e8 --- /dev/null +++ b/src/muse/故事世界/模型.py @@ -0,0 +1,123 @@ +"""故事事实的身份与证据外壳；动态内容由S04结构描述。""" + +from typing import Literal + +from pydantic import ConfigDict +from pydantic.dataclasses import dataclass + +from muse.共享.错误 import Muse错误 + + +class 事实错误(Muse错误): + def __init__(self, 错误码: str, 说明: str): + self.错误码 = 错误码 + super().__init__(说明) + + +@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid")) +class 引文锚点: + paragraph_id: str + start: int + end: int + quote: str + + def __post_init__(self): + if ( + not self.paragraph_id + or not self.quote + or self.start < 0 + or self.end - self.start != len(self.quote) + ): + raise 事实错误("FACT_QUOTE_INVALID", "引文需要段落身份、实际文字和一致的码点区间") + + +@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid")) +class 叙事窗口: + from_chapter_id: str + to_chapter_id: str + + +@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid")) +class 事实条目: + proposal_id: str + object_id: str + type_id: str + content: dict + evidence: tuple[引文锚点, ...] + instance_family: Literal["entity", "relation"] = "entity" + planned_window: 叙事窗口 | None = None + + def __post_init__(self): + if not self.proposal_id or not self.object_id or not self.type_id or not self.evidence: + raise 事实错误("FACT_INVALID", "事实提案需要稳定身份、登记类型与实际正文依据") + + +@dataclass(frozen=True, slots=True) +class 引用对象: + object_id: str + work_id: str + type_id: str + state: Literal["confirmed", "approved_in_batch", "unconfirmed"] = "confirmed" + available: bool = True + + +@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid")) +class 事实结构: + schema_id: str + base_version: int + effective_schema_hash: str + projection_version: str + extension_version: int | None = None + + +@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid")) +class 正文依据: + chapter_id: str + branch_id: str + revision: int + document_hash: str + evidence: tuple[引文锚点, ...] + + def __post_init__(self): + if self.revision < 1 or not self.document_hash or not self.evidence: + raise 事实错误("FACT_INVALID", "正文依据需要确切的已保存版本、哈希和引文") + + +@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid")) +class 知情记录: + actor_id: str + from_chapter_id: str + + +@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid")) +class 事实提案请求: + work_id: str + object_id: str + expected_revision: int + schema: 事实结构 + content: dict + sources: tuple[正文依据, ...] + effective_from: str | None + instance_family: Literal["entity", "relation"] = "entity" + independent: bool = False + knowledge_mode: Literal["fact", "belief", "plan", "setting"] = "fact" + known_to: tuple[知情记录, ...] = () + reader_visible: bool = True + planned_window: 叙事窗口 | None = None + + def __post_init__(self): + if not self.work_id or not self.object_id or self.expected_revision < 0: + raise 事实错误("FACT_INVALID", "事实提案需要作品、对象和预期版本") + if not self.independent and not self.sources: + raise 事实错误("FACT_INVALID", "抽取事实需要正文依据；独立设定必须由作者明确声明") + if len({k.actor_id for k in self.known_to}) != len(self.known_to): + raise 事实错误("FACT_INVALID", "同一事实版本的角色获知位置不能重复") + + +@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid")) +class 事实定位: + proposal_id: str + proposal_revision: int + proposal_hash: str + target_ref: str + expected_revision: int diff --git a/src/muse/故事世界/设定修订.py b/src/muse/故事世界/设定修订.py new file mode 100644 index 0000000..ec0a610 --- /dev/null +++ b/src/muse/故事世界/设定修订.py @@ -0,0 +1,322 @@ +"""S01中的B02事实写入；结构与来源均由所属公开接口核对。""" + +from dataclasses import asdict, dataclass + +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 作者动作, 依赖引用, 修改影响, 参与操作, 固定哈希, 审阅内容, 提交计划 + + +def 事实授权(作者: str, work_id: str) -> 授权摘要: + return 授权摘要(固定哈希({"author": 作者}), 固定哈希({"work": work_id, "owner": "B02"})) + + +def 读取事实结构(连, 作者: str, work_id: str, 选择: 事实结构, 家族: str): + 元数据 = 元数据服务(连) + 结构 = 元数据.读取有效结构( + 选择.schema_id, 选择.base_version, work_id=work_id, extension_version=选择.extension_version + ) + 类型 = 元数据.读取类型(结构.基础.type_id) + if 选择实例维护者(类型, 家族) != "B02": + raise 事实错误("SCOPE_DENIED", "结构不属于本书事实生命周期") + 投影 = 投影字段( + 结构, + 元数据.当前策略(类型.type_id), + 事实授权(作者, work_id), + 字段用途名="userEditable", + 内容用途="planning", + 运行用途="production", + ) + return 结构, 类型, 投影 + + +def 核验提案内容(连, 作者: str, 请求: 事实提案请求): + 目录 = 读取作品范围(连, 作者, 请求.work_id) + 章序 = {c["chapter_id"]: c["position"] for c in 目录["chapters"]} + if 请求.effective_from is not None and 请求.effective_from not in 章序: + raise 事实错误("FACT_TIME_INVALID", "事实生效位置不属于本书目录") + if 请求.effective_from is None and not 请求.independent: + raise 事实错误("FACT_TIME_INVALID", "正文事实需要明确的生效章节") + if 请求.planned_window is not None: + 核对叙事窗口(请求.planned_window, 章序) + 存储 = 故事存储(连) + 当前 = 存储.读取对象(请求.object_id) + if 当前 is not None and 当前["work_id"] != 请求.work_id: + raise 事实错误("SCOPE_DENIED", "事实对象不属于本书") + if (当前["current_revision"] if 当前 else 0) != 请求.expected_revision: + raise 事实错误("REVISION_CONFLICT", "事实对象已被其他决定更新") + 结构, 类型, 投影 = 读取事实结构(连, 作者, 请求.work_id, 请求.schema, 请求.instance_family) + if 结构.effective_schema_hash != 请求.schema.effective_schema_hash: + raise 事实错误("SCHEMA_STALE", "事实结构与请求的固定版本不一致") + if 投影.projection_version != 请求.schema.projection_version: + raise 事实错误("PROJECTION_STALE", "事实可编辑字段已经变化") + if 当前 is not None and (当前["type_id"], 当前["instance_family"]) != ( + 类型.type_id, + 请求.instance_family, + ): + raise 事实错误("SCOPE_DENIED", "现有对象不能被提案改为其他类型或生命周期") + 值 = 校验字段变更( + 结构, + 投影, + 当前["content"] if 当前 else {}, + tuple(字段变更((key,), "set", val) for key, val in 请求.content.items()), + ) + 依据 = [读取依据状态(连, 作者, 请求.work_id, 源) for 源 in 请求.sources] + if not 请求.independent and not any(s["current"] for s in 依据): + raise 事实错误("SOURCE_STALE", "事实已失去当前正文依据") + # 回忆可在后章揭示早期事件；证据何时可见由时点查询独立控制。 + 引用版本 = {} + 截止 = 章序[请求.effective_from] if 请求.effective_from else 0 + for 需求 in 值.引用需求: + for id_ in 需求.值 if isinstance(需求.值, list) else [需求.值]: + 对象 = ( + 存储.读取时点对象(请求.work_id, id_, 章序, 截止) if isinstance(id_, str) else None + ) + if ( + 对象 is None + or 对象["work_id"] != 请求.work_id + or 对象["type_id"] not in 需求.规则.目标类型 + or 请求.knowledge_mode in {"fact", "setting"} + and 对象["knowledge_mode"] not in {"fact", "setting"} + ): + raise 事实错误( + "FACT_REFERENCE_INVALID", "字段引用在该时点不可用、类型不符或不是已发生事实" + ) + if not 当前依据可用(连, 作者, 对象): + raise 事实错误("FACT_REFERENCE_INVALID", "引用对象已失去当前依据") + 引用版本[id_] = 对象["current_revision"] + 引用数 = sum(len(v.值) if isinstance(v.值, list) else 1 for v in 值.引用需求) + if 请求.instance_family == "relation" and 引用数 < 2: + raise 事实错误("FACT_REFERENCE_INVALID", "关系结构必须提供真实端点") + for 知情 in 请求.known_to: + if 知情.from_chapter_id not in 章序: + raise 事实错误("FACT_TIME_INVALID", "角色获知位置必须属于本书目录") + 对象 = 存储.读取时点对象(请求.work_id, 知情.actor_id, 章序, 章序[知情.from_chapter_id]) + if ( + 对象 is None + or 对象["instance_family"] != "entity" + or 对象["knowledge_mode"] not in {"fact", "setting"} + or not 当前依据可用(连, 作者, 对象) + ): + raise 事实错误("FACT_REFERENCE_INVALID", "知情主体必须是获知时点内有效的本书实体") + 引用版本[知情.actor_id] = 对象["current_revision"] + 引用来源 = 读取引用依赖(连, 作者, 请求.work_id, 引用版本, 章序, 截止, 请求.known_to)[1] + return 值.值, 结构, 类型, 投影, ([] if 请求.independent else 依据) + 引用来源, 引用版本, 目录 + + +def 读取引用依赖(连, 作者: str, work_id: str, 对象ID集, 章序: dict[str, int], 截止: int, 知情范围): + """引用内容的有效性也依赖其正文；S01需共同保护这些当前指针。""" + 版本 = {} + 来源 = [] + for id_ in 对象ID集: + 对象 = 故事存储(连).读取对象(id_) + 版本[id_] = 对象["current_revision"] if 对象 else 0 + if 对象 is not None: + if 对象["work_id"] != work_id: + raise 事实错误("SCOPE_DENIED", "引用对象不属于本书") + 时点集 = {截止} | {章序[k.from_chapter_id] for k in 知情范围 if k.actor_id == id_} + for 时点 in sorted(时点集): + 时点对象 = 故事存储(连).读取时点对象(work_id, id_, 章序, 时点) + if 时点对象 is not None and not 时点对象["independent"]: + for 原 in 时点对象["sources"]: + 来源.append( + 读取依据状态( + 连, 作者, work_id, TypeAdapter(正文依据).validate_python(原) + ) + ) + return 版本, 来源 + + +@dataclass(frozen=True) +class 世界写入: + 作者: str + 请求: 事实提案请求 + + +class 世界参与者: + 身份 = "B02.write_object" + 请求类型 = 世界写入 + + def 提交(self, 连, 操作): + 请求 = 操作.请求 + 内容, 结构, 类型, _, _, _, _ = 核验提案内容(连, 操作.作者, 请求) + 元数据 = 元数据服务(连) + target = "world:" + 请求.object_id + if 故事存储(连).读取对象(请求.object_id) is None: + 绑定 = 元数据.建立初始绑定(target, 请求.work_id, 结构) + else: + 绑定 = 元数据.读取绑定(target) + 元数据.保护提交( + target, + 请求.schema.effective_schema_hash, + 请求.schema.projection_version, + 事实授权(操作.作者, 请求.work_id), + 字段用途名="userEditable", + 内容用途="planning", + 运行用途="production", + ) + 存储 = 故事存储(连) + 系统版本 = 存储.占系统版本(请求.work_id) + return 存储.写对象(操作.作者, 请求, 类型.type_id, 内容, 绑定, 系统版本, 固定哈希(内容)) + + +@dataclass(frozen=True) +class 提案保存: + proposal_id: str + 请求: 事实提案请求 + + +@dataclass(frozen=True) +class 提案写入: + 作者: str + 输入: 提案保存 + + +def 核对提案归属(连, 作者: str, proposal_id: str): + 行 = 故事存储(连).读取提案(proposal_id) + if 行 is None: + raise 事实错误("FACT_PROPOSAL_NOT_FOUND", "事实提案不存在") + 读取作品范围(连, 作者, 行["work_id"]) + 请求 = TypeAdapter(事实提案请求).validate_python(行["request"]) + if 固定哈希([行["request"], 行["proposed_content"]]) != 行["proposal_hash"]: + raise 事实错误("FACT_INVALID", "已保存提案与内容哈希不一致") + return 行, 请求 + + +def 形成依赖( + 请求, 目录, 依据, 引用版本, proposal_id: str, proposal_revision: int, object_revision: int +): + 依赖 = { + ("B01.directory", "directory:" + 请求.work_id): str(目录["revision"]), + ("B02.object", "world:" + 请求.object_id): str(object_revision), + ("B02.proposal", proposal_id): str(proposal_revision), + } + for 源 in 依据: + 依赖[("B05.document", 源["document_id"])] = str(源["current_revision"]) + for id_, 版 in 引用版本.items(): + 依赖[("B02.object", "world:" + id_)] = str(版) + return tuple(依赖引用(类, id_, 版) for (类, id_), 版 in sorted(依赖.items())) + + +class 事实候选入口: + 身份 = "B02.propose_fact" + 请求类型 = 提案保存 + + def 查看(self, 连, 身份, 命令): + 输入 = 命令.业务请求 + 请求 = 输入.请求 + 内容, 结构, _, 投影, 依据, 引用, 目录 = 核验提案内容(连, 身份.作者, 请求) + 原 = 故事存储(连).读取提案(输入.proposal_id) + 版 = 原["current_revision"] if 原 else 0 + return 审阅内容( + 输入.proposal_id, + 版, + None, + None, + None, + 固定哈希([asdict(请求), 内容]), + 结构.effective_schema_hash, + 投影.projection_version, + 固定哈希(依据), + ("proposal",), + 形成依赖(请求, 目录, 依据, 引用, 输入.proposal_id, 版, 请求.expected_revision), + ) + + def 准备(self, 连, 身份, 命令, 当前): + return 提交计划((参与操作("B02.write_proposal", 提案写入(身份.作者, 命令.业务请求)),)) + + +class 提案参与者: + 身份 = "B02.write_proposal" + 请求类型 = 提案写入 + + def 提交(self, 连, 操作): + 请求 = 操作.输入.请求 + 内容, *_ = 核验提案内容(连, 操作.作者, 请求) + return 故事存储(连).写提案( + 操作.输入.proposal_id, 请求, 内容, 固定哈希([asdict(请求), 内容]) + ) + + +class 事实决定入口: + 身份 = "B02.decide_fact" + 请求类型 = 事实定位 + + def 查看(self, 连, 身份, 命令): + 定位 = 命令.业务请求 + 行, 请求 = 核对提案归属(连, 身份.作者, 定位.proposal_id) + if (行["revision"], 行["proposal_hash"]) != (定位.proposal_revision, 定位.proposal_hash): + raise 事实错误("REVISION_CONFLICT", "提案版本与被审内容不一致") + target = "world:" + 请求.object_id + if 定位.target_ref != target: + raise 事实错误("SCOPE_DENIED", "提案与事实对象不一致") + 对象 = 故事存储(连).读取对象(请求.object_id) + if 对象 is not None and 对象["work_id"] != 请求.work_id: + raise 事实错误("SCOPE_DENIED", "事实对象不属于本书") + 版 = 对象["current_revision"] if 对象 else 0 + 目录 = 读取作品范围(连, 身份.作者, 请求.work_id) + 依据 = [读取依据状态(连, 身份.作者, 请求.work_id, s) for s in 请求.sources] + 结构, _, 投影 = 读取事实结构(连, 身份.作者, 请求.work_id, 请求.schema, 请求.instance_family) + # 依赖由已保存提案的完整规范值识别，不相信客户端自报预检。 + from muse.元数据.接口 import 校验值 + + 引用 = {} + for 需 in 校验值(结构, 行["proposed_content"]).引用需求: + for id_ in 需.值 if isinstance(需.值, list) else [需.值]: + obj = 故事存储(连).读取对象(id_) if isinstance(id_, str) else None + if isinstance(id_, str): + 引用[id_] = obj["current_revision"] if obj else 0 + for 知情 in 请求.known_to: + obj = 故事存储(连).读取对象(知情.actor_id) + 引用[知情.actor_id] = obj["current_revision"] if obj else 0 + 章序 = {c["chapter_id"]: c["position"] for c in 目录["chapters"]} + 截止 = 章序[请求.effective_from] if 请求.effective_from else 0 + 引用, 其他依据 = 读取引用依赖(连, 身份.作者, 请求.work_id, 引用, 章序, 截止, 请求.known_to) + 依据 = ([] if 请求.independent else 依据) + 其他依据 + return 审阅内容( + target, + 版, + 定位.proposal_id, + 行["revision"], + 行["proposal_hash"], + 固定哈希([对象["content_hash"] if 对象 else None, 行["proposed_content"]]), + 结构.effective_schema_hash, + 投影.projection_version, + 固定哈希([依据, 引用, 目录]), + ("object",), + 形成依赖(请求, 目录, 依据, 引用, 定位.proposal_id, 行["revision"], 版), + ) + + def 准备(self, 连, 身份, 命令, 当前): + if 命令.动作 in {作者动作.拒绝, 作者动作.暂缓}: + return 提交计划(()) + if 命令.动作 is not 作者动作.采纳: + raise 事实错误("FACT_INVALID", "事实提案需要明确的作者决定") + _, 请求 = 核对提案归属(连, 身份.作者, 命令.业务请求.proposal_id) + 核验提案内容(连, 身份.作者, 请求) + return 提交计划( + (参与操作("B02.write_object", 世界写入(身份.作者, 请求)),), + (修改影响(当前.目标, "fact_changed", str(当前.数据版本), str(当前.数据版本 + 1)),), + ) + + +class 世界对象依赖: + 种类 = "B02.object" + + def 锁定当前(self, 连, 对象ID): + if not 对象ID.startswith("world:"): + raise 事实错误("SCOPE_DENIED", "不是故事世界对象身份") + return str(故事存储(连).锁对象(对象ID.removeprefix("world:"))) + + +class 世界提案依赖: + 种类 = "B02.proposal" + + def 锁定当前(self, 连, 对象ID): + return str(故事存储(连).锁提案(对象ID)) diff --git a/src/muse/故事世界/连续性.py b/src/muse/故事世界/连续性.py new file mode 100644 index 0000000..8138e05 --- /dev/null +++ b/src/muse/故事世界/连续性.py @@ -0,0 +1,20 @@ +"""故事时间约束使用稳定章节身份和指定目录版本的顺序。""" + +from muse.故事世界.模型 import 事实错误, 叙事窗口 + + +def 核对叙事窗口(窗口: 叙事窗口, 章节顺序: dict[str, int]) -> None: + 起 = 章节顺序.get(窗口.from_chapter_id) + 止 = 章节顺序.get(窗口.to_chapter_id) + if 起 is None or 止 is None or 起 > 止: + raise 事实错误("FACT_TIME_INVALID", "计划窗口需要本书存在且顺序正确的起止章节") + + +def 选择时点记录(记录: list[dict], 章节顺序: dict[str, int], 截止: int) -> dict[str, dict]: + """记录按系统版本倒序；每个对象取叙事时点内最新的已确认版本。""" + 结果 = {} + for 行 in 记录: + 起点 = 章节顺序.get(行["effective_from"]) if 行["effective_from"] else 0 + if 起点 is not None and 起点 <= 截止: + 结果.setdefault(行["object_id"], 行) + return 结果 diff --git a/tests/单元/test_事实时点与有效性.py b/tests/单元/test_事实时点与有效性.py new file mode 100644 index 0000000..fc0de4e --- /dev/null +++ b/tests/单元/test_事实时点与有效性.py @@ -0,0 +1,331 @@ +"""逐条承接引文行为；不执行旧Skill脚本、不规范化正文。""" + +from dataclasses import asdict, replace + +import pytest +from pydantic import TypeAdapter, ValidationError + +from muse.元数据.接口 import ( + 值规则, + 元数据错误, + 合成结构, + 字段定义, + 引用规则, + 类型定义, + 结构定义, + 读取内置种子, +) +from muse.故事世界.接口 import ( + 事实条目, + 事实错误, + 保留名称提及, + 叙事窗口, + 定位字面引文, + 引用对象, + 校验事实批次, + 核对原文引文, + 解析类型选择, +) +from muse.正文写作.接口 import 换行节点, 文本节点, 正文草稿, 段落 + +原增量正文 = "林深把黑纹缠上手臂，异种核心在胸腔里低鸣。何岚站在舱门口没有说话。" +原抽取正文 = "林深走进舰桥，何岚把黑色钥匙交给他。" + + +def 单段(内容): + return 正文草稿((段落("p1", (文本节点(内容),)),)) + + +def test_quote_not_found_fails_closed__3554e5(): + with pytest.raises(事实错误) as 错误: + 定位字面引文(单段(原增量正文), "正文里不存在的句子") + assert 错误.value.错误码 == "FACT_QUOTE_NOT_FOUND" + + +def test_quote_is_normalized_before_matching__505864(): + # 保留原LC-e3eca2942895样本；格式1不允许把字面反斜杠改成真正换行。 + escaped_body = 原增量正文 + "\\n舱灯闪了一下" + quote = "\n舱灯闪了一下" + with pytest.raises(事实错误) as 错误: + 定位字面引文(单段(escaped_body), quote) + assert 错误.value.错误码 == "FACT_QUOTE_NOT_FOUND" + assert escaped_body.endswith("\\n舱灯闪了一下") + 原文 = 正文草稿((段落("p1", (文本节点(原增量正文), 换行节点(), 文本节点("舱灯闪了一下"))),)) + 证据 = 定位字面引文(原文, quote) + assert 核对原文引文(原文, 证据) == quote + assert 证据.start == len(原增量正文) and 证据.end == len(原增量正文 + quote) + + +def test_normalize_rejects_fabricated_evidence__9e1858(): + 原输入 = { + "type": "item", + "name": "黑钥匙", + "brief": "钥匙", + "fields": {}, + "evidence": "正文没有这句", + } + with pytest.raises(事实错误) as 错误: + 定位字面引文(单段(原抽取正文), 原输入["evidence"]) + assert 错误.value.错误码 == "FACT_QUOTE_NOT_FOUND" + + +def test_normalize_binds_evidence_and_aliases__1b9888(): + 原输入 = { + "type": "人物", + "name": "林深", + "brief": "机师", + "fields": {}, + "evidence": "林深走进舰桥", + } + 种子 = next(s for s in 读取内置种子() if s.类型.type_id == "character") + # 此旧场景中的中文类型名在输入目录登记；解析代码没有固定中文别名表。 + 类型 = replace(种子.类型, 名称="人物") + 选中 = 解析类型选择(原输入["type"], (类型,)) + 正文 = 单段(原抽取正文) + 提案 = 事实条目( + "proposal-lin", + "character-lin", + 选中.type_id, + {"名称": 原输入["name"], "一句话摘要": 原输入["brief"], **原输入["fields"]}, + (定位字面引文(正文, 原输入["evidence"]),), + ) + 结果 = 校验事实批次( + (提案,), + 作品ID="synthetic-work", + 结构目录={"character": (类型, 合成结构(种子.结构))}, + 正文=正文, + 可用对象={}, + ) + assert 结果[0].type_id == "character" + assert 结果[0].evidence[0].quote in 原抽取正文 + assert 结果[0].content["名称"] == "林深" + + +旧增量样本 = { + "character_location_changed": ( + {"characterName": "林深", "toLocation": "舱室底层"}, + "异种核心在胸腔里低鸣", + ), + "character_knowledge_added": ( + {"characterName": "何岚", "knowledge": "林深体内有异种核心"}, + "何岚站在舱门口没有说话", + ), + "relationship_changed": ( + {"fromName": "林深", "toName": "何岚", "relation": "互相提防"}, + "何岚站在舱门口没有说话", + ), + "hook_advanced": ( + { + "hookId": "hook-voice", + "action": "advanced", + "dueWindow": {"fromChapter": 5, "toChapter": 8}, + }, + "异种核心在胸腔里低鸣", + ), + "timeline_event_added": ({"event": "隔离舱首次审讯结束"}, "何岚站在舱门口没有说话"), + "setting_added": ({"factType": "污染规则", "text": "黑纹扩散不可逆"}, "林深把黑纹缠上手臂"), +} +名称映射 = { + "林深": "character-lin", + "何岚": "character-he", + "舱室底层": "location-bottom", + "指挥舱": "location-command", + "hook-voice": "hook-voice", + "h1": "hook-one", +} + + +@pytest.fixture +def 旧增量环境(): + # 六行保留旧输入种类；它们仅是本用例的动态登记，不是生产类型白名单。 + 字符 = 值规则("text") + + def 引用(type_id): + return 值规则("reference", 引用=引用规则((type_id,))) + + 角色, 地点, 伏笔 = 引用("character"), 引用("location"), 引用("foreshadowing") + 字段表 = { + "character_location_changed": [ + ("角色", "characterName", 角色), + ("目标地点", "toLocation", 地点), + ], + "character_knowledge_added": [ + ("角色", "characterName", 角色), + ("新增认知", "knowledge", 字符), + ], + "relationship_changed": [ + ("甲方", "fromName", 角色), + ("乙方", "toName", 角色), + ("关系", "relation", 字符), + ], + "hook_advanced": [ + ("伏笔", "hookId", 伏笔), + ("动作", "action", 值规则("enum", 枚举=("planted", "advanced", "paid"))), + ], + "timeline_event_added": [("事件", "event", 字符)], + "setting_added": [("设定类型", "factType", 字符), ("内容", "text", 字符)], + } + 结构目录 = {} + for id_, fields in 字段表.items(): + 家族 = "relation" if id_ == "relationship_changed" else "entity" + 类型 = 类型定义(id_, id_, "world", 家族, (家族,)) + 结构 = 结构定义( + id_, + id_, + 1, + tuple( + 字段定义(id_ + "." + key, key, key, rule, 必填=True, 别名=(alias,)) + for key, alias, rule in fields + ), + ) + 结构目录[id_] = (类型, 合成结构(结构)) + 对象 = { + id_: 引用对象( + id_, + "synthetic-work", + "character" + if id_.startswith("character-") + else "location" + if id_.startswith("location-") + else "foreshadowing", + ) + for id_ in 名称映射.values() + } + return { + "作品ID": "synthetic-work", + "结构目录": 结构目录, + "正文": 单段(原增量正文), + "可用对象": 对象, + "章节顺序": {f"chapter-{i}": i for i in range(1, 13)}, + } + + +def 旧样本提案( + 类型名="character_location_changed", *, payload=None, quote=None, delta_id="delta-1" +): + 原字段, 原引文 = 旧增量样本.get(类型名, 旧增量样本["character_location_changed"]) + 内容 = dict(原字段 if payload is None else payload) + 窗口 = 内容.pop("dueWindow", None) + # 已知旧名称在输入边界转换为固定对象ID；生产事实不由显示名拼接身份。 + for key in ("characterName", "fromName", "toName", "toLocation", "hookId"): + if key in 内容: + 内容[key] = 名称映射[内容[key]] + return 事实条目( + delta_id, + 内容.get("hookId", "object-" + delta_id), + 类型名, + 内容, + (定位字面引文(单段(原增量正文), 原引文 if quote is None else quote),), + instance_family="relation" if 类型名 == "relationship_changed" else "entity", + planned_window=叙事窗口(f"chapter-{窗口['fromChapter']}", f"chapter-{窗口['toChapter']}") + if 窗口 + else None, + ) + + +@pytest.mark.parametrize("delta_type", tuple(旧增量样本), ids=tuple(旧增量样本)) +def test_each_delta_type_has_working_contract__d0e17d(旧增量环境, delta_type): + 原字段, 引文 = 旧增量样本[delta_type] + 提案 = 旧样本提案(delta_type, payload=原字段, quote=引文) + 结果 = 校验事实批次((提案,), **旧增量环境) + assert 结果[0].type_id == delta_type and 结果[0].object_id + assert 结果[0].evidence[0].quote == 引文 + + +def test_subject_key_derivation__e42f4b(旧增量环境): + 提案 = 旧样本提案( + "relationship_changed", + payload={"fromName": "林深", "toName": "何岚", "relation": "同盟"}, + quote="何岚站在舱门口没有说话", + ) + 结果 = 校验事实批次((提案,), **旧增量环境)[0] + assert (结果.content["甲方"], 结果.content["乙方"]) == ("character-lin", "character-he") + hook = 旧样本提案( + "hook_advanced", + payload={"hookId": "hook-voice", "action": "planted"}, + quote="异种核心在胸腔里低鸣", + ) + assert 校验事实批次((hook,), **旧增量环境)[0].object_id == "hook-voice" + + +def test_closed_field_set_and_type_enum__42dd3b(旧增量环境): + extra = asdict(旧样本提案()) + extra["extraField"] = 1 + with pytest.raises(ValidationError): + TypeAdapter(事实条目).validate_python(extra) + bad_type = 旧样本提案("character_resurrected") + with pytest.raises(事实错误) as 错误: + 校验事实批次((bad_type,), **旧增量环境) + assert 错误.value.错误码 == "FACT_TYPE_UNKNOWN" + + +def test_payload_contract_enforced_per_type__a30962(旧增量环境): + missing = 旧样本提案(payload={"characterName": "林深"}) + with pytest.raises(元数据错误): + 校验事实批次((missing,), **旧增量环境) + action = 旧样本提案( + "hook_advanced", + payload={"hookId": "h1", "action": "detonated"}, + quote="异种核心在胸腔里低鸣", + ) + with pytest.raises(元数据错误): + 校验事实批次((action,), **旧增量环境) + window = 旧样本提案( + "hook_advanced", + payload={ + "hookId": "h1", + "action": "planted", + "dueWindow": {"fromChapter": 9, "toChapter": 3}, + }, + quote="异种核心在胸腔里低鸣", + ) + with pytest.raises(事实错误) as 错误: + 校验事实批次((window,), **旧增量环境) + assert 错误.value.错误码 == "FACT_TIME_INVALID" + + +def test_batch_rejects_duplicate_ids__b430df(旧增量环境): + 首 = 旧样本提案(delta_id="delta-1") + 后 = 旧样本提案(delta_id="delta-1", payload={"characterName": "何岚", "toLocation": "指挥舱"}) + with pytest.raises(事实错误) as 错误: + 校验事实批次((首, 后), **旧增量环境) + assert 错误.value.错误码 == "FACT_DUPLICATE_ID" + assert [ + x.proposal_id for x in 校验事实批次((首, replace(后, proposal_id="delta-2")), **旧增量环境) + ] == ["delta-1", "delta-2"] + + +def test_salvage_drops_unbound_relation_and_keeps_exact_name_evidence__8101bb(): + 原输出 = { + "entities": [ + { + "type": "item", + "name": "黑色钥匙", + "brief": "钥匙", + "fields": {}, + "evidence": "模型编造", + }, + { + "type": "item", + "name": "不存在的物件", + "brief": "?", + "fields": {}, + "evidence": "模型编造", + }, + ], + "relations": [ + { + "source": "林深", + "target": "何岚", + "type": "伙伴", + "description": "?", + "evidence": "模型编造", + } + ], + "state": {}, + } + 结果 = 保留名称提及(原输出, 单段(原抽取正文), tuple(s.类型 for s in 读取内置种子())) + assert [x["name"] for x in 结果["mentions"]] == ["黑色钥匙"] + assert 结果["mentions"][0]["evidence"].quote == "黑色钥匙" + assert 结果["relations"] == [] and 结果["mechanical_drops"] == {"entities": 1, "relations": 1} + assert "brief" not in 结果["mentions"][0] and 结果["state"] == "needs_confirmation" diff --git a/tests/集成/test_故事事实确认与时点.py b/tests/集成/test_故事事实确认与时点.py new file mode 100644 index 0000000..afe58a4 --- /dev/null +++ b/tests/集成/test_故事事实确认与时点.py @@ -0,0 +1,715 @@ +"""真实元数据、正文、B02与S01共同验证事实来源和时点，不调用模型。""" + +from dataclasses import replace + +import pytest +from pydantic import TypeAdapter + +from muse.作品规划.接口 import 档案保存 +from muse.元数据.接口 import 元数据服务, 导入内置结构 +from muse.共享.调用身份 import 内容用途, 用途, 调用身份 +from muse.共享.错误 import Muse错误 +from muse.启动 import 构建 +from muse.故事世界.接口 import 事实定位, 事实提案请求, 定位字面引文, 正文依据 +from muse.正式变更.接口 import 作者动作 +from muse.正文写作.接口 import 文本节点, 正文草稿, 段落 +from muse.配置 import 应用配置 + +pytestmark = pytest.mark.数据库 + + +@pytest.fixture +def 故事环境(应用测试库): + with 应用测试库[用途.维护].连接() as 连, 连.transaction(): + 导入内置结构(元数据服务(连)) + 装配 = 构建(应用配置(应用测试库[用途.生产].引用, "synthetic")) + 作者 = 调用身份("author", None, 用途.生产, 内容用途.规划) + 作品 = 装配.要求作品() + 结构 = 作品.新档案结构(作者, "story-work", "work_core", 1) + 作品.保存档案(作者, "create-story", 档案保存("story-work", 0, {"名称": "合成故事世界"}, 结构)) + for i in range(1, 4): + 作品.添加章节( + 作者, f"chapter-{i}", "story-work", f"chapter-{i}", f"第{i}幕", 预期目录版本=i - 1 + ) + 正文 = 装配.要求正文() + for i, text in [ + (1, "林深停在码头，何岚站在船旁。"), + (2, "何岚知道林深已经登船。"), + (3, "林深来到北岸。"), + ]: + 正文.保存人工( + 作者, f"body-{i}", f"chapter-{i}", 0, 正文草稿((段落(f"p{i}", (文本节点(text),)),)) + ) + return 装配, 作者, 应用测试库 + + +def 来源(装配, 作者, chapter, quote): + 记录 = 装配.要求正文().读取正文(作者, chapter) + 原文 = TypeAdapter(正文草稿).validate_python(记录["document"]) + return 正文依据( + chapter, "main", 记录["revision"], 记录["document_hash"], (定位字面引文(原文, quote),) + ) + + +def 提案( + 装配, + 作者, + object_id, + *, + sources=(), + independent=False, + effective="chapter-1", + mode="fact", + known_to=(), +): + 世界 = 装配.要求故事世界() + 结构 = 世界.新事实结构(作者, "story-work", object_id, "character", 1) + 请求 = 事实提案请求( + "story-work", + object_id, + 0, + 结构, + {"名称": object_id}, + tuple(sources), + effective, + independent=independent, + knowledge_mode=mode, + known_to=tuple({"actor_id": id_, "from_chapter_id": effective} for id_ in known_to), + ) + 回执 = 世界.提出事实(作者, "propose-" + object_id, 请求) + 记录 = 世界.读取提案(作者, 回执["results"][0]["proposal_id"]) + return 记录, 事实定位( + 记录["proposal_id"], 记录["revision"], 记录["proposal_hash"], "world:" + object_id, 0 + ) + + +def 确认(装配, 作者, 定位, command): + 世界 = 装配.要求故事世界() + 审阅 = 世界.打开事实审阅(作者, 定位) + return 世界.决定事实( + 作者, command, 定位, 作者动作.采纳, 审阅["review_id"], 审阅["review_hash"], ("object",) + ) + + +def test_事实确认与来源失效保留独立设定__ac1001(故事环境): + 装配, 作者, _ = 故事环境 + 源 = 来源(装配, 作者, "chapter-1", "林深停在码头") + _, 位置 = 提案(装配, 作者, "person-dependent", sources=(源,)) + assert 装配.要求故事世界().查询世界(作者, "story-work", "chapter-1")["facts"] == [] + 回执 = 确认(装配, 作者, 位置, "confirm-dependent") + assert 回执["results"][0]["revision"] == 1 + _, 独立 = 提案(装配, 作者, "person-independent", independent=True) + 确认(装配, 作者, 独立, "confirm-independent") + 初 = 装配.要求故事世界().查询世界(作者, "story-work", "chapter-1") + assert {v["object_id"] for v in 初["facts"]} == {"person-dependent", "person-independent"} + 装配.要求正文().保存人工( + 作者, + "changed-source", + "chapter-1", + 1, + 正文草稿((段落("p1", (文本节点("码头上没有人。"),)),)), + ) + 后 = 装配.要求故事世界().查询世界(作者, "story-work", "chapter-1") + assert {v["object_id"] for v in 后["facts"]} == {"person-independent"} + assert {v["object_id"]: v["reason"] for v in 后["omitted"]}[ + "person-dependent" + ] == "SOURCE_STALE" + + +def test_叙事时点与系统确认截止分开__ac1002(故事环境): + 装配, 作者, _ = 故事环境 + _, 一 = 提案(装配, 作者, "person-before", independent=True) + 确认(装配, 作者, 一, "confirm-before") + 切点 = 装配.要求故事世界().查询世界(作者, "story-work", "chapter-1")["system_revision"] + _, 后 = 提案( + 装配, + 作者, + "person-future", + sources=(来源(装配, 作者, "chapter-3", "林深来到北岸"),), + effective="chapter-3", + ) + 确认(装配, 作者, 后, "confirm-future") + assert { + v["object_id"] + for v in 装配.要求故事世界().查询世界(作者, "story-work", "chapter-1")["facts"] + } == {"person-before"} + assert { + v["object_id"] + for v in 装配.要求故事世界().查询世界(作者, "story-work", "chapter-3", 系统版本=切点)[ + "facts" + ] + } == {"person-before"} + assert { + v["object_id"] + for v in 装配.要求故事世界().查询世界(作者, "story-work", "chapter-3")["facts"] + } == {"person-before", "person-future"} + + +def test_后文披露早期事件和角色猜测分别查询__ac1003(故事环境): + 装配, 作者, _ = 故事环境 + _, 早期 = 提案( + 装配, + 作者, + "past-recalled", + sources=(来源(装配, 作者, "chapter-3", "林深来到北岸"),), + effective="chapter-1", + ) + 确认(装配, 作者, 早期, "confirm-recalled") + 世界 = 装配.要求故事世界() + assert [v["object_id"] for v in 世界.查询世界(作者, "story-work", "chapter-1")["facts"]] == [ + "past-recalled" + ] + 读者 = 世界.查询世界(作者, "story-work", "chapter-1", 视角="reader") + assert 读者["facts"] == [] and 读者["omitted"][0]["reason"] == "FUTURE_EVIDENCE" + assert len(世界.查询世界(作者, "story-work", "chapter-3", 视角="reader")["facts"]) == 1 + _, 主体 = 提案(装配, 作者, "knower", independent=True) + 确认(装配, 作者, 主体, "confirm-knower") + _, 猜测 = 提案(装配, 作者, "guess", independent=True, mode="belief", known_to=("knower",)) + 确认(装配, 作者, 猜测, "confirm-guess") + _, 计划 = 提案(装配, 作者, "future-plan", independent=True, mode="plan", known_to=("knower",)) + 确认(装配, 作者, 计划, "confirm-plan") + 角色 = 世界.查询世界(作者, "story-work", "chapter-1", 视角="character", 知情主体ID="knower") + assert [x["object_id"] for x in 角色["beliefs"]] == ["guess"] + assert [x["object_id"] for x in 角色["plans"]] == ["future-plan"] + assert 角色["facts"] == [] + + +def test_其他充分正文依据保持已确认事实有效__ac1004(故事环境): + 装配, 作者, _ = 故事环境 + 一 = 来源(装配, 作者, "chapter-1", "林深停在码头") + 二 = 来源(装配, 作者, "chapter-3", "林深来到北岸") + _, 定位 = 提案(装配, 作者, "two-sources", sources=(一, 二)) + 确认(装配, 作者, 定位, "confirm-two-sources") + 装配.要求正文().保存人工( + 作者, + "change-one-source", + "chapter-1", + 1, + 正文草稿((段落("p1", (文本节点("码头上没有人。"),)),)), + ) + 当前 = 装配.要求故事世界().查询世界(作者, "story-work", "chapter-3") + assert [v["object_id"] for v in 当前["facts"]] == ["two-sources"] + + +def test_源变后旧审阅不能确认但可以重新审阅拒绝__ac1005(故事环境): + 装配, 作者, _ = 故事环境 + 世界 = 装配.要求故事世界() + _, 定位 = 提案( + 装配, 作者, "stale-review", sources=(来源(装配, 作者, "chapter-1", "林深停在码头"),) + ) + 审阅 = 世界.打开事实审阅(作者, 定位) + 装配.要求正文().保存人工( + 作者, + "change-reviewed-source", + "chapter-1", + 1, + 正文草稿((段落("p1", (文本节点("没有人停在码头。"),)),)), + ) + with pytest.raises(Muse错误) as 捕获: + 世界.决定事实( + 作者, + "stale-accept", + 定位, + 作者动作.采纳, + 审阅["review_id"], + 审阅["review_hash"], + ("object",), + ) + assert 捕获.value.错误码 == "SOURCE_STALE" + 新审阅 = 世界.打开事实审阅(作者, 定位) + 回执 = 世界.决定事实( + 作者, "reject-stale", 定位, 作者动作.拒绝, 新审阅["review_id"], 新审阅["review_hash"], () + ) + assert ( + 回执["action"] == "reject" and 世界.查询世界(作者, "story-work", "chapter-1")["facts"] == [] + ) + + +def test_新动态类型与计划窗口贯通真实持久化__ac1006(故事环境): + from muse.元数据.接口 import 值规则, 字段定义, 类型定义, 结构定义 + from muse.故事世界.接口 import 叙事窗口 + + 装配, 作者, 库 = 故事环境 + with 库[用途.维护].连接() as 连, 连.transaction(): + 元数据 = 元数据服务(连) + 元数据.登记类型(类型定义("landing_point", "登船点", "world", "entity", ("entity",))) + 结构 = 结构定义( + "landing_point", + "landing_point", + 1, + ( + 字段定义("landing.name", "名称", "名称", 值规则("text"), 必填=True), + 字段定义( + "landing.hints", "位置提示", "位置提示", 值规则("list", 元素=值规则("text")) + ), + ), + ) + 元数据.登记结构候选(结构) + 元数据.发布结构("landing_point", 1, 结构.内容哈希) + 世界 = 装配.要求故事世界() + 选择 = 世界.新事实结构(作者, "story-work", "landing-one", "landing_point", 1) + 请求 = 事实提案请求( + "story-work", + "landing-one", + 0, + 选择, + {"名称": "码头", "位置提示": ["船旁"]}, + (来源(装配, 作者, "chapter-1", "林深停在码头，何岚站在船旁"),), + "chapter-1", + planned_window=叙事窗口("chapter-2", "chapter-3"), + ) + with pytest.raises(Muse错误) as 捕获: + 世界.提出事实( + 作者, "bad-window", replace(请求, planned_window=叙事窗口("chapter-3", "chapter-1")) + ) + assert 捕获.value.错误码 == "FACT_TIME_INVALID" + 回执 = 世界.提出事实(作者, "dynamic-proposal", 请求) + 原 = 世界.读取提案(作者, 回执["results"][0]["proposal_id"]) + 确认( + 装配, + 作者, + 事实定位(原["proposal_id"], 1, 原["proposal_hash"], "world:landing-one", 0), + "confirm-dynamic", + ) + 实体 = 世界.查询世界(作者, "story-work", "chapter-1")["facts"][0] + assert 实体["type_id"] == "landing_point" and 实体["content"] == 请求.content + assert 实体["planned_window"] == {"from_chapter_id": "chapter-2", "to_chapter_id": "chapter-3"} + 读者 = 世界.查询世界(作者, "story-work", "chapter-1", 视角="reader")["facts"][0] + assert "planned_window" not in 读者 + + +def test_HTTP事实提案确认与查询使用真实作者会话__ac1007(故事环境, tmp_path): + from dataclasses import asdict + + from fastapi.testclient import TestClient + + from muse.接入.http.应用 import 创建应用 + from muse.配置 import 服务配置 + + 装配, 作者, 库 = 故事环境 + 口令 = tmp_path / "fact-author.txt" + 口令.write_text("synthetic-fact-author") + 口令.chmod(0o600) + 配置 = 应用配置( + 库[用途.生产].引用, + "synthetic", + HTTP=服务配置( + str(口令), + 作者ID="author", + 公开地址="http://testserver", + 允许来源=("http://testserver",), + ), + ) + with TestClient(创建应用(配置), headers={"origin": "http://testserver"}) as client: + assert client.get("/api/v1/works/story-work/facts").status_code == 401 + assert ( + client.post("/api/v1/session", json={"password": "synthetic-fact-author"}).status_code + == 200 + ) + frame = client.get( + "/api/v1/works/story-work/facts/schema", + params={"object_id": "http-person", "schema_id": "character"}, + ) + assert frame.status_code == 200, frame.text + req = { + "work_id": "story-work", + "object_id": "http-person", + "expected_revision": 0, + "schema": frame.json(), + "content": {"名称": "林深"}, + "sources": [asdict(来源(装配, 作者, "chapter-1", "林深停在码头"))], + "effective_from": "chapter-1", + } + creation = {"command_id": "http-fact", "proposal": req} + result = client.post("/api/v1/works/story-work/fact-proposals", json=creation) + assert result.status_code == 200, result.text + assert ( + client.post("/api/v1/works/story-work/fact-proposals", json=creation).json() + == result.json() + ) + pid = result.json()["results"][0]["proposal_id"] + item = client.get(f"/api/v1/fact-proposals/{pid}").json() + assert client.get("/api/v1/works/story-work/facts").json()["facts"] == [] + review_req = { + "proposal_id": pid, + "proposal_revision": 1, + "proposal_hash": item["proposal_hash"], + "target_ref": "world:http-person", + "expected_revision": 0, + } + review = client.post(f"/api/v1/fact-proposals/{pid}/reviews", json=review_req) + assert review.status_code == 200, review.text + decision = { + **review_req, + "command_id": "http-fact-confirm", + "action": "adopt", + "author_review_id": review.json()["review_id"], + "review_hash": review.json()["review_hash"], + "approved_changes": ["object"], + } + assert ( + client.post( + f"/api/v1/fact-proposals/{pid}/decisions", json={**decision, "decided_by": "author"} + ).status_code + == 422 + ) + accepted = client.post(f"/api/v1/fact-proposals/{pid}/decisions", json=decision) + assert accepted.status_code == 200, accepted.text + assert ( + client.post(f"/api/v1/fact-proposals/{pid}/decisions", json=decision).json() + == accepted.json() + ) + facts = client.get( + "/api/v1/works/story-work/facts", params={"as_of": "chapter-1", "view": "reader"} + ).json() + assert facts["facts"][0]["content"]["名称"] == "林深" and facts["facts"][0]["revision"] == 1 + + +def 关系提案(装配, 作者, object_id="relation-main"): + 世界 = 装配.要求故事世界() + _, 甲 = 提案( + 装配, 作者, "person-left", sources=(来源(装配, 作者, "chapter-1", "林深停在码头"),) + ) + 确认(装配, 作者, 甲, "confirm-left") + _, 乙 = 提案(装配, 作者, "person-right", independent=True) + 确认(装配, 作者, 乙, "confirm-right") + schema = 世界.新事实结构( + 作者, "story-work", object_id, "character_relation", 1, 家族="relation" + ) + req = 事实提案请求( + "story-work", + object_id, + 0, + schema, + {"甲方": "person-left", "乙方": "person-right"}, + (来源(装配, 作者, "chapter-2", "何岚知道林深已经登船"),), + "chapter-2", + instance_family="relation", + ) + return req + + +def test_关系端点和正文引文必须真实__ac1008(故事环境): + 装配, 作者, _ = 故事环境 + 世界 = 装配.要求故事世界() + req = 关系提案(装配, 作者) + with pytest.raises(Muse错误): + 世界.提出事实(作者, "missing-endpoint", replace(req, content={"甲方": "person-left"})) + with pytest.raises(Muse错误) as 错: + 世界.提出事实( + 作者, + "unknown-endpoint", + replace(req, content={"甲方": "person-left", "乙方": "missing-person"}), + ) + assert 错.value.错误码 == "FACT_REFERENCE_INVALID" + anchor = replace( + req.sources[0].evidence[0], quote="正文没有这条依据句", end=len("正文没有这条依据句") + ) + with pytest.raises(Muse错误) as 错: + 世界.提出事实( + 作者, "fake-quote", replace(req, sources=(replace(req.sources[0], evidence=(anchor,)),)) + ) + assert 错.value.错误码 == "FACT_QUOTE_NOT_FOUND" + result = 世界.提出事实(作者, "real-relation", req) + item = 世界.读取提案(作者, result["results"][0]["proposal_id"]) + 确认( + 装配, + 作者, + 事实定位(item["proposal_id"], 1, item["proposal_hash"], "world:relation-main", 0), + "confirm-relation", + ) + assert any( + v["object_id"] == "relation-main" + for v in 世界.查询世界(作者, "story-work", "chapter-2")["facts"] + ) + + +def test_关系确认同时保护端点的正文依据__ac1009(故事环境, monkeypatch): + from concurrent.futures import ThreadPoolExecutor + from threading import Event + from time import monotonic, sleep + + from muse.故事世界.存储 import 故事存储 + + 装配, 作者, 库 = 故事环境 + 世界 = 装配.要求故事世界() + req = 关系提案(装配, 作者) + result = 世界.提出事实(作者, "relation-with-lock", req) + item = 世界.读取提案(作者, result["results"][0]["proposal_id"]) + loc = 事实定位(item["proposal_id"], 1, item["proposal_hash"], "world:relation-main", 0) + review = 世界.打开事实审阅(作者, loc) + entered, release = Event(), Event() + pids = [] + original = 故事存储.写对象 + + def pause_write(self, *args, **kwargs): + # 只暂停调度，随后调用真实存储；不替换数据库行为或回执。 + pids.append(self.连.info.backend_pid) + entered.set() + assert release.wait(5) + return original(self, *args, **kwargs) + + monkeypatch.setattr(故事存储, "写对象", pause_write) + with ThreadPoolExecutor(2) as pool: + accepted = pool.submit( + 世界.决定事实, + 作者, + "locked-confirm", + loc, + 作者动作.采纳, + review["review_id"], + review["review_hash"], + ("object",), + ) + assert entered.wait(5) + edited = pool.submit( + 装配.要求正文().保存人工, + 作者, + "parallel-source-write", + "chapter-1", + 1, + 正文草稿((段落("p1", (文本节点("新的码头段落。"),)),)), + ) + try: + end = monotonic() + 3 + blocked = False + while monotonic() < end and not blocked: + with 库[用途.维护].连接() as conn: + blocked = conn.execute( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity " + "WHERE %s=ANY(pg_blocking_pids(pid)))", + (pids[0],), + ).fetchone()[0] + if not blocked: + sleep(0.01) + assert blocked and not edited.done() + finally: + release.set() + assert accepted.result(timeout=5)["results"][0]["object_id"] == "relation-main" + assert edited.result(timeout=5)["results"][0]["revision"] == 2 + + +def test_跨作品来源和提案目标拒绝__ac1010(故事环境): + 装配, 作者, _ = 故事环境 + 作品 = 装配.要求作品() + schema = 作品.新档案结构(作者, "another-work", "work_core", 1) + 作品.保存档案( + 作者, "create-another", 档案保存("another-work", 0, {"名称": "另一部合成作品"}, schema) + ) + 作品.添加章节( + 作者, "another-chapter", "another-work", "another-chapter", "另一章", 预期目录版本=0 + ) + 装配.要求正文().保存人工( + 作者, + "another-body", + "another-chapter", + 0, + 正文草稿((段落("other-p", (文本节点("另一部作品的角色。"),)),)), + ) + 世界 = 装配.要求故事世界() + frame = 世界.新事实结构(作者, "story-work", "wrong-source", "character", 1) + req = 事实提案请求( + "story-work", + "wrong-source", + 0, + frame, + {"名称": "另一部作品的角色"}, + (来源(装配, 作者, "another-chapter", "另一部作品的角色"),), + "chapter-1", + ) + with pytest.raises(Muse错误) as err: + 世界.提出事实(作者, "cross-source", req) + assert err.value.错误码 == "SCOPE_DENIED" + _, loc = 提案(装配, 作者, "own-person", independent=True) + with pytest.raises(Muse错误) as err: + 世界.打开事实审阅(作者, replace(loc, target_ref="world:another-person")) + assert err.value.错误码 == "SCOPE_DENIED" + assert 世界.查询世界(作者, "story-work", "chapter-1")["facts"] == [] + + +def test_事实确认失败回滚且原命令可恢复__ac1011(故事环境): + import psycopg + + 装配, 作者, 库 = 故事环境 + _, loc = 提案(装配, 作者, "rollback-person", independent=True) + 世界 = 装配.要求故事世界() + review = 世界.打开事实审阅(作者, loc) + with 库[用途.维护].连接() as conn: + conn.execute("""CREATE FUNCTION reject_world_decision() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'synthetic world decision failure'; END $$; + CREATE TRIGGER reject_world_decision BEFORE INSERT ON muse_candidate_decision + FOR EACH ROW EXECUTE FUNCTION reject_world_decision();""") + args = ( + 作者, + "world-retry", + loc, + 作者动作.采纳, + review["review_id"], + review["review_hash"], + ("object",), + ) + with pytest.raises(psycopg.Error): + 世界.决定事实(*args) + assert 世界.查询世界(作者, "story-work", "chapter-1")["system_revision"] == 0 + with 库[用途.维护].连接() as conn: + assert conn.execute("SELECT count(*) FROM muse_world_object").fetchone()[0] == 0 + conn.execute("DROP TRIGGER reject_world_decision ON muse_candidate_decision") + receipt = 世界.决定事实(*args) + assert 世界.决定事实(*args) == receipt + assert receipt["results"][0]["system_revision"] == 1 + + +def test_事实修订追加版本且历史系统截止仍可回查__ac1012(故事环境): + 装配, 作者, _ = 故事环境 + 世界 = 装配.要求故事世界() + _, loc = 提案(装配, 作者, "versioned-person", independent=True) + 首 = 确认(装配, 作者, loc, "confirm-v1") + frame = 世界.新事实结构(作者, "story-work", "versioned-person", "character", 1) + req = 事实提案请求( + "story-work", + "versioned-person", + 1, + frame, + {"名称": "作者修订后的名字"}, + (), + "chapter-2", + independent=True, + ) + created = 世界.提出事实(作者, "revised-proposal", req) + p = 世界.读取提案(作者, created["results"][0]["proposal_id"]) + 确认( + 装配, + 作者, + 事实定位(p["proposal_id"], 1, p["proposal_hash"], "world:versioned-person", 1), + "confirm-v2", + ) + assert ( + 世界.查询世界(作者, "story-work", "chapter-2")["facts"][0]["content"]["名称"] + == "作者修订后的名字" + ) + 早期 = 世界.查询世界(作者, "story-work", "chapter-1")["facts"][0] + assert 早期["revision"] == 1 + 历史 = 世界.查询世界( + 作者, "story-work", "chapter-2", 系统版本=首["results"][0]["system_revision"] + )["facts"][0] + assert 历史["revision"] == 1 and 历史["content"] == 早期["content"] + with pytest.raises(Muse错误) as err: + 世界.提出事实(作者, "old-version-write", req) + assert err.value.错误码 == "REVISION_CONFLICT" + + +@pytest.mark.parametrize( + "模式,生效", + [("fact", "chapter-3"), ("plan", "chapter-1"), ("belief", "chapter-1")], + ids=["future", "plan", "belief"], +) +def test_事实引用不借用未来或猜测计划实体__ac1013(故事环境, 模式, 生效): + 装配, 作者, _ = 故事环境 + 世界 = 装配.要求故事世界() + req = 关系提案(装配, 作者) + _, loc = 提案(装配, 作者, "unavailable-endpoint", independent=True, mode=模式, effective=生效) + 确认(装配, 作者, loc, "confirm-unavailable-endpoint") + with pytest.raises(Muse错误) as 错: + 世界.提出事实( + 作者, + "fact-with-unavailable-endpoint", + replace(req, content={"甲方": "person-left", "乙方": "unavailable-endpoint"}), + ) + assert 错.value.错误码 == "FACT_REFERENCE_INVALID" + # 规划仍能引用其时点内的规划对象，不能把计划伪装成事实。 + if 模式 == "plan": + assert 世界.提出事实( + 作者, + "planned-relation", + replace( + req, + knowledge_mode="plan", + content={"甲方": "person-left", "乙方": "unavailable-endpoint"}, + ), + )["results"] + + +def test_历史事实按保存结构解释并应用当前策略__ac1014(故事环境): + from muse.元数据.接口 import 值规则, 合成结构, 启用命令, 字段定义, 投影字段 + from muse.故事世界.设定修订 import 事实授权 + + 装配, 作者, 库 = 故事环境 + 世界 = 装配.要求故事世界() + _, loc = 提案(装配, 作者, "historical-schema", independent=True) + 确认(装配, 作者, loc, "confirm-historical-schema") + before = 世界.查询世界(作者, "story-work", "chapter-1")["facts"][0] + with 库[用途.维护].连接() as 连, 连.transaction(): + meta = 元数据服务(连) + old = meta.读取结构("character", 1) + new = replace( + old, + schema_version=2, + 父版本=1, + 字段=old.字段 + (字段定义("character.new", "新提示", "新提示", 值规则("text")),), + ) + meta.登记结构候选(new) + meta.发布结构("character", 2, new.内容哈希) + effective = meta.读取有效结构("character", 2) + meta.启用绑定( + "world:historical-schema", + "story-work", + effective, + 启用命令( + "schema-addition", + effective.effective_schema_hash, + 1, + "author", + "world:historical-schema", + ), + ) + expected = 投影字段( + 合成结构(old), + meta.当前策略("character"), + 事实授权("author", "story-work"), + 字段用途名="uiVisible", + 内容用途="planning", + 运行用途="production", + ) + after = 世界.查询世界(作者, "story-work", "chapter-1")["facts"][0] + assert after["content"] == before["content"] + assert after["schema_binding"]["base_version"] == 1 + assert after["projection_version"] == expected.projection_version + + +def test_角色获知时点独立于事件生效__ac1015(故事环境): + 装配, 作者, _ = 故事环境 + 世界 = 装配.要求故事世界() + _, 主体 = 提案(装配, 作者, "late-knower", independent=True) + 确认(装配, 作者, 主体, "confirm-late-knower") + schema = 世界.新事实结构(作者, "story-work", "late-known-event", "character", 1) + 请求 = 事实提案请求( + "story-work", + "late-known-event", + 0, + schema, + {"名称": "林深登船"}, + (来源(装配, 作者, "chapter-1", "林深停在码头"),), + "chapter-1", + known_to=({"actor_id": "late-knower", "from_chapter_id": "chapter-3"},), + ) + 回执 = 世界.提出事实(作者, "propose-late-knowledge", 请求) + 原 = 世界.读取提案(作者, 回执["results"][0]["proposal_id"]) + 确认( + 装配, + 作者, + 事实定位(原["proposal_id"], 1, 原["proposal_hash"], "world:late-known-event", 0), + "confirm-late-knowledge", + ) + assert ( + 世界.查询世界(作者, "story-work", "chapter-1", 视角="character", 知情主体ID="late-knower")[ + "facts" + ] + == [] + ) + assert [ + v["object_id"] + for v in 世界.查询世界( + 作者, "story-work", "chapter-3", 视角="character", 知情主体ID="late-knower" + )["facts"] + ] == ["late-known-event"] diff --git a/数据库/迁移/V0006__故事世界.sql b/数据库/迁移/V0006__故事世界.sql new file mode 100644 index 0000000..359a26a --- /dev/null +++ b/数据库/迁移/V0006__故事世界.sql @@ -0,0 +1,52 @@ +-- B02本书实例、不可变事实版本与提案；作者决定仍由S01保存。 +CREATE TABLE public.muse_world_clock ( + work_id text PRIMARY KEY REFERENCES public.muse_work(work_id), + system_revision bigint NOT NULL CHECK(system_revision>=0) +); +CREATE TABLE public.muse_world_object ( + object_id text PRIMARY KEY, + work_id text NOT NULL REFERENCES public.muse_work(work_id), + type_id text NOT NULL, + instance_family text NOT NULL CHECK(instance_family IN ('entity','relation')), + current_revision bigint NOT NULL CHECK(current_revision>0) +); +CREATE TABLE public.muse_world_version ( + object_id text NOT NULL REFERENCES public.muse_world_object(object_id), + revision bigint NOT NULL CHECK(revision>0), + system_revision bigint NOT NULL CHECK(system_revision>0), + content jsonb NOT NULL, + content_hash text NOT NULL, + schema_binding jsonb NOT NULL, + sources jsonb NOT NULL, + independent boolean NOT NULL, + effective_from text REFERENCES public.muse_chapter(chapter_id), + knowledge_mode text NOT NULL CHECK(knowledge_mode IN ('fact','belief','plan','setting')), + known_to jsonb NOT NULL, + reader_visible boolean NOT NULL, + planned_window jsonb, + author_id text NOT NULL, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + PRIMARY KEY(object_id,revision) +); +ALTER TABLE public.muse_world_object ADD CONSTRAINT world_current_version_fk + FOREIGN KEY(object_id,current_revision) REFERENCES public.muse_world_version(object_id,revision) + DEFERRABLE INITIALLY DEFERRED; +CREATE TABLE public.muse_fact_proposal ( + proposal_id uuid PRIMARY KEY, + work_id text NOT NULL REFERENCES public.muse_work(work_id), + object_id text NOT NULL, + current_revision bigint NOT NULL CHECK(current_revision>0) +); +CREATE TABLE public.muse_fact_proposal_version ( + proposal_id uuid NOT NULL REFERENCES public.muse_fact_proposal(proposal_id), + revision bigint NOT NULL CHECK(revision>0), + request jsonb NOT NULL, + proposed_content jsonb NOT NULL, + proposal_hash text NOT NULL, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + PRIMARY KEY(proposal_id,revision) +); +ALTER TABLE public.muse_fact_proposal ADD CONSTRAINT fact_proposal_current_version_fk + FOREIGN KEY(proposal_id,current_revision) REFERENCES public.muse_fact_proposal_version(proposal_id,revision) + DEFERRABLE INITIALLY DEFERRED; +CREATE INDEX world_version_system_idx ON public.muse_world_version(object_id,system_revision DESC);