diff --git a/api/models.py b/api/models.py index b14e4f79..bd63649e 100644 --- a/api/models.py +++ b/api/models.py @@ -4033,6 +4033,68 @@ def _sync_sidecar_from_state_db_if_newer(session) -> bool: ) if not state_messages: return False + + # The Agent row is intentionally hidden from display while the saved + # handoff proof is active. Materialize the exact WebUI-owned prompt + # before reconciling that row away, so clearing pending state cannot + # erase the only durable copy of the submitted text/attachments. + pending_identity = _validated_webui_pending_user_timestamp_identity( + locked, + getattr(locked, '_webui_pending_user_timestamp_identity', None), + ) + # When a legacy sidecar has no explicit context_messages, the shared + # context reconciler falls back to messages. Capture that model-facing + # merge before adding the display-only WebUI owner row below. + merged_context = ( + reconciled_state_db_messages_for_session( + locked, + prefer_context=True, + state_messages=state_messages, + ) + if pending_identity is not None + else None + ) + if pending_identity is not None: + pending_timestamp = pending_identity[1] + pending_text = locked.pending_user_message + pending_source = getattr(locked, 'pending_user_source', None) or 'webui' + pending_attachments = list(getattr(locked, 'pending_attachments', None) or []) + pending_row = { + 'role': 'user', + 'content': pending_text, + 'timestamp': pending_timestamp, + } + stamp_message_source(pending_row, pending_source) + if str(pending_source or '').strip().lower() == 'fork': + pending_row['_fork_child_turn'] = locked.session_id + if pending_attachments: + pending_row['attachments'] = pending_attachments + + existing_pending = next(( + message for message in locked_messages + if isinstance(message, dict) + and ( + _message_exact_timestamp_details(message) + == (pending_timestamp, True) + and (message.get('_source') or 'webui') == pending_source + and _message_matches_pending_text(message, pending_text) + ) + ), None) + if existing_pending is not None: + # A prior eager checkpoint already owns this exact proved turn; + # refresh it from the still-authoritative pending fields. + existing_pending.update(pending_row) + if pending_attachments: + existing_pending['attachments'] = pending_attachments + else: + existing_pending.pop('attachments', None) + else: + # The proof makes this submitted row authoritative even if a + # clock adjustment places it before the surviving sidecar tail. + if not _insert_state_message_chronologically(locked_messages, pending_row): + locked_messages.insert(0, pending_row) + locked.messages = locked_messages + merged_messages = reconciled_state_db_messages_for_session( locked, state_messages=state_messages, @@ -4043,11 +4105,12 @@ def _sync_sidecar_from_state_db_if_newer(session) -> bool: # recover — leave the sidecar untouched rather than rewriting in place. if len(merged_messages) <= locked_count: return False - merged_context = reconciled_state_db_messages_for_session( - locked, - prefer_context=True, - state_messages=state_messages, - ) + if merged_context is None: + merged_context = reconciled_state_db_messages_for_session( + locked, + prefer_context=True, + state_messages=state_messages, + ) # Mutate + persist the freshly-loaded, locked object. Because we hold the # lock and reloaded under it, this save cannot clobber a concurrent @@ -10787,12 +10850,15 @@ def _project_native_image_payload_conflicts_for_display( 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 + if conflict is not None and _message_identity_compatible(message, conflict[0]): + incoming, sidecar_owner = conflict + is_sidecar_owner = ( + _message_identity_compatible(message, sidecar_owner) + and _session_message_api_content_key(message) + == _session_message_api_content_key(sidecar_owner) + ) + if not is_sidecar_owner: + continue visible_messages.append(message) return visible_messages diff --git a/api/routes.py b/api/routes.py index 85551061..71ef658e 100644 --- a/api/routes.py +++ b/api/routes.py @@ -10065,7 +10065,35 @@ def _merged_session_messages_for_display(session, cli_messages=None) -> list: -def _merged_webui_lineage_messages_for_display(session, messages=None) -> list: +_LINEAGE_PARENT_SESSION_UNSET = object() + + +def _webui_lineage_parent_session_for_display(session): + """Load the immediate parent only for display-eligible continuations.""" + parent_id = str(getattr(session, "parent_session_id", "") or "").strip() + if not parent_id: + return None + if ( + str(getattr(session, "compression_recovery_source_session_id", "") or "").strip() + and str(getattr(session, "compression_recovery_action", "") or "").strip() + ): + return None + source = str(getattr(session, "session_source", "") or "").strip().lower() + relationship = str(getattr(session, "relationship_type", "") or "").strip().lower() + if source == "fork" or relationship == "child_session": + return None + try: + return get_session(parent_id, metadata_only=False) + except Exception: + return None + + +def _merged_webui_lineage_messages_for_display( + session, + messages=None, + *, + parent_session=_LINEAGE_PARENT_SESSION_UNSET, +) -> list: """Include immediate parent-only rows when a WebUI continuation sidecar is partial. Compression/continuation sessions should render as one conversation. Most @@ -10076,23 +10104,9 @@ def _merged_webui_lineage_messages_for_display(session, messages=None) -> list: subset of their parent. """ primary_messages = list(messages if messages is not None else (getattr(session, "messages", []) or [])) - parent_id = str(getattr(session, "parent_session_id", "") or "").strip() - if not parent_id: - return primary_messages - if ( - str(getattr(session, "compression_recovery_source_session_id", "") or "").strip() - and str(getattr(session, "compression_recovery_action", "") or "").strip() - ): - return primary_messages - source = str(getattr(session, "session_source", "") or "").strip().lower() - relationship = str(getattr(session, "relationship_type", "") or "").strip().lower() - if source == "fork" or relationship == "child_session": - return primary_messages - try: - parent = get_session(parent_id, metadata_only=False) - except Exception: - return primary_messages - parent_messages = list(getattr(parent, "messages", []) or []) + if parent_session is _LINEAGE_PARENT_SESSION_UNSET: + parent_session = _webui_lineage_parent_session_for_display(session) + parent_messages = list(getattr(parent_session, "messages", []) or []) if not parent_messages: return primary_messages if _messages_start_with_visible_prefix(primary_messages, parent_messages): @@ -10572,7 +10586,6 @@ from api.models import ( new_session, all_sessions, title_from, - _write_session_index, SESSION_INDEX_FILE, _active_state_db_path, load_projects, @@ -13333,9 +13346,11 @@ def _handle_session_get(handler, parsed) -> bool: state_db_messages, ) sidecar_messages = _webui_sidecar_lineage_messages_for_display(s) + lineage_parent = _webui_lineage_parent_session_for_display(s) projection_sidecar_messages = _merged_webui_lineage_messages_for_display( s, sidecar_messages, + parent_session=lineage_parent, ) _all_msgs = merge_session_messages_append_only( sidecar_messages, @@ -13343,7 +13358,11 @@ def _handle_session_get(handler, parsed) -> bool: truncation_watermark=getattr(s, "truncation_watermark", None), truncation_boundary=getattr(s, "truncation_boundary", None), ) - _all_msgs = _merged_webui_lineage_messages_for_display(s, _all_msgs) + _all_msgs = _merged_webui_lineage_messages_for_display( + s, + _all_msgs, + parent_session=lineage_parent, + ) _all_msgs = _project_native_image_payload_conflicts_for_display( projection_sidecar_messages, state_db_messages, diff --git a/docs/rfcs/webui-run-state-consistency-contract.md b/docs/rfcs/webui-run-state-consistency-contract.md index 252756e8..d7e0ffcf 100644 --- a/docs/rfcs/webui-run-state-consistency-contract.md +++ b/docs/rfcs/webui-run-state-consistency-contract.md @@ -183,7 +183,12 @@ and 5; it does not mark every run-state boundary implemented. native-image and scalar text-attachment rows, and never changes model context. If multiple user rows share that timestamp, omit the whole ambiguous display bucket until the turn settles; keep all rows in model - context. Do not identify the row by its text. + 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. 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 diff --git a/tests/test_native_image_turn_display_context.py b/tests/test_native_image_turn_display_context.py index b3d44166..cf3de3a0 100644 --- a/tests/test_native_image_turn_display_context.py +++ b/tests/test_native_image_turn_display_context.py @@ -433,11 +433,23 @@ def test_get_session_projects_parent_only_payload_conflict_without_losing_parent response.update(payload=payload, status=status) or payload ), ) + original_get_session = routes.get_session + parent_loads = [] + + def get_session_with_fresh_parent(requested_id, metadata_only=False): + if requested_id == parent_id: + parent_session = models.Session.load(parent_id) + parent_loads.append(parent_session) + return parent_session + return original_get_session(requested_id, metadata_only=metadata_only) + + monkeypatch.setattr(routes, "get_session", get_session_with_fresh_parent) routes._handle_session_get( None, SimpleNamespace(path="/api/session", query=f"session_id={session_id}&resolve_model=0"), ) + assert len(parent_loads) == 1 assert response["status"] == 200 public_messages = response["payload"]["session"]["messages"] mirror_rows = [message for message in public_messages if message.get("content") == mirror] @@ -452,6 +464,9 @@ def test_get_session_projects_parent_only_payload_conflict_without_losing_parent prefer_context=True, state_messages=state_rows, ) + assert "Unrelated parent-only row" not in [ + message.get("content") for message in replay_context + ] replay_rows = [ message for message in _sanitize_messages_for_agent(replay_context) if message.get("content") == mirror diff --git a/tests/test_session_lost_response_regression.py b/tests/test_session_lost_response_regression.py index f37da008..bbbc2dd2 100644 --- a/tests/test_session_lost_response_regression.py +++ b/tests/test_session_lost_response_regression.py @@ -818,6 +818,83 @@ 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_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 +961,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 +971,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 +1007,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):