From 16cf5897e9011731c6ac91af63864fe2b210c7ee Mon Sep 17 00:00:00 2001 From: DavidMetcalfe <80915+DavidMetcalfe@users.noreply.github.com> Date: Wed, 23 Sep 2026 21:46:36 -0700 Subject: [PATCH] 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 | 12 ++ .../test_queued_prompt_persistence.py | 198 ++++++++++++++++++ .../test_tui_gateway_queue_on_busy.py | 12 +- tui_gateway/methods_prompt.py | 3 +- tui_gateway/session_auto_continue.py | 95 ++++++++- tui_gateway/session_workdir.py | 48 +++-- 6 files changed, 339 insertions(+), 29 deletions(-) create mode 100644 tests/tui_gateway/test_queued_prompt_persistence.py diff --git a/hermes_state_messages.py b/hermes_state_messages.py index bd30a95b5a..94245a8554 100644 --- a/hermes_state_messages.py +++ b/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"] diff --git a/tests/tui_gateway/test_queued_prompt_persistence.py b/tests/tui_gateway/test_queued_prompt_persistence.py new file mode 100644 index 0000000000..45837eb016 --- /dev/null +++ b/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() diff --git a/tests/tui_gateway/test_tui_gateway_queue_on_busy.py b/tests/tui_gateway/test_tui_gateway_queue_on_busy.py index d0206c7b71..2cbaac0977 100644 --- a/tests/tui_gateway/test_tui_gateway_queue_on_busy.py +++ b/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} ] diff --git a/tui_gateway/methods_prompt.py b/tui_gateway/methods_prompt.py index b48eb66bf5..047bca461b 100644 --- a/tui_gateway/methods_prompt.py +++ b/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") diff --git a/tui_gateway/session_auto_continue.py b/tui_gateway/session_auto_continue.py index f4825f6c07..258aac4785 100644 --- a/tui_gateway/session_auto_continue.py +++ b/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 diff --git a/tui_gateway/session_workdir.py b/tui_gateway/session_workdir.py index c9196756b9..c26a164913 100644 --- a/tui_gateway/session_workdir.py +++ b/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: