from __future__ import annotations import json from collections import Counter from typing import Any from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION, RUNTIME_SCHEMA_VERSION from content_agent.findings import fail as _fail from content_agent.integrations.runtime_files import RUNTIME_FILENAMES from content_agent.interfaces import RuntimeFileStore JSON_FILES = { "source_context.json", "pattern_seed_pack.json", "final_output.json", "strategy_review.json", } JSONL_FILES = set(RUNTIME_FILENAMES) - JSON_FILES POLICY_RUN_FILES = { "pattern_seed_pack.json", "search_queries.jsonl", "discovered_content_items.jsonl", "content_media_records.jsonl", "pattern_recall_evidence.jsonl", "rule_decisions.jsonl", "walk_actions.jsonl", "run_events.jsonl", "source_path_records.jsonl", "search_clues.jsonl", "final_output.json", "strategy_review.json", } RAW_PAYLOAD_FILES = { "search_queries.jsonl", "discovered_content_items.jsonl", "content_media_records.jsonl", "pattern_recall_evidence.jsonl", "rule_decisions.jsonl", "walk_actions.jsonl", "run_events.jsonl", "source_path_records.jsonl", "search_clues.jsonl", "strategy_review.json", } FORBIDDEN_PAYLOAD_KEYS = { "password", "token", "access_token", "refresh_token", "api_key", "apikey", "secret", "dsn", "authorization", "cookie", "session", "credential", } DECISION_ACTIONS = { "ADD_TO_CONTENT_POOL", "KEEP_CONTENT_FOR_REVIEW", "REJECT_CONTENT", "TECHNICAL_RETRY_REQUIRED", } EFFECT_STATUSES = {"success", "pending", "failed", "rule_blocked"} WALK_STATUSES = {"success", "pending", "failed", "skipped", "rule_blocked"} V4_SCORECARD_SCHEMA_VERSION = "v4_scorecard.v1" V4_GEMINI_QUERY_RELEVANCE_SCHEMA_VERSION = "v4_gemini_query_relevance.v1" V4_SCORE_THRESHOLD_FALLBACK = { "pool_total": 70, "pool_query": 70, "review_total": 55, "review_query": 55, "walk_query": 70, "walk_platform": 65, "walk_total": 70, } V4_LEGACY_FIELD_BLOCKLIST = { "fit_senior_50plus", "fit_confidence", "relevance_score", "platform_heat", "age_50_plus_level", } DECISION_REPLAY_REQUIRED_FIELDS = { "policy_bundle_hash", "rule_pack_id", "rule_pack_version", "dispatch_id", "strategy_version", } SOURCE_EVIDENCE_FIELDS = { "source_kind", "pattern_source_system", "case_id_type", "source_post_id", "pattern_execution_id", "mining_config_id", "itemset_ids", "itemset_items", "support", "absolute_support", "matched_post_ids", "video_ids", "case_ids", "seed_terms", "discovery_start_source", "previous_discovery_step", "origin_path_id", "run_id", "policy_run_id", "discovered_platform_content_id", "source_certainty", "validation_status", } def validate_run(run_id: str, runtime: RuntimeFileStore) -> dict[str, Any]: findings: list[dict[str, Any]] = [] data = _load_files(run_id, runtime, findings) if any(finding["level"] == "fail" for finding in findings): return _result(run_id, findings) _check_run_ids(run_id, data, findings) _check_schema_versions(data, findings) _check_policy_run_ids(data, findings) _check_raw_payloads(data, findings) _check_unique_ids(data, findings) _check_references(data, findings) _check_pattern_recall_evidence(data, findings) _check_source_evidence(data, findings) _check_source_paths(data, findings) _check_summary(data, findings) _check_completeness(data, findings) _check_v4_score_contract(data, findings) _check_v4_walk_gate_contract(data, findings) _check_v4_walk_action_consumption(data, findings) _check_v4_final_output_explanation(data, findings) _check_v4_strategy_review_explanation(data, findings) _check_v4_action_thresholds(data, findings) _check_v4_gemini_failure_contract(data, findings) _check_v4_legacy_field_blocklist(data, findings) return _result(run_id, findings) def compute_final_output_completeness( final_output: dict[str, Any], decisions: list[dict[str, Any]], source_path_records: list[dict[str, Any]], ) -> dict[str, Any]: findings: list[str] = [] decision_ids = {decision.get("decision_id") for decision in decisions} path_ids = {path.get("source_path_record_id") for path in source_path_records} decision_asset_paths = { (path.get("decision_id"), path.get("to_node_id")): path for path in source_path_records if path.get("source_path_type") == "decision_to_asset" } for asset in final_output.get("content_assets", []): decision_id = asset.get("decision_id") content_id = asset.get("platform_content_id") asset_path_ids = set(asset.get("source_path_record_ids", [])) if decision_id not in decision_ids: findings.append(f"content_asset missing decision: {content_id}") missing_paths = sorted(asset_path_ids - path_ids) if missing_paths: findings.append(f"content_asset missing paths: {content_id}") decision_asset = decision_asset_paths.get((decision_id, content_id)) if not decision_asset: findings.append(f"content_asset missing decision_to_asset: {content_id}") elif decision_asset.get("source_path_record_id") not in asset_path_ids: findings.append(f"content_asset omits decision_to_asset: {content_id}") for record in final_output.get("review_records", []): if record.get("decision_id") not in decision_ids: findings.append(f"review_record missing decision: {record.get('platform_content_id')}") if set(record.get("source_path_record_ids", [])) - path_ids: findings.append(f"review_record missing paths: {record.get('platform_content_id')}") for record in final_output.get("reject_records", []): if record.get("decision_id") not in decision_ids: findings.append(f"reject_record missing decision: {record.get('decision_target_id')}") if set(record.get("source_path_record_ids", [])) - path_ids: findings.append(f"reject_record missing paths: {record.get('decision_target_id')}") for record in final_output.get("technical_retry_records", []): if record.get("decision_id") not in decision_ids: findings.append( f"technical_retry_record missing decision: {record.get('decision_target_id')}" ) if set(record.get("source_path_record_ids", [])) - path_ids: findings.append( f"technical_retry_record missing paths: {record.get('decision_target_id')}" ) final_decision_ids = { record.get("decision_id") for record in final_output.get("decision_records", []) } if decision_ids - final_decision_ids: findings.append("decision_records incomplete") for author_asset in final_output.get("author_assets", []): if set(author_asset.get("decision_ids", [])) - decision_ids: findings.append(f"author_asset missing decisions: {author_asset.get('author_asset_id')}") if set(author_asset.get("source_path_record_ids", [])) - path_ids: findings.append(f"author_asset missing paths: {author_asset.get('author_asset_id')}") evidence_refs = author_asset.get("evidence_refs") or {} if evidence_refs.get("decision_ids") != author_asset.get("decision_ids"): findings.append(f"author_asset evidence incomplete: {author_asset.get('author_asset_id')}") required_sections = [ "content_assets", "author_assets", "review_records", "decision_records", "search_clues", "reject_records", "summary", ] missing_sections = [section for section in required_sections if section not in final_output] findings.extend(f"final_output missing section: {section}" for section in missing_sections) complete = not findings return { "validation_status": "pass" if complete else "fail", "run_path_complete": complete, "trace_complete": complete, "findings_summary": findings, } def _load_files( run_id: str, runtime: RuntimeFileStore, findings: list[dict[str, Any]], ) -> dict[str, Any]: run_dir = runtime.run_dir(run_id) data: dict[str, Any] = {} for filename in RUNTIME_FILENAMES: path = run_dir / filename if not path.exists(): _fail(findings, "file_missing", f"missing runtime file: {filename}") continue try: if filename in JSON_FILES: data[filename] = json.loads(path.read_text(encoding="utf-8")) else: data[filename] = [ json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip() ] except json.JSONDecodeError as exc: _fail(findings, "json_parse_failed", f"{filename} cannot parse: {exc}") return data def _check_run_ids(run_id: str, data: dict[str, Any], findings: list[dict[str, Any]]) -> None: for filename, value in data.items(): rows = value if isinstance(value, list) else [value] for row in rows: if isinstance(row, dict) and row.get("run_id") != run_id: _fail(findings, "run_id_mismatch", f"{filename} has mismatched run_id") def _check_schema_versions(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: for filename, value in data.items(): if filename in JSON_FILES: if isinstance(value, dict) and value.get("schema_version") != RUNTIME_SCHEMA_VERSION: _fail(findings, "schema_version_missing", f"{filename} has bad schema_version") continue rows = value if isinstance(value, list) else [] for row in rows: if row.get("record_schema_version") != RUNTIME_RECORD_SCHEMA_VERSION: _fail( findings, "record_schema_version_missing", f"{filename} has bad record_schema_version", ) def _check_policy_run_ids(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: if "policy_run_id" in data.get("source_context.json", {}): _fail(findings, "policy_run_id_unexpected", "source_context.json must stay run-level") policy_run_id = data.get("pattern_seed_pack.json", {}).get("policy_run_id") if not policy_run_id: _fail(findings, "policy_run_id_missing", "pattern_seed_pack.json missing policy_run_id") return for filename in POLICY_RUN_FILES: value = data.get(filename) rows = value if isinstance(value, list) else [value] for row in rows: if isinstance(row, dict) and row.get("policy_run_id") != policy_run_id: _fail( findings, "policy_run_id_mismatch", f"{filename} has mismatched policy_run_id", ) def _check_raw_payloads(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: for filename in RAW_PAYLOAD_FILES: value = data.get(filename) rows = value if isinstance(value, list) else [value] for row in rows: if not isinstance(row, dict): continue raw_payload = row.get("raw_payload") if not isinstance(raw_payload, dict) or not raw_payload: _fail(findings, "raw_payload_missing", f"{filename} missing raw_payload") continue forbidden_paths = _find_forbidden_payload_keys(raw_payload) if forbidden_paths: _fail( findings, "raw_payload_forbidden_key", f"{filename} raw_payload contains forbidden keys: {forbidden_paths}", ) def _find_forbidden_payload_keys(value: Any, prefix: str = "raw_payload") -> list[str]: if isinstance(value, dict): paths: list[str] = [] for key, child in value.items(): child_path = f"{prefix}.{key}" if str(key).lower() in FORBIDDEN_PAYLOAD_KEYS: paths.append(child_path) paths.extend(_find_forbidden_payload_keys(child, child_path)) return paths if isinstance(value, list): paths = [] for index, child in enumerate(value): paths.extend(_find_forbidden_payload_keys(child, f"{prefix}[{index}]")) return paths return [] def _check_unique_ids(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: checks = [ ("search_queries.jsonl", "search_query_id"), ("discovered_content_items.jsonl", "content_discovery_id"), ("discovered_content_items.jsonl", "platform_content_id"), ("pattern_recall_evidence.jsonl", "recall_evidence_id"), ("rule_decisions.jsonl", "decision_id"), ("rule_decisions.jsonl", "decision_target_id"), ("walk_actions.jsonl", "walk_action_id"), ("source_path_records.jsonl", "source_path_record_id"), ] for filename, field in checks: values = [row.get(field) for row in data.get(filename, [])] duplicates = [value for value, count in Counter(values).items() if value and count > 1] if duplicates: _fail(findings, "duplicate_id", f"{filename}.{field} duplicates: {duplicates}") def _check_references(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: search_query_ids = {row["search_query_id"] for row in data.get("search_queries.jsonl", [])} items = data.get("discovered_content_items.jsonl", []) content_ids = {row["platform_content_id"] for row in items} decisions = data.get("rule_decisions.jsonl", []) decision_ids = {row["decision_id"] for row in decisions} walk_action_ids = {row["walk_action_id"] for row in data.get("walk_actions.jsonl", [])} path_ids = {row["source_path_record_id"] for row in data.get("source_path_records.jsonl", [])} for item in items: if item.get("search_query_id") not in search_query_ids: _fail( findings, "missing_search_query_ref", f"content item has unknown search_query_id: {item.get('search_query_id')}", ) for media in data.get("content_media_records.jsonl", []): if not media.get("platform"): _fail(findings, "platform_missing", "content_media_records.jsonl missing platform") if media.get("platform_content_id") not in content_ids: _fail( findings, "missing_content_ref", f"media has unknown platform_content_id: {media.get('platform_content_id')}", ) for evidence in data.get("pattern_recall_evidence.jsonl", []): if evidence.get("platform_content_id") not in content_ids: _fail( findings, "missing_content_ref", f"recall evidence has unknown platform_content_id: {evidence.get('platform_content_id')}", ) for decision in decisions: if decision.get("decision_target_id") not in content_ids: _fail( findings, "missing_content_ref", f"decision has unknown decision_target_id: {decision.get('decision_target_id')}", ) if decision.get("decision_action") not in DECISION_ACTIONS: _fail( findings, "bad_decision_action", f"unsupported decision_action: {decision.get('decision_action')}", ) if decision.get("search_query_effect_status") not in EFFECT_STATUSES: _fail( findings, "bad_effect_status", f"unsupported decision effect status: {decision.get('search_query_effect_status')}", ) replay_data = decision.get("decision_replay_data") or {} missing_replay_fields = [ field for field in DECISION_REPLAY_REQUIRED_FIELDS if not replay_data.get(field) ] if missing_replay_fields: _fail( findings, "decision_replay_incomplete", f"decision {decision.get('decision_id')} missing replay fields: {missing_replay_fields}", ) for clue in data.get("search_clues.jsonl", []): if clue.get("search_query_effect_status") not in EFFECT_STATUSES: _fail( findings, "bad_effect_status", f"unsupported query effect status: {clue.get('search_query_effect_status')}", ) if not clue.get("query_aggregation_id"): _fail( findings, "missing_query_aggregation_id", f"search clue missing query_aggregation_id: {clue.get('clue_id')}", ) for action in data.get("walk_actions.jsonl", []): missing_action_fields = [ field for field in ["walk_action_id", "edge_id", "walk_action", "walk_status"] if not action.get(field) ] if missing_action_fields: _fail( findings, "walk_action_incomplete", f"walk action missing fields: {missing_action_fields}", ) if action.get("walk_status") not in WALK_STATUSES: _fail( findings, "bad_walk_status", f"unsupported walk_status: {action.get('walk_status')}", ) decision_id = action.get("decision_id") if decision_id and decision_id not in decision_ids: _fail(findings, "missing_decision_ref", f"walk action has unknown decision_id: {decision_id}") for path in data.get("source_path_records.jsonl", []): decision_id = path.get("decision_id") if decision_id and decision_id not in decision_ids: _fail(findings, "missing_decision_ref", f"path has unknown decision_id: {decision_id}") walk_action_id = (path.get("raw_payload") or {}).get("walk_action_id") or path.get("walk_action_id") if walk_action_id and walk_action_id not in walk_action_ids: _fail( findings, "missing_walk_action_ref", f"path has unknown walk_action_id: {walk_action_id}", ) final_output = data.get("final_output.json", {}) for asset in final_output.get("content_assets", []): if asset.get("decision_id") not in decision_ids: _fail( findings, "missing_decision_ref", f"asset has unknown decision_id: {asset.get('decision_id')}", ) for path_id in asset.get("source_path_record_ids", []): if path_id not in path_ids: _fail(findings, "missing_path_ref", f"asset has unknown path_id: {path_id}") for record in final_output.get("review_records", []): if record.get("decision_id") not in decision_ids: _fail( findings, "missing_decision_ref", f"review_records has unknown decision_id: {record.get('decision_id')}", ) for path_id in record.get("source_path_record_ids", []): if path_id not in path_ids: _fail( findings, "missing_path_ref", f"review_records has unknown path_id: {path_id}", ) for section in ["reject_records", "technical_retry_records", "decision_records"]: for row in final_output.get(section, []): if row.get("decision_id") not in decision_ids: _fail( findings, "missing_decision_ref", f"{section} has unknown decision_id: {row.get('decision_id')}", ) for path_id in row.get("source_path_record_ids", []): if path_id not in path_ids: _fail(findings, "missing_path_ref", f"{section} has unknown path_id: {path_id}") for author_asset in final_output.get("author_assets", []): for decision_id in author_asset.get("decision_ids", []): if decision_id not in decision_ids: _fail( findings, "missing_decision_ref", f"author_assets has unknown decision_id: {decision_id}", ) for path_id in author_asset.get("source_path_record_ids", []): if path_id not in path_ids: _fail( findings, "missing_path_ref", f"author_assets has unknown path_id: {path_id}", ) evidence_refs = author_asset.get("evidence_refs") or {} if evidence_refs.get("decision_ids") != author_asset.get("decision_ids"): _fail( findings, "author_asset_evidence_incomplete", f"author asset evidence refs do not match decisions: {author_asset.get('author_asset_id')}", ) def _check_pattern_recall_evidence( data: dict[str, Any], findings: list[dict[str, Any]], ) -> None: evidence_rows = data.get("pattern_recall_evidence.jsonl", []) evidence_by_id = {row.get("recall_evidence_id"): row for row in evidence_rows} for item in data.get("discovered_content_items.jsonl", []): pattern_match = item.get("pattern_match_result") or {} # V3(M3):桥接键 pattern_recall 已退役;改以"是否被判定过"(judge_status 存在)为准—— # 每条经 Gemini 判定的内容必须能解析到真实 evidence 行,否则视为血缘损坏。 if not pattern_match.get("judge_status"): continue evidence_id = pattern_match.get("pattern_recall_evidence_id") evidence = evidence_by_id.get(evidence_id) if not evidence: _fail( findings, "pattern_recall_evidence_missing", f"matched item cannot find recall evidence: {evidence_id}", ) # V3(M2):判定改为 Gemini 直读,不再有 decode 强证据词/分类树路径; # matched_terms/matched_category_paths 不再适用,故不再强制。 def _check_source_evidence(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: source_context = data.get("source_context.json", {}) evidence_pack = source_context.get("ext_data", {}).get("evidence_pack", {}) for decision in data.get("rule_decisions.jsonl", []): _check_one_source_evidence( findings, decision.get("source_evidence") or {}, evidence_pack, f"decision {decision.get('decision_id')}", ) _check_final_output_source_evidence_refs(data, findings) _check_final_decision_coverage(data, findings) def _check_final_output_source_evidence_refs( data: dict[str, Any], findings: list[dict[str, Any]], ) -> None: decisions_by_id = { decision.get("decision_id"): decision for decision in data.get("rule_decisions.jsonl", []) if decision.get("decision_id") } final_output = data.get("final_output.json", {}) for section in [ "content_assets", "reject_records", "review_records", "technical_retry_records", "decision_records", ]: for record in final_output.get(section, []): decision_id = record.get("decision_id") decision = decisions_by_id.get(decision_id) if not decision: continue ref = record.get("source_evidence_ref") if ref is None: if "source_evidence" in record: continue _fail( findings, "source_evidence_ref_missing", f"{section} {decision_id} missing source_evidence_ref", ) continue if not isinstance(ref, dict): _fail( findings, "source_evidence_ref_invalid", f"{section} {decision_id} has invalid source_evidence_ref", ) continue if ref.get("decision_id") != decision_id: _fail( findings, "source_evidence_ref_mismatch", f"{section} {decision_id} ref points to wrong decision", ) target_id = record.get("platform_content_id") or record.get("decision_target_id") expected_target = decision.get("decision_target_id") if target_id and expected_target and ref.get("decision_target_id") != expected_target: _fail( findings, "source_evidence_ref_mismatch", f"{section} {decision_id} ref points to wrong target", ) def _check_one_source_evidence( findings: list[dict[str, Any]], source_evidence: dict[str, Any], evidence_pack: dict[str, Any], label: str, ) -> None: missing_fields = sorted(field for field in SOURCE_EVIDENCE_FIELDS if field not in source_evidence) if missing_fields: _fail(findings, "source_evidence_missing_fields", f"{label} missing {missing_fields}") # V3(M4):分类树 category/element binding 已随 decode 链路退役,不再作为血缘门槛。 scalar_fields = [ "pattern_execution_id", "source_post_id", "pattern_source_system", "case_id_type", "mining_config_id", "support", "absolute_support", "source_certainty", "validation_status", ] list_fields = [ "itemset_ids", "itemset_items", "matched_post_ids", "video_ids", "case_ids", "seed_terms", ] for field in scalar_fields + list_fields: if source_evidence.get(field) != evidence_pack.get(field): _fail(findings, "source_evidence_mismatch", f"{label} mismatched {field}") if ( "decode_case_ids" in source_evidence and source_evidence.get("decode_case_ids") != evidence_pack.get("decode_case_ids") ): _fail(findings, "source_evidence_mismatch", f"{label} mismatched decode_case_ids") platform_content_id = source_evidence.get("discovered_platform_content_id") if platform_content_id: if platform_content_id == source_evidence.get("source_post_id"): _fail(findings, "source_evidence_content_pollution", f"{label} rewrites source_post_id") if platform_content_id in (source_evidence.get("matched_post_ids") or []): _fail( findings, "source_evidence_content_pollution", f"{label} rewrites matched_post_ids", ) def _check_final_decision_coverage(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: decision_ids = {decision["decision_id"] for decision in data.get("rule_decisions.jsonl", [])} final_decision_ids = { record.get("decision_id") for record in data.get("final_output.json", {}).get("decision_records", []) } missing = sorted(decision_ids - final_decision_ids) if missing: _fail(findings, "final_decision_missing", f"final_output decision_records missing {missing}") def _check_source_paths(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: evidence_pack = data.get("source_context.json", {}).get("ext_data", {}).get("evidence_pack", {}) pattern_execution_id = evidence_pack.get("pattern_execution_id") paths = data.get("source_path_records.jsonl", []) path_by_id = {path["source_path_record_id"]: path for path in paths} pattern_query_paths = { path["to_node_id"]: path for path in paths if path.get("source_path_type") == "pattern_to_search_query" } query_content_paths = { path["to_node_id"]: path for path in paths if path.get("source_path_type") == "search_query_to_content" } decision_asset_paths = { (path.get("decision_id"), path.get("to_node_id")): path for path in paths if path.get("source_path_type") == "decision_to_asset" } for decision in data.get("rule_decisions.jsonl", []): _check_content_source_path( findings, label=f"decision {decision.get('decision_id')}", platform_content_id=decision.get("decision_target_id"), pattern_execution_id=pattern_execution_id, pattern_query_paths=pattern_query_paths, query_content_paths=query_content_paths, ) for asset in data.get("final_output.json", {}).get("content_assets", []): platform_content_id = asset.get("platform_content_id") path_ids = set(asset.get("source_path_record_ids", [])) query_content = query_content_paths.get(platform_content_id) if not query_content or query_content.get("source_path_record_id") not in path_ids: _fail( findings, "source_path_broken", f"asset lacks search_query_to_content path: {platform_content_id}", ) continue pattern_query = pattern_query_paths.get(query_content.get("from_node_id")) if not pattern_query or pattern_query.get("source_path_record_id") not in path_ids: _fail( findings, "source_path_broken", f"asset lacks pattern_to_search_query path: {platform_content_id}", ) continue if pattern_query.get("from_node_id") != pattern_execution_id: _fail( findings, "source_path_broken", f"asset path starts from wrong pattern: {platform_content_id}", ) for path_id in path_ids: if path_id not in path_by_id: _fail(findings, "source_path_broken", f"asset path missing: {path_id}") decision_asset = decision_asset_paths.get( (asset.get("decision_id"), platform_content_id) ) if not decision_asset: _fail( findings, "decision_to_asset_missing", f"asset lacks decision_to_asset path: {platform_content_id}", ) continue if decision_asset.get("source_path_record_id") not in path_ids: _fail( findings, "decision_to_asset_missing", f"asset source paths omit decision_to_asset: {platform_content_id}", ) if decision_asset.get("from_node_type") != "RuleDecision": _fail( findings, "decision_to_asset_broken", f"asset decision_to_asset starts from wrong node: {platform_content_id}", ) if decision_asset.get("from_node_id") != asset.get("decision_id"): _fail( findings, "decision_to_asset_broken", f"asset decision_to_asset has wrong decision id: {platform_content_id}", ) if decision_asset.get("to_node_type") != "ContentAsset": _fail( findings, "decision_to_asset_broken", f"asset decision_to_asset ends at wrong node: {platform_content_id}", ) for record in data.get("final_output.json", {}).get("review_records", []): platform_content_id = record.get("platform_content_id") path_ids = set(record.get("source_path_record_ids", [])) query_content = query_content_paths.get(platform_content_id) if not query_content or query_content.get("source_path_record_id") not in path_ids: _fail( findings, "source_path_broken", f"review record lacks search_query_to_content path: {platform_content_id}", ) continue pattern_query = pattern_query_paths.get(query_content.get("from_node_id")) if not pattern_query or pattern_query.get("source_path_record_id") not in path_ids: _fail( findings, "source_path_broken", f"review record lacks pattern_to_search_query path: {platform_content_id}", ) def _check_content_source_path( findings: list[dict[str, Any]], label: str, platform_content_id: Any, pattern_execution_id: Any, pattern_query_paths: dict[str, dict[str, Any]], query_content_paths: dict[str, dict[str, Any]], ) -> None: query_content = query_content_paths.get(platform_content_id) if not query_content: _fail( findings, "source_path_broken", f"{label} lacks search_query_to_content path: {platform_content_id}", ) return pattern_query = pattern_query_paths.get(query_content.get("from_node_id")) if not pattern_query: _fail( findings, "source_path_broken", f"{label} lacks pattern_to_search_query path: {platform_content_id}", ) return if pattern_query.get("from_node_id") != pattern_execution_id: _fail( findings, "source_path_broken", f"{label} path starts from wrong pattern: {platform_content_id}", ) def _check_summary(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: decisions = data.get("rule_decisions.jsonl", []) action_counts = Counter(decision.get("decision_action") for decision in decisions) summary = data.get("final_output.json", {}).get("summary", {}) expected = { "pooled_content_count": action_counts["ADD_TO_CONTENT_POOL"], "review_content_count": action_counts["KEEP_CONTENT_FOR_REVIEW"], "pending_content_count": 0, "rejected_content_count": action_counts["REJECT_CONTENT"], } if ( action_counts["TECHNICAL_RETRY_REQUIRED"] or "technical_retry_content_count" in summary ): expected["technical_retry_content_count"] = action_counts["TECHNICAL_RETRY_REQUIRED"] for field, value in expected.items(): if summary.get(field) != value: _fail( findings, "summary_mismatch", f"summary.{field} expected {value}, got {summary.get(field)}", ) clue_counts = Counter() for clue in data.get("search_clues.jsonl", []): clue_counts["ADD_TO_CONTENT_POOL"] += clue.get("pooled_content_count", 0) clue_counts["KEEP_CONTENT_FOR_REVIEW"] += clue.get("review_content_count", 0) clue_counts["REJECT_CONTENT"] += clue.get("rejected_content_count", 0) clue_counts["TECHNICAL_RETRY_REQUIRED"] += clue.get("technical_retry_content_count", 0) for action, count in action_counts.items(): if action == "TECHNICAL_RETRY_REQUIRED" and not count: continue if clue_counts[action] < count: _fail( findings, "search_clue_mismatch", f"search_clues {action} expected at least {count}, got {clue_counts[action]}", ) def _check_completeness(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: final_output = data.get("final_output.json", {}) expected = compute_final_output_completeness( final_output, data.get("rule_decisions.jsonl", []), data.get("source_path_records.jsonl", []), ) summary = final_output.get("summary", {}) for field in ["run_path_complete", "trace_complete"]: if summary.get(field) != expected[field]: _fail( findings, "completeness_mismatch", f"summary.{field} expected {expected[field]}, got {summary.get(field)}", ) if final_output.get("validation_status") != expected["validation_status"]: _fail( findings, "completeness_mismatch", f"validation_status expected {expected['validation_status']}, got {final_output.get('validation_status')}", ) def _is_v4_contract_record(decision: dict[str, Any]) -> bool: scorecard = decision.get("scorecard") or {} return isinstance(scorecard, dict) and scorecard.get("schema_version") == V4_SCORECARD_SCHEMA_VERSION def _v4_score_values(decision: dict[str, Any]) -> tuple[Any, Any, Any]: scorecard = decision.get("scorecard") or {} return ( scorecard.get("query_relevance_score"), scorecard.get("platform_performance_score"), decision.get("score"), ) def _v4_score_thresholds(decision: dict[str, Any]) -> dict[str, Any]: scorecard = decision.get("scorecard") or {} thresholds = scorecard.get("score_thresholds") if not isinstance(thresholds, dict): return dict(V4_SCORE_THRESHOLD_FALLBACK) return {**V4_SCORE_THRESHOLD_FALLBACK, **thresholds} def _v4_walk_threshold_message(thresholds: dict[str, Any]) -> str: return ( f"query>={thresholds['walk_query']}/" f"platform>={thresholds['walk_platform']}/" f"score>={thresholds['walk_total']}" ) def _check_v4_score_contract(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: for decision in data.get("rule_decisions.jsonl", []): if not _is_v4_contract_record(decision): continue scorecard = decision.get("scorecard") or {} query_score, platform_score, total_score = _v4_score_values(decision) score_missing = bool(scorecard.get("score_missing")) technical_retry = decision.get("decision_reason_code") in { "v4_technical_retry_needed", "portrait_unavailable", "portrait_incomplete", } if score_missing and technical_retry: if not isinstance(scorecard.get("missing_observable_fields"), list): _fail( findings, "v4_missing_observable_fields_invalid", f"decision {decision.get('decision_id')} missing_observable_fields must be a list", ) continue for field_name, value in [ ("query_relevance_score", query_score), ("platform_performance_score", platform_score), ("score", total_score), ]: if not _is_number(value) or not 0 <= value <= 100: _fail( findings, "v4_score_contract_invalid", f"decision {decision.get('decision_id')} has invalid {field_name}: {value}", ) if not isinstance(scorecard.get("missing_observable_fields"), list): _fail( findings, "v4_missing_observable_fields_invalid", f"decision {decision.get('decision_id')} missing_observable_fields must be a list", ) if _is_number(query_score) and _is_number(platform_score) and _is_number(total_score): # M9B:抖音(scorecard 带 fifty_plus_status)走 35/35/30 或 35/35(not_attempted); # 非抖音(无该字段)维持 50/50。 fifty_status = scorecard.get("fifty_plus_status") fifty_score = scorecard.get("fifty_plus_score") if fifty_status == "ok" and _is_number(fifty_score): expected = query_score * 0.35 + platform_score * 0.35 + fifty_score * 0.30 elif fifty_status == "not_attempted": expected = query_score * 0.35 + platform_score * 0.35 else: expected = query_score * 0.5 + platform_score * 0.5 if abs(total_score - expected) > 0.01: _fail( findings, "v4_score_total_mismatch", f"decision {decision.get('decision_id')} score expected {expected}, got {total_score}", ) def _check_v4_walk_gate_contract(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: for decision in data.get("rule_decisions.jsonl", []): if not _is_v4_contract_record(decision): continue replay_data = decision.get("decision_replay_data") or {} if "allow_walk" not in replay_data: _fail( findings, "v4_allow_walk_missing", f"decision {decision.get('decision_id')} missing decision_replay_data.allow_walk", ) continue if not isinstance(replay_data.get("allow_walk"), bool): _fail( findings, "v4_allow_walk_invalid", f"decision {decision.get('decision_id')} allow_walk must be boolean", ) continue query_score, platform_score, total_score = _v4_score_values(decision) thresholds = _v4_score_thresholds(decision) expected_allow_walk = ( _is_number(query_score) and _is_number(platform_score) and _is_number(total_score) and query_score >= thresholds["walk_query"] and platform_score >= thresholds["walk_platform"] and total_score >= thresholds["walk_total"] ) if replay_data["allow_walk"] != expected_allow_walk: _fail( findings, "v4_allow_walk_threshold_mismatch", f"decision {decision.get('decision_id')} allow_walk does not match {_v4_walk_threshold_message(thresholds)}", ) def _check_v4_walk_action_consumption(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: v4_decisions_by_id = { decision.get("decision_id"): decision for decision in data.get("rule_decisions.jsonl", []) if _is_v4_contract_record(decision) and decision.get("decision_id") } if not v4_decisions_by_id: return for action in data.get("walk_actions.jsonl", []): edge_id = action.get("edge_id") if edge_id not in {"hashtag_to_query", "author_to_works"}: continue raw_payload = action.get("raw_payload") or {} decision_id = raw_payload.get("decision_id") or action.get("decision_id") decision = v4_decisions_by_id.get(decision_id) if not decision: continue missing_fields = [ field for field in ["decision_id", "allow_walk", "allow_walk_reason", "walk_gate_snapshot"] if field not in raw_payload ] if missing_fields: _fail( findings, "v4_walk_action_gate_context_missing", f"walk action {action.get('walk_action_id')} missing V4 gate context fields: {missing_fields}", ) continue expected_allow_walk = (decision.get("decision_replay_data") or {}).get("allow_walk") if raw_payload.get("allow_walk") != expected_allow_walk: _fail( findings, "v4_walk_action_allow_walk_mismatch", f"walk action {action.get('walk_action_id')} allow_walk does not match decision {decision_id}", ) if action.get("walk_status") == "success" and expected_allow_walk is not True: _fail( findings, "v4_walk_action_allow_walk_denied", f"walk action {action.get('walk_action_id')} succeeded for allow_walk=false decision {decision_id}", ) if action.get("reason_code") == "v4_allow_walk_denied" and expected_allow_walk is not False: _fail( findings, "v4_walk_action_deny_mismatch", f"walk action {action.get('walk_action_id')} denies walk but decision {decision_id} is not allow_walk=false", ) def _check_v4_final_output_explanation( data: dict[str, Any], findings: list[dict[str, Any]], ) -> None: v4_decisions_by_id = { decision.get("decision_id"): decision for decision in data.get("rule_decisions.jsonl", []) if _is_v4_contract_record(decision) and decision.get("decision_id") } if not v4_decisions_by_id: return records_by_decision_id = _final_output_records_by_decision_id( data.get("final_output.json", {}) ) for decision_id, decision in v4_decisions_by_id.items(): section_records = records_by_decision_id.get(decision_id) or {} decision_records = section_records.get("decision_records") or [] if not decision_records: _fail( findings, "v4_final_output_explanation_missing", f"final_output decision_records missing V4 explanation record for decision {decision_id}", ) continue for section, records in section_records.items(): for record in records: _check_v4_explanation_record( decision, record, f"final_output.{section}", findings, ) def _final_output_records_by_decision_id( final_output: dict[str, Any], ) -> dict[str, dict[str, list[dict[str, Any]]]]: by_decision_id: dict[str, dict[str, list[dict[str, Any]]]] = {} for section in [ "content_assets", "review_records", "reject_records", "technical_retry_records", "decision_records", ]: for record in final_output.get(section, []) or []: decision_id = record.get("decision_id") if not decision_id: for candidate in record.get("decision_ids") or []: by_decision_id.setdefault(candidate, {}).setdefault(section, []).append(record) continue by_decision_id.setdefault(decision_id, {}).setdefault(section, []).append(record) return by_decision_id def _check_v4_explanation_record( decision: dict[str, Any], record: dict[str, Any], section: str, findings: list[dict[str, Any]], ) -> None: explanation = record.get("v4_explanation") decision_id = decision.get("decision_id") if not isinstance(explanation, dict) or not explanation: _fail( findings, "v4_final_output_explanation_missing", f"{section} record for decision {decision_id} missing v4_explanation", ) return if explanation.get("schema_version") != "v4_decision_explanation.v1": _fail( findings, "v4_final_output_explanation_mismatch", f"{section} record for decision {decision_id} has invalid V4 explanation schema", ) scorecard = decision.get("scorecard") or {} replay_data = decision.get("decision_replay_data") or {} expected_values = { "scorecard_schema_version": scorecard.get("schema_version"), "query_relevance_score": scorecard.get("query_relevance_score"), "platform_performance_score": scorecard.get("platform_performance_score"), "score": decision.get("score"), "decision_reason_code": decision.get("decision_reason_code"), "allow_walk": replay_data.get("allow_walk"), "allow_walk_reason": replay_data.get("allow_walk_reason"), } for field_name, expected in expected_values.items(): if expected is None and field_name not in explanation: continue if not _v4_values_equal(explanation.get(field_name), expected): _fail( findings, "v4_final_output_explanation_mismatch", f"{section} record for decision {decision_id} has mismatched {field_name}", ) if not isinstance(explanation.get("missing_observable_fields"), list): _fail( findings, "v4_final_output_explanation_mismatch", f"{section} record for decision {decision_id} missing_observable_fields must be a list", ) def _check_v4_strategy_review_explanation( data: dict[str, Any], findings: list[dict[str, Any]], ) -> None: if not any(_is_v4_contract_record(decision) for decision in data.get("rule_decisions.jsonl", [])): return strategy_review = data.get("strategy_review.json") or {} if not strategy_review: return v4_summary = strategy_review.get("v4_summary") if not isinstance(v4_summary, dict): _fail( findings, "v4_strategy_review_explanation_missing", "strategy_review missing v4_summary for V4 decisions", ) return required_fields = [ "schema_version", "score_buckets", "allow_walk_distribution", "walk_gate_review", ] missing = [field for field in required_fields if field not in v4_summary] if missing: _fail( findings, "v4_strategy_review_explanation_missing", f"strategy_review.v4_summary missing fields: {missing}", ) return if v4_summary.get("schema_version") != "v4_strategy_review_summary.v1": _fail( findings, "v4_strategy_review_explanation_invalid", "strategy_review.v4_summary has invalid schema_version", ) if not isinstance(v4_summary.get("score_buckets"), dict): _fail( findings, "v4_strategy_review_explanation_invalid", "strategy_review.v4_summary.score_buckets must be an object", ) allow_walk_distribution = v4_summary.get("allow_walk_distribution") if not isinstance(allow_walk_distribution, dict) or not { "allowed", "denied", "missing", } <= set(allow_walk_distribution): _fail( findings, "v4_strategy_review_explanation_invalid", "strategy_review.v4_summary.allow_walk_distribution is incomplete", ) walk_gate_review = v4_summary.get("walk_gate_review") if not isinstance(walk_gate_review, dict) or "v4_gate_distribution" not in walk_gate_review: _fail( findings, "v4_strategy_review_explanation_invalid", "strategy_review.v4_summary.walk_gate_review is incomplete", ) def _check_v4_action_thresholds(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: for decision in data.get("rule_decisions.jsonl", []): if not _is_v4_contract_record(decision): continue if decision.get("decision_reason_code") == "v4_technical_retry_needed": continue query_score, _platform_score, total_score = _v4_score_values(decision) if not (_is_number(query_score) and _is_number(total_score)): continue action = decision.get("decision_action") thresholds = _v4_score_thresholds(decision) if action == "ADD_TO_CONTENT_POOL": ok = query_score >= thresholds["pool_query"] and total_score >= thresholds["pool_total"] elif action == "KEEP_CONTENT_FOR_REVIEW": ok = ( query_score >= thresholds["review_query"] and total_score >= thresholds["review_total"] and not ( query_score >= thresholds["pool_query"] and total_score >= thresholds["pool_total"] ) ) elif action == "REJECT_CONTENT": ok = query_score < thresholds["review_query"] or total_score < thresholds["review_total"] else: continue if not ok: _fail( findings, "v4_action_threshold_mismatch", f"decision {decision.get('decision_id')} action {action} conflicts with query/score thresholds", ) def _check_v4_gemini_failure_contract(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: for evidence in data.get("pattern_recall_evidence.jsonl", []): summary = evidence.get("evidence_summary") or {} if not ( isinstance(summary, dict) and summary.get("schema_version") == V4_GEMINI_QUERY_RELEVANCE_SCHEMA_VERSION ): continue if summary.get("final_status") != "failed": continue missing = [ field for field in ["failure_type", "retry_count", "http_status_code", "final_status"] if field not in summary ] if missing: _fail( findings, "v4_gemini_failure_incomplete", f"recall evidence {evidence.get('recall_evidence_id')} missing failure fields: {missing}", ) if not isinstance(summary.get("retry_count"), int) or summary.get("retry_count", 0) < 1: _fail( findings, "v4_gemini_failure_retry_invalid", f"recall evidence {evidence.get('recall_evidence_id')} retry_count must be >=1", ) def _check_v4_legacy_field_blocklist(data: dict[str, Any], findings: list[dict[str, Any]]) -> None: for decision in data.get("rule_decisions.jsonl", []): if not _is_v4_contract_record(decision): continue legacy_paths = _find_legacy_field_paths( { "scorecard": decision.get("scorecard"), "decision_replay_data": decision.get("decision_replay_data"), "raw_payload": decision.get("raw_payload"), }, f"decision {decision.get('decision_id')}", ) if legacy_paths: _fail( findings, "v4_legacy_field_present", f"V4 decision contains legacy fields: {legacy_paths}", ) for evidence in data.get("pattern_recall_evidence.jsonl", []): summary = evidence.get("evidence_summary") or {} if not ( isinstance(summary, dict) and summary.get("schema_version") == V4_GEMINI_QUERY_RELEVANCE_SCHEMA_VERSION ): continue legacy_paths = _find_legacy_field_paths( {"evidence_summary": summary, "raw_payload": evidence.get("raw_payload")}, f"recall evidence {evidence.get('recall_evidence_id')}", ) if legacy_paths: _fail( findings, "v4_legacy_field_present", f"V4 recall evidence contains legacy fields: {legacy_paths}", ) def _find_legacy_field_paths(value: Any, prefix: str) -> list[str]: if isinstance(value, dict): paths: list[str] = [] for key, child in value.items(): child_path = f"{prefix}.{key}" if key in V4_LEGACY_FIELD_BLOCKLIST: paths.append(child_path) paths.extend(_find_legacy_field_paths(child, child_path)) return paths if isinstance(value, list): paths = [] for index, child in enumerate(value): paths.extend(_find_legacy_field_paths(child, f"{prefix}[{index}]")) return paths return [] def _is_number(value: Any) -> bool: return isinstance(value, (int, float)) and not isinstance(value, bool) def _v4_values_equal(left: Any, right: Any) -> bool: if _is_number(left) and _is_number(right): return abs(float(left) - float(right)) <= 0.01 return left == right def _result(run_id: str, findings: list[dict[str, Any]]) -> dict[str, Any]: return { "run_id": run_id, "status": "fail" if any(finding["level"] == "fail" for finding in findings) else "pass", "findings": findings, }