diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 9fb894287a33e..9ffdcc200e914 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -2779,6 +2779,117 @@ def repair_tool_call(agent, tool_name: str) -> str | None:
 
 
 
+# Placeholder substituted for an empty non-final message that would otherwise
+# make the provider reject the whole request. Kept identical to the stub-
+# creation placeholder in chat_completion_helpers so a healed transcript reads
+# consistently whether the empty turn was caught at write time or send time.
+_INTERRUPTED_PLACEHOLDER = "[response interrupted]"
+
+
+def _msg_has_payload(msg: Dict[str, Any]) -> bool:
+    """True if ``msg`` carries anything the API treats as non-empty content.
+
+    Covers string content, non-empty multimodal content lists, tool_calls,
+    tool_call_id linkage (tool results), and reasoning payloads. Mirrors the
+    emptiness checks used by ``AIAgent._is_thinking_only_assistant`` but is
+    role-agnostic so it can vet user/assistant/tool turns uniformly.
+    """
+    content = msg.get("content")
+    if isinstance(content, str):
+        if content.strip():
+            return True
+    elif isinstance(content, list):
+        for block in content:
+            if isinstance(block, dict):
+                # any typed block (text/image/tool_use/document/...) counts,
+                # as long as a text block is not itself blank
+                if block.get("type") == "text":
+                    if isinstance(block.get("text"), str) and block["text"].strip():
+                        return True
+                    continue
+                return True
+            elif block:
+                return True
+    elif content not in (None, ""):
+        return True
+    # Structural payloads that make an "empty-content" message still valid.
+    if msg.get("tool_calls"):
+        return True
+    if isinstance(msg.get("reasoning_content"), str) and msg["reasoning_content"].strip():
+        return True
+    if msg.get("reasoning") or msg.get("reasoning_details"):
+        return True
+    return False
+
+
+def repair_empty_non_final_messages(
+    messages: List[Dict[str, Any]],
+) -> List[Dict[str, Any]]:
+    """Heal empty-content non-final messages before they reach the provider.
+
+    Root-cause context: a stream that dies with 0 recovered characters (peer
+    reset, stall-kill) could persist an assistant turn with ``content=None``
+    and no tool_calls. The Anthropic message schema — and the litellm/Bedrock
+    proxies in front of it — reject ANY request whose transcript contains an
+    empty non-final message:
+
+        "all messages must have non-empty content except for the optional
+         final assistant message"  (HTTP 400 INVALID_REQUEST_BODY)
+
+    Once such a message lands mid-transcript it poisons EVERY subsequent turn
+    of that session until it scrolls out of context. The write-time guard in
+    ``chat_completion_helpers`` stops NEW stubs, but sessions already carrying
+    one (persisted before the guard, or fed in from a host history) stay stuck
+    and previously needed a manual DB edit + gateway restart to recover.
+
+    This pass is the self-healing counterpart: it runs unconditionally on the
+    per-call ``api_messages`` copy, so a poisoned transcript repairs itself
+    IN MEMORY on the very next send — no restart, no DB surgery. The final
+    message is left untouched (an empty final assistant turn is legal). The
+    stored conversation history is never mutated; only the wire copy is
+    repaired, so the UI/session trace stays faithful.
+
+    Repair strategy is substitution, not deletion: dropping a mid-transcript
+    turn can break role alternation and tool-call pairing, whereas an honest
+    minimal placeholder keeps the sequence intact and reads correctly as an
+    interrupted turn on replay.
+    """
+    if not messages or len(messages) < 2:
+        return messages
+
+    repaired: List[Dict[str, Any]] = []
+    healed = 0
+    last_idx = len(messages) - 1
+    for idx, msg in enumerate(messages):
+        if (
+            idx != last_idx
+            and isinstance(msg, dict)
+            # tool results are validated by their own orphan/pairing pass; an
+            # empty tool result is a separate (and rarer) concern.
+            and msg.get("role") in ("assistant", "user")
+            and not _msg_has_payload(msg)
+        ):
+            # Shallow-copy so stored history / prompt caching stays byte-stable.
+            fixed = dict(msg)
+            fixed["content"] = _INTERRUPTED_PLACEHOLDER
+            repaired.append(fixed)
+            healed += 1
+        else:
+            repaired.append(msg)
+
+    if healed:
+        _ra().logger.warning(
+            "Pre-call sanitizer: healed %d empty non-final message(s) by "
+            "substituting placeholder content — an empty-content turn was in "
+            "the transcript and would 400 the request ('messages must have "
+            "non-empty content' / INVALID_REQUEST_BODY). Self-recovering the "
+            "poisoned transcript in memory; no restart needed.",
+            healed,
+        )
+        return repaired
+    return messages
+
+
 def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     """Fix orphaned tool_call / tool_result pairs before every LLM call.
 
@@ -2799,6 +2910,15 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
         filtered.append(msg)
     messages = filtered
 
+    # --- Heal empty-content non-final messages (self-recovery) ---
+    # A dead stream can leave an empty assistant stub (or an empty user turn)
+    # mid-transcript; the provider then 400s EVERY subsequent request until it
+    # scrolls out. Repair it here, on the per-call copy, so a poisoned session
+    # recovers itself in memory on the next send — no restart, no DB edit.
+    # Done first so a substituted turn participates normally in the tool-pair
+    # and dedup passes below.
+    messages = repair_empty_non_final_messages(messages)
+
     # --- Drop empty / malformed tool_calls arrays on assistant messages ---
     # An assistant message carrying ``tool_calls: []`` (an empty array) — or a
     # non-list value under the key — is semantically identical to an assistant
diff --git a/tests/run_agent/test_message_sequence_repair.py b/tests/run_agent/test_message_sequence_repair.py
index 816af59fd891f..589f5fbf4f4c1 100644
--- a/tests/run_agent/test_message_sequence_repair.py
+++ b/tests/run_agent/test_message_sequence_repair.py
@@ -770,3 +770,101 @@ def test_sanitize_preserves_populated_tool_calls():
     out = sanitize_api_messages(list(messages))
     assistant = [m for m in out if m.get("role") == "assistant"][0]
     assert [tc["id"] for tc in assistant["tool_calls"]] == ["call_Z"]
+
+
+# ── Self-recovery: heal empty-content non-final messages ──────────────────
+# Repro of the production incident: a dead stream persisted an empty-content
+# assistant stub mid-transcript, and every later request 400'd with
+# "all messages must have non-empty content except for the optional final
+# assistant message" (INVALID_REQUEST_BODY). sanitize_api_messages now heals
+# such turns on the per-call copy so the session recovers itself in memory.
+
+def test_sanitize_heals_empty_assistant_stub_between_tool_and_user():
+    """The exact production shape: tool -> EMPTY assistant (finish=length) ->
+    user recovery. The empty assistant would 400 the whole request; it must be
+    substituted with non-empty placeholder content, not left empty."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    messages = [
+        {"role": "user", "content": "do a thing"},
+        {"role": "assistant", "content": None, "tool_calls": [
+            {"id": "call_A", "type": "function",
+             "function": {"name": "foo", "arguments": "{}"}}]},
+        {"role": "tool", "tool_call_id": "call_A", "content": "result"},
+        {"role": "assistant", "content": None},  # <-- poison stub (non-final)
+        {"role": "user", "content": "[System: previous response was cut off]"},
+    ]
+    out = sanitize_api_messages(list(messages))
+    # Every non-final message now has non-empty content.
+    for m in out[:-1]:
+        c = m.get("content")
+        assert m.get("tool_calls") or (isinstance(c, str) and c.strip()) or (
+            isinstance(c, list) and c), f"empty non-final survived: {m}"
+    # The stub specifically became the placeholder.
+    stub = out[3]
+    assert stub["role"] == "assistant"
+    assert stub["content"] == "[response interrupted]"
+
+
+def test_sanitize_heals_empty_user_message():
+    """An empty user turn mid-transcript is equally invalid and is healed."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    messages = [
+        {"role": "assistant", "content": "hello"},
+        {"role": "user", "content": ""},          # <-- empty user (non-final)
+        {"role": "assistant", "content": "still here"},
+    ]
+    out = sanitize_api_messages(list(messages))
+    assert out[1]["content"] == "[response interrupted]"
+
+
+def test_sanitize_allows_empty_final_assistant():
+    """Negative control: an empty FINAL assistant message is legal per the API
+    ('except for the optional final assistant message') and must NOT be
+    touched."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    messages = [
+        {"role": "user", "content": "hi"},
+        {"role": "assistant", "content": ""},  # final — allowed empty
+    ]
+    out = sanitize_api_messages(list(messages))
+    assert out[-1]["content"] == ""  # untouched
+
+
+def test_sanitize_empty_heal_is_non_destructive():
+    """Healing must not mutate the caller's persisted dicts — only the per-call
+    copy is repaired, so the stored trajectory keeps the true empty turn."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    original_stub = {"role": "assistant", "content": None}
+    messages = [
+        {"role": "user", "content": "q"},
+        original_stub,
+        {"role": "user", "content": "recovery"},
+    ]
+    sanitize_api_messages(list(messages))
+    assert original_stub["content"] is None  # untouched in-place
+
+
+def test_sanitize_preserves_reasoning_only_and_toolcall_turns():
+    """Negative control: a non-final assistant turn that is 'empty content' but
+    carries reasoning OR tool_calls is valid payload and must NOT be rewritten
+    to the placeholder (that would clobber real model output)."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    messages = [
+        {"role": "user", "content": "q"},
+        {"role": "assistant", "content": None, "tool_calls": [
+            {"id": "call_R", "type": "function",
+             "function": {"name": "foo", "arguments": "{}"}}]},
+        {"role": "tool", "tool_call_id": "call_R", "content": "r"},
+        {"role": "user", "content": "next"},
+    ]
+    out = sanitize_api_messages(list(messages))
+    # tool_call assistant keeps its tool_calls, content not clobbered to text
+    tc_asst = out[1]
+    assert tc_asst.get("tool_calls") and tc_asst["tool_calls"][0]["id"] == "call_R"
+    assert tc_asst["content"] != "[response interrupted]"
+