d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
125 lines
6.4 KiB
Python
125 lines
6.4 KiB
Python
"""抽取版本与补偿：批量原子登记、失败窗口断点续跑、版本不互抹与观察对照。"""
|
||
|
||
import pytest
|
||
|
||
from muse.任务运行.接口 import 任务状态
|
||
from muse.共享.错误 import Muse错误
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
|
||
|
||
def _来源(工具, 标题: str, 章数: int = 4) -> dict:
|
||
正文 = "".join(f"第{i}回 风\n内容{'甲' * 20}。\n" for i in range(1, 章数 + 1))
|
||
return 工具.导入来源(标题, 正文)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w18-18b03c",
|
||
environment="真实隔离 PG",
|
||
given="确定性输入或隔离数据库",
|
||
when="批量原子登记与版本身份",
|
||
then=["行为可复验", "失败与歧义如实呈现"],
|
||
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
|
||
)
|
||
def test_批量原子登记与版本身份__18b03c(研究环境):
|
||
工具 = 研究环境
|
||
来源 = _来源(工具, "样章：原子") # 3 章 → 2 窗（窗口章数默认 3？默认 3 章 1 窗）
|
||
任务 = 工具.发起(来源["source_id"], 理由="题材接近", 窗口章数=2)
|
||
快照 = 工具.装配.任务运行.读取任务(任务["task_id"])
|
||
计划窗数 = len(快照.冻结输入["输入"]["窗计划"])
|
||
for 序 in range(计划窗数):
|
||
工具.剧本.append(
|
||
工具.分析输出([{"name": f"实体{序}", "type": "人物", "evidence": ["内容甲"]}])
|
||
)
|
||
终态 = 工具.跑任务(任务["task_id"])
|
||
assert 终态.状态 == 任务状态.已完成
|
||
版本步 = next(s for s in 终态.步骤 if s["step_id"] == "分析版本登记")
|
||
版本 = 版本步["checkpoint"]["版本"]
|
||
assert 版本["版本ID"] == 任务["版本ID"], "检查点版本身份应与发起预计算一致"
|
||
assert 版本["revision"] == 来源["revision"]
|
||
assert len(版本["完成窗"]) == 计划窗数
|
||
assert 版本["样本依据"]["理由"] == "题材接近"
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w18-18b04d",
|
||
environment="真实隔离 PG",
|
||
given="确定性输入或隔离数据库",
|
||
when="失败窗口断点续跑只补缺失窗",
|
||
then=["行为可复验", "失败与歧义如实呈现"],
|
||
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
|
||
)
|
||
def test_失败窗口断点续跑只补缺失窗__18b04d(研究环境):
|
||
工具 = 研究环境
|
||
来源 = _来源(工具, "样章：断点")
|
||
任务 = 工具.发起(来源["source_id"])
|
||
快照 = 工具.装配.任务运行.读取任务(任务["task_id"])
|
||
计划 = 快照.冻结输入["输入"]["窗计划"]
|
||
assert len(计划) >= 2, "用例需要至少两个窗口"
|
||
# 首窗有效、次窗合同无效：窗口一结果经检查点持久化，任务失败停可解释断点。
|
||
工具.剧本.append(工具.分析输出([{"name": "首窗实体", "type": "人物", "evidence": ["内容甲"]}]))
|
||
工具.剧本.append({"类型": "文本", "文本": {"chapters": []}})
|
||
with pytest.raises((Muse错误, RuntimeError, OSError)):
|
||
工具.跑任务(任务["task_id"])
|
||
失败态 = 工具.装配.任务运行.读取任务(任务["task_id"])
|
||
assert 失败态.状态 == 任务状态.已失败
|
||
中间检查点 = next(s for s in 失败态.步骤 if s["step_id"] == "逐窗提取")
|
||
已存窗 = set((中间检查点.get("checkpoint") or {}).get("窗口结果") or {})
|
||
assert "w001" in 已存窗, "首窗成果应在失败前已持久化"
|
||
assert "w002" not in 已存窗, "失败窗不应伪称完成"
|
||
# 恢复：只补缺失窗；剧本只放一条，若重跑已存窗会因剧本耗尽拿到空对象而再次失败。
|
||
运行 = 工具.装配.任务运行
|
||
运行.控制任务(
|
||
任务["task_id"],
|
||
"post-author",
|
||
任务状态.已失败,
|
||
"恢复",
|
||
命令ID=f"resume-{任务['task_id'][:8]}",
|
||
)
|
||
工具.剧本.append(工具.分析输出([{"name": "次窗实体", "type": "地点", "evidence": ["内容甲"]}]))
|
||
终态 = 工具.跑任务(任务["task_id"])
|
||
assert 终态.状态 == 任务状态.已完成
|
||
版本步 = next(s for s in 终态.步骤 if s["step_id"] == "分析版本登记")
|
||
版本 = 版本步["checkpoint"]["版本"]
|
||
assert sorted(版本["完成窗"]) == sorted(窗["窗ID"] for 窗 in 计划)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w18-18b05e",
|
||
environment="真实隔离 PG",
|
||
given="确定性输入或隔离数据库",
|
||
when="来源前进旧版本结论保留且新拆书指向新版本",
|
||
then=["行为可复验", "失败与歧义如实呈现"],
|
||
contract="docs/系统架构/新版设计/模块设计/B03-资料研究.md",
|
||
)
|
||
def test_来源前进旧版本结论保留且新拆书指向新版本__18b05e(研究环境):
|
||
工具 = 研究环境
|
||
来源 = _来源(工具, "样章：漂移")
|
||
任务 = 工具.发起(来源["source_id"])
|
||
快照 = 工具.装配.任务运行.读取任务(任务["task_id"])
|
||
for _ in 快照.冻结输入["输入"]["窗计划"]:
|
||
工具.剧本.append(
|
||
工具.分析输出([{"name": "旧版实体", "type": "人物", "evidence": ["内容甲"]}])
|
||
)
|
||
# 外部漂移：拆书进行中来源前进到 r2；r1 冻结分析照常完成，不被抹掉。
|
||
工具.导入来源("样章：漂移", "第1回 风\n内容换新版本。" * 3)
|
||
终态 = 工具.跑任务(任务["task_id"])
|
||
assert 终态.状态 == 任务状态.已完成
|
||
旧版本 = next(s for s in 终态.步骤 if s["step_id"] == "分析版本登记")["checkpoint"]["版本"]
|
||
assert 旧版本["revision"] == 1 and 旧版本["content_hash"] == 来源["content_hash"]
|
||
# 新拆书取最新版本：版本身份不同，旧结论仍可回查。
|
||
新任务 = 工具.发起(来源["source_id"])
|
||
assert 新任务["版本ID"] != 旧版本["版本ID"]
|
||
新快照 = 工具.装配.任务运行.读取任务(新任务["task_id"])
|
||
assert 新快照.冻结输入["输入"]["拆书"]["revision"] == 2
|
||
for _ in 新快照.冻结输入["输入"]["窗计划"]:
|
||
工具.剧本.append(
|
||
工具.分析输出([{"name": "新版实体", "type": "人物", "evidence": ["内容换新"]}])
|
||
)
|
||
assert 工具.跑任务(新任务["task_id"]).状态 == 任务状态.已完成
|
||
from muse.资料研究.分析版本 import 版本台账
|
||
|
||
台账 = 版本台账(工具.装配.任务运行.列出任务("post-author"))
|
||
本源 = [行 for 行 in 台账 if 行["source_id"] == 来源["source_id"]]
|
||
assert {行["版本ID"] for 行 in 本源} == {旧版本["版本ID"], 新任务["版本ID"]}
|