from __future__ import annotations import copy from typing import Any, Callable from content_agent.business_modules.run_record.validation import validate_run from content_agent.integrations.runtime_files import LocalRuntimeFileStore, RUNTIME_FILENAMES RUN_ID = "run_v4_contract" POLICY_RUN_ID = "policy_v4_contract" def test_v4_contract_runtime_passes_validate_run(tmp_path): runtime = _write_runtime(tmp_path) result = validate_run(RUN_ID, runtime) assert result["status"] == "pass" def test_v4_contract_does_not_apply_to_v3_scorecard_records(tmp_path): def mutate(data: dict[str, Any]) -> None: decision = data["rule_decisions.jsonl"][0] decision["score"] = None decision["scorecard"] = { "total_score": None, "fit_senior_50plus": True, "relevance_score": 0.91, } decision["decision_replay_data"].pop("allow_walk") runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert result["status"] == "pass" def test_v4_score_contract_rejects_bad_total(tmp_path): def mutate(data: dict[str, Any]) -> None: data["rule_decisions.jsonl"][0]["score"] = 92 _refresh_final_output_explanations(data) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert _check_ids(result) == ["v4_score_total_mismatch"] def test_v4_score_contract_accepts_score_weights(tmp_path): def mutate(data: dict[str, Any]) -> None: data["rule_decisions.jsonl"][0]["scorecard"]["score_weights"] = { "query_relevance": 0.5, "platform_performance": 0.5, } _refresh_final_output_explanations(data) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert result["status"] == "pass" def test_v4_score_contract_rejects_bad_score_weights(tmp_path): def mutate(data: dict[str, Any]) -> None: data["rule_decisions.jsonl"][0]["scorecard"]["score_weights"] = { "query_relevance": 0.4, "platform_performance": 0.4, } _refresh_final_output_explanations(data) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert _check_ids(result) == ["v4_score_total_mismatch"] def test_v4_score_contract_not_attempted_score_weights_are_not_normalized(tmp_path): def mutate(data: dict[str, Any]) -> None: decision = data["rule_decisions.jsonl"][0] decision["score"] = 70 decision["scorecard"]["query_relevance_score"] = 100 decision["scorecard"]["platform_performance_score"] = 100 decision["scorecard"]["fifty_plus_status"] = "not_attempted" decision["scorecard"]["score_weights"] = { "query_relevance": 0.35, "platform_performance": 0.35, } decision["decision_replay_data"]["walk_gate_snapshot"] = { "query_relevance_score": 100, "platform_performance_score": 100, "score": 70, } _refresh_final_output_explanations(data) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert result["status"] == "pass" def test_v4_walk_gate_rejects_allow_walk_below_threshold(tmp_path): def mutate(data: dict[str, Any]) -> None: decision = data["rule_decisions.jsonl"][0] decision["score"] = 65 decision["scorecard"]["query_relevance_score"] = 60 decision["scorecard"]["platform_performance_score"] = 70 _refresh_final_output_explanations(data) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert "v4_allow_walk_threshold_mismatch" in _check_ids(result) def test_v4_walk_gate_rejects_allow_walk_false_at_passing_threshold(tmp_path): def mutate(data: dict[str, Any]) -> None: data["rule_decisions.jsonl"][0]["decision_replay_data"]["allow_walk"] = False _refresh_final_output_explanations(data) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert "v4_allow_walk_threshold_mismatch" in _check_ids(result) def test_v4_walk_action_rejects_success_when_allow_walk_false(tmp_path): def mutate(data: dict[str, Any]) -> None: decision = data["rule_decisions.jsonl"][0] decision["score"] = 70 decision["scorecard"]["platform_performance_score"] = 60 decision["decision_replay_data"]["allow_walk"] = False decision["decision_replay_data"]["allow_walk_reason"] = "v4_query_and_platform_pass" decision["decision_replay_data"]["walk_gate_snapshot"] = { "query_relevance_score": 80, "platform_performance_score": 60, "score": 70, } _refresh_final_output_explanations(data) data["walk_actions.jsonl"].append( _walk_action( "walk_page_001", "hashtag_to_query", "success", { "decision_id": "decision_001", "allow_walk": False, "allow_walk_reason": "v4_query_and_platform_pass", "walk_gate_snapshot": decision["decision_replay_data"]["walk_gate_snapshot"], }, ) ) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert "v4_walk_action_allow_walk_denied" in _check_ids(result) def test_v4_walk_action_rejects_missing_gate_snapshot(tmp_path): def mutate(data: dict[str, Any]) -> None: data["walk_actions.jsonl"].append( _walk_action( "walk_tag_001", "hashtag_to_query", "success", { "decision_id": "decision_001", "allow_walk": True, "allow_walk_reason": "query>=70/platform>=65/score>=70", }, ) ) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert "v4_walk_action_gate_context_missing" in _check_ids(result) def test_v4_action_thresholds_reject_conflicting_action(tmp_path): def mutate(data: dict[str, Any]) -> None: decision = data["rule_decisions.jsonl"][0] decision["decision_action"] = "KEEP_CONTENT_FOR_REVIEW" runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert "v4_action_threshold_mismatch" in _check_ids(result) def test_v4_validator_uses_decision_score_thresholds(tmp_path): def mutate(data: dict[str, Any]) -> None: decision = data["rule_decisions.jsonl"][0] decision["score"] = 67.5 decision["scorecard"]["query_relevance_score"] = 70 decision["scorecard"]["platform_performance_score"] = 65 decision["scorecard"]["score_thresholds"] = { "pool_total": 65, "pool_query": 65, "review_total": 55, "review_query": 55, "walk_query": 65, "walk_platform": 65, "walk_total": 65, } decision["decision_action"] = "ADD_TO_CONTENT_POOL" decision["decision_replay_data"]["allow_walk"] = True decision["decision_replay_data"]["allow_walk_reason"] = "query>=65/platform>=65/score>=65" decision["decision_replay_data"]["walk_gate_snapshot"] = { "query_relevance_score": 70, "platform_performance_score": 65, "score": 67.5, } _refresh_final_output_explanations(data) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert result["status"] == "pass" def test_v4_gemini_failure_requires_structured_failure_fields(tmp_path): def mutate(data: dict[str, Any]) -> None: summary = data["pattern_recall_evidence.jsonl"][0]["evidence_summary"] summary.clear() summary.update( { "schema_version": "v4_gemini_query_relevance.v1", "final_status": "failed", "retry_count": 0, } ) runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert { "v4_gemini_failure_incomplete", "v4_gemini_failure_retry_invalid", } <= set(_check_ids(result)) def test_v4_legacy_field_blocklist_rejects_v4_records_only(tmp_path): def mutate(data: dict[str, Any]) -> None: data["rule_decisions.jsonl"][0]["scorecard"]["platform_heat"] = 88 data["pattern_recall_evidence.jsonl"][0]["evidence_summary"]["relevance_score"] = 0.9 runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert _check_ids(result).count("v4_legacy_field_present") == 2 def test_v4_final_output_requires_explanation(tmp_path): def mutate(data: dict[str, Any]) -> None: data["final_output.json"]["decision_records"][0].pop("v4_explanation") runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert "v4_final_output_explanation_missing" in _check_ids(result) def test_v4_final_output_explanation_must_match_decision(tmp_path): def mutate(data: dict[str, Any]) -> None: data["final_output.json"]["decision_records"][0]["v4_explanation"]["allow_walk"] = False runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert "v4_final_output_explanation_mismatch" in _check_ids(result) def test_v4_strategy_review_requires_summary_when_generated(tmp_path): def mutate(data: dict[str, Any]) -> None: data["strategy_review.json"].pop("v4_summary") runtime = _write_runtime(tmp_path, mutate) result = validate_run(RUN_ID, runtime) assert "v4_strategy_review_explanation_missing" in _check_ids(result) def _write_runtime( tmp_path, mutate: Callable[[dict[str, Any]], None] | None = None, ) -> LocalRuntimeFileStore: runtime = LocalRuntimeFileStore(tmp_path / "runtime") runtime.prepare_run(RUN_ID) data = _runtime_payload() if mutate: mutate(data) for filename in RUNTIME_FILENAMES: value = data[filename] if isinstance(value, list): runtime.append_jsonl(RUN_ID, filename, value) else: runtime.write_json(RUN_ID, filename, value) return runtime def _runtime_payload() -> dict[str, Any]: evidence_pack = { "pattern_source_system": "pg_pattern_v2", "case_id_type": "post_id", "source_kind": "pattern_itemset", "source_post_id": "post_001", "pattern_execution_id": 581, "mining_config_id": 2082, "itemset_ids": [1608352], "itemset_items": ["毛主席", "感人"], "support": 5, "absolute_support": 5, "matched_post_ids": ["post_001", "post_002"], "video_ids": ["video_001"], "case_ids": ["case_001"], "seed_terms": ["父爱感悟"], "discovery_start_source": "pattern_seed", "previous_discovery_step": "search_query_generated", "origin_path_id": "path_origin_001", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "source_certainty": "high", "validation_status": "validated", } source_evidence = _source_evidence(evidence_pack, "content_001") path_ids = ["path_pattern_query", "path_query_content", "path_decision_asset"] decision = { "record_schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "decision_id": "decision_001", "policy_bundle_id": "policy_bundle_v4", "rule_pack_id": "douyin_content_discovery_rule_pack_v4", "rule_pack_version": "4.0.0", "strategy_version": "V4", "decision_target_type": "content", "decision_target_id": "content_001", "decision_action": "ADD_TO_CONTENT_POOL", "decision_reason_code": "v4_query_and_platform_pass", "search_query_effect_status": "success", "score": 75, "scorecard": { "schema_version": "v4_scorecard.v1", "query_relevance_score": 80, "platform_performance_score": 70, "missing_observable_fields": [], }, "source_evidence": copy.deepcopy(source_evidence), "decision_replay_data": { "policy_bundle_hash": "hash_v4", "rule_pack_id": "douyin_content_discovery_rule_pack_v4", "rule_pack_version": "4.0.0", "dispatch_id": "dispatch_v4", "strategy_version": "V4", "allow_walk": True, "allow_walk_reason": "query>=70/platform>=65/score>=70", "walk_gate_snapshot": { "query_relevance_score": 80, "platform_performance_score": 70, "score": 75, }, }, "raw_payload": {"decision_id": "decision_001", "v4_contract": True}, } v4_explanation = _v4_explanation(decision) return { "source_context.json": { "schema_version": "runtime_record.v1", "run_id": RUN_ID, "demand_content_id": "123", "ext_data": {"evidence_pack": evidence_pack}, }, "pattern_seed_pack.json": { "schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "itemsets": [{"itemset_id": 1608352}], "seed_terms": ["父爱感悟"], }, "search_queries.jsonl": [ { "record_schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "search_query_id": "query_001", "search_query": "父爱感悟", "search_query_generation_method": "v4_seed", "discovery_start_source": "pattern_seed", "previous_discovery_step": "search_query_generated", "search_query_effect_status": "success", "pattern_seed_ref": {"query_source_type": "seed"}, "raw_payload": {"query_source_refs": [{"query_source_type": "seed"}]}, } ], "discovered_content_items.jsonl": [ { "record_schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "content_discovery_id": "discovery_001", "search_query_id": "query_001", "platform": "douyin", "platform_content_id": "content_001", "content_url": "https://example.com/content_001", "statistics": {"share_count": 10}, "tags": ["父爱"], "source_evidence": copy.deepcopy(source_evidence), "pattern_match_result": { "judge_status": "ok", "pattern_recall_evidence_id": "recall_001", }, "platform_raw_payload": {}, "raw_payload": {"platform_content_id": "content_001"}, } ], "content_media_records.jsonl": [ { "record_schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "media_record_id": "media_001", "platform": "douyin", "platform_content_id": "content_001", "media_status": "available", "raw_payload": {"platform_content_id": "content_001"}, } ], "pattern_recall_evidence.jsonl": [ { "record_schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "recall_evidence_id": "recall_001", "content_discovery_id": "discovery_001", "platform_content_id": "content_001", "recall_status": "judged", "evidence_summary": { "schema_version": "v4_gemini_query_relevance.v1", "final_status": "success", "query_relevance_score": 80, "reason": "契合 query", }, "raw_payload": {"recall_evidence_id": "recall_001"}, } ], "rule_decisions.jsonl": [decision], "walk_actions.jsonl": [], "run_events.jsonl": [], "source_path_records.jsonl": [ _path("path_pattern_query", "pattern_to_search_query", "Pattern", 581, "SearchQuery", "query_001"), _path("path_query_content", "search_query_to_content", "SearchQuery", "query_001", "Content", "content_001"), _path( "path_decision_asset", "decision_to_asset", "RuleDecision", "decision_001", "ContentAsset", "content_001", decision_id="decision_001", ), ], "search_clues.jsonl": [ { "record_schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "clue_id": "clue_001", "search_query_id": "query_001", "search_query": "父爱感悟", "result_count": 1, "pooled_content_count": 1, "review_content_count": 0, "pending_content_count": 0, "rejected_content_count": 0, "search_query_effect_status": "success", "query_aggregation_id": "agg_query_success", "raw_payload": {"clue_id": "clue_001"}, } ], "final_output.json": { "schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "validation_status": "pass", "content_assets": [ { "platform_content_id": "content_001", "decision_id": "decision_001", "source_path_record_ids": path_ids, "source_evidence": copy.deepcopy(source_evidence), "v4_explanation": copy.deepcopy(v4_explanation), } ], "author_assets": [], "review_records": [], "decision_records": [ { "decision_id": "decision_001", "score": 75, "scorecard": copy.deepcopy(decision["scorecard"]), "decision_replay_data": copy.deepcopy(decision["decision_replay_data"]), "source_evidence": copy.deepcopy(source_evidence), "v4_explanation": copy.deepcopy(v4_explanation), } ], "search_clues": [], "reject_records": [], "summary": { "pooled_content_count": 1, "review_content_count": 0, "pending_content_count": 0, "rejected_content_count": 0, "run_path_complete": True, "trace_complete": True, }, }, "strategy_review.json": { "schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "summary": {}, "v4_summary": { "schema_version": "v4_strategy_review_summary.v1", "score_buckets": { "pool": 1, "review": 0, "reject": 0, "technical_retry": 0, "unknown": 0, }, "allow_walk_distribution": { "allowed": 1, "denied": 0, "missing": 0, }, "walk_gate_review": { "v4_gate_distribution": { "allowed": 0, "denied": 0, "missing": 0, } }, }, "raw_payload": {"strategy_review_id": "review_001"}, }, } def _v4_explanation(decision: dict[str, Any]) -> dict[str, Any]: scorecard = decision["scorecard"] replay = decision["decision_replay_data"] return { "schema_version": "v4_decision_explanation.v1", "scorecard_schema_version": scorecard["schema_version"], "query_relevance_score": scorecard["query_relevance_score"], "platform_performance_score": scorecard["platform_performance_score"], "score": decision["score"], "platform_performance_components": scorecard.get("platform_performance_components", []), "missing_observable_fields": scorecard.get("missing_observable_fields", []), "decision_action": decision["decision_action"], "decision_reason_code": decision["decision_reason_code"], "search_query_effect_status": decision["search_query_effect_status"], "allow_walk": replay["allow_walk"], "allow_walk_reason": replay["allow_walk_reason"], "walk_gate_snapshot": replay["walk_gate_snapshot"], } def _refresh_final_output_explanations(data: dict[str, Any]) -> None: decisions = {decision["decision_id"]: decision for decision in data["rule_decisions.jsonl"]} for section in ["content_assets", "review_records", "reject_records", "decision_records"]: for record in data["final_output.json"].get(section, []): decision = decisions.get(record.get("decision_id")) if not decision: continue record["v4_explanation"] = _v4_explanation(decision) if section == "decision_records": record["score"] = decision.get("score") record["scorecard"] = copy.deepcopy(decision.get("scorecard")) record["decision_replay_data"] = copy.deepcopy( decision.get("decision_replay_data") ) def _source_evidence(evidence_pack: dict[str, Any], platform_content_id: str) -> dict[str, Any]: evidence = copy.deepcopy(evidence_pack) evidence["discovered_platform_content_id"] = platform_content_id return evidence def _path( path_id: str, path_type: str, from_type: str, from_id: Any, to_type: str, to_id: Any, decision_id: str | None = None, ) -> dict[str, Any]: return { "record_schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "source_path_record_id": path_id, "source_path_type": path_type, "from_node_type": from_type, "from_node_id": from_id, "to_node_type": to_type, "to_node_id": to_id, "decision_id": decision_id, "raw_payload": {"source_path_record_id": path_id}, } def _walk_action( action_id: str, edge_id: str, status: str, raw_payload: dict[str, Any], ) -> dict[str, Any]: return { "record_schema_version": "runtime_record.v1", "run_id": RUN_ID, "policy_run_id": POLICY_RUN_ID, "walk_action_id": action_id, "edge_id": edge_id, "edge_type": "query", "from_node_type": "SearchQuery", "from_node_id": "query_001", "to_node_type": "SearchQuery", "to_node_id": "query_002", "walk_action": "fetch_next_page", "walk_status": status, "budget_tier": "normal", "depth": 1, "reason_code": None, "raw_payload": raw_payload, } def _check_ids(result: dict[str, Any]) -> list[str]: return [finding["check_id"] for finding in result["findings"]]