fix(tui_gateway): persist a busy-queued prompt at accept, re-place its row at turn start

A prompt accepted while the agent was busy lived only in the in-memory
queue: session.resume's cold read lacked it until its turn ran and a
backend restart lost it permanently. _handle_busy_submit now writes the
user row through the same #111868 machinery as an idle submit (extracted
as _write_submit_user_row; the durable dict rides the QUEUE ENVELOPE,
never the shared session slot the in-flight turn may own), a text-only
merge syncs the already-written row's content in place, and
_drain_queued_prompt re-places the row at the transcript end before
dispatch (fresh write + deactivate_message on the early row) so the
stored raw order stays [u, a, u, a] instead of glueing the two user
turns under repair_alternation, and the drained turn adopts the fresh
row instead of appending a duplicate. Cancel/crash keep the trailing
user row (the documented interrupted shape). display_kind rides the
envelope from prompt.submit through both writes.

(cherry picked from commit 32792f99d582608d68d40bc12930b295af0d332e)

hermes_state_messages.py

@@ -958,6 +958,18 @@ class SessionMessagesMixin:
             "UPDATE messages SET content = ? WHERE id = ? AND session_id = ? AND role = 'user' AND active = 1",
             (self._encode_content(content), row_id, session_id))
 
+    def deactivate_message(self, session_id: str, row_id: int) -> int:
+        """Deactivate ONE known row (id-addressed, idempotent; returns the affected row count). Used by
+        the queued-prompt drain: the row written at accept time sits ahead of the in-flight turn's
+        assistant reply, and the drain re-appends an identical row at the transcript end — leaving the
+        early row active would put two user rows before that reply and the alternation repair would
+        glue the two turns into one. The durable row is preserved (inactive), never deleted."""
+        if not session_id or isinstance(row_id, bool) or not isinstance(row_id, int) or row_id <= 0:
+            return 0
+        return self._write_rowcount(
+            "UPDATE messages SET active = 0 WHERE id = ? AND session_id = ?",
+            (row_id, session_id))
+
     def _display_dedupe_key(self, row) -> Tuple[Any, ...]:
         """Historical display identity, including normalized live content from user handoff carriers."""
         dedupe_content = row["content"]

tests/tui_gateway/test_queued_prompt_persistence.py

@@ -0,0 +1,198 @@
+"""A prompt accepted while the agent is busy is durable at ACCEPT time.
+
+``_handle_busy_submit`` used to keep the queued turn in memory only: a cold transcript read
+(``session.resume``) lacked the message until its turn ran, and a backend restart lost it
+permanently. The accept now writes the user row through the same #111868 machinery the idle
+submit uses, keeps the envelope's row content in sync when a text-only arrival merges, and
+re-places the row at the transcript end when the queued turn actually dispatches, so the
+stored raw transcript stays [user, assistant, user, assistant] and the turn adopts its row
+instead of writing a duplicate.
+"""
+
+import types
+
+from agent.turn_context import _stage_turn_user_message
+from hermes_state import SessionDB
+from run_agent import AIAgent
+from tui_gateway import server
+
+
+def _desktop_session(monkeypatch, db):
+    monkeypatch.setattr(server, "_get_db", lambda: db)
+    monkeypatch.setattr(server, "_schedule_agent_build", lambda _sid: None)
+    monkeypatch.setattr(server, "_schedule_session_cap_enforcement", lambda: None)
+    monkeypatch.setattr(server, "_register_session_cwd", lambda _session: None)
+    resp = server.handle_request({"id": "c", "method": "session.create", "params": {"cols": 96, "source": "desktop"}})
+    assert "result" in resp, resp
+    sid = resp["result"]["session_id"]
+    server._sessions[sid]["agent"] = types.SimpleNamespace()
+    return sid, resp["result"]["stored_session_id"]
+
+
+def _busy(session, in_flight="prompt A"):
+    with session["history_lock"]:
+        session["running"] = True
+        server._start_inflight_turn(session, in_flight)
+
+
+def _active_rows(db, key):
+    return db.get_messages_as_conversation(key, repair_alternation=True, include_row_ids=True)
+
+
+def _flush_agent(db, key):
+    """Agent shell owning the real flush (the crash persist at turn start runs this same code)."""
+    agent = types.SimpleNamespace(
+        _session_db=db, _session_db_created=True, _persist_disabled=False, session_id=key,
+        _session_persist_lock=None, _flushed_db_message_ids=set(), _flushed_db_message_session_id=None,
+        _last_flushed_db_idx=0, _persist_user_message_idx=None, _persist_user_message_override=None,
+        _persist_user_message_timestamp=None, _pending_cli_user_message=None)
+    agent._ensure_db_session = lambda: None
+    agent._flush_messages_to_session_db = AIAgent._flush_messages_to_session_db.__get__(agent, AIAgent)
+    agent._flush_messages_to_session_db_unlocked = AIAgent._flush_messages_to_session_db_unlocked.__get__(agent, AIAgent)
+    return agent
+
+
+def _run_turn(session, db, key, text, reply):
+    """The turn body the real ``_run_prompt_submit`` runs once the agent is ready: adopt the
+    staged row, crash-persist the user turn, flush the reply (mirrors test_submit_time_user_row)."""
+    agent = _flush_agent(db, key)
+    server._adopt_submit_user_row(session, agent, text, text)
+    user_msg, _pending = _stage_turn_user_message(agent, text, text, None, None, None, None)
+    messages = [user_msg]
+    agent._persist_user_message_idx = 0
+    agent._flush_messages_to_session_db(messages, [])
+    agent._flush_messages_to_session_db(messages + [{"role": "assistant", "content": reply}], [])
+
+
+def _accept_busy_then_run_both_turns(monkeypatch, tmp_path, queued_text="queued text QUEUED-MARKER"):
+    """Real two-turn flow: turn A's row is in the transcript, turn A is live, B is accepted busy,
+    A's reply lands, the real ``_drain_queued_prompt`` dispatches B and B's turn body (adopt +
+    crash-persist + flush) writes reply B. Returns ``(db, sid, key)`` in steady state."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid, key = _desktop_session(monkeypatch, db)
+    session = server._sessions[sid]
+    server._ensure_session_db_row(session)  # the lazy row a real first submit would have written
+    db.append_message(key, "user", content="prompt A")  # turn A's row, as A's turn would have written it
+    _busy(session)
+    resp = server._handle_busy_submit("r1", sid, session, queued_text, "ws-1", queued=True, display_kind=None)
+    assert resp["result"]["status"] == "queued"
+    db.append_message(key, "assistant", content="reply A")  # turn A concludes
+    with session["history_lock"]:
+        session["running"] = False
+        server._clear_inflight_turn(session)
+    monkeypatch.setattr(server, "_run_prompt_submit",
+                        lambda rid, s, sess, text, **kw: _run_turn(sess, db, key, text, "reply B"))
+    assert server._drain_queued_prompt("r2", sid, session) is True
+    return db, sid, key
+
+
+def test_busy_accept_writes_the_queued_user_row_immediately(monkeypatch, tmp_path):
+    """RED for the bug: accept acked {"status": "queued"} with no DB write, so a cold read
+    (session.resume) or a restart saw nothing until the queued turn actually ran."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid, key = _desktop_session(monkeypatch, db)
+    session = server._sessions[sid]
+    try:
+        _busy(session)
+        resp = server._handle_busy_submit("r1", sid, session, "queued text QUEUED-MARKER", "ws-1",
+                                          queued=True, display_kind=None)
+        assert resp["result"]["status"] == "queued"
+        # The turn has not run: the cold resume read must already see the accepted message.
+        assert any(r["role"] == "user" and "queued text QUEUED-MARKER" in str(r["content"])
+                   for r in _active_rows(db, key))
+    finally:
+        server._sessions.pop(sid, None)
+        db.close()
+
+
+def test_queued_turn_replays_as_its_own_turn_after_the_live_turn(monkeypatch, tmp_path):
+    """The accept-time row lands BEFORE the in-flight turn's assistant rows (raw [uA, uB, aA]);
+    the drain must re-place it at the transcript end, so the repaired projection keeps FOUR
+    separate messages instead of gluing the two user turns into one."""
+    db, sid, key = _accept_busy_then_run_both_turns(monkeypatch, tmp_path)
+    try:
+        assert [(r["role"], r["content"]) for r in _active_rows(db, key)] == [
+            ("user", "prompt A"), ("assistant", "reply A"),
+            ("user", "queued text QUEUED-MARKER"), ("assistant", "reply B")]
+        # Exactly one ACTIVE row carries the queued text; the accept-time row survives inactive.
+        active = [r for r in _active_rows(db, key) if "QUEUED-MARKER" in str(r["content"])]
+        assert len(active) == 1
+        every = db.get_messages_as_conversation(key, include_inactive=True, include_row_ids=True)
+        superseded = [r for r in every if "QUEUED-MARKER" in str(r["content"]) and r["_row_id"] != active[0]["_row_id"]]
+        assert len(superseded) == 1  # durable history, never deleted
+    finally:
+        server._sessions.pop(sid, None)
+        db.close()
+
+
+def test_queued_prompt_survives_a_backend_restart(monkeypatch, tmp_path):
+    """A queued prompt is durable at accept: a brand-new SessionDB on the same file (the shape a
+    restarted backend opens) reads the queued message back."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid, key = _desktop_session(monkeypatch, db)
+    session = server._sessions[sid]
+    try:
+        _busy(session)
+        server._handle_busy_submit("r1", sid, session, "queued text QUEUED-MARKER", "ws-1",
+                                   queued=True, display_kind=None)
+        fresh = SessionDB(db_path=tmp_path / "state.db")  # a restarted backend opens a new handle
+        try:
+            assert any(r["role"] == "user" and "queued text QUEUED-MARKER" in str(r["content"])
+                       for r in fresh.get_messages_as_conversation(key, repair_alternation=True,
+                                                                   include_row_ids=True))
+        finally:
+            fresh.close()
+    finally:
+        server._sessions.pop(sid, None)
+        db.close()
+
+
+def test_merged_queue_text_updates_the_written_row_in_place(monkeypatch, tmp_path):
+    """A text-only arrival merges into the queued envelope; the already-written row must not
+    lag the envelope, or cold readers see half the merged prompt and adoption stops matching."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid, key = _desktop_session(monkeypatch, db)
+    session = server._sessions[sid]
+    try:
+        _busy(session)
+        assert server._handle_busy_submit("r1", sid, session, "first QUEUED-MARKER-A", "ws-1",
+                                          queued=True, display_kind=None)["result"]["status"] == "queued"
+        assert server._handle_busy_submit("r2", sid, session, "second", "ws-1",
+                                          queued=True, display_kind=None)["result"]["status"] == "queued"
+        assert session["queued_prompt"]["text"] == "first QUEUED-MARKER-A\n\nsecond"
+        rows = [r for r in _active_rows(db, key) if r["role"] == "user"]
+        assert len(rows) == 1  # merge syncs the ONE row, never appends a second
+        assert rows[0]["content"] == session["queued_prompt"]["text"]
+    finally:
+        server._sessions.pop(sid, None)
+        db.close()
+
+
+def test_cleared_queue_keeps_the_queued_row_as_the_interrupted_shape(monkeypatch, tmp_path):
+    """A cancelled queued prompt keeps its trailing user row: a trailing user row with no reply
+    is the documented interrupted-transcript shape (never delete it on cancel)."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid, key = _desktop_session(monkeypatch, db)
+    session = server._sessions[sid]
+    try:
+        _busy(session)
+        server._handle_busy_submit("r1", sid, session, "queued text QUEUED-MARKER", "ws-1",
+                                   queued=True, display_kind=None)
+        server._ac_set_queue(session, [])  # Stop / queue clear
+        assert not session.get("queued_prompt") and not session.get("queued_prompts")
+        rows = _active_rows(db, key)
+        assert rows and rows[-1]["role"] == "user" and "queued text QUEUED-MARKER" in str(rows[-1]["content"])
+    finally:
+        server._sessions.pop(sid, None)
+        db.close()
+
+
+def test_drained_turn_adopts_the_replaced_row_and_writes_no_duplicate(monkeypatch, tmp_path):
+    """Four active rows after the whole flow (uA, aA, uB, aB): the drained turn adopted the
+    re-placed row instead of appending a fifth."""
+    db, sid, key = _accept_busy_then_run_both_turns(monkeypatch, tmp_path)
+    try:
+        assert len(db.get_messages_as_conversation(key)) == 4
+    finally:
+        server._sessions.pop(sid, None)
+        db.close()

tests/tui_gateway/test_tui_gateway_queue_on_busy.py

@@ -30,6 +30,12 @@ def _session(agent=None, **extra):
     }
 
 
+def _visible(envelope):
+    """Queue envelope without the underscore-internal durable fields a busy accept now rides on it
+    (``_submit_user_row``/``_queued_display_kind``): these tests assert the public envelope shape."""
+    return None if envelope is None else {k: v for k, v in envelope.items() if not k.startswith("_")}
+
+
 # ── _enqueue_prompt ────────────────────────────────────────────────────────
 
 
@@ -246,7 +252,7 @@ def test_hard_interrupt_queue_path_scrubs_stale_inflight_self_duplicate(monkeypa
     resp = server._handle_busy_submit("r1", "sid", session, "Q", "ws-1")
 
     assert resp["result"]["status"] == "queued"
-    assert session.get("queued_prompt") == {"text": "Q", "transport": "ws-1"}
+    assert _visible(session.get("queued_prompt")) == {"text": "Q", "transport": "ws-1"}
     assert not session.get("queued_prompts")
     # Interrupt is async-threaded; policy still enqueued Q after scrubbing P.
 
@@ -578,7 +584,7 @@ def test_busy_submit_claims_attached_image_for_queued_turn(monkeypatch):
     assert redirected == []
     assert not interrupted.wait(0.1)
     assert session["attached_images"] == []
-    assert session["queued_prompt"] == {
+    assert _visible(session["queued_prompt"]) == {
         "text": "is this B?",
         "image_paths": ["/tmp/b.png"],
         "transport": None,
@@ -607,7 +613,7 @@ def test_busy_image_prompts_keep_b_and_c_attachments_in_submission_order(monkeyp
         server._methods["prompt.submit"]("c", {"session_id": "sid", "text": "C"})
 
         assert session["queued_prompt"]["image_paths"] == ["/tmp/b.png"]
-        assert session["queued_prompts"] == [
+        assert [_visible(e) for e in session["queued_prompts"]] == [
             {"text": "C", "image_paths": ["/tmp/c.png"], "transport": None}
         ]
 

tui_gateway/methods_prompt.py

@@ -649,7 +649,8 @@ def _(rid, params: dict) -> dict:
             # for `running` to clear and resubmits with the truncation intact.
             return _err(rid, 4009, "session busy")
         busy_response = _handle_busy_submit(
-            rid, sid, session, text, busy_transport, queued=bool(params.get("queued")), turn_author=turn_author)
+            rid, sid, session, text, busy_transport, queued=bool(params.get("queued")), turn_author=turn_author,
+            display_kind=display_kind)
         if busy_response is not None:
             return busy_response
     raw_rebind_ids = params.get("rebind_survivor_row_ids")

tui_gateway/session_auto_continue.py

@@ -134,11 +134,12 @@ def _ac_inflight_original(session: dict) -> str:
 
 
 def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[str] | None = None,
-                    turn_author: dict | None = None) -> None:
+                    turn_author: dict | None = None) -> dict | None:
     """Queue a message for the next turn. Text-only arrivals share a slot and merge losslessly (like the
     consecutive-user merge in ``repair_message_sequence``); image-bearing and authored ones stay separate
     envelopes so attachment chronology and the sender survive. ``transport`` is pinned so the drained turn
-    streams to its sender."""
+    streams to its sender. Returns the envelope dict the text landed in (the merged head on a merge)
+    so the caller can attach durable state to it; None when the text was dropped as a duplicate."""
     image_paths = list(image_paths or [])
     # Scrub live-turn self-duplicates first so the text merge below can't glue "{original}\n\n{later}" and re-fire the
     # original after a correction settles.
@@ -147,7 +148,7 @@ def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[
     text_only = not image_paths and isinstance(text, str)
     # A text-only self-copy of the live prompt would restart it on drain; an authored copy is another sender's message.
     if text_only and not turn_author and text.strip() == _ac_inflight_original(session) != "":
-        return
+        return None
     queued = {"text": text, "transport": transport, **({"image_paths": image_paths} if image_paths else {}),
               **({"turn_author": turn_author} if turn_author else {})}
     existing = session.get("queued_prompt")
@@ -156,10 +157,12 @@ def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[
             and not session.get("queued_prompts")):
         prev = existing["text"]
         existing["text"] = f"{prev}\n\n{text}" if prev and text else (prev or text)
-    elif existing:
+        return existing
+    if existing:
         session.setdefault("queued_prompts", []).append(queued)
     else:
         session["queued_prompt"] = queued
+    return queued
 
 
 def _sanitize_queued_entry_vs_inflight_user(entry: Any, original: str) -> dict | None:
@@ -273,8 +276,79 @@ def _session_compression_in_flight(session: dict) -> bool:
     return isinstance(holder, str) and bool(holder)
 
 
+def _persist_queued_user_row(session: dict, envelope: dict, display_kind: str | None) -> None:
+    """Make a queued prompt durable the moment it is accepted. Writes the user row through the same
+    #111868 machinery as an idle submit (so a cold ``session.resume`` sees it and a restart cannot lose
+    it) and attaches the durable dict to the QUEUE ENVELOPE — never ``session["_submit_user_row"]``, the
+    shared slot a possibly-still-staged in-flight turn owns. A text-only merge into an existing envelope
+    rewrites the already-written row (and the staged dict) to the merged text in place, so cold readers
+    see the merged text and the drained turn's ``_adopt_submit_user_row`` content match keeps working.
+    A failed write stages nothing on the envelope; the drained turn then writes its own row as before.
+    Caller holds ``history_lock`` (the durable row, the envelope text and the queue must move together —
+    a drain cannot claim between the merge and the write)."""
+    # The queue path bypasses prompt.submit's lazy row creation: the first message of a draft session can
+    # be a queued one, and the message insert needs its sessions row.
+    # Isolated compute-host turns are EXCLUDED exactly as prompt.submit excludes them (an isolated
+    # dispatch returns before ``_persist_session_row_for_submit``): the worker process owns that turn's
+    # persistence, and a local accept-time row would double it on drain.
+    if _session_uses_compute_host(session):
+        return
+    _ensure_session_db_row(session)
+    staged = envelope.get("_submit_user_row")
+    if isinstance(staged, dict) and isinstance(staged.get("_row_id"), int):
+        # Merge sync: the envelope text grew; the already-written row must not lag it.
+        if staged.get("content") != envelope.get("text"):
+            with _session_db(session) as db:
+                if db is None:
+                    return
+                try:
+                    db.set_user_message_content(session.get("session_key"), staged["_row_id"], envelope["text"])
+                except Exception:
+                    logger.debug("queued-prompt row merge update failed", exc_info=True)
+                    return
+            staged["content"] = envelope["text"]
+        return
+    staged = _write_submit_user_row(session, envelope.get("text"), display_kind)
+    if staged is not None:
+        envelope["_submit_user_row"] = staged
+        if display_kind:
+            envelope["_queued_display_kind"] = display_kind
+
+
+def _replace_queued_user_row_for_turn(session: dict, queued: dict) -> None:
+    """Re-place a queued prompt's accept-time row at the transcript END before dispatching its turn.
+
+    The accept-time write lands BEFORE the in-flight turn's assistant rows (raw ``[uA, uB, aA]``), and
+    ``repair_alternation`` would glue the two user turns into one. So on drain: write an identical row
+    with a fresh timestamp at the end via the normal ``_persist_submit_user_row`` — it slots
+    ``session["_submit_user_row"]``, so the turn's ``_adopt_submit_user_row`` adopts it and the flush
+    writes no second row (the exact existing contract) — then deactivate the early row (durable
+    history, never deleted). Steady-state raw order: ``[uA, aA, uB, aB]``. Caller holds
+    ``history_lock`` so a concurrent submit cannot interleave its own row between the two writes.
+    """
+    early = queued.get("_submit_user_row")
+    if not (isinstance(early, dict) and isinstance(early.get("_row_id"), int)):
+        return  # no accept-time row (write failed / pre-feature envelope): the turn persists as before
+    # Append the replacement FIRST: if that write fails nothing is deactivated, the accept-time row
+    # stays active (the message stays visible) and the turn's crash persist persists it as before.
+    _persist_submit_user_row(session, queued.get("text"), queued.get("_queued_display_kind"))
+    fresh = session.get("_submit_user_row")
+    if not (isinstance(fresh, dict) and isinstance(fresh.get("_row_id"), int)):
+        return  # re-append wrote nothing: keep the accept-time row active
+    queued["_submit_user_row"] = fresh  # the envelope follows the live row (a retry drains cleanly)
+    with _session_db(session) as db:
+        if db is None:
+            return
+        try:
+            db.deactivate_message(session.get("session_key"), early["_row_id"])
+        except Exception:
+            # Both rows briefly active merges in projection but never loses the message; deleting or
+            # losing text would be worse.
+            logger.debug("queued-prompt row re-placement deactivate failed", exc_info=True)
+
+
 def _handle_busy_submit(rid, sid: str, session: dict, text: Any, transport: Any, queued: bool = False,
-                        turn_author: dict | None = None) -> dict | None:
+                        turn_author: dict | None = None, display_kind: str | None = None) -> dict | None:
     """Apply ``display.busy_input_mode`` to a mid-turn prompt instead of rejecting it (rejection made clients busy-retry
     and drop sends): ``interrupt`` (default) → redirect, falling back to hard interrupt + queue; ``queue`` → queue only;
     ``steer`` → inject after the current atomic action. ``queued=True`` (client queue drain) forces queue mode: a "run
@@ -310,7 +384,11 @@ def _handle_busy_submit(rid, sid: str, session: dict, text: Any, transport: Any,
             if image_paths:
                 session["attached_images"] = image_paths + list(session.get("attached_images", []))
             return None
-        _enqueue_prompt(session, text, transport, image_paths=image_paths, turn_author=turn_author)
+        envelope = _enqueue_prompt(session, text, transport, image_paths=image_paths, turn_author=turn_author)
+        # Durable AT ACCEPT (not when the turn runs): a cold resume sees the queued message and a
+        # backend restart cannot lose it. Lives on the envelope, never the shared session slot.
+        if envelope is not None:
+            _persist_queued_user_row(session, envelope, display_kind)
         session["last_active"] = time.time()
     # Attachments need their own model invocation: queue without cancelling so the user gets both results in order.
     # ``steer`` must NEVER escalate to a hard interrupt: it would kill the live turn AND drop ``AIAgent._pending_steer``
@@ -353,6 +431,11 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
     kwargs: dict = {"queued_prompt_generation": queue_generation}
     if queued.get("image_paths"):
         kwargs["image_paths"] = queued["image_paths"]
+    # Re-place the accept-time row (if any) at the transcript END before the turn's rows follow it,
+    # and slot the fresh row for adoption. Under history_lock so a concurrent submit can't interleave
+    # its own row write between the re-append and the deactivation.
+    with session["history_lock"]:
+        _replace_queued_user_row_for_turn(session, queued)
     # The compute-host frame has no author field, so only the inline runner receives it.
     author_kwargs = {"turn_author": queued["turn_author"]} if queued.get("turn_author") else {}
     dispatch_failed = False

tui_gateway/session_workdir.py

@@ -345,6 +345,33 @@ def _persist_branch_seed(session: dict) -> None:
             _workdir_reraise_disk_full(exc, "branch seed persist failed")
 
 
+def _write_submit_user_row(session: dict, text: Any, display_kind: str | None):
+    """Write the submitted user turn to the transcript and RETURN the durable dict (stamped
+    ``_DB_PERSISTED_MARKER``/``_row_id``) WITHOUT slotting it on the session. The write half of
+    :func:`_persist_submit_user_row`, shared by the busy-queue accept (which attaches the dict to
+    the queue envelope, never the shared session slot a possibly-still-staged in-flight turn owns).
+    Returns None when nothing was written (no key / non-text / store unavailable / failed write)."""
+    key = session.get("session_key")
+    if not key or not isinstance(text, str) or not text.strip():
+        return None
+    from agent.context_compressor import _DB_PERSISTED_MARKER
+    from agent.message_metadata import stamp_message_timestamp
+    staged = stamp_message_timestamp({"role": "user", "content": text})
+    if display_kind:
+        staged["display_kind"] = display_kind
+    with _session_db(session) as db:
+        if db is None:
+            return None
+        try:
+            staged["_row_id"] = db.append_message(
+                key, "user", content=text, display_kind=display_kind, timestamp=staged["timestamp"])
+        except Exception as exc:
+            _workdir_reraise_disk_full(exc, "submit-time user row persist failed")
+            return None
+    staged[_DB_PERSISTED_MARKER] = True
+    return staged
+
+
 def _persist_submit_user_row(session: dict, text: Any, display_kind: str | None) -> None:
     """Write the submitted user turn at send time, before the agent build and turn: the agent's own
     crash persist only runs once the build finished, so quitting a frozen app during a slow first build
@@ -353,25 +380,8 @@ def _persist_submit_user_row(session: dict, text: Any, display_kind: str | None)
     ``_stage_turn_user_message`` and the flush writes no second row. A failed write stages nothing:
     the turn's crash persist then writes the row as before."""
     session.pop("_submit_user_row", None)  # a failed/unsupported write must not acknowledge an older send
-    key = session.get("session_key")
-    if not key or not isinstance(text, str) or not text.strip():
-        return
-    from agent.context_compressor import _DB_PERSISTED_MARKER
-    from agent.message_metadata import stamp_message_timestamp
-    staged = stamp_message_timestamp({"role": "user", "content": text})
-    if display_kind:
-        staged["display_kind"] = display_kind
-    with _session_db(session) as db:
-        if db is None:
-            return
-        try:
-            staged["_row_id"] = db.append_message(
-                key, "user", content=text, display_kind=display_kind, timestamp=staged["timestamp"])
-        except Exception as exc:
-            _workdir_reraise_disk_full(exc, "submit-time user row persist failed")
-            return
-    staged[_DB_PERSISTED_MARKER] = True
-    session["_submit_user_row"] = staged
+    if (staged := _write_submit_user_row(session, text, display_kind)) is not None:
+        session["_submit_user_row"] = staged
 
 
 def _adopt_submit_user_row(session: dict, agent, persist_user_message: Any, text: Any) -> None: