73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
"""T10 rollup + prefix cache 前缀稳定性测试。
|
||
|
||
验证交流文本的"恒定前缀"（version/request_id/query/meta/brief）在写入
|
||
progress/issues/decisions 后保持不变 —— 这是 llama-server --cache-reuse
|
||
命中、降低 prefill 开销的前提（D2 / 6.2 / T10）。
|
||
"""
|
||
import json
|
||
|
||
from router_system.workspace import Workspace
|
||
|
||
BRIEF = {
|
||
"goal": "实现快排",
|
||
"constraints": ["标准库"],
|
||
"tags": ["code"],
|
||
"acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}],
|
||
"plan": [{"id": "s1", "task": "实现", "deps": [], "done_criteria": "可运行"}],
|
||
}
|
||
|
||
|
||
def _ws():
|
||
ws = Workspace.new("abc123def456", "写个快排")
|
||
ws.apply_brief(BRIEF)
|
||
return ws
|
||
|
||
|
||
def test_prefix_stable_across_writes():
|
||
ws = _ws()
|
||
sig0 = ws.prefix_signature()
|
||
ws.add_progress("s1", "done", "完成", "a://s1.py")
|
||
ws.add_issue("s1", "a://f.py#L1", "obs", "exp", "try", "ask")
|
||
ws.add_decision("i1", "reply")
|
||
ws.mark_round()
|
||
ws.add_budget(input_tokens=100, output_tokens=20)
|
||
assert ws.prefix_signature() == sig0 # 前缀不随写操作变化
|
||
|
||
|
||
def test_prefix_changes_with_brief():
|
||
ws = _ws()
|
||
sig = ws.prefix_signature()
|
||
ws2 = Workspace.new("abc123def456", "写个快排")
|
||
ws2.apply_brief({**BRIEF, "goal": "不同的目标"})
|
||
assert ws2.prefix_signature() != sig
|
||
|
||
|
||
def test_prefix_changes_with_query():
|
||
ws = _ws()
|
||
sig = ws.prefix_signature()
|
||
ws2 = Workspace.new("abc123def456", "另一个问题")
|
||
ws2.apply_brief(BRIEF)
|
||
assert ws2.prefix_signature() != sig
|
||
|
||
|
||
def test_rollup_keeps_prefix_stable():
|
||
ws = _ws()
|
||
ws.add_progress("s1", "done", "ok", "a://s1.py")
|
||
sig_before = ws.prefix_signature()
|
||
ws.rollup()
|
||
assert ws.prefix_signature() == sig_before
|
||
|
||
|
||
def test_serialized_leading_is_stable():
|
||
ws = _ws()
|
||
head_before = _leading_json(ws)
|
||
ws.add_progress("s1", "done", "ok", "a://s1.py")
|
||
ws.add_issue("s1", "a://f.py#L1", "obs", "exp", "try", "ask")
|
||
assert _leading_json(ws) == head_before
|
||
|
||
|
||
def _leading_json(ws: Workspace) -> str:
|
||
d = ws.data
|
||
stable = {k: d.get(k) for k in ("version", "request_id", "query", "meta", "brief")}
|
||
return json.dumps(stable, ensure_ascii=False, sort_keys=True)
|