Skip to content

Commit 077f1f5

Browse files
CuSO41108CuSO41108
committed
fix(plan): pause non-replanable node failures
1 parent 62d2c6d commit 077f1f5

5 files changed

Lines changed: 304 additions & 67 deletions

File tree

‎src/agent_app/plan/agent_runner.py‎

Lines changed: 56 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -54,50 +54,71 @@ def __call__(self, context: PlanNodeContext) -> NodeExecutionResult:
5454
plan_revision_id=context.revision.id,
5555
plan_node_id=context.node.id,
5656
)
57-
if result.pending_action is not None or result.task_status == "waiting_user":
58-
pending = result.pending_action
59-
return NodeExecutionResult(
60-
status="waiting_approval",
61-
metadata={
62-
"pending_action_id": None if pending is None else pending.id,
63-
"kind": None if pending is None else pending.kind,
64-
},
65-
)
66-
if result.stop_reason == "max_tool_rounds_exceeded" or result.task_status == "paused":
57+
return node_execution_result_from_turn(context, result)
58+
59+
60+
def node_execution_result_from_turn(
61+
context: PlanNodeContext,
62+
result: Any,
63+
) -> NodeExecutionResult:
64+
"""Apply the same durable node outcome policy to new and resumed turns."""
65+
66+
if result.pending_action is not None or result.task_status == "waiting_user":
67+
pending = result.pending_action
68+
return NodeExecutionResult(
69+
status="waiting_approval",
70+
metadata={
71+
"pending_action_id": None if pending is None else pending.id,
72+
"kind": None if pending is None else pending.kind,
73+
},
74+
)
75+
if result.stop_reason == "max_tool_rounds_exceeded" or result.task_status == "paused":
76+
return NodeExecutionResult(
77+
status="paused",
78+
error="Node execution window exhausted; explicit continuation is required.",
79+
metadata={
80+
"stop_reason": result.stop_reason,
81+
},
82+
)
83+
if result.stop_reason == "repeated_deterministic_tool_failure":
84+
return NodeExecutionResult(
85+
status="paused",
86+
error=(
87+
"repeated_deterministic_tool_failure: "
88+
+ (result.final_text or "The same non-retryable tool target failed again.")
89+
),
90+
metadata={
91+
"stop_reason": result.stop_reason,
92+
"failure_category": "deterministic_tool_failure",
93+
},
94+
)
95+
if result.success:
96+
validation = _validate_acceptance_evidence(context, result.tool_runs)
97+
if validation["status"] != "passed":
6798
return NodeExecutionResult(
6899
status="paused",
69-
error="Node execution window exhausted; explicit continuation is required.",
70-
metadata={
71-
"stop_reason": result.stop_reason,
72-
},
73-
)
74-
if result.success:
75-
validation = _validate_acceptance_evidence(context, result.tool_runs)
76-
if validation["status"] != "passed":
77-
return NodeExecutionResult(
78-
status="failed",
79-
error=_acceptance_failure_message(context, validation),
80-
evidence_refs=tuple(validation["evidence_refs"]),
81-
metadata={
82-
"stop_reason": "acceptance_evidence_missing",
83-
"failure_category": "acceptance_not_met",
84-
"acceptance_validation": validation,
85-
},
86-
)
87-
return NodeExecutionResult(
88-
status="completed",
89-
output=result.final_text,
100+
error=_acceptance_failure_message(context, validation),
90101
evidence_refs=tuple(validation["evidence_refs"]),
91102
metadata={
92-
"stop_reason": result.stop_reason,
103+
"stop_reason": "acceptance_evidence_missing",
104+
"failure_category": "acceptance_not_met",
93105
"acceptance_validation": validation,
94106
},
95107
)
96108
return NodeExecutionResult(
97-
status="failed",
98-
error=result.final_text or result.stop_reason or "AgentLoop node execution failed.",
99-
metadata={"stop_reason": result.stop_reason},
109+
status="completed",
110+
output=result.final_text,
111+
evidence_refs=tuple(validation["evidence_refs"]),
112+
metadata={
113+
"stop_reason": result.stop_reason,
114+
"acceptance_validation": validation,
115+
},
100116
)
117+
return NodeExecutionResult(
118+
status="failed",
119+
error=result.final_text or result.stop_reason or "AgentLoop node execution failed.",
120+
metadata={"stop_reason": result.stop_reason},
121+
)
101122

102123

103124
def build_node_prompt(context: PlanNodeContext) -> str:

‎src/agent_app/plan/service.py‎

Lines changed: 114 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,11 @@
55
from typing import Any
66
from uuid import uuid4
77

8-
from agent_app.plan.agent_runner import PlanAgentNodeRunner, build_node_prompt
8+
from agent_app.plan.agent_runner import (
9+
PlanAgentNodeRunner,
10+
build_node_prompt,
11+
node_execution_result_from_turn,
12+
)
913
from agent_app.plan.executor import PlanExecutionResult, PlanExecutor, PlanNodeContext
1014
from agent_app.plan.graph import PlanGraph, resource_claims_for_node
1115
from agent_app.plan.planner import PlanPlanner, PlanPlanningError, PlannerAttemptHook
@@ -42,6 +46,13 @@ class ToolActionResolutionResult:
4246
decision: RecoveryDecision
4347

4448

49+
_CONTINUATION_BUDGET_PAUSE_REASONS = {
50+
"max_tool_rounds_exceeded",
51+
"acceptance_evidence_missing",
52+
"repeated_deterministic_tool_failure",
53+
}
54+
55+
4556
class PlanTaskService:
4657
"""Coordinate planning, PlanGraph persistence, and bounded node execution."""
4758

@@ -478,7 +489,7 @@ def resume(self, *, task_id: str, decision: RecoveryDecision | None = None) -> P
478489

479490
task = self._require_task(task_id)
480491
if fresh.kind == RecoveryKind.PAUSED:
481-
if task.stop_reason == "max_tool_rounds_exceeded":
492+
if task.stop_reason in _CONTINUATION_BUDGET_PAUSE_REASONS:
482493
if task.budget.used_continuations >= task.budget.max_continuations:
483494
failed = self._tasks.fail(task.id, reason="continuation_budget_exceeded")
484495
return self._finalize_failed_execution(
@@ -503,7 +514,7 @@ def resume(self, *, task_id: str, decision: RecoveryDecision | None = None) -> P
503514
)
504515
task = self._tasks.consume_continuation(
505516
task.id,
506-
reason="resume_after_max_tool_rounds",
517+
reason=f"resume_after_{task.stop_reason}",
507518
)
508519
else:
509520
task = self._tasks.resume(
@@ -932,17 +943,22 @@ def handle_approval(
932943
),
933944
)
934945

935-
next_status = "completed" if approved and turn_result.success else "failed"
936-
result_record = {
937-
"status": next_status,
938-
"output": turn_result.final_text,
939-
"error": None if next_status == "completed" else (
940-
turn_result.final_text or turn_result.stop_reason or "approval_rejected"
941-
),
942-
"metadata": {
943-
"approval": "approved" if approved else "rejected",
944-
"stop_reason": turn_result.stop_reason,
945-
},
946+
if approved:
947+
node_outcome = node_execution_result_from_turn(context, turn_result)
948+
next_status = node_outcome.status
949+
result_record = node_outcome.to_record()
950+
else:
951+
next_status = "failed"
952+
result_record = {
953+
"status": next_status,
954+
"output": turn_result.final_text,
955+
"error": turn_result.final_text or turn_result.stop_reason or "approval_rejected",
956+
"evidence_refs": [],
957+
"metadata": {"stop_reason": turn_result.stop_reason},
958+
}
959+
result_record["metadata"] = {
960+
**result_record.get("metadata", {}),
961+
"approval": "approved" if approved else "rejected",
946962
}
947963
revision = self._plan_store.update_node_status(
948964
revision.id,
@@ -1019,17 +1035,12 @@ def handle_user_message(
10191035
),
10201036
)
10211037

1022-
next_status = "completed" if turn_result.success else "failed"
1023-
result_record = {
1024-
"status": next_status,
1025-
"output": turn_result.final_text,
1026-
"error": None if next_status == "completed" else (
1027-
turn_result.final_text or turn_result.stop_reason or "user_message_resume_failed"
1028-
),
1029-
"metadata": {
1030-
"resume_kind": "ask_user",
1031-
"stop_reason": turn_result.stop_reason,
1032-
},
1038+
node_outcome = node_execution_result_from_turn(context, turn_result)
1039+
next_status = node_outcome.status
1040+
result_record = node_outcome.to_record()
1041+
result_record["metadata"] = {
1042+
**result_record.get("metadata", {}),
1043+
"resume_kind": "ask_user",
10331044
}
10341045
revision = self._plan_store.update_node_status(
10351046
revision.id,
@@ -1108,6 +1119,12 @@ def _execute_and_reconcile(
11081119
replan_error=exc,
11091120
)
11101121
if execution.status == "paused":
1122+
if task.status == "running":
1123+
task = self._pause_for_node_recovery(
1124+
task.id,
1125+
revision=after,
1126+
execution=execution,
1127+
)
11111128
return PlanTaskResult(
11121129
task=task,
11131130
revision=self._plan_store.get_revision_by_id(execution.revision.id),
@@ -1120,6 +1137,78 @@ def _execute_and_reconcile(
11201137
latest = self._plan_store.get_revision_by_id(execution.revision.id)
11211138
return PlanTaskResult(task=task, revision=latest, execution=execution)
11221139

1140+
def _pause_for_node_recovery(
1141+
self,
1142+
task_id: str,
1143+
*,
1144+
revision: PlanRevision,
1145+
execution: PlanExecutionResult,
1146+
) -> TaskState:
1147+
node_id = execution.waiting_node_id
1148+
node_result = revision.node_results.get(node_id or "", {})
1149+
metadata = node_result.get("metadata", {}) if isinstance(node_result, dict) else {}
1150+
if not isinstance(metadata, dict):
1151+
metadata = {}
1152+
stop_reason = str(metadata.get("stop_reason") or "plan_node_paused")
1153+
failure_category = str(metadata.get("failure_category") or "node_recovery")
1154+
task = self._tasks.pause_for_recovery(task_id, reason=stop_reason)
1155+
latest = self._sessions.get_latest_checkpoint(task_id)
1156+
run = self._sessions.create_execution_run(
1157+
task_id=task_id,
1158+
agent_id="plan_runtime",
1159+
scope=f"plan_node_recovery:{node_id or 'unknown'}",
1160+
max_tool_rounds=1,
1161+
parent_checkpoint_id=None if latest is None else latest.id,
1162+
)
1163+
state = {
1164+
"phase": "plan_node_recovery",
1165+
"plan_id": revision.graph.id,
1166+
"revision": revision.graph.revision,
1167+
"revision_id": revision.id,
1168+
"node_id": node_id,
1169+
"stop_reason": stop_reason,
1170+
"failure_category": failure_category,
1171+
"failure_detail": execution.failure_reason,
1172+
"recoverable": True,
1173+
}
1174+
checkpoint = self._sessions.create_checkpoint(
1175+
task_id=task_id,
1176+
run_id=run.id,
1177+
cursor="paused_by_user",
1178+
status="paused_by_user",
1179+
state=state,
1180+
)
1181+
self._sessions.update_execution_run(
1182+
run.id,
1183+
status="paused_by_user",
1184+
stop_reason=stop_reason,
1185+
)
1186+
self._sessions.append_task_trace(
1187+
task_id,
1188+
"checkpoint",
1189+
{
1190+
"checkpoint_id": checkpoint.id,
1191+
"run_id": run.id,
1192+
"parent_checkpoint_id": checkpoint.parent_checkpoint_id,
1193+
"cursor": checkpoint.cursor,
1194+
"status": checkpoint.status,
1195+
**state,
1196+
},
1197+
)
1198+
self._sessions.append_task_trace(
1199+
task_id,
1200+
"plan_node_recovery_available",
1201+
{
1202+
"checkpoint_id": checkpoint.id,
1203+
"node_id": node_id,
1204+
"stop_reason": stop_reason,
1205+
"failure_category": failure_category,
1206+
"recoverable": True,
1207+
"continuation_command": f"/continue {task_id[:8]}",
1208+
},
1209+
)
1210+
return task
1211+
11231212
def _finalize_failed_execution(
11241213
self,
11251214
task_id: str,

‎tests/integration/test_cli_flow.py‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -454,7 +454,12 @@ def test_cli_resumes_plan_ask_user_with_natural_language_answer(self, mock_from_
454454
]),
455455
]
456456
)
457-
second_model = _FakeModelClient([_text_response("README inspected")])
457+
second_model = _FakeModelClient([
458+
_tool_call_response([
459+
ToolCall(id="ask-read", name="file_read", arguments={"path": "README.md"}),
460+
]),
461+
_text_response("README inspected"),
462+
])
458463
mock_from_config.side_effect = [first_model, second_model]
459464

460465
first_stdout = io.StringIO()

‎tests/unit/test_plan_agent_runner.py‎

Lines changed: 26 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ def test_runner_rejects_success_without_required_tool_evidence(self) -> None:
102102

103103
outcome = PlanAgentNodeRunner(loop)(_context())
104104

105-
self.assertEqual(outcome.status, "failed")
105+
self.assertEqual(outcome.status, "paused")
106106
self.assertIn("acceptance_evidence_missing", outcome.error or "")
107107
self.assertEqual(outcome.metadata["failure_category"], "acceptance_not_met")
108108

@@ -123,7 +123,7 @@ def test_runner_rejects_code_search_with_no_matches(self) -> None:
123123
_context(allowed_tools=["code_search"])
124124
)
125125

126-
self.assertEqual(outcome.status, "failed")
126+
self.assertEqual(outcome.status, "paused")
127127
self.assertEqual(outcome.evidence_refs, ())
128128

129129
def test_runner_requires_task_uuid_to_appear_in_tool_evidence(self) -> None:
@@ -146,7 +146,7 @@ def test_runner_requires_task_uuid_to_appear_in_tool_evidence(self) -> None:
146146
_context(objective=f"Locate Trace for Task {task_uuid}.")
147147
)
148148

149-
self.assertEqual(outcome.status, "failed")
149+
self.assertEqual(outcome.status, "paused")
150150
self.assertEqual(
151151
outcome.metadata["acceptance_validation"]["missing_anchors"],
152152
[task_uuid],
@@ -195,9 +195,31 @@ def test_edit_node_requires_successful_write_evidence(self) -> None:
195195
)
196196
)
197197

198-
self.assertEqual(outcome.status, "failed")
198+
self.assertEqual(outcome.status, "paused")
199199
self.assertIn("replace_in_file", outcome.error or "")
200200

201+
def test_runner_pauses_repeated_deterministic_tool_failure(self) -> None:
202+
loop = _FakeLoop(
203+
TurnResult(
204+
session_id="session-1",
205+
final_text="Search path not found: 'agent_app'.",
206+
stop_reason="repeated_deterministic_tool_failure",
207+
tool_runs=[],
208+
success=False,
209+
task_id="task-1",
210+
task_status="running",
211+
)
212+
)
213+
214+
outcome = PlanAgentNodeRunner(loop)(_context())
215+
216+
self.assertEqual(outcome.status, "paused")
217+
self.assertEqual(
218+
outcome.metadata["failure_category"],
219+
"deterministic_tool_failure",
220+
)
221+
self.assertIn("Search path not found", outcome.error or "")
222+
201223
def test_runner_turns_pending_user_action_into_waiting_approval(self) -> None:
202224
loop = _FakeLoop(
203225
TurnResult(

0 commit comments

Comments
 (0)