test_failure_contract.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152
  1. from __future__ import annotations
  2. import json
  3. import pytest
  4. from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolResult
  5. def test_failure_detail_is_bounded_json_safe_and_round_trips() -> None:
  6. failure = FailureDetail(
  7. code="INPUT_SCOPE_MISMATCH",
  8. message="x" * 3_000,
  9. disposition=FailureDisposition.REPLAN_TASK,
  10. source_tool="save_candidate",
  11. details={"task_id": "task-1", "scope": ["paragraph", 3]},
  12. )
  13. assert len(failure.message) == 2_000
  14. assert FailureDetail.from_dict(failure.to_dict()) == failure
  15. assert failure.fingerprint() == FailureDetail(
  16. code="INPUT_SCOPE_MISMATCH",
  17. message="different wording",
  18. disposition=FailureDisposition.REPLAN_TASK,
  19. source_tool="save_candidate",
  20. details={"task_id": "task-2", "scope": ["paragraph", 3]},
  21. ).fingerprint()
  22. def test_failure_detail_rejects_non_json_details() -> None:
  23. with pytest.raises(ValueError, match="JSON values"):
  24. FailureDetail(
  25. code="INVALID",
  26. message="invalid",
  27. disposition=FailureDisposition.RETRY_CALL,
  28. details={"value": object()},
  29. )
  30. def test_tool_result_exposes_structured_failure_to_model() -> None:
  31. failure = FailureDetail(
  32. code="REJECTED",
  33. message="retry with another value",
  34. disposition=FailureDisposition.RETRY_CALL,
  35. source_tool="sample",
  36. )
  37. result = ToolResult(title="rejected", output="context", failure=failure)
  38. payload = json.loads(result.to_llm_message())
  39. assert payload["failure"] == failure.to_dict()
  40. assert payload["output"] == "context"
  41. assert ToolExecutionError(failure).failure == failure