52 lines
3.2 KiB
Python
52 lines
3.2 KiB
Python
"""Pure checks adapted from A2A operation knowledge and test readback boundaries.
|
|
|
|
Reference: learning/packages/contracts/python/penguin_learning_v1/operation_chain.py
|
|
(`_references`, `graph_integrity`, `validate_operation_candidate`) and
|
|
learning/src/learning_m0/test_readback.py. No source package or database is imported.
|
|
This reduced contract covers ordered, read-only steps, not the A2A operation DTO.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from .db import digest
|
|
|
|
|
|
def evidence_graph_errors(steps: list[dict], bindings: list[dict], current: dict[str, dict]) -> list[str]:
|
|
"""Check the closed set of immutable before/after evidence references."""
|
|
errors = []
|
|
references = [step.get(key) for step in steps for key in ("evidence_id", "after_evidence_id")]
|
|
expected = {reference for reference in references if isinstance(reference, str) and reference}
|
|
ids = [item.get("observation_id") for item in bindings]
|
|
if any(not isinstance(value, str) or not value for value in ids) or len(ids) != len(set(ids)):
|
|
errors.append("冻结证据标识缺失或重复")
|
|
if set(ids) != expected:
|
|
errors.append("候选步骤与冻结证据引用不完整对应")
|
|
for position, step in enumerate(steps, 1):
|
|
if not step.get("evidence_id") or not step.get("after_evidence_id"):
|
|
errors.append(f"第 {position} 步缺少动作前后证据")
|
|
elif step["evidence_id"] == step["after_evidence_id"]:
|
|
errors.append(f"第 {position} 步动作前后必须是分别采集的证据")
|
|
for binding in bindings:
|
|
reference = binding.get("observation_id")
|
|
actual = current.get(reference)
|
|
if actual is None:
|
|
errors.append("冻结证据缺少同任务的原始记录")
|
|
elif digest(actual) != binding.get("content_digest"):
|
|
errors.append("来源证据与候选冻结摘要不一致")
|
|
return errors
|
|
|
|
|
|
def output_assertion_checks(actual: dict, expected: dict) -> list[dict]:
|
|
"""Deterministic assertions on saved outputs; this is not an independent read."""
|
|
checks = []
|
|
for name, expectation in expected.items():
|
|
value = actual.get(name)
|
|
valid = isinstance(value, list) and all(isinstance(item, dict) and isinstance(item.get("text"), str) and set(item) <= {"text", "href"} and ("href" not in item or isinstance(item["href"], str)) for item in value)
|
|
checks.append({"name": f"output_schema:{name}", "passed": valid, "message": "读取输出结构符合固定契约" if valid else f"输出 {name} 不符合文本读取结构"})
|
|
# A URL containing the keyword is not evidence that the rendered result
|
|
# text contains it. Only the explicitly read text participates.
|
|
text = "\n".join(item["text"] for item in value) if valid else ""
|
|
checks.append({"name": f"output_present:{name}", "passed": bool(text.strip()), "message": f"输出 {name} 未读到内容" if not text.strip() else "已读到输出"})
|
|
for item in [expectation] if isinstance(expectation, str) else expectation:
|
|
checks.append({"name": f"output_contains:{name}", "passed": item in text, "message": "预期文本匹配" if item in text else f"输出 {name} 未包含预期文本：{item[:100]}"})
|
|
return checks
|