d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
587 lines
25 KiB
Python
587 lines
25 KiB
Python
"""任务上下文实际消费的冻结快照：幂等、输入变化新快照、漂移失效与任务追溯。
|
||
|
||
合成流程驱动真实 S02、B01/B02/B05 与隔离 PG；不调用模型、不证明完整创作链。
|
||
"""
|
||
|
||
from dataclasses import asdict
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from psycopg.types.json import Jsonb
|
||
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 应用配置
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
|
||
工具集 = ("read_history_document", "query_work_plan", "query_facts_as_of")
|
||
|
||
|
||
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 快照环境(内置结构测试库):
|
||
应用测试库 = 内置结构测试库
|
||
登记 = 流程登记()
|
||
装配 = 构建(应用配置(应用测试库[用途.生产].引用, "test"), 流程=登记)
|
||
with 装配.生命周期():
|
||
作者 = 调用身份("freeze-author", None, 用途.生产, 内容用途.生成)
|
||
作品, 正文 = 装配.要求作品(), 装配.要求正文()
|
||
结构选择 = 作品.新档案结构(作者, "freeze-work", "work_core", 1)
|
||
作品.保存档案(
|
||
作者,
|
||
"create-freeze-work",
|
||
档案保存(
|
||
"freeze-work",
|
||
0,
|
||
{"名称": "旧码头", "题材定位": "现实", "主题立意": "还信"},
|
||
结构选择,
|
||
),
|
||
)
|
||
for i in range(1, 4):
|
||
作品.添加章节(
|
||
作者, f"add-ch-{i}", "freeze-work", f"ch-{i}", f"第{i}章", 预期目录版本=i - 1
|
||
)
|
||
正文.保存人工(
|
||
作者,
|
||
"body-ch-1",
|
||
"ch-1",
|
||
0,
|
||
正文草稿((段落("p-ch-1", (文本节点("林深停在码头，雨还没停。"),)),)),
|
||
)
|
||
|
||
规划 = 装配.要求规划()
|
||
|
||
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:
|
||
选择 = 世界.新事实结构(作者, "freeze-work", 对象, 类型, 1)
|
||
回执 = 世界.提出事实(
|
||
作者,
|
||
"propose-" + 对象,
|
||
事实提案请求(
|
||
"freeze-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", "林深停在码头"),))
|
||
父 = 保存并确认(
|
||
规划保存(
|
||
"freeze-work",
|
||
0,
|
||
规划.新规划结构(作者, "freeze-work", "outline", 1),
|
||
{"名称": "旧码头"},
|
||
),
|
||
"outline",
|
||
)
|
||
细纲 = 保存并确认(
|
||
规划保存(
|
||
"freeze-work",
|
||
0,
|
||
规划.新规划结构(作者, "freeze-work", "fine_outline", 1),
|
||
细纲内容(父["plan_id"]),
|
||
chapter_id="ch-3",
|
||
),
|
||
"fine-outline",
|
||
)
|
||
|
||
def 读取材料(工具) -> tuple[dict, tuple[来源绑定, ...]]:
|
||
"""经任务只读工具取真实材料与来源绑定；顺序固定为细纲、事实、历史。"""
|
||
材料: dict[str, str] = {}
|
||
绑定: list[来源绑定] = []
|
||
计划 = 工具.调用(工具请求("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"]
|
||
for 结果 in (计划, 事实, 历史):
|
||
绑定.extend(
|
||
来源绑定(s.来源ID, s.数据版本, s.结构哈希, s.投影版本) for s in 结果.来源
|
||
)
|
||
return 材料, tuple(绑定)
|
||
|
||
def 冻结并输出(上下文) -> 步骤结果:
|
||
工具 = 装配.要求只读工具(上下文.领取)
|
||
材料, 绑定 = 读取材料(工具)
|
||
上限 = 上下文.任务.冻结输入["冻结上下文"]["budget"]["context_bytes"]
|
||
装配结果 = 装配材料(材料, tuple(材料), 字节上限=上限)
|
||
回执 = 装配.要求上下文().冻结上下文(上下文.任务.任务ID, 装配结果, 绑定)
|
||
return 步骤结果(
|
||
{
|
||
"快照": {
|
||
"snapshot_id": 回执.快照ID,
|
||
"input_hash": 回执.输入哈希,
|
||
"created": 回执.新建,
|
||
},
|
||
"装配": asdict(装配结果),
|
||
}
|
||
)
|
||
|
||
登记.登记处理器(
|
||
步骤处理器(
|
||
"synthetic-freeze",
|
||
"1",
|
||
冻结并输出,
|
||
"freeze/v1",
|
||
"freeze/v1",
|
||
保护职责="范围核对",
|
||
允许工具=工具集,
|
||
)
|
||
)
|
||
登记.登记类型("synthetic-freeze", 必需保护=("范围核对",))
|
||
登记.登记恢复检查(
|
||
"synthetic-freeze", "1", lambda 任务: 装配.要求上下文().读取范围(任务.任务ID)
|
||
)
|
||
运行 = 装配.任务运行
|
||
assert 运行 is not None
|
||
流程服务(运行, 登记).发布(
|
||
流程定义(
|
||
"synthetic-freeze", "1", (步骤计划("read", "synthetic-freeze", "1", 工具=工具集),)
|
||
)
|
||
)
|
||
|
||
def 开始(范围=None):
|
||
if 范围 is None:
|
||
范围 = 装配.要求上下文().授权任务范围(
|
||
作者,
|
||
"freeze-work",
|
||
2,
|
||
({"chapter_id": "ch-1", "revision": 1},),
|
||
目标章ID="ch-3",
|
||
含事实=True,
|
||
)
|
||
id_ = 运行.创建任务(
|
||
任务请求(
|
||
"synthetic-freeze",
|
||
str(uuid4()),
|
||
作者.作者,
|
||
用途.生产,
|
||
内容用途.生成,
|
||
{},
|
||
"test",
|
||
"test",
|
||
{
|
||
"source_scope": 范围,
|
||
"schema_versions": {},
|
||
"authorization": "synthetic-author-choice",
|
||
"budget": {"context_bytes": 4096},
|
||
"stop_conditions": {"max_reads": 3},
|
||
},
|
||
作品ID="freeze-work",
|
||
),
|
||
"synthetic-freeze",
|
||
"1",
|
||
)
|
||
领取 = 运行.领取步骤("freeze-test", ["synthetic-freeze"], 租期秒=120)
|
||
assert 领取 is not None and 领取.任务ID == id_
|
||
return id_, 领取, 装配.要求只读工具(领取)
|
||
|
||
yield {
|
||
"装配": 装配,
|
||
"库组": 应用测试库,
|
||
"作者": 作者,
|
||
"正文": 正文,
|
||
"规划": 规划,
|
||
"开始": 开始,
|
||
"读取材料": 读取材料,
|
||
"保存并确认": 保存并确认,
|
||
"父": 父,
|
||
"细纲": 细纲,
|
||
"确认事实": 确认事实,
|
||
}
|
||
|
||
|
||
def 事件绑定(装配, task_id: str) -> tuple[来源绑定, ...]:
|
||
"""从工具读取事件的真实留痕重建来源绑定；不经过处理器内存。"""
|
||
事件 = 装配.任务运行.续接事件(task_id).事件
|
||
return tuple(
|
||
来源绑定(r["source_id"], r["revision"], r["schema_hash"], r["projection_version"])
|
||
for e in 事件
|
||
if e.类型 == 事件类型.工具读取
|
||
for r in e.载荷["source_refs"]
|
||
)
|
||
|
||
|
||
def 重建装配(装配记录: dict) -> 材料装配结果:
|
||
return 材料装配结果(
|
||
入选来源=tuple(装配记录["入选来源"]),
|
||
文本=装配记录["文本"],
|
||
字节数=装配记录["字节数"],
|
||
省略=tuple(tuple(项) for 项 in 装配记录["省略"]),
|
||
材料区间=tuple(tuple(项) for 项 in 装配记录.get("材料区间", ())),
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-snapshot-lifecycle",
|
||
environment="真实隔离PostgreSQL",
|
||
given="授权任务范围并完成工具读取与材料装配的真实任务",
|
||
when="服务端冻结上下文并按任务回读快照",
|
||
then="回读内容与冻结一致；快照与任务、来源版本绑定可追溯",
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_完整链路冻结回读一致且任务可追溯__b09311(快照环境):
|
||
环境 = 快照环境
|
||
装配, 开始 = 环境["装配"], 环境["开始"]
|
||
task_id, 领取, _ = 开始()
|
||
结果 = 装配.任务运行.执行一步(领取)
|
||
回执 = 结果.输出["快照"]
|
||
assert 回执["created"] is True
|
||
记录集 = 装配.要求上下文().读取任务快照(task_id)
|
||
assert len(记录集) == 1
|
||
快照 = 记录集[0]
|
||
assert 快照.快照ID == 回执["snapshot_id"]
|
||
assert 快照.任务ID == task_id and 快照.作品ID == "freeze-work"
|
||
assert 快照.输入哈希 == 回执["input_hash"]
|
||
assert 快照.当前状态 == "有效" and 快照.失效说明 is None
|
||
记录 = 结果.输出["装配"]
|
||
assert 快照.装配["入选"] == list(记录["入选来源"])
|
||
assert 快照.装配["省略"] == []
|
||
assert 快照.装配["文本"] == 记录["文本"]
|
||
assert 快照.装配["字节数"] == 记录["字节数"] == len(记录["文本"].encode("utf-8"))
|
||
assert 快照.装配["文本哈希"]
|
||
绑定 = {b.来源ID: b for b in 快照.来源绑定}
|
||
assert {环境["细纲"]["plan_id"], "lin-shen"} <= 绑定.keys()
|
||
assert all((b.数据版本, b.结构哈希, b.投影版本) for b in 快照.来源绑定)
|
||
按身份 = 装配.要求上下文().读取快照(回执["snapshot_id"])
|
||
assert (按身份.输入哈希, 按身份.任务ID) == (快照.输入哈希, task_id)
|
||
# 消费绑定与 S02 工具读取留痕一致，不另造第二套事件账本
|
||
assert set(事件绑定(装配, task_id)) == set(快照.来源绑定)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-snapshot-idempotent",
|
||
environment="真实隔离PostgreSQL",
|
||
given="同一任务同一输入的两次冻结",
|
||
when="重复调用冻结",
|
||
then="同一快照身份与哈希，表内仅一行，不改写",
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_重复冻结同一输入幂等为一行__b09312(快照环境):
|
||
环境 = 快照环境
|
||
装配, 开始 = 环境["装配"], 环境["开始"]
|
||
task_id, 领取, _ = 开始()
|
||
结果 = 装配.任务运行.执行一步(领取)
|
||
首次 = 结果.输出["快照"]
|
||
再次 = 装配.要求上下文().冻结上下文(
|
||
task_id, 重建装配(结果.输出["装配"]), 事件绑定(装配, task_id)
|
||
)
|
||
assert 再次.快照ID == 首次["snapshot_id"]
|
||
assert 再次.输入哈希 == 首次["input_hash"]
|
||
assert 再次.新建 is False
|
||
assert len(装配.要求上下文().读取任务快照(task_id)) == 1
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-snapshot-new-input",
|
||
environment="真实隔离PostgreSQL",
|
||
given="装配输入（入选或省略）变化后再次冻结",
|
||
when="冻结并回读新旧快照",
|
||
then="产生新快照新身份；旧快照字段原样保留",
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_装配输入变化产生新快照且不改写旧快照__b09313(快照环境):
|
||
环境 = 快照环境
|
||
装配, 开始, 读取材料 = 环境["装配"], 环境["开始"], 环境["读取材料"]
|
||
task_id, _, 工具 = 开始()
|
||
服务 = 装配.要求上下文()
|
||
材料, 绑定 = 读取材料(工具)
|
||
全套 = 装配材料(材料, tuple(材料), 字节上限=4096)
|
||
# 同一批真实材料、压减一字节预算：末位历史正文被整项省略
|
||
压减 = 装配材料(材料, (), 字节上限=全套.字节数 - 1)
|
||
assert 压减.省略 == (("历史:ch-1", "BUDGET_EXCEEDED"),)
|
||
首次 = 服务.冻结上下文(task_id, 全套, 绑定)
|
||
再次 = 服务.冻结上下文(task_id, 压减, 绑定)
|
||
assert 再次.新建 is True
|
||
assert 再次.快照ID != 首次.快照ID and 再次.输入哈希 != 首次.输入哈希
|
||
快照集 = 服务.读取任务快照(task_id)
|
||
assert len(快照集) == 2
|
||
旧 = 服务.读取快照(首次.快照ID)
|
||
assert 旧.装配["字节数"] == 全套.字节数 and 旧.装配["省略"] == []
|
||
新 = 服务.读取快照(再次.快照ID)
|
||
assert 新.装配["省略"] == [["历史:ch-1", "BUDGET_EXCEEDED"]]
|
||
assert 新.装配["字节数"] == 压减.字节数 < 全套.字节数
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-snapshot-drift-invalid",
|
||
environment="真实隔离PostgreSQL",
|
||
given="快照冻结后细纲重确认、正文重存或事实系统版本前进",
|
||
when="回读快照",
|
||
then="当前状态标注 SOURCE_STALE 与失效说明；历史快照字段不被改写",
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
@pytest.mark.parametrize(
|
||
"漂移源",
|
||
["plan", "document", "facts"],
|
||
ids=["plan-revised", "document-revised", "facts-advanced"],
|
||
)
|
||
def test_来源漂移后回读明确失效且历史不改写__b09314(快照环境, 漂移源):
|
||
环境 = 快照环境
|
||
装配, 作者, 开始 = 环境["装配"], 环境["作者"], 环境["开始"]
|
||
task_id, 领取, _ = 开始()
|
||
结果 = 装配.任务运行.执行一步(领取)
|
||
身份 = 结果.输出["快照"]["snapshot_id"]
|
||
服务 = 装配.要求上下文()
|
||
冻结前 = 服务.读取快照(身份)
|
||
assert 冻结前.当前状态 == "有效"
|
||
if 漂移源 == "plan":
|
||
内容 = 细纲内容(环境["父"]["plan_id"])
|
||
内容["chapterGoal"] = "作者改过的目标。"
|
||
环境["保存并确认"](
|
||
规划保存(
|
||
"freeze-work",
|
||
1,
|
||
环境["规划"].新规划结构(作者, "freeze-work", "fine_outline", 1),
|
||
内容,
|
||
chapter_id="ch-3",
|
||
),
|
||
"revise-outline",
|
||
)
|
||
elif 漂移源 == "document":
|
||
环境["正文"].保存人工(
|
||
作者, "freeze-rev-2", "ch-1", 1, 正文草稿((段落("p-2", (文本节点("改过的正文"),)),))
|
||
)
|
||
else:
|
||
环境["确认事实"]("late-comer", "character", {"名称": "后来的事实"}, 独立=True)
|
||
漂移后 = 服务.读取快照(身份)
|
||
assert 漂移后.当前状态 == "SOURCE_STALE"
|
||
assert 漂移后.失效说明 and "需重新授权" in 漂移后.失效说明
|
||
assert 漂移后.快照ID == 身份
|
||
assert 漂移后.输入哈希 == 冻结前.输入哈希
|
||
assert 漂移后.装配 == 冻结前.装配
|
||
assert 漂移后.来源绑定 == 冻结前.来源绑定
|
||
assert len(服务.读取任务快照(task_id)) == 1
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-snapshot-reauthorize",
|
||
environment="真实隔离PostgreSQL",
|
||
given="来源版本变化后按新版本重新授权任务范围",
|
||
when="再次冻结",
|
||
then="经重新授权形成新快照；不静默续用旧输入",
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_来源版本变化经重新授权产生新快照__b09315(快照环境):
|
||
环境 = 快照环境
|
||
装配, 作者, 开始 = 环境["装配"], 环境["作者"], 环境["开始"]
|
||
task_1, 领取_1, _ = 开始()
|
||
旧 = 装配.任务运行.执行一步(领取_1).输出["快照"]
|
||
内容 = 细纲内容(环境["父"]["plan_id"])
|
||
内容["chapterGoal"] = "作者改过的目标。"
|
||
环境["保存并确认"](
|
||
规划保存(
|
||
"freeze-work",
|
||
1,
|
||
环境["规划"].新规划结构(作者, "freeze-work", "fine_outline", 1),
|
||
内容,
|
||
chapter_id="ch-3",
|
||
),
|
||
"revise-outline",
|
||
)
|
||
范围 = 装配.要求上下文().授权任务范围(
|
||
作者,
|
||
"freeze-work",
|
||
2,
|
||
({"chapter_id": "ch-1", "revision": 1},),
|
||
目标章ID="ch-3",
|
||
含事实=True,
|
||
)
|
||
task_2, 领取_2, _ = 开始(范围)
|
||
结果 = 装配.任务运行.执行一步(领取_2)
|
||
新 = 结果.输出["快照"]
|
||
assert 新["snapshot_id"] != 旧["snapshot_id"] and 新["input_hash"] != 旧["input_hash"]
|
||
服务 = 装配.要求上下文()
|
||
旧快照 = 服务.读取快照(旧["snapshot_id"])
|
||
新快照 = 服务.读取快照(新["snapshot_id"])
|
||
assert 旧快照.当前状态 == "SOURCE_STALE"
|
||
assert 新快照.当前状态 == "有效"
|
||
assert 新快照.任务ID == task_2 and 旧快照.任务ID == task_1
|
||
旧绑定 = {b.来源ID: b.数据版本 for b in 旧快照.来源绑定}
|
||
新绑定 = {b.来源ID: b.数据版本 for b in 新快照.来源绑定}
|
||
旧计划ID = 环境["细纲"]["plan_id"]
|
||
新计划ID = 新快照.冻结范围["target_plan"]["plan_id"]
|
||
if 新计划ID == 旧计划ID:
|
||
assert 新绑定[新计划ID] != 旧绑定[旧计划ID]
|
||
else:
|
||
assert 新计划ID not in 旧绑定
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-snapshot-binding-denied",
|
||
environment="真实隔离PostgreSQL",
|
||
given="携带冻结范围以外来源绑定的冻结请求",
|
||
when="调用冻结",
|
||
then="明确拒绝，不落部分快照",
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_绑定越出冻结范围拒绝冻结__b09316(快照环境):
|
||
环境 = 快照环境
|
||
装配, 开始, 读取材料 = 环境["装配"], 环境["开始"], 环境["读取材料"]
|
||
task_id, _, 工具 = 开始()
|
||
材料, 绑定 = 读取材料(工具)
|
||
计划绑定 = next(b for b in 绑定 if b.来源ID == 环境["细纲"]["plan_id"])
|
||
文本 = 材料["目标章细纲"]
|
||
越界装配 = 材料装配结果(
|
||
入选来源=("越界材料",), 文本=文本, 字节数=len(文本.encode("utf-8")), 省略=()
|
||
)
|
||
越界绑定 = (计划绑定, 来源绑定("not-in-scope", "1", "hash", "proj"))
|
||
with pytest.raises(上下文错误) as 异常:
|
||
装配.要求上下文().冻结上下文(task_id, 越界装配, 越界绑定)
|
||
assert 异常.value.错误码 == "SCOPE_DENIED"
|
||
assert 装配.要求上下文().读取任务快照(task_id) == ()
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-ctx-snapshot-conflict-row",
|
||
environment="真实隔离PostgreSQL",
|
||
given="存储级直插与既有 snapshot_id 相同但 input_hash/载荷不同的行",
|
||
when="再次以同身份冻结/读取",
|
||
then=["CONTEXT_SNAPSHOT_CONFLICT 拒绝；原快照行不被改写"],
|
||
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
||
)
|
||
def test_存储级同身份不同输入哈希冲突拒绝且不改写原行__b09317(快照环境):
|
||
"""同一快照身份对应不同输入哈希时存储层拒绝写入且不改写原行。
|
||
|
||
快照身份由任务与输入哈希派生，公开路径不可达本分支；经维护连接
|
||
直插同 ID 但不同 input_hash 与载荷的行后重放真实冻结，触发防御校验。
|
||
直插必须落在任务所在库，另借新库会因外键不成立而失败。
|
||
"""
|
||
环境 = 快照环境
|
||
应用测试库 = 环境["库组"]
|
||
装配, 开始, 读取材料 = 环境["装配"], 环境["开始"], 环境["读取材料"]
|
||
task_id, _, 工具 = 开始()
|
||
服务 = 装配.要求上下文()
|
||
材料, 绑定 = 读取材料(工具)
|
||
装配结果 = 装配材料(材料, tuple(材料), 字节上限=4096)
|
||
范围 = 服务.读取范围(task_id)
|
||
哈希 = 输入哈希(快照载荷(task_id, 范围, 装配结果, 绑定))
|
||
身份 = 快照身份(task_id, 哈希)
|
||
占位哈希 = "deadbeef" * 8
|
||
assert 占位哈希 != 哈希
|
||
占位装配 = {"入选": [], "省略": [], "字节数": 1, "文本哈希": 占位哈希, "文本": "占位"}
|
||
with 应用测试库[用途.维护].连接() as 连, 连.transaction():
|
||
连.execute(
|
||
"INSERT INTO muse_context_snapshot (snapshot_id,run_purpose,task_id,work_id,"
|
||
"input_hash,frozen_scope,assembly,source_bindings,created_by) "
|
||
"VALUES (%s,'production',%s,%s,%s,%s,%s,%s,%s)",
|
||
(
|
||
身份,
|
||
task_id,
|
||
"freeze-work",
|
||
占位哈希,
|
||
Jsonb(范围),
|
||
Jsonb(占位装配),
|
||
Jsonb([]),
|
||
"conflict-probe",
|
||
),
|
||
)
|
||
with pytest.raises(上下文错误) as 异常:
|
||
服务.冻结上下文(task_id, 装配结果, 绑定)
|
||
assert 异常.value.错误码 == "CONTEXT_SNAPSHOT_CONFLICT"
|
||
assert 身份 in 异常.value.说明
|
||
# 原行不被改写：仍是直插的占位哈希与占位装配，任务名下仅此一行
|
||
快照集 = 服务.读取任务快照(task_id)
|
||
assert len(快照集) == 1
|
||
assert 快照集[0].快照ID == 身份 and 快照集[0].输入哈希 == 占位哈希
|
||
assert 快照集[0].装配["文本"] == "占位" and 快照集[0].来源绑定 == ()
|
||
# 冲突只拒绝同身份不同哈希：不同输入的新冻结仍正常落库
|
||
压减 = 装配材料(材料, (), 字节上限=装配结果.字节数 - 1)
|
||
新回执 = 服务.冻结上下文(task_id, 压减, 绑定)
|
||
assert 新回执.新建 is True and 新回执.快照ID != 身份
|
||
assert len(服务.读取任务快照(task_id)) == 2
|