test_p8_no_auto_mutation.py 2.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
  1. from content_agent.business_modules import learning_review
  2. from content_agent.integrations.runtime_files import LocalRuntimeFileStore
  3. from tests.test_p8_performance_feedback import FeedbackRuntime
  4. from tests.test_p8_strategy_review import _write_minimal_runtime
  5. def test_strategy_review_does_not_rewrite_runtime_inputs(tmp_path):
  6. runtime = LocalRuntimeFileStore(tmp_path / "runtime")
  7. run_id = "run_no_auto"
  8. policy_run_id = "policy_no_auto"
  9. runtime.prepare_run(run_id)
  10. _write_minimal_runtime(runtime, run_id, policy_run_id)
  11. final_before = runtime.read_json(run_id, "final_output.json")
  12. decisions_before = runtime.read_jsonl(run_id, "rule_decisions.jsonl")
  13. clues_before = runtime.read_jsonl(run_id, "search_clues.jsonl")
  14. review = learning_review.run(run_id, policy_run_id, runtime)
  15. assert runtime.read_json(run_id, "final_output.json") == final_before
  16. assert runtime.read_jsonl(run_id, "rule_decisions.jsonl") == decisions_before
  17. assert runtime.read_jsonl(run_id, "search_clues.jsonl") == clues_before
  18. assert all(item["requires_human_approval"] is True for item in review["recommendations"])
  19. forbidden_patch_fields = {
  20. "auto_" + "apply",
  21. "rule_pack_" + "patch",
  22. "walk_strategy_" + "patch",
  23. "budget_" + "patch",
  24. "publish_job_" + "patch",
  25. }
  26. for item in review["recommendations"]:
  27. assert forbidden_patch_fields.isdisjoint(item)
  28. def test_strategy_review_with_feedback_still_does_not_emit_auto_patches(tmp_path):
  29. run_id = "run_feedback_no_auto"
  30. policy_run_id = "policy_feedback_no_auto"
  31. runtime = FeedbackRuntime(
  32. tmp_path / "runtime",
  33. feedback_rows=[
  34. {
  35. "run_id": run_id,
  36. "policy_run_id": policy_run_id,
  37. "feedback_id": "feedback_no_auto",
  38. "feedback_status": "available",
  39. "completion_rate": 0.8,
  40. }
  41. ],
  42. )
  43. runtime.prepare_run(run_id)
  44. _write_minimal_runtime(runtime, run_id, policy_run_id)
  45. review = learning_review.run(run_id, policy_run_id, runtime)
  46. forbidden_patch_fields = {
  47. "auto_" + "apply",
  48. "rule_pack_" + "patch",
  49. "walk_strategy_" + "patch",
  50. "budget_" + "patch",
  51. "threshold_" + "patch",
  52. "publish_job_" + "patch",
  53. }
  54. assert all(item["requires_human_approval"] is True for item in review["recommendations"])
  55. for item in review["recommendations"]:
  56. assert forbidden_patch_fields.isdisjoint(item)