fix(agent-team): planner turn headroom + classified retry-once (Phase A)

The planner's single-shot Claude call intermittently failed with "Reached
maximum number of turns (1)" — it needs slightly more headroom than the
clarifier to finish emitting its JSON. Phase A of the planner-reliability plan:

- plan_node now invokes with max_turns=4 (allowed_tools stays []; the extra
  turns buy completion, not exploration).
- build_plan_prompt instructs the model to use no tools and return only JSON
  (a tool_use would consume the single turn before the plan is emitted).
- plan_node auto-retries the model call exactly once on a TRANSIENT failure
  (turn-cap exhaustion or an empty reply), and fails fast on DETERMINISTIC ones
  (malformed JSON, missing/blank phases) — a retry would just reproduce those.

review_loop_llm.py is intentionally GPT-4.1 cross-family (no claude_invoke), so
it gets no turn-budget change. verifier_llm.py does use claude_invoke but is P3
build/verify scope — left for a follow-up.

Tests: max_turns passthrough; retry on turn-cap and on empty; no retry on
malformed JSON; the no-tools prompt line. 1387 passed.

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
 

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