diff --git a/CHANGELOG.md b/CHANGELOG.md index d016d921..84be4784 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,16 @@ last tool result. After the turn settled, that wrapper stayed visible in the chat transcript. The settled-transcript writeback now scrubs the consumed wrapper from the rows it is built from. (#7610 by @webtecnica) +- **Saving a very large session no longer reads and parses the whole file just to count messages.** + The #1558 backup safeguard in `Session.save()` needs the on-disk message count; it obtained it by + loading the entire sidecar. On a real 203 MB / 266,940-message session that made every save + expensive. The count is now taken without a full parse, and the shrink-backup behaviour is + unchanged. (#7578 by @rodrigogs) +- **A cancelled or recovered turn no longer duplicates an answer that was already saved (#6366).** + When a completed assistant turn had been persisted and a later cancel/recovery path ran for the + same pending turn, recovery could append a duplicate user turn plus a `_partial` clone of the + journal. Recovery now stops once the transcript has already advanced past the pending turn. + (#7682 by @happy5318) - **The live stream reports which model actually served the turn.** A new additive `runtime_model` SSE event carries the model (and provider, when known) that the Agent reported while producing diff --git a/api/gateway_chat.py b/api/gateway_chat.py index c66175d5..0dfa7227 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -38,6 +38,7 @@ from api.config import ( from api.helpers import _redact_text, redact_session_data from api.models import clear_process_wakeup_pause, get_session, merge_session_messages_append_only from api.run_journal import RunJournalWriter, bound_run_journal_snapshot_args +from api.turn_journal import append_turn_journal_event_for_stream logger = logging.getLogger(__name__) @@ -1404,6 +1405,21 @@ def _run_gateway_chat_streaming( if cancel_event.is_set(): _restore_cancelled_success_writeback() return + # #6366 re-gate: record the durable same-stream completion + # event in the crash-safe turn journal. The run journal's + # terminal state is only reached on its own ``stream_end`` + # write path, so a Gateway run whose terminal write is lost + # would otherwise leave no completion evidence at all and + # stale-cancel recovery would re-append a duplicate + # recovered row after the valid final answer. + try: + append_turn_journal_event_for_stream( + session_id, + stream_id, + {"event": "completed", "created_at": time.time()}, + ) + except Exception: + logger.debug("Failed to append completed turn journal event", exc_info=True) success_writeback_committed = True try: from api.goals import evaluate_goal_after_turn, has_active_goal diff --git a/api/models.py b/api/models.py index 0322a97b..c9c0318a 100644 --- a/api/models.py +++ b/api/models.py @@ -1107,6 +1107,14 @@ _METADATA_PREFIX_MAX_BYTES = 1024 * 1024 #: even when the stop key sits at 2 KB. _METADATA_PREFIX_FIRST_STAGE_BYTES = 64 * 1024 +# Marks a sidecar as written by the CURRENT writer contract, where the +# persisted `message_count` equals len(messages) by construction (both keys +# land in the same atomic write). save()'s bounded-prefix shrink check trusts +# a persisted count ONLY when this marker sits next to it in the file; see +# _prefix_message_count(). Bump the value whenever the writer contract changes +# meaning, so counts from an older contract fall back to the full parse. +_MESSAGE_COUNT_MARKER = 1 + def _read_metadata_json_prefix(path, max_prefix_bytes=_METADATA_PREFIX_MAX_BYTES): """Read only the metadata portion before the large arrays. @@ -1505,6 +1513,16 @@ class Session: # legacy-format reader that stops at a scene key still finds the count. # The full anchor_activity_scenes bodies serialize AFTER messages. meta['message_count'] = len(self.messages or []) + # _mc_v marks this file as written by the current writer contract, + # where `message_count` equals len(messages) by construction and both + # keys land in the same atomic write. save()'s shrink guard takes the + # bounded-prefix shortcut ONLY for marked files; an unmarked count + # (a legacy pre-#5854 sidecar, a sidecar materialized by an older + # recovery writer, any foreign writer) gets the full parse, so a stale + # count from outside this contract can never read a real shrink as a + # growth and skip the #1558 backup. One save re-marks the file, so the + # fast path still covers steady state. + meta['_mc_v'] = _MESSAGE_COUNT_MARKER meta['anchor_scene_index'] = _anchor_scene_index_from_records(self.anchor_activity_scenes) # Keep the in-memory fingerprint aligned with what we just persisted, so a # later metadata-only reload of THIS object (or any fingerprint reader) @@ -1517,7 +1535,7 @@ class Session: meta['anchor_activity_scenes'] = self.anchor_activity_scenes if isinstance(self.anchor_activity_scenes, dict) else {} # Fields not in METADATA_FIELDS (e.g. last_usage) go at the end. Exclude # the keys we placed explicitly above so they aren't emitted twice. - _placed = {'message_count', 'anchor_scene_index', 'messages', 'tool_calls', 'anchor_activity_scenes'} + _placed = {'message_count', '_mc_v', 'anchor_scene_index', 'messages', 'tool_calls', 'anchor_activity_scenes'} extra = {k: v for k, v in self.__dict__.items() if k not in METADATA_FIELDS and k not in _placed and not k.startswith('_')} @@ -1537,12 +1555,46 @@ class Session: # their .bak get restored automatically. try: if self.path.exists(): - existing_text = self.path.read_text(encoding='utf-8') - try: - existing = json.loads(existing_text) - existing_msg_count = len(existing.get('messages') or []) - except (json.JSONDecodeError, ValueError): - existing_msg_count = -1 # corrupt → always back up + # The on-disk count, without reading the body. + # + # The decision below is a function of ONE integer -- how many + # messages the file on disk holds -- and save() already writes + # that integer into the metadata prefix, before `messages`, as + # `message_count` (see METADATA_FIELDS above; load_metadata_only + # and the sidebar freshness check read it the same way). So read + # THAT through a bounded 64 KiB prefix instead of the whole file. + # Measured before this: a 203,439,398-byte sidecar cost 20,377 ms + # (17,453 in read_text, 2,924 in json.loads) to yield one integer, + # on EVERY save -- including the grow-saves that never back + # anything up. The prefix read is O(64 KiB), and because the count + # is part of the bytes on disk it travels with any rewrite of them. + # + # An in-memory "I wrote this, stat says nothing changed" cache is + # NOT sufficient here, and was removed after review: (inode, size, + # mtime_ns) is not a content identity. A same-length in-place + # rewrite inside one mtime tick keeps all three fields -- ext4 + # stamps mtime from a coarse clock, so two writes in the same tick + # share one mtime_ns -- and a stale cached count then reads a real + # shrink as a growth and skips the #1558 backup. The prefix count + # cannot be fooled that way. + # + # Every unknown falls through to the full read + parse below: a + # legacy (pre-#5854) sidecar whose count is not in the prefix, a + # count written without the current writer's _mc_v marker (an + # older writer's count can be stale relative to the messages + # array next to it), a corrupt or truncated prefix, a file with + # no top-level `messages` key at all, or metadata alone that + # overflows the budget. + # Fail-open is the contract -- never "assume no shrink". + existing_text = None + existing_msg_count = _prefix_message_count(self.path) + if existing_msg_count is None: + existing_text = self.path.read_text(encoding='utf-8') + try: + existing = json.loads(existing_text) + existing_msg_count = len(existing.get('messages') or []) + except (json.JSONDecodeError, ValueError): + existing_msg_count = -1 # corrupt → always back up incoming_msg_count = len(self.messages or []) if ( existing_msg_count > 0 @@ -1560,6 +1612,10 @@ class Session: return if existing_msg_count > incoming_msg_count: bak_path = self.path.with_suffix('.json.bak') + if existing_text is None: + # The .bak body is the one thing that needs the full text, + # and a shrink is the one time it is needed. + existing_text = self.path.read_text(encoding='utf-8') # SHOULD-FIX #2 (Opus): atomic write via tmp+replace, # mirroring the main save() pattern below. Prevents a # torn .bak from a crash mid-write or a concurrent @@ -2442,6 +2498,191 @@ def _latest_user_matches_pending_text(messages, pending_text): return False +def _pending_active_turn_token(session): + """Return the exact WebUI active-turn token for the pending turn. + + The token is ``":"`` — the same value the + eager user-message checkpoint and the agent-result merge stamp onto the + materialized user row (``stamp_message_source``). It is unforgeable + per-turn identity: two distinct streams that submit the same prompt + inside the same second produce different tokens. + """ + from api.process_event_utils import build_active_turn_token + + return build_active_turn_token( + getattr(session, 'active_stream_id', None), + getattr(session, 'pending_started_at', None), + ) + + +def _transcript_user_row_is_pending_turn(message, session, pending_token) -> bool: + """Return True only when a transcript user row provably IS the pending turn. + + Identity is bound to the active-turn token, never to integer-second + timestamp equality: two different streams that send the same prompt + within one second truncate to the same ``int(timestamp)``, so the + checkpoint matcher alone would accept the *other* stream's row as the + pending turn. A token mismatch therefore always loses, and a row that + carries no token at all cannot be proven to be this turn — which means + the caller must recover, not suppress. + """ + if not isinstance(message, dict) or message.get('role') != 'user': + return False + row_token = message.get('_active_turn_token') + if pending_token: + if not row_token: + # Ambiguous legacy identity: the pending turn has a resolvable + # token but this row predates token stamping. We cannot prove it + # is the same turn, so recovery must run. + return False + return row_token == pending_token + if row_token: + # The row belongs to a different, token-bearing turn. Never let a + # same-second timestamp collision override that mismatch. + return False + # Neither side carries a token (legacy/dead-stream pending state): fall + # back to the strict content + timestamp + source + attachments + # checkpoint, which is the only identity signal available. + return _message_matches_pending_checkpoint( + message, + getattr(session, 'pending_user_message', None), + getattr(session, 'pending_started_at', None), + getattr(session, 'pending_user_source', None), + getattr(session, 'pending_attachments', None), + ) + + +def _pending_turn_has_final_assistant_answer(messages, start_idx: int) -> bool: + """Return True when the matched user turn ends in a genuine final answer. + + Reuses the established final-answer semantics + (``_assistant_message_has_final_visible_text``) instead of "an assistant + row exists". An assistant row only settles the turn when it carries real + visible answer text: empty rows, tool-call-only rows, interim + ``_partial`` progress rows, rows recovered from the run journal and + compaction reference cards are all rejected, so a turn that was + interrupted mid-tool-execution is never mistaken for a completed one. + + The scan stops at the next real user row — a final answer that belongs + to a *later* turn must not settle this pending turn. + """ + from api.streaming import ( + _assistant_message_has_final_visible_text, + _is_synthetic_control_message, + ) + + for later in messages[start_idx + 1:]: + if not isinstance(later, dict): + continue + if _is_synthetic_control_message(later): + continue + role = later.get('role') + if role == 'user': + if is_context_compression_marker(later): + # Synthetic compaction cards are not a user-turn boundary. + continue + return False + if role != 'assistant': + # Tool rows never prove a final answer. + continue + if ( + later.get('_partial') + or later.get('_error') + or later.get('_recovered') + or later.get('_recovered_from_run_journal') + ): + continue + if is_context_compression_marker(later): + continue + if _assistant_message_has_final_visible_text(later): + return True + return False + + +def _transcript_already_advanced_past_pending(session) -> bool: + """#6366: a stale pending user message whose own user row already appears + in the durable transcript, followed by a genuine final assistant answer + inside that same user-turn boundary, means the transcript has already + advanced past this pending turn. Returning True here lets the recovery + path skip the recovered user / ``_partial`` clone / journal replay / + generic no-response error that would otherwise append after the valid + final answer. The tail-only check at line ~3485 (``session.messages[-1]``) + misses this case: when the pending turn's user row is older than + the transcript tail, the tail is a *newer* assistant row that + can never match the pending **user** checkpoint, and the recovery + path falls through into the append branch. + + Identity is bound to the pending stream's exact active-turn token + (``_transcript_user_row_is_pending_turn``), so a repeated prompt from a + different stream — even one submitted in the same integer second — + can never be matched against this pending turn and silently dropped. + + Completion requires a genuine final visible assistant answer + (``_pending_turn_has_final_assistant_answer``): empty, tool-call-only, + interim ``_partial``, journal-recovered and compaction tails all keep + recovery active, because that is precisely the interrupted-turn shape + the recovery path exists for. + + The transcript heuristic alone is not durable: an unflagged interim + prose row ("Let me check the logs first.") followed by tool rows that + were never followed by a final answer in the transcript passes the + predicate as a "completed" turn, but the stream may have died mid- + tool and the run journal was never marked done. 9/23 re-gate: also + require durable same-stream terminal evidence. The run journal + cannot supply that evidence on this path — the caller in + ``_apply_core_sync_or_error_marker`` already returns early whenever + the run journal reports ``completed`` for the stream, so the branch + that reaches this predicate can never observe that state (which is + why the previous run-journal gate was unreachable dead code). The + evidence therefore has to come from the turn journal, which records + the exact-stream ``completed`` event for the pending turn + (``_turn_journal_records_completion``). Without it, fall through to + the normal journal-recovery path so the partial output is replayed + and the interruption is marked. + + The suppression is therefore deliberately conservative: whenever the + turn's identity, its completion, or its durable terminal evidence + cannot be positively proven, this helper returns False and the + prompt is recovered instead of discarded. A false negative leaves a + visible cosmetic duplicate; a false positive silently destroys a + prompt or a response. + + Idempotent: this is a pure read that does not mutate the session. + Running recovery twice therefore produces the same outcome. + """ + pending_text = getattr(session, 'pending_user_message', None) + if not pending_text: + return False + messages = getattr(session, 'messages', None) + if not isinstance(messages, list): + return False + pending_token = _pending_active_turn_token(session) + for idx, message in enumerate(messages): + if not _transcript_user_row_is_pending_turn(message, session, pending_token): + continue + # Found the pending turn's own user row at ``idx``. A genuine + # final answer inside that turn's boundary is necessary but not + # sufficient: durable same-stream terminal evidence must also + # exist, otherwise unflagged interim prose + # ("Let me check the logs first.") followed by tool rows can be + # mistaken for a finished turn. + if not _pending_turn_has_final_assistant_answer(messages, idx): + continue + stream_id = getattr(session, 'active_stream_id', None) + # The turn journal — not the run journal — carries the + # completion evidence that is still observable here: the caller + # (``_apply_core_sync_or_error_marker``) already consumed the + # run-journal ``completed`` state on its own early return, so a + # run-journal gate at this depth can never fire. + if not _turn_journal_records_completion(session, stream_id): + # No durable same-stream terminal evidence — the turn may + # have died mid-tool. Fall through to the journal-recovery + # path instead of suppressing it. + return False + return True + return False + + def _partial_message_signature(message: dict) -> tuple: """Return a stable identity for partial assistant markers recovered on load.""" if not isinstance(message, dict): @@ -2643,6 +2884,54 @@ def _run_journal_terminal_state(session, stream_id: str | None) -> str | None: return str(terminal.get('terminal_state') or '') or None +def _turn_journal_records_completion(session, stream_id: str | None) -> bool: + """Return True when the crash-safe turn journal proves the turn for + ``stream_id`` reached its terminal ``completed`` state. + + The turn journal is written per stream/turn id by the exact-stream + workers and by :mod:`api.routes` on submission, so a ``completed`` + event for the *pending stream id* is completion evidence that + survives a run-journal write failure: the run journal only reaches + ``done`` / ``stream_end`` on its own write path, and a process that + dies between the turn journal write and the run-journal terminal + append leaves the run journal without terminal state while the turn + journal still records that the turn actually finished. + + Bound strictly to ``stream_id`` — a ``completed`` turn belonging to + another stream of the same session never settles this one — and to + the latest turn recorded for that stream, so a turn that was + cancelled or crashed (``interrupted``) keeps recovery active even + when an earlier turn of the same stream completed normally. + """ + if not stream_id: + return False + try: + from api.turn_journal import ( + derive_turn_journal_states, + read_turn_journal, + ) + journal = read_turn_journal(session.session_id) + states, _ = derive_turn_journal_states(journal.get('events') or []) + except Exception: + return False + latest: tuple[float, str, dict] | None = None + for turn_id, event in states.items(): + if str(event.get('stream_id') or '') != str(stream_id): + continue + try: + created_at = float(event.get('created_at') or 0) + except (TypeError, ValueError): + created_at = 0.0 + # ``>=`` on both elements keeps the winner deterministic when two + # turns of the same stream share a ``created_at``. + candidate = (created_at, turn_id, event) + if latest is None or candidate[:2] >= latest[:2]: + latest = candidate + if latest is None: + return False + return latest[2].get('event') == 'completed' + + def _recoverable_unsaved_gateway_terminal_error( session, stream_id: str | None, @@ -3519,6 +3808,29 @@ def _apply_core_sync_or_error_marker( ) return True if not _tail_user_already_checkpointed: + # #6366 re-gate: when the durable transcript has already + # advanced past this pending turn into a newer settled + # user/assistant boundary, the recovery path must NOT + # append a recovered user row + ``_partial`` clone + + # journal replay + generic no-response error after the + # valid final answer. The tail-only check above misses + # that case (the tail is a newer assistant row that can + # never match the pending user checkpoint). Clear only + # the stale pending fields and return; the transcript is + # already correct and durable. + if _transcript_already_advanced_past_pending(session): + session.active_stream_id = None + session.pending_user_message = None + session.pending_attachments = [] + session.pending_started_at = None + session.pending_user_source = None + session.save(touch_updated_at=touch_updated_at) + logger.info( + "Session %s: cleared stale pending state for stream %s — transcript already advanced past this turn", + sid, + _stream_id, + ) + return True _append_recovered_pending_turn(session, timestamp=_recovered_ts) else: recovered = { @@ -4239,6 +4551,57 @@ def _sidecar_stat_signature(path): int(st.st_size), int(getattr(st, 'st_ctime_ns', int(st.st_ctime * 1_000_000_000)))) +_MESSAGE_COUNT_MARKER = 1 + + +def _prefix_message_count(path): + """The sidecar's own persisted ``message_count``, from a bounded prefix. + + The #1558 shrink check in save() needs exactly one integer -- how many + messages the file on disk holds -- and the writer puts that integer in the + metadata prefix, before ``messages``, precisely so readers can have it + without parsing the body (see METADATA_FIELDS). Reading the prefix costs + O(64 KiB) against O(file size) for ``read_text`` + ``json.loads`` (measured + 2026-09-15: 20,377 ms for one integer on a 203,439,398-byte sidecar). + + Content-derived on purpose. A stat identity cannot stand in for it: an + in-place rewrite of the same length inside one mtime tick keeps inode, size + *and* mtime_ns, so a cached count would read a real shrink as a growth and + skip the backup. The count is part of the bytes, so any rewrite of them + rewrites it too. + + Gated on the writer marker (``_mc_v``, review 2026-09-22): a count is only + trusted when the SAME write that produced the messages array also vouched + for the count. Anything unmarked -- a legacy pre-#5854 sidecar, a sidecar + materialized by an older recovery writer whose denormalized count could be + stale against its rows, any foreign writer -- returns None and the caller + falls back to the full read + parse. After one save by the current writer + the file carries the marker and the fast path resumes. + + Returns None -- meaning "the caller must fall back to the full read + + parse" -- for every shape that does not carry a usable MARKED count: an + unmarked or wrong-marker file, a legacy sidecar whose count is not in the + prefix, a file with no top-level ``messages`` key at all, a corrupt or + truncated prefix, an unreadable path, or metadata alone that overflows the + budget. + """ + try: + raw = _read_metadata_json_prefix(path) + except (OSError, ValueError): + return None + if not raw: + return None + try: + parsed = json.loads(raw) + except (json.JSONDecodeError, ValueError): + return None + if not isinstance(parsed, dict): + return None + if parsed.get('_mc_v') != _MESSAGE_COUNT_MARKER: + return None + return _parse_nonnegative_int(parsed.get('message_count')) + + def _legacy_sidecar_facts_get(sid): """Return cached authoritative facts for a LEGACY sidecar, or None (#5854). diff --git a/api/session_recovery.py b/api/session_recovery.py index d2d46cb2..d9b09479 100644 --- a/api/session_recovery.py +++ b/api/session_recovery.py @@ -601,7 +601,14 @@ def _state_db_row_to_sidecar(row: dict) -> dict: 'session_id': row.get('id'), 'title': row.get('title') or 'Recovered WebUI Session', 'workspace': workspace_value if isinstance(workspace_value, str) else '', - 'message_count': row.get('message_count') if isinstance(row.get('message_count'), int) else len(messages), + # state.db is canonical for the recovered message rows. Its denormalized + # sessions.message_count is absent from older schemas (read above as 0) + # and can be stale after recovery, so never copy it into a sidecar whose + # `messages` array we just materialized. Session.save() uses this field + # to decide whether an overwrite shrinks the transcript; it must match + # the payload exactly or the #1558 backup / empty-snapshot guards can be + # bypassed. + 'message_count': len(messages if isinstance(messages, list) else []), 'worktree_path': row.get('worktree_path') or None, 'worktree_branch': row.get('worktree_branch') or None, 'worktree_repo_root': row.get('worktree_repo_root') or None, diff --git a/tests/test_issue6366_stale_cancel_recovery.py b/tests/test_issue6366_stale_cancel_recovery.py new file mode 100644 index 00000000..a76b3d73 --- /dev/null +++ b/tests/test_issue6366_stale_cancel_recovery.py @@ -0,0 +1,1424 @@ +"""Tests for #6366: stale-cancel recovery must not append duplicate +rows when the durable transcript has already advanced past the +pending turn — and must never silently discard a distinct repeated +prompt or an interrupted tool-running turn. + +Bug shape: the recovery path at ``api.models.py::_apply_core_sync_or_error_marker`` +checks only ``session.messages[-1]`` against the pending user +checkpoint. When a stale pending turn's user row is older than the +transcript tail, the tail is a *newer* settled assistant row that +can never match the pending **user** checkpoint, and the recovery +path falls through into the append branch. It then appends a +recovered user row, a ``_partial`` clone of the old journal, and a +generic ``No response from provider`` error — all after the valid +final answer. The duplicates survive reload. + +Fix: add a transcript-advance detector that scans the durable +messages for the pending turn's own user row at any index and, when +found, checks for a genuine final assistant answer inside that turn's +boundary. If one exists, the recovery path must clear the stale +pending fields without appending any rows. The helper is a pure read +so recovery is idempotent: running it twice produces the same +outcome. + +Two guard rails keep the suppression from destroying data: + +* **Identity** is bound to the pending stream's exact active-turn + token, never to integer-second timestamp equality. Two different + streams that send the same prompt within one second truncate to the + same second, so the checkpoint matcher alone would match the other + stream's row and the branch would then clear the only durable copy + of the prompt. A token mismatch — or a token-less legacy row — + always loses, so an unprovable identity recovers instead of + suppressing. + +* **Completion** reuses the established final-answer semantics + (``_assistant_message_has_final_visible_text``) rather than "an + assistant row exists". Empty, tool-call-only, interim ``_partial`` + and compaction tails all keep recovery active, so a turn that was + interrupted mid-tool-execution is replayed and marked instead of + being treated as already advanced. + +Because a false negative leaves a visible cosmetic duplicate while a +false positive silently destroys a prompt, the suppression defaults to +recovering whenever identity or completion cannot be proven. The +full-recovery cases below exercise ``_apply_core_sync_or_error_marker`` +itself, because neither guard rail is reachable through the predicate +in isolation — the suppression decision only takes effect on the real +recovery path. +""" + +import json + +import pytest + +import api.models as models +from api.models import ( + SESSIONS, + Session, + _apply_core_sync_or_error_marker, + _pending_turn_has_final_assistant_answer, + _run_journal_terminal_state, + _transcript_already_advanced_past_pending, + _transcript_user_row_is_pending_turn, + _turn_journal_records_completion, +) +from api.run_journal import append_run_event +from api.turn_journal import append_turn_journal_event_for_stream + + +@pytest.fixture(autouse=True) +def _isolate_session_state(tmp_path, monkeypatch): + session_dir = tmp_path / "sessions" + session_dir.mkdir() + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + SESSIONS.clear() + yield + SESSIONS.clear() + + +class _FakeSession: + """Minimal stand-in for api.models.Session — only the fields the + new helpers read.""" + + def __init__( + self, + pending=None, + messages=None, + *, + stream_id=None, + started_at=None, + source=None, + attachments=None, + session_id="fake-session-6366", + ): + self.session_id = session_id + self.pending_user_message = pending + self.messages = list(messages or []) + self.active_stream_id = stream_id + self.pending_started_at = started_at + self.pending_user_source = source + self.pending_attachments = attachments + + +def _token(stream_id, started_at): + return f"{stream_id}:{float(started_at):.17g}" + + +# ── _transcript_user_row_is_pending_turn identity cases ───────────────── + + +def test_user_row_matches_only_with_identical_active_turn_token(): + """Identity is unforgeable: the row's ``_active_turn_token`` must equal + the token derived from the pending stream id + started_at.""" + s = _FakeSession( + pending="hello world", + stream_id="stream-a", + started_at=222.5, + ) + own_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + } + other_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_active_turn_token": "stream-b:222.5", + } + assert _transcript_user_row_is_pending_turn(own_row, s, _token("stream-a", 222.5)) is True + assert _transcript_user_row_is_pending_turn(other_row, s, _token("stream-a", 222.5)) is False + + +def test_user_row_without_token_never_matches_a_token_bearing_pending_turn(): + """A legacy row that predates token stamping cannot be proven to be the + pending turn, so the caller must recover rather than suppress.""" + s = _FakeSession( + pending="hello world", + stream_id="stream-a", + started_at=222.5, + ) + legacy_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + } + assert _transcript_user_row_is_pending_turn(legacy_row, s, _token("stream-a", 222.5)) is False + + +def test_token_bearing_row_never_matches_a_token_less_pending_turn(): + """The mirror case: a token-bearing row belongs to a real turn, so it + must not be claimed by a pending turn that has no resolvable token.""" + s = _FakeSession( + pending="hello world", + stream_id=None, + started_at=None, + ) + token_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + } + assert _transcript_user_row_is_pending_turn(token_row, s, None) is False + + +def test_legacy_pending_turn_falls_back_to_strict_checkpoint_match(): + """With no token on either side the strict content + timestamp + source + + attachments checkpoint is the only identity signal, and it still + distinguishes a same-second repeated prompt from the pending turn.""" + s = _FakeSession( + pending="hello world", + stream_id=None, + started_at=222, + source="webui", + attachments=[{"name": "a.png"}], + ) + same_turn_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "a.png"}], + } + other_turn_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "old.png"}], + } + assert _transcript_user_row_is_pending_turn(same_turn_row, s, None) is True + assert _transcript_user_row_is_pending_turn(other_turn_row, s, None) is False + + +# ── _pending_turn_has_final_assistant_answer completion cases ─────────── + + +def test_final_answer_requires_visible_answer_text(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + {"role": "assistant", "content": "Here is the final answer."}, + ], + 0, + ) is True + + +def test_final_answer_rejects_empty_assistant_row(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + {"role": "assistant", "content": ""}, + ], + 0, + ) is False + + +def test_final_answer_rejects_tool_call_only_row(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": "I found a source and will inspect it.", + "tool_calls": [{"id": "call_1"}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "result"}, + ], + 0, + ) is False + + +def test_final_answer_rejects_interim_partial_row(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": "Still working on it…", + "_partial": True, + }, + ], + 0, + ) is False + + +def test_final_answer_rejects_compaction_marker_row(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + ], + 0, + ) is False + + +def test_final_answer_accepts_answer_after_compaction_marker(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + {"role": "assistant", "content": "Here is the final answer."}, + ], + 0, + ) is True + + +def test_final_answer_does_not_leak_across_a_later_user_turn(): + """A final answer belonging to a *later* turn must not settle this one.""" + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + {"role": "user", "content": "later question"}, + {"role": "assistant", "content": "Here is the final answer."}, + ], + 0, + ) is False + + +def test_final_answer_tolerates_tool_rows_before_the_answer(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]}, + {"role": "tool", "tool_call_id": "call_1", "content": "result"}, + {"role": "assistant", "content": "Here is the final answer."}, + ], + 0, + ) is True + + +# ── _transcript_already_advanced_past_pending unit cases ─────────────── + + +def test_advances_past_pending_when_user_matched_and_settled_assistant_follows(): + """Canonical bug shape: a stale pending turn's user row sits at + an older index, and a newer settled assistant row follows. + Recovery must not append duplicate rows after the assistant. + Identity comes from the pending stream's own active-turn token; + the journal is marked ``done`` (terminal_state ``completed``) so + the suppression gate is satisfied.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "a.png"}], + "_active_turn_token": "stream-a:222.5", + }, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[{"name": "a.png"}], + ) + _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + assert _transcript_already_advanced_past_pending(s) is True + + +def test_advances_past_pending_ignores_partial_clone_after_user(): + """A ``_partial`` clone after the pending turn's user row does NOT count + as a settled advance — the canonical bug shape is the partial + clone followed by a generic no-response error, both of which + the previous recovery path appended. Only a genuine final visible + answer represents a real advance.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 1, + "_active_turn_token": "stream-a:1.5", + }, + {"role": "assistant", "content": "answer", "timestamp": 2, "_partial": True}, + {"role": "assistant", "content": "no response", "timestamp": 3, "_error": True}, + ], + stream_id="stream-a", + started_at=1.5, + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_advances_past_pending_ignores_recovered_assistant_row(): + """A previously-recovered assistant row is not a settled answer + either; the recovery path may have produced it on an earlier + pass. Skip it and keep looking forward.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 1, + "_active_turn_token": "stream-a:1.5", + }, + {"role": "assistant", "content": "answer", "timestamp": 2, "_recovered": True}, + ], + stream_id="stream-a", + started_at=1.5, + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_only_user_row_matches(): + """The pending turn's user row exists but no final assistant answer has + been settled. Recovery is the legitimate path; the helper + returns False.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 1, + "_active_turn_token": "stream-a:1.5", + }, + ], + stream_id="stream-a", + started_at=1.5, + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_no_user_row_matches(): + """No transcript row carries the pending turn's token. The bug shape + here is the older non-matching rows + a stale pending. The + recovery path is legitimate; the helper returns False.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 1, + "_active_turn_token": "stream-z:1.5", + }, + {"role": "assistant", "content": "answer", "timestamp": 2}, + ], + stream_id="stream-a", + started_at=1.5, + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_no_pending_text(): + """An empty pending text means there is nothing to compare + against. The helper returns False without scanning the + transcript.""" + s = _FakeSession( + pending=None, + messages=[ + {"role": "assistant", "content": "answer", "timestamp": 2}, + ], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_no_messages(): + """An empty transcript is the cold-start case; recovery is the + legitimate path. The helper returns False.""" + s = _FakeSession(pending="hello world", messages=[]) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_advances_past_pending_with_unrelated_intervening_rows(): + """The pending turn's user row followed by a tool row and THEN a final + answer counts as a real advance. The helper scans past non-assistant + roles correctly, and the journal is marked ``done`` so the durable + terminal evidence gate is satisfied.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [], + "_active_turn_token": "stream-a:222.5", + }, + {"role": "tool", "content": "tool output"}, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[], + ) + _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + assert _transcript_already_advanced_past_pending(s) is True + + +def test_helper_is_pure_read_idempotent(): + """Idempotence: running the helper twice on the same session + state returns the same result. The helper does not mutate the + session, so recovery is automatically safe to run twice. Journal + is marked ``done`` so the durable terminal evidence gate passes + on both invocations.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [], + "_active_turn_token": "stream-a:222.5", + }, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[], + ) + _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + first = _transcript_already_advanced_past_pending(s) + second = _transcript_already_advanced_past_pending(s) + assert first is True + assert second is True + # And the session state was not mutated. + assert s.pending_user_message == "hello world" + assert len(s.messages) == 2 + + +def test_does_not_advance_when_repeated_prompt_has_different_token(): + """#6366 regression guard for repeated-prompt identity: a *new* + user turn that happens to repeat the same prompt text as an + older transcript row is NOT a stale advance. The active-turn + token tells a stale duplicate from a legitimate re-send, even + when both truncate to the same integer second.""" + s = _FakeSession( + pending="repeat this", + messages=[ + # Older user row from a DIFFERENT stream, same text and the + # same integer second. + { + "role": "user", + "content": "repeat this", + "timestamp": 111, + "_source": "webui", + "attachments": [{"name": "old.png"}], + "_active_turn_token": "stream-old:111.5", + }, + {"role": "assistant", "content": "Earlier answer"}, + # The current turn's own row — pending, no answer yet. + { + "role": "user", + "content": "repeat this", + "timestamp": 111, + "_source": "webui", + "attachments": [{"name": "current.png"}], + "_active_turn_token": "stream-a:111.5", + }, + ], + stream_id="stream-a", + started_at=111.5, + source="webui", + attachments=[{"name": "current.png"}], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_advances_past_pending_with_exact_token_match(): + """The strict-match path: a transcript user row that carries the + pending stream's own active-turn token, followed by a genuine final + answer, is the canonical #6366 bug shape and must suppress the + recovery append.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "a.png"}], + "_active_turn_token": "stream-a:222.5", + }, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[{"name": "a.png"}], + ) + _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + assert _transcript_already_advanced_past_pending(s) is True + + +def test_does_not_advance_when_final_answer_belongs_to_a_later_turn(): + """A final answer that sits after a newer user row belongs to that + newer turn, so it cannot prove the pending turn advanced.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + }, + {"role": "user", "content": "a different question", "timestamp": 224}, + {"role": "assistant", "content": "answer", "timestamp": 225}, + ], + stream_id="stream-a", + started_at=222.5, + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_tool_execution_was_interrupted(): + """A turn interrupted mid-tool-execution ends with a tool-call-only + assistant row and a tool result — not a final answer. Recovery must + still replay the journal and mark the interruption.""" + s = _FakeSession( + pending="run the migration", + messages=[ + { + "role": "user", + "content": "run the migration", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + }, + { + "role": "assistant", + "content": "I will run the migration now.", + "tool_calls": [{"id": "call_1"}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "partial output"}, + ], + stream_id="stream-a", + started_at=222.5, + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_unflagged_interim_prose_precedes_tools(): + """9/22 re-gate regression: an unflagged interim prose row + ("Let me check the logs first.") followed by tool rows passes the + transcript predicate (``_pending_turn_has_final_assistant_answer`` + returns True at the first visible assistant row), but the run + journal was never marked ``done`` — the stream may have died mid- + tool. Without durable same-stream terminal evidence the suppression + gate must stay closed, so ``_transcript_already_advanced_past_pending`` + returns False and the recovery path is free to run.""" + s = _FakeSession( + pending="check the failing test", + messages=[ + { + "role": "user", + "content": "check the failing test", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + }, + # The bug shape: an unflagged assistant row with visible + # text and no tool_calls. The transcript heuristic accepts + # this as a final answer (it carries visible text and has + # no _partial / _error / _recovered marker), but the row + # is actually pre-final interim prose that the repository + # journals as ``interim_assistant`` while the model + # decides what tool to call. The tool_calls that + # produced the following tool rows live on a *later* row + # in the original stream — the journal has not yet + # recorded them because the WebUI process died. + { + "role": "assistant", + "content": "Let me check the logs first.", + }, + # The stream then ran tools before dying. The transcript + # therefore has tool rows after the unflagged prose. + {"role": "tool", "tool_call_id": "call_1", "content": "log tail"}, + {"role": "tool", "tool_call_id": "call_2", "content": "log grep"}, + ], + stream_id="stream-a", + started_at=222.5, + ) + # The journal has tokens and an interim_assistant checkpoint, but + # no ``done`` / ``stream_end`` — the run is still mid-tool from + # the journal's point of view. + _write_journal( + s.session_id, "stream-a", + [ + ("token", {"text": "Let me check the logs first."}), + ("interim_assistant", {"text": "Let me check the logs first."}), + ("tool", {"name": "terminal", "tid": "call_1", "preview": "tail -n 50"}), + ("tool_complete", {"name": "terminal", "tid": "call_1", "preview": "log tail"}), + ("tool", {"name": "terminal", "tid": "call_2", "preview": "grep ERROR"}), + ], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_advance_when_journal_marked_done_after_interim_prose(): + """Companion positive control: when the run journal IS marked + ``done`` (terminal_state ``completed``) for the same stream, the + suppression gate opens even though the transcript carries the + unflagged interim prose + tool shape. Without the journal evidence + the predicate would still hold the gate closed, so this case pins + the boundary between the two new branches.""" + s = _FakeSession( + pending="check the failing test", + messages=[ + { + "role": "user", + "content": "check the failing test", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + }, + { + "role": "assistant", + "content": "Let me check the logs first.", + }, + {"role": "tool", "tool_call_id": "call_1", "content": "log tail"}, + {"role": "assistant", "content": "The logs show a stale lock.", "timestamp": 223}, + ], + stream_id="stream-a", + started_at=222.5, + ) + _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + assert _transcript_already_advanced_past_pending(s) is True + + +# ── turn-journal completion evidence (9/23 re-gate) ─────────────────── +# +# The suppression predicate used to require a *run-journal* terminal +# ``completed`` state. Its caller in ``_apply_core_sync_or_error_marker`` +# already returns early on that same condition, so by the time the +# predicate ran the run-journal branch could never be true and the +# predicate could never suppress anything. The completion evidence +# therefore has to come from the turn journal — the crash-safe journal +# that records the exact-stream ``completed`` event for the pending turn. + + +_TURN_JOURNAL_CLOCK = {"tick": 0} + + +def _write_turn_journal(session_id, stream_id, events): + """Append events to the turn journal for ``stream_id``, reusing the + stream's turn id exactly like ``append_turn_journal_event_for_stream`` + does in production. + + ``created_at`` is taken from a module-level monotonic clock so events + appended across several calls — and across turn ids — keep a strict + ordering; a per-call sequence would hand the same timestamp to events + of different turns, leaving the derived "latest turn per stream" + ambiguous. + """ + written = [] + for event_name, payload in events: + _TURN_JOURNAL_CLOCK["tick"] += 1 + written.append( + append_turn_journal_event_for_stream( + session_id, + stream_id, + { + "event": event_name, + "created_at": 1_700_000_000.0 + _TURN_JOURNAL_CLOCK["tick"], + **payload, + }, + ) + ) + return written + + +def test_turn_journal_helper_reports_completed_for_matching_stream(): + """The helper reads the turn journal and reports completion only for + the stream whose ``completed`` event was recorded.""" + session_id = "issue6366_turn_journal_helper" + _write_turn_journal( + session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + probe = _FakeSession( + pending="hello world", + messages=[], + stream_id="stream-a", + started_at=222.5, + session_id=session_id, + ) + assert _turn_journal_records_completion(probe, "stream-a") is True + assert _turn_journal_records_completion(probe, "stream-b") is False + + +def test_turn_journal_helper_rejects_interrupted_turn(): + """An ``interrupted`` turn is not a completed turn, even when the + journal also carries an earlier ``completed`` event for another + turn of the same stream.""" + session_id = "issue6366_turn_journal_interrupted" + _write_turn_journal( + session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + _write_turn_journal( + session_id, + "stream-a", + [("submitted", {}), ("interrupted", {"reason": "cancelled"})], + ) + probe = _FakeSession( + pending="hello world", + messages=[], + stream_id="stream-a", + started_at=222.5, + session_id=session_id, + ) + assert _turn_journal_records_completion(probe, "stream-a") is False + + +def test_advances_past_pending_from_turn_journal_without_run_journal_terminal(): + """9/23 re-gate root case: the run journal never wrote a terminal + event (the stream died before ``done``/``stream_end`` reached it), + but the turn journal recorded the exact-stream ``completed`` event. + + The transcript alone proves the turn produced a genuine final + answer, and the turn journal proves the turn reached its terminal + state, so the recovery path must suppress the duplicate recovered + user row + journal clone + interruption marker. + """ + session_id = "issue6366_turn_journal_completed" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "a.png"}], + "_active_turn_token": "stream-a:222.5", + }, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[{"name": "a.png"}], + session_id=session_id, + ) + # The run journal holds visible progress but NO terminal event. + _write_journal( + session_id, + "stream-a", + [ + ("token", {"text": "answer"}), + ], + ) + # The turn journal has the exact-stream completion event. + _write_turn_journal( + session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + assert _run_journal_terminal_state(s, "stream-a") != "completed" + assert _turn_journal_records_completion(s, "stream-a") is True + assert _transcript_already_advanced_past_pending(s) is True + + +def test_full_recovery_suppresses_duplicates_on_turn_journal_completion(tmp_path): + """Full-path 9/23 re-gate regression: the run journal never wrote a + terminal state, but the turn journal recorded the exact-stream + ``completed`` event. Recovery must clear the stale pending state + without appending the recovered user row, the journal clone, or the + interruption marker after the valid final answer.""" + session_id = "issue6366_full_recovery_turn_journal_completed" + stream_id = "stream-turn-journal-completed" + started_at = 1_700_002_100.5 + + session = Session( + session_id=session_id, + title="Completed per turn journal, silent run journal", + messages=[ + { + "role": "user", + "content": "summarise the diff", + "timestamp": int(started_at), + "_source": "webui", + "attachments": [], + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + context_messages=[ + {"role": "user", "content": "summarise the diff"}, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + pending_user_message="summarise the diff", + pending_started_at=started_at, + pending_user_source="webui", + pending_attachments=[], + active_stream_id=stream_id, + ) + # The run journal has visible progress but no terminal event, so the + # line-3671 early return cannot fire and the suppression predicate + # is now the only gate that can keep the transcript clean. + _write_journal( + session_id, + stream_id, + [("token", {"text": "The diff adds a guard."})], + ) + # The turn journal recorded the exact-stream terminal completion. + _write_turn_journal( + session_id, + stream_id, + [("submitted", {}), ("completed", {})], + ) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Nothing was appended after the valid final answer. + assert json.dumps(session.messages, ensure_ascii=False) == before + assert not any( + message.get("_recovered") or message.get("_recovered_from_run_journal") + for message in session.messages + ) + assert not any(message.get("_error") for message in session.messages) + # The stale pending state was cleared. + assert session.pending_user_message is None + assert session.active_stream_id is None + assert session.pending_started_at is None + assert session.pending_user_source is None + assert session.pending_attachments == [] + + +def test_full_recovery_appends_rows_on_turn_journal_completion_loss(tmp_path): + """Mirror control: when neither the run journal nor the turn journal + has terminal completion evidence, the same transcript shape must + still recover — the suppression must not widen to cover a turn whose + completion can no longer be proven.""" + session_id = "issue6366_full_recovery_no_turn_journal_completion" + stream_id = "stream-no-turn-journal-completion" + started_at = 1_700_002_200.5 + + session = Session( + session_id=session_id, + title="Silent run journal, no turn journal completion", + messages=[ + { + "role": "user", + "content": "summarise the diff", + "timestamp": int(started_at), + "_source": "webui", + "attachments": [], + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + context_messages=[ + {"role": "user", "content": "summarise the diff"}, + ], + pending_user_message="summarise the diff", + pending_started_at=started_at, + pending_user_source="webui", + pending_attachments=[], + active_stream_id=stream_id, + ) + # Neither journal carries terminal completion evidence. + _write_journal( + session_id, + stream_id, + [("token", {"text": "The diff adds a guard."})], + ) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Recovery ran: the journal replay is visible again and the turn is + # marked interrupted instead of being silently declared finished. + assert json.dumps(session.messages, ensure_ascii=False) != before + assert any( + message.get("_recovered_from_run_journal") + and "The diff adds a guard." in str(message.get("content", "")) + for message in session.messages + ) + assert any( + message.get("_error") and message.get("type") == "interrupted" + for message in session.messages + ) + + +# ── full _apply_core_sync_or_error_marker recovery cases ──────────────── +# +# Neither guard rail above is reachable through the predicate in +# isolation: the suppression decision only takes effect where +# ``_apply_core_sync_or_error_marker`` calls it. These probes drive the +# real repair path end to end. + + +def _write_journal(session_id, stream_id, events): + for event_name, payload in events: + append_run_event(session_id, stream_id, event_name, payload) + + +def test_full_recovery_keeps_distinct_same_second_repeated_prompt(tmp_path): + """Two distinct streams send the same prompt inside one integer second. + + The earlier stream's turn is fully settled in the durable transcript. + The later stream is the pending one and was cancelled before it ever + checkpointed a user row. The repair must recover the pending prompt + instead of matching it against the earlier stream's row — which the + integer-second-truncating checkpoint matcher happily accepts — and + silently clearing the only durable copy of it. + """ + session_id = "issue6366_same_second_repeat" + stale_stream = "stream-stale" + pending_stream = "stream-pending" + # Both turns started inside the same integer second. + shared_second = 1_700_000_000 + + session = Session( + session_id=session_id, + title="Same-second repeated prompt", + messages=[ + { + "role": "user", + "content": "run the migration", + "timestamp": shared_second + 0.1, + "_active_turn_token": f"{stale_stream}:{shared_second + 0.1:.17g}", + }, + {"role": "assistant", "content": "Migration finished."}, + ], + context_messages=[ + {"role": "user", "content": "run the migration"}, + {"role": "assistant", "content": "Migration finished."}, + ], + pending_user_message="run the migration", + pending_started_at=shared_second + 0.6, + active_stream_id=pending_stream, + ) + _write_journal( + session_id, + pending_stream, + [("token", {"text": "Checking the migration status."})], + ) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=pending_stream, + ) is True + + # The pending prompt survives — recovery materialized it as a recovered + # user row instead of dropping it against the other stream's checkpoint. + recovered_users = [ + message + for message in session.messages + if message.get("_recovered") and message.get("content") == "run the migration" + ] + assert len(recovered_users) == 1 + # The journal's visible partial work was replayed, not suppressed. + assert any( + message.get("_recovered_from_run_journal") + and message.get("content") == "Checking the migration status." + for message in session.messages + ) + # And the interruption is marked so the user can see what happened. + assert any( + message.get("_error") and message.get("type") == "interrupted" + for message in session.messages + ) + # The earlier settled turn was left untouched. + assert session.messages[1]["content"] == "Migration finished." + + +def test_full_recovery_keeps_legacy_same_second_repeated_prompt(tmp_path): + """Legacy transcript with no tokens on either side: two identical + prompts inside one second must still be told apart, otherwise the + second prompt is matched against the first's checkpoint and dropped.""" + session_id = "issue6366_legacy_same_second_repeat" + stream_id = "stream-legacy-pending" + shared_second = 1_700_000_100 + + session = Session( + session_id=session_id, + title="Legacy same-second repeated prompt", + messages=[ + # The other stream's row: same text, same integer second, and + # distinguishable only by its different attachments. + { + "role": "user", + "content": "retry the upload", + "timestamp": shared_second, + "_source": "webui", + "attachments": [{"name": "first.png"}], + }, + {"role": "assistant", "content": "Upload done."}, + ], + context_messages=[ + {"role": "user", "content": "retry the upload"}, + {"role": "assistant", "content": "Upload done."}, + ], + pending_user_message="retry the upload", + pending_started_at=shared_second, + pending_user_source="webui", + pending_attachments=[{"name": "second.png"}], + active_stream_id=stream_id, + ) + _write_journal(session_id, stream_id, [("token", {"text": "Re-running upload."})]) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + recovered_users = [ + message + for message in session.messages + if message.get("_recovered") and message.get("content") == "retry the upload" + ] + assert len(recovered_users) == 1 + assert recovered_users[0].get("attachments") == [{"name": "second.png"}] + assert any( + message.get("_recovered_from_run_journal") + and message.get("content") == "Re-running upload." + for message in session.messages + ) + + +@pytest.mark.parametrize( + "tail_rows", + [ + pytest.param( + [{"role": "assistant", "content": ""}], + id="empty-assistant-tail", + ), + pytest.param( + [ + { + "role": "assistant", + "content": "I will inspect the file now.", + "tool_calls": [{"id": "call_1"}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "partial output"}, + ], + id="tool-call-only-tail", + ), + pytest.param( + [ + { + "role": "assistant", + "content": "Still working on the first step…", + "_partial": True, + }, + ], + id="interim-progress-tail", + ), + pytest.param( + [ + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + ], + id="compaction-tail", + ), + ], +) +def test_full_recovery_runs_journal_replay_for_non_final_tails(tmp_path, tail_rows): + """An interrupted turn whose tail is not a genuine final answer must NOT + be treated as completed: journal replay has to run and the turn has to + be marked as interrupted. + """ + session_id = f"issue6366_tail_{tail_rows[-1].get('_partial', 'x')}" + session_id = "issue6366_non_final_tail" + stream_id = "stream-interrupted-tools" + started_at = 1_700_000_500.5 + + session = Session( + session_id=session_id, + title="Interrupted tool-running turn", + messages=[ + { + "role": "user", + "content": "inspect the failing test", + "timestamp": int(started_at), + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + *tail_rows, + ], + context_messages=[ + {"role": "user", "content": "inspect the failing test"}, + ], + pending_user_message="inspect the failing test", + pending_started_at=started_at, + active_stream_id=stream_id, + ) + _write_journal( + session_id, + stream_id, + [ + ("reasoning", {"text": "Reading the assertion first."}), + ("token", {"text": "The assertion never ran."}), + ( + "tool", + { + "name": "terminal", + "preview": "pytest tests/test_x.py", + "args": {"command": "pytest tests/test_x.py"}, + }, + ), + ], + ) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Journal replay ran: the visible partial output is back in the transcript. + assert any( + message.get("_recovered_from_run_journal") + and message.get("content") == "The assertion never ran." + for message in session.messages + ) + # The interrupted turn is marked, so the user sees what happened. + assert any( + message.get("_error") + and message.get("type") == "interrupted" + and "partial output above was recovered" + in str(message.get("content", "")) + for message in session.messages + ) + # Recovery actually appended rows rather than clearing pending state only. + assert json.dumps(session.messages, ensure_ascii=False) != before + assert session.messages[0]["content"] == "inspect the failing test" + + +def test_full_recovery_suppresses_duplicates_only_for_genuine_final_answer(tmp_path): + """Positive control for the suppression branch: when the pending turn's + own user row is followed by a genuine final answer AND the run journal + has the same-stream ``done`` event (durable terminal evidence), the + repair clears the stale pending state and appends nothing. + """ + session_id = "issue6366_suppression_control" + stream_id = "stream-stale-cancelled" + started_at = 1_700_000_900.25 + + session = Session( + session_id=session_id, + title="Already-answered pending turn", + messages=[ + { + "role": "user", + "content": "summarise the diff", + "timestamp": int(started_at), + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + context_messages=[ + {"role": "user", "content": "summarise the diff"}, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + pending_user_message="summarise the diff", + pending_started_at=started_at, + active_stream_id=stream_id, + ) + # Same-stream durable terminal evidence — the run journal was marked + # ``done`` (terminal_state ``completed``) so the suppression gate + # opens and the transcript-only advance is accepted as terminal. + # No visible token events are written so the line-3648 path's + # ``_append_journaled_partial_output`` has nothing to append. + _write_journal(session_id, stream_id, [("done", {})]) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Nothing was appended after the valid final answer. + assert json.dumps(session.messages, ensure_ascii=False) == before + assert not any( + message.get("_recovered") or message.get("_recovered_from_run_journal") + for message in session.messages + ) + assert not any(message.get("_error") for message in session.messages) + # The stale pending state was cleared. + assert session.pending_user_message is None + assert session.active_stream_id is None + assert session.pending_started_at is None + assert session.pending_user_source is None + assert session.pending_attachments == [] + + +def test_full_recovery_runs_journal_replay_for_unflagged_interim_prose(tmp_path): + """9/22 re-gate regression, end-to-end: the durable transcript + already advanced past the pending turn into an unflagged interim + prose row followed by tool rows, but the run journal was never + marked ``done``. Before the fix, + ``_transcript_already_advanced_past_pending`` returned True at the + first visible assistant row, ``_apply_core_sync_or_error_marker`` + cleared the pending state, and the user saw a half-finished turn + with the journal's partial output and the interruption marker + both suppressed. The fix requires durable same-stream terminal + evidence (``done`` / ``stream_end`` → ``completed``) before + suppressing recovery, so the run journal is replayed and the + ``interrupted`` marker is appended for this unflagged-prose + shape. + """ + session_id = "issue6366_unflagged_interim_prose" + stream_id = "stream-unflagged-interim" + started_at = 1_700_001_200.75 + + session = Session( + session_id=session_id, + title="Unflagged interim prose followed by tool rows", + messages=[ + { + "role": "user", + "content": "check the failing test", + "timestamp": int(started_at), + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + # The bug shape: an unflagged assistant row that the + # transcript heuristic accepts as a "final answer" because + # it carries visible text and no _partial / _error / + # _recovered / _recovered_from_run_journal hint. The + # stream had just started talking about what it was about + # to do, then ran tools. The tool_calls that produced the + # following tool rows live on a later row in the original + # stream — the WebUI process died before the tool-call + # assistant row was committed, so the transcript currently + # shows just the pre-final prose followed by tool result + # rows. + { + "role": "assistant", + "content": "Let me check the logs first.", + }, + {"role": "tool", "tool_call_id": "call_1", "content": "log tail"}, + {"role": "tool", "tool_call_id": "call_2", "content": "log grep"}, + ], + context_messages=[ + {"role": "user", "content": "check the failing test"}, + ], + pending_user_message="check the failing test", + pending_started_at=started_at, + active_stream_id=stream_id, + ) + # The run journal has visible progress — the unflagged prose was + # emitted as a token, the run was checkpointed as interim_assistant, + # and one of the tool rows finished. Crucially there is no + # ``done`` / ``stream_end`` event: the stream died mid-tool, so + # the journal has no durable terminal evidence. + _write_journal( + session_id, stream_id, + [ + ("reasoning", {"text": "Reading the assertion first."}), + ("token", {"text": "Let me check the logs first."}), + ("interim_assistant", {"text": "checkpoint"}), + ( + "tool", + { + "name": "terminal", + "tid": "call_1", + "preview": "tail -n 50", + "args": {"command": "tail -n 50"}, + }, + ), + ( + "tool_complete", + { + "name": "terminal", + "tid": "call_1", + "preview": "log tail", + "duration": 0.05, + }, + ), + ], + ) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Recovery actually appended rows rather than clearing pending + # state only. + assert json.dumps(session.messages, ensure_ascii=False) != before + # The journal's visible partial output is back in the transcript: + # the "Let me check the logs first." prose that was emitted over + # SSE before the WebUI process died is now visible again. + assert any( + message.get("_recovered_from_run_journal") + and "Let me check the logs first." in str(message.get("content", "")) + for message in session.messages + ) + # And the interrupted turn is marked, so the user can see the + # stream was cut off rather than presented with a half-finished + # turn that looks complete. + interrupted_marker = next( + ( + message + for message in session.messages + if message.get("_error") and message.get("type") == "interrupted" + ), + None, + ) + assert interrupted_marker is not None + assert "partial output above was recovered" in str( + interrupted_marker.get("content", ""), + ) + # Pending state was cleared (recovery always clears it) and no + # duplicate user row was appended ahead of the journal replay — + # the original user row is still the first message. + assert session.messages[0]["content"] == "check the failing test" + assert session.pending_user_message is None + assert session.active_stream_id is None diff --git a/tests/test_save_count_without_full_parse.py b/tests/test_save_count_without_full_parse.py new file mode 100644 index 00000000..543ace41 --- /dev/null +++ b/tests/test_save_count_without_full_parse.py @@ -0,0 +1,386 @@ +"""A save must not read and parse the whole sidecar just to count its messages. + +The #1558 backup safeguard in `Session.save()` compares the on-disk message +count with the incoming one, and writes `.json.bak` only when the save +would SHRINK the array. Sound. But it obtained the on-disk count by +`json.loads(self.path.read_text())` -- the entire file -- on EVERY save, +including the 99.9% of saves that grow the conversation and never back +anything up. `session_recovery._msg_count()` did the same on every boot, for +every live file that has a `.bak`. + +Measured in production on 2026-09-15, in the running container: + + sessions/fa3bca34a0c6.json 203,439,398 bytes / 266,940 messages + read_text 17,453 ms + json.loads 2,924 ms + TOTAL to obtain one integer: 20,377 ms <- paid on every save + +That was the 39,357 ms `/api/session` request stuck at stage +`t6_after_json_write`, after the read-side re-parse (#4633 recurrence) had +already been fixed. The cost scales linearly with the file, and the file grows +for as long as the conversation does. + +What the check needs is ONE integer -- how many messages the file on disk +holds -- and `save()` already writes that integer into the metadata prefix, +before `messages`, as `message_count`, exactly so readers can have it without +parsing the body (that is how `load_metadata_only()` and the sidebar freshness +check work). So `save()` reads that, through the same bounded 64 KiB prefix +reader, and falls back to the full read + parse whenever the prefix does not +carry a usable count. The `.bak` body is the one thing that still needs the +full text, and a shrink is the one time it is needed. + +The count is read from the FILE, not cached in the object, and that is the +point. An earlier revision of this PR trusted an in-memory (inode, size, +mtime_ns) signature: "this object wrote the file and stat says nothing changed, +so its length is already known". Review (nesquena-hermes, 2026-09-21) showed +that is not a content identity -- a same-length in-place rewrite inside one +mtime tick keeps all three fields, because ext4 stamps mtime from a coarse +clock, so a stale cached count reads a real shrink as a growth and skips the +#1558 backup. That is the test at the bottom of this file. A count read from +the bytes cannot be fooled that way: any rewrite of the bytes rewrites it. + +Fail-open is the contract these tests pin -- whenever the prefix has no usable +count (a legacy pre-#5854 sidecar, a file with no top-level `messages` key, a +corrupt or truncated prefix, metadata that overflows the budget), the behaviour +must be exactly today's full parse, never "assume no shrink". + +`session_recovery._msg_count()` pays the same cost at boot and is left alone +on purpose: its "torn file -> -1" contract is what makes recovery restore a +`.bak` over a truncated live file, and no bounded read can prove a file is +not truncated. +""" +import json +import os +import pathlib +from collections import OrderedDict + +import pytest + +import api.models as M + + +@pytest.fixture +def session_store(tmp_path, monkeypatch): + sdir = tmp_path / "sessions" + sdir.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(M, "SESSION_DIR", sdir) + monkeypatch.setattr(M, "SESSIONS", OrderedDict()) + return sdir + + +def _msgs(n): + return [{"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}"} for i in range(n)] + + +def _make(session_store, sid, n): + s = M.Session(session_id=sid, title="T", workspace=str(session_store.parent), + model="glm", messages=_msgs(n)) + s.save() + return s + + +def _spy_full_reads(monkeypatch, target): + """Count `Path.read_text()` calls on ONE path. + + `read_text` is the whole-file read; the metadata-prefix reader uses `open()` + with a byte budget and never goes through it. So this counter is exactly + "how many times was this file read in full", which is the cost being removed. + """ + calls = {"n": 0} + real = pathlib.Path.read_text + target = pathlib.Path(target) + + def counting(self, *a, **k): + if pathlib.Path(self) == target: + calls["n"] += 1 + return real(self, *a, **k) + + monkeypatch.setattr(type(pathlib.Path()), "read_text", counting) + return calls + + +# ── save(): the grow path must be free of full reads ──────────────────────── + +def test_grow_save_does_not_read_the_existing_file_in_full(session_store, monkeypatch): + """The common case. Appending messages must not read the previous file + in full -- this object wrote that file and it is untouched, so the on-disk + count is already known.""" + s = _make(session_store, "g1", 5) + calls = _spy_full_reads(monkeypatch, s.path) + + s.messages = _msgs(7) + s.save() + + assert calls["n"] == 0, f"grow-save read the existing sidecar in full {calls['n']}x" + assert not s.path.with_suffix(".json.bak").exists(), "a grow-save must not produce a backup" + assert len(M.Session.load("g1").messages) == 7 + + +def test_same_size_save_does_not_read_the_existing_file_in_full(session_store, monkeypatch): + """Metadata-only saves (title, flags, stream state) keep the array as-is. + They are the most frequent save of all and must be just as cheap.""" + s = _make(session_store, "g2", 5) + calls = _spy_full_reads(monkeypatch, s.path) + + s.title = "renamed" + s.save() + + assert calls["n"] == 0 + assert not s.path.with_suffix(".json.bak").exists() + + +# ── save(): the safeguards must be untouched ──────────────────────────────── + +def test_shrink_save_still_writes_bak_with_the_pre_shrink_content(session_store, monkeypatch): + """#1558 must keep working. A shrink still produces a `.bak` holding the + pre-shrink array -- and reads the file in full exactly once, for that body. + Not twice: knowing the count must not be followed by a redundant parse.""" + s = _make(session_store, "s1", 5) + calls = _spy_full_reads(monkeypatch, s.path) + + s.messages = _msgs(3) + s.save() + + reads_by_save = calls["n"] # taken before load() below, which reads the file too + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "a shrinking save must leave a recoverable backup" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5 + assert len(M.Session.load("s1").messages) == 3 + assert reads_by_save == 1, f"shrink-save read the sidecar in full {reads_by_save}x; the .bak body needs exactly one" + + +def test_empty_active_snapshot_is_still_refused(session_store, monkeypatch): + """The other guard in the same block: an empty array with a live stream or + pending prompt must NOT overwrite a populated file (the #1558 data-loss + shape). This object was neither loaded nor saved by this process, so it has + no signature to trust and refusing may cost today's single full read -- + never more, and it must still refuse.""" + _make(session_store, "r1", 5) + empty = M.Session(session_id="r1", title="T", workspace=str(session_store.parent), + model="glm", messages=[], active_stream_id="a" * 32, + pending_user_message="still typing") + calls = _spy_full_reads(monkeypatch, empty.path) + + empty.save() + + reads_by_save = calls["n"] # taken before load() below, which reads the file too + assert len(M.Session.load("r1").messages) == 5, "the populated file must survive" + assert reads_by_save <= 1, f"refusing read the file {reads_by_save}x; an unknown identity costs at most one" + + +# ── save(): fail-open whenever the prefix carries no usable count ─────────── + +def _write_legacy_without_count(session_store, sid, n): + """A pre-#5854 sidecar: no persisted `message_count` at all.""" + doc = {"session_id": sid, "title": "Legacy", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "messages": _msgs(n)} + p = session_store / f"{sid}.json" + p.write_text(json.dumps(doc, indent=2), encoding="utf-8") + return p + + +def test_legacy_sidecar_without_message_count_still_backs_up_on_shrink(session_store): + """The prefix has no count to give, so this must fall back to the full + parse and keep the safeguard. Silently treating "no count" as "no shrink" + would reopen #1558.""" + _write_legacy_without_count(session_store, "l1", 5) + s = M.Session(session_id="l1", title="Legacy", workspace=str(session_store.parent), + model="glm", messages=_msgs(3)) + + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "fail-open: a legacy file must still be backed up on shrink" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5 + + +def test_file_changed_on_disk_since_last_save_falls_back_and_still_backs_up(session_store): + """An externally grown file. This process saved 3 messages; something else + then rewrote the file with 6 (an external appender, another process, a + restore). The next save with 4 is a real shrink relative to DISK, and the + count that governs it is the one the file now carries (6), not the 3 this + process remembers: the `.bak` must hold the 6-message array.""" + s = _make(session_store, "x1", 3) + external = {"session_id": "x1", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "message_count": 6, "messages": _msgs(6)} + s.path.write_text(json.dumps(external, indent=2), encoding="utf-8") + + s.messages = _msgs(4) + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "an externally grown file must still be backed up on shrink" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 + + +def test_file_changed_on_disk_without_a_count_still_backs_up_on_shrink(session_store): + """The same external rewrite, but the replacement carries no + `message_count` (an older writer, a hand-edited file). No prefix count -> + the full parse runs and the 6-message array is still backed up.""" + s = _make(session_store, "x2", 3) + external = {"session_id": "x2", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "messages": _msgs(6)} + s.path.write_text(json.dumps(external, indent=2), encoding="utf-8") + + s.messages = _msgs(4) + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "fail-open: a count-less external rewrite must still be backed up" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 + + +def test_unmarked_stale_count_from_a_foreign_writer_still_backs_up_on_shrink(session_store): + """A persisted count with no `_mc_v` marker is not trusted. + + The exact shape review (nesquena-hermes, 2026-09-22) pinned: a sidecar + materialized by an OLDER writer can carry a count stale against the + messages array beside it (the recovery writer used to copy a denormalized + state.db count over real rows). Here the foreign file says 2 but really + holds 6; a save of 4 must not read that 2 as "the save grows the file" and + skip the #1558 backup. The marker gate makes the prefix reader refuse the + unmarked count, the full parse sees the real array, and the 6 messages + land in the `.bak`. Remove the gate in `_prefix_message_count` and this + test fails with no backup written. + """ + s = _make(session_store, "x3", 3) + foreign = {"session_id": "x3", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "message_count": 2, "messages": _msgs(6)} # stale count, NO _mc_v + s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8") + + s.messages = _msgs(4) + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "an unmarked stale count must not hide the shrink" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 + assert len(M.Session.load("x3").messages) == 4 + + +def test_wrong_marker_version_still_backs_up_on_shrink(session_store): + """A count marked by a DIFFERENT writer contract is equally untrusted.""" + s = _make(session_store, "x4", 3) + foreign = {"session_id": "x4", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "_mc_v": M._MESSAGE_COUNT_MARKER + 1, + "message_count": 2, "messages": _msgs(6)} + s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8") + + s.messages = _msgs(4) + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "a count from another writer version must not hide the shrink" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 + + +def test_first_save_remarks_the_file_and_the_fast_path_resumes(session_store, monkeypatch): + """The gate costs one full parse, once. After a save by the current writer + the file carries `_mc_v`, and the cheap prefix path is back in service.""" + s = _make(session_store, "x5", 3) + foreign = {"session_id": "x5", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "message_count": 2, "messages": _msgs(6)} # unmarked foreign file + s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8") + + s.messages = _msgs(6) + s.save() # equal-count save; unmarked prefix count ignored, full parse, no shrink + + assert json.loads(s.path.read_text(encoding="utf-8"))["_mc_v"] == M._MESSAGE_COUNT_MARKER + + calls = _spy_full_reads(monkeypatch, s.path) + s.messages = _msgs(8) + s.save() + assert calls["n"] == 0, "a marked file must take the bounded-prefix path again" + assert not s.path.with_suffix(".json.bak").exists() + + +def _tool_partial(ts=123): + """The exact shape the #2592 collapse recognises (copied from its test).""" + return {"role": "assistant", "content": "", "_partial": True, "timestamp": ts, + "reasoning": "same reasoning", + "_partial_tool_calls": [{"name": "execute_code", + "args": {"code": "raise RuntimeError('boom')"}, + "done": True, "is_error": True, "duration": 3.87}]} + + +def test_collapse_self_heal_on_load_still_backs_up_the_pre_collapse_array(session_store): + """#2592: load() de-duplicates adjacent partials and immediately saves the + shorter transcript, and that save must produce a `.bak` because it shrinks + the array on purpose. The count that decides it comes from the file's own + prefix, which the writer stamped with the length it wrote (5) -- the + ON-DISK length, not the post-collapse length of the object (3). Read the + object's own length instead and this backup silently disappears, so the + fixture carries the count a real save() would have written.""" + doc = {"session_id": "h1", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "_mc_v": M._MESSAGE_COUNT_MARKER, + "message_count": 5, + "messages": [{"role": "user", "content": "run this"}, + _tool_partial(), _tool_partial(), _tool_partial(), + {"role": "assistant", "content": "**Task cancelled.**", "_error": True}]} + (session_store / "h1.json").write_text(json.dumps(doc), encoding="utf-8") + + loaded = M.Session.load("h1") + + assert sum(1 for m in loaded.messages if m.get("_partial")) == 1, "the collapse must have fired" + assert len(loaded.messages) == 3 + bak = (session_store / "h1.json").with_suffix(".json.bak") + assert bak.exists(), "the self-heal shrink must leave the pre-collapse array recoverable" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5 + + +# ── the collision a stat identity cannot see (review, 2026-09-21) ─────────── + +def test_same_length_in_place_rewrite_inside_one_mtime_tick_still_backs_up(session_store): + """A same-length, same-mtime, same-inode rewrite must not hide a shrink. + + The removed revision of this PR trusted an in-memory (inode, size, + mtime_ns) identity: "this object wrote the file, stat says nothing changed, + so the on-disk length is the one I remember". A stat tuple is not a content + identity. Another writer replaces the sidecar IN PLACE with a longer + transcript that occupies the same number of bytes, inside one mtime tick, + so all three fields still match -- and the next shrinking save then reads + its stale count as "growing", overwrites the longer transcript and writes + no `.bak`. That is the #1558 data-loss shape the safeguard exists to + prevent, reachable on the branch, which is why the identity cache is gone. + + The count now comes from the file's own metadata prefix, which any rewrite + of the bytes has to carry, so the shrink is seen for what it is. + """ + p = session_store / "i1.json" + base = {"session_id": "i1", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0} + + def _render(n): + return json.dumps({**base, "_mc_v": M._MESSAGE_COUNT_MARKER, + "message_count": n, "messages": _msgs(n)}, + indent=2, ensure_ascii=False) + + width = max(len(_render(2)), len(_render(5))) + short, grown = _render(2).ljust(width), _render(5).ljust(width) + assert len(short) == len(grown) == width + p.write_text(short, encoding="utf-8") + + loaded = M.Session.load("i1") # the only version this object knows + before = p.stat() + + p.write_text(grown, encoding="utf-8") # same inode, same byte length... + os.utime(p, ns=(before.st_atime_ns, before.st_mtime_ns)) + after = p.stat() + assert (after.st_ino, after.st_size, after.st_mtime_ns) == \ + (before.st_ino, before.st_size, before.st_mtime_ns), \ + "premise: the rewrite must be invisible to (inode, size, mtime_ns)" + assert len(loaded.messages) == 2, "premise: this object remembers 2 messages" + + loaded.messages = _msgs(3) + loaded.save() + + bak = p.with_suffix(".json.bak") + assert bak.exists(), "the 5-message transcript must be recoverable" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5 + assert len(M.Session.load("i1").messages) == 3 diff --git a/tests/test_session_db_sidecar_reconciliation.py b/tests/test_session_db_sidecar_reconciliation.py index 8b00892b..ba27e574 100644 --- a/tests/test_session_db_sidecar_reconciliation.py +++ b/tests/test_session_db_sidecar_reconciliation.py @@ -45,9 +45,54 @@ def test_recover_missing_sidecars_from_state_db_materializes_webui_row(tmp_path) assert data["parent_session_id"] == "parent-1" assert data["source_tag"] == "webui" assert data["session_source"] == "webui" + assert data["message_count"] == 2 assert [m["content"] for m in data["messages"]] == ["message 1", "message 2"] +def test_recovered_sidecar_derives_message_count_from_materialized_rows(tmp_path, monkeypatch): + """Recovery must not copy a stale state.db summary into a live sidecar. + + `Session.save()` relies on sidecar `message_count` for both #1558 guards. + A recovery row whose denormalized DB count says zero but whose canonical + message rows contain five messages must therefore refuse an empty active + snapshot and preserve a five-message backup on a later shrink. + """ + import api.models as models + + sid = _make_state_db(tmp_path / "state.db", sid="recovered_stale_count", messages=5) + with sqlite3.connect(tmp_path / "state.db") as conn: + conn.execute("UPDATE sessions SET message_count = 0 WHERE id = ?", (sid,)) + + assert recover_missing_sidecars_from_state_db(tmp_path, tmp_path / "state.db")["materialized"] == 1 + sidecar = tmp_path / f"{sid}.json" + assert json.loads(sidecar.read_text(encoding="utf-8"))["message_count"] == 5 + + monkeypatch.setattr(models, "SESSION_DIR", tmp_path) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", tmp_path / "_index.json") + active_empty = models.Session( + session_id=sid, + title="Recovered from DB", + workspace=str(tmp_path), + model="openai/gpt-5", + messages=[], + active_stream_id="a" * 32, + pending_user_message="still typing", + ) + active_empty.save() + assert len(json.loads(sidecar.read_text(encoding="utf-8"))["messages"]) == 5 + + shrinking = models.Session( + session_id=sid, + title="Recovered from DB", + workspace=str(tmp_path), + model="openai/gpt-5", + messages=[{"role": "user", "content": f"replacement {i}"} for i in range(3)], + ) + shrinking.save() + backup = sidecar.with_suffix(".json.bak") + assert len(json.loads(backup.read_text(encoding="utf-8"))["messages"]) == 5 + + def test_recover_missing_sidecars_from_state_db_skips_deleted_webui_tombstone(tmp_path, monkeypatch): import api.models as _m monkeypatch.setattr(_m, "SESSION_DIR", tmp_path) diff --git a/tests/test_webui_gateway_chat_backend.py b/tests/test_webui_gateway_chat_backend.py index 226fee6c..88b78f6b 100644 --- a/tests/test_webui_gateway_chat_backend.py +++ b/tests/test_webui_gateway_chat_backend.py @@ -25,6 +25,7 @@ from api.gateway_chat import ( webui_chat_backend_mode, webui_gateway_chat_enabled, ) +from api.turn_journal import derive_turn_journal_states, read_turn_journal def test_gateway_chat_backend_is_default_off_for_truthy_values(): @@ -397,6 +398,73 @@ def test_gateway_chat_worker_translates_sse_and_persists_session(tmp_path, monke assert all(len(item) == 3 and item[2] for item in events) +def test_gateway_chat_worker_records_turn_journal_completion(tmp_path, monkeypatch): + """#6366 re-gate: a successful Gateway run must record durable + same-stream completion evidence in the crash-safe turn journal. + + The stale-cancel recovery predicate derives its completion evidence + from the turn journal, so a Gateway run that persisted its final + answer but lost the run journal's terminal write would otherwise + have no completion evidence at all — and recovery would re-append a + duplicate recovered row after the valid final answer. + """ + session_dir = tmp_path / "sessions" + session_dir.mkdir() + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + monkeypatch.setattr(models, "SESSIONS", OrderedDict()) + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def __iter__(self): + yield b'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n' + yield b"data: [DONE]\n\n" + + monkeypatch.setenv("HERMES_WEBUI_GATEWAY_BASE_URL", "http://gateway.local") + monkeypatch.setenv("HERMES_WEBUI_GATEWAY_API_KEY", "secret-token") + monkeypatch.setattr( + gateway_chat.urllib.request, + "urlopen", + lambda req, timeout=0: FakeResponse(), + ) + + s = new_session() + stream_id = "stream-gateway-turn-journal-completed" + s.active_stream_id = stream_id + s.pending_user_message = "Say hello" + s.pending_attachments = [] + s.pending_started_at = 456 + s.save() + channel = create_stream_channel() + STREAMS[stream_id] = channel + + gateway_chat._run_gateway_chat_streaming( + s.session_id, + "Say hello", + "test-model", + str(tmp_path), + stream_id, + [], + ) + + journal = read_turn_journal(s.session_id, session_dir=session_dir) + states, _ = derive_turn_journal_states(journal.get("events") or []) + stream_events = [ + event + for event in states.values() + if str(event.get("stream_id") or "") == stream_id + ] + assert stream_events, "expected turn journal events for the gateway stream" + assert any( + event.get("event") == "completed" for event in stream_events + ), "the gateway success writeback must record a completed turn journal event" + + def test_gateway_chat_worker_classifies_terminal_provider_error_without_text(tmp_path, monkeypatch): """Gateway terminal errors must survive an empty assistant stream.""" from unittest.mock import MagicMock