"""``message.complete.persisted_turn``: the committed row addresses of the agent's current turn. Shared by every host that publishes a turn's completion (the TUI gateway's in-process turn and the session authority's TurnRunner turn), so a client sees one receipt contract whichever host ran the turn. Clients key on it to bind a streamed reply to its stored row (``SessionMessage.row_id``) instead of comparing words. """ from __future__ import annotations from typing import Any, Callable from agent.context_compressor import _DB_PERSISTED_MARKER def committed_row_id(message: Any) -> int | None: row_id = message.get("_row_id") if isinstance(message, dict) else None return row_id if message.get(_DB_PERSISTED_MARKER) and type(row_id) is int and row_id > 0 else None def persisted_turn_receipt( messages: Any, start: Any, history: list, raw: Any, status: str, *, compression_unchanged: bool, prefix_row_matches: Callable[[Any, Any], bool], ) -> dict | None: """Report committed row addresses, never a text/timestamp search for a matching turn. ``start`` is the agent's persistence cursor (``_persist_user_message_idx``), re-anchored during compaction. A partial receipt can address its surviving rows, but cannot retire a client's entire streamed turn. Full coverage additionally requires the unchanged pre-turn prefix (``prefix_row_matches(history_row, turn_row)`` for every row before ``start``), no compaction and no redirected user boundary. """ if (not isinstance(messages, list) or type(start) is not int or not 0 <= start < len(messages) or not isinstance(messages[start], dict) or messages[start].get("role") != "user"): return None tail = messages[start:] anchor_id = committed_row_id(tail[0]) if any(before is tail[0] or (anchor_id is not None and anchor_id == committed_row_id(before)) for before in history): return None # preflight returned the old transcript, not a new turn row_ids = [rid for message in tail if (rid := committed_row_id(message)) is not None] if not row_ids: return None receipt: dict = {"row_ids": row_ids, "complete": False} if (user_row_id := committed_row_id(tail[0])) is not None: receipt["user_row_id"] = user_row_id # Every rendered user row of the turn in order (the prompt, then each steer/redirect row), so a # client binds its optimistic bubbles even when the turn holds more than one prompt row. Only a # fully committed list is published: a gap would shift every later pairing. users = [message for message in tail if message.get("role") == "user" and message.get("display_kind") != "hidden"] user_row_ids = [committed_row_id(message) for message in users] if users and None not in user_row_ids: receipt["user_row_ids"] = user_row_ids last = tail[-1] # Equality only verifies the structurally selected final row's body: it never selects an identity. if (status == "complete" and last.get("role") == "assistant" and not last.get("tool_calls") and last.get("content") == raw and (final_id := committed_row_id(last)) is not None): receipt["final_assistant_row_id"] = final_id prefix_unchanged = start == len(history) and all( prefix_row_matches(before, after) for before, after in zip(history, messages[:start])) receipt["complete"] = bool( prefix_unchanged and compression_unchanged and len(row_ids) == len(tail) and sum(message.get("role") == "user" for message in tail) == 1 and "final_assistant_row_id" in receipt) return receipt