fix(chat): preserve journal recovery after a dead image turn

Count a proved WebUI-owned pending row in the display baseline before
accepting a state.db self-heal. Without assistant output past that turn,
leave pending state for the run journal to restore partial text and the
interruption marker instead of clearing it prematurely.

Assisted-by: Hermes Agent:gpt-6-sol
Assisted-by: Codex:gpt-6-luna
Assisted-by: Claude Code:claude-opus-5-5

api/models.py

@@ -4095,15 +4095,14 @@ def _sync_sidecar_from_state_db_if_newer(session) -> bool:
                     locked_messages.insert(0, pending_row)
             locked.messages = locked_messages
 
+        display_baseline_count = len(locked_messages)
         merged_messages = reconciled_state_db_messages_for_session(
             locked,
             state_messages=state_messages,
         )
-        # The reconciler is append-only: a genuine state.db advance (output the
-        # lost stream never wrote back) shows up as MORE rows than the sidecar.
-        # A merged length not greater than the sidecar means nothing new to
-        # recover — leave the sidecar untouched rather than rewriting in place.
-        if len(merged_messages) <= locked_count:
+        # The baseline includes any proven WebUI-owned pending display row. Only
+        # a later state.db row is an advance worth committing and clearing pending.
+        if len(merged_messages) <= display_baseline_count:
             return False
         if merged_context is None:
             merged_context = reconciled_state_db_messages_for_session(

docs/rfcs/webui-run-state-consistency-contract.md

@@ -185,10 +185,12 @@ and 5; it does not mark every run-state boundary implemented.
    ambiguous display bucket until the turn settles; keep all rows in model
    context. Do not identify the row by its text. If a stream dies before
    settlement, state.db self-heal must save the submitted prompt and attachments
-   as a visible sidecar row before clearing pending metadata; the Agent row
-   remains available in model context. A partial continuation must use one
-   consistent parent snapshot when projecting a conflicting provider payload
-   onto its sidecar-owned display row.
+   as a visible sidecar row before clearing pending metadata only when there is
+   genuine state.db output beyond that submitted turn. Otherwise, leave pending
+   state intact for journaled partial-output and interruption-marker recovery.
+   The Agent row remains available in model context. A partial continuation
+   must use one consistent parent snapshot when projecting a conflicting
+   provider payload onto its sidecar-owned display row.
    Match settled native-image scalar projections only with trusted turn and
    durable-row identity, never the marker alone. A durable row ID proves row
    identity, not provider-payload freshness: when the sidecar and state.db have

tests/test_session_lost_response_regression.py

@@ -895,6 +895,92 @@ def test_get_session_sync_keeps_pending_image_prompt_when_agent_row_is_proven(mo
     assert agent_only_note in str(enriched_rows[0]["content"])
 
 
+def test_state_db_pending_image_row_alone_does_not_skip_stale_journal_recovery(
+    monkeypatch, hermes_home,
+):
+    sid = "state_pending_image_without_output_sid"
+    stream_id = "dead_pending_image_without_output_stream"
+    timestamp = time.time() - models._REPAIR_STALE_PENDING_GRACE_SECONDS - 5
+    prompt = "Describe the attached image"
+    attachment = {"name": "sample.png", "mime": "image/png", "is_image": True}
+    old_messages = [
+        {"role": "user", "content": "old question", "timestamp": timestamp - 2},
+        {"role": "assistant", "content": "old answer", "timestamp": timestamp - 1},
+    ]
+    state_messages = [
+        *old_messages,
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": prompt},
+                {"type": "image_url", "image_url": {"url": "data:image/png;base64,YWJj"}},
+            ],
+            "timestamp": timestamp,
+        },
+    ]
+    session = Session(
+        session_id=sid,
+        title="Pending image without output",
+        messages=old_messages,
+        context_messages=old_messages,
+        active_stream_id=stream_id,
+        pending_user_message=prompt,
+        pending_attachments=[attachment],
+        pending_started_at=timestamp,
+        pending_user_source="webui",
+        _webui_pending_user_timestamp_identity=(stream_id, timestamp),
+    )
+    session.save()
+    models.SESSIONS[sid] = session
+
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_summary",
+        lambda sid_arg, profile=None: {
+            "message_count": len(state_messages),
+            "last_message_at": timestamp,
+        },
+    )
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_messages",
+        lambda sid_arg, **kwargs: list(state_messages),
+    )
+    append_run_event(sid, stream_id, "token", {"text": "Recovered partial text."})
+
+    # Cached get_session runs state.db self-heal, then the real stale-stream
+    # cleanup must still get the pending turn so it can replay the run journal.
+    loaded = models.get_session(sid)
+    assert loaded is session
+    assert loaded.active_stream_id == stream_id
+    assert loaded.pending_user_message == prompt
+    assert loaded.pending_attachments == [attachment]
+
+    import api.routes as routes
+
+    assert routes._clear_stale_stream_state(loaded) is True
+
+    reloaded = Session.load(sid)
+    submitted_rows = [
+        message for message in reloaded.messages
+        if message.get("role") == "user" and message.get("content") == prompt
+    ]
+    assert len(submitted_rows) == 1
+    assert submitted_rows[0]["attachments"] == [attachment]
+    assert any(
+        message.get("_recovered_from_run_journal")
+        and "Recovered partial text." in message.get("content", "")
+        for message in reloaded.messages
+    )
+    markers = [
+        message for message in reloaded.messages
+        if message.get("_error") and message.get("type") == "interrupted"
+    ]
+    assert len(markers) == 1
+    assert "**Response interrupted.**" in markers[0]["content"]
+    assert "partial output above was recovered" in markers[0]["content"]
+
+
 def test_get_session_does_not_sync_while_stream_is_still_live(monkeypatch):
     """A still-running worker owns its own writeback; do not race it.