fix(#6611): make regeneration start atomic (#6677)

* fix(#6611): make regeneration transaction canonical

* fix(#6611): close cross-provider regeneration blockers

* fix(#6611): harden regeneration transaction boundaries

* fix(#6611): bind canonical context and fork lineage

* fix(#6611): restore active turn token stamping

* test(#6611): make regeneration regressions CI portable

* test(#6611): preserve the tracked regeneration error fixture

* fix(#6611): preserve extracted session harness contracts

* fix(regeneration): preserve concurrent and imported turn ownership (#6611)

* fix(regeneration): honor imported turn ownership safely (#6611)

* test(regeneration): cover both session lock outcomes (#6611)

* test(regeneration): clear unused transaction fixtures (#6611)

* fix(#6611): consume public active-turn marker

* ci: re-trigger (Playwright install hung ~2.5h on GH runners, warm-cache retry)

---------

Co-authored-by: nesquena-hermes <nesquena+hermes@gmail.com>
Co-authored-by: n <a@n>

api/gateway_chat.py

@@ -909,6 +909,7 @@ def _run_gateway_chat_streaming(
     *,
     model_provider=None,
     goal_related=False,
+    regeneration=False,
 ):
     """Bridge a WebUI chat turn through Hermes Gateway's API server.
 
@@ -1279,18 +1280,29 @@ def _run_gateway_chat_streaming(
             # same sort key; later transcript merges can then fall back to
             # role/content ordering instead of turn order.
             assistant_ts = now + 0.000001
-            user_msg = {"role": "user", "content": str(msg_text or ""), "timestamp": now}
             pending_source = getattr(s, "pending_user_source", None) or "webui"
-            if pending_source != "webui":
-                user_msg["_source"] = pending_source
-            if attachments:
-                user_msg["attachments"] = list(attachments)
+            from api.streaming import _active_turn_authority, _materialize_active_turn_user
+
+            active_turn_identity = _active_turn_authority(s, stream_id, msg_text)
+            user_msg = _materialize_active_turn_user(
+                active_turn_identity,
+                str(msg_text or ""),
+                pending_source,
+            )
+            user_msg["timestamp"] = float(
+                active_turn_identity.get("timestamp") or now
+            )
             assistant_msg = {"role": "assistant", "content": assistant_text, "timestamp": assistant_ts}
             saved_reasoning = STREAM_REASONING_TEXT.get(stream_id, "")
             if saved_reasoning:
                 assistant_msg["reasoning"] = saved_reasoning
             previous_messages = list(getattr(s, "messages", None) or [])
-            previous_context = list(getattr(s, "context_messages", None) or getattr(s, "messages", None) or [])
+            stored_context = getattr(s, "context_messages", None)
+            previous_context = list(
+                stored_context
+                if isinstance(stored_context, list) and (regeneration or stored_context)
+                else getattr(s, "messages", None) or []
+            )
             previous_process_wakeup_pause = dict(getattr(s, "process_wakeup_pause", {}) or {})
             # Stamp stable ids on the two new rows (shared with the display merge
             # below) so display and model-context copies share an id for the

api/helpers.py

@@ -19,6 +19,9 @@ _PUBLIC_MESSAGE_INTERNAL_FIELDS = frozenset({
     "_state_db_row_id",
     "_db_row_id",
     "state_db_row_id",
+    "_active_turn_token",
+    "_active_turn_user",
+    "_fork_child_turn",
 })
 
 
@@ -1114,8 +1117,14 @@ def scrub_internal_replay_fields(
     return result
 
 
-def _public_message_projection(message, *, _enabled: bool):
+def _public_message_projection(message, *, _enabled: bool, _active_turn_token=None):
     """Return one public transcript message without internal replay fields."""
+    is_active = (
+        isinstance(message, dict)
+        and message.get("role") == "user"
+        and _active_turn_token is not None
+        and message.get("_active_turn_token") == _active_turn_token
+    )
     message = scrub_internal_replay_fields([message], message_records=True)[0]
     if not isinstance(message, dict):
         return _redact_value(message, _enabled=_enabled)
@@ -1131,13 +1140,22 @@ def _public_message_projection(message, *, _enabled: bool):
             ]
         else:
             item[key] = _redact_value(value, _enabled=_enabled)
+    if is_active:
+        item["_active_turn_user"] = True
     return item
 
 
-def _redact_messages(messages, *, _enabled: bool):
+def _redact_messages(messages, *, _enabled: bool, _active_turn_token=None):
     if not isinstance(messages, list):
         return _redact_value(messages, _enabled=_enabled)
-    return [_public_message_projection(message, _enabled=_enabled) for message in messages]
+    return [
+        _public_message_projection(
+            message,
+            _enabled=_enabled,
+            _active_turn_token=_active_turn_token,
+        )
+        for message in messages
+    ]
 
 
 def _redact_tool_calls(tool_calls, *, _enabled: bool):
@@ -1152,10 +1170,12 @@ def _redact_nested_message_containers(value, *, _enabled: bool):
         return _redact_value(scrubbed, _enabled=_enabled)
     result = {}
     for key, child in scrubbed.items():
-        if key == "messages" and isinstance(child, list):
+        if key in {"messages", "context_messages"} and isinstance(child, list):
             result[key] = _redact_messages(child, _enabled=_enabled)
         elif key == "tool_calls" and isinstance(child, list):
             result[key] = _redact_tool_calls(child, _enabled=_enabled)
+        elif key == "runtime_journal_snapshot" and isinstance(child, dict):
+            result[key] = _redact_nested_message_containers(child, _enabled=_enabled)
         else:
             result[key] = _redact_value(child, _enabled=_enabled)
     return result
@@ -1193,30 +1213,21 @@ def _copy_json_value(value):
 
 
 def redact_session_data(session_dict: dict) -> dict:
-    """Redact credentials from message content, tool data, and session sidecars.
-
-    Applies to: messages[], tool_calls[], todo_state, runtime_journal_snapshot,
-    and title.
-    The underlying session file is not modified; redaction is response-layer only.
-
-    Reads the ``api_redact_enabled`` setting ONCE for the entire response and
-    threads it through to avoid hundreds of settings.json reads per session
-    payload (a 50-message session has hundreds of nested strings). When the
-    setting is disabled this is also a fast path: the recursion still walks
-    but every string returns early.
-    """
+    """Redact credentials in the public session response without mutation."""
     from api.config import load_settings
     _enabled = bool(load_settings().get("api_redact_enabled", True))
     if not isinstance(session_dict, dict):
         return {}
     result = {}
+    from api.process_event_utils import build_active_turn_token
+    _active_turn_token = build_active_turn_token(session_dict.get("active_stream_id"), session_dict.get("pending_started_at"))
     for key, value in session_dict.items():
         if key in _PUBLIC_MESSAGE_INTERNAL_FIELDS:
             continue
         if key == 'title' and isinstance(value, str):
             result[key] = _redact_text(value, _enabled=_enabled)
         elif key in {'messages', 'context_messages'}:
-            result[key] = _redact_messages(value, _enabled=_enabled)
+            result[key] = _redact_messages(value, _enabled=_enabled, _active_turn_token=_active_turn_token)
         elif key == 'tool_calls' and isinstance(value, list):
             result[key] = _redact_tool_calls(value, _enabled=_enabled)
         elif key in {'todo_state', 'runtime_journal_snapshot'}:

api/routes.py

@@ -2889,6 +2889,7 @@ from api.config import (
     ACTIVE_RUNS_LOCK,
     register_stream_owner,
     register_session_writeback_owner,
+    clear_session_writeback_owner_if_owned,
     stream_owner_session_id,
     unregister_stream_owner,
     CHAT_LOCK,
@@ -13208,6 +13209,26 @@ def handle_get(handler, parsed) -> bool:
             ):
                 raw["is_cli_session"] = False
                 raw["read_only"] = True
+            imported_turn_marker = any(
+                isinstance(row, dict) and row.get("_active_turn_token")
+                for row in _all_msgs
+            )
+            if (
+                not raw.get("read_only")
+                and not _truncated
+                and (not raw.get("is_cli_session") or imported_turn_marker)
+            ):
+                from api.session_ops import regeneration_authority, regeneration_state
+                canonical_state = regeneration_state(s)
+                revision = regeneration_authority(
+                    s,
+                    rows=canonical_state[0],
+                    context=canonical_state[1],
+                    full_transcript=True,
+                    canonical_state=canonical_state,
+                )
+                if revision:
+                    raw["regeneration_revision"] = revision
             redact = redact_session_data(raw)
             _t5 = _time.monotonic()
             if _diag: _diag.stage("t5_after_redact")
@@ -21763,6 +21784,8 @@ def _checkpoint_user_message_for_eager_session_save(s, msg: str, attachments, st
             latest_text = " ".join(str(latest.get("content") or "").split())
             msg_text = " ".join(str(msg or "").split())
             if latest_text == msg_text:
+                if str(source or "").strip().lower() == "fork":
+                    latest["_fork_child_turn"] = s.session_id
                 return
     user_msg = {"role": "user", "content": msg}
     from api.process_event_utils import build_active_turn_token, stamp_message_source
@@ -21772,6 +21795,8 @@ def _checkpoint_user_message_for_eager_session_save(s, msg: str, attachments, st
         source,
         active_turn_token=build_active_turn_token(getattr(s, "active_stream_id", None), started_at),
     )
+    if str(source or "").strip().lower() == "fork":
+        user_msg["_fork_child_turn"] = s.session_id
     if isinstance(started_at, (int, float)) and started_at > 0:
         user_msg["timestamp"] = float(started_at)
     if attachments:
@@ -21799,6 +21824,9 @@ def _provisional_title_from_prompt(prompt: str, fallback: str = "Untitled") -> s
     return title_from([{"role": "user", "content": text}], fallback) or fallback
 
 
+_RETAINED_CONTEXT_USER_UNSET = object()
+
+
 def _prepare_chat_start_session_for_stream(
     s,
     *,
@@ -21810,6 +21838,9 @@ def _prepare_chat_start_session_for_stream(
     stream_id: str,
     started_at: float | None = None,
     source: str = "webui",
+    retained_user=None,
+    retained_context_user=_RETAINED_CONTEXT_USER_UNSET,
+    defer_save: bool = False,
 ):
     """Persist chat-start state according to webui.session_save_mode.
 
@@ -21820,6 +21851,11 @@ def _prepare_chat_start_session_for_stream(
     a normal session message. Empty sessions are never saved here because this
     helper only runs after a non-empty message is validated.
     """
+    effective_source = (
+        "fork"
+        if str(getattr(s, "session_source", None) or "").strip().lower() == "fork"
+        else source
+    )
     s.workspace = workspace
     s.model = model
     s.model_provider = model_provider
@@ -21829,21 +21865,62 @@ def _prepare_chat_start_session_for_stream(
     s.pending_user_message = msg
     s.pending_attachments = attachments
     s.pending_started_at = started_at if started_at is not None else time.time()
-    s.pending_user_source = source
+    s.pending_user_source = effective_source
+    if retained_user is not None:
+        from api.process_event_utils import build_active_turn_token
+
+        retained_user["timestamp"] = s.pending_started_at
+        active_turn_token = build_active_turn_token(stream_id, s.pending_started_at)
+        retained_user["_active_turn_token"] = active_turn_token
+        if str(effective_source or "").strip().lower() == "fork":
+            retained_user["_fork_child_turn"] = s.session_id
+        if retained_context_user is not _RETAINED_CONTEXT_USER_UNSET:
+            if retained_context_user is not None and not any(
+                row is retained_context_user
+                for row in list(getattr(s, "context_messages", None) or [])
+            ):
+                raise RuntimeError("regeneration retained context row is not installed")
+            if isinstance(retained_context_user, dict):
+                retained_context_user["timestamp"] = s.pending_started_at
+                retained_context_user["_active_turn_token"] = active_turn_token
+                if str(effective_source or "").strip().lower() == "fork":
+                    retained_context_user["_fork_child_turn"] = s.session_id
+        else:
+            retained_id = retained_user.get("id") or retained_user.get("message_id")
+            retained_old_timestamp = retained_user.get("timestamp")
+            retained_old_content = retained_user.get("content")
+            for context_row in reversed(list(getattr(s, "context_messages", None) or [])):
+                if not isinstance(context_row, dict) or context_row.get("role") != "user":
+                    continue
+                context_id = context_row.get("id") or context_row.get("message_id")
+                id_match = retained_id is not None and context_id == retained_id
+                old_shape_match = (
+                    retained_old_timestamp is not None
+                    and context_row.get("timestamp") == retained_old_timestamp
+                    and context_row.get("content") == retained_old_content
+                )
+                if not (id_match or old_shape_match):
+                    continue
+                context_row["timestamp"] = s.pending_started_at
+                context_row["_active_turn_token"] = active_turn_token
+                if str(effective_source or "").strip().lower() == "fork":
+                    context_row["_fork_child_turn"] = s.session_id
+                break
     current_title = getattr(s, "title", None)
-    if _is_default_or_empty_session_title(current_title):
+    if retained_user is None and _is_default_or_empty_session_title(current_title):
         provisional_title = _provisional_title_from_prompt(msg, current_title or "Untitled")
         if provisional_title and not _is_default_or_empty_session_title(provisional_title):
             s.title = provisional_title
-    if get_webui_session_save_mode() == "eager":
+    if retained_user is None and get_webui_session_save_mode() == "eager":
         _checkpoint_user_message_for_eager_session_save(
             s,
             msg,
             attachments,
             s.pending_started_at,
-            source=source,
+            source=effective_source,
         )
-    s.save()
+    if not defer_save:
+        s.save()
 
 
 def _is_hidden_empty_session(s) -> bool:
@@ -21891,6 +21968,247 @@ def _active_stream_blocks_chat_start(session, stream_id: str | None) -> bool:
     return False
 
 
+def _start_regeneration_stream_locked(
+    s,
+    *,
+    turn,
+    workspace: str,
+    model: str,
+    model_provider,
+    normalized_model: bool,
+    diag,
+    goal_related: bool,
+    source: str,
+    moa_config,
+    backend_is_gateway: bool,
+):
+    """Commit a retained-row regeneration before releasing its real worker."""
+    from api.session_ops import (
+        RegenerationUnavailable,
+        apply_regeneration_plan,
+        plan_regeneration,
+        restore_regeneration_state,
+        snapshot_regeneration_state,
+    )
+
+    try:
+        plan = plan_regeneration(
+            s, expected_revision=turn.revision, lock_held=True
+        )
+        turn = plan.turn
+    except RegenerationUnavailable as exc:
+        return {
+            "error": str(exc),
+            "code": exc.code,
+            "_status": exc.status,
+        }
+    # Snapshot only after lock-held authority validation, before mutation.
+    snapshot = snapshot_regeneration_state(s)
+    if compression_recovery_payload_for_session(s):
+        clear_compression_recovery(s)
+    stream_id = uuid.uuid4().hex
+    gateway_starting = False
+    thread_started = False
+    save_attempted = False
+    accepted = False
+    journal_event = {}
+    release_worker = threading.Event()
+    abort_worker = threading.Event()
+    worker_thread = None
+
+    worker_target = (
+        _run_gateway_chat_streaming if backend_is_gateway else _run_agent_streaming
+    )
+    worker_kwargs = {
+        "model_provider": model_provider,
+        "goal_related": goal_related,
+    }
+    if backend_is_gateway:
+        worker_kwargs["regeneration"] = True
+    if moa_config and not backend_is_gateway:
+        worker_kwargs["moa_config"] = moa_config
+
+    def _gated_worker():
+        release_worker.wait()
+        if abort_worker.is_set():
+            return
+        worker_target(
+            s.session_id,
+            turn.message_text,
+            model,
+            workspace,
+            stream_id,
+            copy.deepcopy(turn.attachments),
+            **worker_kwargs,
+        )
+
+    def _cleanup_owned_start():
+        if goal_related:
+            STREAM_GOAL_RELATED.pop(stream_id, None)
+        with STREAMS_LOCK:
+            STREAMS.pop(stream_id, None)
+        unregister_stream_owner(stream_id)
+        clear_session_writeback_owner_if_owned(s.session_id, stream_id)
+        if gateway_starting:
+            try:
+                from api.gateway_chat import (
+                    _clear_gateway_run_starting,
+                    _finish_gateway_run_starting,
+                )
+
+                _finish_gateway_run_starting(stream_id)
+                _clear_gateway_run_starting(stream_id)
+            except Exception:
+                logger.debug(
+                    "Failed to clear compensated gateway start %s",
+                    stream_id,
+                    exc_info=True,
+                )
+
+    try:
+        applied, retained_context_user = apply_regeneration_plan(
+            s,
+            plan,
+            return_context_user=True,
+        )
+        if not applied:
+            restore_regeneration_state(s, snapshot)
+            return {
+                "error": "Session changed while regeneration was being prepared.",
+                "code": "stale_regeneration_revision",
+                "_status": 409,
+            }
+        retained_user = s.messages[-1]
+        msg = turn.message_text
+        attachments = copy.deepcopy(turn.attachments)
+        was_hidden_empty_session = _is_hidden_empty_session(s)
+        _prepare_chat_start_session_for_stream(
+            s,
+            msg=msg,
+            attachments=attachments,
+            workspace=workspace,
+            model=model,
+            model_provider=model_provider,
+            stream_id=stream_id,
+            source=turn.source,
+            retained_user=retained_user,
+            retained_context_user=retained_context_user,
+            defer_save=True,
+        )
+
+        diag.stage("turn_journal_submitted") if diag else None
+        from api.turn_journal import append_turn_journal_event
+
+        journal_event = append_turn_journal_event(
+            s.session_id,
+            {
+                "event": "submitted",
+                "stream_id": stream_id,
+                "role": "user",
+                "content": msg,
+                "attachments": attachments,
+                "workspace": workspace,
+                "model": model,
+                "model_provider": model_provider,
+                "created_at": s.pending_started_at,
+            },
+        )
+        diag.stage("stream_registration") if diag else None
+        stream = create_stream_channel()
+        register_stream_owner(stream_id, s.session_id)
+        with STREAMS_LOCK:
+            STREAMS[stream_id] = stream
+        if goal_related:
+            STREAM_GOAL_RELATED[stream_id] = True
+        if backend_is_gateway:
+            from api.gateway_chat import _mark_gateway_run_starting
+
+            gateway_starting = True
+            _mark_gateway_run_starting(stream_id)
+
+        diag.stage("worker_thread_start") if diag else None
+        worker_thread = threading.Thread(target=_gated_worker, daemon=True)
+        worker_thread.start()
+        thread_started = True
+        save_attempted = True
+        s.save()
+        accepted = True
+        set_last_workspace(workspace)
+        release_worker.set()
+    except Exception as exc:
+        abort_worker.set()
+        release_worker.set()
+        if (
+            thread_started
+            and worker_thread is not None
+            and callable(getattr(worker_thread, "join", None))
+        ):
+            worker_thread.join(timeout=1)
+        _cleanup_owned_start()
+        if accepted:
+            if journal_event:
+                try:
+                    append_turn_journal_event(
+                        s.session_id,
+                        {
+                            "event": "interrupted",
+                            "stream_id": stream_id,
+                            "turn_id": journal_event.get("turn_id"),
+                            "reason": "post_acceptance_workspace_failure",
+                        },
+                    )
+                except Exception:
+                    logger.warning("Failed to close accepted regeneration journal", exc_info=True)
+            exc._regeneration_accepted = True
+            raise
+        restore_regeneration_state(s, snapshot)
+        if save_attempted:
+            try:
+                s.save(touch_updated_at=False)
+            except Exception:
+                logger.exception(
+                    "Failed to persist compensated regeneration for %s",
+                    s.session_id,
+                )
+        if journal_event:
+            try:
+                append_turn_journal_event(
+                    s.session_id,
+                    {
+                        "event": "interrupted",
+                        "stream_id": stream_id,
+                        "turn_id": journal_event.get("turn_id"),
+                        "reason": "start_compensated",
+                    },
+                )
+            except Exception:
+                logger.warning(
+                    "Failed to close compensated turn journal event",
+                    exc_info=True,
+                )
+        raise
+
+    release_worker.set()
+    if was_hidden_empty_session:
+        publish_session_list_changed(
+            "session_new",
+            profile=getattr(s, "profile", None),
+            session_id=getattr(s, "session_id", None),
+        )
+    response = {
+        "stream_id": stream_id,
+        "session_id": s.session_id,
+        "pending_started_at": s.pending_started_at,
+        "turn_id": journal_event.get("turn_id"),
+        "title": s.title,
+    }
+    if normalized_model:
+        response["effective_model"] = model
+    if model_provider:
+        response["effective_model_provider"] = model_provider
+    return response
+
+
 def _active_run_stream_for_session(session_id: str | None) -> str | None:
     """Return a live worker stream for this session even if sidecar stream id is clear.
 
@@ -22024,6 +22342,7 @@ def _start_chat_stream_for_session(
     source: str = "webui",
     moa_config=None,
     external_runtime_owned: bool | None = None,
+    regeneration=None,
 ):
     """Persist pending state, register an SSE channel, and start an agent turn."""
     if external_runtime_owned is None:
@@ -22092,6 +22411,20 @@ def _start_chat_stream_for_session(
                         "_status": 409,
                     }
                 needs_stale_cleanup = False
+                if regeneration is not None:
+                    return _start_regeneration_stream_locked(
+                        s,
+                        turn=regeneration,
+                        workspace=workspace,
+                        model=model,
+                        model_provider=model_provider,
+                        normalized_model=normalized_model,
+                        diag=diag,
+                        goal_related=goal_related,
+                        source=source,
+                        moa_config=moa_config,
+                        backend_is_gateway=backend_is_gateway,
+                    )
                 stream_id = uuid.uuid4().hex
                 diag.stage("save_pending_state") if diag else None
                 was_hidden_empty_session = _is_hidden_empty_session(s)
@@ -22222,6 +22555,7 @@ def _chat_start_response_from_run_start(result):
         "effective_model",
         "effective_model_provider",
         "error",
+        "code",
         "active_stream_id",
         "_status",
     ):
@@ -22258,6 +22592,7 @@ def _start_run(
     diag=None,
     moa_config=None,
     gateway_chat_enabled: bool | None = None,
+    regeneration=None,
 ):
     """Shared start-run helper for /api/chat/start and start_session_turn.
 
@@ -22286,6 +22621,8 @@ def _start_run(
     )
 
     if runtime_adapter_enabled() or runtime_adapter_runner_enabled():
+        if regeneration is not None and runtime_adapter_runner_enabled():
+            return {"error": "Regeneration is not supported by the runner backend.", "code": "unsupported_regeneration_backend", "_status": 409}
         def _legacy_start_run(request: StartRunRequest) -> dict:
             return _start_chat_stream_for_session(
                 s,
@@ -22299,6 +22636,7 @@ def _start_run(
                 source=request.source or source,
                 moa_config=moa_config,
                 external_runtime_owned=gateway_chat_enabled,
+                regeneration=regeneration,
             )
 
         def _legacy_adapter_factory():
@@ -22340,6 +22678,7 @@ def _start_run(
         source=source,
         moa_config=moa_config,
         external_runtime_owned=gateway_chat_enabled,
+        regeneration=regeneration,
     )
 
 
@@ -22992,6 +23331,14 @@ def _handle_chat_start(handler, body, diag=None):
                 {"status": "suppressed", "reason": "silent_control_message"},
                 status=200,
             )
+        if body.get("regenerate") is True:
+            from api.runtime_adapter import runtime_adapter_runner_enabled
+
+            if runtime_adapter_runner_enabled():
+                return j(handler, {
+                    "error": "Regeneration is not supported by the runner backend.",
+                    "code": "unsupported_regeneration_backend",
+                }, status=409)
         # Reject a stale local Agent runtime before materialising, claiming, or
         # mutating any session state. Gateway-backed turns run in the gateway's
         # process and do not depend on this WebUI process's imported checkout.
@@ -23000,7 +23347,10 @@ def _handle_chat_start(handler, body, diag=None):
             return j(handler, stale_response, status=409)
         diag.stage("get_session") if diag else None
         try:
-            s = _get_or_materialize_session(body["session_id"], refresh_cli_messages=True)
+            s = _get_or_materialize_session(
+                body["session_id"],
+                refresh_cli_messages=body.get("regenerate") is not True,
+            )
         except KeyError:
             # No WebUI sidecar. If this is a foreign-origin session (CLI,
             # TUI, Desktop) with recoverable state.db messages, claim it by
@@ -23092,12 +23442,31 @@ def _handle_chat_start(handler, body, diag=None):
                 s.profile = requested_profile
             else:
                 return bad(handler, "Session not found", 404)
+        regeneration = None
+        if body.get("regenerate") is True:
+            if any(key in body for key in ("message", "attachments", "keep_count", "prompt", "prompt_index")):
+                return j(handler, {"error": "regeneration accepts only regeneration_revision", "code": "invalid_regeneration_request"}, status=400)
+            if not isinstance(body.get("regeneration_revision"), str):
+                return j(handler, {"error": "regeneration_revision is required", "code": "stale_regeneration_revision"}, status=409)
+            try:
+                from api.session_ops import plan_regeneration, RegenerationUnavailable
+                regeneration = plan_regeneration(
+                    s, expected_revision=body["regeneration_revision"]
+                )
+            except RegenerationUnavailable as exc:
+                return j(handler, {"error": str(exc), "code": exc.code}, status=exc.status)
+            msg = regeneration.turn.message_text
+            attachments = copy.deepcopy(regeneration.turn.attachments)[:20]
+        else:
+            msg = None
+            attachments = None
         diag.stage("normalize_message") if diag else None
-        msg = str(body.get("message", "")).strip()
+        msg = str(msg if msg is not None else body.get("message", "")).strip()
         if not msg:
             return bad(handler, "message is required")
         diag.stage("normalize_attachments") if diag else None
-        attachments = _normalize_chat_attachments(body.get("attachments") or [])[:20]
+        if attachments is None:
+            attachments = _normalize_chat_attachments(body.get("attachments") or [])[:20]
         recovery = compression_recovery_payload_for_session(s)
         if recovery and not attachments and is_generic_continuation_intent(msg):
             return j(
@@ -23113,7 +23482,10 @@ def _handle_chat_start(handler, body, diag=None):
             )
         diag.stage("resolve_workspace") if diag else None
         try:
-            workspace = _resolve_chat_workspace_with_recovery(s, body.get("workspace"))
+            if regeneration is not None:
+                workspace = _resolve_chat_workspace_for_regeneration(s, body.get("workspace"))
+            else:
+                workspace = _resolve_chat_workspace_with_recovery(s, body.get("workspace"))
         except WorkspaceBindingPersistenceError as e:
             return bad(handler, str(e), 500)
         except ValueError as e:
@@ -23156,7 +23528,7 @@ def _handle_chat_start(handler, body, diag=None):
         # compares). This survives same-model follow-up sends (the onchange marker
         # is one-shot) yet can't outlive a real switch.
         try:
-            if explicit_model_pick:
+            if explicit_model_pick and regeneration is None:
                 from api.models import model_explicit_pick_signature as _mk_sig
                 s.model_explicit_pick_signature = _mk_sig(model, model_provider)
         except Exception:
@@ -23215,6 +23587,7 @@ def _handle_chat_start(handler, body, diag=None):
             "route": "/api/chat/start",
             "diag": diag,
             "gateway_chat_enabled": gateway_chat_enabled,
+            "regeneration": regeneration,
         }
         if not gateway_chat_enabled and moa_config is not None:
             start_run_kwargs["moa_config"] = moa_config
@@ -23231,7 +23604,7 @@ def _handle_chat_start(handler, body, diag=None):
                 return restore_err
             return None
 
-        if recovery:
+        if recovery and regeneration is None:
             recovery_cleared_for_start = copy.deepcopy(recovery)
             clear_compression_recovery(s)
         try:
@@ -23239,8 +23612,9 @@ def _handle_chat_start(handler, body, diag=None):
                 s,
                 **start_run_kwargs,
             )
-        except Exception:
-            _restore_cleared_recovery()
+        except Exception as exc:
+            if not getattr(exc, "_regeneration_accepted", False):
+                _restore_cleared_recovery()
             raise
         # Map adapter-selection NotImplementedError (501) onto the legacy
         # bad-request response shape that this route exposed historically
@@ -23283,6 +23657,17 @@ def _resolve_chat_workspace_with_recovery(s, requested_workspace) -> str:
     return str(persisted.workspace)
 
 
+def _resolve_chat_workspace_for_regeneration(s, requested_workspace) -> str:
+    """Resolve regeneration's workspace without persisting before start acceptance."""
+    if requested_workspace not in (None, ""):
+        return str(resolve_trusted_workspace(requested_workspace))
+    workspace, _recovered = resolve_implicit_workspace_with_recovery(
+        getattr(s, "workspace", None),
+        get_last_workspace,
+    )
+    return str(workspace)
+
+
 def _normalize_chat_attachments(raw_attachments):
     """Normalize attachment payloads from the browser.
 

api/session_ops.py

@@ -9,17 +9,382 @@ from __future__ import annotations
 import json
 import logging
 import uuid
+import copy
+import hashlib
+import math
+from dataclasses import dataclass
+from contextlib import nullcontext
 from bisect import bisect_left
 from typing import Any
 
 from api.config import LOCK, _get_session_agent_lock
 from api.models import get_session, SESSIONS
+from api.agent_sessions import normalize_agent_session_source
 
 logger = logging.getLogger(__name__)
 
 AUTO_TITLE_LABELS = {'untitled', 'new chat'}
 
 
+class RegenerationUnavailable(Exception):
+    def __init__(self, code: str, status: int = 409, message: str | None = None):
+        super().__init__(message or code)
+        self.code = code
+        self.status = status
+
+
+def _regeneration_source_class(value):
+    raw = str(value or "").strip().lower()
+    if not raw:
+        return ""
+    if raw == "fork":
+        return "fork"
+    normalized = normalize_agent_session_source(raw).get("session_source")
+    return str(normalized or raw).strip().lower()
+
+
+def _regeneration_source_allowed(value):
+    return _regeneration_source_class(value) in {"webui", "fork"}
+
+
+def _selected_regeneration_turn_owned(session, row) -> bool:
+    """Accept only a final row whose provenance proves WebUI ownership."""
+    if getattr(session, "read_only", False) or not isinstance(row, dict):
+        return False
+    session_source = _regeneration_source_class(getattr(session, "session_source", None))
+    imported_session = bool(
+        getattr(session, "is_cli_session", False)
+        or session_source not in {"", "webui", "fork"}
+    )
+    raw_sources = (
+        getattr(session, "raw_source", None),
+        getattr(session, "source_tag", None),
+    )
+    row_source = row.get("_source") or row.get("source")
+    if session_source == "fork":
+        if any(source and not _regeneration_source_allowed(source) for source in raw_sources):
+            return False
+        if row_source and not _regeneration_source_allowed(row_source):
+            return False
+        return bool(
+            getattr(session, "parent_session_id", None)
+            and row.get("_fork_child_turn") == getattr(session, "session_id", None)
+        )
+    if imported_session:
+        token = row.get("_active_turn_token")
+        if not isinstance(token, str) or not token.strip() or ":" not in token:
+            return False
+        stream_id, started_at = token.rsplit(":", 1)
+        try:
+            started = float(started_at)
+        except (TypeError, ValueError):
+            return False
+        from api.process_event_utils import build_active_turn_token
+        if not math.isfinite(started) or started <= 0:
+            return False
+        if build_active_turn_token(stream_id, started) != token:
+            return False
+    else:
+        if any(source and not _regeneration_source_allowed(source) for source in raw_sources):
+            return False
+        if row_source and not _regeneration_source_allowed(row_source):
+            return False
+    return True
+
+
+@dataclass(frozen=True)
+class RegenerationTurn:
+    user_index: int
+    assistant_index: int
+    message: dict
+    message_text: str
+    attachments: list
+    source: str
+    message_count: int
+    revision: str
+    row_digest: str
+
+
+@dataclass(frozen=True)
+class RegenerationPlan:
+    canonical_rows: list
+    canonical_context: list
+    turn: RegenerationTurn
+    revision: str
+    row_digest: str
+    message_count: int
+    truncation_boundary: int
+
+
+def plan_regeneration(session, *, expected_revision=None, lock_held=False):
+    """Prepare one canonical display/context pair for a locked regeneration."""
+    lock_context = nullcontext() if lock_held else _get_session_agent_lock(session.session_id)
+    with lock_context:
+        rows, context = regeneration_state(session)
+        revision = regeneration_revision_for(rows, session=session, context=context)
+        if expected_revision is not None and expected_revision != revision:
+            raise RegenerationUnavailable("stale_regeneration_revision")
+        turn = resolve_regeneration_turn(
+            rows, session=session, expected_revision=revision,
+            lock_held=True, context=context,
+        )
+        return RegenerationPlan(
+            canonical_rows=copy.deepcopy(rows),
+            canonical_context=copy.deepcopy(context),
+            turn=turn,
+            revision=revision,
+            row_digest=turn.row_digest,
+            message_count=len(rows),
+            truncation_boundary=turn.user_index + 1,
+        )
+
+
+def apply_regeneration_plan(
+    session,
+    plan: RegenerationPlan,
+    *,
+    return_context_user: bool = False,
+):
+    """Install the prepared pair and truncate it without a second authority read."""
+    def _result(success, context_user=None):
+        return (success, context_user) if return_context_user else success
+
+    if not isinstance(plan, RegenerationPlan):
+        return _result(False)
+    rows = copy.deepcopy(plan.canonical_rows)
+    context = copy.deepcopy(plan.canonical_context)
+    if len(rows) != plan.message_count or plan.truncation_boundary != plan.turn.user_index + 1:
+        return _result(False)
+    if regeneration_revision_for(rows, session=session, context=context) != plan.revision:
+        return _result(False)
+    session.messages = rows
+    session.context_messages = context
+    current = session.messages[plan.turn.user_index]
+    if not isinstance(current, dict) or current.get("role") != "user":
+        return _result(False)
+    truncate_session_at_keep(session, plan.truncation_boundary)
+    prepared_context, context_boundary_index = truncate_context_for_display_keep(
+        context,
+        rows,
+        plan.truncation_boundary,
+        return_boundary_index=True,
+    )
+    session.context_messages = prepared_context if prepared_context or not context else context[: plan.truncation_boundary]
+    retained_context_user = None
+    if context_boundary_index is not None:
+        for context_row in reversed(session.context_messages[: context_boundary_index + 1]):
+            if isinstance(context_row, dict) and context_row.get("role") == "user":
+                retained_context_user = context_row
+                break
+    return _result(True, retained_context_user)
+
+
+def snapshot_regeneration_state(session):
+    return copy.deepcopy(session.__dict__)
+
+
+def restore_regeneration_state(session, snapshot):
+    session.__dict__.clear()
+    session.__dict__.update(copy.deepcopy(snapshot))
+
+
+def regeneration_revision_for(rows, *, session=None, context=None) -> str:
+    """Hash the canonical writable transcript and its aligned context."""
+    payload = json.dumps(
+        {
+            "session_id": str(getattr(session, "session_id", "") or "") if session is not None else "",
+            "messages": list(rows or []),
+            "context_messages": list(context or []),
+            "truncation_watermark": getattr(session, "truncation_watermark", None) if session is not None else None,
+            "truncation_boundary": getattr(session, "truncation_boundary", None) if session is not None else None,
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def regeneration_transcript(session, *, state_messages=None):
+    """Return the state.db-reconciled transcript used by every authority consumer."""
+    if state_messages is None:
+        return regeneration_state(session)[0]
+    from api.models import reconciled_state_db_messages_for_session
+    return reconciled_state_db_messages_for_session(session, state_messages=state_messages)
+
+
+def regeneration_context(session):
+    return regeneration_state(session)[1]
+
+
+def regeneration_state(session):
+    """Read one immutable state.db snapshot and reconcile both transcript views."""
+    from api.models import (
+        get_state_db_session_messages,
+        reconciled_state_db_messages_for_session,
+    )
+
+    state_messages = get_state_db_session_messages(
+        getattr(session, "session_id", None),
+        profile=getattr(session, "profile", None),
+    )
+    return (
+        reconciled_state_db_messages_for_session(
+            session,
+            state_messages=state_messages,
+        ),
+        reconciled_state_db_messages_for_session(
+            session,
+            prefer_context=True,
+            state_messages=state_messages,
+        ),
+    )
+
+
+def regeneration_revision(session) -> str:
+    rows, context = regeneration_state(session)
+    return regeneration_revision_for(
+        rows,
+        session=session,
+        context=context,
+    )
+
+
+def regeneration_authority(
+    session,
+    rows=None,
+    *,
+    context=None,
+    full_transcript=True,
+    canonical_state=None,
+):
+    """Mint a revision only for a complete, writable, canonical transcript."""
+    if not full_transcript:
+        return None
+    if getattr(session, "active_stream_id", None) or getattr(session, "pending_user_message", None):
+        return None
+    canonical_rows, canonical_context = canonical_state or regeneration_state(session)
+    rows = list(canonical_rows if rows is None else rows)
+    if not rows:
+        return None
+    if rows != canonical_rows:
+        return None
+    if context is not None and list(context or []) != canonical_context:
+        return None
+    try:
+        resolve_regeneration_turn(
+            canonical_rows,
+            session=session,
+            context=canonical_context,
+        )
+    except RegenerationUnavailable:
+        return None
+    return regeneration_revision_for(
+        canonical_rows,
+        session=session,
+        context=canonical_context,
+    )
+
+
+def resolve_regeneration_turn(
+    rows,
+    *,
+    session=None,
+    expected_revision=None,
+    lock_held=False,
+    context=None,
+):
+    """Select the current session's final complete local exchange under its lock."""
+    legacy_session_call = session is None and not isinstance(rows, (list, tuple))
+    legacy_context = None
+    if legacy_session_call:
+        session = rows
+        rows, legacy_context = regeneration_state(session)
+    lock_context = (
+        _get_session_agent_lock(session.session_id)
+        if legacy_session_call and not lock_held
+        else nullcontext()
+    )
+    with lock_context:
+        rows = list(rows or [])
+        if context is None:
+            context = legacy_context
+        if context is None:
+            _, context = regeneration_state(session)
+        context = list(context)
+        revision = regeneration_revision_for(rows, session=session, context=context)
+        if expected_revision is not None and expected_revision != revision:
+            raise RegenerationUnavailable("stale_regeneration_revision")
+        if getattr(session, "active_stream_id", None):
+            raise RegenerationUnavailable("session_active")
+        if getattr(session, "pending_user_message", None):
+            raise RegenerationUnavailable("session_active")
+        assistant_index = next(
+            (
+                index
+                for index in range(len(rows) - 1, -1, -1)
+                if isinstance(rows[index], dict)
+                and rows[index].get("role") == "assistant"
+                and _assistant_message_has_final_visible_text(rows[index])
+            ),
+            None,
+        )
+        if assistant_index is not None:
+            index = next(
+                (
+                    candidate
+                    for candidate in range(assistant_index - 1, -1, -1)
+                    if isinstance(rows[candidate], dict)
+                    and rows[candidate].get("role") == "user"
+                ),
+                None,
+            )
+        else:
+            index = None
+        if index is not None:
+            if any(
+                isinstance(row, dict) and row.get("role") == "user"
+                for row in rows[assistant_index + 1:]
+            ):
+                raise RegenerationUnavailable("no_regenerable_turn", 400)
+            if any(
+                isinstance(row, dict) and row.get("role") in {"assistant", "tool"}
+                for row in rows[assistant_index + 1:]
+            ):
+                raise RegenerationUnavailable("no_regenerable_turn", 400)
+            row = rows[index]
+            if not _selected_regeneration_turn_owned(session, row):
+                raise RegenerationUnavailable("regeneration_read_only", 403)
+            content = _extract_text(row.get("content", ""))
+            if content:
+                row_digest = hashlib.sha256(
+                    json.dumps(
+                        row,
+                        sort_keys=True,
+                        separators=(",", ":"),
+                        default=str,
+                    ).encode("utf-8")
+                ).hexdigest()
+                return RegenerationTurn(
+                    index,
+                    assistant_index,
+                    copy.deepcopy(row),
+                    content,
+                    copy.deepcopy(row.get("attachments") or []),
+                    str(row.get("_source") or "webui"),
+                    len(rows),
+                    revision,
+                    row_digest,
+                )
+    raise RegenerationUnavailable("no_regenerable_turn", 400)
+
+
+def _assistant_message_has_final_visible_text(message) -> bool:
+    from api.streaming import _assistant_message_has_final_visible_text as _has_final_text
+
+    return _has_final_text(message)
+
+
 def _live_active_stream_id(session) -> str | None:
     """Return session.active_stream_id ONLY if that stream is live in THIS
     process; else None.
@@ -119,16 +484,21 @@ def truncate_context_for_display_keep(
     context_messages: list | None,
     full_messages: list | None,
     keep: int,
+    *,
+    return_boundary_index: bool = False,
 ) -> list:
     """Align model context with display prefix ``full_messages[:keep]``."""
+    def _result(rows, boundary_index=None):
+        return (rows, boundary_index) if return_boundary_index else rows
+
     if keep <= 0:
-        return []
+        return _result([])
     ctx = context_messages if isinstance(context_messages, list) else []
     msgs = full_messages if isinstance(full_messages, list) else []
     if not ctx:
-        return []
+        return _result([])
     if len(msgs) == 0:
-        return []
+        return _result([])
     # Only the perfectly-parallel case (display and context row-for-row) can be
     # sliced at the raw display index. When the two arrays differ in length —
     # in EITHER direction — they have diverged and need alignment:
@@ -145,7 +515,7 @@ def truncate_context_for_display_keep(
     # strips unanswered tool_calls; gateway: it forwards no tool_calls/tool rows
     # at all), so we do not re-do that trimming here.
     if len(ctx) == len(msgs):
-        return ctx[:keep]
+        return _result(ctx[:keep], min(keep, len(ctx)) - 1)
 
     def _row_signature(row: Any) -> tuple[str, ...] | None:
         if not isinstance(row, dict):
@@ -374,8 +744,8 @@ def truncate_context_for_display_keep(
                 and isinstance(msgs[keep - 1], dict)
                 and msgs[keep - 1].get('role') == 'user'
             ):
-                return ctx[:last_kept + 1]
-            return ctx[:first_unkept]
+                return _result(ctx[:last_kept + 1], last_kept)
+            return _result(ctx[:first_unkept], first_unkept - 1)
         if last_kept is not None:
             ambiguous_first_unkept = ambiguous_matches[keep]
             if (
@@ -383,8 +753,8 @@ def truncate_context_for_display_keep(
                 and isinstance(msgs[keep - 1], dict)
                 and msgs[keep - 1].get('role') != 'user'
             ):
-                return ctx[:ambiguous_first_unkept]
-            return ctx[:last_kept + 1]
+                return _result(ctx[:ambiguous_first_unkept], ambiguous_first_unkept - 1)
+            return _result(ctx[:last_kept + 1], last_kept)
 
         # Both boundary rows were ambiguous/unmatched (common in large sessions
         # where context rows have lost their id/timestamp so the matcher can't
@@ -403,14 +773,15 @@ def truncate_context_for_display_keep(
             for i in range(keep - 1, -1, -1):
                 resolved = matches[i] if matches[i] is not None else ambiguous_matches[i]
                 if resolved is not None:
-                    return ctx[:resolved + 1]
+                    return _result(ctx[:resolved + 1], resolved)
 
     # Final fallback preserves #5096 behavior when alignment is unreliable
     # (no display row resolved to a context index, or keep >= len(msgs)).
     prefix_len = max(0, len(ctx) - len(msgs))
     prefix = ctx[:prefix_len]
     suffix = ctx[prefix_len:]
-    return prefix + suffix[:keep]
+    result = prefix + suffix[:keep]
+    return _result(result, len(result) - 1 if result else None)
 
 
 def truncate_session_at_keep(session, keep: int) -> tuple[int, int]:
@@ -618,7 +989,18 @@ def _extract_text(content: Any) -> str:
     if isinstance(content, list):
         parts = []
         for p in content:
-            if isinstance(p, dict) and p.get('type') == 'text':
-                parts.append(p.get('text', ''))
+            if not isinstance(p, dict):
+                continue
+            part_type = str(p.get('type') or '').lower()
+            if part_type not in ('', 'text', 'input_text', 'output_text'):
+                continue
+            part_text = (
+                p.get('text')
+                or p.get('content')
+                or p.get('input_text')
+                or p.get('output_text')
+                or ''
+            )
+            parts.append(str(part_text))
         return ' '.join(parts)
     return str(content)

api/streaming.py

@@ -156,6 +156,29 @@ def _session_payload_with_full_messages(session, *, tool_calls=None):
     attach_todo_state(raw, messages)
     if tool_calls is not None:
         raw['tool_calls'] = tool_calls
+    try:
+        from api.session_ops import (
+            regeneration_authority,
+            regeneration_state,
+        )
+        canonical_messages, canonical_context = regeneration_state(session)
+        revision = regeneration_authority(
+            session,
+            rows=canonical_messages,
+            context=canonical_context,
+            full_transcript=True,
+            canonical_state=(canonical_messages, canonical_context),
+        )
+        if revision:
+            messages = list(canonical_messages)
+            raw['messages'] = messages
+            raw['message_count'] = len(messages)
+            attach_todo_state(raw, messages)
+            raw['regeneration_revision'] = revision
+        else:
+            raw.pop('regeneration_revision', None)
+    except Exception:
+        raw.pop('regeneration_revision', None)
     return raw
 
 
@@ -1530,12 +1553,29 @@ def _clean_synthetic_control_messages_with_provenance(messages):
 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)
+    token = build_active_turn_token(stream_id, getattr(session, 'pending_started_at', None))
+    checkpoint = (
+        next(
+            (
+                copy.deepcopy(message)
+                for message in reversed(list(getattr(session, 'messages', None) or []))
+                if isinstance(message, dict)
+                and message.get('role') == 'user'
+                and message.get('_active_turn_token') == token
+            ),
+            None,
+        )
+        if token
+        else None
+    )
     return {
-        'token': build_active_turn_token(stream_id, getattr(session, 'pending_started_at', None)),
+        'session_id': getattr(session, 'session_id', None),
+        'token': token,
         '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 []),
+        'checkpoint': checkpoint,
         'current_turn_user_idx': None,
         'turn_id': '',
     }
@@ -1628,7 +1668,7 @@ def _mark_active_turn_checkpoint_in_history(messages, identity, msg_text, *, all
     if (
         not isinstance(message, dict)
         or message.get('role') != 'user'
-        or _normalize_user_text(message.get('content')) != _normalize_user_text(expected_text)
+        or _normalize_user_text(_message_text(message.get('content'))) != _normalize_user_text(expected_text)
     ):
         return messages, False
     _mark_active_turn_checkpoint(message, identity)
@@ -1675,19 +1715,24 @@ def _find_active_turn_checkpoint_index(result_messages, previous_context, identi
         if (
             isinstance(message, dict)
             and message.get('role') == 'user'
-            and _normalize_user_text(message.get('content')) == _normalize_user_text(expected_text)
+            and _normalize_user_text(_message_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,
-    }
+    checkpoint = identity.get('checkpoint') if isinstance(identity, dict) else None
+    message = (
+        copy.deepcopy(checkpoint)
+        if isinstance(checkpoint, dict) and checkpoint.get('role') == 'user'
+        else {
+            'role': 'user',
+            'content': identity.get('text') if isinstance(identity, dict) else msg_text,
+        }
+    )
     if isinstance(identity, dict):
-        if identity.get('timestamp') is not None:
+        if identity.get('timestamp') is not None and message.get('timestamp') is None:
             message['timestamp'] = identity['timestamp']
         if identity.get('attachments'):
             message['attachments'] = copy.deepcopy(identity['attachments'])
@@ -1696,8 +1741,15 @@ def _materialize_active_turn_user(identity, msg_text, source):
             identity.get('source') or source or 'webui',
             active_turn_token=identity.get('token'),
         )
+        if str(identity.get('source') or source or '').strip().lower() == 'fork':
+            child_session_id = identity.get('session_id')
+            if child_session_id:
+                message['_fork_child_turn'] = child_session_id
     else:
-        stamp_message_source(message, source)
+        stamp_message_source(
+            message,
+            source,
+        )
     return message
 
 
@@ -1713,7 +1765,18 @@ def _settle_current_turn_boundary(previous_context, result_messages, identity, m
         msg_text,
     )
     if _checkpoint_idx is not None:
-        _mark_active_turn_checkpoint(result_messages[_checkpoint_idx], identity)
+        existing_checkpoint = result_messages[_checkpoint_idx]
+        if isinstance(identity.get('checkpoint'), dict):
+            retained_checkpoint = _materialize_active_turn_user(identity, msg_text, source)
+            if (
+                retained_checkpoint.get('id') is None
+                and isinstance(existing_checkpoint, dict)
+                and existing_checkpoint.get('id') is not None
+            ):
+                retained_checkpoint['id'] = existing_checkpoint['id']
+            result_messages[_checkpoint_idx] = retained_checkpoint
+        else:
+            _mark_active_turn_checkpoint(existing_checkpoint, identity)
         return result_messages
     previous_context = list(previous_context or [])
     if _messages_have_prefix(result_messages, previous_context):
@@ -7379,7 +7442,7 @@ def _materialize_pending_user_turn_before_error(session) -> bool:
         except (TypeError, ValueError):
             return False
         return (
-            _normalize_user_text(existing.get('content')) == _normalize_user_text(pending_text)
+            _normalize_user_text(_message_text(existing.get('content'))) == _normalize_user_text(pending_text)
             and existing_ts == recovered_ts
             and existing_source == pending_source
             and list(existing.get('attachments') or []) == pending_attachments
@@ -7393,6 +7456,8 @@ def _materialize_pending_user_turn_before_error(session) -> bool:
         'timestamp': recovered_ts,
         '_recovered': True,
     }
+    if str(pending_source or '').strip().lower() == 'fork':
+        recovered['_fork_child_turn'] = session.session_id
     stamp_message_source(recovered, pending_source)
     if pending_attachments:
         recovered['attachments'] = pending_attachments

static/messages.js

@@ -1953,6 +1953,52 @@ async function send(){
   }finally{ _sendInProgress=false; _sendInProgressSid=null; }
 }
 
+async function startRegeneration(sessionId, regenerationRevision){
+  const sid=String(sessionId||'');
+  if(!sid||!regenerationRevision||!S.session||S.session.session_id!==sid)return;
+  const snapshot=Array.isArray(S.messages)?S.messages.slice():[];
+  let assistantIndex=-1;
+  let userIndex=-1;
+  for(let i=snapshot.length-1;i>=0;i--){
+    if(assistantIndex<0&&snapshot[i]?.role==='assistant'){assistantIndex=i;continue;}
+    if(assistantIndex>=0&&snapshot[i]?.role==='user'){userIndex=i;break;}
+  }
+  if(userIndex<0)return;
+  const retained=Object.assign({},snapshot[userIndex],{_pending:true});
+  S.messages=snapshot.slice(0,userIndex+1);
+  S.messages[userIndex]=retained;
+  renderMessages();setBusy(true);
+  if(typeof ensureLiveWorklogShell==='function')ensureLiveWorklogShell();
+  else if(typeof appendThinking==='function')appendThinking('',{pending:true});
+  try{
+    const response=await api('/api/chat/start',{method:'POST',body:JSON.stringify({
+      session_id:sid,regenerate:true,regeneration_revision:regenerationRevision
+    })});
+    if(!S.session||S.session.session_id!==sid)return;
+    const streamId=response&&response.stream_id;
+    if(!streamId)throw new Error('Regeneration did not start a stream.');
+    S.activeStreamId=streamId;
+    S.session.active_stream_id=streamId;
+    S.session.regeneration_revision=null;
+    if(typeof response.pending_started_at==='number')S.session.pending_started_at=response.pending_started_at;
+    if(response.title&&typeof applySessionTitleUpdate==='function')applySessionTitleUpdate(sid,response.title);
+    if(!INFLIGHT[sid])INFLIGHT[sid]={messages:S.messages.slice(),uploaded:[],toolCalls:[]};
+    markInflight(sid,streamId);
+    if(typeof saveInflightState==='function')saveInflightState(sid,{streamId,messages:S.messages.slice(),uploaded:[],toolCalls:[]});
+    if(typeof showLiveRunStatus==='function')showLiveRunStatus(sid,{startedAt:S.session.pending_started_at||Date.now()/1000});
+    if(typeof updateSendBtn==='function')updateSendBtn();
+    if(typeof renderSessionList==='function')void renderSessionList();
+    attachLiveStream(sid,streamId,[]);
+  }catch(error){
+    if(S.session&&S.session.session_id===sid){
+      S.messages=snapshot;delete INFLIGHT[sid];
+      if(typeof clearInflightState==='function')clearInflightState(sid);
+      removeThinking();renderMessages();setBusy(false);setComposerStatus('');
+    }
+    throw error;
+  }
+}
+
 const LIVE_STREAMS={};
 const _STREAM_NOTIFICATION_BACKGROUND={};
 
@@ -6139,7 +6185,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
           const _prevCost=(S.session&&S.session.estimated_cost)||0;
           const _prevCacheRead=(S.session&&S.session.cache_read_tokens)||0;
           const _prevCacheWrite=(S.session&&S.session.cache_write_tokens)||0;
-          S.session=d.session;S.messages=_carryForwardEphemeralTurnFields(S.messages||[], d.session.messages||[]);if(typeof _messagesTruncated!=='undefined')_messagesTruncated=!!d.session._messages_truncated;
+          S.session=d.session;S.messages=_carryForwardEphemeralTurnFields(S.messages||[], d.session.messages||[]);if(typeof _adoptRegenerationRevision==='function')_adoptRegenerationRevision(d.session);if(typeof _messagesTruncated!=='undefined')_messagesTruncated=!!d.session._messages_truncated;
           // #4720: reset _oldestIdx (full-load symmetry; keeps the #4613 anchor aligned).
           if(typeof _oldestIdx!=='undefined')_oldestIdx=d.session._messages_offset||0;
           S.messages=_filterRecoveryControlMessages(S.messages || []);
@@ -6578,6 +6624,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
           } else if(d.session&&typeof d.session==='object'){
             S.session=d.session;
             const _nextMsgs3018=(d.session.messages||[]).filter(m=>m&&m.role);
+            if(typeof _adoptRegenerationRevision==='function')_adoptRegenerationRevision(d.session);
             _attachProjectedAnchorSceneToLastAssistant(_nextMsgs3018);
             S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018);
             if(S.session&&S.session.session_id){
@@ -6818,6 +6865,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
             : (typeof _messageUserUnpinned!=='undefined' && _messageUserUnpinned));
         S.session=sessionPayload;
         const _nextMsgs3018=(sessionPayload.messages||[]).filter(m=>m&&m.role);
+        if(typeof _adoptRegenerationRevision==='function')_adoptRegenerationRevision(sessionPayload);
         _attachProjectedAnchorSceneToLastAssistant(_nextMsgs3018);
         S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018);
         if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session);
@@ -6990,6 +7038,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
           try{localStorage.setItem('hermes-webui-session',S.session.session_id);}catch(_){}
           if(typeof _setActiveSessionUrl==='function') _setActiveSessionUrl(S.session.session_id);
         }
+        if(typeof _adoptRegenerationRevision==='function')_adoptRegenerationRevision(session);
         const _markerOnlyAssistantError=_replaceMarkerOnlyAssistantWithStreamError(S.messages);
         if(_markerOnlyAssistantError&&typeof showToast==='function') showToast('No response received after context compression. Please retry.',5000,'error');
         const hasMessageToolMetadata=S.messages.some(m=>{

static/sessions.js

@@ -189,6 +189,15 @@ function _clearRememberedNewChatDraftSession(sid) {
   } catch (_) {}
 }
 
+function _adoptRegenerationRevision(sessionPayload){
+  if(!S||!S.session||!sessionPayload||typeof sessionPayload!=='object') return;
+  if(Object.prototype.hasOwnProperty.call(sessionPayload,'regeneration_revision')){
+    S.session.regeneration_revision=sessionPayload.regeneration_revision;
+  }else{
+    delete S.session.regeneration_revision;
+  }
+}
+
 async function _restoreRememberedNewChatDraftSession() {
   let sid = '';
   try { sid = localStorage.getItem(NEW_CHAT_DRAFT_SESSION_KEY) || ''; } catch (_) { sid = ''; }
@@ -1491,7 +1500,7 @@ async function newSession(flash, options={}){
     if(consumedExplicitModelOverride&&typeof _clearEmptyComposerModelOverride==='function'){
       _clearEmptyComposerModelOverride();
     }
-    S.session=data.session;S.messages=data.session.messages||[];
+    S.session=data.session;if(typeof _adoptRegenerationRevision==='function') _adoptRegenerationRevision(data.session);S.messages=data.session.messages||[];
     S._pendingSessionToolsets=null;
     if(_sessionSourceFilter==='cli') _sessionSourceFilter='webui';
     if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session);
@@ -1974,6 +1983,7 @@ async function loadSession(sid){
     return loadSession(continuationSid,{...opts,skipLineageResolve:true,skipContinuationResolve:true,force:true,_preloadNotified:true});
   }
   S.session=data.session;
+  if(typeof _adoptRegenerationRevision==='function') _adoptRegenerationRevision(data.session);
   if(typeof _clearEmptyComposerModelOverride==='function') _clearEmptyComposerModelOverride();
   // Loading a real existing session abandons any pre-session toolset override
   // staged on the empty composer before any deferred refresh work runs.
@@ -3209,6 +3219,7 @@ async function _ensureMessagesLoaded(sid, opts) {
     );
   }
   if(S.session&&S.session.session_id===sid){
+    if(typeof _adoptRegenerationRevision==='function') _adoptRegenerationRevision(data.session);
     S.session.message_count=Number(data.session.message_count || msgs.length);
     S.lastUsage={...(data.session.last_usage||S.lastUsage||{})};
     // Phase 2: the messages=1 response carries the canonical cold-load
@@ -3905,6 +3916,11 @@ async function _ensureAllMessagesLoaded() {
     _syncToolCallsForLoadedMessages(msgs, data.session.tool_calls);
     if (S.session && S.session.session_id === sid) {
       S.session.message_count = Number(data.session.message_count || msgs.length);
+      if (Object.prototype.hasOwnProperty.call(data.session, 'regeneration_revision')) {
+        S.session.regeneration_revision = data.session.regeneration_revision;
+      } else {
+        delete S.session.regeneration_revision;
+      }
     }
   } finally {
     _loadingOlder = false;

static/ui.js

@@ -491,7 +491,7 @@ async function startCompressionRecovery(btn){
     if(!sid) throw new Error('Compression recovery did not return a session.');
     try{localStorage.setItem('hermes-webui-session',sid);}catch(_){}
     if(typeof loadSession==='function') await loadSession(sid,{preserveActiveInput:false});
-    else if(data.session){S.session=data.session;S.messages=data.session.messages||[];syncTopbar();renderMessages();}
+    else if(data.session){S.session=data.session;if(typeof _adoptRegenerationRevision==='function')_adoptRegenerationRevision(data.session);S.messages=data.session.messages||[];syncTopbar();renderMessages();}
     if(typeof renderSessionList==='function') await renderSessionList();
     if(typeof _setActiveSessionUrl==='function') _setActiveSessionUrl(sid);
     if(typeof showToast==='function') showToast((data&&data.message)||'Started focused continuation.',3000,'success');
@@ -9982,6 +9982,7 @@ async function refreshSession() {
   try {
     const data = await api(`/api/session?session_id=${encodeURIComponent(S.session.session_id)}`);
     S.session = data.session;
+    if(typeof _adoptRegenerationRevision==='function') _adoptRegenerationRevision(data.session);
     S.messages = data.session.messages || [];
     _messagesTruncated = !!data.session._messages_truncated;
     _oldestIdx = data.session._messages_offset || 0;
@@ -10825,14 +10826,14 @@ function _activeTurnTokenMatches(msg, session){
  * same text twice in a row (a plain "继续" follow-up) legitimately gets two
  * identical user turns, and matching on text would swallow the new one. The
  * discriminator is therefore exact identity, never proximity: the active turn's
- * row either carries the server-stamped `_active_turn_token` (stream_id +
- * started_at — unique to this turn), or its timestamp equals `pending_started_at`
- * within a precision-only epsilon that absorbs float/state.db drift but never a
- * full second. A whole-second (or sub-second) mismatch is ambiguous and returns
- * null so the caller materializes the pending turn — the transient duplicate is
- * harmless, hiding a turn + moving its attachments is not. Text equality is
- * still required downstream, so a false match needs identical text AND an exact
- * identity signal.
+ * public row carries `_active_turn_user`, while private rows carry the server-
+ * stamped `_active_turn_token` (stream_id + started_at — unique to this turn),
+ * or their timestamp equals `pending_started_at` within a precision-only epsilon
+ * that absorbs float/state.db drift but never a full second. A whole-second (or
+ * sub-second) mismatch is ambiguous and returns null so the caller materializes
+ * the pending turn — the transient duplicate is harmless, hiding a turn + moving
+ * its attachments is not. Text equality is still required downstream, so a false
+ * match needs identical text AND an exact identity signal.
  */
 function _pendingActiveTurnUserMessage(messages, session){
   const startedAt=Number(session?.pending_started_at);
@@ -10842,6 +10843,8 @@ function _pendingActiveTurnUserMessage(messages, session){
     const msg=list[i];
     if(!msg||String(msg.role||'')!=='user') continue;
     if(typeof _isContextCompactionMessage==='function'&&_isContextCompactionMessage(msg)) continue;
+    // Public projections replace the private token with this authoritative marker.
+    if(msg._active_turn_user===true) return msg;
     // Unambiguous: the row carries the active turn's exact token
     // (stream_id + started_at) stamped by the server's eager-checkpoint path.
     if(typeof _activeTurnTokenMatches==='function'&&_activeTurnTokenMatches(msg,session)) return msg;
@@ -19286,30 +19289,25 @@ async function submitEdit(msgIdx, newText) {
 
 async function regenerateResponse(btn) {
   if(!S.session || S.busy) return;
-  const row = btn.closest('[data-msg-idx]');
-  if(!row) return;
-  const assistantIdx = parseInt(row.dataset.msgIdx, 10);
-  const absoluteKeepCount = _oldestIdx + assistantIdx;
+  const row=btn&&btn.closest&&btn.closest('[data-msg-idx]');
+  if(!row)return;
+  const clickedAbsoluteIndex=_oldestIdx+parseInt(row.dataset.msgIdx,10);
   const initialSid = S.session.session_id;
-  let lastUserText = '';
-  for(let i = assistantIdx - 1; i >= 0; i--) {
-    const m = S.messages[i];
-    if(m && m.role === 'user') { lastUserText = msgContent(m); break; }
-  }
-  if(!lastUserText) return;
   if(typeof _ensureAllMessagesLoaded==='function'){
     await _ensureAllMessagesLoaded();
   }
   if(!S.session || S.session.session_id !== initialSid) return;
+  if(!S.session.regeneration_revision){ setStatus(t('regen_failed')); return; }
+  let latestAssistantIndex=-1;
+  for(let i=S.messages.length-1;i>=0;i--){
+    if(S.messages[i]?.role==='assistant'){latestAssistantIndex=i;break;}
+  }
+  if(clickedAbsoluteIndex!==latestAssistantIndex){
+    setStatus(t('regen_failed'));
+    return;
+  }
   try {
-    await api('/api/session/truncate', {method:'POST', body:JSON.stringify({
-      session_id: initialSid,
-      keep_count: absoluteKeepCount
-    })});
-    S.messages = S.messages.slice(0, absoluteKeepCount);
-    renderMessages();
-    $('msg').value = lastUserText;
-    await send();
+    await startRegeneration(initialSid, S.session.regeneration_revision);
   } catch(e) { setStatus(t('regen_failed') + e.message); }
 }
 
@@ -21408,7 +21406,7 @@ async function promptNewFile(targetDir = S.currentDir || '.'){
       // System-minted session (#6022): explicit worktree:false — creating a
       // file from a blank page must not inherit the config worktree default.
       const r=await api('/api/session/new',{method:'POST',body:JSON.stringify({workspace:ws,worktree:false})});
-      if(r&&r.session){S._pendingSessionToolsets=null;S.session=r.session;S.messages=[];syncTopbar();renderMessages();await renderSessionList();}
+      if(r&&r.session){S._pendingSessionToolsets=null;S.session=r.session;if(typeof _adoptRegenerationRevision==='function') _adoptRegenerationRevision(r.session);S.messages=[];syncTopbar();renderMessages();await renderSessionList();}
     }catch(e){setStatus(t('create_failed')+e.message);return;}
   }
   if(!S.session)return;
@@ -21441,7 +21439,7 @@ async function promptNewFolder(targetDir = S.currentDir || '.'){
       // System-minted session (#6022): explicit worktree:false — creating a
       // folder from a blank page must not inherit the config worktree default.
       const r=await api('/api/session/new',{method:'POST',body:JSON.stringify({workspace:ws,worktree:false})});
-      if(r&&r.session){S._pendingSessionToolsets=null;S.session=r.session;S.messages=[];syncTopbar();renderMessages();await renderSessionList();}
+      if(r&&r.session){S._pendingSessionToolsets=null;S.session=r.session;if(typeof _adoptRegenerationRevision==='function') _adoptRegenerationRevision(r.session);S.messages=[];syncTopbar();renderMessages();await renderSessionList();}
     }catch(e){setStatus(t('folder_create_failed')+e.message);return;}
   }
   if(!S.session)return;

tests/_issue6611_fixture.py

@@ -0,0 +1,22 @@
+import json
+from pathlib import Path
+
+
+_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "issue6611_regeneration_rows.json"
+_EXPECTED_ROWS = [
+    {"role": "user", "content": "same prompt"},
+    {"role": "assistant", "content": "provider failed", "_error": True},
+]
+
+
+def load_issue6611_fixture():
+    fixture = json.loads(_FIXTURE_PATH.read_text(encoding="utf-8"))
+    assert set(fixture) == {"issue", "rows"}
+    assert fixture["issue"] == 6611
+    assert isinstance(fixture["rows"], list)
+    assert fixture["rows"] == _EXPECTED_ROWS
+    assert all(set(row) == {"role", "content"} for row in fixture["rows"][:1])
+    assert set(fixture["rows"][1]) == {"role", "content", "_error"}
+    assert [row["role"] for row in fixture["rows"]] == ["user", "assistant"]
+    assert fixture["rows"][1]["_error"] is True
+    return fixture

tests/fixtures/issue6611_regeneration_rows.json

@@ -0,0 +1,14 @@
+{
+  "issue": 6611,
+  "rows": [
+    {
+      "role": "user",
+      "content": "same prompt"
+    },
+    {
+      "role": "assistant",
+      "content": "provider failed",
+      "_error": true
+    }
+  ]
+}

tests/test_issue4728_new_chat_default_model.py

@@ -13,9 +13,9 @@ NODE = shutil.which("node")
 _DRIVER_SRC = r"""
 const fs = require('fs');
 
-function extractNewSession(src) {
-  const start = src.indexOf('async function newSession(');
-  if (start < 0) throw new Error('newSession not found');
+function extractFunction(src, signature) {
+  const start = src.indexOf(signature);
+  if (start < 0) throw new Error(signature + ' not found');
   let depth = 0;
   let bodyStart = src.indexOf('{', src.indexOf(')', start));
   for (let i = bodyStart; i < src.length; i++) {
@@ -26,7 +26,7 @@ function extractNewSession(src) {
       if (depth === 0) return src.slice(start, i + 1);
     }
   }
-  throw new Error('newSession body not closed');
+  throw new Error(signature + ' body not closed');
 }
 
 const src = fs.readFileSync(process.argv[2], 'utf8');
@@ -156,7 +156,8 @@ globalThis.api = async (url, opts) => {
   };
 };
 
-eval(extractNewSession(src));
+eval(extractFunction(src, 'function _adoptRegenerationRevision('));
+eval(extractFunction(src, 'async function newSession('));
 
 (async () => {
   await newSession(false, {});

tests/test_issue6611_regenerate_browser.py

@@ -0,0 +1,145 @@
+import json
+import subprocess
+from pathlib import Path
+
+from tests._issue6611_fixture import load_issue6611_fixture
+
+
+ROOT = Path(__file__).parents[1]
+
+
+def _start_regeneration_source():
+    source = (ROOT / "static" / "messages.js").read_text(encoding="utf-8")
+    start = source.index("async function startRegeneration(")
+    end = source.index("\nconst LIVE_STREAMS=", start)
+    return source[start:end]
+
+
+def _run_node(scenario, *, with_metadata=False):
+    function_source = _start_regeneration_source()
+    initial_messages = load_issue6611_fixture()["rows"]
+    if with_metadata:
+        initial_messages[0].update({"attachments": ["proof.txt"], "custom": "keep"})
+    messages_json = json.dumps(initial_messages)
+    script = f"""
+const result={{renders:0,busy:[],attached:[],bodies:[],thinking:0}};
+let S={{session:{{session_id:'s1',regeneration_revision:'rev-1'}},messages:{messages_json}}};
+const INFLIGHT={{}};
+function renderMessages(){{result.renders++;}}
+function setBusy(v){{result.busy.push(v);}}
+function ensureLiveWorklogShell(){{result.thinking++;}}
+function appendThinking(){{result.thinking++;}}
+function removeThinking(){{result.thinking--;}}
+function setComposerStatus(){{}}
+function clearInflightState(){{}}
+function markInflight(sid,streamId){{result.marked=[sid,streamId];}}
+function saveInflightState(){{}}
+function showLiveRunStatus(){{}}
+function updateSendBtn(){{}}
+function renderSessionList(){{}}
+function applySessionTitleUpdate(){{}}
+function attachLiveStream(sid,streamId,files){{result.attached.push([sid,streamId,files]);}}
+{function_source}
+async function api(_path, options){{
+  result.bodies.push(JSON.parse(options.body));
+  if('{scenario}'==='reject') throw new Error('typed rejection');
+  if('{scenario}'==='switch'){{
+    S.session={{session_id:'s2'}};
+    S.messages=[{{role:'user',content:'other session'}}];
+  }}
+  return {{stream_id:'stream-1',pending_started_at:123,title:'Title'}};
+}}
+(async()=>{{
+  try{{await startRegeneration('s1','rev-1');}}catch(error){{result.error=error.message;}}
+  result.messages=S.messages;
+  result.session=S.session;
+  result.inflight=Object.keys(INFLIGHT);
+  process.stdout.write(JSON.stringify(result));
+}})();
+"""
+    completed = subprocess.run(
+        ["node", "-e", script],
+        cwd=ROOT,
+        text=True,
+        capture_output=True,
+        check=True,
+    )
+    return json.loads(completed.stdout)
+
+
+def test_reporter_flow_keeps_one_prompt_and_adopts_one_accepted_stream():
+    result = _run_node("success", with_metadata=True)
+    assert [row["role"] for row in result["messages"]] == ["user"]
+    assert result["messages"][0]["content"] == "same prompt"
+    assert result["messages"][0]["attachments"] == ["proof.txt"]
+    assert result["messages"][0]["custom"] == "keep"
+    assert result["attached"] == [["s1", "stream-1", []]]
+    assert result["bodies"] == [
+        {"session_id": "s1", "regenerate": True, "regeneration_revision": "rev-1"}
+    ]
+
+
+def test_issue_artifact_regeneration_leaves_one_user_row():
+    fixture = load_issue6611_fixture()
+    assert fixture["issue"] == 6611
+    artifact_rows = fixture["rows"]
+    result = _run_node("success")
+    assert [row["role"] for row in result["messages"]].count("user") == 1
+    assert [row["role"] for row in result["messages"]] == ["user"]
+    assert result["messages"][0]["content"] == artifact_rows[0]["content"]
+
+
+def test_normal_full_load_adopts_and_clears_regeneration_revision():
+    source = (ROOT / "static" / "sessions.js").read_text(encoding="utf-8")
+    start = source.index("function _adoptRegenerationRevision(")
+    end = source.index("\n}\n\nasync function _restoreRememberedNewChatDraftSession", start) + 2
+    function_source = source[start:end]
+    script = f"""
+let S={{session:{{session_id:'s1',regeneration_revision:'old'}}}};
+{function_source}
+_adoptRegenerationRevision({{session_id:'s1',regeneration_revision:'fresh'}});
+if(S.session.regeneration_revision!=='fresh') throw new Error('fresh revision was not adopted');
+_adoptRegenerationRevision({{session_id:'s1'}});
+if(Object.prototype.hasOwnProperty.call(S.session,'regeneration_revision')) throw new Error('stale revision survived replacement');
+process.stdout.write('revision adoption ok');
+"""
+    result = subprocess.run(["node", "-e", script], cwd=ROOT, text=True, capture_output=True, check=True)
+    assert result.stdout == "revision adoption ok"
+
+
+def test_typed_rejection_restores_the_complete_local_transcript():
+    result = _run_node("reject")
+    assert [row["role"] for row in result["messages"]] == ["user", "assistant"]
+    assert result["error"] == "typed rejection"
+    assert result["busy"][-1] is False
+    assert result["attached"] == []
+
+
+def test_delayed_response_never_attaches_to_a_newly_selected_session():
+    result = _run_node("switch")
+    assert result["session"]["session_id"] == "s2"
+    assert result["messages"] == [{"role": "user", "content": "other session"}]
+    assert result["attached"] == []
+
+
+def test_regenerate_response_has_no_truncate_or_generic_send_reentry():
+    source = (ROOT / "static" / "ui.js").read_text(encoding="utf-8")
+    body = source[
+        source.index("async function regenerateResponse"):
+        source.index("// postProcessRenderedMessages")
+    ]
+    assert "startRegeneration(initialSid" in body
+    assert "/api/session/truncate" not in body
+    assert "await send(" not in body
+
+
+def test_regenerate_response_loads_the_full_session_before_requiring_revision():
+    source = (ROOT / "static" / "ui.js").read_text(encoding="utf-8")
+    body = source[
+        source.index("async function regenerateResponse"):
+        source.index("// postProcessRenderedMessages")
+    ]
+    assert "if(!S.session || S.busy || !S.session.regeneration_revision) return;" not in body
+    assert body.index("await _ensureAllMessagesLoaded()") < body.index(
+        "if(!S.session.regeneration_revision)"
+    )

tests/test_issue6611_regenerate_turn_identity.py

@@ -0,0 +1,63 @@
+import copy
+
+from api.models import Session
+from api.session_ops import resolve_regeneration_turn, regeneration_revision
+
+
+def _session():
+    return Session(
+        session_id="issue6611",
+        messages=[
+            {
+                "role": "user",
+                "content": "same prompt",
+                "id": "u1",
+                "_source": "webui",
+                "timestamp": 123,
+                "attachments": [
+                    {
+                        "name": "a.txt",
+                        "path": "C:/uploads/a.txt",
+                        "mime": "text/plain",
+                        "size": 7,
+                        "is_image": False,
+                        "upload_id": "upload-1",
+                    }
+                ],
+            },
+            {"role": "assistant", "content": "failed"},
+        ],
+        context_messages=[{"role": "user", "content": "same prompt", "id": "u1"}, {"role": "assistant", "content": "failed"}],
+    )
+
+
+def test_regenerate_errored_turn_yields_one_user_row():
+    session = _session()
+    turn = resolve_regeneration_turn(session)
+    assert [row["role"] for row in session.messages[: turn.user_index + 1]] == ["user"]
+    assert turn.message["id"] == "u1"
+
+
+def test_retained_row_preserves_attachments_and_identity():
+    session = _session()
+    turn = resolve_regeneration_turn(session)
+    retained = copy.deepcopy(turn.message)
+    assert retained["id"] == "u1"
+    assert retained["timestamp"] == 123
+    assert retained["attachments"] == [
+        {
+            "name": "a.txt",
+            "path": "C:/uploads/a.txt",
+            "mime": "text/plain",
+            "size": 7,
+            "is_image": False,
+            "upload_id": "upload-1",
+        }
+    ]
+    assert regeneration_revision(session)
+
+
+def test_generic_operations_have_no_regeneration_coordinate():
+    session = _session()
+    assert "regeneration_revision" not in session.__dict__
+    assert session.messages[-1]["content"] == "failed"

tests/test_issue6611_regeneration_authority.py

@@ -0,0 +1,436 @@
+import json
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from api.helpers import redact_session_data
+from api.models import Session
+from api.session_ops import (
+    RegenerationUnavailable,
+    regeneration_authority,
+    regeneration_revision_for,
+    regeneration_state,
+    resolve_regeneration_turn,
+)
+from api.streaming import _session_payload_with_full_messages
+from tests.js_source_extract import extract_function
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _session():
+    return Session(
+        session_id="authority6611",
+        messages=[{"role": "user", "content": "p", "_source": "webui"}, {"role": "assistant", "content": "a"}],
+        context_messages=[{"role": "user", "content": "p"}, {"role": "assistant", "content": "a"}],
+    )
+
+
+def test_all_terminal_payloads_carry_fresh_revision():
+    session = _session()
+    payload = _session_payload_with_full_messages(session)
+    assert payload["regeneration_revision"] == regeneration_authority(session, rows=session.messages)
+    assert payload["regeneration_revision"] == regeneration_revision_for(session.messages, session=session, context=session.context_messages)
+
+
+def test_get_revision_consumer_delegates_imported_ownership_to_shared_authority():
+    from pathlib import Path
+
+    source = (Path(__file__).parents[1] / "api" / "routes.py").read_text(encoding="utf-8")
+    get_start = source.index("imported_turn_marker = any(")
+    body = source[get_start:get_start + 900]
+    assert "regeneration_authority(" in body
+    assert "imported_turn_marker" in body
+
+
+def test_private_active_turn_token_never_public():
+    session = _session()
+    session.messages[-2]["_active_turn_token"] = "private"
+    public = redact_session_data(session.compact() | {"messages": session.messages})
+    assert "_active_turn_token" not in str(public)
+
+
+def test_private_active_turn_token_is_redacted_in_nested_context_and_journal():
+    session = _session()
+    session_dict = session.compact() | {
+        "messages": session.messages,
+        "context_messages": [
+            {
+                "role": "user",
+                "content": "p",
+                "_active_turn_token": "secret",
+                "_fork_child_turn": "authority6611",
+            }
+        ],
+        "runtime_journal_snapshot": {
+            "context_messages": [
+                {
+                    "role": "user",
+                    "content": "p",
+                    "_active_turn_token": "secret",
+                    "_fork_child_turn": "authority6611",
+                }
+            ]
+        },
+    }
+    public = redact_session_data(session_dict)
+    assert public["context_messages"][0].get("_active_turn_token") is None
+    assert public["context_messages"][0].get("_fork_child_turn") is None
+    assert public["runtime_journal_snapshot"]["context_messages"][0].get("_active_turn_token") is None
+    assert public["runtime_journal_snapshot"]["context_messages"][0].get("_fork_child_turn") is None
+
+
+def test_public_active_turn_marker_matches_only_the_active_user_row():
+    from api.process_event_utils import build_active_turn_token
+
+    session = _session()
+    session.active_stream_id = "stream-active"
+    session.pending_started_at = 123.0
+    token = build_active_turn_token(session.active_stream_id, session.pending_started_at)
+    session.messages[0]["_active_turn_token"] = token
+    session.messages.insert(0, {"role": "user", "content": "older", "_active_turn_token": "other"})
+    public = redact_session_data(
+        session.compact()
+        | {
+            "active_stream_id": session.active_stream_id,
+            "pending_started_at": session.pending_started_at,
+            "messages": session.messages,
+            "context_messages": [
+                {"role": "user", "content": "older", "_active_turn_token": "other"},
+                {"role": "user", "content": "p", "_active_turn_token": token},
+            ],
+        }
+    )
+    assert public["messages"][1]["_active_turn_user"] is True
+    assert "_active_turn_user" not in public["messages"][0]
+    assert public["context_messages"][1]["_active_turn_user"] is True
+    assert "_active_turn_token" not in str(public)
+
+
+def test_public_active_turn_marker_is_consumed_by_browser_with_timestamp_drift():
+    node = shutil.which("node")
+    if node is None:
+        pytest.skip("node is required for the browser projection probe")
+
+    from api.process_event_utils import build_active_turn_token
+
+    stream_id = "projection-browser-stream"
+    started_at = 123.5
+    token = build_active_turn_token(stream_id, started_at)
+    public = redact_session_data(
+        {
+            "active_stream_id": stream_id,
+            "pending_started_at": started_at,
+            "messages": [
+                {
+                    "role": "user",
+                    "content": "same prompt",
+                    "timestamp": 100.0,
+                    "_active_turn_token": token,
+                }
+            ],
+        }
+    )
+    projected = public["messages"][0]
+    assert projected["_active_turn_user"] is True
+    assert "_active_turn_token" not in projected
+
+    ui_source = (ROOT / "static" / "ui.js").read_text(encoding="utf-8")
+    functions = "\n".join(
+        extract_function(ui_source, name)
+        for name in (
+            "_messageTimestampSeconds",
+            "_activeTurnTokenMatches",
+            "_pendingActiveTurnUserMessage",
+        )
+    )
+    script = f"""
+const _PENDING_ACTIVE_TURN_TS_EPSILON=1e-6;
+{functions}
+const messages={json.dumps([projected])};
+const session={{active_stream_id:{json.dumps(stream_id)},pending_started_at:{started_at + 1}}};
+const selected=_pendingActiveTurnUserMessage(messages,session);
+if(selected!==messages[0]) throw new Error('projected active row was not selected');
+process.stdout.write(JSON.stringify({{marker:selected._active_turn_user,token:selected._active_turn_token||null}}));
+"""
+    result = subprocess.run([node, "-e", script], text=True, capture_output=True, check=True)
+    assert json.loads(result.stdout) == {"marker": True, "token": None}
+
+
+def test_parent_or_foreign_source_refuses_authority():
+    session = _session()
+    session.messages[-2]["_source"] = "cron"
+    assert regeneration_authority(session, rows=session.messages) is None
+    try:
+        resolve_regeneration_turn(session)
+    except RegenerationUnavailable as exc:
+        assert exc.code == "regeneration_read_only"
+    else:
+        raise AssertionError("foreign source was accepted")
+
+
+def test_writable_imported_session_accepts_only_a_marked_final_user_turn(monkeypatch, tmp_path):
+    from api.process_event_utils import build_active_turn_token
+    from api import models as models_api
+
+    session = _session()
+    session.session_source = "cli"
+    session.is_cli_session = True
+    session.raw_source = "cli"
+    session.source_tag = "cli"
+    session.messages[-2]["_source"] = "cli"
+    session.active_stream_id = "imported-stream"
+    session.pending_started_at = 123.0
+    token = build_active_turn_token(session.active_stream_id, session.pending_started_at)
+    session.active_stream_id = None
+    session.pending_started_at = None
+    session.messages[-2]["_active_turn_token"] = token
+    monkeypatch.setattr(models_api, "SESSION_DIR", tmp_path)
+    monkeypatch.setattr(models_api, "SESSION_INDEX_FILE", tmp_path / "_index.json")
+    session.save(touch_updated_at=False)
+    session = Session.load(session.session_id)
+    revision = regeneration_authority(session)
+    assert revision
+    assert resolve_regeneration_turn(session, expected_revision=revision).message["content"] == "p"
+    assert _session_payload_with_full_messages(session)["regeneration_revision"] == revision
+
+
+def test_get_and_terminal_consumers_emit_imported_marked_revision(monkeypatch):
+    from types import SimpleNamespace
+    from urllib.parse import urlparse
+    from api import models as models_api
+    from api import routes
+    from api.process_event_utils import build_active_turn_token
+
+    session = _session()
+    session.session_source = "desktop"
+    session.is_cli_session = True
+    session.raw_source = "desktop"
+    session.source_tag = "desktop"
+    session.messages[-2]["_source"] = "desktop"
+    session.messages[-2]["_active_turn_token"] = build_active_turn_token("imported-stream", 123.0)
+    revision = regeneration_authority(session)
+    captured = {}
+
+    def capture(_handler, data, status=200, **_kwargs):
+        captured["data"] = data
+        captured["status"] = status
+
+    monkeypatch.setattr(routes, "get_session", lambda *_args, **_kwargs: session)
+    monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *_args: True)
+    monkeypatch.setattr(routes, "_clear_stale_stream_state", lambda *_args: None)
+    monkeypatch.setattr(routes, "_session_requires_cli_metadata_lookup", lambda *_args: False)
+    monkeypatch.setattr(routes, "get_state_db_session_messages", lambda *_args, **_kwargs: [])
+    monkeypatch.setattr(models_api, "get_state_db_session_messages", lambda *_args, **_kwargs: [])
+    monkeypatch.setattr(routes, "_resolve_effective_session_model_for_display", lambda *_args: None)
+    monkeypatch.setattr(routes, "_resolve_effective_session_model_provider_for_display", lambda *_args: None)
+    monkeypatch.setattr(routes, "j", capture)
+
+    routes.handle_get(
+        SimpleNamespace(_safe_webui_print=lambda *_args: None),
+        urlparse(f"/api/session?session_id={session.session_id}&messages=1&resolve_model=0"),
+    )
+
+    assert captured["status"] == 200
+    assert captured["data"]["session"]["regeneration_revision"] == revision
+    assert _session_payload_with_full_messages(session)["regeneration_revision"] == revision
+
+
+def test_get_and_terminal_consumers_omit_imported_unowned_revision(monkeypatch):
+    from types import SimpleNamespace
+    from urllib.parse import urlparse
+    from api import models as models_api
+    from api import routes
+
+    session = _session()
+    session.session_source = "desktop"
+    session.is_cli_session = True
+    session.raw_source = "desktop"
+    session.source_tag = "desktop"
+    session.messages[-2]["_source"] = "desktop"
+    captured = {}
+
+    def capture(_handler, data, status=200, **_kwargs):
+        captured["data"] = data
+        captured["status"] = status
+
+    monkeypatch.setattr(routes, "get_session", lambda *_args, **_kwargs: session)
+    monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *_args: True)
+    monkeypatch.setattr(routes, "_clear_stale_stream_state", lambda *_args: None)
+    monkeypatch.setattr(routes, "_session_requires_cli_metadata_lookup", lambda *_args: False)
+    monkeypatch.setattr(routes, "get_state_db_session_messages", lambda *_args, **_kwargs: [])
+    monkeypatch.setattr(models_api, "get_state_db_session_messages", lambda *_args, **_kwargs: [])
+    monkeypatch.setattr(routes, "_resolve_effective_session_model_for_display", lambda *_args: None)
+    monkeypatch.setattr(routes, "_resolve_effective_session_model_provider_for_display", lambda *_args: None)
+    monkeypatch.setattr(routes, "j", capture)
+
+    routes.handle_get(
+        SimpleNamespace(_safe_webui_print=lambda *_args: None),
+        urlparse(f"/api/session?session_id={session.session_id}&messages=1&resolve_model=0"),
+    )
+
+    assert captured["status"] == 200
+    assert "regeneration_revision" not in captured["data"]["session"]
+    assert "regeneration_revision" not in _session_payload_with_full_messages(session)
+
+
+@pytest.mark.parametrize(
+    "marker,read_only,expected",
+    [(None, False, None), ("bad-token", False, None), ("valid-earlier", False, None), ("valid-final", True, None)],
+)
+def test_imported_turn_matrix_rejects_unowned_or_read_only_rows(marker, read_only, expected):
+    from api.process_event_utils import build_active_turn_token
+
+    session = _session()
+    session.session_source = "desktop"
+    session.is_cli_session = True
+    session.read_only = read_only
+    valid = build_active_turn_token("imported-stream", 123.0)
+    if marker == "valid-earlier":
+        session.messages[-2]["_active_turn_token"] = valid
+        session.messages.append({"role": "user", "content": "foreign", "_source": "desktop"})
+        session.messages.append({"role": "assistant", "content": "foreign answer"})
+    elif marker == "valid-final":
+        session.messages[-2]["_active_turn_token"] = valid
+    elif marker:
+        session.messages[-2]["_active_turn_token"] = marker
+    assert regeneration_authority(session) is expected
+    assert "regeneration_revision" not in _session_payload_with_full_messages(session)
+
+
+def test_fork_child_has_regeneration_authority():
+    session = _session()
+    session.session_source = "fork"
+    session.parent_session_id = "parent-6611"
+    session.messages[-2]["_fork_child_turn"] = session.session_id
+    revision = regeneration_authority(session)
+    assert revision
+    assert resolve_regeneration_turn(session, expected_revision=revision).source == "webui"
+
+
+def test_fork_of_fork_copied_parent_marker_refuses_authority():
+    session = _session()
+    session.session_source = "fork"
+    session.parent_session_id = "parent-6611"
+    session.messages[-2]["_fork_child_turn"] = "parent-6611"
+    assert regeneration_authority(session) is None
+    try:
+        resolve_regeneration_turn(session)
+    except RegenerationUnavailable as exc:
+        assert exc.code == "regeneration_read_only"
+    else:
+        raise AssertionError("parent fork marker was accepted by child")
+
+
+def test_current_fork_child_materialization_binds_and_accepts_its_new_turn(monkeypatch):
+    from api import routes
+
+    session = Session(
+        session_id="fork-child-current-6611",
+        messages=[
+            {"role": "user", "content": "parent prompt", "_fork_child_turn": "parent-6611"},
+            {"role": "assistant", "content": "parent answer"},
+        ],
+        context_messages=[
+            {"role": "user", "content": "parent prompt"},
+            {"role": "assistant", "content": "parent answer"},
+        ],
+        session_source="fork",
+        parent_session_id="parent-6611",
+    )
+    monkeypatch.setattr(routes, "register_session_writeback_owner", lambda *_args: None)
+    monkeypatch.setattr(routes, "get_webui_session_save_mode", lambda: "eager")
+    routes._prepare_chat_start_session_for_stream(
+        session,
+        msg="current child prompt",
+        attachments=[],
+        workspace="C:/workspace",
+        model="model",
+        model_provider="provider",
+        stream_id="fork-current-stream",
+        started_at=123.0,
+        defer_save=True,
+    )
+    assert session.messages[-1]["_fork_child_turn"] == session.session_id
+    session.messages.append({"role": "assistant", "content": "current answer"})
+    session.active_stream_id = None
+    session.pending_user_message = None
+    session.pending_attachments = []
+    session.pending_started_at = None
+    session.pending_user_source = None
+    revision = regeneration_authority(session)
+    assert revision
+    assert resolve_regeneration_turn(session, expected_revision=revision).message["content"] == "current child prompt"
+
+
+def test_fork_without_child_lineage_refuses_authority():
+    session = _session()
+    session.session_source = "fork"
+    session.parent_session_id = None
+    assert regeneration_authority(session) is None
+    try:
+        resolve_regeneration_turn(session)
+    except RegenerationUnavailable as exc:
+        assert exc.code == "regeneration_read_only"
+    else:
+        raise AssertionError("parent-only fork state was accepted")
+
+
+def test_authority_withholds_a_noncanonical_display_projection():
+    session = _session()
+    canonical_rows, canonical_context = regeneration_state(session)
+    projected_rows = canonical_rows + [{"role": "tool", "content": "stitched parent"}]
+    assert regeneration_authority(
+        session,
+        rows=projected_rows,
+        context=canonical_context,
+    ) is None
+
+
+def test_terminal_payload_embeds_the_rows_it_hashes(monkeypatch):
+    session = _session()
+    canonical_rows = [
+        {"role": "user", "content": "recovered", "_source": "webui"},
+        {"role": "assistant", "content": "answer"},
+    ]
+    canonical_context = list(canonical_rows)
+    monkeypatch.setattr(
+        "api.session_ops.regeneration_state",
+        lambda _session: (canonical_rows, canonical_context),
+    )
+    payload = _session_payload_with_full_messages(session)
+    assert payload["messages"] == canonical_rows
+    assert payload["message_count"] == len(canonical_rows)
+    assert payload["regeneration_revision"] == regeneration_authority(
+        session,
+        rows=canonical_rows,
+        context=canonical_context,
+    )
+
+
+def test_recovered_display_context_pair_survives_local_and_gateway_apply(monkeypatch):
+    session = _session()
+    canonical_rows = [
+        {"role": "user", "content": "recovered", "id": "u-recovered", "_source": "webui"},
+        {"role": "assistant", "content": "failed"},
+    ]
+    canonical_context = [
+        {"role": "system", "content": "recovered context only"},
+        *canonical_rows,
+    ]
+    monkeypatch.setattr(
+        "api.session_ops.regeneration_state",
+        lambda _session: (canonical_rows, canonical_context),
+    )
+    from api.session_ops import apply_regeneration_plan, plan_regeneration
+
+    plan = plan_regeneration(session)
+    assert apply_regeneration_plan(session, plan)
+    assert session.messages == canonical_rows[:1]
+    assert any(row.get("content") == "recovered" for row in session.context_messages)
+    payload = _session_payload_with_full_messages(session)
+    assert payload["messages"] == canonical_rows
+    assert payload["message_count"] == len(canonical_rows)

tests/test_issue6611_regeneration_settlement.py

@@ -0,0 +1,67 @@
+import copy
+import pytest
+
+from api.models import Session
+from api.session_ops import RegenerationUnavailable, apply_regeneration_plan, plan_regeneration, regeneration_revision
+
+
+def _session():
+    return Session(
+        session_id="settlement6611",
+        messages=[
+            {"role": "user", "content": "old", "id": "u0", "_source": "webui"},
+            {"role": "assistant", "content": "old answer"},
+            {"role": "user", "content": "latest", "id": "u1", "_source": "webui"},
+            {"role": "assistant", "content": "error"},
+        ],
+        context_messages=[
+            {"role": "user", "content": "old", "id": "u0"},
+            {"role": "assistant", "content": "old answer"},
+            {"role": "user", "content": "latest", "id": "u1"},
+            {"role": "assistant", "content": "error"},
+        ],
+    )
+
+
+def test_recovered_prefix_survives_regeneration_truncation():
+    session = _session()
+    plan = plan_regeneration(session)
+    assert apply_regeneration_plan(session, plan)
+    assert [row["content"] for row in session.messages] == ["old", "old answer", "latest"]
+
+
+def test_stale_revision_rejects_without_mutation():
+    session = _session()
+    before = copy.deepcopy(session.__dict__)
+    with pytest.raises(RegenerationUnavailable):
+        plan_regeneration(session, expected_revision="stale-revision")
+    assert session.__dict__ == before
+
+
+def test_double_fire_revision_changes_after_winner():
+    session = _session()
+    plan = plan_regeneration(session)
+    assert apply_regeneration_plan(session, plan)
+    assert regeneration_revision(session) != plan.revision
+
+
+def test_trailing_tool_state_is_not_regenerable_as_an_older_exchange():
+    session = Session(
+        session_id="settlement6611-tool-tail",
+        messages=[
+            {"role": "user", "content": "prompt", "_source": "webui"},
+            {"role": "assistant", "content": "visible answer"},
+            {"role": "tool", "content": "unfinished tool result"},
+        ],
+        context_messages=[
+            {"role": "user", "content": "prompt"},
+            {"role": "assistant", "content": "visible answer"},
+            {"role": "tool", "content": "unfinished tool result"},
+        ],
+    )
+    try:
+        plan_regeneration(session)
+    except RegenerationUnavailable as exc:
+        assert exc.code == "no_regenerable_turn"
+    else:
+        raise AssertionError("trailing tool state was treated as a completed exchange")

tests/test_issue6611_regeneration_transaction.py

@@ -0,0 +1,573 @@
+import copy
+import queue
+import threading
+from types import SimpleNamespace
+from pathlib import Path
+
+import pytest
+
+from api.models import Session
+from api.session_ops import (
+    RegenerationUnavailable,
+    RegenerationPlan,
+    apply_regeneration_plan,
+    plan_regeneration,
+    restore_regeneration_state,
+    snapshot_regeneration_state,
+)
+
+from tests._issue6611_fixture import load_issue6611_fixture
+
+
+def _session():
+    rows = [
+        {"role": "user", "content": "prompt", "id": "u1", "_source": "webui"},
+        {"role": "assistant", "content": "failed"},
+    ]
+    return Session(
+        session_id="transaction6611",
+        messages=copy.deepcopy(rows),
+        context_messages=copy.deepcopy(rows),
+        workspace="C:/workspace",
+    )
+
+
+def test_plan_installs_rows_and_context_as_one_prepared_pair():
+    session = _session()
+    plan = plan_regeneration(session)
+    assert isinstance(plan, RegenerationPlan)
+    assert plan.canonical_rows == session.messages
+    assert plan.canonical_context == session.context_messages
+    applied, retained_context_user = apply_regeneration_plan(
+        session,
+        plan,
+        return_context_user=True,
+    )
+    assert applied
+    assert retained_context_user is session.context_messages[0]
+    assert session.messages == plan.canonical_rows[: plan.truncation_boundary]
+    assert session.context_messages == plan.canonical_context[: plan.truncation_boundary]
+    assert any(row.get("content") == "prompt" for row in session.context_messages)
+    assert [row["role"] for row in session.messages] == ["user"]
+
+
+def test_apply_consumes_prepared_pair_without_ambient_authority_read(monkeypatch):
+    session = _session()
+    plan = plan_regeneration(session)
+    monkeypatch.setattr(
+        "api.session_ops.regeneration_state",
+        lambda _session: (_ for _ in ()).throw(AssertionError("ambient read")),
+    )
+    session.messages[0]["content"] = "changed"
+    assert apply_regeneration_plan(session, plan)
+    assert session.messages == plan.canonical_rows[: plan.truncation_boundary]
+
+
+def test_complete_session_snapshot_restores_every_attribute():
+    session = _session()
+    session.compression_state = {"marker": "kept"}
+    session._anchor_scene_index = 7
+    before = copy.deepcopy(session.__dict__)
+    snapshot = snapshot_regeneration_state(session)
+    session.messages.clear()
+    session.context_messages.clear()
+    session.compression_state["marker"] = "changed"
+    session._anchor_scene_index = 99
+    restore_regeneration_state(session, snapshot)
+    assert session.__dict__ == before
+
+
+def test_persisted_preacceptance_rollback_restores_exact_snapshot(monkeypatch):
+    session = _session()
+    snapshot = snapshot_regeneration_state(session)
+    persisted = []
+    monkeypatch.setattr(Session, "save", lambda self, **_kwargs: persisted.append(copy.deepcopy(self.__dict__)))
+    session.active_stream_id = "stale"
+    session.pending_user_message = "changed"
+    restore_regeneration_state(session, snapshot)
+    session.save(touch_updated_at=False)
+    assert persisted == [snapshot]
+
+
+def test_noop_rejection_does_not_need_persisted_rollback():
+    session = _session()
+    snapshot = snapshot_regeneration_state(session)
+    restore_regeneration_state(session, snapshot)
+    assert session.__dict__ == snapshot
+
+
+def test_locked_stale_plan_does_not_restore_a_request_time_snapshot(monkeypatch):
+    from api import routes
+
+    session = _session()
+    session.active_stream_id = "accepted-after-browser-validation"
+    current_state = copy.deepcopy(session.__dict__)
+
+    def stale_plan(*_args, **_kwargs):
+        raise RegenerationUnavailable("stale_regeneration_revision")
+
+    monkeypatch.setattr("api.session_ops.plan_regeneration", stale_plan)
+    result = routes._start_regeneration_stream_locked(
+        session,
+        turn=SimpleNamespace(revision="old-revision"),
+        workspace="C:/workspace",
+        model="model",
+        model_provider="provider",
+        normalized_model=False,
+        diag=None,
+        goal_related=False,
+        source="webui",
+        moa_config=None,
+        backend_is_gateway=False,
+    )
+    assert result["code"] == "stale_regeneration_revision"
+    assert session.__dict__ == current_state
+
+
+def test_locked_unexpected_plan_error_does_not_restore_a_request_time_snapshot(monkeypatch):
+    from api import routes
+
+    session = _session()
+    session.active_stream_id = "accepted-after-browser-validation"
+    current_state = copy.deepcopy(session.__dict__)
+
+    def unexpected_plan(*_args, **_kwargs):
+        raise RuntimeError("plan failed")
+
+    monkeypatch.setattr("api.session_ops.plan_regeneration", unexpected_plan)
+    with pytest.raises(RuntimeError, match="plan failed"):
+        routes._start_regeneration_stream_locked(
+            session,
+            turn=SimpleNamespace(revision="old-revision"),
+            workspace="C:/workspace",
+            model="model",
+            model_provider="provider",
+            normalized_model=False,
+            diag=None,
+            goal_related=False,
+            source="webui",
+            moa_config=None,
+            backend_is_gateway=False,
+    )
+    assert session.__dict__ == current_state
+
+
+def test_locked_preacceptance_exception_restores_the_transaction_snapshot(monkeypatch):
+    from api import routes
+
+    session = _session()
+    plan = plan_regeneration(session)
+    before = copy.deepcopy(session.__dict__)
+
+    def fail_prepare(*_args, **_kwargs):
+        raise RuntimeError("prepare failed")
+
+    monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", fail_prepare)
+    with pytest.raises(RuntimeError, match="prepare failed"):
+        routes._start_regeneration_stream_locked(
+            session,
+            turn=plan.turn,
+            workspace="C:/workspace",
+            model="model",
+            model_provider="provider",
+            normalized_model=False,
+            diag=None,
+            goal_related=False,
+            source="webui",
+            moa_config=None,
+            backend_is_gateway=False,
+        )
+    assert session.__dict__ == before
+
+
+def test_locked_postacceptance_workspace_exception_does_not_restore_turn(monkeypatch):
+    from api import routes
+    import api.turn_journal as turn_journal
+
+    session = _session()
+    plan = plan_regeneration(session)
+    monkeypatch.setattr(routes, "register_session_writeback_owner", lambda *_args: None)
+    monkeypatch.setattr(routes, "clear_session_writeback_owner_if_owned", lambda *_args: None)
+    monkeypatch.setattr(routes, "register_stream_owner", lambda *_args: None)
+    monkeypatch.setattr(routes, "unregister_stream_owner", lambda *_args: None)
+    monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
+    monkeypatch.setattr(turn_journal, "append_turn_journal_event", lambda *_args, **_kwargs: {"turn_id": "turn-6611"})
+    monkeypatch.setattr(Session, "save", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda *_args: (_ for _ in ()).throw(RuntimeError("workspace failed")))
+
+    class FakeThread:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def start(self):
+            return None
+
+        def join(self, timeout=None):
+            return None
+
+    monkeypatch.setattr(routes.threading, "Thread", FakeThread)
+    before = copy.deepcopy(session.__dict__)
+    with pytest.raises(RuntimeError, match="workspace failed") as raised:
+        routes._start_regeneration_stream_locked(
+            session,
+            turn=plan.turn,
+            workspace="C:/workspace",
+            model="model",
+            model_provider="provider",
+            normalized_model=False,
+            diag=None,
+            goal_related=False,
+            source="webui",
+            moa_config=None,
+            backend_is_gateway=False,
+        )
+    assert raised.value._regeneration_accepted is True
+    assert session.active_stream_id is not None
+    assert session.__dict__ != before
+
+
+def test_chat_start_losing_regeneration_preserves_locked_send_winner(monkeypatch, tmp_path):
+    from api import routes
+    from api import models as models_api
+    import api.runtime_adapter as runtime_adapter
+
+    session = _session()
+    session.session_id = "route-race-6611"
+    session.model_explicit_pick_signature = "before-regeneration"
+    revision = plan_regeneration(session).revision
+    monkeypatch.setattr(models_api, "SESSION_DIR", tmp_path)
+    monkeypatch.setattr(models_api, "SESSION_INDEX_FILE", tmp_path / "_index.json")
+    monkeypatch.setattr(runtime_adapter, "runtime_adapter_runner_enabled", lambda: False)
+    monkeypatch.setattr(routes, "_get_or_materialize_session", lambda *_args, **_kwargs: session)
+    monkeypatch.setattr(routes, "_agent_runtime_barrier_response", lambda **_kwargs: None)
+    monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *_args: True)
+    monkeypatch.setattr(routes, "_get_active_profile_name", lambda: "default")
+    monkeypatch.setattr(routes, "_read_profile_model_config", lambda *_args: (None, None, {}))
+    monkeypatch.setattr(routes, "_resolve_chat_workspace_for_regeneration", lambda *_args: "C:/workspace")
+    monkeypatch.setattr(routes, "get_config_snapshot", lambda: {})
+    monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda *_args: False)
+    monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", lambda *_args, **_kwargs: ("model", "provider", False))
+    monkeypatch.setattr(routes, "_repair_foreign_session_model_provider", lambda *_args, **_kwargs: "provider")
+    monkeypatch.setattr(routes, "compression_recovery_payload_for_session", lambda *_args: None)
+
+    winner_started = threading.Event()
+    winner_finished = threading.Event()
+
+    def lose_after_locked_send(*_args, **_kwargs):
+        def normal_send_winner():
+            with routes._get_session_agent_lock(session.session_id):
+                session.active_stream_id = "winner-stream"
+                session.pending_user_message = "winner prompt"
+                session.pending_started_at = 222.0
+                session.pending_user_source = "webui"
+                session.save(touch_updated_at=False)
+                winner_started.set()
+                winner_finished.set()
+
+        thread = threading.Thread(target=normal_send_winner)
+        thread.start()
+        assert winner_started.wait(1)
+        assert winner_finished.wait(1)
+        thread.join(timeout=1)
+        return {
+            "error": "stale regeneration revision",
+            "code": "stale_regeneration_revision",
+            "_status": 409,
+        }
+
+    captured = {}
+    monkeypatch.setattr(routes, "_start_run", lose_after_locked_send)
+    monkeypatch.setattr(
+        routes,
+        "j",
+        lambda _handler, data, status=200, **_kwargs: captured.update(data=data, status=status),
+    )
+    routes._handle_chat_start(
+        None,
+        {
+            "session_id": session.session_id,
+            "regenerate": True,
+            "regeneration_revision": revision,
+            "explicit_model_pick": True,
+        },
+    )
+    assert captured["status"] == 409
+    assert (
+        session.active_stream_id,
+        session.pending_user_message,
+        session.pending_started_at,
+        session.pending_user_source,
+    ) == ("winner-stream", "winner prompt", 222.0, "webui")
+    assert session.model_explicit_pick_signature == "before-regeneration"
+    reloaded = Session.load(session.session_id)
+    assert (
+        reloaded.active_stream_id,
+        reloaded.pending_user_message,
+        reloaded.pending_started_at,
+        reloaded.pending_user_source,
+    ) == ("winner-stream", "winner prompt", 222.0, "webui")
+
+
+def test_regeneration_lock_winner_blocks_a_later_normal_send(monkeypatch):
+    from api import routes
+
+    session = _session()
+    accepted = threading.Event()
+    normal_checked = threading.Event()
+    normal_result = {}
+
+    monkeypatch.setattr(routes, "_agent_runtime_barrier_response", lambda **_kwargs: None)
+    monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *_args: None)
+
+    def accepted_regeneration(value, **_kwargs):
+        value.active_stream_id = "accepted-regeneration"
+        value.pending_user_message = "regenerated prompt"
+        accepted.set()
+
+        def ordinary_send_observer():
+            with routes._get_session_agent_lock(value.session_id):
+                normal_result["active_stream_id"] = value.active_stream_id
+                normal_result["pending_user_message"] = value.pending_user_message
+                normal_checked.set()
+
+        thread = threading.Thread(target=ordinary_send_observer)
+        thread.start()
+        assert not normal_checked.wait(0.05)
+        return {"stream_id": "accepted-regeneration", "session_id": value.session_id}
+
+    monkeypatch.setattr(routes, "_start_regeneration_stream_locked", accepted_regeneration)
+    result = routes._start_chat_stream_for_session(
+        session,
+        msg="ignored",
+        attachments=[],
+        workspace="C:/workspace",
+        model="model",
+        model_provider="provider",
+        external_runtime_owned=False,
+        regeneration=SimpleNamespace(revision="revision"),
+    )
+
+    assert accepted.is_set()
+    assert result["stream_id"] == "accepted-regeneration"
+    assert normal_checked.wait(1)
+    assert normal_result == {
+        "active_stream_id": "accepted-regeneration",
+        "pending_user_message": "regenerated prompt",
+    }
+
+
+def test_prepare_mirrors_active_turn_token_to_context_before_timestamp_mutation(monkeypatch):
+    from api import routes
+
+    session = _session()
+    retained_user = session.messages[0]
+    session.context_messages = [
+        {"role": "user", "content": "prompt"},
+        {"role": "user", "content": "prompt"},
+    ]
+    retained_context_user = session.context_messages[1]
+    monkeypatch.setattr(routes, "register_session_writeback_owner", lambda *_args: None)
+    routes._prepare_chat_start_session_for_stream(
+        session,
+        msg="prompt",
+        attachments=[],
+        workspace="C:/workspace",
+        model="model",
+        model_provider="provider",
+        stream_id="token-parity-stream",
+        started_at=99.0,
+        retained_user=retained_user,
+        retained_context_user=retained_context_user,
+        defer_save=True,
+    )
+    assert retained_user["timestamp"] == 99.0
+    assert "timestamp" not in session.context_messages[0]
+    assert "_active_turn_token" not in session.context_messages[0]
+    assert retained_context_user["timestamp"] == 99.0
+    assert retained_context_user["_active_turn_token"] == retained_user["_active_turn_token"]
+
+
+def test_prepare_marks_new_fork_turn_in_eager_materialization(monkeypatch):
+    from api import routes
+    from api.streaming import _materialize_active_turn_user
+
+    session = Session(
+        session_id="fork-prepare-6611",
+        messages=[],
+        context_messages=[],
+        session_source="fork",
+        parent_session_id="parent-6611",
+    )
+    monkeypatch.setattr(routes, "register_session_writeback_owner", lambda *_args: None)
+    monkeypatch.setattr(routes, "get_webui_session_save_mode", lambda: "eager")
+    routes._prepare_chat_start_session_for_stream(
+        session,
+        msg="new fork prompt",
+        attachments=[],
+        workspace="C:/workspace",
+        model="model",
+        model_provider="provider",
+        stream_id="fork-prepare-stream",
+        started_at=99.0,
+        defer_save=True,
+    )
+    assert len(session.messages) == 1
+    assert session.messages[0]["_fork_child_turn"] == session.session_id
+
+    deferred = Session(
+        session_id="fork-deferred-6611",
+        messages=[],
+        context_messages=[],
+        session_source="fork",
+        parent_session_id="parent-6611",
+    )
+    routes._prepare_chat_start_session_for_stream(
+        deferred,
+        msg="deferred fork prompt",
+        attachments=[],
+        workspace="C:/workspace",
+        model="model",
+        model_provider="provider",
+        stream_id="fork-deferred-stream",
+        started_at=100.0,
+        defer_save=True,
+    )
+    materialized = _materialize_active_turn_user(
+        {
+            "text": deferred.pending_user_message,
+            "source": deferred.pending_user_source,
+            "timestamp": deferred.pending_started_at,
+            "session_id": deferred.session_id,
+        },
+        deferred.pending_user_message,
+        deferred.pending_user_source,
+    )
+    assert materialized["_fork_child_turn"] == deferred.session_id
+
+
+def test_issue_artifact_rows_follow_production_regeneration_and_error_settlement(monkeypatch, tmp_path):
+    """Exercise production plan/apply/prepare/materializer/save-reload helpers, not HTTP integration."""
+    from api import routes
+    from api import models as models_api
+    from api.session_ops import apply_regeneration_plan, plan_regeneration
+    from api.streaming import _materialize_pending_user_turn_before_error
+
+    rows = load_issue6611_fixture()["rows"]
+    session = Session(
+        session_id="artifact-production-6611",
+        messages=copy.deepcopy(rows),
+        context_messages=copy.deepcopy(rows),
+        workspace="C:/workspace",
+    )
+    plan = plan_regeneration(session)
+    assert apply_regeneration_plan(session, plan)
+    monkeypatch.setattr(routes, "register_session_writeback_owner", lambda *_args: None)
+    routes._prepare_chat_start_session_for_stream(
+        session,
+        msg=rows[0]["content"],
+        attachments=[],
+        workspace="C:/workspace",
+        model="model",
+        model_provider="provider",
+        stream_id="artifact-production-stream",
+        started_at=99.0,
+        retained_user=session.messages[-1],
+        defer_save=True,
+    )
+    assert [row["role"] for row in session.messages].count("user") == 1
+    assert session.messages[0]["content"] == rows[0]["content"]
+    assert _materialize_pending_user_turn_before_error(session) is False
+    session.active_stream_id = None
+    session.pending_user_message = None
+    session.pending_attachments = []
+    session.pending_started_at = None
+    session.pending_user_source = None
+    session.messages.append(copy.deepcopy(rows[1]))
+    assert [row["role"] for row in session.messages] == ["user", "assistant"]
+    assert [row["content"] for row in session.messages if row["role"] == "user"] == [rows[0]["content"]]
+    monkeypatch.setattr(models_api, "SESSION_DIR", tmp_path)
+    monkeypatch.setattr(models_api, "SESSION_INDEX_FILE", tmp_path / "_index.json")
+    session.save(touch_updated_at=False)
+    reloaded = Session.load(session.session_id)
+    assert reloaded is not None
+    assert [row["role"] for row in reloaded.messages] == ["user", "assistant"]
+    assert [row["content"] for row in reloaded.messages if row["role"] == "user"] == [rows[0]["content"]]
+
+
+def test_locked_start_always_replans_after_browser_validation():
+    source = (Path(__file__).parents[1] / "api" / "routes.py").read_text(encoding="utf-8")
+    start = source.index("def _start_regeneration_stream_locked(")
+    end = source.index("def _active_run_stream_for_session", start)
+    body = source[start:end]
+    assert "plan = plan_regeneration(" in body
+    assert "expected_revision=turn.revision" in body
+    assert "lock_held=True" in body
+
+
+def test_regeneration_preview_has_no_request_snapshot_or_outer_restore_owner():
+    source = (Path(__file__).parents[1] / "api" / "routes.py").read_text(encoding="utf-8")
+    chat_start = source.index("def _handle_chat_start(")
+    chat_end = source.index("def _resolve_chat_workspace_with_recovery", chat_start)
+    body = source[chat_start:chat_end]
+    assert "regeneration_snapshot = snapshot_regeneration_state" not in body
+    assert "transaction_snapshot" not in body
+    assert "_restore_regeneration_preacceptance" not in body
+    regeneration_preview = body[body.index('if body.get("regenerate") is True:'):body.index('diag.stage("normalize_message")')]
+    assert "_clear_stale_stream_state" not in regeneration_preview
+    assert "model_explicit_pick_signature" not in regeneration_preview
+
+
+def test_concurrent_normal_winner_survives_regeneration_409_in_memory_and_after_reload(monkeypatch, tmp_path):
+    from api import routes
+    from api import models as models_api
+
+    session = _session()
+    session.session_id = "race-6611"
+    session.active_stream_id = "stale-stream"
+    session.pending_user_message = "stale prompt"
+    session.pending_started_at = 111.0
+    session.pending_user_source = "webui"
+    monkeypatch.setattr(models_api, "SESSION_DIR", tmp_path)
+    monkeypatch.setattr(models_api, "SESSION_INDEX_FILE", tmp_path / "_index.json")
+    session.save(touch_updated_at=False)
+
+    winner_started = threading.Event()
+    winner_finished = threading.Event()
+
+    def stale_plan(*_args, **_kwargs):
+        def normal_winner():
+            winner_started.set()
+            session.active_stream_id = "winner-stream"
+            session.pending_user_message = "winner prompt"
+            session.pending_started_at = 222.0
+            session.pending_user_source = "webui"
+            session.save(touch_updated_at=False)
+            winner_finished.set()
+
+        threading.Thread(target=normal_winner).start()
+        assert winner_started.wait(1)
+        assert winner_finished.wait(1)
+        raise RegenerationUnavailable("stale_regeneration_revision")
+
+    monkeypatch.setattr("api.session_ops.plan_regeneration", stale_plan)
+    result = routes._start_regeneration_stream_locked(
+        session,
+        turn=SimpleNamespace(revision="old-revision"),
+        workspace="C:/workspace",
+        model="model",
+        model_provider="provider",
+        normalized_model=False,
+        diag=None,
+        goal_related=False,
+        source="webui",
+        moa_config=None,
+        backend_is_gateway=False,
+    )
+    assert result["_status"] == 409
+    assert (session.active_stream_id, session.pending_user_message, session.pending_started_at, session.pending_user_source) == (
+        "winner-stream", "winner prompt", 222.0, "webui"
+    )
+    reloaded = Session.load(session.session_id)
+    assert (reloaded.active_stream_id, reloaded.pending_user_message, reloaded.pending_started_at, reloaded.pending_user_source) == (
+        "winner-stream", "winner prompt", 222.0, "webui"
+    )

tests/test_issue_edit_regenerate_absolute_keep_count.py

@@ -28,10 +28,11 @@ def test_submit_edit_uses_absolute_keep_count():
     assert "keep_count: absoluteKeepCount" in body
 
 
-def test_regenerate_uses_absolute_keep_count():
+def test_regenerate_delegates_to_atomic_start_without_client_truncation():
     body = _function_body(UI_JS, "regenerateResponse")
-    assert re.search(r"absoluteKeepCount\s*=\s*_oldestIdx\s*\+\s*assistantIdx", body)
-    assert "keep_count: absoluteKeepCount" in body
+    assert "await startRegeneration(initialSid, S.session.regeneration_revision)" in body
+    assert "/api/session/truncate" not in body
+    assert "await send()" not in body
 
 
 def test_submit_edit_captures_absolute_before_await():