#!/usr/bin/env python3 """Read-only validator for every visible data-source Inspector in real runs.""" from __future__ import annotations import argparse import hashlib import json import sys import urllib.error import urllib.parse import urllib.request from collections import Counter from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any HIDDEN_BUSINESS_KEYS = {"technical", "changes"} def request_json(base_url: str, path: str) -> tuple[int, Any]: url = f"{base_url.rstrip('/')}{path}" try: with urllib.request.urlopen(url, timeout=60) as response: return response.status, json.load(response) except urllib.error.HTTPError as exc: try: payload = json.load(exc) except Exception: payload = {"detail": str(exc)} return exc.code, payload def quote_ref(value: str) -> str: return urllib.parse.quote(value, safe="") def collect_detail_refs(value: Any) -> list[str]: found: list[str] = [] def visit(node: Any) -> None: if isinstance(node, dict): for key, child in node.items(): if key == "detailRef" and isinstance(child, str) and child: found.append(child) visit(child) elif isinstance(node, list): for child in node: visit(child) visit(value) return list(dict.fromkeys(found)) def pointer_lookup(value: Any, pointer: str) -> tuple[bool, Any]: if not pointer: return True, value current = value for raw in pointer.lstrip("/").split("/"): part = raw.replace("~1", "/").replace("~0", "~") if isinstance(current, list): try: current = current[int(part)] except (ValueError, IndexError): return False, None elif isinstance(current, dict) and part in current: current = current[part] else: return False, None return True, current def digest(value: Any) -> str: serialized = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str) return hashlib.sha256(serialized.encode("utf-8")).hexdigest() def validate_binding(binding: dict[str, Any], sources: dict[str, Any]) -> list[str]: errors: list[str] = [] source_id = str(binding.get("sourceId") or "") selector = binding.get("selector") or {} resolution = binding.get("resolution") source = sources.get(source_id) if source is None: return [f"binding {binding.get('id')} 的 sourceId {source_id} 未返回"] if resolution == "resolved" and source.get("resultState") in {"missing", "unavailable"}: errors.append( f"binding {binding.get('id')} 标记 resolved，但来源状态为 {source.get('resultState')}" ) if selector.get("kind") == "unresolved": code = str(selector.get("code") or "") reason = str(selector.get("reason") or "") if not reason: errors.append(f"binding {binding.get('id')} 的 unresolved 原因为空") if not code: errors.append(f"binding {binding.get('id')} 的 unresolved 缺少原因码") if code in {"binding-map-missing", "field-path-mismatch"}: errors.append(f"binding {binding.get('id')} 存在可视化映射错误 {code}") elif code not in { "source-record-not-saved", "event-body-not-saved", "unsafe-association", "reliable-log-anchor-unavailable", }: errors.append(f"binding {binding.get('id')} 使用了未允许的缺失原因码 {code}") if code in {"source-record-not-saved", "event-body-not-saved"} and source.get("resultState") not in {"missing", "unavailable"}: errors.append( f"binding {binding.get('id')} 声称原始记录未保存，" f"但来源状态为 {source.get('resultState')}" ) return errors if selector.get("kind") == "json-pointer": found, _selected = pointer_lookup(source.get("rawRecord"), selector.get("path") or "") if not found: errors.append(f"binding {binding.get('id')} 无法反向定位 {selector.get('path')}") elif selector.get("kind") == "text-span": found, text = pointer_lookup(source.get("rawRecord"), selector.get("path") or "") if not found or not isinstance(text, str): errors.append(f"binding {binding.get('id')} 的原文字段不存在") else: characters = list(text) start = int(selector.get("startCodePoint") or 0) end = int(selector.get("endCodePoint") or 0) if "".join(characters[start:end]) != selector.get("exactText"): errors.append(f"binding {binding.get('id')} 的原文区间不匹配") if digest(text) != selector.get("sourceDigest"): errors.append(f"binding {binding.get('id')} 的 sourceDigest 不匹配") elif selector.get("kind") == "members": members = selector.get("members") or [] if not members: errors.append(f"binding {binding.get('id')} 的聚合成员为空") for member in members: errors.extend(validate_binding({ "id": f"{binding.get('id')}:member", "sourceId": member.get("sourceId"), "selector": member.get("selector"), "resolution": "resolved", }, sources)) return errors def selected_value(binding: dict[str, Any], source: dict[str, Any]) -> tuple[bool, Any]: selector = binding.get("selector") or {} kind = selector.get("kind") raw = source.get("rawRecord") if kind == "json-pointer": return pointer_lookup(raw, str(selector.get("path") or "")) if kind == "whole-record": return True, raw if kind == "text-span": found, text = pointer_lookup(raw, str(selector.get("path") or "")) if not found or not isinstance(text, str): return False, None start = int(selector.get("startCodePoint") or 0) end = int(selector.get("endCodePoint") or 0) return True, "".join(list(text)[start:end]) return False, None def equivalent(left: Any, right: Any) -> bool: return json.dumps(left, ensure_ascii=False, sort_keys=True, default=str) == json.dumps( right, ensure_ascii=False, sort_keys=True, default=str ) def validate_business_value( binding: dict[str, Any], source: dict[str, Any], business_value: Any ) -> list[str]: """Direct bindings must select the exact value shown in column one.""" if (binding.get("transform") or {}).get("kind") != "direct": return [] found, selected = selected_value(binding, source) if found and not equivalent(selected, business_value): return [ f"binding {binding.get('id')} 声明直接展示，但 selector 值与业务项不一致" ] return [] def validate_empty_sources(sources: dict[str, Any]) -> list[str]: errors: list[str] = [] for source_id, source in sources.items(): if source.get("resultState") != "empty": continue kind = source.get("kind") locator = source.get("locator") or {} raw = source.get("rawRecord") if kind == "database" and not ( raw == [] and int(locator.get("resultCount") or 0) == 0 ): errors.append(f"空结果来源 {source_id} 无法证明数据库查询返回 0 条") if kind == "runtime-event" and not isinstance(raw, dict): errors.append(f"空结果来源 {source_id} 没有保存运行 Event") return errors def validate_view(payload: dict[str, Any]) -> tuple[Counter, list[str]]: counts: Counter = Counter() errors: list[str] = [] if payload.get("schemaVersion") != "inspector-source-v2": return counts, ["schemaVersion 不是 inspector-source-v2"] business = payload.get("businessProjection") or {} sources = payload.get("sources") or {} for source in sources.values(): counts[f"source:{source.get('resultState') or 'present'}"] += 1 actual_selectors: list[str] = [] for module in payload.get("modules") or []: bindings = {item.get("id"): item for item in module.get("bindings") or []} for row in module.get("rows") or []: counts["rows"] += 1 business_selector = str(row.get("businessSelector") or "") actual_selectors.append(business_selector) found, business_value = pointer_lookup(business, business_selector) if not found: errors.append(f"业务行 {row.get('id')} 无法定位 {row.get('businessSelector')}") ids = row.get("bindingIds") or [] if not ids: errors.append(f"业务行 {row.get('id')} 没有 binding") for binding_id in ids: binding = bindings.get(binding_id) if binding is None: errors.append(f"业务行 {row.get('id')} 引用了不存在的 binding {binding_id}") continue counts[str(binding.get("resolution") or "unknown")] += 1 errors.extend(validate_binding(binding, sources)) source = sources.get(str(binding.get("sourceId") or "")) or {} errors.extend(validate_business_value(binding, source, business_value)) errors.extend(validate_log_sources(sources)) errors.extend(validate_empty_sources(sources)) expected_selectors = visible_business_selectors(business) missing_selectors = [value for value in expected_selectors if value not in actual_selectors] if missing_selectors: errors.append(f"旧业务 Inspector 的可见项没有生成对照行: {missing_selectors}") return counts, errors def validate_log_sources(sources: dict[str, Any]) -> list[str]: errors: list[str] = [] for source_id, source in sources.items(): if source.get("kind") != "log-anchor": continue locator = source.get("locator") or {} event_id = locator.get("eventId") if event_id is None: errors.append(f"日志来源 {source_id} 没有 eventId") continue event = sources.get(f"event:{event_id}") or {} event_raw = event.get("rawRecord") or {} expected_msg_id = str(event_raw.get("msg_id") or locator.get("msgId") or "") if not expected_msg_id: errors.append(f"Event {event_id} 没有 msg_id，不应附带日志来源") continue raw_log = source.get("rawRecord") rendered = raw_log if isinstance(raw_log, str) else json.dumps(raw_log, ensure_ascii=False) if f"msg_id={expected_msg_id}" not in rendered and f"msg_id:{expected_msg_id}" not in rendered: errors.append(f"日志来源 {source_id} 与 Event {event_id} 的 msg_id 不一致") return errors def visible_business_selectors(projection: dict[str, Any]) -> list[str]: """Mirror the readable Inspector's visible leaves, including fixed empty notices.""" selectors: list[str] = [] if projection.get("question") not in (None, ""): selectors.append("/question") for section_index, section in enumerate(projection.get("businessSections") or []): if not isinstance(section, dict): continue root = f"/businessSections/{section_index}" if isinstance(section.get("items"), list): selectors.extend(f"{root}/items/{index}" for index, _ in enumerate(section["items"])) elif section.get("content") is not None: selectors.append(f"{root}/content") for block_index, block in enumerate(projection.get("blocks") or []): if not isinstance(block, dict): continue root = f"/blocks/{block_index}" kind = str(block.get("type") or "summary") if kind in {"summary", "notice"}: selectors.append(f"{root}/value") elif kind == "items": selectors.extend(f"{root}/items/{index}" for index, _ in enumerate(block.get("items") or [])) elif kind == "implementation-plan": if block.get("mode") is not None: selectors.append(f"{root}/mode") for route_index, route in enumerate(block.get("routes") or []): if not isinstance(route, dict): continue for key in ("target", "method", "action", "emphasis", "pathType"): if route.get(key) is not None: selectors.append(f"{root}/routes/{route_index}/{key}") if block.get("reasoning") is not None: selectors.append(f"{root}/reasoning") elif kind == "creative-process": for step_index, step in enumerate(block.get("steps") or []): if not isinstance(step, dict): continue for key in ("action", "summary", "plan", "reasoning"): if step.get(key) is not None: selectors.append(f"{root}/steps/{step_index}/{key}") elif kind == "evaluation-branches": for branch_index, branch in enumerate(block.get("branches") or []): if not isinstance(branch, dict): continue for key in ("conclusion", "achievements", "problems", "evidence"): if branch.get(key) not in (None, [], ""): selectors.append(f"{root}/branches/{branch_index}/{key}") elif kind in {"evaluation-matrix", "comparison"}: selectors.extend(f"{root}/rows/{index}" for index, _ in enumerate(block.get("rows") or [])) elif kind == "artifact": selectors.extend(f"{root}/refs/{index}" for index, _ in enumerate(block.get("refs") or [])) else: selectors.append(root) if projection.get("detailKind") == "retrieval-event": conditions = projection.get("queryConditions") or [] selectors.extend( [f"/queryConditions/{index}/value" for index, _ in enumerate(conditions)] or ["/queryConditions"] ) outcome = projection.get("outcome") or {} selectors.extend( f"/outcome/{key}" for key in ("message", "count", "errorKind") if outcome.get(key) is not None ) selectors.extend(f"/items/{index}" for index, _ in enumerate(projection.get("items") or [])) completeness = projection.get("completeness") or {} if completeness.get("bodyAvailable") is False: selectors.append("/completeness/bodyAvailable") if completeness.get("truncated"): selectors.append( "/completeness/omittedCharacters" if completeness.get("omittedCharacters") is not None else "/completeness/truncated" ) if isinstance(projection.get("activities"), list): activities = projection.get("activities") or [] selectors.extend( [f"/activities/{index}" for index, _ in enumerate(activities)] or ["/missingActivityNotice" if projection.get("missingActivityNotice") else "/activities"] ) return selectors def validate_detail(base_url: str, run_id: int, detail_ref: str) -> tuple[str, Counter, list[str], bool]: encoded = quote_ref(detail_ref) status, payload = request_json( base_url, f"/api/script-builds/{run_id}/inspector-view/{encoded}" ) if status == 404: return detail_ref, Counter(), [], False if status != 200: return detail_ref, Counter(), [f"inspector-view HTTP {status}: {payload}"], True counts, errors = validate_view(payload) old_status, old = request_json( base_url, f"/api/script-builds/{run_id}/activities/{encoded}" ) if old_status == 200: expected = {key: value for key, value in old.items() if key not in HIDDEN_BUSINESS_KEYS} if payload.get("businessProjection") != expected: errors.append("第一列与旧业务 Inspector 投影不一致") return detail_ref, counts, errors, True def validate_run(base_url: str, run_id: int, *, workers: int) -> dict[str, Any]: status, execution = request_json(base_url, f"/api/script-builds/{run_id}/execution-view") if status != 200: return {"runId": run_id, "fatal": f"execution-view HTTP {status}: {execution}"} refs = collect_detail_refs(execution) totals: Counter = Counter() failures: list[dict[str, Any]] = [] checked = 0 with ThreadPoolExecutor(max_workers=max(1, workers)) as executor: futures = { executor.submit(validate_detail, base_url, run_id, detail_ref): detail_ref for detail_ref in refs } for future in as_completed(futures): detail_ref, counts, errors, did_check = future.result() checked += int(did_check) totals.update(counts) if errors: failures.append({"detailRef": detail_ref, "errors": errors}) failures.sort(key=lambda item: refs.index(item["detailRef"])) return { "runId": run_id, "visibleRefs": len(refs), "checkedInspectors": checked, "counts": dict(totals), "failures": failures, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--base-url", default="http://127.0.0.1:8788") parser.add_argument("--runs", type=int, nargs="+", default=[422, 443, 444, 446]) parser.add_argument("--output") parser.add_argument("--workers", type=int, default=8) args = parser.parse_args() results = [validate_run(args.base_url, run_id, workers=args.workers) for run_id in args.runs] report = {"baseUrl": args.base_url, "runs": results} rendered = json.dumps(report, ensure_ascii=False, indent=2) print(rendered) if args.output: with open(args.output, "w", encoding="utf-8") as handle: handle.write(rendered + "\n") return 1 if any(result.get("fatal") or result.get("failures") for result in results) else 0 if __name__ == "__main__": sys.exit(main())