content-find-agent-new/tests/test_v4_m3_scoring_replay.py

from __future__ import annotations

import json

from content_agent.business_modules.run_record.validation import validate_run
from content_agent.integrations.database_runtime import DatabaseRuntimeStore
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from tests.gemini_helpers import FakeGeminiVideoClient, fake_gemini_fail, fake_gemini_pool
from tests.replay_harness import replay_case
from tests.test_database_runtime import FakeConnection, _config, _insert_values


def test_v4_m3_scoring_replay_produces_v4_runtime_contract(tmp_path):
    runtime_root = tmp_path / "runtime"
    artifacts = replay_case(
        "real_id45",
        runtime_root=runtime_root,
        gemini_video_client=FakeGeminiVideoClient(default_result=fake_gemini_pool()),
    )

    validation = validate_run(artifacts.run_id, LocalRuntimeFileStore(runtime_root))

    assert artifacts.state["status"] == "success"
    assert validation["status"] == "pass"
    assert artifacts.files["final_output.json"]["validation_status"] == "pass"
    assert artifacts.files["discovered_content_items.jsonl"]
    assert artifacts.files["pattern_recall_evidence.jsonl"]
    assert artifacts.decisions

    for item in artifacts.files["discovered_content_items.jsonl"]:
        assert "statistics" in item
        assert "platform_raw_payload" in item
        assert "raw_payload" in item

    for evidence in artifacts.files["pattern_recall_evidence.jsonl"]:
        summary = evidence["evidence_summary"]
        assert summary["schema_version"] == "v4_gemini_query_relevance.v1"
        assert summary["final_status"] in {"ok", "success"}
        assert "query_relevance_score" in summary
        assert "fit_senior_50plus" not in summary
        assert "relevance_score" not in summary

    for decision in artifacts.decisions:
        scorecard = decision["scorecard"]
        replay_data = decision["decision_replay_data"]
        assert scorecard["schema_version"] == "v4_scorecard.v1"
        assert "platform_performance_score" in scorecard
        assert "platform_performance_components" in scorecard
        assert "missing_observable_fields" in scorecard
        assert "allow_walk" in replay_data
        assert "walk_gate_snapshot" in replay_data
        assert "platform_heat" not in scorecard


def test_v4_m3_scoring_replay_routes_technical_failure_to_retry(tmp_path):
    failed_result = fake_gemini_fail()
    failed_result["response_body_summary"] = {
        "http_status_code": 502,
        "json_top_level_keys": ["error"],
        "text_excerpt": "Provider returned error",
    }
    artifacts = replay_case(
        "real_id45",
        runtime_root=tmp_path / "runtime",
        gemini_video_client=FakeGeminiVideoClient(default_result=failed_result),
    )

    assert artifacts.summary["pooled_content_count"] == 0
    assert artifacts.summary["rejected_content_count"] == 0
    assert artifacts.summary["review_content_count"] == 0
    assert artifacts.summary["technical_retry_content_count"] == len(artifacts.decisions)
    assert {d["decision_reason_code"] for d in artifacts.decisions} == {"v4_technical_retry_needed"}
    assert {d["decision_action"] for d in artifacts.decisions} == {"TECHNICAL_RETRY_REQUIRED"}
    assert all(d["decision_replay_data"]["allow_walk"] is False for d in artifacts.decisions)
    assert all(d["scorecard"]["schema_version"] == "v4_scorecard.v1" for d in artifacts.decisions)
    first_evidence = artifacts.files["pattern_recall_evidence.jsonl"][0]
    assert first_evidence["evidence_summary"]["response_body_summary"]["http_status_code"] == 502
    assert first_evidence["raw_payload"]["response_body_summary"]["text_excerpt"] == "Provider returned error"


def test_v4_m3_db_runtime_preserves_scoring_json_containers(tmp_path):
    artifacts = replay_case(
        "real_id45",
        runtime_root=tmp_path / "runtime",
        gemini_video_client=FakeGeminiVideoClient(default_result=fake_gemini_pool()),
    )
    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    store.append_jsonl(
        artifacts.run_id,
        "discovered_content_items.jsonl",
        artifacts.files["discovered_content_items.jsonl"][:1],
    )
    store.append_jsonl(
        artifacts.run_id,
        "pattern_recall_evidence.jsonl",
        artifacts.files["pattern_recall_evidence.jsonl"][:1],
    )
    store.append_jsonl(artifacts.run_id, "rule_decisions.jsonl", artifacts.decisions[:1])

    inserted = [(_table_name(sql), _insert_values(sql, params)) for sql, params in connection.statements]
    content_row = _only(inserted, "content_agent_discovered_content_items")
    evidence_row = _only(inserted, "content_agent_pattern_recall_evidence")
    decision_row = _only(inserted, "content_agent_rule_decisions")

    content_raw = json.loads(content_row["raw_payload"])
    assert json.loads(content_row["statistics"]) == content_raw["statistics"]
    assert json.loads(content_row["platform_raw_payload"]) == content_raw["platform_raw_payload"]

    evidence_summary = json.loads(evidence_row["evidence_summary"])
    evidence_raw = json.loads(evidence_row["raw_payload"])
    assert evidence_summary["schema_version"] == "v4_gemini_query_relevance.v1"
    for key, value in evidence_summary.items():
        assert evidence_raw[key] == value

    scorecard = json.loads(decision_row["scorecard"])
    replay_data = json.loads(decision_row["decision_replay_data"])
    decision_raw = json.loads(decision_row["raw_payload"])
    assert scorecard["schema_version"] == "v4_scorecard.v1"
    assert replay_data["allow_walk"] in {True, False}
    assert "scorecard" not in decision_raw
    assert "decision_replay_data" not in decision_raw
    assert decision_raw["record_schema_version"] == "runtime_record.v1"


def _table_name(sql: str) -> str:
    return sql.split("`", 2)[1]


def _only(rows: list[tuple[str, dict]], table: str) -> dict:
    matches = [values for row_table, values in rows if row_table == table]
    assert len(matches) == 1
    return matches[0]