diff --git a/api/models.py b/api/models.py index 4a17a371..b14e4f79 100644 --- a/api/models.py +++ b/api/models.py @@ -10717,6 +10717,86 @@ def merge_session_messages_append_only( _STRUCTURED_IDENTITY_MEMO.reset(token) +def _project_native_image_payload_conflicts_for_display( + sidecar_messages, + state_messages, + merged_messages, +): + """Project a proven same-row native-image conflict onto its sidecar bubble.""" + sidecar_messages = list(sidecar_messages or []) + state_messages = list(state_messages or []) + if not sidecar_messages or not state_messages: + return merged_messages + + sidecar_row_id_counts = collections.Counter( + row_id + for message in sidecar_messages + if (row_id := _state_db_row_identity(message)) is not None + ) + state_row_id_counts = collections.Counter( + row_id + for message in state_messages + if (row_id := _state_db_row_identity(message)) is not None + ) + display_conflicts = {} + for incoming in state_messages: + if ( + not isinstance(incoming, dict) + or incoming.get(_WEBUI_UNMATCHED_NATIVE_IMAGE_MIRROR_FIELD) is not True + ): + continue + row_id, row_id_valid = _state_db_row_identity_details(incoming) + timestamp, timestamp_valid = _message_exact_timestamp_details(incoming) + content = incoming.get("content") + if ( + not row_id_valid + or row_id is None + or sidecar_row_id_counts[row_id] != 1 + or state_row_id_counts[row_id] != 1 + or not timestamp_valid + or timestamp is None + or str(incoming.get("role") or "").lower() != "user" + or not isinstance(content, str) + ): + continue + incoming_api_content = _session_message_api_content_key(incoming) + if incoming_api_content is None: + continue + sidecar_owner = next(( + message for message in sidecar_messages + if isinstance(message, dict) + and _state_db_row_identity_details(message) == (row_id, True) + and _message_identity_compatible(message, incoming) + and message.get("content") == content + and _message_exact_timestamp_details(message) == (timestamp, True) + and _session_message_api_content_key(message) + not in (None, incoming_api_content) + ), None) + if sidecar_owner is not None: + display_conflicts[(row_id, timestamp, content)] = (incoming, sidecar_owner) + + if not display_conflicts: + return merged_messages + visible_messages = [] + for message in merged_messages: + if isinstance(message, dict) and isinstance(message.get("content"), str): + row_id, row_id_valid = _state_db_row_identity_details(message) + timestamp, timestamp_valid = _message_exact_timestamp_details(message) + conflict = ( + display_conflicts.get((row_id, timestamp, message["content"])) + if row_id_valid and row_id is not None and timestamp_valid + else None + ) + if ( + conflict is not None + and message is not conflict[1] + and _message_identity_compatible(message, conflict[0]) + ): + continue + visible_messages.append(message) + return visible_messages + + def _merge_session_messages_append_only_impl( sidecar_messages: list, state_messages: list, @@ -11538,80 +11618,11 @@ def reconciled_state_db_messages_for_session( incoming_provenance="state_db", ) if not prefer_context: - # Keep the append-only merge's conflicting provider payloads in model - # context. For display only, a marked native-image mirror can share the - # existing sidecar bubble when its durable identity, exact timestamp, - # and exact visible user content all agree. - sidecar_row_id_counts = collections.Counter( - row_id - for message in local_messages - if (row_id := _state_db_row_identity(message)) is not None + reconciled_messages = _project_native_image_payload_conflicts_for_display( + local_messages, + state_messages, + reconciled_messages, ) - state_row_id_counts = collections.Counter( - row_id - for message in state_messages or () - if (row_id := _state_db_row_identity(message)) is not None - ) - - display_conflicts = {} - for incoming in state_messages or (): - if ( - not isinstance(incoming, dict) - or incoming.get(_WEBUI_UNMATCHED_NATIVE_IMAGE_MIRROR_FIELD) is not True - ): - continue - row_id, row_id_valid = _state_db_row_identity_details(incoming) - timestamp, timestamp_valid = _message_exact_timestamp_details(incoming) - content = incoming.get("content") - if ( - not row_id_valid - or row_id is None - or sidecar_row_id_counts[row_id] != 1 - or state_row_id_counts[row_id] != 1 - or not timestamp_valid - or timestamp is None - or str(incoming.get("role") or "").lower() != "user" - or not isinstance(content, str) - ): - continue - incoming_api_content = _session_message_api_content_key(incoming) - if incoming_api_content is None: - continue - sidecar_owner = next(( - message for message in local_messages - if isinstance(message, dict) - and _state_db_row_identity_details(message) == (row_id, True) - and _message_identity_compatible(message, incoming) - and message.get("content") == content - and _message_exact_timestamp_details(message) == (timestamp, True) - and _session_message_api_content_key(message) - not in (None, incoming_api_content) - ), None) - if sidecar_owner is not None: - display_conflicts[(row_id, timestamp, content)] = ( - incoming, - sidecar_owner, - ) - - if display_conflicts: - visible_messages = [] - for message in reconciled_messages: - if isinstance(message, dict) and isinstance(message.get("content"), str): - row_id, row_id_valid = _state_db_row_identity_details(message) - timestamp, timestamp_valid = _message_exact_timestamp_details(message) - conflict = ( - display_conflicts.get((row_id, timestamp, message["content"])) - if row_id_valid and row_id is not None and timestamp_valid - else None - ) - if ( - conflict is not None - and message is not conflict[1] - and _message_identity_compatible(message, conflict[0]) - ): - continue - visible_messages.append(message) - reconciled_messages = visible_messages return _state_db_session_messages_result( reconciled_messages, state_revision, diff --git a/api/routes.py b/api/routes.py index 55a16815..0eb4867a 100644 --- a/api/routes.py +++ b/api/routes.py @@ -9304,6 +9304,7 @@ def _limited_webui_messages_for_display_with_sidecar( ) if not state_db_messages: return sidecar_messages + # NOTE: do not short-circuit to the sidecar when state.db has no strictly # newer rows. A state.db row whose timestamp is at-or-before the sidecar's # newest (recovery / edited-in-place / missing-timestamp cases) is still @@ -9364,6 +9365,11 @@ def _limited_webui_messages_for_display_with_sidecar( truncation_boundary=getattr(session, "truncation_boundary", None), incoming_provenance="state_db", ) + merged = _project_native_image_payload_conflicts_for_display( + sidecar_messages, + state_db_messages, + merged, + ) if cache_key is not None: _state_key = cache_key[4] _streaming_key = ( @@ -10580,6 +10586,7 @@ from api.models import ( get_state_db_session_message_keys_before_timestamp, get_state_db_session_summary, merge_session_messages_append_only, + _project_native_image_payload_conflicts_for_display, _suppress_native_image_display_mirrors, _reconcile_api_content_sidecars, _enrich_sidebar_lineage_metadata, @@ -13325,12 +13332,18 @@ def _handle_session_get(handler, parsed) -> bool: s, state_db_messages, ) + sidecar_messages = _webui_sidecar_lineage_messages_for_display(s) _all_msgs = merge_session_messages_append_only( - _webui_sidecar_lineage_messages_for_display(s), + sidecar_messages, state_db_messages, truncation_watermark=getattr(s, "truncation_watermark", None), truncation_boundary=getattr(s, "truncation_boundary", None), ) + _all_msgs = _project_native_image_payload_conflicts_for_display( + sidecar_messages, + state_db_messages, + _all_msgs, + ) _all_msgs = _merged_webui_lineage_messages_for_display(s, _all_msgs) else: if is_messaging_session and cli_messages: diff --git a/tests/test_native_image_turn_display_context.py b/tests/test_native_image_turn_display_context.py index 40402502..b6e6fa7b 100644 --- a/tests/test_native_image_turn_display_context.py +++ b/tests/test_native_image_turn_display_context.py @@ -227,6 +227,149 @@ def _write_state_db(path, session_id, rows): ) +@pytest.mark.parametrize("msg_limit", [None, 5]) +def test_get_session_projects_marked_payload_conflict_in_full_and_limited_paths( + monkeypatch, tmp_path, msg_limit, +): + import api.config + import api.models as models + import api.session_ops + from api import routes + + session_id = f"native-image-route-{msg_limit}" + timestamp = 880.0 + seed, identity, _ = _settle_image_turn( + session_id=session_id, + timestamp=timestamp, + agent_row_id=41, + ) + context_user = next( + message for message in seed.context_messages + if message.get("_active_turn_token") == identity["token"] + ) + mirror = _durable_agent_content(context_user["content"]) + sidecar_payload = "SIDE-CAR-PROVIDER-PAYLOAD" + state_payload = "STATE-DB-PROVIDER-PAYLOAD" + sidecar_attachments = [{ + "name": "sidecar-owned.png", + "mime": "image/png", + "is_image": True, + }] + sidecar_mirror = { + "role": "user", + "content": mirror, + "timestamp": timestamp, + "_state_db_row_id": 1, + "api_content": sidecar_payload, + "attachments": sidecar_attachments, + } + seed.messages.append(dict(sidecar_mirror)) + seed.context_messages.append(dict(sidecar_mirror)) + + session_dir = tmp_path / "sessions" + session_dir.mkdir() + db_path = tmp_path / "state.db" + _write_state_db(db_path, session_id, [(mirror, timestamp, state_payload)]) + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + monkeypatch.setattr(routes, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path) + session = models.Session( + session_id=session_id, + workspace="/fixture", + model="fixture-model", + context_length=128_000, + messages=seed.messages, + context_messages=seed.context_messages, + source_tag="webui", + session_source="webui", + ) + session.save(skip_index=True) + with models.LOCK: + models.SESSIONS.pop(session_id, None) + + monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *_: True) + monkeypatch.setattr(routes, "_active_stream_ids", lambda: set()) + monkeypatch.setattr(routes, "_lookup_cli_session_metadata", lambda *_: {}) + monkeypatch.setattr(routes, "find_run_summary", lambda *_: None) + monkeypatch.setattr(api.config, "load_settings", lambda: {"api_redact_enabled": False}) + monkeypatch.setattr(api.session_ops, "regeneration_state", lambda _session: ([], [])) + monkeypatch.setattr( + api.session_ops, + "regeneration_authority", + lambda *_args, **_kwargs: None, + ) + response = {} + monkeypatch.setattr( + routes, + "j", + lambda _handler, payload, status=200, **_kwargs: ( + response.update(payload=payload, status=status) or payload + ), + ) + query = f"session_id={session_id}&resolve_model=0" + if msg_limit is not None: + query += f"&msg_limit={msg_limit}" + routes._handle_session_get(None, SimpleNamespace(path="/api/session", query=query)) + + assert response["status"] == 200 + public_session = response["payload"]["session"] + public_messages = public_session["messages"] + mirror_rows = [message for message in public_messages if message.get("content") == mirror] + assert len(mirror_rows) == 1, f"expected one visible mirror, found {len(mirror_rows)}" + assert all("api_content" not in message for message in mirror_rows) + assert public_session["message_count"] == 3 + assert mirror_rows[0]["attachments"] == sidecar_attachments + + state_rows = models.get_state_db_session_messages(session_id) + replay_context = models.reconciled_state_db_messages_for_session( + session, + prefer_context=True, + state_messages=state_rows, + ) + replay_rows = [ + message for message in _sanitize_messages_for_agent(replay_context) + if message.get("content") == mirror + ] + assert len(replay_rows) == 2 + assert {message["api_content"] for message in replay_rows} == { + sidecar_payload, + state_payload, + } + + +def test_limited_conflict_projection_keeps_state_row_when_owner_is_outside_slice(): + from api import routes + + timestamp = 881.0 + mirror = "Describe this image [screenshot]" + sidecar_slice = [{"role": "user", "content": "Earlier submission", "timestamp": 879.0}] + marked_state_row = { + "role": "user", + "content": mirror, + "timestamp": timestamp, + "_state_db_row_id": 42, + "api_content": "STATE-DB-PROVIDER-PAYLOAD", + "_webui_unmatched_native_image_mirror": True, + } + session = SimpleNamespace( + session_id="limited-conflict-owner-outside-page", + messages=sidecar_slice, + context_messages=[], + ) + merged = routes._limited_webui_messages_for_display_with_sidecar( + session, + sidecar_slice, + [marked_state_row], + state_db_signature=None, + msg_before=1, + ) + assert any(message.get("content") == "Earlier submission" for message in merged) + state_rows = [message for message in merged if message.get("content") == mirror] + assert len(state_rows) == 1 + assert state_rows[0]["api_content"] == marked_state_row["api_content"] + + @pytest.mark.parametrize("msg_limit", [None, 5]) @pytest.mark.parametrize("attachment_mode", ["native_image", "text_attachment"]) def test_get_session_keeps_pending_agent_projection_private_but_in_context(