d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
556 lines
22 KiB
Python
556 lines
22 KiB
Python
"""确定性回放与依赖复检的真实链路：探索留痕、冻结、回放一致、漂移失效、恢复阻断。
|
|
|
|
真实 PG + S02 全链路，不调用模型；漂移用例各自指明失效项，不做字符串包含冒充。
|
|
"""
|
|
|
|
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 应用配置
|
|
|
|
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 装配.生命周期():
|
|
服务 = 装配.要求上下文()
|
|
作者 = 调用身份("replay-author", None, 用途.生产, 内容用途.生成)
|
|
作品, 正文 = 装配.要求作品(), 装配.要求正文()
|
|
结构选择 = 作品.新档案结构(作者, "replay-work", "work_core", 1)
|
|
作品.保存档案(
|
|
作者,
|
|
"create-replay-work",
|
|
档案保存(
|
|
"replay-work",
|
|
0,
|
|
{"名称": "旧码头", "题材定位": "现实", "主题立意": "还信"},
|
|
结构选择,
|
|
),
|
|
)
|
|
for i in range(1, 4):
|
|
作品.添加章节(
|
|
作者, f"add-ch-{i}", "replay-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:
|
|
选择 = 世界.新事实结构(作者, "replay-work", 对象, 类型, 1)
|
|
回执 = 世界.提出事实(
|
|
作者,
|
|
"propose-" + 对象,
|
|
事实提案请求(
|
|
"replay-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", "林深停在码头"),))
|
|
父 = 保存并确认(
|
|
规划保存(
|
|
"replay-work",
|
|
0,
|
|
规划.新规划结构(作者, "replay-work", "outline", 1),
|
|
{"名称": "旧码头"},
|
|
),
|
|
"outline",
|
|
)
|
|
保存并确认(
|
|
规划保存(
|
|
"replay-work",
|
|
0,
|
|
规划.新规划结构(作者, "replay-work", "fine_outline", 1),
|
|
细纲内容(父["plan_id"]),
|
|
chapter_id="ch-3",
|
|
),
|
|
"fine-outline",
|
|
)
|
|
|
|
探索留痕: dict = {"清单": (), "首读": ()}
|
|
|
|
def 探索并冻结(上下文) -> 步骤结果:
|
|
工具 = 装配.要求只读工具(上下文.领取)
|
|
清单: tuple[探索条目, ...] = ()
|
|
首读: list[tuple[str, dict]] = []
|
|
绑定集: dict[str, 来源绑定] = {}
|
|
材料: dict[str, str] = {}
|
|
|
|
def 读(调用ID: str, 名称: str, 参数: dict):
|
|
nonlocal 清单
|
|
结果 = 工具.调用(工具请求(调用ID, 名称, 参数))
|
|
清单 = 记录探索(清单, 调用ID, 名称, 参数, 结果)
|
|
首读.append((调用ID, 结果.内容))
|
|
for 源 in 结果.来源:
|
|
绑定集.setdefault(
|
|
源.来源ID, 来源绑定(源.来源ID, 源.数据版本, 源.结构哈希, 源.投影版本)
|
|
)
|
|
return 结果
|
|
|
|
计划 = 读("plan-1", "query_work_plan", {})
|
|
事实 = 读("facts-1", "query_facts_as_of", {})
|
|
读("hist-1", "read_history_document", {"chapter_id": "ch-1", "revision": 1, "end": 6})
|
|
读("hist-2", "read_history_document", {"chapter_id": "ch-1", "revision": 1, "start": 6})
|
|
材料["目标章细纲"] = 计划.内容["content"]["chapterGoal"]
|
|
for 项 in 事实.内容["facts"]:
|
|
名称 = 项["content"].get("名称")
|
|
if 名称:
|
|
材料["事实:" + 项["object_id"]] = 名称
|
|
材料["历史:ch-1"] = 首读[2][1]["text"] + 首读[3][1]["text"]
|
|
上限 = 上下文.任务.冻结输入["冻结上下文"]["budget"]["context_bytes"]
|
|
回执 = 服务.冻结上下文(
|
|
上下文.任务.任务ID,
|
|
装配材料(材料, tuple(材料), 字节上限=上限),
|
|
tuple(绑定集.values()),
|
|
)
|
|
探索留痕["清单"] = 清单
|
|
探索留痕["首读"] = tuple(首读)
|
|
return 步骤结果(
|
|
{
|
|
"快照": {
|
|
"snapshot_id": 回执.快照ID,
|
|
"input_hash": 回执.输入哈希,
|
|
"created": 回执.新建,
|
|
}
|
|
}
|
|
)
|
|
|
|
登记.登记处理器(
|
|
步骤处理器(
|
|
"synthetic-replay",
|
|
"1",
|
|
探索并冻结,
|
|
"replay/v1",
|
|
"replay/v1",
|
|
保护职责="范围核对",
|
|
允许工具=工具集,
|
|
)
|
|
)
|
|
登记.登记类型("synthetic-replay", 必需保护=("范围核对",))
|
|
# 恢复检查注册真实依赖重验：授权、来源、结构与版本，不再只是只读范围的 lambda
|
|
登记.登记恢复检查("synthetic-replay", "1", 依赖重验检查(服务))
|
|
运行 = 装配.任务运行
|
|
assert 运行 is not None
|
|
流程服务(运行, 登记).发布(
|
|
流程定义(
|
|
"synthetic-replay", "1", (步骤计划("read", "synthetic-replay", "1", 工具=工具集),)
|
|
)
|
|
)
|
|
|
|
def 开始(范围=None):
|
|
if 范围 is None:
|
|
范围 = 服务.授权任务范围(
|
|
作者,
|
|
"replay-work",
|
|
2,
|
|
({"chapter_id": "ch-1", "revision": 1},),
|
|
目标章ID="ch-3",
|
|
含事实=True,
|
|
)
|
|
id_ = 运行.创建任务(
|
|
任务请求(
|
|
"synthetic-replay",
|
|
str(uuid4()),
|
|
作者.作者,
|
|
用途.生产,
|
|
内容用途.生成,
|
|
{},
|
|
"test",
|
|
"test",
|
|
{
|
|
"source_scope": 范围,
|
|
"schema_versions": {},
|
|
"authorization": "synthetic-author-choice",
|
|
"budget": {"context_bytes": 4096},
|
|
"stop_conditions": {"max_reads": 6},
|
|
},
|
|
作品ID="replay-work",
|
|
),
|
|
"synthetic-replay",
|
|
"1",
|
|
)
|
|
领取 = 运行.领取步骤("replay-test", ["synthetic-replay"], 租期秒=120)
|
|
assert 领取 is not None and 领取.任务ID == id_
|
|
return id_, 领取, 装配.要求只读工具(领取)
|
|
|
|
def 改细纲() -> None:
|
|
内容 = 细纲内容(父["plan_id"])
|
|
内容["chapterGoal"] = "作者改过的目标。"
|
|
保存并确认(
|
|
规划保存(
|
|
"replay-work",
|
|
1,
|
|
规划.新规划结构(作者, "replay-work", "fine_outline", 1),
|
|
内容,
|
|
chapter_id="ch-3",
|
|
),
|
|
"revise-outline",
|
|
)
|
|
|
|
def 改正文() -> None:
|
|
正文.保存人工(
|
|
作者, "freeze-rev-2", "ch-1", 1, 正文草稿((段落("p-2", (文本节点("改过的正文"),)),))
|
|
)
|
|
|
|
def 前进事实() -> None:
|
|
确认事实("late-comer", "character", {"名称": "后来的事实"}, 独立=True)
|
|
|
|
yield {
|
|
"装配": 装配,
|
|
"服务": 服务,
|
|
"作者": 作者,
|
|
"规划": 规划,
|
|
"正文": 正文,
|
|
"开始": 开始,
|
|
"改细纲": 改细纲,
|
|
"改正文": 改正文,
|
|
"前进事实": 前进事实,
|
|
"探索留痕": 探索留痕,
|
|
}
|
|
|
|
|
|
def 工具读取事件数(装配, task_id: str) -> int:
|
|
return sum(1 for e in 装配.任务运行.续接事件(task_id).事件 if e.类型 == 事件类型.工具读取)
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-ctx-replay-deterministic",
|
|
environment="真实隔离PostgreSQL",
|
|
given="已完成探索并冻结的真实任务",
|
|
when="按清单确定性回放",
|
|
then="回放与首读逐项一致（含中文码点片段）；回放不追加工具读取留痕",
|
|
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
|
)
|
|
def test_回放与首读逐项一致且不再留痕__b7e0c1(回放环境):
|
|
环境 = 回放环境
|
|
装配, 服务 = 环境["装配"], 环境["服务"]
|
|
task_id, 领取, _ = 环境["开始"]()
|
|
结果 = 装配.任务运行.执行一步(领取)
|
|
清单 = 环境["探索留痕"]["清单"]
|
|
首读 = 环境["探索留痕"]["首读"]
|
|
assert [项.调用ID for 项 in 清单] == ["plan-1", "facts-1", "hist-1", "hist-2"]
|
|
留痕数 = 工具读取事件数(装配, task_id)
|
|
assert 留痕数 == 4
|
|
回放结果 = 服务.回放探索(task_id, 清单)
|
|
# 回放确定性：逐项与首读完全相等，含中文码点片段
|
|
assert [(条.调用ID, 条.内容) for 条 in 回放结果.逐项] == list(首读)
|
|
assert 回放结果.逐项[2].内容["text"] == "林深停在码头"
|
|
assert 回放结果.逐项[3].内容["text"] == "，雨还没停。"
|
|
assert [条.消费哈希 for 条 in 回放结果.逐项] == [项.消费哈希 for 项 in 清单]
|
|
assert [条.来源 for 条 in 回放结果.逐项] == [项.来源引用 for 项 in 清单]
|
|
# 回放只用固定材料：不追加工具读取留痕
|
|
assert 工具读取事件数(装配, task_id) == 留痕数
|
|
# 冻结快照路径与任务内冻结范围路径同一结果
|
|
快照 = 服务.读取任务快照(task_id)[0]
|
|
assert 快照.快照ID == 结果.输出["快照"]["snapshot_id"]
|
|
快照回放 = 回放(
|
|
服务.数据库,
|
|
快照.冻结范围,
|
|
工具范围从冻结(task_id, 快照.冻结范围),
|
|
清单,
|
|
服务.规划引用解析,
|
|
)
|
|
assert 快照回放.逐项 == 回放结果.逐项
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-ctx-replay-source-drift",
|
|
environment="真实隔离PostgreSQL",
|
|
given="回放前正文重存、细纲重确认或事实系统版本前进",
|
|
when="按清单回放",
|
|
then="SOURCE_STALE 且指明漂移调用与来源",
|
|
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
|
)
|
|
@pytest.mark.parametrize(
|
|
"漂移源",
|
|
["document", "plan", "facts"],
|
|
ids=["document-revised", "plan-reconfirmed", "facts-advanced"],
|
|
)
|
|
def test_来源漂移后回放明确失效并指明项__b7e0c2(回放环境, 漂移源):
|
|
环境 = 回放环境
|
|
装配, 服务 = 环境["装配"], 环境["服务"]
|
|
task_id, 领取, _ = 环境["开始"]()
|
|
装配.任务运行.执行一步(领取)
|
|
清单 = 环境["探索留痕"]["清单"]
|
|
if 漂移源 == "document":
|
|
环境["改正文"]()
|
|
指明 = ("hist-1", "ch-1")
|
|
elif 漂移源 == "plan":
|
|
环境["改细纲"]()
|
|
指明 = ("plan-1", "ch-3", "细纲")
|
|
else:
|
|
环境["前进事实"]()
|
|
指明 = ("facts-1", "事实系统版本")
|
|
with pytest.raises(上下文错误) as 异常:
|
|
服务.回放探索(task_id, 清单)
|
|
assert 异常.value.错误码 == "SOURCE_STALE"
|
|
for 项 in 指明:
|
|
assert 项 in 异常.value.说明
|
|
# 失效不采纳内容：快照历史不改写，仍只有一条冻结记录
|
|
assert len(服务.读取任务快照(task_id)) == 1
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-ctx-replay-out-of-scope",
|
|
environment="真实隔离PostgreSQL",
|
|
given="越界参数的清单",
|
|
when="经真实任务回放",
|
|
then="明确拒绝，不返回部分材料",
|
|
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
|
)
|
|
def test_越界参数清单回放拒绝__b7e0c3(回放环境):
|
|
环境 = 回放环境
|
|
装配, 服务 = 环境["装配"], 环境["服务"]
|
|
task_id, 领取, _ = 环境["开始"]()
|
|
装配.任务运行.执行一步(领取)
|
|
清单 = 环境["探索留痕"]["清单"]
|
|
历史 = next(项 for 项 in 清单 if 项.工具名 == "read_history_document")
|
|
|
|
def 越界条目(调用ID: str, 参数: dict) -> 探索条目:
|
|
return 探索条目(
|
|
调用ID=调用ID,
|
|
工具名="read_history_document",
|
|
顺序=len(清单) + 1,
|
|
参数=参数,
|
|
来源引用=历史.来源引用,
|
|
消费哈希=历史.消费哈希,
|
|
)
|
|
|
|
for 参数 in (
|
|
{"chapter_id": "ch-1", "revision": 2},
|
|
{"chapter_id": "ch-2", "revision": 1},
|
|
{"chapter_id": "ch-1", "revision": 1, "as_of": 3},
|
|
):
|
|
with pytest.raises(上下文错误) as 异常:
|
|
服务.回放探索(task_id, 清单 + (越界条目("越界", 参数),))
|
|
assert 异常.value.错误码 == "SCOPE_DENIED"
|
|
assert "越界" in 异常.value.说明
|
|
# 未追加越界条目的原清单仍可回放
|
|
服务.回放探索(task_id, 清单)
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-ctx-deps-recheck-blocks",
|
|
environment="真实隔离PostgreSQL",
|
|
given="注册真实依赖重验检查的任务在来源漂移后",
|
|
when="重验执行范围",
|
|
then="抛错阻断，不静默续用旧输入",
|
|
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
|
)
|
|
def test_依赖重验检查漂移后阻断任务重验__b7e0c4(回放环境):
|
|
环境 = 回放环境
|
|
装配 = 环境["装配"]
|
|
运行 = 装配.任务运行
|
|
task_id, 领取, _ = 环境["开始"]()
|
|
运行.重验执行范围(领取) # 漂移前真实检查通过
|
|
环境["前进事实"]()
|
|
with pytest.raises(上下文错误) as 异常:
|
|
运行.重验执行范围(领取)
|
|
assert 异常.value.错误码 == "SOURCE_STALE"
|
|
assert "事实系统版本" in 异常.value.说明 and "来源锁定失效" in 异常.value.说明
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-ctx-deps-lock-status",
|
|
environment="真实隔离PostgreSQL",
|
|
given="有效与漂移两种冻结范围",
|
|
when="核对依赖",
|
|
then="有效输出逐来源当前版本锁定依据；漂移输出具体变化项",
|
|
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
|
)
|
|
def test_核对依赖输出锁定依据与变化项__b7e0c5(回放环境):
|
|
环境 = 回放环境
|
|
装配, 服务 = 环境["装配"], 环境["服务"]
|
|
task_id, 领取, _ = 环境["开始"]()
|
|
装配.任务运行.执行一步(领取)
|
|
快照 = 服务.读取任务快照(task_id)[0]
|
|
with 服务.数据库.连接(只读=True) as 连:
|
|
状态 = 核对依赖(连, 快照.冻结范围, 服务.规划引用解析, 快照.来源绑定)
|
|
assert 状态.有效 is True and 状态.变化项 == ()
|
|
锁定 = {项.身份: 项 for 项 in 状态.锁定依据}
|
|
目标 = 快照.冻结范围["target_plan"]
|
|
assert 锁定[目标["plan_id"]].种类 == "plan"
|
|
assert 锁定[目标["plan_id"]].版本 == str(目标["revision"])
|
|
assert 锁定[目标["plan_id"]].哈希 == 目标["content_hash"]
|
|
assert 锁定[目标["plan_id"]].投影版本
|
|
文档ID = 快照.冻结范围["documents"][0]["document_id"]
|
|
assert 锁定[文档ID].种类 == "document" and 锁定[文档ID].哈希
|
|
assert any(项.种类 == "facts" for 项 in 状态.锁定依据)
|
|
环境["前进事实"]()
|
|
with 服务.数据库.连接(只读=True) as 连:
|
|
漂移 = 核对依赖(连, 快照.冻结范围, 服务.规划引用解析, 快照.来源绑定)
|
|
assert 漂移.有效 is False
|
|
assert any("事实系统版本" in 项 for 项 in 漂移.变化项)
|
|
assert not any(项.种类 == "facts" for 项 in 漂移.锁定依据)
|
|
# 文稿漂移连带细纲引用失效时，仍输出结构化变化项而不是向外抛错
|
|
环境["改正文"]()
|
|
with 服务.数据库.连接(只读=True) as 连:
|
|
文稿漂移 = 核对依赖(连, 快照.冻结范围, 服务.规划引用解析, 快照.来源绑定)
|
|
assert 文稿漂移.有效 is False
|
|
assert any("ch-1" in 项 and "前进到 2" in 项 for 项 in 文稿漂移.变化项)
|
|
assert not any(项.种类 == "document" for 项 in 文稿漂移.锁定依据)
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-ctx-replay-auth-denied",
|
|
environment="真实隔离PostgreSQL",
|
|
given="任务冻结范围的授权/身份不一致（恢复冻结范围失败）",
|
|
when="调用回放探索",
|
|
then=["SCOPE_DENIED 拒绝回放，不沿用已撤回或不一致的授权"],
|
|
contract="docs/系统架构/新版设计/模块设计/B09-上下文.md",
|
|
)
|
|
def test_回放授权重验失败拒绝回放__b7e0c6(回放环境):
|
|
"""恢复冻结范围失败（篡改冻结用途或他人作者）时回放整体 SCOPE_DENIED。
|
|
|
|
回放入口先按当前授权重验任务冻结范围；身份或用途不一致即拒绝，
|
|
不沿用已记录清单的读取权限。同一清单在授权一致的任务上仍可回放。
|
|
"""
|
|
环境 = 回放环境
|
|
装配, 服务, 开始, 作者 = 环境["装配"], 环境["服务"], 环境["开始"], 环境["作者"]
|
|
正常ID, 领取, _ = 开始()
|
|
装配.任务运行.执行一步(领取)
|
|
清单 = 环境["探索留痕"]["清单"]
|
|
范围 = 服务.读取范围(正常ID)
|
|
篡改集 = (
|
|
("run_purpose", "evaluation", "任务用途与冻结范围用途不一致"),
|
|
("authorized_by", "someone-else", "冻结范围与任务作者不一致"),
|
|
)
|
|
for 键, 值, 指明 in 篡改集:
|
|
# 篡改任务不经工具领取（领取路径会在供给工具时先行拒绝），直接入库后回放
|
|
越权ID = 装配.任务运行.创建任务(
|
|
任务请求(
|
|
"synthetic-replay",
|
|
str(uuid4()),
|
|
作者.作者,
|
|
用途.生产,
|
|
内容用途.生成,
|
|
{},
|
|
"test",
|
|
"test",
|
|
{
|
|
"source_scope": {**范围, 键: 值},
|
|
"schema_versions": {},
|
|
"authorization": "synthetic-author-choice",
|
|
"budget": {"context_bytes": 4096},
|
|
"stop_conditions": {"max_reads": 6},
|
|
},
|
|
作品ID="replay-work",
|
|
),
|
|
"synthetic-replay",
|
|
"1",
|
|
)
|
|
with pytest.raises(上下文错误) as 异常:
|
|
服务.回放探索(越权ID, 清单)
|
|
assert 异常.value.错误码 == "SCOPE_DENIED"
|
|
assert "任务授权重验未通过" in 异常.value.说明 and 指明 in 异常.value.说明
|
|
# 对照：同一真实清单在授权一致的任务上仍可回放，拒绝只来自授权重验
|
|
服务.回放探索(正常ID, 清单)
|