refactor(agent): one durable-row rule for the flush and the sidecar stamp; trim tests

The turn-start stamp had grown its own copy of the "what does the current
user row hold" rule (persist override = clean transcript, live content =
wire bytes = sidecar when they differ) that _db_flush_row already
implements. Two copies drift; extract durable_user_row_content() in
session_persistence and call it from both.

Also: reuse _persist_lock() instead of a third open-coded lock/nullcontext
ladder; drop the hasattr guard on set_latest_user_api_content (it predates
this fix and exists on every SessionDB); cut the comment to the WHY;
trim the new test file from 18 cases to the 7 invariants (real close
flush E2E, repeated-"ok" positional protection, API-only pre-flushed
turn, normal path writes nothing, compaction keeps positional, store
guards). Still 3 red / 4 green when agent/turn_context.py is swapped
for main's copy.

agent/session_persistence.py

@@ -7,7 +7,7 @@ import logging
 import re
 from contextlib import nullcontext
 
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 from agent.context_compressor import (
     COMPRESSED_SUMMARY_METADATA_KEY,
@@ -74,6 +74,19 @@ def _override_replaces_content(msg: Dict, content: Any, override: Any) -> bool:
     )
 
 
+def durable_user_row_content(agent, msg: Dict, content: Any, api_content: Any) -> Tuple[Any, Any]:
+    """``(content, api_content)`` as the current turn's user row is written: the persist override is the
+    clean transcript, the live content is what the wire sent — so when they differ and nothing else was
+    injected, the live bytes ARE the sidecar. Shared by the flush and the turn-start stamp so the stamp
+    matches the row the flush wrote."""
+    override = getattr(agent, "_persist_user_message_override", None)
+    if _override_replaces_content(msg, content, override):
+        if api_content is None and isinstance(content, str) and content != override:
+            api_content = content
+        content = override
+    return content, api_content
+
+
 def _summary_display_kind(msg: Dict) -> Any:
     """Standalone handoffs are hidden so they never occupy the active user slot in retry/undo dispatch;
     merge-into-tail carriers keep their prior visibility."""
@@ -143,12 +156,7 @@ def _db_flush_row(agent, msg: Dict, is_current_turn_user: bool) -> Dict[str, Any
     api_content = msg.get("api_content") if isinstance(msg.get("api_content"), str) else None
     timestamp = msg.get("timestamp")
     if is_current_turn_user and role == "user":
-        override = getattr(agent, "_persist_user_message_override", None)
-        if _override_replaces_content(msg, content, override):
-            # Live content is what the wire sent, the override is the clean transcript; keep the sent bytes.
-            if api_content is None and isinstance(content, str) and content != override:
-                api_content = content
-            content = override
+        content, api_content = durable_user_row_content(agent, msg, content, api_content)
         ov_timestamp = getattr(agent, "_persist_user_message_timestamp", None)
         timestamp = timestamp if ov_timestamp is None else ov_timestamp
     if api_content == content: