fix: local delivery runner ships a typed reason for every failure

The `--run-delivery` catch-all in tools/bot_mode_dm.py printed a structured
`{error, reason}` only for target_busy; any other exception went to stderr as
prose, so the sender's completion notification had no reason to branch on
(auth failure vs. transient) for the local lane while the relay lane had one.

The vocabulary-guarded helper the relay lane introduced moves beside the
vocabulary it guards (tools/bot_failure_reasons.delivery_failure_reason) and
both lanes call it: an exception's own `reason` is trusted only when it is
`target_busy` or in ALL_REASONS, otherwise the text is classified. The runner
now prints `{error, reason}` JSON on stdout for every exception (exit 1 kept).

Test: a classifiable, an unclassifiable and an out-of-vocabulary-reason
exception in --run-delivery each yield JSON with a vocabulary reason.

tests/tools/test_bot_turn_lock.py

@@ -393,3 +393,31 @@ def test_every_relay_refusal_carries_its_typed_reason(tmp_path, monkeypatch, fai
 
     assert out["error"]["code"] == code
     assert out["error"]["data"]["reason"] == reason
+
+
+@pytest.mark.parametrize(
+    ("failure", "reason"),
+    [
+        (RuntimeError("Error code: 401 - invalid api key"), "provider_auth_or_access"),
+        (RuntimeError("something nobody has a rule for"), "unknown"),
+        (_WithReason("CERTIFICATE_VERIFY_FAILED", "ssl handshake failed"), "unknown"),
+    ],
+    ids=["classifiable-failure", "unclassifiable-failure", "reason-outside-the-vocabulary"],
+)
+def test_delivery_main_reports_every_failure_as_typed_json(tmp_path, monkeypatch, capsys, failure, reason):
+    """The local lane's runner stdout IS the sender's completion notification. A failure other
+    than target_busy used to reach the sender as stderr prose with no reason, so it could not
+    tell an auth failure from a transient one; it now rides the same vocabulary as the relay."""
+    dm = tmp_path / "dm.txt"
+    dm.write_text("hi", encoding="utf-8")
+
+    def _raise(*args, **kwargs):
+        raise failure
+
+    monkeypatch.setattr(bot_mode_dm, "_run_delivery", _raise)
+
+    rc = bot_mode_dm._delivery_main(["--run-delivery", "query-file", str(dm), "hermes", "-p", "ops", "chat"])
+
+    assert rc == 1
+    payload = json.loads(capsys.readouterr().out.strip())
+    assert payload == {"error": str(failure), "reason": reason}

tools/bot_failure_reasons.py

@@ -96,3 +96,19 @@ def classify_agent_error(text: str) -> str:
             if pattern.search(raw):
                 return code
     return UNKNOWN
+
+
+def delivery_failure_reason(error: BaseException) -> str:
+    """The typed reason for a delivery refusal, kept inside the documented vocabulary.
+
+    An exception's own ``reason`` is trusted only when it names a real code: ``TurnBusyError``
+    carries ``target_busy``, but ``reason`` is also a stdlib attribute on ``ssl.SSLError`` and
+    ``urllib.error.URLError``, and forwarding one of those would put free text where consumers
+    expect a closed set. Anything else is classified like every other failure. Shared by the
+    relay lane (``bot_relay.deliver``) and the local runner (``bot_mode_dm --run-delivery``).
+    """
+    # 'target_busy' extends the structured refusal enum and predates ALL_REASONS.
+    supplied = str(getattr(error, "reason", "") or "").strip()
+    if supplied == "target_busy" or supplied in ALL_REASONS:
+        return supplied
+    return classify_agent_error(str(error))

tools/bot_mode_dm.py

@@ -690,13 +690,12 @@ def _delivery_main(args: list[str]) -> int:
             profile_home, argv = Path(argv[1]), argv[2:]
         return _run_delivery(argv, rest[1], stdin_file=rest[0] == "stdin", profile_home=profile_home, author=author)
     except Exception as exc:
-        # 'target_busy': the queued delivery gave up after its bounded wait — surface the
-        # structured payload on stdout so the completion notification carries it back.
-        if getattr(exc, "reason", "") == "target_busy":
-            # See #93091.
-            print(json.dumps({"error": str(exc), "reason": "target_busy"}))
-        else:
-            print(f"message_agent delivery failed: {type(exc).__name__}: {exc}", file=sys.stderr)
+        # Every refusal ships a typed reason on stdout so the completion notification carries it
+        # back to the sender (#93091): 'target_busy' from the queue's bounded wait, otherwise the
+        # same vocabulary-guarded classification the relay lane applies.
+        from tools.bot_failure_reasons import delivery_failure_reason
+
+        print(json.dumps({"error": str(exc), "reason": delivery_failure_reason(exc)}))
         return 1
 
 

tui_gateway/methods_bot_relay.py

@@ -11,6 +11,7 @@ import subprocess
 from pathlib import Path
 
 # Defined beside the sender-side waiter budget so the two Python sides cannot drift (#93911).
+from tools.bot_failure_reasons import delivery_failure_reason
 from tools.bot_relay import TURN_ATTEMPT_TIMEOUT_SECONDS
 
 from .method_ctx import HandlerRegistry
@@ -34,23 +35,6 @@ def _run_delivery(profile: str, tmp: str, env: dict | None = None) -> subprocess
         errors="replace", timeout=TURN_ATTEMPT_TIMEOUT_SECONDS, env=env)
 
 
-def _delivery_failure_reason(error: BaseException) -> str:
-    """The typed reason for a refusal, kept inside the documented vocabulary.
-
-    An exception's own ``reason`` is trusted only when it names a real code: ``TurnBusyError``
-    carries ``target_busy``, but ``reason`` is also a stdlib attribute on ``ssl.SSLError`` and
-    ``urllib.error.URLError``, and forwarding one of those would put free text where consumers
-    expect a closed set. Anything else is classified like every other failure.
-    """
-    from tools.bot_failure_reasons import ALL_REASONS, classify_agent_error
-
-    # 'target_busy' extends the structured refusal enum and predates ALL_REASONS.
-    supplied = str(getattr(error, "reason", "") or "").strip()
-    if supplied == "target_busy" or supplied in ALL_REASONS:
-        return supplied
-    return classify_agent_error(str(error))
-
-
 @method("bot_relay.roster.sync")
 def _(rid, params: dict, _root=_relay_root) -> dict:
     """Replace this gateway's view of agents on OTHER connections → ``{count}`` accepted rows
@@ -75,7 +59,7 @@ def _(rid, params: dict, _root=_relay_root) -> dict:
 
 @method("bot_relay.deliver")
 def _(rid, params: dict, _root=_relay_root, _run=_run_delivery,
-      _failure_reason=_delivery_failure_reason) -> dict:
+      _failure_reason=delivery_failure_reason) -> dict:
     """Deliver a relayed DM (``profile``, attribution-prefixed ``message``) into a Bot Chat ON THIS
     GATEWAY via the one-turn ``hermes -p <profile> chat -c "Bot Chat"`` transport local DMs use →
     ``{reply}``. Blocking by design (Desktop relay worker; the RPC pool keeps it off the reader)."""