Merge pull request #6503 from rodboev/pr/6481-verification-evidence-contamination

fix(#6481): settle verification turns from exact turn ownership

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

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)

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,
@@ -1429,6 +1430,380 @@ def _drop_synthetic_control_messages(messages):
     ]
 
 
+def _clean_synthetic_control_messages_with_provenance(messages):
+    """Remove synthetic rows while retaining marked verification-nudge provenance."""
+    raw_messages = list(messages or [])
+    has_verification_nudge = any(
+        isinstance(message, dict)
+        and message.get('role') == 'user'
+        and _is_synthetic_control_message(message)
+        for message in raw_messages
+    )
+    return _drop_synthetic_control_messages(raw_messages), has_verification_nudge
+
+
+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 {
+        '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 _coerce_current_turn_user_idx(value):
+    if isinstance(value, bool) or value is None:
+        return None
+    try:
+        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_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, *, allow_index_fallback=True):
+    messages = list(messages or [])
+    if not isinstance(identity, dict):
+        return messages, False
+    if _active_turn_has_checkpoint(messages, identity):
+        return messages, True
+    if not allow_index_fallback:
+        return messages, False
+    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 _owner_projection_current_turn_row(messages, identity):
+    messages = list(messages or [])
+    if not isinstance(identity, dict):
+        return None
+    for message in messages:
+        if _active_turn_token_matches(message, identity):
+            return copy.deepcopy(message)
+    return None
+
+
+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):
+    message = {
+        'role': 'user',
+        'content': identity.get('text') if isinstance(identity, dict) else msg_text,
+    }
+    if isinstance(identity, dict):
+        if identity.get('timestamp') is not None:
+            message['timestamp'] = identity['timestamp']
+        if identity.get('attachments'):
+            message['attachments'] = copy.deepcopy(identity['attachments'])
+        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 _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
+    _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
+    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
+        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]
+        + [_materialize_active_turn_user(identity, msg_text, source)]
+        + result_messages[insert_at:]
+    )
+
+
+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'),
+        allow_index_fallback=False,
+    )
+    if _active_turn_has_checkpoint(display, identity):
+        return display, context
+    checkpoint = _owner_projection_current_turn_row(
+        context,
+        identity,
+    )
+    if checkpoint is None and identity.get('token'):
+        checkpoint = _materialize_active_turn_user(
+            identity,
+            identity.get('text'),
+            identity.get('source') or 'webui',
+        )
+        context.append(copy.deepcopy(checkpoint))
+    if checkpoint is not None and not _active_turn_has_checkpoint(display, identity):
+        display.append(copy.deepcopy(checkpoint))
+    return display, context
+
+
+def _prepare_marker_clean_writeback(
+    previous_context_messages,
+    result_messages,
+    active_turn_identity=None,
+):
+    """Return marker-cleaned rows, next context rows, and nudge provenance."""
+    cleaned, has_verification_nudge = _clean_synthetic_control_messages_with_provenance(
+        result_messages
+    )
+    provenance = {
+        'verification_nudge_seen': has_verification_nudge,
+        'active_turn_identity': copy.deepcopy(active_turn_identity),
+    }
+    if not result_messages:
+        return (
+            cleaned,
+            list(previous_context_messages or []),
+            provenance,
+        )
+    if cleaned:
+        return (
+            cleaned,
+            _restore_reasoning_metadata(previous_context_messages, cleaned),
+            provenance,
+        )
+    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
+        else list(next_context_messages or [])
+    )
+    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, *, active_turn_identity=None):
+    """Return True only when the token-owned current turn already has visible assistant prose."""
+    if not isinstance(active_turn_identity, dict) or not active_turn_identity.get('token'):
+        return False
+    current_turn_seen = False
+    for msg in messages or []:
+        if not isinstance(msg, dict) or _is_synthetic_control_message(msg):
+            continue
+        if _active_turn_token_matches(msg, active_turn_identity):
+            current_turn_seen = True
+            continue
+        if not current_turn_seen:
+            continue
+        role = msg.get('role')
+        if role == 'assistant' and _assistant_message_has_final_visible_text(msg) and not msg.get('_error'):
+            return True
+        if role == 'user':
+            return False
+    return False
+
+
 def _agent_result_tool_limit_reached(result) -> bool:
     """Return True when current-turn metadata says the tool iteration cap fired."""
     if not isinstance(result, dict):
@@ -4653,6 +5028,7 @@ def _deduplicate_context_messages(messages):
         return messages
     seen = set()
     deduped = []
+    user_exact_index = {}
     for msg in messages:
         if _is_context_compression_marker(msg):
             marker_key = (
@@ -4671,6 +5047,26 @@ def _deduplicate_context_messages(messages):
             deduped.append(msg)
             continue
         key = _message_identity(msg)
+        if isinstance(msg, dict) and msg.get('role') == 'user' and key is not None:
+            user_exact_key = (
+                key,
+                msg.get('timestamp'),
+                msg.get('_ts'),
+                msg.get('_source') or 'webui',
+                json.dumps(msg.get('attachments') or [], sort_keys=True, ensure_ascii=False),
+            )
+            prior_exact_idx = user_exact_index.get(user_exact_key)
+            if msg.get('_active_turn_token'):
+                if prior_exact_idx is not None:
+                    deduped[prior_exact_idx] = msg
+                    continue
+                if key in seen:
+                    deduped.append(msg)
+                    user_exact_index[user_exact_key] = len(deduped) - 1
+                    continue
+            elif prior_exact_idx is not None:
+                continue
+            user_exact_index[user_exact_key] = len(deduped)
         if key is not None and key in seen:
             continue
         if key is not None:
@@ -5174,6 +5570,19 @@ def _dedupe_replayed_context_messages(previous_context, result_messages, msg_tex
                 if candidates:
                     candidates = _strip_replayed_context_items(previous_context, candidates)
                 return previous_context + candidates
+        assistant_or_tool_only_result = bool(result_messages) and all(
+            _is_context_compression_marker(m)
+            or (
+                isinstance(m, dict)
+                and m.get('role') in ('assistant', 'tool')
+            )
+            for m in result_messages
+        )
+        if assistant_or_tool_only_result:
+            candidates = _strip_replayed_prefix(previous_context, result_messages)
+            if candidates:
+                candidates = _strip_replayed_context_items(previous_context, candidates)
+            return previous_context + candidates
         return result_messages
     candidates = result_messages[len(previous_context):]
     # Strip stale merges only from the new-turn candidate slice so that
@@ -5681,7 +6090,14 @@ def _advance_truncation_watermark_after_commit(session) -> None:
     session.truncation_watermark = time.time()
 
 
-def _merge_display_messages_after_agent_result(previous_display, previous_context, result_messages, msg_text, source: str = "webui"):
+def _merge_display_messages_after_agent_result(
+    previous_display,
+    previous_context,
+    result_messages,
+    msg_text,
+    source: str = "webui",
+    verification_nudge_provenance=None,
+):
     """Keep UI transcript durable while allowing model context to compact.
 
     If Hermes Agent returns a normal append-only history, append that delta to
@@ -5730,6 +6146,25 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex
     previous_display = _deduped
     previous_context = list(previous_context or [])
     result_messages = list(result_messages or [])
+    if isinstance(verification_nudge_provenance, dict):
+        _writeback_provenance = verification_nudge_provenance
+    else:
+        _writeback_provenance = {
+            'verification_nudge_seen': bool(verification_nudge_provenance),
+            'active_turn_identity': None,
+        }
+    _verification_nudge_seen = _writeback_provenance.get('verification_nudge_seen', False) or any(
+        isinstance(message, dict)
+        and message.get('role') == 'user'
+        and _is_synthetic_control_message(message)
+        for message in result_messages
+    )
+    _active_turn_identity = _writeback_provenance.get('active_turn_identity')
+    previous_display, _ = _align_current_turn_display(
+        previous_display,
+        previous_context,
+        _active_turn_identity,
+    )
     # 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)
@@ -5887,7 +6322,19 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex
             candidates = _strip_replayed_prefix(previous_context, candidates)
     else:
         current_user_idx = _find_current_user_turn(result_messages, msg_text)
-        turn_candidates = result_messages[current_user_idx:] if current_user_idx is not None else []
+        assistant_or_tool_only_result = bool(result_messages) and all(
+            _is_context_compression_marker(m)
+            or (
+                isinstance(m, dict)
+                and m.get('role') in ('assistant', 'tool')
+            )
+            for m in result_messages
+        )
+        turn_candidates = (
+            result_messages[current_user_idx:]
+            if current_user_idx is not None
+            else result_messages if assistant_or_tool_only_result else []
+        )
         # Normalize stale merges only in the current-turn slice.
         if msg_text and previous_user_tail:
             turn_candidates = _strip_stale_user_merge_from_messages(
@@ -5905,17 +6352,24 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex
         _message_identity(m) == current_user_key or _looks_like_current_user_turn(m, msg_text)
         for m in candidates
     )
-    current_user_already_checkpointed = bool(
-        merged
+    current_user_already_checkpointed = (
+        _active_turn_has_checkpoint(merged, _active_turn_identity)
+        if _active_turn_identity
+        else False
+    )
+    verification_user_already_checkpointed = bool(
+        _verification_nudge_seen
+        and _active_turn_identity
         and (
-            _message_identity(merged[-1]) == current_user_key
-            or _looks_like_current_user_turn(merged[-1], msg_text)
+            _active_turn_has_checkpoint(previous_display, _active_turn_identity)
+            or _active_turn_has_checkpoint(previous_context, _active_turn_identity)
         )
     )
     if (
         current_user_key is not None
         and not current_user_in_candidates
         and not current_user_already_checkpointed
+        and not verification_user_already_checkpointed
         and any(
             isinstance(m, dict) and m.get('role') in ('assistant', 'tool')
             for m in candidates
@@ -5927,8 +6381,9 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex
         # directly would make the assistant bubble appear attached to the prior
         # exchange and then clear the pending prompt. Materialize the current
         # turn at the transcript boundary before the assistant/tool response.
-        current_user_msg = {'role': 'user', 'content': msg_text}
-        stamp_message_source(current_user_msg, source)
+        current_user_msg = _materialize_active_turn_user(
+            _active_turn_identity, msg_text, source
+        )
         insert_at = 0
         while insert_at < len(candidates) and _is_context_compression_marker(candidates[insert_at]):
             insert_at += 1
@@ -6057,12 +6512,25 @@ def _turn_transcript_lacks_final_assistant_answer(
     msg_text,
     source: str = "webui",
     drop_replayed_assistant: bool = False,
+    active_turn_identity=None,
 ) -> bool:
     """Return True when an already-merged transcript still lacks a final assistant answer."""
     merged_messages = list(merged_messages or [])
     previous_display = list(previous_display or [])
-    current_user_idx = _find_current_user_turn(merged_messages, msg_text)
-    if current_user_idx is None or current_user_idx < len(previous_display):
+    current_user_token_idx = next(
+        (
+            idx for idx, message in enumerate(merged_messages)
+            if _active_turn_token_matches(message, active_turn_identity)
+        ),
+        None,
+    )
+    current_user_idx = current_user_token_idx
+    if current_user_idx is None:
+        current_user_idx = _find_current_user_turn(merged_messages, msg_text)
+    if current_user_idx is None or (
+        current_user_token_idx is None
+        and current_user_idx < len(previous_display)
+    ):
         # The active turn lives after the durable transcript boundary. If the
         # merged display only exposes an older user row, materialize the pending
         # prompt so a replayed assistant row cannot satisfy the wrong turn.
@@ -6109,15 +6577,26 @@ def _merged_transcript_lacks_final_assistant_answer(
     msg_text,
     source: str = "webui",
     drop_replayed_assistant: bool = False,
+    active_turn_identity=None,
 ) -> bool:
     """Return True when the current turn still lacks a final assistant answer."""
     previous_display = list(previous_display or [])
+    _verification_nudge_seen = any(
+        isinstance(message, dict)
+        and message.get('role') == 'user'
+        and _is_synthetic_control_message(message)
+        for message in result_messages or []
+    )
     merged_messages = _merge_display_messages_after_agent_result(
         previous_display,
         previous_context,
         _restore_reasoning_metadata(previous_display, result_messages),
         msg_text,
         source=source,
+        verification_nudge_provenance={
+            'verification_nudge_seen': _verification_nudge_seen,
+            'active_turn_identity': active_turn_identity,
+        },
     )
     return _turn_transcript_lacks_final_assistant_answer(
         merged_messages,
@@ -6125,6 +6604,7 @@ def _merged_transcript_lacks_final_assistant_answer(
         msg_text,
         source=source,
         drop_replayed_assistant=drop_replayed_assistant,
+        active_turn_identity=active_turn_identity,
     )
 
 
@@ -7607,6 +8087,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_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
@@ -8988,12 +9469,18 @@ def _run_agent_streaming(
                     state_messages=_external_state_messages,
                 ) or []
             )
-            _previous_context_messages = _new_turn_context_from_messages(
+            _previous_owner_context_messages = list(
                 reconciled_state_db_messages_for_session(
                     s,
                     prefer_context=True,
                     state_messages=_external_state_messages,
-                ),
+                ) or []
+            )
+            _previous_owner_context_messages = _deduplicate_context_messages(
+                _previous_owner_context_messages
+            )
+            _previous_context_messages = _new_turn_context_from_messages(
+                _previous_owner_context_messages,
                 msg_text,
             )
             # Dedup before feeding to agent — merge_session_messages_append_only
@@ -9066,6 +9553,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
@@ -9101,6 +9589,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
@@ -9192,7 +9685,9 @@ def _run_agent_streaming(
                         return
                 with _stream_writeback_stage(_writeback_timings, "merge_result"):
                     _tool_limit_reached = _agent_result_tool_limit_reached(result)
-                    _result_messages = result.get('messages') or _previous_context_messages
+                    _result_messages = result.get('messages')
+                    if _result_messages is None:
+                        _result_messages = _previous_context_messages
                     _result_messages = _drop_synthetic_max_iteration_summary_requests(
                         _result_messages,
                         enabled=_tool_limit_reached,
@@ -9231,35 +9726,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
-                    _next_context_messages = _restore_reasoning_metadata(
-                        _previous_context_messages,
-                        _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,
-                    )
-                    s.context_messages = _deduplicate_context_messages(_next_context_messages)
-                    s.messages = _merge_display_messages_after_agent_result(
+                    _result_messages = _settle_result_messages(
+                        s,
                         _previous_messages,
-                        _previous_context_messages,
-                        _restore_display_reasoning_metadata(_previous_messages, _result_messages),
+                        _previous_owner_context_messages,
+                        _result_messages,
                         msg_text,
-                        source=getattr(s, 'pending_user_source', None) or 'webui',
+                        _turn_pending_source,
+                        _active_turn_identity,
                     )
-                    _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 <function_calls>...</function_calls>
                 # in the raw response text; this must be removed before the content is
@@ -9403,7 +9878,7 @@ def _run_agent_streaming(
                 # workspace-aware helper from this branch while still
                 # preserving the pre-turn length for downstream self-heal
                 # checks introduced on master.
-                _all_result_messages = result.get('messages') or []
+                _all_result_messages = _drop_synthetic_control_messages(result.get('messages') or [])
                 _prev_len = len(_previous_context_messages)
                 _assistant_added = _assistant_reply_added_after_current_turn(
                     _all_result_messages,
@@ -9434,12 +9909,27 @@ def _run_agent_streaming(
                 )
                 _saved_transcript_lacks_final_answer = _merged_transcript_lacks_final_assistant_answer(
                     _previous_messages,
-                    _previous_context_messages,
+                    _previous_owner_context_messages,
                     _all_result_messages,
                     msg_text,
                     source=getattr(s, 'pending_user_source', None) or 'webui',
                     drop_replayed_assistant=_drop_replayed_assistant,
+                    active_turn_identity=_active_turn_identity,
                 )
+                if (
+                    not _all_result_messages
+                    and _current_turn_already_has_visible_assistant_answer(
+                        _align_current_turn_display(
+                            _previous_messages,
+                            _previous_owner_context_messages,
+                            _active_turn_identity,
+                        )[0],
+                        active_turn_identity=_active_turn_identity,
+                    )
+                ):
+                    _saved_transcript_lacks_final_answer = False
+                if not _assistant_added and not _saved_transcript_lacks_final_answer:
+                    _assistant_added = True
                 _is_agent_result_terminal = _agent_result_terminal_failure(result)
                 _terminal_failure = (
                     _captured_terminal_failure
@@ -9547,10 +10037,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:
@@ -9571,36 +10067,22 @@ def _run_agent_streaming(
                                 # evaluates False on next conceptual pass.
                                 # Since we're in a flat block, directly run the
                                 # post-result merge logic here.
-                                _result_messages = result.get('messages') or _previous_context_messages
+                                _result_messages = result.get('messages')
+                                if _result_messages is None:
+                                    _result_messages = _previous_context_messages
                                 _result_messages = _drop_synthetic_max_iteration_summary_requests(
                                     _result_messages,
                                     enabled=_agent_result_tool_limit_reached(result),
                                 )
-                                _next_context_messages = _restore_reasoning_metadata(
-                                    _previous_context_messages,
-                                    _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,
-                                )
-                                s.context_messages = _deduplicate_context_messages(_next_context_messages)
-                                s.messages = _merge_display_messages_after_agent_result(
+                                _result_messages = _settle_result_messages(
+                                    s,
                                     _previous_messages,
-                                    _previous_context_messages,
-                                    _restore_reasoning_metadata(_previous_messages, _result_messages),
+                                    _previous_owner_context_messages,
+                                    _result_messages,
                                     msg_text,
-                                    source=getattr(s, 'pending_user_source', None) or 'webui',
+                                    _turn_pending_source,
+                                    _active_turn_identity,
                                 )
-                                _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
@@ -10779,10 +11261,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:
@@ -10799,31 +11287,18 @@ def _run_agent_streaming(
                                         getattr(s, 'active_stream_id', None),
                                     )
                                     return
-                                _result_messages = _heal_result.get('messages') or _previous_context_messages
-                                _next_context_messages = _restore_reasoning_metadata(
-                                    _previous_context_messages, _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,
-                                )
-                                s.context_messages = _deduplicate_context_messages(_next_context_messages)
-                                s.messages = _merge_display_messages_after_agent_result(
+                                _result_messages = _heal_result.get('messages')
+                                if _result_messages is None:
+                                    _result_messages = _previous_context_messages
+                                _result_messages = _settle_result_messages(
+                                    s,
                                     _previous_messages,
-                                    _previous_context_messages,
-                                    _restore_reasoning_metadata(_previous_messages, _result_messages),
+                                    _previous_owner_context_messages,
+                                    _result_messages,
                                     msg_text,
-                                    source=getattr(s, 'pending_user_source', None) or 'webui',
+                                    _turn_pending_source,
+                                    _active_turn_identity,
                                 )
-                                _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

tests/test_issue4729_reasoning_sse_coalesce.py

@@ -97,10 +97,16 @@ def test_reasoning_buffer_flushed_at_every_boundary():
     # terminal catch-all: flush right after the agent run returns, AND a guaranteed-exit
     # flush in the inner finally so exception / retry paths (which bypass the post-run
     # flush) still emit the tail before the outer apperror.
-    assert re.search(
-        r"agent\.run_conversation\(\*\*_run_conversation_kwargs\)\n\s*# #4729",
-        STREAMING,
-    ), "must flush after agent.run_conversation() returns (success path, before done)"
+    run_idx = STREAMING.index("result = agent.run_conversation(**_run_conversation_kwargs)")
+    resolve_idx = STREAMING.index(
+        "_active_turn_identity = _resolve_active_turn_authority(",
+        run_idx,
+    )
+    flush_idx = STREAMING.index("_flush_reasoning_buffer()", resolve_idx)
+    assert run_idx < resolve_idx < flush_idx, (
+        "must flush after agent.run_conversation() returns, after turn authority is resolved, "
+        "and before terminal SSE delivery"
+    )
     # the inner finally must also flush (covers exception/retry exits)
     fin = STREAMING[STREAMING.index("        finally:\n            # #4729"):]
     fin = fin[:700]

tests/test_issue5141_terminal_failure_transcript_evaluator.py

@@ -93,7 +93,14 @@ def test_turn_evaluator_materializes_pending_user_after_display_boundary():
 def test_merged_wrapper_delegates_to_turn_evaluator():
     calls = []
 
-    def _fake_evaluator(merged_messages, previous_display, msg_text, source="webui", drop_replayed_assistant=False):
+    def _fake_evaluator(
+        merged_messages,
+        previous_display,
+        msg_text,
+        source="webui",
+        drop_replayed_assistant=False,
+        active_turn_identity=None,
+    ):
         calls.append(
             {
                 "merged_len": len(list(merged_messages or [])),
@@ -101,6 +108,7 @@ def test_merged_wrapper_delegates_to_turn_evaluator():
                 "msg_text": msg_text,
                 "source": source,
                 "drop_replayed_assistant": drop_replayed_assistant,
+                "active_turn_identity": active_turn_identity,
             }
         )
         return True
@@ -125,4 +133,5 @@ def test_merged_wrapper_delegates_to_turn_evaluator():
     assert calls[0]["msg_text"] == "hello"
     assert calls[0]["source"] == "cli"
     assert calls[0]["drop_replayed_assistant"] is True
+    assert calls[0]["active_turn_identity"] is None
     assert calls[0]["merged_len"] >= 1

tests/test_issue6481_verification_evidence_phantom.py

@@ -0,0 +1,475 @@
+"""Regression coverage for issue #6481 verification follow-up contracts."""
+
+import types
+
+from api import streaming
+
+
+ATTEMPTED_ANSWER = "The failing test is fixed."
+CORRECTIVE_ANSWER = "Verification failed. I fixed the parser and reran the tests."
+VERIFICATION_EVIDENCE = {
+    "status": "passed",
+    "kind": "terminal",
+    "scope": "workspace",
+    "canonical_command": "pytest tests/test_app.py -q",
+}
+
+
+def _current_turn_prefix(*, include_attempted_answer):
+    messages = [
+        {
+            "role": "user",
+            "content": "Fix the failing test.",
+            "_active_turn_token": "stream:turn",
+        },
+        {
+            "role": "assistant",
+            "content": [{"type": "tool_use", "name": "terminal", "input": {"cmd": "pytest"}}],
+        },
+        {
+            "role": "tool",
+            "tool_call_id": "terminal",
+            "content": "Tests passed.",
+            "verification_evidence": VERIFICATION_EVIDENCE,
+        },
+    ]
+    if include_attempted_answer:
+        messages.append({"role": "assistant", "content": ATTEMPTED_ANSWER})
+    return messages
+
+
+def _verification_followup(marker, corrective=CORRECTIVE_ANSWER):
+    return [
+        {"role": "user", "content": "[System: verify the workspace]", marker: True},
+        {"role": "assistant", "content": corrective},
+    ]
+
+
+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,
+    previous_context=None,
+    msg_text="Fix the failing test.",
+    writeback_provenance=None,
+):
+    previous = list(previous or [{"role": "user", "content": "Fix the failing test."}])
+    previous_context = list(previous_context if previous_context is not None else previous)
+    return streaming._merge_display_messages_after_agent_result(
+        previous,
+        previous_context,
+        result_messages,
+        msg_text,
+        verification_nudge_provenance=writeback_provenance,
+    )
+
+
+def _role_content_sequence(messages):
+    return [(message["role"], message.get("content")) for message in messages]
+
+
+def _expected_current_turn(*, include_attempted_answer):
+    expected = _current_turn_prefix(include_attempted_answer=include_attempted_answer)
+    return expected + [{"role": "assistant", "content": CORRECTIVE_ANSWER}]
+
+
+def test_corrective_followup_survives_after_attempted_answer_for_both_markers():
+    for marker in ("_verification_stop_synthetic", "_pre_verify_synthetic"):
+        previous = _current_turn_prefix(include_attempted_answer=True)
+        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(
+            _expected_current_turn(include_attempted_answer=True)
+        )
+        assert [message.get("content") for message in merged].count("Fix the failing test.") == 1
+        assert all(not streaming._is_synthetic_control_message(message) for message in merged)
+
+
+def test_delta_only_followup_preserves_corrective_answer():
+    for marker in ("_verification_stop_synthetic", "_pre_verify_synthetic"):
+        previous = _current_turn_prefix(include_attempted_answer=True)
+        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(
+            _expected_current_turn(include_attempted_answer=True)
+        )
+        assert [message.get("content") for message in merged].count("Fix the failing test.") == 1
+
+
+def test_first_post_nudge_answer_survives_when_no_attempted_answer_exists():
+    for marker in ("_verification_stop_synthetic", "_pre_verify_synthetic"):
+        previous = _current_turn_prefix(include_attempted_answer=False)
+        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(
+            _expected_current_turn(include_attempted_answer=False)
+        )
+        assert [message.get("content") for message in merged].count("Fix the failing test.") == 1
+
+
+def test_unmarked_adjacent_assistant_rows_remain_unchanged():
+    messages = [
+        {"role": "user", "content": "Give me two alternatives."},
+        {"role": "assistant", "content": "First alternative."},
+        {"role": "assistant", "content": "Second alternative."},
+    ]
+
+    assert streaming._drop_synthetic_control_messages(messages) == messages
+
+
+def test_unmarked_assistant_only_new_turn_materializes_current_user_row():
+    previous = [
+        {"role": "user", "content": "Earlier request."},
+        {"role": "assistant", "content": "Earlier answer."},
+    ]
+    result = [{"role": "assistant", "content": "New answer."}]
+
+    merged = _merge(result, previous=previous, msg_text="New request.")
+
+    assert _role_content_sequence(merged) == [
+        ("user", "Earlier request."),
+        ("assistant", "Earlier answer."),
+        ("user", "New request."),
+        ("assistant", "New answer."),
+    ]
+
+
+def test_unmarked_repeated_identical_prompt_materializes_new_user_row():
+    previous = [
+        {"role": "user", "content": "Fix the failing test."},
+        {"role": "assistant", "content": "The first attempt."},
+    ]
+    result = [{"role": "assistant", "content": "The second attempt."}]
+
+    merged = _merge(result, previous=previous)
+
+    assert _role_content_sequence(merged) == [
+        ("user", "Fix the failing test."),
+        ("assistant", "The first attempt."),
+        ("user", "Fix the failing test."),
+        ("assistant", "The second attempt."),
+    ]
+
+
+def test_marked_repeated_old_prompt_materializes_deferred_current_turn():
+    prompt = "Fix the failing test."
+    previous = [
+        {"role": "user", "content": prompt, "timestamp": 1.0},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    corrective = "Verification failed. I fixed the parser and reran the tests."
+
+    for marker in ("_verification_stop_synthetic", "_pre_verify_synthetic"):
+        merged = _merge(
+            previous
+            + [
+                {"role": "user", "content": "[System: verify the workspace]", marker: True},
+                {"role": "assistant", "content": corrective},
+            ],
+            previous=previous,
+            msg_text=prompt,
+            writeback_provenance=_writeback_provenance(
+                prompt=prompt,
+                timestamp=2.0,
+                current_turn_user_idx=len(previous),
+                token="direct-stream:2",
+            ),
+        )
+
+        assert _role_content_sequence(merged) == [
+            ("user", prompt),
+            ("assistant", "The earlier attempt is complete."),
+            ("user", prompt),
+            ("assistant", corrective),
+        ]
+        assert [message.get("content") for message in merged].count(prompt) == 2
+
+
+def test_context_only_exact_checkpoint_does_not_suppress_display_boundary():
+    prompt = "Fix the failing test."
+    previous_display = [
+        {"role": "user", "content": prompt, "timestamp": 1.0},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    previous_context = previous_display + [
+        {
+            "role": "user",
+            "content": prompt,
+            "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_current_turn_display(
+        previous_display,
+        previous_context,
+        _writeback_provenance(
+            prompt=prompt,
+            timestamp=2.0,
+            current_turn_user_idx=2,
+            token="direct-stream:2",
+        )["active_turn_identity"],
+    )
+
+    merged = _merge(
+        [
+            {"role": "user", "content": "[System: verify the workspace]", "_verification_stop_synthetic": True},
+            {"role": "assistant", "content": corrective},
+        ],
+        previous=aligned_display,
+        previous_context=previous_context,
+        msg_text=prompt,
+        writeback_provenance=_writeback_provenance(
+            prompt=prompt,
+            timestamp=2.0,
+            current_turn_user_idx=2,
+            token="direct-stream:2",
+        ),
+    )
+
+    assert _role_content_sequence(merged) == [
+        ("user", prompt),
+        ("assistant", "The earlier attempt is complete."),
+        ("user", prompt),
+        ("assistant", corrective),
+    ]
+
+
+def test_divergent_context_index_does_not_relabel_same_text_display_row():
+    prompt = "Fix the failing test."
+    token = "direct-stream:2"
+    previous_display = [
+        {"role": "user", "content": prompt, "timestamp": 1.0},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    previous_context = [
+        {"role": "user", "content": prompt, "timestamp": 1.0, "attachments": []},
+        {
+            "role": "user",
+            "content": prompt,
+            "timestamp": 2.0,
+            "attachments": [],
+            "_active_turn_token": token,
+        },
+    ]
+
+    aligned_display, _ = streaming._align_current_turn_display(
+        previous_display,
+        previous_context,
+        _writeback_provenance(
+            prompt=prompt,
+            timestamp=2.0,
+            current_turn_user_idx=0,
+            token=token,
+        )["active_turn_identity"],
+    )
+
+    assert _role_content_sequence(aligned_display) == [
+        ("user", prompt),
+        ("assistant", "The earlier attempt is complete."),
+        ("user", prompt),
+    ]
+    assert aligned_display[0].get("_active_turn_token") is None
+    assert aligned_display[-1]["_active_turn_token"] == token
+
+
+def test_missing_exact_checkpoint_materializes_current_turn_without_retokening_history():
+    prompt = "Fix the failing test."
+    token = "direct-stream:1.9"
+    previous_display = [
+        {"role": "user", "content": prompt, "timestamp": 1.1},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    previous_context = [
+        {"role": "user", "content": prompt, "timestamp": 1.1, "attachments": []},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+
+    aligned_display, aligned_context = streaming._align_current_turn_display(
+        previous_display,
+        previous_context,
+        _writeback_provenance(
+            prompt=prompt,
+            timestamp=1.9,
+            current_turn_user_idx=0,
+            token=token,
+        )["active_turn_identity"],
+    )
+
+    expected = [
+        ("user", prompt),
+        ("assistant", "The earlier attempt is complete."),
+        ("user", prompt),
+    ]
+    assert _role_content_sequence(aligned_display) == expected
+    assert _role_content_sequence(aligned_context) == expected
+    assert aligned_display[0].get("_active_turn_token") is None
+    assert aligned_context[0].get("_active_turn_token") is None
+    assert aligned_display[-1]["_active_turn_token"] == token
+    assert aligned_context[-1]["_active_turn_token"] == token
+
+
+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,
+    }

tests/test_issue765_streaming_persistence.py

@@ -314,10 +314,10 @@ class TestIssue765FollowupHardening:
         t2.join(timeout=5)
 
         assert not errors, f"Concurrent same-session saves should not fail: {errors}"
-        assert len(replace_sources) == 2, f"Expected 2 replace calls, got {replace_sources}"
+        assert len(replace_sources) >= 2, f"Expected replace calls, got {replace_sources}"
         assert len(set(replace_sources)) == 2, (
-            "Concurrent same-session saves must use distinct temp files; "
-            f"got {replace_sources}"
+            "Concurrent same-session saves must use distinct temp files even if Windows-safe "
+            f"replace retries one of them; got {replace_sources}"
         )
         data = json.loads(s.path.read_text(encoding="utf-8"))
         assert data["session_id"] == "same_sid"
@@ -337,14 +337,14 @@ class TestIssue765FollowupHardening:
             "with _agent_lock:\n"
             "                if not ephemeral and not _stream_writeback_is_current(s, stream_id):"
         )
-        save_idx = src.find("_deduplicate_context_messages(_next_context_messages)")
+        save_idx = src.find("_result_messages = _settle_result_messages(")
 
         assert stop_idx != -1, "Success path must stop the checkpoint thread"
         assert join_idx != -1, "Success path must join the checkpoint thread"
         assert lock_idx != -1, "Success path must serialize mutation with _agent_lock"
-        assert save_idx != -1, "Success path restore/mutation block not found"
+        assert save_idx != -1, "Success path settlement block not found"
         assert stop_idx < join_idx < lock_idx <= save_idx, (
-            "Checkpoint stop/join must happen before the success-path session mutation block"
+            "Checkpoint stop/join must happen before the success-path settlement block"
         )
 
     def test_silent_failure_path_does_not_reacquire_agent_lock(self):

tests/test_run_journal_streaming_static.py

@@ -29,11 +29,14 @@ def test_streaming_journals_sse_events_before_queue_delivery():
 def test_streaming_compacts_all_successful_agent_result_writebacks():
     src = Path("api/streaming.py").read_text(encoding="utf-8")
     run_src = src[src.index("def _run_agent_streaming("):]
+    settle_src = src[src.index("def _settle_result_messages("):]
+    settle_src = settle_src[: settle_src.index("\n\ndef _current_turn_already_has_visible_assistant_answer(")]
 
-    # Normal completion plus both credential self-heal retry-success paths must
-    # each compact after committing their own result shape. The retries use
-    # different local variable names, so guard the durable invariant itself.
-    assert run_src.count("_compact_session_image_parts_for_persistence(s)") == 3
+    # Normal completion plus both credential self-heal retry-success paths now
+    # route through one shared settlement helper, which owns the compaction.
+    assert "def _settle_result_messages(" in src
+    assert "_compact_session_image_parts_for_persistence(session)" in settle_src
+    assert run_src.count("_settle_result_messages(") == 3
 
 
 def test_visible_process_echo_compare_ignores_all_whitespace():

tests/test_sprint42.py

@@ -813,9 +813,9 @@ def test_streaming_restores_prior_reasoning_metadata_after_followup():
     src = (REPO / 'api' / 'streaming.py').read_text(encoding="utf-8")
     assert "def _restore_reasoning_metadata(" in src, \
         "streaming.py must define a helper to restore prior reasoning metadata"
-    assert "_next_context_messages" in src and "s.context_messages" in src, \
+    assert "next_context_messages" in src and "_deduplicate_context_messages(next_context_messages)" in src, \
         "streaming.py must restore prior reasoning metadata into model context"
-    assert "s.messages = _merge_display_messages_after_agent_result(" in src, \
+    assert "session.messages = _merge_display_messages_after_agent_result(" in src, \
         "streaming.py must merge restored result messages into the visible transcript"
     assert "updated_messages.insert(safe_pos, copy.deepcopy(prev_msg))" in src, \
         "streaming.py must reinsert dropped reasoning-only assistant messages"

tests/test_stale_stream_writeback.py

@@ -174,7 +174,7 @@ def test_success_path_checks_stream_ownership_before_persisting_result():
     src = Path("api/streaming.py").read_text(encoding="utf-8")
     guard = "if not ephemeral and not _stream_writeback_is_current(s, stream_id):"
     guard_pos = src.find(guard)
-    result_merge_pos = src.find("_result_messages = result.get('messages') or _previous_context_messages")
+    result_merge_pos = src.find("_result_messages = result.get('messages')")
     compression_pos = src.find("Handle context compression side effects")
 
     assert guard_pos != -1
@@ -192,7 +192,7 @@ def test_self_heal_retry_success_checks_stream_ownership_before_writeback():
     guard = "if not ephemeral and not _stream_writeback_is_current(s, stream_id):"
 
     assert guard in block
-    assert block.index(guard) < block.index("_result_messages = _heal_result.get('messages') or _previous_context_messages")
+    assert block.index(guard) < block.index("_result_messages = _heal_result.get('messages')")
     assert block.index(guard) < block.index("s.save()")
 
 

tests/test_tool_limit_terminal_state.py

@@ -4,7 +4,10 @@ import sys
 import types
 from pathlib import Path
 
+import pytest
+
 from api import models
+from api.process_event_utils import build_active_turn_token
 from api import streaming
 from api.models import Session
 
@@ -19,6 +22,12 @@ def _run_streaming_with_fake_agent(
     *,
     prior_messages=None,
     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()
@@ -40,22 +49,31 @@ def _run_streaming_with_fake_agent(
 
     session_id = "tool_limit_session"
     stream_id = "stream-tool-limit"
+    session_messages = list(prior_messages or [])
+    session_context_messages = list(prior_context_messages or [])
+    for history in (session_messages, session_context_messages):
+        for message in history:
+            if isinstance(message, dict) and message.get("role") == "user":
+                message.setdefault("timestamp", 1.0)
+                message.setdefault("_source", "webui")
+                message.setdefault("attachments", [])
     session = Session(
         session_id=session_id,
         title="Tool limit test",
         workspace=str(tmp_path),
         model="gpt-4o",
-        messages=list(prior_messages or []),
-        context_messages=list(prior_context_messages or []),
+        messages=session_messages,
+        context_messages=session_context_messages,
     )
     session.active_stream_id = stream_id
-    session.pending_user_message = "Do the long task."
-    session.pending_started_at = 1.0
+    session.pending_user_message = msg_text
+    session.pending_started_at = pending_started_at
     session.save()
     models.SESSIONS[session_id] = session
     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):
@@ -70,8 +88,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):
@@ -87,9 +112,15 @@ 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="Do the long task.",
+            msg_text=msg_text,
             model="gpt-4o",
             workspace=str(tmp_path),
             stream_id=stream_id,
@@ -255,6 +286,408 @@ def test_streaming_tool_limit_with_final_answer_persists_clean_done_state(tmp_pa
     assert assistant["_statusCard"]["title"] == "Tool iteration limit reached"
 
 
+@pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"])
+@pytest.mark.parametrize("delta_only", [False, True])
+def test_verification_nudge_is_removed_while_corrective_followup_persists(
+    tmp_path,
+    monkeypatch,
+    marker,
+    delta_only,
+):
+    corrective = "Verification failed. I fixed the parser and reran the tests."
+    prior_turn = [
+        {"role": "user", "content": "Fix the failing test."},
+        {"role": "assistant", "content": [{"type": "tool_use", "name": "terminal"}]},
+        {
+            "role": "tool",
+            "tool_call_id": "terminal",
+            "content": "Tests passed.",
+            "verification_evidence": {
+                "status": "passed",
+                "kind": "terminal",
+                "scope": "workspace",
+                "canonical_command": "pytest tests/test_app.py -q",
+            },
+        },
+        {"role": "assistant", "content": "The failing test is fixed."},
+    ]
+    result_messages = [
+        {"role": "user", "content": "[System: verify the workspace]", marker: True},
+        {"role": "assistant", "content": corrective},
+    ]
+    if not delta_only:
+        result_messages = prior_turn + result_messages
+    result = {"messages": result_messages}
+
+    _events, payload = _run_streaming_with_fake_agent(
+        tmp_path,
+        monkeypatch,
+        result,
+        prior_messages=prior_turn,
+        prior_context_messages=prior_turn,
+        msg_text="Fix the failing test.",
+        current_turn_user_idx=0,
+    )
+
+    expected_sequence = [
+        (message["role"], message.get("content"))
+        for message in prior_turn
+    ] + [("assistant", corrective)]
+    for field in ("messages", "context_messages"):
+        serialized = json.dumps(payload[field])
+        assert marker not in serialized
+        assert "[System: verify the workspace]" not in serialized
+        assert [
+            (message["role"], message.get("content"))
+            for message in payload[field]
+        ] == expected_sequence
+        assert [message.get("content") for message in payload[field]].count(
+            "Fix the failing test."
+        ) == 1
+
+
+@pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"])
+def test_deferred_repeated_old_prompt_marked_followup_materializes_current_turn(
+    tmp_path,
+    monkeypatch,
+    marker,
+):
+    prompt = "Fix the failing test."
+    old_turn = [
+        {"role": "user", "content": prompt, "timestamp": 1.0},
+        {"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=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 = [
+        ("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_eager_exact_checkpoint_does_not_duplicate_repeated_prompt(
+    tmp_path,
+    monkeypatch,
+    marker,
+):
+    prompt = "Fix the failing test."
+    old_turn = [
+        {"role": "user", "content": prompt, "timestamp": 1.0},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    current_checkpoint = [
+        {
+            "role": "user",
+            "content": 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."
+    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=2.0,
+        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_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": build_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_shared_settlement_owner_handles_exception_style_delta(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 = [
+        {
+            "role": "user",
+            "content": "Fix the failing test.",
+            "_active_turn_token": build_active_turn_token("stream-tool-limit", 1.0),
+        },
+        {"role": "assistant", "content": [{"type": "tool_use", "name": "terminal"}]},
+        {
+            "role": "tool",
+            "tool_call_id": "terminal",
+            "content": "Tests passed.",
+            "verification_evidence": {
+                "status": "passed",
+                "kind": "terminal",
+                "scope": "workspace",
+                "canonical_command": "pytest tests/test_app.py -q",
+            },
+        },
+        {"role": "assistant", "content": "The failing test is fixed."},
+    ]
+    result = {
+        "messages": [
+            {"role": "user", "content": "[System: verify the workspace]", marker: True},
+        ],
+    }
+
+    _events, payload = _run_streaming_with_fake_agent(
+        tmp_path,
+        monkeypatch,
+        result,
+        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"):
+        assert payload[field] == payload["messages"]
+        serialized = json.dumps(payload[field])
+        assert marker not in serialized
+        assert "[System: verify the workspace]" not in serialized
+        assert "The failing test is fixed." in serialized
+
+
 def test_streaming_tool_limit_partial_without_final_answer_emits_no_final_apperror(tmp_path, monkeypatch):
     result = {
         "status": "partial",
@@ -479,3 +912,169 @@ def test_streaming_empty_result_messages_do_not_treat_prior_assistant_as_current
     assert apperror_payloads[-1]["type"] == "no_response"
     assert not [payload for event, payload in events if event == "done"]
     assert payload["messages"][-1]["_error"] is True
+
+
+def test_streaming_empty_result_same_text_does_not_treat_prior_assistant_as_current_answer(
+    tmp_path,
+    monkeypatch,
+):
+    prompt = "Fix the failing test."
+    prior = [
+        {"role": "user", "content": prompt, "timestamp": 1.1},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    result = {"messages": []}
+
+    events, payload = _run_streaming_with_fake_agent(
+        tmp_path,
+        monkeypatch,
+        result,
+        prior_messages=prior,
+        prior_context_messages=prior,
+        msg_text=prompt,
+        pending_started_at=1.9,
+        current_turn_user_idx=len(prior),
+    )
+
+    apperror_payloads = [event_payload for event, event_payload in events if event == "apperror"]
+    assert apperror_payloads, "expected silent-failure apperror"
+    assert apperror_payloads[-1]["type"] == "no_response"
+    assert not [event_payload for event, event_payload in events if event == "done"]
+    assert [(message["role"], message.get("content")) for message in payload["messages"][:-1]] == [
+        ("user", prompt),
+        ("assistant", "The earlier attempt is complete."),
+        ("user", prompt),
+    ]
+    assert payload["messages"][-1]["role"] == "assistant"
+    assert payload["messages"][-1]["_error"] is True
+
+
+def test_streaming_empty_result_without_exact_checkpoint_materializes_current_user(
+    tmp_path,
+    monkeypatch,
+):
+    prompt = "Fix the failing test."
+    prior = [
+        {"role": "user", "content": prompt, "timestamp": 1.1},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    result = {"messages": []}
+
+    events, payload = _run_streaming_with_fake_agent(
+        tmp_path,
+        monkeypatch,
+        result,
+        prior_messages=prior,
+        prior_context_messages=prior,
+        msg_text=prompt,
+        pending_started_at=1.9,
+        current_turn_user_idx=0,
+    )
+
+    apperror_payloads = [event_payload for event, event_payload in events if event == "apperror"]
+    assert apperror_payloads, "expected silent-failure apperror"
+    assert apperror_payloads[-1]["type"] == "no_response"
+    assert not [event_payload for event, event_payload in events if event == "done"]
+    assert [(message["role"], message.get("content")) for message in payload["messages"][:-1]] == [
+        ("user", prompt),
+        ("assistant", "The earlier attempt is complete."),
+        ("user", prompt),
+    ]
+    assert payload["messages"][0].get("_active_turn_token") is None
+    assert payload["messages"][2]["_active_turn_token"] == build_active_turn_token(
+        "stream-tool-limit",
+        1.9,
+    )
+    assert payload["messages"][-1]["role"] == "assistant"
+    assert payload["messages"][-1]["_error"] is True
+
+
+def test_streaming_empty_result_divergent_context_index_does_not_reuse_historical_assistant(
+    tmp_path,
+    monkeypatch,
+):
+    prompt = "Fix the failing test."
+    current_token = build_active_turn_token("stream-tool-limit", 1.9)
+    prior = [
+        {"role": "user", "content": prompt, "timestamp": 1.0},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    prior_context = [
+        {"role": "user", "content": prompt, "timestamp": 1.0, "attachments": []},
+        {
+            "role": "user",
+            "content": prompt,
+            "timestamp": 1.9,
+            "attachments": [],
+            "_active_turn_token": current_token,
+        },
+    ]
+    result = {"messages": []}
+
+    events, payload = _run_streaming_with_fake_agent(
+        tmp_path,
+        monkeypatch,
+        result,
+        prior_messages=prior,
+        prior_context_messages=prior_context,
+        msg_text=prompt,
+        pending_started_at=1.9,
+        current_turn_user_idx=0,
+    )
+
+    apperror_payloads = [event_payload for event, event_payload in events if event == "apperror"]
+    assert apperror_payloads, "expected silent-failure apperror"
+    assert apperror_payloads[-1]["type"] == "no_response"
+    assert not [event_payload for event, event_payload in events if event == "done"]
+    assert [(message["role"], message.get("content")) for message in payload["messages"][:-1]] == [
+        ("user", prompt),
+        ("assistant", "The earlier attempt is complete."),
+        ("user", prompt),
+    ]
+    assert payload["messages"][0].get("_active_turn_token") is None
+    assert payload["messages"][-1]["role"] == "assistant"
+    assert payload["messages"][-1]["_error"] is True
+
+
+def test_streaming_tool_limit_empty_result_without_exact_checkpoint_materializes_current_user(
+    tmp_path,
+    monkeypatch,
+):
+    prompt = "Fix the failing test."
+    prior = [
+        {"role": "user", "content": prompt, "timestamp": 1.1},
+        {"role": "assistant", "content": "The earlier attempt is complete."},
+    ]
+    result = {
+        "turn_exit_reason": "max_iterations_reached(30/30)",
+        "messages": [],
+    }
+
+    events, payload = _run_streaming_with_fake_agent(
+        tmp_path,
+        monkeypatch,
+        result,
+        prior_messages=prior,
+        prior_context_messages=prior,
+        msg_text=prompt,
+        pending_started_at=1.9,
+        current_turn_user_idx=0,
+    )
+
+    apperror_payloads = [event_payload for event, event_payload in events if event == "apperror"]
+    assert apperror_payloads, "expected tool-limit apperror"
+    assert apperror_payloads[-1]["type"] == "tool_limit_reached"
+    assert apperror_payloads[-1]["terminal_state"] == "tool_limit_reached"
+    assert not [event_payload for event, event_payload in events if event == "done"]
+    assert [(message["role"], message.get("content")) for message in payload["messages"][:-1]] == [
+        ("user", prompt),
+        ("assistant", "The earlier attempt is complete."),
+        ("user", prompt),
+    ]
+    assert payload["messages"][0].get("_active_turn_token") is None
+    assert payload["messages"][2]["_active_turn_token"] == build_active_turn_token(
+        "stream-tool-limit",
+        1.9,
+    )
+    assert payload["messages"][-1]["role"] == "assistant"
+    assert payload["messages"][-1]["_error"] is True