diff --git a/api/models.py b/api/models.py index 80098e71..4a17a371 100644 --- a/api/models.py +++ b/api/models.py @@ -11537,6 +11537,81 @@ def reconciled_state_db_messages_for_session( truncation_boundary=getattr(session, "truncation_boundary", None), 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 + ) + 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/streaming.py b/api/streaming.py index f4df3f2f..050ba1d0 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -9137,10 +9137,42 @@ def _add_supported_run_conversation_kwarg(callable_obj, kwargs, name, value): return False +def _register_pending_user_timestamp_identity( + callable_obj, session, persist_user_timestamp, +): + """Persist the pending-turn proof while the caller owns the session lock.""" + if session is None: + return + try: + timestamp_parameter = inspect.signature(callable_obj).parameters.get( + "persist_user_timestamp" + ) + except (TypeError, ValueError): + timestamp_parameter = None + has_timestamp_contract = ( + timestamp_parameter is not None + and timestamp_parameter.kind is not inspect.Parameter.POSITIONAL_ONLY + ) + identity = ( + _validated_webui_pending_user_timestamp_identity( + session, + (getattr(session, "active_stream_id", None), persist_user_timestamp), + ) + if has_timestamp_contract + else None + ) + previous = getattr(session, "_webui_pending_user_timestamp_identity", None) + session._webui_pending_user_timestamp_identity = identity + session_id = getattr(session, "session_id", None) + save = getattr(session, "save", None) + if callable(save) and session_id and previous != identity: + # Persist before run_conversation can write this turn to state.db. + save(touch_updated_at=False, skip_index=True) + + def _build_run_conversation_kwargs( callable_obj, *, - session=None, user_message, system_message, conversation_history, @@ -9149,7 +9181,7 @@ def _build_run_conversation_kwargs( persist_user_message, persist_user_timestamp, ): - """Build one rolling-compatible Agent invocation contract. + """Build one rolling-compatible Agent invocation contract without mutation. ``persist_user_timestamp`` is signature-gated (#6935): an older hermes-agent whose ``run_conversation()`` predates the kwarg must not @@ -9174,45 +9206,6 @@ def _build_run_conversation_kwargs( "conversation_history_revision", conversation_history_revision, ) - if session is not None: - try: - timestamp_parameter = inspect.signature(callable_obj).parameters.get( - "persist_user_timestamp" - ) - except (TypeError, ValueError): - timestamp_parameter = None - has_timestamp_contract = ( - timestamp_parameter is not None - and timestamp_parameter.kind is not inspect.Parameter.POSITIONAL_ONLY - ) - session_id = getattr(session, "session_id", None) - save = getattr(session, "save", None) - lock = ( - _get_session_agent_lock(session_id) - if callable(save) and session_id - else contextlib.nullcontext() - ) - with lock: - candidate = ( - _validated_webui_pending_user_timestamp_identity( - session, - ( - getattr(session, "active_stream_id", None), - kwargs.get("persist_user_timestamp"), - ), - ) - if has_timestamp_contract - else None - ) - # A retry using an older/opaque Agent must revoke earlier proof; - # row identity cannot be carried across a changed invocation contract. - identity = candidate - previous = getattr(session, "_webui_pending_user_timestamp_identity", None) - session._webui_pending_user_timestamp_identity = identity - if callable(save) and session_id and previous != identity: - # Persist the proof while holding the session mutation lock, - # before run_conversation can write this turn to state.db. - save(touch_updated_at=False, skip_index=True) return kwargs @@ -11763,9 +11756,13 @@ def _run_agent_streaming( _agent_msg_text = "\n\n".join([*_process_notifications, msg_text]).strip() user_message = _build_native_multimodal_message(workspace_ctx, _agent_msg_text, attachments, workspace, cfg=_cfg, active_provider=(resolved_provider or ""), active_model=(resolved_model or ""), requested_provider=(_session_requested_provider or ""), profile=(getattr(s, "profile", None) or Path(_profile_home))) _persistent_state_before = _persistent_state_snapshot(_profile_home) + with _agent_lock: + _persist_user_timestamp = getattr(s, 'pending_started_at', None) + _register_pending_user_timestamp_identity( + agent.run_conversation, s, _persist_user_timestamp + ) _run_conversation_kwargs = _build_run_conversation_kwargs( agent.run_conversation, - session=s, user_message=user_message, system_message=workspace_system_msg, conversation_history=_sanitize_messages_for_agent( @@ -11779,7 +11776,7 @@ def _run_agent_streaming( conversation_history_revision=_conversation_history_revision, task_id=session_id, persist_user_message=msg_text, - persist_user_timestamp=getattr(s, 'pending_started_at', None), + persist_user_timestamp=_persist_user_timestamp, ) # Only pass moa_config when a /moa override is actually active, so a # normal send never trips a TypeError on an older hermes-agent whose @@ -12343,9 +12340,17 @@ def _run_agent_streaming( _heal_context_messages, _heal_conversation_history_revision, ) = _refresh_context_and_revision_from_state_db() + _heal_persist_user_timestamp = getattr( + s, 'pending_started_at', None + ) + # The returned-error path already owns _agent_lock. + _register_pending_user_timestamp_identity( + agent.run_conversation, + s, + _heal_persist_user_timestamp, + ) _heal_kwargs = _build_run_conversation_kwargs( agent.run_conversation, - session=s, user_message=user_message, system_message=workspace_system_msg, conversation_history=_sanitize_messages_for_agent( @@ -12361,7 +12366,7 @@ def _run_agent_streaming( ), task_id=session_id, persist_user_message=msg_text, - persist_user_timestamp=getattr(s, 'pending_started_at', None), + persist_user_timestamp=_heal_persist_user_timestamp, ) if moa_config is not None: _heal_kwargs["moa_config"] = moa_config @@ -13695,9 +13700,17 @@ def _run_agent_streaming( _heal_context_messages, _heal_conversation_history_revision, ) = _refresh_context_and_revision_from_state_db() + with _agent_lock: + _heal_persist_user_timestamp = getattr( + s, 'pending_started_at', None + ) + _register_pending_user_timestamp_identity( + _heal_agent.run_conversation, + s, + _heal_persist_user_timestamp, + ) _heal_kwargs2 = _build_run_conversation_kwargs( _heal_agent.run_conversation, - session=s, user_message=user_message, system_message=workspace_system_msg, conversation_history=_sanitize_messages_for_agent( @@ -13713,7 +13726,7 @@ def _run_agent_streaming( ), task_id=session_id, persist_user_message=msg_text, - persist_user_timestamp=getattr(s, 'pending_started_at', None), + persist_user_timestamp=_heal_persist_user_timestamp, ) if moa_config is not None: _heal_kwargs2["moa_config"] = moa_config diff --git a/docs/rfcs/webui-run-state-consistency-contract.md b/docs/rfcs/webui-run-state-consistency-contract.md index a005c797..252756e8 100644 --- a/docs/rfcs/webui-run-state-consistency-contract.md +++ b/docs/rfcs/webui-run-state-consistency-contract.md @@ -187,10 +187,13 @@ and 5; it does not mark every run-state boundary implemented. 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 - conflicting nonempty `api_content`, preserve both versions for display and - replay without mutating either. Fill a missing payload from the other copy; - repeated reconciliation must remain bounded and idempotent. Preserve - genuinely distinct or ambiguous rows when matching settled projections. + conflicting nonempty `api_content`, preserve both versions for model-context + replay without mutating either. For visible display, a marked mirror may + share the existing sidecar bubble only when its valid durable row ID, exact + timestamp, and exact visible user content match; keep the sidecar-owned row + and its display metadata. Distinct row IDs, ambiguous or invalid identities, + and different visible user text remain separate. Fill a missing payload from + the other copy; repeated reconciliation must remain bounded and idempotent. Agent state.db alone cannot restore the original attachment if the WebUI sidecar is lost. Visible interim assistant progress must remain visible timeline content; a diff --git a/tests/test_compression_snapshot_revision.py b/tests/test_compression_snapshot_revision.py index 7d0b9411..be1c57a4 100644 --- a/tests/test_compression_snapshot_revision.py +++ b/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: diff --git a/tests/test_native_image_turn_display_context.py b/tests/test_native_image_turn_display_context.py index b31b1411..40402502 100644 --- a/tests/test_native_image_turn_display_context.py +++ b/tests/test_native_image_turn_display_context.py @@ -236,6 +236,10 @@ def test_get_session_keeps_pending_agent_projection_private_but_in_context( import api.models as models import api.session_ops from api import routes + from api.streaming import ( + _build_run_conversation_kwargs, + _register_pending_user_timestamp_identity, + ) timestamp = time.time() stream_id = "pending-webui-stream" @@ -304,11 +308,10 @@ def test_get_session_keeps_pending_agent_projection_private_but_in_context( ], ) - from api.streaming import _build_run_conversation_kwargs - + with api.config._get_session_agent_lock(session_id): + _register_pending_user_timestamp_identity(agent_run, session, timestamp) run_kwargs = _build_run_conversation_kwargs( agent_run, - session=session, user_message=prompt, system_message="system", conversation_history=[], @@ -434,7 +437,11 @@ def test_get_session_keeps_pending_agent_projection_private_but_in_context( def test_pending_row_identity_requires_timestamp_handoff_to_active_agent(): from api.models import _suppress_native_image_display_mirrors - from api.streaming import _build_run_conversation_kwargs + from api.streaming import ( + _build_run_conversation_kwargs, + _register_pending_user_timestamp_identity, + _get_session_agent_lock, + ) def modern_run(user_message, persist_user_timestamp=None): return None @@ -453,7 +460,6 @@ def test_pending_row_identity_requires_timestamp_handoff_to_active_agent(): pending_user_source="webui", ) kwargs = { - "session": session, "user_message": "prompt", "system_message": "system", "conversation_history": [], @@ -462,11 +468,15 @@ def test_pending_row_identity_requires_timestamp_handoff_to_active_agent(): "persist_user_message": "prompt", "persist_user_timestamp": timestamp, } + with _get_session_agent_lock("session"): + _register_pending_user_timestamp_identity(modern_run, session, timestamp) modern = _build_run_conversation_kwargs(modern_run, **kwargs) assert modern["persist_user_timestamp"] == timestamp assert session._webui_pending_user_timestamp_identity == ( "stream-identity", timestamp, ) + with _get_session_agent_lock("session"): + _register_pending_user_timestamp_identity(modern_run, session, timestamp) rebuilt = _build_run_conversation_kwargs(modern_run, **kwargs) assert rebuilt["persist_user_timestamp"] == timestamp assert session._webui_pending_user_timestamp_identity == ( @@ -474,6 +484,8 @@ def test_pending_row_identity_requires_timestamp_handoff_to_active_agent(): ) # Credential-heal fallback can replace the callable mid-turn. Proof from # the first invocation must not authorize rows from an older Agent. + with _get_session_agent_lock("session"): + _register_pending_user_timestamp_identity(legacy_run, session, timestamp) fallback = _build_run_conversation_kwargs(legacy_run, **kwargs) assert "persist_user_timestamp" not in fallback assert session._webui_pending_user_timestamp_identity is None @@ -484,14 +496,18 @@ def test_pending_row_identity_requires_timestamp_handoff_to_active_agent(): pending_user_message="prompt", pending_user_source="webui", ) - legacy = _build_run_conversation_kwargs( - legacy_run, **{**kwargs, "session": legacy_session} - ) + with _get_session_agent_lock("session"): + _register_pending_user_timestamp_identity( + legacy_run, legacy_session, timestamp + ) + legacy = _build_run_conversation_kwargs(legacy_run, **kwargs) assert "persist_user_timestamp" not in legacy assert legacy_session._webui_pending_user_timestamp_identity is None - opaque = _build_run_conversation_kwargs( - opaque_run, **{**kwargs, "session": legacy_session} - ) + with _get_session_agent_lock("session"): + _register_pending_user_timestamp_identity( + opaque_run, legacy_session, timestamp + ) + opaque = _build_run_conversation_kwargs(opaque_run, **kwargs) assert "persist_user_timestamp" in opaque assert legacy_session._webui_pending_user_timestamp_identity is None row = {"role": "user", "content": "Agent-only memory", "timestamp": timestamp} @@ -1204,12 +1220,18 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery( if message.get("_active_turn_token") == identity["token"] ) mirror = _durable_agent_content(context_user["content"]) + display_attachments = [{ + "name": "sidecar-owned.png", + "mime": "image/png", + "is_image": True, + }] sidecar_mirror_row = { "role": "user", "content": mirror, "timestamp": timestamp, "_state_db_row_id": 42, "api_content": sidecar_payload, + "attachments": display_attachments, } session.messages.append(dict(sidecar_mirror_row)) session.context_messages.append(dict(sidecar_mirror_row)) @@ -1243,10 +1265,9 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery( message for message in context if message.get("_state_db_row_id") == 42 ] - assert len(display_row_42) == 2 - assert {message["api_content"] for message in display_row_42} == { - sidecar_payload, state_payload, - } + assert len(display_row_42) == 1 + assert display_row_42[0]["api_content"] == sidecar_payload + assert display_row_42[0]["attachments"] == display_attachments assert len(context_row_42) == 2 assert {message["api_content"] for message in context_row_42} == { sidecar_payload, state_payload, @@ -1254,8 +1275,9 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery( public = public_session_projection({"messages": display})["messages"] public_mirrors = [message for message in public if message.get("content") == mirror] - assert len(public_mirrors) == 2 + assert len(public_mirrors) == 1 assert all("api_content" not in message for message in public_mirrors) + assert public_mirrors[0]["attachments"] == display_attachments display_image_turn = next( message for message in display if message.get("_active_turn_token") == identity["token"] @@ -1277,14 +1299,8 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery( message for message in session.messages if message.get("_state_db_row_id") == 42 ] - if first_public is None: - assert len(live_row_42) == 1 - assert live_row_42[0]["api_content"] == sidecar_payload - else: - assert len(live_row_42) == 2 - assert {message["api_content"] for message in live_row_42} == { - sidecar_payload, state_payload, - } + assert len(live_row_42) == 1 + assert live_row_42[0]["api_content"] == sidecar_payload replay_image_turn = next( message for message in replay if isinstance(message.get("content"), list) @@ -1306,6 +1322,106 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery( session.context_messages = context +def test_native_image_payload_conflict_with_distinct_visible_text_stays_separate(): + import api.models as models + + timestamp = 865.0 + session, identity, _ = _settle_image_turn(timestamp=timestamp, agent_row_id=41) + context_user = next( + message for message in session.context_messages + if message.get("_active_turn_token") == identity["token"] + ) + mirror = _durable_agent_content(context_user["content"]) + sidecar_row = { + "role": "user", + "content": mirror, + "timestamp": timestamp, + "_state_db_row_id": 42, + "api_content": "SIDECAR-PAYLOAD", + } + session.messages.append(sidecar_row) + state_row = { + "role": "user", + "content": "A separate visible submission", + "timestamp": timestamp, + "_state_db_row_id": 42, + "api_content": "STATE-DB-PAYLOAD", + "_webui_unmatched_native_image_mirror": True, + } + + display = models.reconciled_state_db_messages_for_session( + session, + state_messages=[state_row], + ) + same_row = [ + message for message in display + if message.get("_state_db_row_id") == 42 + ] + assert {message["content"] for message in same_row} == { + mirror, + "A separate visible submission", + } + + +def test_marked_native_image_conflict_keeps_ambiguous_sidecar_row_id_visible(): + import api.models as models + + timestamp = 867.0 + session, identity, _ = _settle_image_turn(timestamp=timestamp, agent_row_id=41) + context_user = next( + message for message in session.context_messages + if message.get("_active_turn_token") == identity["token"] + ) + mirror = _durable_agent_content(context_user["content"]) + sidecar_rows = [ + { + "role": "user", + "content": mirror, + "timestamp": timestamp, + "_state_db_row_id": 42, + "api_content": "SIDECAR-PAYLOAD-ONE", + "attachments": [{"name": "first.png"}], + }, + { + "role": "user", + "content": mirror, + "timestamp": timestamp, + "_state_db_row_id": 42, + "api_content": "SIDECAR-PAYLOAD-TWO", + "attachments": [{"name": "second.png"}], + }, + ] + session.messages.extend(sidecar_rows) + state_row = { + "role": "user", + "content": mirror, + "timestamp": timestamp, + "_state_db_row_id": 42, + "api_content": "STATE-DB-PAYLOAD", + } + marked = models._suppress_native_image_display_mirrors(session, [state_row]) + assert marked[0]["_webui_unmatched_native_image_mirror"] is True + + display = models.reconciled_state_db_messages_for_session( + session, + state_messages=[state_row], + ) + same_row = [ + message for message in display + if message.get("_state_db_row_id") == 42 + ] + assert len(same_row) == 3 + assert {message["api_content"] for message in same_row} == { + "SIDECAR-PAYLOAD-ONE", + "SIDECAR-PAYLOAD-TWO", + "STATE-DB-PAYLOAD", + } + assert [message.get("attachments") for message in same_row[:2]] == [ + [{"name": "first.png"}], + [{"name": "second.png"}], + ] + + def test_marked_native_image_rows_with_distinct_ids_survive_recovery(): import api.models as models