161 lines
7.0 KiB
Python
161 lines
7.0 KiB
Python
# ============================================================
|
||
# round-15 黄金测试：trace_chain 接入 schedule.run（矩阵 114 行）
|
||
# 覆盖：ALGO_RUN 审计携带 traceChainHash（格式正确）、
|
||
# 同输入/同种子/同版本可复算、元素/顺序/输入变化断链。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
|
||
from server.agent_core.evidence import EvidenceItem, trace_chain
|
||
from server.aps_domain.workflow import handle_intent
|
||
from server.contracts import IntentResult
|
||
from server.state.seed import seed_world
|
||
|
||
_HEX64 = re.compile(r"^[0-9a-f]{64}$")
|
||
|
||
|
||
class _MemStore:
|
||
"""最小内存 store：满足 _run_schedule 对 data/next_id/save 的依赖。"""
|
||
|
||
def __init__(self, data):
|
||
self.data = data
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
key = f"_c_{kind}"
|
||
self.data[key] = self.data.get(key, 0) + 1
|
||
return self.data[key]
|
||
|
||
def save(self):
|
||
pass
|
||
|
||
|
||
def _run_schedule_once(*, strategy: str = "COMPREHENSIVE") -> tuple[dict, _MemStore]:
|
||
"""在全新种子世界上跑一次 schedule.run，返回 (rationale, store)。"""
|
||
store = _MemStore(seed_world())
|
||
|
||
async def _run():
|
||
await handle_intent(
|
||
store, "t",
|
||
IntentResult(intent="schedule.run",
|
||
params={"strategy": strategy, "engine": "RULE"},
|
||
confidence=1.0, source="RULE_FAST"))
|
||
|
||
asyncio.run(_run())
|
||
runs = [e for e in store.data["auditEvents"]
|
||
if e["category"] == "ALGO_RUN" and e["action"] == "schedule.run"]
|
||
assert runs, "应产生一条 ALGO_RUN schedule.run 审计"
|
||
return runs[-1]["rationale"], store
|
||
|
||
|
||
def test_schedule_run_audit_carries_trace_chain_hash():
|
||
"""1) schedule.run 审计含 traceChainHash/traceCount/traceSummary（非空、格式正确）。"""
|
||
rationale, _ = _run_schedule_once()
|
||
chain_hash = rationale["traceChainHash"]
|
||
assert isinstance(chain_hash, str) and chain_hash, "traceChainHash 非空"
|
||
assert _HEX64.match(chain_hash), "traceChainHash 应为 64 位小写十六进制 SHA-256"
|
||
assert rationale["traceCount"] == 3, "链应含 schedule-version/run/algorithm 三条"
|
||
summary = rationale["traceSummary"]
|
||
assert isinstance(summary, list) and len(summary) == 3
|
||
assert [it["kind"] for it in summary] == ["schedule-version", "run", "algorithm"]
|
||
assert summary[0]["ref"] and summary[1]["ref"].startswith("run-")
|
||
|
||
|
||
def test_schedule_run_trace_recomputable_same_input():
|
||
"""2) 可复算：同输入/同种子/同版本再次求解得到相同 chainHash。"""
|
||
h1, _ = _run_schedule_once()
|
||
h2, _ = _run_schedule_once()
|
||
assert h1["traceChainHash"] == h2["traceChainHash"]
|
||
assert h1["traceCount"] == h2["traceCount"] == 3
|
||
assert h1["traceSummary"] == h2["traceSummary"]
|
||
|
||
|
||
def test_trace_chain_breaks_on_element_or_order_change():
|
||
"""3) 断链：EvidenceItem 元素或顺序变化后 chainHash 不同（纯函数直接验证）。"""
|
||
items = [
|
||
EvidenceItem(kind="schedule-version", ref="1", version="1"),
|
||
EvidenceItem(kind="run", ref="run-1", runId="1", engine="RULE",
|
||
meta={"strategy": "COMPREHENSIVE"}),
|
||
EvidenceItem(kind="algorithm", ref="RULE", version="RULE", inputsHash="abc"),
|
||
]
|
||
base = trace_chain(items)["chainHash"]
|
||
# 元素内容变化（version 不同 → 断链）
|
||
changed = [
|
||
EvidenceItem(kind="schedule-version", ref="2", version="2"),
|
||
EvidenceItem(kind="run", ref="run-1", runId="1", engine="RULE",
|
||
meta={"strategy": "COMPREHENSIVE"}),
|
||
EvidenceItem(kind="algorithm", ref="RULE", version="RULE", inputsHash="abc"),
|
||
]
|
||
assert trace_chain(changed)["chainHash"] != base
|
||
# 顺序变化 → 断链
|
||
assert trace_chain(list(reversed(items)))["chainHash"] != base
|
||
# 数量变化（删元素）→ 断链
|
||
assert trace_chain(items[:-1])["chainHash"] != base
|
||
|
||
|
||
def test_schedule_run_trace_breaks_on_input_change():
|
||
"""3+) 端到端断链：输入（策略）变化后审计 chainHash 不同。"""
|
||
h_comp, _ = _run_schedule_once(strategy="COMPREHENSIVE")
|
||
h_deliv, _ = _run_schedule_once(strategy="DELIVERY_FIRST")
|
||
assert h_comp["traceChainHash"] != h_deliv["traceChainHash"]
|
||
assert h_comp["traceCount"] == h_deliv["traceCount"] == 3
|
||
|
||
|
||
def test_all_engines_recomputable_and_evidence_has_schedule_version():
|
||
"""round-16：RULE/CP/HYBRID/GA 四引擎 schedule.run 均可复算，
|
||
evidence_refs 含 schedule-version 引用（对齐 round-15 计划假设 2）。"""
|
||
from server.aps_domain.workflow import _run_schedule
|
||
from server.contracts import IntentResult
|
||
from server.state.seed import seed_world
|
||
|
||
for engine in ("RULE", "CP", "HYBRID", "GA"):
|
||
r1, _ = _run_schedule_once_engine(engine)
|
||
r2, _ = _run_schedule_once_engine(engine)
|
||
assert r1["traceChainHash"] == r2["traceChainHash"], f"{engine} 应可复算"
|
||
assert any(ref.startswith("schedule-version:") for ref in r1["evidence"]), \
|
||
f"{engine} evidence 应含 schedule-version 引用"
|
||
assert r1["traceCount"] == r2["traceCount"] == 3
|
||
|
||
|
||
def _run_schedule_once_engine(engine: str) -> tuple[dict, _MemStore]:
|
||
"""在全新种子世界上以指定引擎跑一次 schedule.run，返回 (rationale, store)。"""
|
||
store = _MemStore(seed_world())
|
||
|
||
async def _run():
|
||
await handle_intent(
|
||
store, "t",
|
||
IntentResult(intent="schedule.run",
|
||
params={"strategy": "COMPREHENSIVE", "engine": engine},
|
||
confidence=1.0, source="RULE_FAST"))
|
||
|
||
asyncio.run(_run())
|
||
runs = [e for e in store.data["auditEvents"]
|
||
if e["category"] == "ALGO_RUN" and e["action"] == "schedule.run"]
|
||
assert runs, f"应生成一条 ALGO_RUN schedule.run 审计（{engine}）"
|
||
return runs[-1]["rationale"], store
|
||
|
||
|
||
def test_trace_chain_hash_ignores_wallclock():
|
||
"""round-16：墙钟 solveTimeSec 不进入链哈希（同输入跨引擎可复算的根基）。"""
|
||
from server.aps_domain.workflow import _build_run_trace
|
||
from server.contracts import ScheduleResult, IntentResult
|
||
from server.engines.base import EngineParams
|
||
|
||
store = _MemStore(seed_world())
|
||
params = EngineParams(engineType="CP", strategyTemplate="COMPREHENSIVE",
|
||
planningHorizonDays=14, timeLimitSeconds=8.0)
|
||
intent = IntentResult(intent="schedule.run", params={}, confidence=1.0, source="RULE_FAST")
|
||
|
||
def mk(tsec: float):
|
||
return ScheduleResult(
|
||
versionId=1, versionNo="V1", engineType="CP", strategy="COMPREHENSIVE",
|
||
status="DRAFT", orderCount=7, poCount=7, woCount=30, conflictCount=0,
|
||
totalTardiness=1.0, avgUtilization=0.8, totalCost=0, evidenceRefs=["run:V1"],
|
||
solveStatus="OPTIMAL", solveTimeSec=tsec, optimalityGap=0.0)
|
||
|
||
t1 = _build_run_trace(mk(1.234), params, intent, store)
|
||
t2 = _build_run_trace(mk(5.678), params, intent, store)
|
||
assert t1["chainHash"] == t2["chainHash"], "墙钟不应影响链哈希"
|
||
assert t1["versionRef"] == "schedule-version:1"
|