test_attempt_input_binding.py 3.9 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. import pytest
  2. from agent.orchestration.coordinator import OrchestrationError
  3. from agent.orchestration.models import (
  4. DecisionAction,
  5. FailureCode,
  6. TaskStatus,
  7. ValidationVerdict,
  8. )
  9. from test_coordinator_integration import FakeExecutor, create_task, make_coordinator
  10. async def _accept_child(coordinator, task_id, key):
  11. cycle = (await coordinator.dispatch_tasks("root", [task_id]))[0]
  12. await coordinator.decide_task(
  13. "root",
  14. task_id,
  15. cycle.validation_id,
  16. DecisionAction.ACCEPT,
  17. {"reason": "accepted"},
  18. key,
  19. )
  20. return cycle
  21. @pytest.mark.asyncio
  22. async def test_new_attempt_distinguishes_empty_and_ordered_child_bindings(tmp_path):
  23. executor = FakeExecutor([ValidationVerdict.PASSED, ValidationVerdict.PASSED])
  24. coordinator, store, _ = await make_coordinator(tmp_path, executor)
  25. first = await create_task(coordinator, "first")
  26. second = await create_task(coordinator, "second")
  27. await _accept_child(coordinator, second, "accept-second")
  28. await _accept_child(coordinator, first, "accept-first")
  29. ledger = await store.load("root")
  30. root = ledger.tasks[ledger.root_task_id]
  31. reserved = await coordinator._create_attempt("root", root.task_id, "worker", None)
  32. frozen = (await store.load("root")).attempts[reserved["attempt_id"]]
  33. assert frozen.accepted_child_decision_ids == (
  34. ledger.tasks[first].decision_ids[-1],
  35. ledger.tasks[second].decision_ids[-1],
  36. )
  37. first_attempt = ledger.attempts[ledger.tasks[first].attempt_ids[-1]]
  38. assert first_attempt.accepted_child_decision_ids == ()
  39. @pytest.mark.asyncio
  40. @pytest.mark.parametrize("corruption", ["duplicate", "reverse", "cross", "non_accept"])
  41. async def test_frozen_child_decision_binding_rejects_corruption(tmp_path, corruption):
  42. executor = FakeExecutor([ValidationVerdict.PASSED, ValidationVerdict.PASSED])
  43. coordinator, store, _ = await make_coordinator(tmp_path, executor)
  44. first = await create_task(coordinator, "first")
  45. second = await create_task(coordinator, "second")
  46. await _accept_child(coordinator, first, "accept-first")
  47. await _accept_child(coordinator, second, "accept-second")
  48. ledger = await store.load("root")
  49. root = ledger.tasks[ledger.root_task_id]
  50. reserved = await coordinator._create_attempt("root", root.task_id, "worker", None)
  51. ledger = await store.load("root")
  52. attempt = ledger.attempts[reserved["attempt_id"]]
  53. ids = list(attempt.accepted_child_decision_ids)
  54. if corruption == "duplicate":
  55. attempt.accepted_child_decision_ids = (ids[0], ids[0])
  56. elif corruption == "reverse":
  57. attempt.accepted_child_decision_ids = tuple(reversed(ids))
  58. elif corruption == "cross":
  59. attempt.accepted_child_decision_ids = ("missing-decision",)
  60. else:
  61. decision = ledger.decisions[ids[0]]
  62. object.__setattr__(decision, "action", DecisionAction.RETRY)
  63. with pytest.raises(OrchestrationError):
  64. coordinator._accepted_child_results(ledger, root, attempt)
  65. @pytest.mark.asyncio
  66. async def test_legacy_running_attempt_never_guesses_child_inputs(tmp_path):
  67. executor = FakeExecutor([])
  68. coordinator, store, _ = await make_coordinator(tmp_path, executor)
  69. root_id = (await store.load("root")).root_task_id
  70. reserved = await coordinator._create_attempt("root", root_id, "worker", None)
  71. ledger = await store.load("root")
  72. ledger.attempts[reserved["attempt_id"]].accepted_child_decision_ids = None
  73. await store.commit(ledger, expected_revision=ledger.revision)
  74. result = await coordinator.advance_cycle("root", root_id, reserved["attempt_id"])
  75. recovered = await store.load("root")
  76. attempt = recovered.attempts[reserved["attempt_id"]]
  77. assert result.error == "Legacy running attempt has no frozen child input binding"
  78. assert executor.worker_calls == 0
  79. assert recovered.tasks[root_id].status == TaskStatus.NEEDS_REPLAN
  80. assert attempt.execution_stats.failure_code == FailureCode.PROTOCOL_VIOLATION