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

from content_agent.business_modules import learning_review
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from tests.test_p8_strategy_review import _clue, _write_minimal_runtime


def test_strategy_review_uses_single_run_policy_context(tmp_path):
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    run_id = "run_single"
    policy_run_id = "policy_single"
    other_run_id = "run_other"
    other_policy_run_id = "policy_other"
    runtime.prepare_run(run_id)
    runtime.prepare_run(other_run_id)
    _write_minimal_runtime(runtime, run_id, policy_run_id)
    _write_minimal_runtime(runtime, other_run_id, other_policy_run_id)
    runtime.append_jsonl(
        other_run_id,
        "search_clues.jsonl",
        [_clue(other_run_id, other_policy_run_id, "clue_other", "q_other", "跨 run", "success", 99)],
    )

    review = learning_review.run(run_id, policy_run_id, runtime)

    assert review["data_window"]["scope"] == "single_run"
    assert review["data_window"]["run_id"] == run_id
    assert review["data_window"]["policy_run_id"] == policy_run_id
    assert review["data_window"]["policy_context"]["strategy_version"] == "V1"
    assert review["data_window"]["policy_context"]["walk_strategy_version"] == "V1.0"
    assert review["metric_summary"]["success_query_count"] == 1
    assert all(
        item["clue_id"] != "clue_other"
        for bucket in review["query_review"].values()
        if isinstance(bucket, list)
        for item in bucket
    )