Merge PR #7754 from starship-s/fix/image-turn-display-context

tests/test_7754_image_row_replay_checkpoint.py

@@ -0,0 +1,65 @@
+"""#7754: a marked native-image state row that replays its sidecar row must consume that replay slot.
+
+Before the fix, the row-id fast path for a marked native-image mirror row `continue`d without
+advancing the replay index. The edited checkpoint then never read as consumed, and a rerun reply
+that exists only in state.db was truncated. Master (no marker branch) keeps that reply.
+"""
+
+import api.models as models
+
+
+def _merge(*, marked):
+    field = models._WEBUI_UNMATCHED_NATIVE_IMAGE_MIRROR_FIELD
+    image_row = {
+        "role": "user",
+        "content": "look",
+        "timestamp": 860.0,
+        "_state_db_row_id": 42,
+        "api_content": "PAYLOAD-A",
+    }
+    sidecar = [
+        {"role": "user", "content": "u1", "timestamp": 100.0},
+        {"role": "assistant", "content": "a1", "timestamp": 101.0},
+        dict(image_row),
+        {"role": "assistant", "content": "a_img", "timestamp": 861.0},
+        {"role": "user", "content": "uE", "timestamp": 900.0},
+    ]
+    state_image_row = dict(image_row)
+    if marked:
+        state_image_row[field] = True
+    state = [
+        {"role": "user", "content": "u1", "timestamp": 100.0},
+        {"role": "assistant", "content": "a1", "timestamp": 101.0},
+        state_image_row,
+        {"role": "assistant", "content": "a_img", "timestamp": 861.0},
+        {"role": "user", "content": "u_old", "timestamp": 870.0},
+        {"role": "assistant", "content": "a_old", "timestamp": 871.0},
+        {"role": "user", "content": "uE", "timestamp": 900.0},
+        {"role": "assistant", "content": "aE", "timestamp": 901.0},
+    ]
+    return models.merge_session_messages_append_only(
+        sidecar,
+        state,
+        truncation_watermark=900.0,
+        truncation_boundary=861.0,
+        incoming_provenance="state_db",
+    )
+
+
+def _contents(rows):
+    return [row.get("content") for row in rows]
+
+
+def test_marked_image_row_keeps_state_only_reply_after_edited_checkpoint():
+    merged = _merge(marked=True)
+    assert _contents(merged) == ["u1", "a1", "look", "a_img", "uE", "aE"]
+
+
+def test_marked_and_unmarked_image_rows_merge_identically():
+    assert _contents(_merge(marked=True)) == _contents(_merge(marked=False))
+
+
+def test_removed_rows_between_boundary_and_watermark_stay_out():
+    contents = _contents(_merge(marked=True))
+    assert "u_old" not in contents
+    assert "a_old" not in contents

tests/test_batch_fixes.py

@@ -41,12 +41,13 @@ class TestRootWorkspaceUnblocked:
         assert "'/etc'" in src or 'Path("/etc")' in src or "Path('/etc')" in src
         assert "'/proc'" in src or 'Path("/proc")' in src or "Path('/proc')" in src
 
-    def test_split_guard_present(self):
-        src = read("api/streaming.py")
-        assert "'\\n\\n[Attached files:' in msg_text" in src, (
-            "base_text split must guard against missing '[Attached files:' "
-            "to avoid empty-string on plain messages"
-        )
+    def test_attached_files_suffix_is_optional(self):
+        from api.models import _strip_attached_files_marker
+
+        assert _strip_attached_files_marker("Plain message") == "Plain message"
+        assert _strip_attached_files_marker(
+            "Question\n\n[Attached files: photo.png]"
+        ) == "Question"
 
 
 # ── Group B: custom_providers visibility ─────────────────────────────────────

tests/test_compression_snapshot_revision.py

@@ -1068,6 +1068,7 @@ def test_auth_self_heal_refreshes_revision_after_first_agent_persists_user(
         messages=prior_messages,
         context_messages=prior_messages,
     )
+    _session.pending_user_source = "webui"
     revisions = []
 
     class AuthThenRecoverAgent:
@@ -1080,8 +1081,14 @@ def test_auth_self_heal_refreshes_revision_after_first_agent_persists_user(
             self._last_error = None
             self.stream_delta_callback = _kwargs.get("stream_delta_callback")
 
-        def run_conversation(self, **kwargs):
+        def run_conversation(self, persist_user_timestamp=None, **kwargs):
             type(self).runs += 1
+            assert persist_user_timestamp == 10.0
+            persisted = Session.load(sid)
+            assert persisted._webui_pending_user_timestamp_identity == (
+                stream_id,
+                persist_user_timestamp,
+            )
             revisions.append(kwargs["conversation_history_revision"])
             history = list(kwargs.get("conversation_history") or [])
             if type(self).runs == 1:

tests/test_session_lost_response_regression.py

@@ -818,6 +818,169 @@ def test_get_session_syncs_sidecar_from_newer_state_db_even_when_stream_not_term
     assert reloaded.messages[-1]["content"] == "latest live progress"
 
 
+def test_get_session_sync_keeps_pending_image_prompt_when_agent_row_is_proven(monkeypatch, hermes_home):
+    sid = "state_newer_pending_image_sid"
+    stream_id = "dead_pending_image_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}
+    agent_only_note = "Agent-only recall note"
+    agent_user_content = [
+        {"type": "text", "text": prompt},
+        {"type": "image_url", "image_url": {"url": "data:image/png;base64,YWJj"}},
+        {"type": "text", "text": agent_only_note},
+    ]
+    previous_messages = [
+        {"role": "user", "content": "old question", "timestamp": timestamp - 2},
+        {"role": "assistant", "content": "old answer", "timestamp": timestamp - 1},
+    ]
+    state_messages = [
+        *previous_messages,
+        {"role": "user", "content": agent_user_content, "timestamp": timestamp},
+        {"role": "assistant", "content": "recovered assistant tail", "timestamp": timestamp + 1},
+    ]
+    s = Session(
+        session_id=sid,
+        title="Pending image recovery",
+        messages=previous_messages,
+        context_messages=previous_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),
+    )
+    s.save()
+    models.SESSIONS.pop(sid, None)
+
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_summary",
+        lambda sid_arg, profile=None: {
+            "message_count": len(state_messages),
+            "last_message_at": timestamp + 1,
+        },
+    )
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_messages",
+        lambda sid_arg, **kwargs: list(state_messages),
+    )
+
+    models.get_session(sid)
+    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]["timestamp"] == timestamp
+    assert submitted_rows[0]["attachments"] == [attachment]
+    assert reloaded.messages[-1] == state_messages[-1]
+    assert agent_only_note not in str(reloaded.messages)
+
+    context_rows = models.reconciled_state_db_messages_for_session(
+        reloaded,
+        prefer_context=True,
+        state_messages=state_messages,
+    )
+    enriched_rows = [
+        message for message in context_rows
+        if message.get("role") == "user" and message.get("timestamp") == timestamp
+    ]
+    assert len(enriched_rows) == 1
+    assert enriched_rows[0]["content"] == agent_user_content
+    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.
 
@@ -884,6 +1047,7 @@ def test_sync_save_failure_does_not_mutate_session(monkeypatch):
     """
     sid = "state_sync_save_fail_sid"
     stream_id = "stream_save_fail_no_done"
+    attachment = {"name": "sample.png", "mime": "image/png", "is_image": True}
     s = Session(
         session_id=sid,
         title="Save failure",
@@ -893,7 +1057,10 @@ def test_sync_save_failure_does_not_mutate_session(monkeypatch):
         ],
         active_stream_id=stream_id,
         pending_user_message="new request",
+        pending_attachments=[attachment],
         pending_started_at=102.0,
+        pending_user_source="webui",
+        _webui_pending_user_timestamp_identity=(stream_id, 102.0),
     )
     s.save()
 
@@ -926,7 +1093,13 @@ def test_sync_save_failure_does_not_mutate_session(monkeypatch):
     assert result is False
     assert s.active_stream_id == stream_id
     assert s.pending_user_message == "new request"
+    assert s.pending_attachments == [attachment]
     assert [m["content"] for m in s.messages] == ["old question", "old answer"]
+    reloaded = Session.load(sid)
+    assert reloaded.active_stream_id == stream_id
+    assert reloaded.pending_user_message == "new request"
+    assert reloaded.pending_attachments == s.pending_attachments
+    assert [m["content"] for m in reloaded.messages] == ["old question", "old answer"]
 
 
 def test_sync_persists_recovered_state_db_tail_when_stream_dead(monkeypatch):