diff --git a/anton/core/backends/scratchpad_boot.py b/anton/core/backends/scratchpad_boot.py
index a949752b..d7eaf108 100644
--- a/anton/core/backends/scratchpad_boot.py
+++ b/anton/core/backends/scratchpad_boot.py
@@ -217,6 +217,8 @@ def generate_object(
                 """
                 from anton.core.llm.structured import (
                     build_structured_tool,
+                    looks_truncated,
+                    raise_unusable_tool_call,
                     unwrap_structured_response,
                 )
 
@@ -233,11 +235,24 @@ def generate_object(
                 )
 
                 if not response.tool_calls:
-                    raise ValueError("LLM did not return structured output.")
+                    # Same classification as the async path (ENG-1081).
+                    # Nothing retries here, but the message reaches the model as
+                    # a traceback, so "you ran out of budget" is actionable
+                    # where "did not return structured output" was not.
+                    raise_unusable_tool_call(
+                        response, tool_name=tool["name"], budget=max_tokens
+                    )
 
-                return unwrap_structured_response(
-                    response.tool_calls[0].input, validator_class, is_list
-                )
+                try:
+                    return unwrap_structured_response(
+                        response.tool_calls[0].input, validator_class, is_list
+                    )
+                except Exception:
+                    if looks_truncated(response, max_tokens):
+                        raise_unusable_tool_call(
+                            response, tool_name=tool["name"], budget=max_tokens
+                        )
+                    raise
 
         _scratchpad_llm_instance = _ScratchpadLLM()
 
diff --git a/anton/core/llm/client.py b/anton/core/llm/client.py
index 88c96629..30b79bf0 100644
--- a/anton/core/llm/client.py
+++ b/anton/core/llm/client.py
@@ -199,28 +199,41 @@ async def _generate_object_with(
         """
         from anton.core.llm.structured import (
             build_structured_tool,
+            looks_truncated,
+            raise_unusable_tool_call,
             unwrap_structured_response,
         )
 
         tool, validator_class, is_list = build_structured_tool(schema_class)
 
+        budget = max_tokens or self._max_tokens
+
         response = await provider.complete(
             model=model,
             system=system,
             messages=messages,
             tools=[tool],
             tool_choice={"type": "tool", "name": tool["name"]},
-            max_tokens=max_tokens or self._max_tokens,
+            max_tokens=budget,
         )
 
         if not response.tool_calls:
-            raise ValueError(
-                f"LLM did not return a tool call for forced schema {tool['name']}."
-            )
+            # Shared with the scratchpad's sync twin so both paths classify the
+            # failure identically — see `raise_unusable_tool_call` (ENG-1081).
+            raise_unusable_tool_call(response, tool_name=tool["name"], budget=budget)
 
-        return unwrap_structured_response(
-            response.tool_calls[0].input, validator_class, is_list
-        )
+        try:
+            return unwrap_structured_response(
+                response.tool_calls[0].input, validator_class, is_list
+            )
+        except Exception:
+            # The budget can also run out *inside* the tool call's JSON, so
+            # validation fails here instead. Same cause, same retry (ENG-1081).
+            if looks_truncated(response, budget):
+                raise_unusable_tool_call(
+                    response, tool_name=tool["name"], budget=budget
+                )
+            raise
 
     async def generate_object(
         self,
@@ -263,8 +276,11 @@ async def generate_object(
             input was a list annotation.
 
         Raises:
-            ValueError: If the LLM fails to produce a tool call (rare —
-                forced tool_choice usually prevents this).
+            StructuredOutputError: If the LLM fails to produce a tool call.
+                A ``ValueError`` subclass, so callers that catch
+                ``ValueError`` are unaffected. Check ``.truncated`` to tell a
+                blown ``max_tokens`` budget (retry with more room) from a
+                genuine failure (ENG-1081).
             pydantic.ValidationError: If the tool's input doesn't match
                 the schema.
 
diff --git a/anton/core/llm/provider.py b/anton/core/llm/provider.py
index 7831cd2e..719ba67b 100644
--- a/anton/core/llm/provider.py
+++ b/anton/core/llm/provider.py
@@ -312,6 +312,36 @@ class TokenLimitExceeded(Exception):
     """Raised when the LLM returns 429 due to billing/token limits."""
 
 
+class StructuredOutputError(ValueError):
+    """Raised when a forced-tool-call structured-output call yields no usable call.
+
+    ``truncated`` says whether a retry can help: ``True`` means the model spent
+    the ``max_tokens`` budget narrating before it reached the tool call (the
+    narrating aliases — ``mindshub_air``/``kimi``, ``deepseek``, ``qwen`` — do
+    this deterministically under a tight budget, ENG-1081); ``False`` means the
+    provider errored, refused, or returned nothing, and a bigger budget won't
+    fix it. See `structured.looks_truncated` for how that's decided.
+
+    Subclasses ``ValueError`` so call sites catching the documented ``ValueError``
+    from ``generate_object``/``generate_object_code`` keep working.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        truncated: bool = False,
+        output_tokens: int = 0,
+        max_tokens: int = 0,
+        stop_reason: str | None = None,
+    ):
+        super().__init__(message)
+        self.truncated = truncated
+        self.output_tokens = output_tokens
+        self.max_tokens = max_tokens
+        self.stop_reason = stop_reason
+
+
 class TransientProviderError(ConnectionError):
     """Raised when the provider fails in a way that a retry might fix.
 
diff --git a/anton/core/llm/structured.py b/anton/core/llm/structured.py
index 02372c8e..4f497eb1 100644
--- a/anton/core/llm/structured.py
+++ b/anton/core/llm/structured.py
@@ -27,7 +27,89 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, NoReturn
+
+from .provider import StructuredOutputError
+
+
+def looks_truncated(response, budget: int) -> bool:
+    """True if `response` was cut off by the `max_tokens` budget.
+
+    Token count first, because the MindsHub gateway reports
+    ``finish_reason: "stop"`` at the cap for most aliases (ENG-1082) — the
+    standard ``"length"`` check can't be relied on there. Both dialects are
+    honoured when reported: OpenAI says ``"length"``, Anthropic ``"max_tokens"``.
+    No usage information → ``False``; without evidence we don't buy a retry.
+    """
+    usage = getattr(response, "usage", None)
+    output_tokens = getattr(usage, "output_tokens", 0) or 0
+    stop_reason = getattr(response, "stop_reason", None)
+    return stop_reason in ("length", "max_tokens") or (
+        budget > 0 and output_tokens >= budget
+    )
+
+
+def raise_unusable_tool_call(response, *, tool_name: str, budget: int) -> NoReturn:
+    """Raise `StructuredOutputError` explaining *why* the tool call is unusable.
+
+    Covers both shapes of the same underlying problem:
+
+    - **No tool call at all** — the model narrated in plain ``content`` until
+      the budget ran out.
+    - **A damaged tool call** — the budget ran out *inside* the call's JSON
+      arguments, so ``safe_parse_tool_input`` salvaged a partial dict and set
+      ``parse_error``, and validation then fails. Without this, that case
+      surfaces as a bare ``ValidationError``, which reads like a schema bug and
+      never gets the retry a truncation deserves (ENG-1081).
+
+    Lives here, next to `build_structured_tool`/`unwrap_structured_response`,
+    because both structured-output paths must classify the failure the same
+    way — the async `LLMClient._generate_object_with` and the sync
+    `_ScratchpadLLM.generate_object` in the scratchpad subprocess. Keeping it
+    in one of the two callers is how they drift (ENG-1081).
+
+    A forced tool call can come back empty for two very different reasons:
+
+    - **Truncated** — the model narrated in plain ``content`` and ran out of
+      ``budget`` before reaching the call. Retrying with more room usually
+      works. Models served through MindsHub's Fireworks aliases
+      (``mindshub_air``/``kimi``, ``deepseek``) narrate before acting, so a
+      tight budget fails them.
+    - **Anything else** — the provider errored, refused, or returned nothing.
+      A bigger budget won't help.
+
+    See `looks_truncated` for how truncation is detected.
+
+    Args:
+        response: The provider's ``LLMResponse``.
+        tool_name: Name of the forced tool, for the message.
+        budget: The ``max_tokens`` the call was given.
+
+    Raises:
+        StructuredOutputError: Always. ``.truncated`` carries the verdict.
+    """
+    usage = getattr(response, "usage", None)
+    output_tokens = getattr(usage, "output_tokens", 0) or 0
+    stop_reason = getattr(response, "stop_reason", None)
+    truncated = looks_truncated(response, budget)
+    what = (
+        "returned an unusable tool call for"
+        if getattr(response, "tool_calls", None)
+        else "did not return a tool call for"
+    )
+    detail = (
+        f" (truncated: {output_tokens}/{budget} output tokens spent before the "
+        "call was complete)."
+        if truncated
+        else "."
+    )
+    raise StructuredOutputError(
+        f"LLM {what} forced schema {tool_name}{detail}",
+        truncated=truncated,
+        output_tokens=output_tokens,
+        max_tokens=budget,
+        stop_reason=stop_reason,
+    )
 
 
 def build_structured_tool(schema_class) -> tuple[dict, type, bool]:
diff --git a/anton/core/session.py b/anton/core/session.py
index 77eece9a..c6920bba 100644
--- a/anton/core/session.py
+++ b/anton/core/session.py
@@ -39,6 +39,7 @@
     StreamTaskProgress,
     StreamTextDelta,
     StreamToolResult,
+    StructuredOutputError,
     TokenLimitExceeded,
     ToolCall,
     TransientProviderError,
@@ -158,6 +159,81 @@ class _VerifierVerdict(BaseModel):
     reason: str = Field(description="One brief sentence explaining the verdict.")
 
 
+# Output budgets for the verdict call: first attempt, then the retry used when
+# it comes back truncated (ENG-1081).
+#
+# Models that narrate before acting (`mindshub_air`/`kimi`, `deepseek`, `qwen`)
+# spend the budget on prose and never reach the forced tool call. The original
+# 256 truncated them on essentially every call — 98.6% of `mindshub_air` verdicts
+# in prod returned no tool call, which the fail-safe below turned into a silent
+# "task complete".
+#
+# 2048 is sized from a measured distribution, not one sample: 16 identical calls
+# spanned 245–1654 output tokens (median ~290). That 6.7x per-call spread is also
+# why there is no "this model can't do it" latch — one truncation is a tail
+# sample, not proof about the next turn — and why the 4096 retry exists rather
+# than a single bigger budget. 1024 was measurably too small. Nothing pays for
+# headroom it doesn't use; first-party models answer in 43–115 tokens either way.
+_VERIFIER_TOKEN_BUDGETS = (2048, 4096)
+
+# Appended to the verifier system prompt. Shortens the preamble on narrating
+# models but is not sufficient alone (0/3 at 256 with it), so it pairs with the
+# budgets above. Tool name comes off the schema class so it can't go stale.
+_VERIFIER_NO_PREAMBLE = (
+    f"Call the {_VerifierVerdict.__name__} tool immediately as your first "
+    "action. Do not think out loud, restate the conversation, or explain your "
+    "reasoning before calling it — put your one-sentence justification in the "
+    "tool's `reason` field."
+)
+
+
+def _safe_error_detail(exc: BaseException) -> str:
+    """Describe an exception for logs without copying model or user content.
+
+    Neither ``str(exc)`` nor ``repr(exc)`` is safe here: a Pydantic
+    ``ValidationError`` embeds the rejected ``input_value`` — for the verifier
+    that's model-generated text derived from the user's conversation — and
+    provider exceptions can carry response bodies. Emits the exception type,
+    plus for validation errors the field locations and error codes only, which
+    is what actually identifies the failure.
+    """
+    # This runs *inside* an `except` handler, so it must not raise: an exception
+    # escaping here would turn a gracefully-handled verifier failure into a dead
+    # turn. Everything below is therefore wrapped, including the attribute reads
+    # (a custom exception can expose `status_code`/`errors` as a property that
+    # raises).
+    try:
+        name = type(exc).__name__
+    except Exception:  # pragma: no cover — defensive
+        return "unavailable"
+    try:
+        status = getattr(exc, "status_code", None)
+        if status is not None:
+            return f"{name}(status={status})"
+        errors = getattr(exc, "errors", None)
+        if callable(errors):
+            try:
+                details = errors(
+                    include_input=False, include_url=False, include_context=False
+                )
+            except TypeError:  # pydantic v1 has no include_* kwargs
+                details = errors()
+            # Only `loc` and `type` are ever read — `msg`/`input`/`ctx` can
+            # quote the rejected value.
+            fields = ",".join(
+                f"{'.'.join(str(p) for p in e.get('loc') or ())}:{e.get('type', '?')}"
+                for e in details
+            )
+            if fields:
+                return f"{name}({fields})"
+    except Exception:
+        # Anything odd about this exception object — a property that raises, an
+        # `errors()` that misbehaves — degrades to the type name, never a crash
+        # and never the message.
+        pass
+    return name
+
+
 def _render_tool_result_content(content, cap: int) -> str:
     """Render a tool_result's content as bounded plain text.
 
@@ -2656,25 +2732,64 @@ async def _stream_and_handle_tools(
                 "You are a task-completion verifier. Decide whether the user's "
                 "request is complete, the assistant is waiting on the user, the work "
                 "is unfinished, or the assistant is blocked. Follow the status "
-                "definitions exactly."
+                "definitions exactly.\n\n"
+                # Models that narrate before acting spend the whole budget on
+                # prose and never reach the tool call (ENG-1081). Asking for the
+                # call first shortens the preamble; it does not eliminate it,
+                # which is why the budget below is generous as well.
+                + _VERIFIER_NO_PREAMBLE
             )
-            try:
-                verdict = await self._llm.generate_object_code(
-                    _VerifierVerdict,
-                    system=verifier_system,
-                    messages=verify_messages,
-                    max_tokens=256,
-                )
+            verdict = None
+            for attempt, budget in enumerate(_VERIFIER_TOKEN_BUDGETS):
+                try:
+                    verdict = await self._llm.generate_object_code(
+                        _VerifierVerdict,
+                        system=verifier_system,
+                        messages=verify_messages,
+                        max_tokens=budget,
+                    )
+                    break
+                except StructuredOutputError as exc:
+                    # A truncated verdict is a budget problem, not a verdict: the
+                    # model narrated past `budget` before it reached the tool call
+                    # (ENG-1081). Retry with more room. Any other structured-output
+                    # failure won't be fixed by a bigger budget, so don't pay for it.
+                    retrying = exc.truncated and attempt + 1 < len(
+                        _VERIFIER_TOKEN_BUDGETS
+                    )
+                    _verifier_log.info(
+                        "completion-verifier verdict=%s budget=%d output_tokens=%d "
+                        "stop_reason=%s retrying=%s",
+                        "TRUNCATED" if exc.truncated else "NO_TOOL_CALL",
+                        budget, exc.output_tokens, exc.stop_reason, retrying,
+                    )
+                    if not retrying:
+                        break
+                except Exception as exc:
+                    # Enough to tell the failure modes apart — four used to
+                    # collapse into one "verifier unavailable" line — but never
+                    # the exception message, which can carry conversation
+                    # content (ENG-1081).
+                    _verifier_log.info(
+                        "completion-verifier verdict=ERROR budget=%d error=%s",
+                        budget, _safe_error_detail(exc),
+                    )
+                    break
+
+            if verdict is not None:
                 status = verdict.status
                 reason = verdict.reason.strip()
-            except Exception:
+            else:
                 # Verifier failed — fail safe by treating the turn as done rather
                 # than forcing a continuation the user never asked for.
                 status, reason = "COMPLETE", "verifier unavailable"
 
             _verifier_log.info(
-                "completion-verifier verdict=%s continuation=%d/%d tool_rounds=%d reason=%s",
-                status, continuation, self._max_continuations, tool_round, reason,
+                # No `reason` — it's the model's free-text justification, derived
+                # from the user's conversation, and this is an ordinary app log.
+                # It stays in the Langfuse trace, where that content belongs.
+                "completion-verifier verdict=%s continuation=%d/%d tool_rounds=%d",
+                status, continuation, self._max_continuations, tool_round,
             )
 
             if status in ("COMPLETE", "WAITING"):
diff --git a/tests/e2e/scenarios/test_loop_safety.py b/tests/e2e/scenarios/test_loop_safety.py
index 3f83dcfe..642b5738 100644
--- a/tests/e2e/scenarios/test_loop_safety.py
+++ b/tests/e2e/scenarios/test_loop_safety.py
@@ -50,6 +50,45 @@ def test_continuation_limit_respected(cfg, stub, tmp_path):
     ), f"Budget-exhausted message not found. Request count: {stub.request_count}"
 
 
+@pytest.mark.stub_only
+def test_truncated_verdict_is_retried_not_silently_dropped(cfg, stub, tmp_path):
+    # ENG-1081: the verdict call is a forced tool call, and models that narrate
+    # before acting (mindshub_air/kimi, deepseek) spend the whole budget on prose
+    # and never reach it. That used to raise, get swallowed as a fake COMPLETE,
+    # and end the turn with no message at all. The session must retry with a
+    # bigger budget and then honour the real verdict.
+    stub.queue_tool_call("scratchpad", {"action": "exec", "name": "c", "code": "print(1)"})
+    stub.queue_text("Ran the script. TURN_TEXT")
+    stub.queue_verification_truncated()          # first budget: narrated to the cap
+    stub.queue_verification_incomplete("the summary table is still missing")
+    stub.queue_text("Finished the summary table. RETRY_VERDICT_HONOURED")
+    stub.queue_verification_ok()
+    result = run_anton(["--folder", str(tmp_path)], ["build me a summary", "exit"],
+                       env=base_env(stub), timeout=cfg.timeout(60))
+
+    assert_exit_ok(result)
+    assert_not_output(result, "Traceback (most recent call last)")
+    # The retried verdict (INCOMPLETE) drove a continuation, so the turn kept
+    # working instead of stopping silently on a fabricated COMPLETE.
+    assert_output(result, "RETRY_VERDICT_HONOURED")
+    verdict_calls = [
+        r for r in stub.requests
+        if "task-completion verifier" in json.dumps(r.get("messages", []))
+        or "task-completion verifier" in str(r.get("system", ""))
+    ]
+    assert len(verdict_calls) >= 2, (
+        f"expected a retried verdict call, saw {len(verdict_calls)}. "
+        f"Request count: {stub.request_count}"
+    )
+    # The OpenAI provider sends the budget as `max_completion_tokens`.
+    budgets = [
+        r.get("max_completion_tokens") or r.get("max_tokens") for r in verdict_calls[:2]
+    ]
+    assert budgets[0] < budgets[1], (
+        f"the retry must ask for more room than the first attempt, got {budgets}"
+    )
+
+
 @pytest.mark.stub_only
 def test_waiting_verdict_stops_without_continuation(cfg, stub, tmp_path):
     # ENG-716: a tool-using turn that ends by asking the user a question must
diff --git a/tests/e2e/stub_server.py b/tests/e2e/stub_server.py
index c8c20125..b97bfeff 100644
--- a/tests/e2e/stub_server.py
+++ b/tests/e2e/stub_server.py
@@ -29,6 +29,11 @@ class _Response:
     tool_calls: list[dict] = field(default_factory=list)
     # None = honour request's `stream` flag; True/False = override
     force_streaming: bool | None = None
+    # Reported as `usage.completion_tokens`. Set it equal to the request's
+    # `max_tokens` to emulate a truncated response — which is how the real
+    # MindsHub gateway presents one, since it still says
+    # `finish_reason: "stop"` at the cap (ENG-1082).
+    output_tokens: int = 10
 
 
 class StubServer:
@@ -84,6 +89,22 @@ def _queue_verdict(self, status: str, reason: str) -> "StubServer":
         ))
         return self
 
+    def queue_verification_truncated(self, output_tokens: int = 2048) -> "StubServer":
+        """Queue a verdict call that narrated instead of calling the tool and ran
+        out of budget doing it (ENG-1081).
+
+        Text, no tool call, and `completion_tokens` equal to the verifier's first
+        budget — exactly what `mindshub_air`/`kimi`/`deepseek` return. The
+        session should retry with the larger budget rather than treating this as
+        a verdict.
+        """
+        self._queue.put(_Response(
+            content="Let me analyze this conversation carefully. The user asked for",
+            force_streaming=False,
+            output_tokens=output_tokens,
+        ))
+        return self
+
     def queue_verification_ok(self) -> "StubServer":
         """Queue a COMPLETE verifier verdict."""
         return self._queue_verdict("COMPLETE", "task is done.")
@@ -243,7 +264,11 @@ def _send_json(handler: BaseHTTPRequestHandler, resp: _Response) -> None:
             "message": message,
             "finish_reason": finish_reason,
         }],
-        "usage": {"prompt_tokens": 10, "completion_tokens": 10, "total_tokens": 20},
+        "usage": {
+            "prompt_tokens": 10,
+            "completion_tokens": resp.output_tokens,
+            "total_tokens": 10 + resp.output_tokens,
+        },
     }
     body = json.dumps(data).encode()
     handler.send_response(200)
diff --git a/tests/test_verifier_truncation.py b/tests/test_verifier_truncation.py
new file mode 100644
index 00000000..748e6c52
--- /dev/null
+++ b/tests/test_verifier_truncation.py
@@ -0,0 +1,562 @@
+"""Completion-verifier truncation handling (ENG-1081).
+
+The verdict call is a forced tool call. Models that narrate before acting
+(MindsHub's Fireworks aliases — `mindshub_air`/`kimi`, `deepseek`) spend the
+output budget on plain prose and never reach the call, so a tight `max_tokens`
+fails them deterministically: 98.6% of `mindshub_air` verdict calls in prod
+returned no tool call, and the fail-safe turned each one into a silent
+"task complete" with no message to the user.
+
+Two behaviours are covered here:
+
+1. `_generate_object_with` reports *why* there was no tool call, distinguishing a
+   blown budget (retryable) from a genuine failure — by token count, because the
+   MindsHub gateway reports `finish_reason: "stop"` at the cap (ENG-1082).
+2. The verifier retries a truncated verdict once with a bigger budget, and does
+   NOT spend a retry on a failure a bigger budget can't fix.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from tests.conftest import make_mock_llm
+
+from anton.core.llm.client import LLMClient
+from anton.core.llm.provider import (
+    LLMResponse,
+    StreamComplete,
+    StructuredOutputError,
+    ToolCall,
+    Usage,
+)
+from anton.core.session import (
+    _VERIFIER_TOKEN_BUDGETS,
+    ChatSession,
+    ChatSessionConfig,
+    _VerifierVerdict,
+)
+
+
+@pytest.fixture()
+def workspace():
+    # Keep scratchpad venvs inside the repo workspace (pytest runs sandboxed and
+    # can't write to the real home directory).
+    base = Path(__file__).resolve().parents[1] / ".pytest-workspace"
+    base.mkdir(parents=True, exist_ok=True)
+    return MagicMock(base=base)
+
+
+def _text_response(text: str, output_tokens: int = 20, stop_reason: str = "end_turn") -> LLMResponse:
+    return LLMResponse(
+        content=text,
+        tool_calls=[],
+        usage=Usage(input_tokens=10, output_tokens=output_tokens),
+        stop_reason=stop_reason,
+    )
+
+
+def _scratchpad_response(text: str, code: str = "print(1)") -> LLMResponse:
+    return LLMResponse(
+        content=text,
+        tool_calls=[ToolCall(
+            id="tc_1", name="scratchpad",
+            input={"action": "exec", "name": "main", "code": code},
+        )],
+        usage=Usage(input_tokens=10, output_tokens=20),
+        stop_reason="tool_use",
+    )
+
+
+class _FakeAsyncIter:
+    def __init__(self, items):
+        self._items = items
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if not self._items:
+            raise StopAsyncIteration
+        return self._items.pop(0)
+
+
+# --------------------------------------------------------------------------
+# 1. The client reports *why* the tool call is missing.
+# --------------------------------------------------------------------------
+
+
+def _client_with_response(response: LLMResponse) -> LLMClient:
+    provider = MagicMock()
+    provider.complete = AsyncMock(return_value=response)
+    return LLMClient(
+        planning_provider=provider,
+        planning_model="planner",
+        coding_provider=provider,
+        coding_model="coder",
+    )
+
+
+async def test_no_tool_call_at_the_cap_is_reported_as_truncated():
+    """Prose that spends the whole budget == truncation, even though the
+    gateway calls it `finish_reason: "stop"` (ENG-1082)."""
+    llm = _client_with_response(
+        _text_response("Let me analyze this conversation carefully. The user...",
+                       output_tokens=256, stop_reason="stop")
+    )
+
+    with pytest.raises(StructuredOutputError) as exc_info:
+        await llm.generate_object_code(
+            _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}],
+            max_tokens=256,
+        )
+
+    exc = exc_info.value
+    assert exc.truncated is True
+    assert exc.output_tokens == 256
+    assert exc.max_tokens == 256
+    # Callers that only know the documented ValueError still catch it.
+    assert isinstance(exc, ValueError)
+
+
+@pytest.mark.parametrize("stop_reason", ["length", "max_tokens"])
+async def test_both_provider_dialects_for_truncation(stop_reason):
+    """The gateway/OpenAI dialect says "length"; AnthropicProvider passes
+    Anthropic's own "max_tokens" through raw. Both mean truncated."""
+    llm = _client_with_response(_text_response("narrating…", output_tokens=100,
+                                               stop_reason=stop_reason))
+
+    with pytest.raises(StructuredOutputError) as exc_info:
+        await llm.generate_object_code(
+            _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}],
+            max_tokens=2048,
+        )
+
+    assert exc_info.value.truncated is True
+
+
+async def test_stop_reason_length_is_honoured_below_the_cap():
+    """Gemini reports truncation honestly and can return almost nothing —
+    trust `stop_reason` too, not only the token count."""
+    llm = _client_with_response(_text_response("", output_tokens=9, stop_reason="length"))
+
+    with pytest.raises(StructuredOutputError) as exc_info:
+        await llm.generate_object_code(
+            _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}],
+            max_tokens=256,
+        )
+
+    assert exc_info.value.truncated is True
+
+
+async def test_short_empty_response_is_not_truncated():
+    """A provider that returns nothing well inside the budget is a genuine
+    failure — a bigger budget won't fix it, so it must not be retried."""
+    llm = _client_with_response(_text_response("", output_tokens=5, stop_reason="stop"))
+
+    with pytest.raises(StructuredOutputError) as exc_info:
+        await llm.generate_object_code(
+            _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}],
+            max_tokens=256,
+        )
+
+    assert exc_info.value.truncated is False
+
+
+def _damaged_tool_call_response(output_tokens: int) -> LLMResponse:
+    """A forced tool call that ran out of budget *inside* its JSON arguments.
+
+    `safe_parse_tool_input` repairs what it can and sets `parse_error`, so the
+    call arrives non-empty but incomplete — here missing the required `reason`.
+    """
+    return LLMResponse(
+        content="",
+        tool_calls=[ToolCall(id="tc_v", name="_VerifierVerdict",
+                             input={"status": "WAITING"},
+                             parse_error="Unterminated string starting at: line 1")],
+        usage=Usage(input_tokens=100, output_tokens=output_tokens),
+        stop_reason="stop",
+    )
+
+
+async def test_tool_call_truncated_mid_arguments_is_retryable():
+    """The budget can run out *inside* the tool call, not only before it.
+
+    That arrives as a non-empty but incomplete tool call, so the missing-call
+    check doesn't see it and validation fails instead — which without this
+    would read as a schema bug, skip the retry, and land right back on the
+    silent-stop fail-safe this whole change exists to remove.
+    """
+    llm = _client_with_response(_damaged_tool_call_response(output_tokens=2048))
+
+    with pytest.raises(StructuredOutputError) as exc_info:
+        await llm.generate_object_code(
+            _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}],
+            max_tokens=2048,
+        )
+
+    assert exc_info.value.truncated is True, "must be retryable, not a schema error"
+    assert "unusable tool call" in str(exc_info.value)
+
+
+async def test_schema_mismatch_under_budget_is_not_disguised_as_truncation():
+    """The mirror case: a malformed tool call with budget to spare is a real
+    schema failure. It must keep propagating as a validation error rather than
+    being relabelled truncation and buying a retry that cannot help."""
+    llm = _client_with_response(_damaged_tool_call_response(output_tokens=60))
+
+    with pytest.raises(ValueError) as exc_info:
+        await llm.generate_object_code(
+            _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}],
+            max_tokens=2048,
+        )
+
+    assert not isinstance(exc_info.value, StructuredOutputError), (
+        "a genuine schema mismatch must stay a validation error"
+    )
+
+
+def test_shared_classifier_is_used_by_both_structured_paths():
+    """The async client and the scratchpad's sync twin must classify identically.
+
+    `raise_unusable_tool_call` is the single implementation both call — the whole
+    point of `structured.py`. Guarded here because the previous version of this
+    fix put the logic in `client.py`, which silently left the sync path
+    (exposed to model-written scratchpad code, with a caller-chosen budget)
+    raising a blind ValueError.
+    """
+    import ast
+
+    from anton.core.llm import structured
+
+    assert hasattr(structured, "raise_unusable_tool_call")
+    # Parse the source rather than importing it — `scratchpad_boot` is a
+    # subprocess bootstrap and reads stdin at import time. AST rather than a
+    # substring search, so a passing mention in a comment or a stale import
+    # can't satisfy it: the call must be inside `generate_object` itself.
+    boot_src = (
+        Path(__file__).resolve().parents[1]
+        / "anton" / "core" / "backends" / "scratchpad_boot.py"
+    ).read_text()
+    tree = ast.parse(boot_src)
+    generate_object = next(
+        (n for n in ast.walk(tree)
+         if isinstance(n, ast.FunctionDef) and n.name == "generate_object"),
+        None,
+    )
+    assert generate_object is not None, "sync generate_object not found"
+    called = {
+        n.func.id for n in ast.walk(generate_object)
+        if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
+    }
+    assert "raise_unusable_tool_call" in called, (
+        "the sync scratchpad path must call the shared classifier, not raise its own"
+    )
+    assert "LLM did not return structured output." not in boot_src, (
+        "the old blind ValueError should be gone from the sync path"
+    )
+
+
+def test_classifier_tolerates_a_response_without_usage():
+    """Defensive: a provider response missing `usage` must not blow up the
+    classifier — it just can't prove truncation, which is the safe direction
+    (no retry bought without evidence)."""
+    from anton.core.llm.structured import raise_unusable_tool_call
+
+    class _Bare:
+        content = "some prose"
+
+    with pytest.raises(StructuredOutputError) as exc_info:
+        raise_unusable_tool_call(_Bare(), tool_name="_VerifierVerdict", budget=2048)
+
+    assert exc_info.value.truncated is False
+    assert exc_info.value.output_tokens == 0
+
+
+# --------------------------------------------------------------------------
+# 2. The verifier retries a truncated verdict, once, with more room.
+# --------------------------------------------------------------------------
+
+
+class _ToolThenText:
+    """`plan_stream` fake: one tool round per turn, then plain text.
+
+    Keyed off "have I already used the tool this turn?" rather than a call
+    counter — a counter with `% 2` would silently desync if a turn ever made a
+    third `plan_stream` call (an extra tool round, an internal recovery retry),
+    and the test would then be asserting something other than it claims.
+    Call `next_turn()` between turns.
+    """
+
+    def __init__(self):
+        self.tool_used = False
+
+    def next_turn(self) -> None:
+        self.tool_used = False
+
+    def __call__(self, **kwargs):
+        if not self.tool_used:
+            self.tool_used = True
+            return _FakeAsyncIter([StreamComplete(response=_scratchpad_response("Running."))])
+        return _FakeAsyncIter([StreamComplete(response=_text_response("Done."))])
+
+
+def _session_that_uses_a_tool(mock_llm, workspace) -> ChatSession:
+    """Session whose turn uses one tool, so the completion verifier runs."""
+    mock_llm.plan_stream = _ToolThenText()
+    return ChatSession(ChatSessionConfig(llm_client=mock_llm, workspace=workspace))
+
+
+async def test_truncated_verdict_is_retried_with_a_bigger_budget(workspace):
+    """The narrating-model case: first budget truncates, the retry succeeds, and
+    the retried verdict is the one that counts — no silent 'COMPLETE'."""
+    budgets: list[int] = []
+
+    async def fake_verdict(_schema, *, system, messages, max_tokens):
+        budgets.append(max_tokens)
+        if len(budgets) == 1:
+            raise StructuredOutputError(
+                "no tool call", truncated=True, output_tokens=max_tokens,
+                max_tokens=max_tokens, stop_reason="stop",
+            )
+        return _VerifierVerdict(status="WAITING", reason="asked the user a question")
+
+    mock_llm = make_mock_llm()
+    mock_llm.generate_object_code = AsyncMock(side_effect=fake_verdict)
+
+    session = _session_that_uses_a_tool(mock_llm, workspace)
+    try:
+        async for _ in session.turn_stream("build me a dashboard"):
+            pass
+    finally:
+        await session.close()
+
+    assert budgets == list(_VERIFIER_TOKEN_BUDGETS[:2]), (
+        "a truncated verdict must be retried once, with the larger budget"
+    )
+    # The verdict came from the retry (WAITING → a valid stop), so no
+    # "Continue working" continuation was injected.
+    assert not any(
+        "SYSTEM: Task verification determined this task is not yet complete"
+        in str(m.get("content", ""))
+        for m in session.history
+    )
+
+
+async def test_non_truncated_failure_is_not_retried(workspace):
+    """A failure a bigger budget can't fix costs exactly one call, then falls
+    through to the fail-safe."""
+    calls: list[int] = []
+
+    async def fake_verdict(_schema, *, system, messages, max_tokens):
+        calls.append(max_tokens)
+        raise StructuredOutputError(
+            "no tool call", truncated=False, output_tokens=3,
+            max_tokens=max_tokens, stop_reason="stop",
+        )
+
+    mock_llm = make_mock_llm()
+    mock_llm.generate_object_code = AsyncMock(side_effect=fake_verdict)
+
+    session = _session_that_uses_a_tool(mock_llm, workspace)
+    try:
+        async for _ in session.turn_stream("build me a dashboard"):
+            pass
+    finally:
+        await session.close()
+
+    assert calls == [_VERIFIER_TOKEN_BUDGETS[0]], "must not pay for a hopeless retry"
+
+
+async def test_attempts_are_bounded_and_repeat_each_turn(workspace):
+    """Truncation retries are bounded by the budget list — and are re-tried on a
+    later turn rather than latched off.
+
+    Output length varies ~6.7x per call for an identical request, so one
+    truncation is a tail sample, not proof the model can never fit a verdict.
+    Latching the retry off on that evidence would bring silent stops back.
+    """
+    budgets: list[int] = []
+
+    async def fake_verdict(_schema, *, system, messages, max_tokens):
+        budgets.append(max_tokens)
+        raise StructuredOutputError(
+            "no tool call", truncated=True, output_tokens=max_tokens,
+            max_tokens=max_tokens, stop_reason="stop",
+        )
+
+    mock_llm = make_mock_llm()
+    mock_llm.generate_object_code = AsyncMock(side_effect=fake_verdict)
+
+    session = _session_that_uses_a_tool(mock_llm, workspace)
+    try:
+        async for _ in session.turn_stream("first turn"):
+            pass
+        assert budgets == list(_VERIFIER_TOKEN_BUDGETS), "bounded by the budget list"
+
+        budgets.clear()
+        mock_llm.plan_stream.next_turn()
+        async for _ in session.turn_stream("second turn"):
+            pass
+    finally:
+        await session.close()
+
+    assert budgets == list(_VERIFIER_TOKEN_BUDGETS), (
+        "a later turn gets a fresh chance — the retry is not latched off"
+    )
+
+
+@pytest.mark.parametrize("first_failure", ["no_tool_call", "damaged_tool_call"])
+async def test_truncation_retry_through_the_real_client(workspace, first_failure):
+    """End-to-end: both shapes of a truncated verdict must flow through the real
+    `LLMClient` detection into a session-level retry — prose-at-the-cap with no
+    call at all, and a call cut off inside its own arguments.
+
+    The other session tests raise `StructuredOutputError` by hand, so they would
+    pass even if the detection in `raise_unusable_tool_call` were wrong. This one
+    wires a fake *provider* through the real client, so provider → client →
+    session is the actual code path (ENG-747: don't hand-build error fixtures).
+    """
+    calls: list[int] = []
+
+    async def fake_complete(*, model, system, messages, tools, tool_choice, max_tokens):
+        calls.append(max_tokens)
+        if len(calls) == 1:
+            if first_failure == "damaged_tool_call":
+                # Budget ran out *inside* the call's JSON arguments.
+                return _damaged_tool_call_response(output_tokens=max_tokens)
+            # Narrated right up to the ceiling, never reached the tool call —
+            # and the gateway calls that `finish_reason: "stop"` (ENG-1082).
+            return _text_response("Let me analyze this conversation carefully...",
+                                  output_tokens=max_tokens, stop_reason="stop")
+        return LLMResponse(
+            content="",
+            tool_calls=[ToolCall(id="tc_v", name="_VerifierVerdict",
+                                 input={"status": "WAITING", "reason": "asked the user"})],
+            usage=Usage(input_tokens=100, output_tokens=60),
+            stop_reason="tool_calls",
+        )
+
+    provider = MagicMock()
+    provider.complete = AsyncMock(side_effect=fake_complete)
+    real_client = LLMClient(
+        planning_provider=provider, planning_model="planner",
+        coding_provider=provider, coding_model="coder",
+    )
+
+    # Only the verdict call goes through the real client; the turn itself still
+    # uses the mock so the test stays a unit test.
+    mock_llm = make_mock_llm()
+    mock_llm.generate_object_code = real_client.generate_object_code
+
+    session = _session_that_uses_a_tool(mock_llm, workspace)
+    try:
+        async for _ in session.turn_stream("build me a dashboard"):
+            pass
+    finally:
+        await session.close()
+
+    assert calls == list(_VERIFIER_TOKEN_BUDGETS), (
+        "the real client must classify prose-at-the-cap as truncated and the "
+        "session must retry it with the larger budget"
+    )
+    # The retried verdict (WAITING) stands: no forced continuation.
+    assert not any(
+        "SYSTEM: Task verification determined this task is not yet complete"
+        in str(m.get("content", ""))
+        for m in session.history
+    )
+
+
+async def test_verifier_prompt_forbids_preamble(workspace):
+    """The no-preamble instruction reaches the model. It is not sufficient on its
+    own (0/3 at 256 with it), but it shortens the preamble enough to matter."""
+    seen: dict = {}
+
+    async def fake_verdict(_schema, *, system, messages, max_tokens):
+        seen["system"] = system
+        return _VerifierVerdict(status="COMPLETE", reason="done")
+
+    mock_llm = make_mock_llm()
+    mock_llm.generate_object_code = AsyncMock(side_effect=fake_verdict)
+
+    session = _session_that_uses_a_tool(mock_llm, workspace)
+    try:
+        async for _ in session.turn_stream("build me a dashboard"):
+            pass
+    finally:
+        await session.close()
+
+    assert "immediately as your first action" in seen["system"]
+    assert "Do not think out loud" in seen["system"]
+
+
+# --------------------------------------------------------------------------
+# 3. Verifier logs must not carry conversation content (review finding).
+# --------------------------------------------------------------------------
+
+
+def test_error_detail_never_leaks_the_rejected_value():
+    """`_safe_error_detail` must identify the failure without quoting content.
+
+    A pydantic ValidationError's message embeds the rejected `input_value` — for
+    the verifier that's model-generated text derived from the user's
+    conversation — so `str(exc)` cannot go into ordinary application logs.
+    """
+    import pydantic
+
+    from anton.core.session import _safe_error_detail
+
+    secret = "the user's bank balance is 12345"
+    try:
+        _VerifierVerdict.model_validate({"status": secret, "reason": secret})
+    except pydantic.ValidationError as exc:
+        detail = _safe_error_detail(exc)
+    else:
+        raise AssertionError("expected a ValidationError")
+
+    assert secret not in detail, f"rejected value leaked into the log detail: {detail!r}"
+    assert "12345" not in detail
+    # Still useful: names the exception type and which field failed.
+    assert "ValidationError" in detail
+    assert "status" in detail
+
+
+def test_error_detail_of_a_provider_error_is_type_and_status_only():
+    from anton.core.session import _safe_error_detail
+
+    class _FakeAPIError(Exception):
+        status_code = 503
+
+    detail = _safe_error_detail(_FakeAPIError("upstream said: <full response body>"))
+    assert detail == "_FakeAPIError(status=503)"
+    assert "response body" not in detail
+
+
+def test_error_detail_falls_back_to_the_type_name():
+    from anton.core.session import _safe_error_detail
+
+    detail = _safe_error_detail(RuntimeError("provider hiccup with conversation text"))
+    assert detail == "RuntimeError"
+
+
+def test_error_detail_never_raises_from_inside_an_except_handler():
+    """It runs inside `except`, so raising would turn a handled verifier failure
+    into a dead turn. Hostile exceptions must still produce a string."""
+    from anton.core.session import _safe_error_detail
+
+    class _Hostile(Exception):
+        @property
+        def status_code(self):
+            raise RuntimeError("boom")
+
+    class _HostileErrors(Exception):
+        def errors(self, **kwargs):
+            raise RuntimeError("boom")
+
+    # Degrades to the exception type — never raises, never the message.
+    assert _safe_error_detail(_Hostile()) == "_Hostile"
+    assert _safe_error_detail(_HostileErrors()) == "_HostileErrors"