fix(gateway): retain partial answers after returned errors

Carry the partial-answer flag on failed completion frames and retain a replay body when no text deltas arrived. Preserve streamed segments and keep error-only, successful, and interrupted results out of this path.

Adapted from https://github.com/NousResearch/hermes-agent/pull/76112. The OpenCode terminal-stream reproductions in https://github.com/NousResearch/hermes-agent/pull/85724 are covered by existing routing and usage handling; leave those safeguards unchanged.

Co-authored-by: Jeffgithub0029 <gxalong@gmail.com>
Co-authored-by: brooklyn! <brooklyn.bb.nicholson@gmail.com>

tests/tui_gateway/test_failed_turn_retention.py

@@ -249,6 +249,75 @@ def test_completed_turn_still_clears_inflight(emits, turn_env):
     assert server._inflight_snapshot(session) is None
 
 
+@pytest.mark.parametrize("streamed_prefix", [None, "", "Earlier commentary. "])
+def test_returned_partial_error_keeps_final_response_text(emits, turn_env, monkeypatch, streamed_prefix):
+    """A partial answer survives both the terminal frame and a reconnect that missed it."""
+    text = "Hello! How can I help?"
+    error = "Response remained truncated after 4 continuation attempts"
+
+    def run(message, stream_callback=None, **kwargs):
+        if streamed_prefix is not None:
+            assert stream_callback is not None
+            stream_callback(streamed_prefix + text)
+        return {"final_response": text, "error": error, "failed": True, "partial": True}
+
+    agent = types.SimpleNamespace(
+        session_id="session-key",
+        run_conversation=run,
+        clear_interrupt=lambda: None,
+    )
+    session = _session(agent=agent, running=True)
+    server._start_inflight_turn(session, "hello")
+
+    server._run_prompt_submit("rid", "sid", session, "hello")
+
+    completes = _events(emits, "message.complete")
+    assert len(completes) == 1
+    payload = completes[0]
+    assert payload["status"] == "error"
+    assert payload["partial"] is True
+    assert payload["text"] == text
+    assert payload["error"] == error
+
+    monkeypatch.setattr(server, "_get_db", lambda: None)
+    resumed = server._live_session_payload("sid", session)
+    assert resumed["running"] is False
+    assert resumed["inflight"]["assistant"] == (streamed_prefix or "") + text
+    assert resumed["inflight"]["user"] == "hello"
+    assert resumed["inflight"]["status"] == payload["status"]
+    assert resumed["inflight"]["error"] == payload["error"]
+    assert resumed["inflight"]["error_surface"] == payload["error_surface"]
+    assert resumed["inflight"]["streaming"] is False
+
+
+@pytest.mark.parametrize("result, status", [
+    ({"final_response": "answer", "error": "failure", "failed": True}, "error"),
+    ({"final_response": "", "error": "failure", "partial": True}, "error"),
+    ({"final_response": "  ", "error": "failure", "partial": True}, "error"),
+    ({"final_response": " failure \n", "error": "failure", "partial": True}, "error"),
+    ({"final_response": "answer", "partial": True}, "complete"),
+    ({"final_response": "answer", "error": "failure", "partial": True, "interrupted": True}, "interrupted"),
+])
+def test_only_partial_answers_on_failed_turns_set_flag(emits, turn_env, result, status):
+    agent = types.SimpleNamespace(
+        session_id="session-key",
+        run_conversation=lambda *a, **k: result,
+        clear_interrupt=lambda: None,
+    )
+    session = _session(agent=agent, running=True)
+    server._start_inflight_turn(session, "do the thing")
+
+    server._run_prompt_submit("rid", "sid", session, "do the thing")
+
+    payload = _events(emits, "message.complete")[0]
+    assert payload["status"] == status
+    assert "partial" not in payload
+    if status == "error":
+        assert server._inflight_snapshot(session)["assistant"] == ""
+    else:
+        assert server._inflight_snapshot(session) is None
+
+
 # ── Exception path ─────────────────────────────────────────────────────
 
 

tui_gateway/prompt_turn.py

@@ -778,10 +778,17 @@ def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None,
     if rendered := render_message(raw, cols):
         payload["rendered"] = rendered
     error_value = result.get("error")
+    final_text = result.get("final_response")
+    has_partial_text = bool(
+        result.get("partial") and isinstance(final_text, str)
+        and final_text.strip() and final_text.strip() != str(error_value or "").strip())
     with session["history_lock"]:
         if status == "error":
             # Retain the failed turn: resume's inflight payload is the only carrier of the
             # failure if this frame is lost to a disconnect.
+            if has_partial_text and not (session.get("inflight_turn") or {}).get("assistant"):
+                # Non-streaming results need a replay body too; keep existing streamed segments intact.
+                _append_inflight_delta(session, raw)
             _fail_inflight_turn(session, error_value, error_surface=_error_surface)
             st.error_retained = True
             st.error_detail = _turn_failure_detail(
@@ -791,6 +798,9 @@ def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None,
     if status == "error":
         payload["error"] = str(error_value or raw)
         payload["recoverable"] = True
+        # Desktop distinguishes retained answer text from error copy using this flag.
+        if has_partial_text:
+            payload["partial"] = True
         if _error_surface:
             payload["error_surface"] = _error_surface
     if st.terminal_callback is not None: