d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
490 lines
20 KiB
Python
490 lines
20 KiB
Python
"""目标章细纲与时点事实经 S02 任务以 aiContext 真投影读取；不调用模型。"""
|
||
|
||
from dataclasses import asdict
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
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 流程定义, 流程服务, 流程登记
|
||
from muse.配置 import 应用配置
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
|
||
|
||
def 细纲内容(parent_id: str) -> dict:
|
||
return {
|
||
"targetChapter": 3,
|
||
"sourceRef": {"sourceId": parent_id, "sourceVersion": 1},
|
||
"chapterGoal": "邮差在旧码头确认收信人。",
|
||
"hardConstraints": ["上一章结尾雨还没停。"],
|
||
"keyEvents": [
|
||
{
|
||
"id": "arrival",
|
||
"order": 1,
|
||
"event": "邮差到达旧码头",
|
||
"participants": ["lin-shen"],
|
||
"trigger": "收到退回的信",
|
||
"resultDirection": "确认码头还在使用",
|
||
}
|
||
],
|
||
"adjustableBeats": [],
|
||
"declaredNewFacts": [],
|
||
"mustAppearEntities": ["lin-shen"],
|
||
"chapterEndHook": "仓库的门后有人应声。",
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def 投影环境(内置结构测试库):
|
||
应用测试库 = 内置结构测试库
|
||
with 应用测试库[用途.维护].连接() as 连, 连.transaction():
|
||
元数据 = 元数据服务(连)
|
||
# 动态类型：名称对两种投影可见，秘密备注只对 uiVisible 可见。
|
||
元数据.登记类型(类型定义("landing", "泊位", "world", "entity", ("entity",)))
|
||
结构 = 结构定义(
|
||
"landing",
|
||
"landing",
|
||
1,
|
||
(
|
||
字段定义(
|
||
"landing.name",
|
||
"名称",
|
||
"名称",
|
||
值规则("text"),
|
||
必填=True,
|
||
用途=字段用途(aiContext=True),
|
||
),
|
||
字段定义(
|
||
"landing.note",
|
||
"秘密备注",
|
||
"秘密备注",
|
||
值规则("text"),
|
||
用途=字段用途(uiVisible=True, userEditable=True, aiContext=False),
|
||
),
|
||
),
|
||
)
|
||
元数据.登记结构候选(结构)
|
||
元数据.发布结构("landing", 1, 结构.内容哈希)
|
||
登记 = 流程登记()
|
||
装配 = 构建(应用配置(应用测试库[用途.生产].引用, "test"), 流程=登记)
|
||
with 装配.生命周期():
|
||
作者 = 调用身份("ctx-author", None, 用途.生产, 内容用途.生成)
|
||
作品, 正文 = 装配.要求作品(), 装配.要求正文()
|
||
结构选择 = 作品.新档案结构(作者, "ctx-work", "work_core", 1)
|
||
作品.保存档案(
|
||
作者,
|
||
"create-ctx-work",
|
||
档案保存(
|
||
"ctx-work", 0, {"名称": "旧码头", "题材定位": "现实", "主题立意": "还信"}, 结构选择
|
||
),
|
||
)
|
||
for i in range(1, 5):
|
||
作品.添加章节(
|
||
作者, f"add-ch-{i}", "ctx-work", f"ch-{i}", f"第{i}章", 预期目录版本=i - 1
|
||
)
|
||
for 章, 文本 in [
|
||
("ch-1", "林深停在码头，雨还没停。"),
|
||
("ch-3", "林深来到北岸。"),
|
||
]:
|
||
正文.保存人工(
|
||
作者, f"body-{章}", 章, 0, 正文草稿((段落(f"p-{章}", (文本节点(文本),)),))
|
||
)
|
||
外结构 = 作品.新档案结构(作者, "other-work", "work_core", 1)
|
||
作品.保存档案(
|
||
作者,
|
||
"create-other-work",
|
||
档案保存(
|
||
"other-work", 0, {"名称": "别的书", "题材定位": "现实", "主题立意": "无"}, 外结构
|
||
),
|
||
)
|
||
作品.添加章节(作者, "add-foreign", "other-work", "foreign", "外章", 预期目录版本=0)
|
||
|
||
规划 = 装配.要求规划()
|
||
|
||
def 保存并确认(请求, 命令: str) -> dict:
|
||
回执 = 规划.保存人工候选(作者, 命令 + "-draft", 请求)
|
||
候选 = 规划.读取候选(作者, 回执["results"][0]["candidate_id"])
|
||
定位 = 规划定位(
|
||
候选["candidate_id"],
|
||
候选["revision"],
|
||
候选["candidate_hash"],
|
||
候选["plan_id"],
|
||
请求.expected_revision,
|
||
)
|
||
审阅 = 规划.打开审阅(作者, 定位)
|
||
规划.决定(
|
||
作者,
|
||
命令 + "-accept",
|
||
定位,
|
||
作者动作.采纳,
|
||
审阅["review_id"],
|
||
审阅["review_hash"],
|
||
("plan",),
|
||
)
|
||
return 规划.读取规划(作者, 候选["plan_id"])
|
||
|
||
世界 = 装配.要求故事世界()
|
||
|
||
def 正文源(章: str, 引文: str) -> 正文依据:
|
||
记录 = 正文.读取正文(作者, 章)
|
||
草稿 = TypeAdapter(正文草稿).validate_python(记录["document"])
|
||
return 正文依据(
|
||
章, "main", 记录["revision"], 记录["document_hash"], (定位字面引文(草稿, 引文),)
|
||
)
|
||
|
||
def 确认事实(对象, 类型, 内容, 来源=(), 生效="ch-1", 独立=False) -> None:
|
||
选择 = 世界.新事实结构(作者, "ctx-work", 对象, 类型, 1)
|
||
回执 = 世界.提出事实(
|
||
作者,
|
||
"propose-" + 对象,
|
||
事实提案请求("ctx-work", 对象, 0, 选择, 内容, tuple(来源), 生效, independent=独立),
|
||
)
|
||
提案 = 世界.读取提案(作者, 回执["results"][0]["proposal_id"])
|
||
定位 = 事实定位(
|
||
提案["proposal_id"], 提案["revision"], 提案["proposal_hash"], "world:" + 对象, 0
|
||
)
|
||
审阅 = 世界.打开事实审阅(作者, 定位)
|
||
世界.决定事实(
|
||
作者,
|
||
"confirm-" + 对象,
|
||
定位,
|
||
作者动作.采纳,
|
||
审阅["review_id"],
|
||
审阅["review_hash"],
|
||
("object",),
|
||
)
|
||
|
||
确认事实("lin-shen", "character", {"名称": "林深"}, (正文源("ch-1", "林深停在码头"),))
|
||
确认事实("dock-yard", "landing", {"名称": "旧码头", "秘密备注": "夜里卸货"}, 独立=True)
|
||
确认事实(
|
||
"secret-cargo", "character", {"名称": "夜间货物"}, (正文源("ch-3", "林深来到北岸"),)
|
||
)
|
||
|
||
父 = 保存并确认(
|
||
规划保存(
|
||
"ctx-work", 0, 规划.新规划结构(作者, "ctx-work", "outline", 1), {"名称": "旧码头"}
|
||
),
|
||
"outline",
|
||
)
|
||
细纲 = 保存并确认(
|
||
规划保存(
|
||
"ctx-work",
|
||
0,
|
||
规划.新规划结构(作者, "ctx-work", "fine_outline", 1),
|
||
细纲内容(父["plan_id"]),
|
||
chapter_id="ch-3",
|
||
),
|
||
"fine-outline",
|
||
)
|
||
|
||
def 执行(上下文):
|
||
工具 = 装配.要求只读工具(上下文.领取)
|
||
材料 = {}
|
||
计划 = 工具.调用(工具请求("plan", "query_work_plan", {}))
|
||
材料["目标章细纲"] = 计划.内容["content"]["chapterGoal"]
|
||
事实 = 工具.调用(工具请求("facts", "query_facts_as_of", {}))
|
||
for 项 in 事实.内容["facts"]:
|
||
名称 = 项["content"].get("名称")
|
||
if 名称:
|
||
材料["事实:" + 项["object_id"]] = 名称
|
||
历史 = 工具.调用(
|
||
工具请求("history", "read_history_document", {"chapter_id": "ch-1", "revision": 1})
|
||
)
|
||
材料["历史:ch-1"] = 历史.内容["text"]
|
||
上限 = 上下文.任务.冻结输入["冻结上下文"]["budget"]["context_bytes"]
|
||
return 步骤结果(asdict(装配材料(材料, tuple(材料), 字节上限=上限)))
|
||
|
||
登记.登记处理器(
|
||
步骤处理器(
|
||
"synthetic-projection",
|
||
"1",
|
||
执行,
|
||
"projection/v1",
|
||
"projection/v1",
|
||
保护职责="范围核对",
|
||
允许工具=("read_history_document", "query_work_plan", "query_facts_as_of"),
|
||
)
|
||
)
|
||
登记.登记类型("synthetic-projection", 必需保护=("范围核对",))
|
||
登记.登记恢复检查(
|
||
"synthetic-projection", "1", lambda 任务: 装配.要求上下文().读取范围(任务.任务ID)
|
||
)
|
||
运行 = 装配.任务运行
|
||
assert 运行 is not None
|
||
流程服务(运行, 登记).发布(
|
||
流程定义(
|
||
"synthetic-projection",
|
||
"1",
|
||
(
|
||
步骤计划(
|
||
"read",
|
||
"synthetic-projection",
|
||
"1",
|
||
工具=("read_history_document", "query_work_plan", "query_facts_as_of"),
|
||
),
|
||
),
|
||
)
|
||
)
|
||
|
||
def 开始(范围=None, 任务内容用途=内容用途.生成):
|
||
if 范围 is None:
|
||
范围 = 装配.要求上下文().授权任务范围(
|
||
作者,
|
||
"ctx-work",
|
||
2,
|
||
({"chapter_id": "ch-1", "revision": 1},),
|
||
目标章ID="ch-3",
|
||
含事实=True,
|
||
)
|
||
id_ = 运行.创建任务(
|
||
任务请求(
|
||
"synthetic-projection",
|
||
str(uuid4()),
|
||
作者.作者,
|
||
用途.生产,
|
||
任务内容用途,
|
||
{},
|
||
"test",
|
||
"test",
|
||
{
|
||
"source_scope": 范围,
|
||
"schema_versions": {},
|
||
"authorization": "synthetic-author-choice",
|
||
"budget": {"context_bytes": 2048},
|
||
"stop_conditions": {"max_reads": 3},
|
||
},
|
||
作品ID="ctx-work",
|
||
),
|
||
"synthetic-projection",
|
||
"1",
|
||
)
|
||
领取 = 运行.领取步骤("projection-test", ["synthetic-projection"], 租期秒=120)
|
||
assert 领取 is not None and 领取.任务ID == id_
|
||
return id_, 领取, 装配.要求只读工具(领取)
|
||
|
||
yield {
|
||
"装配": 装配,
|
||
"作者": 作者,
|
||
"开始": 开始,
|
||
"保存并确认": 保存并确认,
|
||
"父": 父,
|
||
"细纲": 细纲,
|
||
"确认事实": 确认事实,
|
||
"正文源": 正文源,
|
||
"规划": 规划,
|
||
}
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-projection-s02-trace",
|
||
environment="真实隔离PostgreSQL",
|
||
given="已确认目标章细纲、多条时点事实与历史正文的真实作品",
|
||
when="经 S02 任务执行装配处理器依次调用三种具名工具并装配材料",
|
||
then=["三种工具读取均留工具读取事件与来源引用；细纲/事实/历史全部进入材料预算入选清单"],
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_细纲与事实经真实任务读取留痕并进入预算__b09201(投影环境):
|
||
环境 = 投影环境
|
||
装配, 开始, 细纲 = 环境["装配"], 环境["开始"], 环境["细纲"]
|
||
task_id, 领取, _ = 开始()
|
||
结果 = 装配.任务运行.执行一步(领取)
|
||
assert set(结果.输出["入选来源"]) == {
|
||
"目标章细纲",
|
||
"事实:dock-yard",
|
||
"事实:lin-shen",
|
||
"事实:secret-cargo",
|
||
"历史:ch-1",
|
||
}
|
||
assert len(结果.输出["入选来源"]) == 5
|
||
读取 = [e for e in 装配.任务运行.续接事件(task_id).事件 if e.类型 == 事件类型.工具读取]
|
||
assert [e.载荷["tool"] for e in 读取] == [
|
||
"query_work_plan",
|
||
"query_facts_as_of",
|
||
"read_history_document",
|
||
]
|
||
计划引用 = 读取[0].载荷["source_refs"][0]
|
||
assert 计划引用["source_id"] == 细纲["plan_id"]
|
||
assert 计划引用["revision"] == "1" and 计划引用["schema_hash"]
|
||
事实身份 = {r["source_id"] for e in 读取 for r in e.载荷["source_refs"]}
|
||
assert {"lin-shen", "dock-yard", "secret-cargo"} <= 事实身份
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-plan-projection",
|
||
environment="真实隔离PostgreSQL",
|
||
given="任务冻结范围为 generation 用途",
|
||
when="调用 query_work_plan 工具",
|
||
then=["返回细纲 aiContext 投影内容、剔除说明与两种用途；来源版本、结构哈希与投影版本完整"],
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_目标细纲按任务用途返回真投影__b09202(投影环境):
|
||
_, _, 工具 = 投影环境["开始"]()
|
||
结果 = 工具.调用(工具请求("plan", "query_work_plan", {}))
|
||
内容 = 结果.内容
|
||
assert 内容["chapter_id"] == "ch-3"
|
||
assert 内容["content"]["chapterGoal"] == "邮差在旧码头确认收信人。"
|
||
assert 内容["content"]["targetChapter"] == 3
|
||
assert 内容["content"]["sourceRef"]["sourceVersion"] == 1
|
||
assert 内容["run_purpose"] == "production"
|
||
assert 内容["content_purpose"] == "generation"
|
||
assert 结果.来源[0].数据版本 == "1"
|
||
assert 结果.来源[0].投影版本
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-facts-view-narrowing",
|
||
environment="真实隔离PostgreSQL",
|
||
given="冻结事实系统版本；一条事实的正文依据位于截止之后；一个不存在的知情主体",
|
||
when="分别以作者、读者与角色视角调用 query_facts_as_of",
|
||
then=[
|
||
"作者视角含全部事实；读者视角省略未来依据并给出 FUTURE_EVIDENCE；未知知情主体 SCOPE_DENIED"
|
||
],
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_时点事实按冻结系统版本与视角收窄返回__b09203(投影环境):
|
||
_, _, 工具 = 投影环境["开始"]()
|
||
作者视角 = 工具.调用(工具请求("facts", "query_facts_as_of", {}))
|
||
对象 = {项["object_id"] for 项 in 作者视角.内容["facts"]}
|
||
assert {"lin-shen", "dock-yard", "secret-cargo"} <= 对象
|
||
assert 作者视角.内容["run_purpose"] == "production"
|
||
assert 作者视角.内容["content_purpose"] == "generation"
|
||
读者视角 = 工具.调用(工具请求("facts", "query_facts_as_of", {"view": "reader"}))
|
||
读者对象 = {项["object_id"] for 项 in 读者视角.内容["facts"]}
|
||
assert "secret-cargo" not in 读者对象 and "lin-shen" in 读者对象
|
||
省略 = {项["object_id"]: 项["reason"] for 项 in 读者视角.内容["omitted"]}
|
||
assert 省略["secret-cargo"] == "FUTURE_EVIDENCE"
|
||
with pytest.raises(上下文错误) as 异常:
|
||
工具.调用(
|
||
工具请求(
|
||
"facts",
|
||
"query_facts_as_of",
|
||
{"view": "character", "knower_id": "missing-actor"},
|
||
)
|
||
)
|
||
assert 异常.value.错误码 == "SCOPE_DENIED"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-projection-drift",
|
||
environment="真实隔离PostgreSQL",
|
||
given="任务范围已冻结细纲身份与事实系统版本",
|
||
when="重新确认细纲新版本或确认新事实后再次调用工具",
|
||
then=["SOURCE_STALE；漂移后不得沿用原任务范围"],
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
@pytest.mark.parametrize("漂移源", ["plan", "facts"], ids=["plan-revised", "facts-advanced"])
|
||
def test_细纲或事实漂移后拒绝沿用原任务__b09204(投影环境, 漂移源):
|
||
环境 = 投影环境
|
||
_, _, 工具 = 环境["开始"]()
|
||
if 漂移源 == "plan":
|
||
内容 = 细纲内容(环境["父"]["plan_id"])
|
||
内容["chapterGoal"] = "作者改过的目标。"
|
||
环境["保存并确认"](
|
||
规划保存(
|
||
"ctx-work",
|
||
1,
|
||
环境["规划"].新规划结构(环境["作者"], "ctx-work", "fine_outline", 1),
|
||
内容,
|
||
chapter_id="ch-3",
|
||
),
|
||
"revise-outline",
|
||
)
|
||
else:
|
||
环境["确认事实"]("late-comer", "character", {"名称": "后来的事实"}, 独立=True)
|
||
with pytest.raises(上下文错误) as 异常:
|
||
工具.调用(
|
||
工具请求(
|
||
"stale",
|
||
"query_work_plan" if 漂移源 == "plan" else "query_facts_as_of",
|
||
{},
|
||
)
|
||
)
|
||
assert 异常.value.错误码 == "SOURCE_STALE"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-aicontext-differs",
|
||
environment="真实隔离PostgreSQL",
|
||
given="一个字段只声明 uiVisible、另一个字段两种用途都可见的动态类型事实",
|
||
when="同一对象分别经工作台查询与 query_facts_as_of 读取",
|
||
then=["工作台可见秘密备注；工具投影剔除该字段并给出 FIELD_NOT_ALLOWED 剔除原因"],
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_aiContext投影与uiVisible差异真实生效__b09205(投影环境):
|
||
环境 = 投影环境
|
||
装配, 作者, 开始 = 环境["装配"], 环境["作者"], 环境["开始"]
|
||
_, _, 工具 = 开始()
|
||
界面 = 装配.要求故事世界().查询世界(作者, "ctx-work", "ch-2")
|
||
泊位 = next(项 for 项 in 界面["facts"] if 项["object_id"] == "dock-yard")
|
||
assert 泊位["content"]["秘密备注"] == "夜里卸货"
|
||
投影 = 工具.调用(工具请求("facts", "query_facts_as_of", {}))
|
||
工具泊位 = next(项 for 项 in 投影.内容["facts"] if 项["object_id"] == "dock-yard")
|
||
assert "秘密备注" not in 工具泊位["content"]
|
||
assert 工具泊位["omitted_fields"]["秘密备注"] == "FIELD_NOT_ALLOWED"
|
||
assert 工具泊位["content"]["名称"] == "旧码头"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-plan-target-boundary",
|
||
environment="真实隔离PostgreSQL",
|
||
given="历史章、截止位置章、跨作品章或无已确认细纲的紧邻章",
|
||
when="以目标章授权任务范围",
|
||
then=["前三种 SCOPE_DENIED（含实际所属作品）；无细纲 PRECONDITION_PENDING"],
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
@pytest.mark.parametrize(
|
||
("目标章", "截止", "错误码"),
|
||
[
|
||
("ch-1", 2, "SCOPE_DENIED"),
|
||
("ch-2", 2, "SCOPE_DENIED"),
|
||
("foreign", 2, "SCOPE_DENIED"),
|
||
("ch-4", 3, "PRECONDITION_PENDING"),
|
||
],
|
||
ids=["past-chapter", "cutoff-chapter", "cross-work", "unconfirmed-outline"],
|
||
)
|
||
def test_目标章不在截止之后紧邻或无细纲拒绝授权__b09206(投影环境, 目标章, 截止, 错误码):
|
||
环境 = 投影环境
|
||
with pytest.raises(上下文错误) as 异常:
|
||
环境["装配"].要求上下文().授权任务范围(
|
||
环境["作者"], "ctx-work", 截止, (), 目标章ID=目标章, 含事实=True
|
||
)
|
||
assert 异常.value.错误码 == 错误码
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-purpose-mismatch",
|
||
environment="真实隔离PostgreSQL",
|
||
given="范围按 generation 授权，任务以 planning 内容用途创建",
|
||
when="领取只读工具装配",
|
||
then=["CONTEXT_SCOPE_INVALID；任务用途与冻结范围用途不一致即失败关闭"],
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_任务用途与冻结范围不一致拒绝__b09207(投影环境):
|
||
with pytest.raises(上下文错误) as 异常:
|
||
投影环境["开始"](任务内容用途=内容用途.规划)
|
||
assert 异常.value.错误码 == "CONTEXT_SCOPE_INVALID"
|