diff --git a/agent-team/agent_team/nodes/planner.py b/agent-team/agent_team/nodes/planner.py index 6d55b65..8118590 100644 --- a/agent-team/agent_team/nodes/planner.py +++ b/agent-team/agent_team/nodes/planner.py @@ -33,6 +33,7 @@ state-transition function that is trivially unit-testable. from __future__ import annotations import json +import logging import re from collections.abc import Callable from typing import Any @@ -40,6 +41,8 @@ from typing import Any from agent_team.billing import ClaudeResult, claude_invoke from agent_team.task_model import Phase, PipelineState, TaskStatus +_log = logging.getLogger(__name__) + # Optional context-provider callable (WS5): () -> str. When injected, its # result is prepended to the planner prompt. Default None = unchanged behavior. ContextProvider = Callable[[], str] @@ -57,6 +60,14 @@ __all__ = [ # Adam (parked + ALARM) instead of looping forever. MAX_PLAN_REVISIONS = 3 +# Per-call turn headroom for the planner's single-shot Claude invoke. The +# subscription invoker defaults to 1 turn, which the planner intermittently +# exhausts before it can finish emitting its JSON ("Reached maximum number of +# turns (1)"). 4 gives enough room to FINISH the reply without inviting the +# model to wander off-task — tools stay disabled (allowed_tools default []), so +# the extra turns buy completion, not exploration. +_PLANNER_MAX_TURNS = 4 + # The reviewer verdict string that sends a plan back to the planner. Kept here # (rather than imported from a review node that does not exist yet) so this leaf # stays self-contained; the review leaf will emit this same token. @@ -206,6 +217,9 @@ def build_plan_prompt( '"summary" (string), "phases" (a non-empty list of objects each with ' '"name" and "steps", where "steps" is a non-empty list of strings). ' "Do not include prose outside the JSON.", + "Do NOT use any tools — respond directly with ONLY the JSON object. A " + "tool call would consume the single available turn before the plan is " + "emitted.", ] return "\n".join(sections) @@ -283,6 +297,22 @@ def _revision_count(state: PipelineState) -> int: return count +def _is_transient_failure(exc: Exception) -> bool: + """Decide whether a planner-call failure is worth retrying exactly once. + + TRANSIENT (retry): the single-shot Claude call hit the turn cap before it + could finish ("Reached maximum number of turns"), or the model returned an + EMPTY reply (``PlannerError`` whose message mentions "empty") — both can + succeed on a fresh attempt. Anything else (malformed JSON, missing/blank + phases) is DETERMINISTIC: a retry would just reproduce the same garbage, so + we let it propagate and fail fast. + """ + message = str(exc).lower() + if "maximum number of turns" in message: + return True + return isinstance(exc, PlannerError) and "empty" in message + + def plan_node( state: PipelineState, config: dict[str, Any] | None = None, @@ -317,8 +347,30 @@ def plan_node( ) prompt = build_plan_prompt(state, context_provider=context_provider) - result: ClaudeResult = claude_invoke(prompt, config=config) - plan = parse_plan(result.text) + + # Auto-retry-once on a TRANSIENT failure only (turn-cap exhaustion or an + # empty reply) — both can clear on a fresh attempt. A DETERMINISTIC failure + # (malformed JSON, missing/blank phases) propagates immediately: re-asking + # the same prompt would just reproduce it. + attempts = 0 + while True: + attempts += 1 + try: + result: ClaudeResult = claude_invoke( + prompt, config=config, max_turns=_PLANNER_MAX_TURNS + ) + plan = parse_plan(result.text) + break + except Exception as exc: # noqa: BLE001 - reclassified below + if attempts == 1 and _is_transient_failure(exc): + _log.warning( + "planner: transient failure on attempt %d (%s); retrying once", + attempts, + exc, + ) + continue + raise + # Record how many times we have planned so review/observability can see it. plan["revision"] = revisions diff --git a/agent-team/tests/test_planner.py b/agent-team/tests/test_planner.py index 988ddad..6e670c1 100644 --- a/agent-team/tests/test_planner.py +++ b/agent-team/tests/test_planner.py @@ -313,3 +313,75 @@ def test_plan_node_unconfigured_invoker_raises() -> None: billing._invoker = billing._unconfigured_invoker with pytest.raises(RuntimeError, match="no invoker bound"): plan_node(_state(plan={"task": "x"})) + + +# --------------------------------------------------------------------------- # +# plan_node — turn headroom + auto-retry-once (Phase A) +# --------------------------------------------------------------------------- # + + +def test_plan_node_passes_max_turns_to_invoke_seam() -> None: + # The single-shot Claude default (1 turn) is flaky; the planner asks for + # headroom so the model can FINISH its JSON. + calls = _bind_invoker(json.dumps(_VALID_PLAN)) + plan_node(_state(plan={"task": "x"})) + assert calls[0]["kw"].get("max_turns") == 4 + + +def _bind_sequenced_invoker( + replies: list[Any], +) -> list[dict[str, Any]]: + """Bind an invoker that, per call, raises if the next item is an Exception + or returns it as the reply text otherwise. Captures its calls.""" + calls: list[dict[str, Any]] = [] + queue = list(replies) + + def _fake(prompt: str, *, mode: BillingMode, **kw: Any) -> ClaudeResult: + calls.append({"prompt": prompt, "mode": mode, "kw": kw}) + item = queue.pop(0) + if isinstance(item, Exception): + raise item + return ClaudeResult(text=item, mode=mode, usage={"input_tokens": 1}) + + billing.set_invoker(_fake) + return calls + + +def test_plan_node_retries_once_on_turn_cap_then_succeeds() -> None: + # First call exhausts the single turn; the retry produces a valid plan. + calls = _bind_sequenced_invoker( + [ + Exception( + "Claude Code returned an error result: " + "Reached maximum number of turns (1)" + ), + json.dumps(_VALID_PLAN), + ] + ) + out = plan_node(_state(plan={"task": "x"})) + assert out["current_phase"] == Phase.REVIEW.value + assert out["plan"]["phases"][0]["name"] == "Phase 1 — bump" + assert len(calls) == 2 # exactly one retry + + +def test_plan_node_retries_once_on_empty_reply_then_succeeds() -> None: + # An EMPTY model reply is transient (PlannerError "empty"): retry once. + calls = _bind_sequenced_invoker([" ", json.dumps(_VALID_PLAN)]) + out = plan_node(_state(plan={"task": "x"})) + assert out["current_phase"] == Phase.REVIEW.value + assert len(calls) == 2 # one retry + + +def test_plan_node_does_not_retry_on_malformed_json() -> None: + # A non-empty but invalid reply is DETERMINISTIC: a retry would reproduce + # it, so the planner fails fast on the first attempt. + calls = _bind_sequenced_invoker(["not json {", json.dumps(_VALID_PLAN)]) + with pytest.raises(PlannerError): + plan_node(_state(plan={"task": "x"})) + assert len(calls) == 1 # no second call + + +def test_build_prompt_includes_no_tools_json_only_instruction() -> None: + prompt = build_plan_prompt(_state(plan={"task": "x"})) + assert "Do NOT use any tools" in prompt + assert "ONLY the JSON object" in prompt