fix(core): legalize malformed tool-call arguments before the wire

A tool call whose `arguments` is not a JSON-object string (an unterminated
string from a non-`length` truncation, or an empty `""` from a no-arg call)
was committed verbatim and replayed on every subsequent send. Strict renderers
that re-parse arguments at render time (vLLM's `deepseek_v4`
`_postprocess_messages` runs `json.loads` on them) reject the whole request
with HTTP 400, wedging the conversation. The only prior guard dropped partial
tool calls on `finish_reason == "length"`; a `stop`/`tool_calls` finish reason
carrying invalid JSON slipped through, and its synthetic "retry" result kept it
from being an orphan, so the orphan-repair pass never touched it.

Add `sanitize_tool_call_arguments`, a wire-neutral legalize pass in lowering
(fold, legalize, repair), normalizing any non-JSON-object `arguments` to `{}`
on the transient wire copy only. The canonical trajectory keeps the raw model
output, so a wedged session self-recovers on its next send. A
`wire_valid_arguments` predicate is shared with a non-destructive
`stream.tool_args_malformed` warning at the stream accumulator, which surfaces
the model-quality problem at production time.

Convert `lowering.py` to structlog so the new pass emits structured events.

(cherry picked from commit 16a68ae6d6)

tests/test_lowering.py

@@ -11,12 +11,16 @@ this pins the behaviour the old ``_anthropic`` ``pc_tool_ids`` /
 
 from __future__ import annotations
 
+import json
 from typing import Any
 
 from turnstone.core.lowering import (
     CANCELLED_TOOL_RESULT,
     _find_orphaned_tool_calls,
     repair_wire_messages,
+    sanitize_tool_call_arguments,
+    tool_args_preview,
+    wire_valid_arguments,
 )
 
 
@@ -180,3 +184,143 @@ def test_repair_does_not_mutate_input() -> None:
     repair_wire_messages(msgs)
     assert len(msgs) == original_len  # caller's list untouched
     assert "tool_calls" in msgs[0]
+
+
+# --------------------------------------------------------------------------- #
+# wire_valid_arguments — the shared "is this renderable" predicate
+# --------------------------------------------------------------------------- #
+def test_wire_valid_arguments_accepts_json_objects() -> None:
+    assert wire_valid_arguments("{}") is True
+    assert wire_valid_arguments('{"command": "ls -la"}') is True
+    assert wire_valid_arguments('  { "a": 1 }\n') is True  # surrounding whitespace ok
+
+
+def test_wire_valid_arguments_rejects_unrenderable() -> None:
+    assert wire_valid_arguments('{"command": "cat /va') is False  # unterminated (the incident)
+    assert wire_valid_arguments("") is False  # empty (no-arg call) — json.loads raises
+    assert wire_valid_arguments("[]") is False  # array, not object
+    assert wire_valid_arguments("5") is False  # bare scalar
+    assert wire_valid_arguments('"hi"') is False  # bare string
+    assert wire_valid_arguments(None) is False  # missing
+    assert wire_valid_arguments({"a": 1}) is False  # raw dict — not a string on the wire
+
+
+def test_wire_valid_arguments_totals_on_deeply_nested_json() -> None:
+    # Deeply-nested JSON makes json.loads raise RecursionError (not a ValueError);
+    # the predicate must return False, not propagate and crash the send.
+    deep = "[" * 5000 + "]" * 5000
+    assert wire_valid_arguments(deep) is False
+
+
+def test_tool_args_preview_stringifies_and_caps() -> None:
+    assert tool_args_preview("x" * 500) == "x" * 120
+    assert tool_args_preview(None) == "None"
+    assert tool_args_preview({"a": 1}) == "{'a': 1}"
+
+
+# --------------------------------------------------------------------------- #
+# sanitize_tool_call_arguments — the legalize pass
+# --------------------------------------------------------------------------- #
+def _call(call_id: str, arguments: Any, name: str = "bash") -> dict[str, Any]:
+    return {"id": call_id, "type": "function", "function": {"name": name, "arguments": arguments}}
+
+
+def _assistant_calls(*calls: dict[str, Any]) -> dict[str, Any]:
+    return {"role": "assistant", "content": "", "tool_calls": list(calls)}
+
+
+def test_sanitize_identity_when_all_valid() -> None:
+    msgs = [_assistant_calls(_call("c1", "{}"), _call("c2", '{"a": 1}')), _tool("c1"), _tool("c2")]
+    # Every arguments already a JSON object → same object returned (allocation-free).
+    assert sanitize_tool_call_arguments(msgs) is msgs
+
+
+def test_sanitize_identity_when_no_tool_calls() -> None:
+    msgs = [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "yo"}]
+    assert sanitize_tool_call_arguments(msgs) is msgs
+
+
+def test_sanitize_legalizes_unterminated_arguments() -> None:
+    # The production incident: deepseek-v4-flash emitted an unterminated args string
+    # with a non-``length`` finish reason, so it was committed and replayed verbatim.
+    msgs = [_assistant_calls(_call("c1", '{"command": "cat /va')), _tool("c1", "retry")]
+    out = sanitize_tool_call_arguments(msgs)
+    assert out is not msgs  # copied on repair
+    assert out[0]["tool_calls"][0]["function"]["arguments"] == "{}"
+    assert json.loads(out[0]["tool_calls"][0]["function"]["arguments"]) == {}
+
+
+def test_sanitize_legalizes_empty_arguments() -> None:
+    # A no-arg tool call sends ``""``; json.loads("") raises, so deepseek_v4 would 400.
+    out = sanitize_tool_call_arguments([_assistant_calls(_call("c1", ""))])
+    assert out[0]["tool_calls"][0]["function"]["arguments"] == "{}"
+
+
+def test_sanitize_legalizes_non_object_json() -> None:
+    out = sanitize_tool_call_arguments([_assistant_calls(_call("c1", "[]"), _call("c2", "5"))])
+    assert [tc["function"]["arguments"] for tc in out[0]["tool_calls"]] == ["{}", "{}"]
+
+
+def test_sanitize_serializes_raw_dict_arguments() -> None:
+    out = sanitize_tool_call_arguments([_assistant_calls(_call("c1", {"command": "ls"}))])
+    got = out[0]["tool_calls"][0]["function"]["arguments"]
+    assert isinstance(got, str) and json.loads(got) == {"command": "ls"}
+
+
+def test_sanitize_falls_back_when_dict_not_serializable() -> None:
+    # Defensive branch: a dict arguments carrying a non-JSON-encodable value
+    # (a set) makes json.dumps raise TypeError — it collapses to "{}", not a crash.
+    out = sanitize_tool_call_arguments([_assistant_calls(_call("c1", {"x": {1, 2, 3}}))])
+    assert out[0]["tool_calls"][0]["function"]["arguments"] == "{}"
+
+
+def test_sanitize_touches_only_the_offending_call() -> None:
+    good = _call("c1", '{"a": 1}')
+    bad = _call("c2", "{oops")
+    out = sanitize_tool_call_arguments([_assistant_calls(good, bad)])
+    # Valid sibling preserved by identity; only the bad call is rebuilt.
+    assert out[0]["tool_calls"][0] is good
+    assert out[0]["tool_calls"][1]["function"]["arguments"] == "{}"
+
+
+def test_sanitize_does_not_mutate_input() -> None:
+    raw = '{"command": "cat /va'
+    bad = _call("c1", raw)
+    msgs = [_assistant_calls(bad)]
+    sanitize_tool_call_arguments(msgs)
+    assert bad["function"]["arguments"] == raw  # caller's dict untouched
+    assert msgs[0]["tool_calls"][0] is bad
+
+
+# --------------------------------------------------------------------------- #
+# legalize ∘ repair — the two send-time validity passes compose
+# --------------------------------------------------------------------------- #
+def test_legalize_then_repair_answered_call() -> None:
+    # Malformed-but-answered (the poison-pill shape): args legalized, no orphan added.
+    msgs = [_assistant_calls(_call("c1", "{bad")), _tool("c1", "retry with valid JSON")]
+    out = repair_wire_messages(sanitize_tool_call_arguments(msgs))
+    assert [m["role"] for m in out] == ["assistant", "tool"]
+    assert json.loads(out[0]["tool_calls"][0]["function"]["arguments"]) == {}
+
+
+def test_legalize_then_repair_orphaned_call() -> None:
+    # Malformed AND unanswered: legalized args + a synthesized cancellation result.
+    msgs = [_assistant_calls(_call("c1", "{bad"))]
+    out = repair_wire_messages(sanitize_tool_call_arguments(msgs))
+    assert [m["role"] for m in out] == ["assistant", "tool"]
+    assert json.loads(out[0]["tool_calls"][0]["function"]["arguments"]) == {}
+    assert out[1]["content"] == CANCELLED_TOOL_RESULT
+
+
+def test_pipeline_every_emitted_arguments_is_a_json_object() -> None:
+    # The end-state invariant a strict renderer relies on.
+    msgs = [
+        _assistant_calls(_call("c1", ""), _call("c2", "{oops"), _call("c3", '{"ok": true}')),
+        _tool("c1"),
+        _tool("c2"),
+        _tool("c3"),
+    ]
+    out = repair_wire_messages(sanitize_tool_call_arguments(msgs))
+    for m in out:
+        for tc in m.get("tool_calls", []):
+            assert isinstance(json.loads(tc["function"]["arguments"]), dict)

tests/test_operator_instruction_declaration.py

@@ -8,6 +8,7 @@ capability-gated emission in ``ChatSession._init_system_messages``.
 
 from __future__ import annotations
 
+import json
 import logging
 from typing import TYPE_CHECKING
 
@@ -330,3 +331,38 @@ class TestEmptyUserTurnDrop:
         assert len(user_turns) == 1
         assert f"[start system-reminder_{nonce}]" in user_turns[0]["content"]
         assert "child done" in user_turns[0]["content"]
+
+
+class TestToolArgumentLegalization:
+    """``_prepare_wire_messages`` legalizes malformed tool-call ``arguments`` so a
+    strict renderer (vLLM ``deepseek_v4``) can ``json.loads`` every arguments string
+    — the sibling send-time validity pass to orphan repair."""
+
+    def test_unterminated_arguments_legalized_on_the_wire(self) -> None:
+        s = make_session()
+        msgs = [
+            {"role": "user", "content": "go"},
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "c1",
+                        "type": "function",
+                        "function": {"name": "bash", "arguments": '{"command": "cat /va'},
+                    }
+                ],
+            },
+            {"role": "tool", "tool_call_id": "c1", "content": "retry with valid JSON"},
+        ]
+        out = s._prepare_wire_messages(msgs)
+        emitted = [
+            tc["function"]["arguments"]
+            for m in out
+            if m.get("role") == "assistant"
+            for tc in m.get("tool_calls", [])
+        ]
+        assert emitted == ["{}"]
+        assert json.loads(emitted[0]) == {}
+        # Canonical input is untouched — legalization is wire-copy only.
+        assert msgs[1]["tool_calls"][0]["function"]["arguments"] == '{"command": "cat /va'

turnstone/core/lowering.py

@@ -6,12 +6,19 @@ trajectory) and the per-provider translators (which own format only — the
 *valid* for an LLM round-trip, so every translator can assume a well-formed
 input and stay a pure format mapping.
 
-This module owns the two provider-neutral lowering passes:
+This module owns the three provider-neutral lowering passes:
 
 * **fold** (representation) — operator-context ``system`` turns are folded into
   the preceding turn as nonce-fenced ``[start system-reminder]`` blocks for models
   without native mid-conversation system support (native models keep them
   inline).  See :func:`fold_system_turns`.
+* **legalize** (validity) — normalizing any tool-call ``arguments`` that isn't a
+  JSON-object string (an unterminated string from a non-``length`` truncation, an
+  empty ``""``, a bare scalar) to ``"{}"`` so a strict renderer (e.g. vLLM's
+  ``deepseek_v4``, which ``json.loads`` the arguments at request-render time)
+  can't reject the whole request.  Mutates the transient wire copy only — the
+  canonical trajectory keeps the raw output.  See
+  :func:`sanitize_tool_call_arguments`.
 * **repair** (validity) — synthesizing cancellation results for orphaned client
   tool calls.  See :func:`repair_wire_messages`.
 
@@ -45,13 +52,14 @@ their own.
 
 from __future__ import annotations
 
-import logging
+import json
 from typing import Any
 
 from turnstone.core import fence
+from turnstone.core.log import get_logger
 from turnstone.core.trajectory import EffectStatus, Turn, dicts_from_turns
 
-logger = logging.getLogger(__name__)
+log = get_logger(__name__)
 
 # The "you cannot tell whether it ran" clause, shared by every cancel
 # disposition surface (this wire-repair fallback AND the session-layer
@@ -161,6 +169,106 @@ def repair_wire_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]
     return out
 
 
+# --------------------------------------------------------------------------- #
+# Legalize — a tool call's ``arguments`` must be a JSON-object string on the wire.
+# --------------------------------------------------------------------------- #
+def wire_valid_arguments(arguments: Any) -> bool:
+    """True when *arguments* is a string that decodes to a JSON object.
+
+    A tool call carries ``arguments`` as an opaque JSON string, and a strict
+    renderer re-parses it at request-render time (vLLM's ``deepseek_v4``
+    ``_postprocess_messages`` does ``json.loads`` on it), so anything that isn't a
+    string decoding to a JSON *object* — an unterminated string from a
+    non-``length`` truncation, an empty ``""``, a bare scalar/array, a raw ``dict``
+    that never got serialized — makes the provider reject the whole request.
+    Shared by the wire legalizer here and the session-layer accumulator's integrity
+    check so the two can't drift on what "valid" means.
+    """
+    if not isinstance(arguments, str):
+        return False
+    try:
+        # json.loads raises JSONDecodeError (already a ValueError, so listing both
+        # was redundant) on malformed JSON, and RecursionError on deeply-nested
+        # JSON — catch both so this predicate is total for any string input.
+        return isinstance(json.loads(arguments), dict)
+    except (json.JSONDecodeError, RecursionError):
+        return False
+
+
+def tool_args_preview(arguments: Any) -> str:
+    """A short, log-safe preview of a tool call's raw ``arguments`` (any type).
+
+    Shared by the wire legalizer and the session-layer accumulator's
+    ``stream.tool_args_malformed`` warning so the two malformed-args log sites can't
+    drift on how much of the raw value they show.
+    """
+    text = arguments if isinstance(arguments, str) else repr(arguments)
+    return text[:120]
+
+
+def _legalized_arguments(arguments: Any) -> str | None:
+    """A wire-valid replacement for *arguments*, or ``None`` if already valid.
+
+    A raw ``dict`` (an internal shape that reached the wire seat) is serialized;
+    anything else that fails :func:`wire_valid_arguments` collapses to ``"{}"``.
+    The value is cosmetic on replay — a malformed call was already answered with a
+    "retry with valid JSON" result, and the model consumes that result, not its own
+    prior arguments — so an empty object drops nothing a strict renderer would keep.
+    """
+    if wire_valid_arguments(arguments):
+        return None
+    if isinstance(arguments, dict):
+        try:
+            return json.dumps(arguments)
+        except (TypeError, ValueError, RecursionError):
+            return "{}"
+    return "{}"
+
+
+def sanitize_tool_call_arguments(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Return *messages* with every assistant tool call's ``arguments`` made
+    wire-valid — the legalize pass (see the module docstring).
+
+    The stream accumulator commits ``arguments`` verbatim, and the only guard that
+    drops a malformed tool call is ``finish_reason == "length"``
+    (``ChatSession._stream_response``); a model that emits invalid JSON with a
+    ``stop`` / ``tool_calls`` finish reason slips through, and one such turn then
+    poison-pills every later request that replays it on a strict renderer.  This
+    legalizes each offending ``arguments`` to a JSON-object string.
+
+    Faithful and cheap, exactly like :func:`repair_wire_messages`: the canonical
+    ``Turn`` trajectory keeps the raw model output (this mutates only the transient
+    wire copy), and the pass is copy-on-write + identity-preserving — a
+    conversation with no malformed call is returned unchanged (same object).
+    """
+    out: list[dict[str, Any]] | None = None  # copy-on-write: None until first fix
+    for idx, msg in enumerate(messages):
+        if msg.get("role") != "assistant" or not msg.get("tool_calls"):
+            continue
+        repaired: list[dict[str, Any]] | None = None
+        for ci, tc in enumerate(msg["tool_calls"]):
+            fn = tc.get("function")
+            if not isinstance(fn, dict):
+                continue
+            replacement = _legalized_arguments(fn.get("arguments"))
+            if replacement is None:
+                continue  # already wire-valid — leave byte-for-byte untouched
+            if repaired is None:
+                repaired = list(msg["tool_calls"])
+            log.debug(
+                "wire.tool_args_legalized",
+                tool=fn.get("name", "?"),
+                call_id=tc.get("id", ""),
+                raw_preview=tool_args_preview(fn.get("arguments")),
+            )
+            repaired[ci] = {**tc, "function": {**fn, "arguments": replacement}}
+        if repaired is not None:
+            if out is None:
+                out = list(messages)
+            out[idx] = {**msg, "tool_calls": repaired}
+    return messages if out is None else out
+
+
 # --------------------------------------------------------------------------- #
 # Fold — operator-context representation (A); runs BEFORE repair on the wire.
 # --------------------------------------------------------------------------- #
@@ -225,7 +333,7 @@ def fold_system_turns(
                     # authorship.  Degrade (still fold) rather than crash the turn —
                     # the harm is OOD voice, not a trust breach (the nonce still
                     # gates operator trust regardless of host turn).
-                    logger.warning(
+                    log.warning(
                         "operator-context system turn (_source=%s) is folding onto "
                         "an assistant turn; operator context should follow a "
                         "user/tool turn",

turnstone/core/session.py

@@ -60,6 +60,9 @@ from turnstone.core.lowering import (
     drop_empty_user_turns,
     fold_system_turns,
     repair_wire_messages,
+    sanitize_tool_call_arguments,
+    tool_args_preview,
+    wire_valid_arguments,
 )
 from turnstone.core.memory import (
     count_messages,
@@ -4287,11 +4290,19 @@ class ChatSession:
         *after* the fold so the fold-path wake turn, which the nudge folds into
         and thereby fills, is kept.
 
+        Before that, :func:`turnstone.core.lowering.sanitize_tool_call_arguments`
+        legalizes any tool-call ``arguments`` that isn't a JSON-object string (a
+        model can emit an unterminated one with a non-``length`` finish reason, and
+        a strict renderer like vLLM's ``deepseek_v4`` ``json.loads`` it and 400s the
+        whole request) — on the wire copy only, so the canonical trajectory keeps
+        the raw output.
+
         Finally, :func:`turnstone.core.lowering.repair_wire_messages`
         synthesizes cancellation results for any orphaned client tool calls so
         the provider translator (the ``C`` layer) never sees an unanswered
         tool call — this is the sole send-time orphan repair; the translators
-        carry none.  Identity-preserving when nothing is orphaned.
+        carry none.  Both final passes are identity-preserving when there is
+        nothing to fix.
         """
         # The lowering passes (fold / drop / repair) are dict-native and the
         # provider translators consume the same dict projection, so the wire prep
@@ -4312,7 +4323,8 @@ class ChatSession:
                 nonce=self._envelope_nonce,
             )
         dropped = drop_empty_user_turns(folded)
-        return repair_wire_messages(dropped)
+        legalized = sanitize_tool_call_arguments(dropped)
+        return repair_wire_messages(legalized)
 
     def _emit_state(self, state: str) -> None:
         """Notify UI of a workstream state transition.
@@ -6545,11 +6557,29 @@ class ChatSession:
 
         if tool_calls_acc:
             self._ensure_tool_call_ids(tool_calls_acc)
-            msg["tool_calls"] = [tool_calls_acc[i] for i in sorted(tool_calls_acc)]
+            ordered = [tool_calls_acc[i] for i in sorted(tool_calls_acc)]
+            msg["tool_calls"] = ordered
+            # Non-destructive integrity signal: the length-guard above drops tool
+            # calls only on ``finish_reason == "length"``, so a model that emits
+            # invalid-JSON arguments with a ``stop`` / ``tool_calls`` finish reason
+            # commits them verbatim.  We keep the raw output (the canonical Turn
+            # stays a faithful record; the wire copy is legalized by
+            # ``lowering.sanitize_tool_call_arguments``) and only flag it here — so a
+            # model-quality problem is visible at the moment it happens, not merely
+            # as a downstream wire legalization on every replay.
+            for tc in ordered:
+                raw_args = tc["function"].get("arguments")
+                if not wire_valid_arguments(raw_args):
+                    log.warning(
+                        "stream.tool_args_malformed",
+                        tool=tc["function"].get("name", "?"),
+                        call_id=tc.get("id", ""),
+                        raw_preview=tool_args_preview(raw_args),
+                    )
             log.info(
                 "stream.tool_calls",
-                count=len(tool_calls_acc),
-                tools=[tool_calls_acc[i]["function"]["name"] for i in sorted(tool_calls_acc)],
+                count=len(ordered),
+                tools=[tc["function"]["name"] for tc in ordered],
             )
 
         # Store raw provider content blocks for multi-turn preservation