From d85979cd690b79330ca8aebe76cd2a8bf897dcdd Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 18 Jun 2026 15:47:06 -0400 Subject: [PATCH] fix(#4373): mark process-wakeup turns as synthetic and filter from visible transcript --- api/models.py | 7 +- api/routes.py | 11 ++- static/ui.js | 2 + tests/test_process_wakeup_synthetic.py | 118 +++++++++++++++++++++++++ 4 files changed, 136 insertions(+), 2 deletions(-) create mode 100644 tests/test_process_wakeup_synthetic.py diff --git a/api/models.py b/api/models.py index 61c5429c..a6994908 100644 --- a/api/models.py +++ b/api/models.py @@ -416,6 +416,9 @@ def _append_recovered_pending_turn(session, *, timestamp: int | None = None) -> 'timestamp': recovered_ts, '_recovered': True, } + pending_source = getattr(session, 'pending_user_source', None) + if pending_source and pending_source != 'webui': + recovered['_source'] = pending_source if session.pending_attachments: recovered['attachments'] = list(session.pending_attachments) session.messages.append(recovered) @@ -621,6 +624,7 @@ class Session: pending_user_message: str=None, pending_attachments=None, pending_started_at=None, + pending_user_source: str=None, context_messages=None, compression_anchor_visible_idx=None, compression_anchor_message_key=None, @@ -668,6 +672,7 @@ class Session: self.pending_user_message = pending_user_message self.pending_attachments = pending_attachments or [] self.pending_started_at = pending_started_at + self.pending_user_source = pending_user_source 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 @@ -743,7 +748,7 @@ class Session: 'input_tokens', 'output_tokens', 'estimated_cost', 'cache_read_tokens', 'cache_write_tokens', 'personality', 'active_stream_id', - 'pending_user_message', 'pending_attachments', 'pending_started_at', + 'pending_user_message', 'pending_attachments', 'pending_started_at', 'pending_user_source', 'compression_anchor_visible_idx', 'compression_anchor_message_key', 'compression_anchor_summary', 'pre_compression_snapshot', 'context_engine', 'compression_anchor_engine', 'compression_anchor_mode', diff --git a/api/routes.py b/api/routes.py index 2001bb26..a6af6c37 100644 --- a/api/routes.py +++ b/api/routes.py @@ -13908,7 +13908,7 @@ def _handle_background(handler, body): return j(handler, {"task_id": task_id, "stream_id": stream_id, "session_id": bg.session_id}) -def _checkpoint_user_message_for_eager_session_save(s, msg: str, attachments, started_at: float | None) -> None: +def _checkpoint_user_message_for_eager_session_save(s, msg: str, attachments, started_at: float | None, source: str = "webui") -> None: """Materialize the current user turn for eager first-turn persistence. The streaming thread still receives ``pending_user_message`` so existing @@ -13926,6 +13926,8 @@ def _checkpoint_user_message_for_eager_session_save(s, msg: str, attachments, st if latest_text == msg_text: return user_msg = {"role": "user", "content": msg} + if source and source != "webui": + user_msg["_source"] = source if isinstance(started_at, (int, float)) and started_at > 0: user_msg["timestamp"] = int(started_at) if attachments: @@ -13963,6 +13965,7 @@ def _prepare_chat_start_session_for_stream( model_provider, stream_id: str, started_at: float | None = None, + source: str = "webui", ): """Persist chat-start state according to webui.session_save_mode. @@ -13980,6 +13983,7 @@ def _prepare_chat_start_session_for_stream( s.pending_user_message = msg s.pending_attachments = attachments s.pending_started_at = started_at if started_at is not None else time.time() + s.pending_user_source = source current_title = getattr(s, "title", None) if _is_default_or_empty_session_title(current_title): provisional_title = _provisional_title_from_prompt(msg, current_title or "Untitled") @@ -13991,6 +13995,7 @@ def _prepare_chat_start_session_for_stream( msg, attachments, s.pending_started_at, + source=source, ) s.save() @@ -14097,6 +14102,7 @@ def _start_chat_stream_for_session( normalized_model: bool = False, diag=None, goal_related: bool = False, + source: str = "webui", ): """Persist pending state, register an SSE channel, and start an agent turn.""" attachments = attachments or [] @@ -14167,6 +14173,7 @@ def _start_chat_stream_for_session( model=model, model_provider=model_provider, stream_id=stream_id, + source=source, ) break if needs_stale_cleanup: @@ -14343,6 +14350,7 @@ def _start_run( model_provider=request.provider or model_provider, normalized_model=normalized_model, diag=diag, + source=request.source or source, ) def _legacy_adapter_factory(): @@ -14381,6 +14389,7 @@ def _start_run( model_provider=model_provider, normalized_model=normalized_model, diag=diag, + source=source, ) diff --git a/static/ui.js b/static/ui.js index 4ba64133..116c367a 100644 --- a/static/ui.js +++ b/static/ui.js @@ -453,6 +453,7 @@ function _cancelMessageVirtualizedRender(){ } function _messageIsRenderable(m){ if(!m||!m.role||m.role==='tool') return false; + if(m._source === 'process_wakeup') return false; if(_isContextCompactionMessage(m)||_isPreservedCompressionTaskListMessage(m)) return false; if(_isRecoveryControlMessage(m)) return false; const hasTc=Array.isArray(m.tool_calls)&&m.tool_calls.length>0; @@ -7285,6 +7286,7 @@ function getPendingSessionMessage(session, messagesOverride=null){ attachments:attachments.length?attachments:undefined, _ts:session?.pending_started_at||Date.now()/1000, _pending:true, + _source:session?.pending_user_source||undefined, }; } async function checkInflightOnBoot(sid) { diff --git a/tests/test_process_wakeup_synthetic.py b/tests/test_process_wakeup_synthetic.py new file mode 100644 index 00000000..6ba18290 --- /dev/null +++ b/tests/test_process_wakeup_synthetic.py @@ -0,0 +1,118 @@ +"""Test coverage for source field threading on process-wakeup synthetic turns.""" + +import time +from api.models import Session, _append_recovered_pending_turn +from api.routes import _checkpoint_user_message_for_eager_session_save + + +def test_append_recovered_pending_turn_stamps_process_wakeup_source(): + """Verify _append_recovered_pending_turn stamps _source when pending_user_source is process_wakeup.""" + s = Session( + session_id="test-session-1", + pending_user_message="[IMPORTANT: Process completed]", + pending_user_source="process_wakeup", + ) + recovered = _append_recovered_pending_turn(s) + assert recovered is not None + assert recovered["role"] == "user" + assert recovered["content"] == "[IMPORTANT: Process completed]" + assert recovered["_source"] == "process_wakeup" + assert recovered["_recovered"] is True + + +def test_append_recovered_pending_turn_skips_webui_source(): + """Verify _append_recovered_pending_turn does NOT stamp _source when source is webui (default).""" + s = Session( + session_id="test-session-2", + pending_user_message="Normal user message", + pending_user_source="webui", + ) + recovered = _append_recovered_pending_turn(s) + assert recovered is not None + assert recovered["role"] == "user" + assert recovered["content"] == "Normal user message" + assert "_source" not in recovered + assert recovered["_recovered"] is True + + +def test_append_recovered_pending_turn_defaults_to_no_source(): + """Verify _append_recovered_pending_turn does NOT stamp _source when pending_user_source is None.""" + s = Session( + session_id="test-session-3", + pending_user_message="Another user message", + pending_user_source=None, + ) + recovered = _append_recovered_pending_turn(s) + assert recovered is not None + assert recovered["role"] == "user" + assert "_source" not in recovered + + +def test_checkpoint_user_message_stamps_process_wakeup_source(): + """Verify eager-path checkpoint stamps _source on message dict when source is process_wakeup.""" + s = Session(session_id="test-session-4") + s.messages = [] + _checkpoint_user_message_for_eager_session_save( + s, + msg="[IMPORTANT: Wakeup prompt]", + attachments=[], + started_at=time.time(), + source="process_wakeup", + ) + assert len(s.messages) == 1 + user_msg = s.messages[0] + assert user_msg["role"] == "user" + assert user_msg["content"] == "[IMPORTANT: Wakeup prompt]" + assert user_msg["_source"] == "process_wakeup" + + +def test_checkpoint_user_message_skips_webui_source(): + """Verify eager-path checkpoint does NOT stamp _source when source is webui (default).""" + s = Session(session_id="test-session-5") + s.messages = [] + _checkpoint_user_message_for_eager_session_save( + s, + msg="Normal user message", + attachments=[], + started_at=time.time(), + source="webui", + ) + assert len(s.messages) == 1 + user_msg = s.messages[0] + assert user_msg["role"] == "user" + assert user_msg["content"] == "Normal user message" + assert "_source" not in user_msg + + +def test_checkpoint_user_message_defaults_to_no_source(): + """Verify eager-path checkpoint does NOT stamp _source when source is omitted (defaults to webui).""" + s = Session(session_id="test-session-6") + s.messages = [] + _checkpoint_user_message_for_eager_session_save( + s, + msg="Another user message", + attachments=[], + started_at=time.time(), + ) + assert len(s.messages) == 1 + user_msg = s.messages[0] + assert user_msg["role"] == "user" + assert user_msg["content"] == "Another user message" + assert "_source" not in user_msg + + +def test_session_pending_user_source_persisted(): + """Verify pending_user_source survives serialization and deserialization.""" + s = Session( + session_id="test-session-7", + pending_user_message="Test message", + pending_user_source="process_wakeup", + ) + s_dict = {k: getattr(s, k, None) for k in [ + 'session_id', 'pending_user_message', 'pending_user_source' + ]} + assert s_dict["pending_user_source"] == "process_wakeup" + + # Reconstruct from dict + s2 = Session(**s_dict) + assert s2.pending_user_source == "process_wakeup"