From 4d035781924459772864d5e0ffc06c31ae1e9898 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 19 Aug 2026 07:11:53 -0400 Subject: [PATCH] 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 Co-authored-by: n --- api/gateway_chat.py | 24 +- api/helpers.py | 45 +- api/routes.py | 411 ++++++++++++- api/session_ops.py | 406 ++++++++++++- api/streaming.py | 87 ++- static/messages.js | 51 +- static/sessions.js | 18 +- static/ui.js | 56 +- tests/_issue6611_fixture.py | 22 + .../fixtures/issue6611_regeneration_rows.json | 14 + .../test_issue4728_new_chat_default_model.py | 11 +- tests/test_issue6611_regenerate_browser.py | 145 +++++ ...test_issue6611_regenerate_turn_identity.py | 63 ++ .../test_issue6611_regeneration_authority.py | 436 +++++++++++++ .../test_issue6611_regeneration_settlement.py | 67 ++ ...test_issue6611_regeneration_transaction.py | 573 ++++++++++++++++++ ...sue_edit_regenerate_absolute_keep_count.py | 9 +- 17 files changed, 2339 insertions(+), 99 deletions(-) create mode 100644 tests/_issue6611_fixture.py create mode 100644 tests/fixtures/issue6611_regeneration_rows.json create mode 100644 tests/test_issue6611_regenerate_browser.py create mode 100644 tests/test_issue6611_regenerate_turn_identity.py create mode 100644 tests/test_issue6611_regeneration_authority.py create mode 100644 tests/test_issue6611_regeneration_settlement.py create mode 100644 tests/test_issue6611_regeneration_transaction.py diff --git a/api/gateway_chat.py b/api/gateway_chat.py index c77f4008..6ee388cf 100644 --- a/api/gateway_chat.py +++ b/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 diff --git a/api/helpers.py b/api/helpers.py index 417bd668..7349cccb 100644 --- a/api/helpers.py +++ b/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'}: diff --git a/api/routes.py b/api/routes.py index 94bbd81e..1937725d 100644 --- a/api/routes.py +++ b/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. diff --git a/api/session_ops.py b/api/session_ops.py index 982e13b5..3a4f985a 100644 --- a/api/session_ops.py +++ b/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) diff --git a/api/streaming.py b/api/streaming.py index b7e2f565..b3c10da9 100644 --- a/api/streaming.py +++ b/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 diff --git a/static/messages.js b/static/messages.js index d6972749..5954902e 100644 --- a/static/messages.js +++ b/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=>{ diff --git a/static/sessions.js b/static/sessions.js index 8c445473..c73131f4 100644 --- a/static/sessions.js +++ b/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; diff --git a/static/ui.js b/static/ui.js index 2b47b8e9..c6a20bf9 100644 --- a/static/ui.js +++ b/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; diff --git a/tests/_issue6611_fixture.py b/tests/_issue6611_fixture.py new file mode 100644 index 00000000..e3594afa --- /dev/null +++ b/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 diff --git a/tests/fixtures/issue6611_regeneration_rows.json b/tests/fixtures/issue6611_regeneration_rows.json new file mode 100644 index 00000000..a03d2d57 --- /dev/null +++ b/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 + } + ] +} diff --git a/tests/test_issue4728_new_chat_default_model.py b/tests/test_issue4728_new_chat_default_model.py index 63c55ca6..8eb921cf 100644 --- a/tests/test_issue4728_new_chat_default_model.py +++ b/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, {}); diff --git a/tests/test_issue6611_regenerate_browser.py b/tests/test_issue6611_regenerate_browser.py new file mode 100644 index 00000000..4b186a9d --- /dev/null +++ b/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)" + ) diff --git a/tests/test_issue6611_regenerate_turn_identity.py b/tests/test_issue6611_regenerate_turn_identity.py new file mode 100644 index 00000000..44ee00e0 --- /dev/null +++ b/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" diff --git a/tests/test_issue6611_regeneration_authority.py b/tests/test_issue6611_regeneration_authority.py new file mode 100644 index 00000000..a519f8c7 --- /dev/null +++ b/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) diff --git a/tests/test_issue6611_regeneration_settlement.py b/tests/test_issue6611_regeneration_settlement.py new file mode 100644 index 00000000..77bb76fd --- /dev/null +++ b/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") diff --git a/tests/test_issue6611_regeneration_transaction.py b/tests/test_issue6611_regeneration_transaction.py new file mode 100644 index 00000000..7f1db4d4 --- /dev/null +++ b/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" + ) diff --git a/tests/test_issue_edit_regenerate_absolute_keep_count.py b/tests/test_issue_edit_regenerate_absolute_keep_count.py index b871c9f9..2640d708 100644 --- a/tests/test_issue_edit_regenerate_absolute_keep_count.py +++ b/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(): @@ -39,4 +40,4 @@ def test_submit_edit_captures_absolute_before_await(): cap = re.search(r"absoluteKeepCount\s*=\s*_oldestIdx\s*\+\s*msgIdx", body) assert cap first_await = re.search(r"\bawait\b", body) - assert first_await and cap.start() < first_await.start() \ No newline at end of file + assert first_await and cap.start() < first_await.start()