W12 本书事实、关系与时间边界基础：本书事实、关系提案、时点边界与角色知情历史。

按 R2 串行阶段整理提交；包内文件为该阶段交付（含后续小增量），状态以工作包清单为准。

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),
+    )

src/muse/故事世界/__init__.py

@@ -0,0 +1 @@
+"""本书实体、关系、时点与事实；业务入口见接口.py。"""

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",
+    }

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()
+        )

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)))

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 结果

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

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

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))

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 结果

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"

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"]

数据库/迁移/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);