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

from __future__ import annotations

from content_agent.run_service import RunService
from content_agent.schemas import RunStartRequest
from tests.test_p0d_p0g import _SpyRuntimeStore
from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE


def test_rule_decisions_and_policy_run_record_include_replay_metadata(tmp_path):
    runtime = _SpyRuntimeStore(tmp_path / "runtime")
    service = RunService(
        runtime=runtime,
        query_variant_client=FakeQueryVariantClient(),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )
    decision = service.read_jsonl(state["run_id"], "rule_decisions.jsonl")[0]
    replay = decision["decision_replay_data"]

    assert replay["policy_bundle_hash"] == state["policy_bundle"]["policy_bundle_hash"]
    assert replay["dispatch_id"] == "dispatch_content"
    assert replay["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"
    assert replay["strategy_version"] == "V4"
    assert replay["effect_mapping_id"] == "map_reject_failed"
    assert replay["allow_walk"] is False
    assert replay["walk_gate_snapshot"]["query_relevance_score"] == 80

    policy_run = runtime.policy_runs[0]
    assert policy_run["policy_bundle_hash"] == state["policy_bundle"]["policy_bundle_hash"]
    assert policy_run["raw_payload"]["dispatch"]["dispatch_id"] == "dispatch_content"
    assert policy_run["raw_payload"]["runtime_status_contract"]["query_effect_status"] == [
        "success",
        "pending",
        "failed",
        "rule_blocked",
    ]
    assert policy_run["decision_summary"]["effect_status_counts"]