6d1f635003
- 场景检索.py：切片语料两路等权RRF（键带slice:前缀），授权join过滤、代次陈旧拒绝、结果带书名预览双路得分；RRF抽公共融合入基础设施 - 参考材料双载体：references节判别式（方法/片段），片段冻结身份带来源内容哈希哨兵，现读核当前版本一致否则SOURCE_STALE - 选段授权slice_references→范围授权→校验→依赖复检→装配→快照绑定全链贯通；片段不登记方法消费 - 范围校验source_ids基集、服务包装、依赖锁定键分派同步；B03接口增读取来源当前指纹（架构门禁禁私表） - 集成a10032-a10036：检索知情候选/未建拒绝/撤回空/片段进快照/来源漂移失效
381 lines
15 KiB
Python
381 lines
15 KiB
Python
"""参考改写真实链路：授权携带参考版本，偏好与参考材料进冻结快照；漂移即失效。"""
|
|
|
|
import json
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
import test_上下文快照与索引写入 as 方法测试
|
|
import test_两阶段写手 as 写手测试
|
|
import test_模型修订任务 as 修订测试
|
|
|
|
from muse.任务运行.接口 import 任务状态
|
|
from muse.作者经验.接口 import 偏好范围, 偏好请求
|
|
from muse.共享.错误 import Muse错误
|
|
from muse.启动 import 装配审校修订
|
|
from muse.正文写作.接口 import 原文选段, 文本节点, 正文草稿, 段落, 段落哈希, 选段写作授权
|
|
from muse.资料研究.接口 import 导入请求, 资料服务
|
|
|
|
pytestmark = pytest.mark.数据库
|
|
生成环境 = 修订测试.生成环境
|
|
|
|
|
|
@pytest.fixture
|
|
def 参考环境(生成环境):
|
|
env = 生成环境
|
|
装配 = env["装配"]
|
|
装配审校修订(
|
|
装配, 写手测试.合成计价(), Path(__file__).parents[2] / "配置/流程模板/审校修订.yaml"
|
|
)
|
|
doc = 正文草稿(
|
|
(段落("p1", (文本节点("机甲的刀锋嗡鸣。"),)), 段落("p2", (文本节点("火花四散。"),)))
|
|
)
|
|
env["正文"].保存人工(env["作者"], "ref-seed", "ch-4", 0, doc)
|
|
saved = env["正文"].读取正文(env["作者"], "ch-4")
|
|
|
|
方法 = 装配.要求知识方法()
|
|
with 方法.数据库.连接() as 连:
|
|
来源 = 资料服务().导入(
|
|
连,
|
|
env["作者"].作者,
|
|
导入请求(
|
|
kind="reference",
|
|
title="机甲打斗参考",
|
|
origin="seed://ref-method",
|
|
content="机甲近身刀法借势卸力，受击反馈优先于光效描写。",
|
|
authorized_uses=("analysis",),
|
|
),
|
|
)
|
|
请求 = 方法测试._提案(str(uuid4()), "机甲近身刀法", str(来源.source_id), owner=env["作者"].作者)
|
|
回执 = 方法.提出方法(env["作者"], str(uuid4()), 请求)
|
|
方法测试._确认(方法, env["作者"], 回执["target_ref"])
|
|
版本 = 方法.读取方法详情(env["作者"], 请求.method_id)["versions"][0]
|
|
|
|
经验 = 装配.要求作者经验()
|
|
经验.保存偏好(
|
|
env["作者"],
|
|
str(uuid4()),
|
|
偏好请求(
|
|
preference_id=str(uuid4()),
|
|
topic="打斗描写",
|
|
text="重视受击反馈，不堆砌形容词。",
|
|
quote="今后重视受击反馈。",
|
|
scope=偏好范围(),
|
|
),
|
|
)
|
|
|
|
选段 = 原文选段("p2", 段落哈希(doc.paragraphs[1]), 0, 5, "火花四散。")
|
|
授权 = 选段写作授权(
|
|
"gen-work",
|
|
"ch-4",
|
|
"main",
|
|
saved["revision"],
|
|
saved["document_hash"],
|
|
"rewrite",
|
|
"改为机甲近身刀法，参考所选方法",
|
|
(选段,),
|
|
(),
|
|
references=(str(版本["version_id"]),),
|
|
)
|
|
env.update(
|
|
授权=授权,
|
|
选段=选段,
|
|
参考版本=str(版本["version_id"]),
|
|
参考方法ID=请求.method_id,
|
|
来源ID=str(来源.source_id),
|
|
)
|
|
return env
|
|
|
|
|
|
def _快照ID(task):
|
|
return next(
|
|
s["checkpoint"]["snapshot_id"]
|
|
for s in task.步骤
|
|
if s.get("checkpoint") and "snapshot_id" in s["checkpoint"]
|
|
)
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w32-a10010",
|
|
environment="隔离PG与合成HTTP提供方",
|
|
given="带参考版本与活跃偏好的选段改写授权",
|
|
when="发起并执行受控修订流程",
|
|
then=["偏好与参考材料入选装配并绑定进快照；写手输入携带材料"],
|
|
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md#参考改写选段改写链路扩展",
|
|
)
|
|
def test_偏好与参考材料进快照与写手输入__a10010(参考环境):
|
|
env = 参考环境
|
|
tid = 修订测试.发起(env, command="ref-revision-1")
|
|
env["剧本"].append(
|
|
{
|
|
"类型": "文本",
|
|
"文本": {
|
|
"edits": [
|
|
{
|
|
"selection_id": env["选段"].selection_id,
|
|
"action": "replace",
|
|
"replacement": "刀锋擦过装甲，反震顺着掌心爬上来。",
|
|
"reason": "合成参考改写",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
task = 修订测试.跑(env, tid)
|
|
assert task.状态 == 任务状态.已完成
|
|
输入 = json.loads(env["收到"][-1]["input"])
|
|
assert "重视受击反馈，不堆砌形容词。" in 输入["装配材料"]
|
|
|
|
快照 = env["装配"].要求上下文().读取快照(_快照ID(task))
|
|
assert 快照.失效说明 is None
|
|
assert "参考方法:" + env["参考版本"] in 快照.装配["入选"]
|
|
assert any(str(绑.来源ID) == env["参考版本"] for 绑 in 快照.来源绑定)
|
|
result = env["装配"].要求审校().读取模型修订(env["作者"], tid)
|
|
assert result["source_kind"] == "selection" and result["outcome"] == "candidate"
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w32-a10011",
|
|
environment="隔离PG与合成HTTP提供方",
|
|
given="携带不存在参考版本的选段授权",
|
|
when="发起受控修订",
|
|
then=["授权范围阶段快速拒绝，不创建任务"],
|
|
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md#参考改写选段改写链路扩展",
|
|
)
|
|
def test_未知参考版本在发起时拒绝__a10011(参考环境):
|
|
env = 参考环境
|
|
env["授权"] = replace(env["授权"], references=("不存在的版本",))
|
|
with pytest.raises(Muse错误):
|
|
修订测试.发起(env, command="ref-revision-bad")
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w32-a10012",
|
|
environment="隔离PG与合成HTTP提供方",
|
|
given="已冻结的参考改写任务",
|
|
when="参考方法推进新版本后读取快照",
|
|
then=["旧参考版本不再当前，快照标注失效且要求重新授权"],
|
|
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md#参考改写选段改写链路扩展",
|
|
)
|
|
def test_参考版本漂移使快照失效__a10012(参考环境):
|
|
env = 参考环境
|
|
tid = 修订测试.发起(env, command="ref-revision-drift")
|
|
svc = env["装配"].任务运行
|
|
claim = svc.领取步骤("revision-worker", ["revision.freeze"])
|
|
assert claim is not None
|
|
svc.执行一步(claim)
|
|
快照ID = _快照ID(svc.读取任务(tid))
|
|
|
|
方法 = env["装配"].要求知识方法()
|
|
新请求 = 方法测试._提案(
|
|
str(uuid4()),
|
|
"机甲近身刀法二版",
|
|
env["来源ID"],
|
|
owner=env["作者"].作者,
|
|
方法ID=env["参考方法ID"],
|
|
)
|
|
回执 = 方法.提出方法(env["作者"], str(uuid4()), 新请求)
|
|
方法测试._确认(方法, env["作者"], 回执["target_ref"])
|
|
|
|
快照 = env["装配"].要求上下文().读取快照(快照ID)
|
|
assert 快照.失效说明 and "参考方法" in 快照.失效说明
|
|
assert "需重新授权" in 快照.失效说明
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w32-a10023",
|
|
environment="隔离PG与合成HTTP提供方",
|
|
given="带入选参考的选段改写授权且修订候选真实交付",
|
|
when="流程完成后查询方法消费台账",
|
|
then=[
|
|
"入选参考按 kind=review 登记消费，关联快照与材料哈希",
|
|
"未入选或未携带参考时不登记",
|
|
"消费计数进入 B09 强信号统计",
|
|
],
|
|
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md#参考改写选段改写链路扩展",
|
|
)
|
|
def test_入选参考交付后登记消费__a10023(参考环境):
|
|
env = 参考环境
|
|
tid = 修订测试.发起(env, command="ref-revision-usage")
|
|
env["剧本"].append(
|
|
{
|
|
"类型": "文本",
|
|
"文本": {
|
|
"edits": [
|
|
{
|
|
"selection_id": env["选段"].selection_id,
|
|
"action": "replace",
|
|
"replacement": "刀锋擦过装甲，反震顺着掌心爬上来。",
|
|
"reason": "合成参考改写",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
task = 修订测试.跑(env, tid)
|
|
assert task.状态 == 任务状态.已完成
|
|
with env["装配"].要求数据库().连接(只读=True) as 连:
|
|
rows = 连.execute(
|
|
"SELECT kind, context_ref, fragment_hash FROM muse_method_usage "
|
|
"WHERE task_id=%s AND method_version_id=%s",
|
|
(tid, env["参考版本"]),
|
|
).fetchall()
|
|
assert len(rows) == 1
|
|
assert rows[0][0] == "review"
|
|
assert rows[0][1] == _快照ID(task)
|
|
# 材料哈希与快照入选的一致性由记录消费内部核对方法消费材料保证，不匹配即拒登。
|
|
assert len(rows[0][2]) == 64
|
|
from muse.知识方法.接口 import 读取消费统计
|
|
|
|
作者ID = env["作者"].作者
|
|
with env["装配"].要求数据库().连接(只读=True) as 连:
|
|
统计 = 读取消费统计(连, 作者ID)
|
|
assert 统计[env["参考版本"]] >= 1
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w32-a10024",
|
|
environment="隔离PG与合成HTTP提供方",
|
|
given="不带参考版本的选段改写授权且候选真实交付",
|
|
when="流程完成后查询方法消费台账",
|
|
then=["无参考即无新增方法消费记录"],
|
|
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md#参考改写选段改写链路扩展",
|
|
)
|
|
def test_无参考修订不登记消费__a10024(参考环境):
|
|
env = 参考环境
|
|
env["授权"] = replace(env["授权"], references=())
|
|
tid = 修订测试.发起(env, command="ref-revision-nousage")
|
|
env["剧本"].append(
|
|
{
|
|
"类型": "文本",
|
|
"文本": {
|
|
"edits": [
|
|
{
|
|
"selection_id": env["选段"].selection_id,
|
|
"action": "replace",
|
|
"replacement": "无参考改写。",
|
|
"reason": "合成",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
task = 修订测试.跑(env, tid)
|
|
assert task.状态 == 任务状态.已完成
|
|
with env["装配"].要求数据库().连接(只读=True) as 连:
|
|
count = 连.execute(
|
|
"SELECT COUNT(*) FROM muse_method_usage WHERE task_id=%s", (tid,)
|
|
).fetchone()[0]
|
|
assert count == 0
|
|
|
|
|
|
@pytest.fixture
|
|
def 片段环境(参考环境):
|
|
env = 参考环境
|
|
from muse.上下文.接口 import 场景检索
|
|
from muse.资料研究.接口 import 构建场景索引, 资料服务
|
|
|
|
with env["装配"].要求数据库().连接() as 连:
|
|
资料服务().导入(
|
|
连,
|
|
env["作者"].作者,
|
|
导入请求(
|
|
kind="reference",
|
|
title="打斗场景参考书",
|
|
origin="seed://slice-rewrite",
|
|
content="机甲的刀锋嗡鸣着划过装甲。\n火花在真空中凝成细线。\n" * 300,
|
|
authorized_uses=("reference_reading",),
|
|
),
|
|
)
|
|
构建场景索引(连, env["作者"].作者)
|
|
env["片段"] = 场景检索(连, "刀锋划过装甲", 数量=1)["结果"][0]
|
|
env["授权"] = replace(env["授权"], references=(), slice_references=(env["片段"]["slice_id"],))
|
|
return env
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w32-a10035",
|
|
environment="隔离PG与合成HTTP提供方",
|
|
given="带片段参考的选段改写授权且候选真实交付",
|
|
when="发起并执行受控修订流程",
|
|
then=[
|
|
"片段原文材料入选装配并绑定进快照，材料键带参考片段前缀",
|
|
"写手输入携带片段材料；不登记方法消费（片段非方法版本）",
|
|
],
|
|
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md#参考改写选段改写链路扩展",
|
|
)
|
|
def test_片段参考材料进快照__a10035(片段环境):
|
|
env = 片段环境
|
|
tid = 修订测试.发起(env, command="slice-revision-1")
|
|
env["剧本"].append(
|
|
{
|
|
"类型": "文本",
|
|
"文本": {
|
|
"edits": [
|
|
{
|
|
"selection_id": env["选段"].selection_id,
|
|
"action": "replace",
|
|
"replacement": "刀锋擦过装甲，反震顺着掌心爬上来。",
|
|
"reason": "合成片段参考改写",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
task = 修订测试.跑(env, tid)
|
|
assert task.状态 == 任务状态.已完成
|
|
输入 = json.loads(env["收到"][-1]["input"])
|
|
assert "参考原文·打斗场景参考书" in 输入["装配材料"]
|
|
|
|
快照 = env["装配"].要求上下文().读取快照(_快照ID(task))
|
|
assert 快照.失效说明 is None
|
|
键 = "参考片段:" + env["片段"]["slice_id"]
|
|
assert 键 in 快照.装配["入选"]
|
|
assert any(
|
|
str(绑.来源ID) == env["片段"]["slice_id"] and 绑.结构哈希 == env["片段"]["内容哈希"]
|
|
for 绑 in 快照.来源绑定
|
|
)
|
|
# 片段不是方法版本：不登记方法消费。
|
|
with env["装配"].要求数据库().连接(只读=True) as 连:
|
|
count = 连.execute(
|
|
"SELECT COUNT(*) FROM muse_method_usage WHERE task_id=%s", (tid,)
|
|
).fetchone()[0]
|
|
assert count == 0
|
|
|
|
|
|
@pytest.mark.case_id(
|
|
"NC-w32-a10036",
|
|
environment="隔离PG与合成HTTP提供方",
|
|
given="已冻结的片段参考改写任务",
|
|
when="参考书原文推进新版本（内容哈希漂移）",
|
|
then=["快照标注失效且要求重新授权"],
|
|
contract="docs/系统架构/新版设计/模块设计/B06-审校修订.md#参考改写选段改写链路扩展",
|
|
)
|
|
def test_片段来源漂移使快照失效__a10036(片段环境):
|
|
env = 片段环境
|
|
tid = 修订测试.发起(env, command="slice-revision-drift")
|
|
svc = env["装配"].任务运行
|
|
claim = svc.领取步骤("revision-worker", ["revision.freeze"])
|
|
assert claim is not None
|
|
svc.执行一步(claim)
|
|
快照ID = _快照ID(svc.读取任务(tid))
|
|
|
|
from muse.资料研究.接口 import 资料服务
|
|
|
|
with env["装配"].要求数据库().连接() as 连:
|
|
资料服务().导入(
|
|
连,
|
|
env["作者"].作者,
|
|
导入请求(
|
|
kind="reference",
|
|
title="打斗场景参考书",
|
|
origin="seed://slice-rewrite",
|
|
content="改写后的新版内容。\n装甲裂缝里的火花。\n" * 300,
|
|
authorized_uses=("reference_reading",),
|
|
),
|
|
)
|
|
快照 = env["装配"].要求上下文().读取快照(快照ID)
|
|
assert 快照.失效说明 and "参考片段" in 快照.失效说明
|
|
assert "需重新授权" in 快照.失效说明
|