153 lines
11 KiB
Python
153 lines
11 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 urllib.parse import urlsplit
|
|
|
|
from .db import digest
|
|
from .learning_contracts import valid_selector
|
|
|
|
|
|
def content_value_valid(value, field: dict) -> bool:
|
|
if not isinstance(value, dict) or set(value) != {"kind", "text", "media", "media_truncated"}:
|
|
return False
|
|
kind, text, media, truncated = (value[key] for key in ("kind", "text", "media", "media_truncated"))
|
|
if not isinstance(kind, str) or kind not in {"text", "media", "mixed"} or not (text is None or isinstance(text, str) and bool(text.strip())) or not isinstance(media, list) or len(media) > field.get("max_media", 4) or type(truncated) is not bool:
|
|
return False
|
|
if any(not isinstance(item, dict) or set(item) != {"kind", "selector"} or not isinstance(item.get("kind"), str) or item["kind"] not in {"image", "video"} or not isinstance(item.get("selector"), str) or not valid_selector(item["selector"], relative=True) for item in media):
|
|
return False
|
|
if len({item["selector"] for item in media}) != len(media) or (truncated and len(media) != field.get("max_media", 4)):
|
|
return False
|
|
return ((kind == "text" and text is not None and not media) or (kind == "media" and text is None and bool(media)) or (kind == "mixed" and text is not None and bool(media)))
|
|
|
|
|
|
def http_url(value) -> bool:
|
|
try:
|
|
parsed = urlsplit(value) if isinstance(value, str) else None
|
|
return bool(parsed and parsed.scheme in {"http", "https"} and parsed.hostname and not parsed.username and not parsed.password)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def step_evidence_references(steps: list[dict]) -> set[str]:
|
|
values = [step.get(key) for step in steps for key in ("evidence_id", "after_evidence_id", "selector_evidence_id")]
|
|
values.extend(reference for step in steps if isinstance(step.get("batch_evidence_ids"), list) for reference in step["batch_evidence_ids"])
|
|
return {reference for reference in values if isinstance(reference, str) and reference}
|
|
|
|
|
|
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 = []
|
|
expected = step_evidence_references(steps)
|
|
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} 步动作前后必须是分别采集的证据")
|
|
batches = step.get("batch_evidence_ids")
|
|
if step.get("batch"):
|
|
config = step["batch"] if isinstance(step["batch"], dict) else {}
|
|
maximum = config.get("max_batches")
|
|
if type(maximum) is not int or not 1 <= maximum <= 5 or not isinstance(batches, list) or not 1 <= len(batches) <= maximum or any(not isinstance(item, str) or not item for item in batches) or len(set(batches)) != len(batches):
|
|
errors.append(f"第 {position} 步缺少真实批次证据")
|
|
else:
|
|
for index, reference in enumerate(batches, 1):
|
|
data = (current.get(reference) or {}).get("data", {})
|
|
batch = data.get("batch") or {}
|
|
if data.get("source") != "action_batch" or batch.get("batch_index") != index or any(type(batch.get(key)) is not int or batch[key] < 0 for key in ("added_count", "duplicate_count")):
|
|
errors.append(f"第 {position} 步批次证据不完整或来源不符")
|
|
elif batches is not None:
|
|
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, contracts: dict | None = None) -> list[dict]:
|
|
"""Deterministic assertions on saved outputs; this is not an independent read."""
|
|
checks = []
|
|
for name, expectation in expected.items():
|
|
value = actual.get(name) if isinstance(actual, dict) else None
|
|
contract = (contracts or {}).get(name) or {}
|
|
if contract.get("kind") == "extract":
|
|
fields = contract.get("fields") or []
|
|
names = {field["name"] for field in fields}
|
|
items = value.get("items") if isinstance(value, dict) else None
|
|
complete_only = contract.get("complete_only") is True
|
|
batch = contract.get("batch")
|
|
allow_empty = contract.get("allow_empty") is True
|
|
has_empty = isinstance(value, dict) and "empty_state" in value
|
|
expected_keys = {"items", "count", "truncated", "source_url", "coverage"} | ({"skipped_incomplete_count"} if complete_only else set())
|
|
if batch:
|
|
expected_keys.add("batch_summary")
|
|
if has_empty:
|
|
expected_keys.add("empty_state")
|
|
skipped = value.get("skipped_incomplete_count", 0) if isinstance(value, dict) else 0
|
|
valid = (isinstance(value, dict) and set(value) == expected_keys
|
|
and isinstance(items, list) and len(items) <= contract.get("limit", 50)
|
|
and type(value.get("count")) is int and value["count"] == len(items)
|
|
and type(value.get("truncated")) is bool
|
|
and type(skipped) is int and skipped >= 0
|
|
and (not skipped or value["truncated"])
|
|
and isinstance(value.get("coverage"), str) and value["coverage"] in {"bounded", "limit_reached", "empty"}
|
|
and http_url(value.get("source_url"))
|
|
and ((value["coverage"] == "empty") == (not items))
|
|
and (value["coverage"] != "limit_reached" or (value["truncated"] and len(items) == contract.get("limit", 50)))
|
|
and all(isinstance(item, dict) and set(item) == names and all(content_value_valid(item[field["name"]], field) if field.get("source") == "content" else isinstance(item[field["name"]], str) or item[field["name"]] is None for field in fields) for item in items))
|
|
if valid:
|
|
valid = all(not field.get("required") or all(content_value_valid(item[field["name"]], field) if field.get("source") == "content" else isinstance(item[field["name"]], str) and item[field["name"]].strip() for item in items) for field in fields)
|
|
if valid and any(item[field["name"]]["media_truncated"] for item in items for field in fields if field.get("source") == "content"):
|
|
valid = value["truncated"] is True
|
|
if valid and batch:
|
|
summary = value.get("batch_summary")
|
|
identities = [item.get(batch["dedupe_field"]) for item in items]
|
|
valid = (value["truncated"] is True and isinstance(summary, dict) and set(summary) == {"batches_read", "duplicates_removed", "stop_reason"}
|
|
and type(summary.get("batches_read")) is int and 1 <= summary["batches_read"] <= batch["max_batches"]
|
|
and type(summary.get("duplicates_removed")) is int and summary["duplicates_removed"] >= 0
|
|
and isinstance(summary.get("stop_reason"), str) and summary["stop_reason"] in {"limit_reached", "boundary", "batch_budget", "time_budget"}
|
|
and (summary["stop_reason"] != "limit_reached" or len(items) == contract.get("limit", 50))
|
|
and all(isinstance(item, str) and item.strip() for item in identities) and len(set(identities)) == len(identities))
|
|
if valid and has_empty:
|
|
proof = value["empty_state"]
|
|
expected_proof = contract.get("empty_state") or {}
|
|
valid = (allow_empty and not batch and not items and not skipped and value["truncated"] is False
|
|
and isinstance(proof, dict) and set(proof) == {"confirmed", "selector", "text"} and proof.get("confirmed") is True
|
|
and proof.get("selector") == expected_proof.get("selector") and proof.get("text") == expected_proof.get("text"))
|
|
if valid and allow_empty and not items and not has_empty:
|
|
valid = False
|
|
values = [item[field["name"]].get("text") if field.get("source") == "content" else item[field["name"]] for item in items for field in fields if field.get("source", "text") != "href"] if valid else []
|
|
text = "\n".join(value for value in values if isinstance(value, str))
|
|
else:
|
|
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)
|
|
text = "\n".join(item["text"] for item in value) if valid else ""
|
|
checks.append({"name": f"output_schema:{name}", "passed": valid, "message": "读取输出结构符合固定契约" if valid else f"输出 {name} 不符合文本读取结构"})
|
|
if isinstance(expectation, dict):
|
|
if expectation == {"state": "nonempty"}:
|
|
nonempty = valid and contract.get("kind") == "extract" and isinstance(value, dict) and type(value.get("count")) is int and value["count"] > 0
|
|
checks.append({"name": f"output_nonempty:{name}", "passed": nonempty, "message": "结构化结果非空，不代表语义判断正确" if nonempty else "未取得有效的非空结构化结果"})
|
|
continue
|
|
confirmed = valid and expectation == {"state": "empty"} and contract.get("allow_empty") is True and isinstance(value, dict) and value.get("empty_state", {}).get("confirmed") is True and value.get("count") == 0
|
|
checks.append({"name": f"output_empty:{name}", "passed": confirmed, "message": "已核对明确空态证据" if confirmed else "未取得符合预期的真实空态证据"})
|
|
continue
|
|
# A URL containing the keyword is not evidence that the rendered result
|
|
# text contains it. Only the explicitly read text participates.
|
|
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
|