From 66438f07bc7660b9f0b8de54ea4475ce0ccfb32c Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Sun, 26 Jul 2026 21:24:48 -0400 Subject: [PATCH] Settle verification turns from exact turn ownership --- api/process_event_utils.py | 23 +- api/routes.py | 10 +- api/streaming.py | 479 ++++++++++-------- ...issue6481_verification_evidence_phantom.py | 206 +++++++- tests/test_tool_limit_terminal_state.py | 225 ++++++++ 5 files changed, 702 insertions(+), 241 deletions(-) diff --git a/api/process_event_utils.py b/api/process_event_utils.py index ccb8fcb0..2d9872f6 100644 --- a/api/process_event_utils.py +++ b/api/process_event_utils.py @@ -4,6 +4,7 @@ from __future__ import annotations from collections import OrderedDict from dataclasses import dataclass import logging +import math import re import threading import time @@ -137,7 +138,25 @@ def attach_wakeup_display_meta(msg: Any, source: Any) -> None: msg["_wakeup_meta"] = meta -def stamp_message_source(msg: Any, source: Any) -> None: +def build_active_turn_token(stream_id: Any, started_at: Any) -> str | None: + """Return the exact eager-row token for one active WebUI turn.""" + if not stream_id: + return None + try: + started = float(started_at) + except (TypeError, ValueError): + return None + if not math.isfinite(started) or started <= 0: + return None + return f"{str(stream_id).strip()}:{started:.17g}" + + +def stamp_message_source( + msg: Any, + source: Any, + *, + active_turn_token: Any = None, +) -> None: """Stamp ``_source`` and any display metadata on a materialized user turn. Single choke point for every path that persists a non-``webui`` user turn @@ -147,6 +166,8 @@ def stamp_message_source(msg: Any, source: Any) -> None: and the cancel outer-finally recovery. ``webui`` turns are left untouched to preserve the existing "``_source`` omitted for the default source" contract. """ + if isinstance(msg, dict) and active_turn_token: + msg["_active_turn_token"] = str(active_turn_token) if not isinstance(msg, dict) or not source or source == "webui": return msg["_source"] = source diff --git a/api/routes.py b/api/routes.py index 65110342..9049fad3 100644 --- a/api/routes.py +++ b/api/routes.py @@ -20929,11 +20929,15 @@ 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} - from api.process_event_utils import stamp_message_source + from api.process_event_utils import build_active_turn_token, stamp_message_source - stamp_message_source(user_msg, source) + stamp_message_source( + user_msg, + source, + active_turn_token=build_active_turn_token(getattr(s, "active_stream_id", None), started_at), + ) if isinstance(started_at, (int, float)) and started_at > 0: - user_msg["timestamp"] = int(started_at) + user_msg["timestamp"] = float(started_at) if attachments: user_msg["attachments"] = list(attachments) s.messages.append(user_msg) diff --git a/api/streaming.py b/api/streaming.py index 4d96c160..1b59a741 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -66,6 +66,7 @@ from api.models import ( ) from api.session_ops import mark_session_title_generated, session_has_manual_title from api.process_event_utils import ( + build_active_turn_token, claim_async_delegation_delivery, complete_async_delegation_delivery, completion_delivery_id, @@ -1441,39 +1442,146 @@ def _clean_synthetic_control_messages_with_provenance(messages): return _drop_synthetic_control_messages(raw_messages), has_verification_nudge -def _active_turn_checkpoint_identity(session, stream_id, msg_text): +def _active_turn_authority(session, stream_id, msg_text): """Capture the stream-owned pending turn before settlement mutates it.""" pending_text = getattr(session, 'pending_user_message', None) return { - 'stream_id': stream_id, + 'token': build_active_turn_token(stream_id, getattr(session, 'pending_started_at', None)), 'text': pending_text if pending_text is not None else msg_text, 'timestamp': getattr(session, 'pending_started_at', None), 'source': getattr(session, 'pending_user_source', None) or 'webui', 'attachments': copy.deepcopy(getattr(session, 'pending_attachments', None) or []), + 'current_turn_user_idx': None, + 'turn_id': '', } -def _active_turn_checkpoint_matches(message, identity): - if not isinstance(identity, dict) or not identity.get('stream_id'): - return False - if identity.get('timestamp') in (None, ''): - return False - if not isinstance(message, dict) or message.get('role') != 'user': - return False +def _coerce_current_turn_user_idx(value): + if isinstance(value, bool) or value is None: + return None try: - return ( - _normalize_user_text(message.get('content')) - == _normalize_user_text(identity.get('text')) - and int(message.get('timestamp')) == int(identity.get('timestamp')) - and (message.get('_source') or 'webui') == (identity.get('source') or 'webui') - and list(message.get('attachments') or []) == list(identity.get('attachments') or []) - ) + idx = int(value) except (TypeError, ValueError): + return None + return idx if idx >= 0 else None + + +def _resolve_active_turn_authority(identity, *, result=None, agent=None): + if not isinstance(identity, dict): + return identity + resolved = dict(identity) + if isinstance(result, dict): + _result_idx = _coerce_current_turn_user_idx(result.get('current_turn_user_idx')) + if _result_idx is not None: + resolved['current_turn_user_idx'] = _result_idx + _result_turn_id = str(result.get('turn_id') or '').strip() + if _result_turn_id: + resolved['turn_id'] = _result_turn_id + if agent is not None: + if resolved.get('current_turn_user_idx') is None: + _agent_idx = _coerce_current_turn_user_idx( + getattr(agent, '_persist_user_message_idx', None) + ) + if _agent_idx is not None: + resolved['current_turn_user_idx'] = _agent_idx + if not resolved.get('turn_id'): + _agent_turn_id = str(getattr(agent, '_current_turn_id', '') or '').strip() + if _agent_turn_id: + resolved['turn_id'] = _agent_turn_id + return resolved + + +def _active_turn_boundary_is_valid(identity): + if not isinstance(identity, dict): return False + if not str(identity.get('turn_id') or '').strip(): + return False + return isinstance(identity.get('current_turn_user_idx'), int) + + +def _active_turn_token_matches(message, identity): + token = identity.get('token') if isinstance(identity, dict) else None + if not token: + return False + return ( + isinstance(message, dict) + and message.get('role') == 'user' + and message.get('_active_turn_token') == token + ) def _active_turn_has_checkpoint(messages, identity): - return any(_active_turn_checkpoint_matches(message, identity) for message in messages or []) + return any(_active_turn_token_matches(message, identity) for message in messages or []) + + +def _mark_active_turn_checkpoint(message, identity): + if ( + isinstance(message, dict) + and message.get('role') == 'user' + and isinstance(identity, dict) + and identity.get('token') + ): + message['_active_turn_token'] = identity['token'] + return message + + +def _mark_active_turn_checkpoint_in_history(messages, identity, msg_text): + messages = list(messages or []) + if not isinstance(identity, dict): + return messages, False + if _active_turn_has_checkpoint(messages, identity): + return messages, True + if not _active_turn_boundary_is_valid(identity): + return messages, False + idx = identity['current_turn_user_idx'] + if idx < 0 or idx >= len(messages): + return messages, False + message = messages[idx] + expected_text = identity.get('text') if identity.get('text') is not None else msg_text + if ( + not isinstance(message, dict) + or message.get('role') != 'user' + or _normalize_user_text(message.get('content')) != _normalize_user_text(expected_text) + ): + return messages, False + _mark_active_turn_checkpoint(message, identity) + return messages, True + + +def _find_active_turn_checkpoint_index(result_messages, previous_context, identity, msg_text): + result_messages = list(result_messages or []) + if not isinstance(identity, dict): + return None + if _active_turn_has_checkpoint(result_messages, identity): + for idx, message in enumerate(result_messages): + if _active_turn_token_matches(message, identity): + return idx + if not _active_turn_boundary_is_valid(identity): + return None + expected_text = identity.get('text') if identity.get('text') is not None else msg_text + candidates = [] + current_turn_user_idx = identity['current_turn_user_idx'] + previous_context = list(previous_context or []) + if _messages_have_prefix(result_messages, previous_context): + candidates.append(current_turn_user_idx) + else: + offset = current_turn_user_idx - len(previous_context) + candidates.extend([offset, current_turn_user_idx]) + seen = set() + for idx in candidates: + if idx in seen or idx is None: + continue + seen.add(idx) + if idx < 0 or idx >= len(result_messages): + continue + message = result_messages[idx] + if ( + isinstance(message, dict) + and message.get('role') == 'user' + and _normalize_user_text(message.get('content')) == _normalize_user_text(expected_text) + ): + return idx + return None def _materialize_active_turn_user(identity, msg_text, source): @@ -1486,40 +1594,50 @@ def _materialize_active_turn_user(identity, msg_text, source): message['timestamp'] = identity['timestamp'] if identity.get('attachments'): message['attachments'] = copy.deepcopy(identity['attachments']) - message['_source'] = identity.get('source') or source or 'webui' + stamp_message_source( + message, + identity.get('source') or source or 'webui', + active_turn_token=identity.get('token'), + ) else: stamp_message_source(message, source) return message -def _ensure_active_turn_boundary(previous_context, result_messages, identity, msg_text, source): +def _settle_current_turn_boundary(previous_context, result_messages, identity, msg_text, source): """Insert the pending turn before assistant/tool output when it is absent.""" result_messages = list(result_messages or []) if not result_messages or not isinstance(identity, dict): return result_messages - if _active_turn_has_checkpoint(result_messages, identity): + _checkpoint_idx = _find_active_turn_checkpoint_index( + result_messages, + previous_context, + identity, + msg_text, + ) + if _checkpoint_idx is not None: + _mark_active_turn_checkpoint(result_messages[_checkpoint_idx], identity) return result_messages - current_user_key = _message_identity({'role': 'user', 'content': msg_text}) - current_start = len(list(previous_context or [])) if _messages_have_prefix( - result_messages, list(previous_context or []) - ) else 0 - if any( - isinstance(message, dict) - and message.get('role') == 'user' - and _message_identity(message) == current_user_key - and not _active_turn_checkpoint_matches(message, identity) - for message in result_messages[current_start:] - ): - return result_messages - if _messages_have_prefix(result_messages, list(previous_context or [])): - insert_at = len(list(previous_context or [])) - elif all( + previous_context = list(previous_context or []) + if _messages_have_prefix(result_messages, previous_context): + insert_at = len(previous_context) + elif _active_turn_boundary_is_valid(identity): + insert_at = identity['current_turn_user_idx'] - len(previous_context) + if insert_at < 0 or insert_at > len(result_messages): + insert_at = identity['current_turn_user_idx'] + if insert_at < 0 or insert_at > len(result_messages): + insert_at = None + else: + insert_at = None + if insert_at is None and all( _is_context_compression_marker(message) or (isinstance(message, dict) and message.get('role') in ('assistant', 'tool')) for message in result_messages ): insert_at = 0 - else: + while insert_at < len(result_messages) and _is_context_compression_marker(result_messages[insert_at]): + insert_at += 1 + if insert_at is None: return result_messages return ( result_messages[:insert_at] @@ -1528,16 +1646,21 @@ def _ensure_active_turn_boundary(previous_context, result_messages, identity, ms ) -def _align_active_turn_boundaries(previous_display, previous_context, identity): +def _align_current_turn_display(previous_display, previous_context, identity): """Make a context-only exact checkpoint visible before shared settlement.""" display = list(previous_display or []) context = list(previous_context or []) if not isinstance(identity, dict): return display, context + display, _ = _mark_active_turn_checkpoint_in_history( + display, + identity, + identity.get('text'), + ) if _active_turn_has_checkpoint(display, identity): return display, context checkpoint = next( - (message for message in context if _active_turn_checkpoint_matches(message, identity)), + (message for message in context if _active_turn_token_matches(message, identity)), None, ) if checkpoint is not None: @@ -1567,6 +1690,69 @@ def _prepare_marker_clean_writeback( return [], list(previous_context_messages or []), provenance +def _settle_result_messages( + session, + previous_messages, + previous_context_messages, + result_messages, + msg_text, + source, + active_turn_identity, +): + ( + result_messages, + next_context_messages, + verification_nudge_provenance, + ) = _prepare_marker_clean_writeback( + previous_context_messages, + result_messages, + active_turn_identity, + ) + if result_messages: + _assign_stable_message_ids( + result_messages, + previous_messages, + previous_context_messages, + ) + next_context_messages = _dedupe_replayed_context_messages( + previous_context_messages, + next_context_messages, + msg_text, + ) + next_context_messages = _settle_current_turn_boundary( + previous_context_messages, + next_context_messages, + active_turn_identity, + msg_text, + source, + ) + session.context_messages = _deduplicate_context_messages(next_context_messages) + if result_messages: + session.context_messages = _settle_current_turn_boundary( + previous_context_messages, + session.context_messages, + active_turn_identity, + msg_text, + source, + ) + previous_display_for_writeback, session.context_messages = _align_current_turn_display( + previous_messages, + session.context_messages, + active_turn_identity, + ) + session.messages = _merge_display_messages_after_agent_result( + previous_display_for_writeback, + previous_context_messages, + _restore_display_reasoning_metadata(previous_messages, result_messages), + msg_text, + source=source, + verification_nudge_provenance=verification_nudge_provenance, + ) + _compact_session_image_parts_for_persistence(session) + _advance_truncation_watermark_after_commit(session) # #3831 + return result_messages + + def _current_turn_already_has_visible_assistant_answer(messages, *, current_user_text=None): """Return True when the current real user turn already has visible assistant prose.""" if current_user_text is not None: @@ -5926,6 +6112,11 @@ def _merge_display_messages_after_agent_result( for message in result_messages ) _active_turn_identity = _writeback_provenance.get('active_turn_identity') + previous_display, _ = _mark_active_turn_checkpoint_in_history( + previous_display, + _active_turn_identity, + msg_text, + ) # Same marker filter for the model-history inputs: the synthetic verify-loop # answer/nudge live in the agent's returned messages and prior context, and # would otherwise slip into the merged transcript as a real delta. (#5334) @@ -6116,29 +6307,14 @@ def _merge_display_messages_after_agent_result( current_user_already_checkpointed = ( _active_turn_has_checkpoint(merged, _active_turn_identity) if _active_turn_identity - else bool( - merged - and ( - _message_identity(merged[-1]) == current_user_key - or _looks_like_current_user_turn(merged[-1], msg_text) - ) - ) + else False ) verification_user_already_checkpointed = bool( _verification_nudge_seen + and _active_turn_identity and ( - ( - _active_turn_identity - and _active_turn_has_checkpoint(previous_display, _active_turn_identity) - ) - or ( - not _active_turn_identity - and msg_text - and any( - _looks_like_current_user_turn(_last_user_row(history), msg_text) - for history in (previous_display, previous_context) - ) - ) + _active_turn_has_checkpoint(previous_display, _active_turn_identity) + or _active_turn_has_checkpoint(previous_context, _active_turn_identity) ) ) if ( @@ -7838,7 +8014,7 @@ def _run_agent_streaming( _turn_session_identity_tokens = _set_turn_session_identity(session_id) s = get_session(session_id) _turn_pending_source = getattr(s, 'pending_user_source', None) or 'webui' - _active_turn_identity = _active_turn_checkpoint_identity(s, stream_id, msg_text) + _active_turn_identity = _active_turn_authority(s, stream_id, msg_text) update_active_run(stream_id, phase="running", session_id=session_id) s.workspace = str(Path(workspace).expanduser().resolve()) _last_persisted_model = None @@ -9298,6 +9474,7 @@ def _run_agent_streaming( ), task_id=session_id, persist_user_message=msg_text, + persist_user_timestamp=getattr(s, 'pending_started_at', None), ) # 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 @@ -9333,6 +9510,11 @@ def _run_agent_streaming( ) _run_conversation_kwargs["user_message"] = user_message result = agent.run_conversation(**_run_conversation_kwargs) + _active_turn_identity = _resolve_active_turn_authority( + _active_turn_identity, + result=result, + agent=agent, + ) # #4729: the run is done — flush any reasoning tail still in the coalescing # buffer (the agent never calls reasoning_callback(None), and a turn can end on # reasoning with no trailing token/tool boundary to trigger a flush) so the last @@ -9447,15 +9629,6 @@ def _run_agent_streaming( ) if isinstance(result, dict): result = {**result, 'messages': _result_messages} - ( - _result_messages, - _next_context_messages, - _verification_nudge_provenance, - ) = _prepare_marker_clean_writeback( - _previous_context_messages, - _result_messages, - _active_turn_identity, - ) if cancel_event.is_set(): _finalize_cancelled_turn(s, ephemeral=False) try: @@ -9472,53 +9645,15 @@ def _run_agent_streaming( logger.debug("Failed to append cancelled turn journal event", exc_info=True) put('cancel', _cancel_event_payload('Cancelled by user')) return - if _result_messages: - # Stamp stable ids on the shared result rows AFTER the context - # restore (so carried-forward ids survive) and BEFORE the - # dedupe/merge build both arrays — including before - # _dedupe_replayed_context_messages deep-copies any - # stale-user repaired boundary row — so display and - # model-context copies of each row share an id for the - # fork/truncate aligner. - _assign_stable_message_ids( - _result_messages, _previous_messages, _previous_context_messages - ) - _next_context_messages = _dedupe_replayed_context_messages( - _previous_context_messages, - _next_context_messages, - msg_text, - ) - _next_context_messages = _ensure_active_turn_boundary( - _previous_context_messages, - _next_context_messages, - _active_turn_identity, - msg_text, - _turn_pending_source, - ) - s.context_messages = _deduplicate_context_messages(_next_context_messages) - if _result_messages: - s.context_messages = _ensure_active_turn_boundary( - _previous_context_messages, - s.context_messages, - _active_turn_identity, - msg_text, - _turn_pending_source, - ) - _previous_display_for_writeback, _ = _align_active_turn_boundaries( + _result_messages = _settle_result_messages( + s, _previous_messages, - s.context_messages, + _previous_context_messages, + _result_messages, + msg_text, + _turn_pending_source, _active_turn_identity, ) - s.messages = _merge_display_messages_after_agent_result( - _previous_display_for_writeback, - _previous_context_messages, - _restore_display_reasoning_metadata(_previous_messages, _result_messages), - msg_text, - source=getattr(s, 'pending_user_source', None) or 'webui', - verification_nudge_provenance=_verification_nudge_provenance, - ) - _compact_session_image_parts_for_persistence(s) - _advance_truncation_watermark_after_commit(s) # #3831 # Strip XML tool-call blocks from assistant message content. # DeepSeek and some other providers emit ... # in the raw response text; this must be removed before the content is @@ -9822,10 +9957,16 @@ def _run_agent_streaming( ), task_id=session_id, persist_user_message=msg_text, + persist_user_timestamp=getattr(s, 'pending_started_at', None), ) if moa_config is not None: _heal_kwargs["moa_config"] = moa_config _heal_result = agent.run_conversation(**_heal_kwargs) + _active_turn_identity = _resolve_active_turn_authority( + _active_turn_identity, + result=_heal_result, + agent=agent, + ) _heal_all_msgs = _heal_result.get('messages') or [] _heal_ok = _has_new_assistant_reply(_heal_all_msgs, _prev_len) or _token_sent except Exception as _retry_exc: @@ -9851,58 +9992,15 @@ def _run_agent_streaming( _result_messages, enabled=_agent_result_tool_limit_reached(result), ) - ( - _result_messages, - _next_context_messages, - _verification_nudge_provenance, - ) = _prepare_marker_clean_writeback( - _previous_context_messages, - _result_messages, - _active_turn_identity, - ) - if _result_messages: - # Mint ids on the shared result rows BEFORE dedupe - # deep-copies any stale-user boundary row, so both - # arrays share the id (#5564). - _assign_stable_message_ids( - _result_messages, _previous_messages, _previous_context_messages - ) - _next_context_messages = _dedupe_replayed_context_messages( - _previous_context_messages, - _next_context_messages, - msg_text, - ) - _next_context_messages = _ensure_active_turn_boundary( - _previous_context_messages, - _next_context_messages, - _active_turn_identity, - msg_text, - _turn_pending_source, - ) - s.context_messages = _deduplicate_context_messages(_next_context_messages) - if _result_messages: - s.context_messages = _ensure_active_turn_boundary( - _previous_context_messages, - s.context_messages, - _active_turn_identity, - msg_text, - _turn_pending_source, - ) - _previous_display_for_writeback, _ = _align_active_turn_boundaries( + _result_messages = _settle_result_messages( + s, _previous_messages, - s.context_messages, + _previous_context_messages, + _result_messages, + msg_text, + _turn_pending_source, _active_turn_identity, ) - s.messages = _merge_display_messages_after_agent_result( - _previous_display_for_writeback, - _previous_context_messages, - _restore_reasoning_metadata(_previous_messages, _result_messages), - msg_text, - source=getattr(s, 'pending_user_source', None) or 'webui', - verification_nudge_provenance=_verification_nudge_provenance, - ) - _compact_session_image_parts_for_persistence(s) - _advance_truncation_watermark_after_commit(s) # #3831 # normal post-result persistence path by # leaving _assistant_added truthy (set below). _assistant_added = True # prevent re-entering guard @@ -11081,10 +11179,16 @@ def _run_agent_streaming( ), task_id=session_id, persist_user_message=msg_text, + persist_user_timestamp=getattr(s, 'pending_started_at', None), ) if moa_config is not None: _heal_kwargs2["moa_config"] = moa_config _heal_result = _heal_agent.run_conversation(**_heal_kwargs2) + _active_turn_identity = _resolve_active_turn_authority( + _active_turn_identity, + result=_heal_result, + agent=_heal_agent, + ) # Retry succeeded — persist the result normally if s is not None: if _checkpoint_stop is not None: @@ -11102,58 +11206,15 @@ def _run_agent_streaming( ) return _result_messages = _heal_result.get('messages') or _previous_context_messages - ( - _result_messages, - _next_context_messages, - _verification_nudge_provenance, - ) = _prepare_marker_clean_writeback( - _previous_context_messages, - _result_messages, - _active_turn_identity, - ) - if _result_messages: - # Mint ids on the shared result rows BEFORE dedupe - # deep-copies any stale-user boundary row, so both - # arrays share the id (#5564). - _assign_stable_message_ids( - _result_messages, _previous_messages, _previous_context_messages - ) - _next_context_messages = _dedupe_replayed_context_messages( - _previous_context_messages, - _next_context_messages, - msg_text, - ) - _next_context_messages = _ensure_active_turn_boundary( - _previous_context_messages, - _next_context_messages, - _active_turn_identity, - msg_text, - _turn_pending_source, - ) - s.context_messages = _deduplicate_context_messages(_next_context_messages) - if _result_messages: - s.context_messages = _ensure_active_turn_boundary( - _previous_context_messages, - s.context_messages, - _active_turn_identity, - msg_text, - _turn_pending_source, - ) - _previous_display_for_writeback, _ = _align_active_turn_boundaries( + _result_messages = _settle_result_messages( + s, _previous_messages, - s.context_messages, + _previous_context_messages, + _result_messages, + msg_text, + _turn_pending_source, _active_turn_identity, ) - s.messages = _merge_display_messages_after_agent_result( - _previous_display_for_writeback, - _previous_context_messages, - _restore_reasoning_metadata(_previous_messages, _result_messages), - msg_text, - source=getattr(s, 'pending_user_source', None) or 'webui', - verification_nudge_provenance=_verification_nudge_provenance, - ) - _compact_session_image_parts_for_persistence(s) - _advance_truncation_watermark_after_commit(s) # #3831 s.save() logger.info('[webui] self-heal (except path): retry succeeded') return # skip error emission diff --git a/tests/test_issue6481_verification_evidence_phantom.py b/tests/test_issue6481_verification_evidence_phantom.py index 3484fa05..12247ed7 100644 --- a/tests/test_issue6481_verification_evidence_phantom.py +++ b/tests/test_issue6481_verification_evidence_phantom.py @@ -1,5 +1,7 @@ """Regression coverage for issue #6481 verification follow-up contracts.""" +import types + from api import streaming @@ -39,6 +41,21 @@ def _verification_followup(marker, corrective=CORRECTIVE_ANSWER): ] +def _writeback_provenance(*, prompt, timestamp, current_turn_user_idx, turn_id="turn-current", token="stream:turn"): + return { + "verification_nudge_seen": True, + "active_turn_identity": { + "token": token, + "text": prompt, + "timestamp": timestamp, + "source": "webui", + "attachments": [], + "current_turn_user_idx": current_turn_user_idx, + "turn_id": turn_id, + }, + } + + def _merge( result_messages, previous=None, @@ -72,6 +89,11 @@ def test_corrective_followup_survives_after_attempted_answer_for_both_markers(): merged = _merge( previous + _verification_followup(marker), previous=previous, + writeback_provenance=_writeback_provenance( + prompt="Fix the failing test.", + timestamp=1.0, + current_turn_user_idx=0, + ), ) assert _role_content_sequence(merged) == _role_content_sequence( @@ -87,6 +109,11 @@ def test_delta_only_followup_preserves_corrective_answer(): merged = _merge( _verification_followup(marker), previous=previous, + writeback_provenance=_writeback_provenance( + prompt="Fix the failing test.", + timestamp=1.0, + current_turn_user_idx=0, + ), ) assert _role_content_sequence(merged) == _role_content_sequence( @@ -101,6 +128,11 @@ def test_first_post_nudge_answer_survives_when_no_attempted_answer_exists(): merged = _merge( previous + _verification_followup(marker), previous=previous, + writeback_provenance=_writeback_provenance( + prompt="Fix the failing test.", + timestamp=1.0, + current_turn_user_idx=0, + ), ) assert _role_content_sequence(merged) == _role_content_sequence( @@ -170,16 +202,12 @@ def test_marked_repeated_old_prompt_materializes_deferred_current_turn(): ], previous=previous, msg_text=prompt, - writeback_provenance={ - "verification_nudge_seen": True, - "active_turn_identity": { - "stream_id": "direct-stream", - "text": prompt, - "timestamp": 2.0, - "source": "webui", - "attachments": [], - }, - }, + writeback_provenance=_writeback_provenance( + prompt=prompt, + timestamp=2.0, + current_turn_user_idx=len(previous), + token="direct-stream:2", + ), ) assert _role_content_sequence(merged) == [ @@ -204,19 +232,19 @@ def test_context_only_exact_checkpoint_does_not_suppress_display_boundary(): "timestamp": 2.0, "_source": "webui", "attachments": [], + "_active_turn_token": "direct-stream:2", }, ] corrective = "Verification failed. I fixed the parser and reran the tests." - aligned_display, _ = streaming._align_active_turn_boundaries( + aligned_display, _ = streaming._align_current_turn_display( previous_display, previous_context, - { - "stream_id": "direct-stream", - "text": prompt, - "timestamp": 2.0, - "source": "webui", - "attachments": [], - }, + _writeback_provenance( + prompt=prompt, + timestamp=2.0, + current_turn_user_idx=2, + token="direct-stream:2", + )["active_turn_identity"], ) merged = _merge( @@ -227,16 +255,12 @@ def test_context_only_exact_checkpoint_does_not_suppress_display_boundary(): previous=aligned_display, previous_context=previous_context, msg_text=prompt, - writeback_provenance={ - "verification_nudge_seen": True, - "active_turn_identity": { - "stream_id": "direct-stream", - "text": prompt, - "timestamp": 2.0, - "source": "webui", - "attachments": [], - }, - }, + writeback_provenance=_writeback_provenance( + prompt=prompt, + timestamp=2.0, + current_turn_user_idx=2, + token="direct-stream:2", + ), ) assert _role_content_sequence(merged) == [ @@ -245,3 +269,129 @@ def test_context_only_exact_checkpoint_does_not_suppress_display_boundary(): ("user", prompt), ("assistant", corrective), ] + + +def test_shared_settlement_owner_contract(): + prompt = "Fix the failing test." + previous = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + result_messages = previous + [ + {"role": "user", "content": "[System: verify the workspace]", "_verification_stop_synthetic": True}, + {"role": "assistant", "content": CORRECTIVE_ANSWER}, + ] + session = types.SimpleNamespace(messages=list(previous), context_messages=list(previous)) + + streaming._settle_result_messages( + session, + previous, + previous, + result_messages, + prompt, + "webui", + _writeback_provenance( + prompt=prompt, + timestamp=1.9, + current_turn_user_idx=len(previous), + token="direct-stream:1.9", + )["active_turn_identity"], + ) + + expected = [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ("assistant", CORRECTIVE_ANSWER), + ] + assert _role_content_sequence(session.messages) == expected + assert _role_content_sequence(session.context_messages) == expected + + +def test_untokened_legacy_row_fails_closed(): + prompt = "Fix the failing test." + previous_display = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + previous_context = previous_display + [ + { + "role": "user", + "content": prompt, + "timestamp": 1.9, + "_source": "webui", + "attachments": [], + }, + ] + + merged = _merge( + [ + {"role": "user", "content": "[System: verify the workspace]", "_verification_stop_synthetic": True}, + {"role": "assistant", "content": CORRECTIVE_ANSWER}, + ], + previous=previous_display, + previous_context=previous_context, + msg_text=prompt, + writeback_provenance=_writeback_provenance( + prompt=prompt, + timestamp=1.9, + current_turn_user_idx=2, + token="direct-stream:1.9", + ), + ) + + assert _role_content_sequence(merged) == [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ("assistant", CORRECTIVE_ANSWER), + ] + + +def test_materialized_webui_user_omits_source(): + message = streaming._materialize_active_turn_user( + _writeback_provenance( + prompt="Fix the failing test.", + timestamp=1.9, + current_turn_user_idx=2, + token="direct-stream:1.9", + )["active_turn_identity"], + "Fix the failing test.", + "webui", + ) + + assert message["role"] == "user" + assert message["content"] == "Fix the failing test." + assert message["_active_turn_token"] == "direct-stream:1.9" + assert "_source" not in message + + +def test_materialized_process_wakeup_user_has_wakeup_meta(): + wakeup_text = ( + "[IMPORTANT: Background process bg-1 completed (exit_code=0).\n" + "Command: python worker.py\n" + "Output:\n" + "done]" + ) + message = streaming._materialize_active_turn_user( + { + "token": "wakeup-stream:1.9", + "text": wakeup_text, + "timestamp": 1.9, + "source": "process_wakeup", + "attachments": [], + "current_turn_user_idx": 0, + "turn_id": "turn-wakeup", + }, + wakeup_text, + "process_wakeup", + ) + + assert message["_source"] == "process_wakeup" + assert message["_active_turn_token"] == "wakeup-stream:1.9" + assert message["_wakeup_meta"] == { + "type": "completion", + "task_id": "bg-1", + "command": "python worker.py", + "exit_code": 0, + } diff --git a/tests/test_tool_limit_terminal_state.py b/tests/test_tool_limit_terminal_state.py index 3cd2f448..abcab3f2 100644 --- a/tests/test_tool_limit_terminal_state.py +++ b/tests/test_tool_limit_terminal_state.py @@ -23,6 +23,10 @@ def _run_streaming_with_fake_agent( prior_context_messages=None, msg_text="Do the long task.", pending_started_at=1.0, + current_turn_user_idx=None, + turn_id="turn-current", + agent_results=None, + enable_auth_retry=False, ): session_dir = tmp_path / "sessions" session_dir.mkdir() @@ -68,6 +72,7 @@ def _run_streaming_with_fake_agent( streaming.SESSIONS[session_id] = session event_queue = queue.Queue() streaming.STREAMS[stream_id] = event_queue + result_queue = list(agent_results or []) class FakeAgent: def __init__(self, **kwargs): @@ -82,8 +87,15 @@ def _run_streaming_with_fake_agent( self.reasoning_config = None self.ephemeral_system_prompt = None self._last_error = None + self._persist_user_message_idx = current_turn_user_idx + self._current_turn_id = turn_id if current_turn_user_idx is not None else "" def run_conversation(self, **kwargs): + if result_queue: + next_result = result_queue.pop(0) + if isinstance(next_result, BaseException): + raise next_result + return next_result return agent_result def interrupt(self, _message): @@ -99,6 +111,12 @@ def _run_streaming_with_fake_agent( m.setattr("api.config.get_config", lambda *_args, **_kwargs: {}) m.setattr("api.config._resolve_cli_toolsets", lambda *_args, **_kwargs: []) m.setitem(sys.modules, "hermes_state", fake_hermes_state) + if enable_auth_retry: + m.setattr( + streaming, + "_attempt_credential_self_heal", + lambda *_args, **_kwargs: {"api_key": "retry-key", "provider": "openai"}, + ) streaming._run_agent_streaming( session_id=session_id, msg_text=msg_text, @@ -307,6 +325,7 @@ def test_verification_nudge_is_removed_while_corrective_followup_persists( prior_messages=prior_turn, prior_context_messages=prior_turn, msg_text="Fix the failing test.", + current_turn_user_idx=0, ) expected_sequence = [ @@ -353,6 +372,45 @@ def test_deferred_repeated_old_prompt_marked_followup_materializes_current_turn( prior_context_messages=old_turn, msg_text=prompt, pending_started_at=2.0, + current_turn_user_idx=len(old_turn), + ) + + expected = [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ("assistant", corrective), + ] + for field in ("messages", "context_messages"): + assert [(message["role"], message.get("content")) for message in payload[field]] == expected + assert [message.get("content") for message in payload[field]].count(prompt) == 2 + assert marker not in json.dumps(payload[field]) + + +@pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"]) +def test_same_second_repeated_prompt_keeps_current_user_turn(tmp_path, monkeypatch, marker): + prompt = "Fix the failing test." + old_turn = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + corrective = "Verification failed. I fixed the parser and reran the tests." + result = { + "messages": old_turn + [ + {"role": "user", "content": "[System: verify the workspace]", marker: True}, + {"role": "assistant", "content": corrective}, + ], + } + + _events, payload = _run_streaming_with_fake_agent( + tmp_path, + monkeypatch, + result, + prior_messages=old_turn, + prior_context_messages=old_turn, + msg_text=prompt, + pending_started_at=1.9, + current_turn_user_idx=len(old_turn), ) expected = [ @@ -385,6 +443,7 @@ def test_eager_exact_checkpoint_does_not_duplicate_repeated_prompt( "timestamp": 2.0, "_source": "webui", "attachments": [], + "_active_turn_token": "stream-tool-limit:2", }, ] corrective = "Verification failed. I fixed the parser and reran the tests." @@ -403,6 +462,7 @@ def test_eager_exact_checkpoint_does_not_duplicate_repeated_prompt( prior_context_messages=old_turn + current_checkpoint, msg_text=prompt, pending_started_at=2.0, + current_turn_user_idx=2, ) expected = [ @@ -417,6 +477,170 @@ def test_eager_exact_checkpoint_does_not_duplicate_repeated_prompt( assert marker not in json.dumps(payload[field]) +@pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"]) +def test_eager_checkpoint_reused_by_exact_token(tmp_path, monkeypatch, marker): + prompt = "Fix the failing test." + old_turn = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + current_checkpoint = [ + { + "role": "user", + "content": prompt, + "timestamp": 1.9, + "_source": "webui", + "attachments": [], + "_active_turn_token": "stream-tool-limit:1.9", + }, + ] + corrective = "Verification failed. I fixed the parser and reran the tests." + result = { + "messages": [ + {"role": "user", "content": "[System: verify the workspace]", marker: True}, + {"role": "assistant", "content": corrective}, + ] + } + + _events, payload = _run_streaming_with_fake_agent( + tmp_path, + monkeypatch, + result, + prior_messages=old_turn, + prior_context_messages=old_turn + current_checkpoint, + msg_text=prompt, + pending_started_at=1.9, + current_turn_user_idx=2, + ) + + expected = [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ("assistant", corrective), + ] + for field in ("messages", "context_messages"): + assert [(message["role"], message.get("content")) for message in payload[field]] == expected + assert [message.get("content") for message in payload[field]].count(prompt) == 2 + assert marker not in json.dumps(payload[field]) + + +@pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"]) +def test_tool_limit_exit_uses_shared_settlement_owner(tmp_path, monkeypatch, marker): + prompt = "Fix the failing test." + old_turn = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + corrective = "Verification failed. I fixed the parser and reran the tests." + result = { + "turn_exit_reason": "max_iterations_reached(30/30)", + "messages": old_turn + [ + {"role": "user", "content": "[System: verify the workspace]", marker: True}, + {"role": "assistant", "content": corrective}, + ], + } + settle_calls = [] + original = streaming._settle_result_messages + + def _spy(*args, **kwargs): + settle_calls.append(True) + return original(*args, **kwargs) + + monkeypatch.setattr(streaming, "_settle_result_messages", _spy) + _events, payload = _run_streaming_with_fake_agent( + tmp_path, + monkeypatch, + result, + prior_messages=old_turn, + prior_context_messages=old_turn, + msg_text=prompt, + pending_started_at=1.9, + current_turn_user_idx=len(old_turn), + ) + + assert len(settle_calls) == 1 + assert [(message["role"], message.get("content")) for message in payload["messages"]] == [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ("assistant", corrective), + ] + + +@pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"]) +def test_exception_exit_uses_shared_settlement_owner(tmp_path, monkeypatch, marker): + prompt = "Fix the failing test." + old_turn = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + corrective = "Verification failed. I fixed the parser and reran the tests." + heal_result = { + "messages": old_turn + [ + {"role": "user", "content": "[System: verify the workspace]", marker: True}, + {"role": "assistant", "content": corrective}, + ], + } + session = types.SimpleNamespace(messages=list(old_turn), context_messages=list(old_turn)) + streaming._settle_result_messages( + session, + old_turn, + old_turn, + heal_result["messages"], + prompt, + "webui", + { + "token": "stream-tool-limit:1.9", + "text": prompt, + "timestamp": 1.9, + "source": "webui", + "attachments": [], + "current_turn_user_idx": len(old_turn), + "turn_id": "turn-current", + }, + ) + + assert [(message["role"], message.get("content")) for message in session.messages] == [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ("assistant", corrective), + ] + + +def test_missing_agent_authority_fails_closed(tmp_path, monkeypatch): + prompt = "Fix the failing test." + old_turn = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + result = { + "messages": old_turn + [ + {"role": "user", "content": "[System: verify the workspace]", "_verification_stop_synthetic": True}, + {"role": "assistant", "content": "Verification failed. I fixed the parser and reran the tests."}, + ], + } + + _events, payload = _run_streaming_with_fake_agent( + tmp_path, + monkeypatch, + result, + prior_messages=old_turn, + prior_context_messages=old_turn, + msg_text=prompt, + pending_started_at=1.9, + current_turn_user_idx=None, + ) + + assert [(message["role"], message.get("content")) for message in payload["messages"]] == [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ("assistant", "Verification failed. I fixed the parser and reran the tests."), + ] + + @pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"]) def test_synthetic_only_verification_delta_preserves_prior_histories(tmp_path, monkeypatch, marker): prior_turn = [ @@ -448,6 +672,7 @@ def test_synthetic_only_verification_delta_preserves_prior_histories(tmp_path, m prior_messages=prior_turn, prior_context_messages=prior_turn, msg_text="Fix the failing test.", + current_turn_user_idx=0, ) for field in ("messages", "context_messages"):