diff --git a/api/models.py b/api/models.py index 10744e17..80098e71 100644 --- a/api/models.py +++ b/api/models.py @@ -1268,6 +1268,36 @@ def _strip_sidebar_heavy_metadata(row: dict) -> dict: return row +def _validated_webui_pending_user_timestamp_identity(session, value): + if not isinstance(value, (tuple, list)) or len(value) != 2: + return None + stream_id = getattr(session, 'active_stream_id', None) + source = str(getattr(session, 'pending_user_source', '') or '').strip().lower() + if ( + not isinstance(stream_id, str) + or not stream_id + or value[0] != stream_id + or not isinstance(getattr(session, 'pending_user_message', None), str) + or not getattr(session, 'pending_user_message', None) + or source not in {'webui', 'fork'} + ): + return None + pending_timestamp, pending_valid = _message_exact_timestamp_details( + {'timestamp': getattr(session, 'pending_started_at', None)} + ) + identity_timestamp, identity_valid = _message_exact_timestamp_details( + {'timestamp': value[1]} + ) + if ( + not pending_valid + or not identity_valid + or pending_timestamp is None + or pending_timestamp != identity_timestamp + ): + return None + return (stream_id, pending_timestamp) + + class Session: def __init__(self, session_id: str=None, title: str='Untitled', workspace=str(DEFAULT_WORKSPACE), created_workspace=None, @@ -1370,6 +1400,11 @@ class Session: self.pending_attachments = pending_attachments or [] self.pending_started_at = pending_started_at self.pending_user_source = pending_user_source + self._webui_pending_user_timestamp_identity = ( + _validated_webui_pending_user_timestamp_identity( + self, kwargs.get('_webui_pending_user_timestamp_identity') + ) + ) self.context_messages = context_messages if isinstance(context_messages, list) else [] self.compression_anchor_visible_idx = compression_anchor_visible_idx self.compression_anchor_message_key = compression_anchor_message_key @@ -1467,6 +1502,11 @@ class Session: ) if touch_updated_at: self.updated_at = time.time() + self._webui_pending_user_timestamp_identity = ( + _validated_webui_pending_user_timestamp_identity( + self, getattr(self, '_webui_pending_user_timestamp_identity', None) + ) + ) # Write metadata fields first so load_metadata_only() can read them # without parsing the full messages array (which may be 400KB+). # Fields are listed in the order they should appear in the JSON file. @@ -1477,6 +1517,7 @@ class Session: 'cache_read_tokens', 'cache_write_tokens', 'personality', 'active_stream_id', 'pending_user_message', 'pending_attachments', 'pending_started_at', 'pending_user_source', + '_webui_pending_user_timestamp_identity', 'compression_anchor_visible_idx', 'compression_anchor_message_key', 'compression_anchor_summary', 'pre_compression_snapshot', 'context_engine', 'compression_anchor_engine', 'compression_anchor_mode', @@ -9237,8 +9278,32 @@ def _suppress_native_image_display_mirrors( state_messages, *, suppress_api_content=True, + suppress_pending_turn=True, ): - """Drop exact Agent image projections backed by one token-paired WebUI turn.""" + """Drop proven pending Agent rows and native-image mirrors from display.""" + if not state_messages: + return state_messages + + if suppress_pending_turn: + identity = _validated_webui_pending_user_timestamp_identity( + session, + getattr(session, "_webui_pending_user_timestamp_identity", None), + ) + if identity is not None: + # A matching timestamp is an Agent handoff identity only because + # the live worker recorded that this exact value was passed via + # persist_user_timestamp. If rows collide on it, hide the whole + # ambiguous bucket from display; model context retains every row. + pending_timestamp = identity[1] + state_messages = [ + message for message in state_messages + if not ( + isinstance(message, dict) + and str(message.get("role") or "").lower() == "user" + and _message_exact_timestamp(message) == pending_timestamp + ) + ] + display_messages = getattr(session, "messages", None) or [] context_messages = getattr(session, "context_messages", None) or [] if not display_messages or not context_messages or not state_messages: @@ -11012,13 +11077,14 @@ def _merge_session_messages_append_only_impl( and incoming_api_content is not None and existing_api_content != incoming_api_content ): - # The unique durable row ID makes state.db authoritative - # for this marked native-image mirror's provider payload. - existing["api_content"] = incoming_api_content - elif existing_api_content is None and incoming_api_content is not None: - _copy_api_content_sidecar(existing, msg) - _merge_session_display_metadata(existing, msg) - continue + # Row identity does not establish which provider payload is + # newer. Fall through so both conflicting versions survive. + pass + else: + if existing_api_content is None and incoming_api_content is not None: + _copy_api_content_sidecar(existing, msg) + _merge_session_display_metadata(existing, msg) + continue if dedup_key in seen_dedup_keys: duplicate = merged_by_dedup_key.get(dedup_key) duplicate_row_id, duplicate_row_id_valid = ( @@ -11418,6 +11484,7 @@ def reconciled_state_db_messages_for_session( session, state_messages, suppress_api_content=not using_context_messages, + suppress_pending_turn=not prefer_context, ) if prefer_context and local_messages: if using_context_messages: diff --git a/api/routes.py b/api/routes.py index 4c8b9fc5..55a16815 100644 --- a/api/routes.py +++ b/api/routes.py @@ -13479,7 +13479,7 @@ def _handle_session_get(handler, parsed) -> bool: "tool_calls": _session_tool_calls, "active_stream_id": getattr(s, "active_stream_id", None), "pending_user_message": getattr(s, "pending_user_message", None), - "pending_attachments": getattr(s, "pending_attachments", []) if load_messages else [], + "pending_attachments": getattr(s, "pending_attachments", []) if (load_messages or getattr(s, "pending_user_message", None)) else [], "pending_started_at": getattr(s, "pending_started_at", None), "pending_user_source": getattr(s, "pending_user_source", None), "context_length": _persisted_cl, @@ -22984,6 +22984,7 @@ def _prepare_chat_start_session_for_stream( s.pending_attachments = attachments s.pending_started_at = started_at if started_at is not None else time.time() s.pending_user_source = effective_source + s._webui_pending_user_timestamp_identity = None if retained_user is not None: from api.process_event_utils import build_active_turn_token diff --git a/api/streaming.py b/api/streaming.py index 2804150a..f4df3f2f 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -79,6 +79,7 @@ from api.models import ( _is_empty_partial_activity_message, _message_exact_timestamp_details, _message_private_identity_compatible, + _validated_webui_pending_user_timestamp_identity, _evict_sessions_over_cap, clear_process_wakeup_pause, get_state_db_session_messages, @@ -9139,6 +9140,7 @@ def _add_supported_run_conversation_kwarg(callable_obj, kwargs, name, value): def _build_run_conversation_kwargs( callable_obj, *, + session=None, user_message, system_message, conversation_history, @@ -9172,6 +9174,45 @@ 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 @@ -11724,6 +11765,7 @@ def _run_agent_streaming( _persistent_state_before = _persistent_state_snapshot(_profile_home) _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( @@ -12303,6 +12345,7 @@ def _run_agent_streaming( ) = _refresh_context_and_revision_from_state_db() _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( @@ -13654,6 +13697,7 @@ def _run_agent_streaming( ) = _refresh_context_and_revision_from_state_db() _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( diff --git a/docs/rfcs/webui-run-state-consistency-contract.md b/docs/rfcs/webui-run-state-consistency-contract.md index 5faf3808..a005c797 100644 --- a/docs/rfcs/webui-run-state-consistency-contract.md +++ b/docs/rfcs/webui-run-state-consistency-contract.md @@ -174,13 +174,25 @@ and 5; it does not mark every run-state boundary implemented. evidence. A WebUI-submitted native-image turn has a separate display owner: keep its exact submitted text and attachment in the visible session row, while the Agent's expanded multipart row remains in `context_messages` for - model replay. Match its scalar state.db projection only with trusted turn - and durable-row identity, never the marker alone. For a uniquely identified - unmatched state.db row already present in the sidecar, conflicting provider - sidecars are two versions of that *one* durable row: show it once and use - the state.db provider payload for replay. Preserve genuinely distinct or - ambiguous rows. Agent state.db alone cannot restore the original attachment - if the WebUI sidecar is lost. + model replay. While the turn is active, the WebUI may hide an Agent user row + from display only after its worker confirms that the active stream's exact + `pending_started_at` value was passed to the Agent as + `persist_user_timestamp`; persist that private proof with the session and + validate it against the pending stream, source, and timestamp after reload. + Never include the proof in public session payloads. This applies to + 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. + 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. + 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 compact Activity disclosure may summarize adjacent tool/debug detail, but it must not be the only place where the user can see emitted progress text. diff --git a/tests/test_native_image_turn_display_context.py b/tests/test_native_image_turn_display_context.py index 6febd954..b31b1411 100644 --- a/tests/test_native_image_turn_display_context.py +++ b/tests/test_native_image_turn_display_context.py @@ -1,7 +1,11 @@ """Native-image turns keep model context private from the visible transcript.""" import json +import shutil import sqlite3 +import subprocess +import time +from pathlib import Path from types import SimpleNamespace import pytest @@ -25,6 +29,21 @@ RECALL_NOTE = "Recall note: the sample is neutral." PLUGIN_NOTE = "Pre-call note: summarize visible details only." +def _js_function_source(src, name): + start = src.find(f"function {name}(") + assert start != -1, f"{name} not found" + brace = src.find("{", start) + depth = 0 + for index in range(brace, len(src)): + if src[index] == "{": + depth += 1 + elif src[index] == "}": + depth -= 1 + if depth == 0: + return src[start:index + 1] + raise AssertionError(f"{name} body unterminated") + + def _native_user_content(text, image_url, extra_text=()): parts = [{"type": "text", "text": f"[Workspace::v1: /fixture]\n{text}"}] if image_url: @@ -208,6 +227,368 @@ def _write_state_db(path, session_id, rows): ) +@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( + monkeypatch, tmp_path, msg_limit, attachment_mode, +): + import api.config + import api.models as models + import api.session_ops + from api import routes + + timestamp = time.time() + stream_id = "pending-webui-stream" + session_id = f"pending-display-{attachment_mode}-{'full' if msg_limit is None else 'limited'}" + prompt = "Recall note: I authored this literal text." + attachments = [ + {"name": "sample.png" if attachment_mode == "native_image" else "sample.txt", + "mime": "image/png" if attachment_mode == "native_image" else "text/plain", + "is_image": attachment_mode == "native_image"} + ] + if attachment_mode == "native_image": + agent_content = _native_user_content( + prompt, IMAGE_A, (RECALL_NOTE, PLUGIN_NOTE), + ) + stored_content = "\x00json:" + json.dumps(agent_content) + api_content = json.dumps(agent_content) + else: + agent_content = f"{prompt}\n\n{RECALL_NOTE}\n\n{PLUGIN_NOTE}" + stored_content = agent_content + api_content = None + + db_path = tmp_path / "state.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id TEXT, role TEXT, " + "content TEXT, timestamp REAL, active INTEGER DEFAULT 1, api_content TEXT)" + ) + + session_dir = tmp_path / "sessions" + session_dir.mkdir() + 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=[], + context_messages=[], + active_stream_id=stream_id, + pending_user_message=prompt, + pending_attachments=attachments, + pending_started_at=timestamp, + pending_user_source="webui", + source_tag="webui", + session_source="webui", + ) + session.save(skip_index=True) + + def agent_run(user_message, persist_user_timestamp=None, **_kwargs): + assert user_message == prompt + assert persist_user_timestamp == timestamp + persisted = models.Session.load(session_id) + assert persisted._webui_pending_user_timestamp_identity == (stream_id, timestamp) + with sqlite3.connect(db_path) as conn: + conn.executemany( + "INSERT INTO messages (id, session_id, role, content, timestamp, api_content) " + "VALUES (?, ?, ?, ?, ?, ?)", + [ + (1, session_id, "user", "A preserved prior turn", timestamp - 2, None), + (2, session_id, "user", stored_content, timestamp, api_content), + (3, session_id, "user", "A distinct later user row", timestamp + 2, None), + (4, "other-profile-session", "user", stored_content, timestamp, api_content), + ], + ) + + from api.streaming import _build_run_conversation_kwargs + + run_kwargs = _build_run_conversation_kwargs( + agent_run, + session=session, + user_message=prompt, + system_message="system", + conversation_history=[], + conversation_history_revision=None, + task_id=session_id, + persist_user_message=prompt, + persist_user_timestamp=timestamp, + ) + agent_run(**run_kwargs) + # A cold Session.load restores only the proof persisted before the Agent + # wrote its pending user row. + session = models.Session.load(session_id) + assert session._webui_pending_user_timestamp_identity == (stream_id, timestamp) + + response = {} + # Simulate a WebUI process restart: no stream or worker remains, but the + # pending timestamp is inside the existing stale-repair grace window. + monkeypatch.setattr(routes, "STREAMS", {}) + monkeypatch.setattr(api.config, "ACTIVE_RUNS", {}) + with models.LOCK: + models.SESSIONS.pop(session_id, None) + monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *_args: True) + monkeypatch.setattr(routes, "_active_stream_ids", lambda: set()) + monkeypatch.setattr(routes, "_lookup_cli_session_metadata", lambda *_args: {}) + monkeypatch.setattr(routes, "find_run_summary", lambda *_args: 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) + 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"] + assert public_session["active_stream_id"] == stream_id + assert public_session["pending_user_message"] == prompt + assert public_session["pending_attachments"] == attachments + assert prompt in json.dumps(public_session) + assert "_webui_pending_user_timestamp_identity" not in json.dumps(public_session) + assert RECALL_NOTE not in json.dumps(public_session) + assert PLUGIN_NOTE not in json.dumps(public_session) + assert all( + RECALL_NOTE not in json.dumps(message) + and PLUGIN_NOTE not in json.dumps(message) + for message in public_messages + ) + assert any(message.get("content") == "A preserved prior turn" for message in public_messages) + assert any(message.get("content") == "A distinct later user row" for message in public_messages) + assert not any(message.get("session_id") == "other-profile-session" for message in public_messages) + + state_rows = models.get_state_db_session_messages(session_id) + context = models.reconciled_state_db_messages_for_session( + session, + prefer_context=True, + state_messages=state_rows, + ) + active_context_rows = [ + message for message in context + if message.get("role") == "user" and message.get("timestamp") == timestamp + ] + assert len(active_context_rows) == 1 + assert RECALL_NOTE in json.dumps(active_context_rows[0]["content"]) + assert PLUGIN_NOTE in json.dumps(active_context_rows[0]["content"]) + if attachment_mode == "native_image": + assert any( + part.get("type") == "image_url" + for part in active_context_rows[0]["content"] + ) + replay_rows = _sanitize_messages_for_agent(context) + assert sum( + RECALL_NOTE in json.dumps(message) + and PLUGIN_NOTE in json.dumps(message) + for message in replay_rows + ) == 1 + assert RECALL_NOTE in json.dumps(replay_rows) + assert PLUGIN_NOTE in json.dumps(replay_rows) + assert not session.messages and not session.context_messages + + # A metadata-only poll may arrive after the full load. It must not erase + # the pending attachment from the browser's session state. + response.clear() + routes._handle_session_get( + None, + SimpleNamespace( + path="/api/session", + query=f"session_id={session_id}&resolve_model=0&messages=0", + ), + ) + metadata = response["payload"]["session"] + assert metadata["pending_user_message"] == prompt + assert metadata["pending_attachments"] == attachments + assert RECALL_NOTE not in json.dumps(metadata) + assert PLUGIN_NOTE not in json.dumps(metadata) + + next_session = models.Session.load(session_id) + monkeypatch.setattr(routes, "register_session_writeback_owner", lambda *_args: None) + monkeypatch.setattr(routes, "get_webui_session_save_mode", lambda: "deferred") + routes._prepare_chat_start_session_for_stream( + next_session, + msg="A separate next turn", + attachments=[], + workspace="/fixture", + model="fixture-model", + model_provider=None, + stream_id="next-webui-stream", + started_at=timestamp + 1, + ) + assert next_session._webui_pending_user_timestamp_identity is None + assert models.Session.load(session_id)._webui_pending_user_timestamp_identity is None + + +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 + + def modern_run(user_message, persist_user_timestamp=None): + return None + + def legacy_run(user_message): + return None + + def opaque_run(user_message, **kwargs): + return None + + timestamp = 1700000000.125 + session = SimpleNamespace( + active_stream_id="stream-identity", + pending_started_at=timestamp, + pending_user_message="prompt", + pending_user_source="webui", + ) + kwargs = { + "session": session, + "user_message": "prompt", + "system_message": "system", + "conversation_history": [], + "conversation_history_revision": None, + "task_id": "session", + "persist_user_message": "prompt", + "persist_user_timestamp": 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, + ) + rebuilt = _build_run_conversation_kwargs(modern_run, **kwargs) + assert rebuilt["persist_user_timestamp"] == timestamp + assert session._webui_pending_user_timestamp_identity == ( + "stream-identity", timestamp, + ) + # Credential-heal fallback can replace the callable mid-turn. Proof from + # the first invocation must not authorize rows from an older Agent. + fallback = _build_run_conversation_kwargs(legacy_run, **kwargs) + assert "persist_user_timestamp" not in fallback + assert session._webui_pending_user_timestamp_identity is None + + legacy_session = SimpleNamespace( + active_stream_id="stream-identity", + pending_started_at=timestamp, + pending_user_message="prompt", + pending_user_source="webui", + ) + legacy = _build_run_conversation_kwargs( + legacy_run, **{**kwargs, "session": legacy_session} + ) + 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} + ) + 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} + assert _suppress_native_image_display_mirrors(legacy_session, [row]) == [row] + + for field, value in ( + ("active_stream_id", "another-stream"), + ("pending_user_source", "cli"), + ("pending_started_at", timestamp + 1), + ("pending_user_message", None), + ): + mismatched = SimpleNamespace( + active_stream_id="stream-identity", + pending_started_at=timestamp, + pending_user_message="prompt", + pending_user_source="webui", + _webui_pending_user_timestamp_identity=("stream-identity", timestamp), + ) + setattr(mismatched, field, value) + assert _suppress_native_image_display_mirrors(mismatched, [row]) == [row] + + +@pytest.mark.skipif(shutil.which("node") is None, reason="node not on PATH") +def test_clean_pending_get_payload_renders_submitted_prompt_and_attachment(): + root = Path(__file__).resolve().parents[1] + ui_js = (root / "static" / "ui.js").read_text(encoding="utf-8") + sessions_js = (root / "static" / "sessions.js").read_text(encoding="utf-8") + helpers = "\n".join( + [ + *[ + _js_function_source(sessions_js, name) + for name in ( + "_messageComparableText", + "_stripAttachedFilesMarker", + "_stripForcedSkillEnvelope", + "_normalizeUserTranscriptText", + "_sameTranscriptMessage", + "_currentTailUserMessage", + "_hasCurrentTailUserDuplicate", + "_mergePendingSessionMessage", + ) + ], + *[ + _js_function_source(ui_js, name) + for name in ( + "_pendingCurrentTailUserMessage", + "_messageTimestampSeconds", + "_activeTurnTokenMatches", + "_pendingActiveTurnUserMessage", + "getPendingSessionMessage", + ) + ], + ] + ) + payload = { + "session_id": "pending-display-session", + "active_stream_id": "pending-webui-stream", + "pending_started_at": 1700000000.125, + "pending_user_source": "webui", + "pending_user_message": "Recall note: I authored this literal text.", + "pending_attachments": [{"name": "literal-note.txt", "mime": "text/plain"}], + "messages": [], + } + assert "_mergePendingSessionMessage(S.session,S.messages)" in sessions_js + assert "_mergePendingSessionMessage(data.session, S.messages)" in ui_js + assert "await refreshSession();" in ui_js + script = f""" +{helpers} +const session={json.dumps(payload)}; +const messages=session.messages; +const inserted=_mergePendingSessionMessage(session,messages); +const duplicate=_mergePendingSessionMessage(session,messages); +process.stdout.write(JSON.stringify({{inserted,duplicate,messages}})); +""" + completed = subprocess.run( + [shutil.which("node"), "-e", script], + cwd=root, + text=True, + capture_output=True, + check=False, + ) + assert completed.returncode == 0, completed.stderr or completed.stdout + result = json.loads(completed.stdout) + assert result["inserted"] is True + assert result["duplicate"] is False + assert result["messages"] == [ + { + "role": "user", + "content": "Recall note: I authored this literal text.", + "attachments": [{"name": "literal-note.txt", "mime": "text/plain"}], + "_ts": 1700000000.125, + "_pending": True, + "_source": "webui", + } + ] + + def test_settlement_reload_and_next_turn_keep_one_clean_bubble_and_rich_context( monkeypatch, tmp_path ): @@ -801,7 +1182,16 @@ def test_marked_native_image_mirror_repairs_malformed_sidecar_once(): assert replay_rows[0]["api_content"] == api_content -def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery(): +@pytest.mark.parametrize( + ("sidecar_payload", "state_payload"), + [ + ("OLDER-SIDECAR-BYTES", "NEWER-DB-BYTES"), + ("NEWER-SIDECAR-BYTES", "OLDER-DB-BYTES"), + ], +) +def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery( + sidecar_payload, state_payload, +): import api.models as models timestamp = 860.0 @@ -814,19 +1204,21 @@ 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"]) - session.messages.append({ + sidecar_mirror_row = { "role": "user", "content": mirror, "timestamp": timestamp, "_state_db_row_id": 42, - "api_content": "OLD-PROVIDER-BYTES", - }) + "api_content": sidecar_payload, + } + session.messages.append(dict(sidecar_mirror_row)) + session.context_messages.append(dict(sidecar_mirror_row)) state_row = { "role": "user", "content": mirror, "timestamp": timestamp, "_state_db_row_id": 42, - "api_content": "NEW-PROVIDER-BYTES", + "api_content": state_payload, } marked = models._suppress_native_image_display_mirrors(session, [state_row]) assert marked[0]["_webui_unmatched_native_image_mirror"] is True @@ -851,15 +1243,19 @@ 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) == 1 - assert display_row_42[0]["api_content"] == "NEW-PROVIDER-BYTES" - assert len(context_row_42) == 1 - assert context_row_42[0]["api_content"] == "NEW-PROVIDER-BYTES" + assert len(display_row_42) == 2 + assert {message["api_content"] for message in display_row_42} == { + sidecar_payload, state_payload, + } + assert len(context_row_42) == 2 + assert {message["api_content"] for message in context_row_42} == { + sidecar_payload, state_payload, + } public = public_session_projection({"messages": display})["messages"] public_mirrors = [message for message in public if message.get("content") == mirror] - assert len(public_mirrors) == 1 - assert "api_content" not in public_mirrors[0] + assert len(public_mirrors) == 2 + assert all("api_content" not in message for message in public_mirrors) display_image_turn = next( message for message in display if message.get("_active_turn_token") == identity["token"] @@ -873,8 +1269,22 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery(): replay = _sanitize_messages_for_agent(context) replay_row_42 = [message for message in replay if message.get("content") == mirror] - assert len(replay_row_42) == 1 - assert replay_row_42[0]["api_content"] == "NEW-PROVIDER-BYTES" + assert len(replay_row_42) == 2 + assert {message["api_content"] for message in replay_row_42} == { + sidecar_payload, state_payload, + } + live_row_42 = [ + 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, + } replay_image_turn = next( message for message in replay if isinstance(message.get("content"), list)