Seeded sessions survive a gateway restart and store their seed once (tui_gateway) (#107549)

* fix(tui-gateway): a seeded session is durable at create, and its seed is written once

session.create accepts opening messages. Three defects sat in that path:

- A seeded session without a parent was never persisted at create, so a
  restart before the first prompt lost it and session.resume answered
  4007. Only branch children (#93959) were persisted up front. The
  same rationale applies to any seeded create: seeded content is
  intent, not an abandoned draft. Parentless seeds now persist their
  row, transcript and client title at create; empty drafts stay lazy.
- _coerce_seed_history dropped display_kind, so a seeded row tagged
  "hidden" (model-facing scaffolding) rendered as a user bubble. The
  coercion keeps "hidden" and only "hidden"; every other kind is
  stamped by the gateway at turn time and is not accepted from the wire.
- A branch child's seed was written twice: _seed_branch_row copied it at
  create but never marked it persisted, so the first prompt's
  _persist_branch_seed appended the copy again. The create path now
  sets _branch_seed_persisted, and the gate is a create-time `seeded`
  stamp instead of parent_session_id, so a resumed session (whose
  history comes from the DB) can never re-append its transcript.

Two invariant tests, both red on main: a parentless seed survives a
gateway restart with the hidden row kept out of the wire transcript and
not re-written by the first-submit path; a branch child's seed is stored
exactly once. The reasoning-fields fixture stamps `seeded`, the flag
session.create sets.

* fix(tui-gateway): a hidden seed row stays out of the list preview and the create count

Live-testing the seeded create on every surface showed two places where
the newly durable hidden row (display_kind="hidden") still surfaced:

- session.list built a session's preview from its first user row with no
  display_kind filter, so a hidden opening row (model-facing scaffolding
  the gateway never paints) became the sidebar preview. The preview
  predicate now skips hidden rows, in every listing query that shares it.
- session.create reported message_count as the raw seed length while its
  messages array already filtered the hidden row (2 vs 1). It now counts
  what is on the wire, the same rule session.resume applies.

Both are covered by the existing seeded-create test: the create count
equals the wire transcript, and the preview of a session whose first
user row is hidden is its first visible user row.

* fix(tui-gateway): a live unpersisted resume counts the wire transcript

session.resume on a live session that has no row yet reported message_count as
the raw history length while its messages array was already filtered, the same
mismatch the previous commit fixed on session.create. Count the wire, as the
cold, deferred and reuse-live resume paths already do.

* chore: retrigger CI (zero-job dispatch failure, auto-heal)

hermes_state_common.py

@@ -74,11 +74,13 @@ _PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL = (f"CASE WHEN SUBSTR({_PREVIEW_MERGED_PRIOR
 _PREVIEW_FORCE_USER_REMAINDER_SQL = _sql_after_marker(_SUMMARY_END_MARKER)
 
 # Pure compaction rows are ineligible; force-user-leading and merged carriers only when authentic content survives.
-_PREVIEW_ELIGIBLE_SQL = (f"((NOT {_PREVIEW_STANDALONE_SUMMARY_SQL} AND NOT {_PREVIEW_MERGED_SUMMARY_SQL})"
+# A display_kind="hidden" row is model-facing scaffolding the gateway never paints; the preview must not paint it either.
+_PREVIEW_ELIGIBLE_SQL = (f"(COALESCE(m.display_kind, '') <> 'hidden'"
+    f" AND ((NOT {_PREVIEW_STANDALONE_SUMMARY_SQL} AND NOT {_PREVIEW_MERGED_SUMMARY_SQL})"
     f" OR ({_PREVIEW_STANDALONE_SUMMARY_SQL} AND INSTR(m.content, {_sql_literal(_SUMMARY_END_MARKER)}) > 0"
     f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_FORCE_USER_REMAINDER_SQL)}) > 0)"
     f" OR ({_PREVIEW_MERGED_SUMMARY_SQL}"
-    f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL)}) > 0))")
+    f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL)}) > 0)))")
 
 # ``_preview_raw`` SELECT for every listing query (scaffolded rows: head + tail around SKILL_EXCERPT_JOINT).
 _PREVIEW_RAW_SELECT = (

tests/test_tui_gateway_server.py

@@ -20973,6 +20973,7 @@ def test_persist_branch_seed_keeps_reasoning_fields(monkeypatch, tmp_path):
         session_key="branch-key",
         parent_session_id="parent-key",
         history=_branch_history(),
+        seeded=True,  # stamped by session.create: this history exists only in memory
     )
     try:
         db.create_session("branch-key", source="tui")

tests/tui_gateway/test_seeded_session_create.py

@@ -0,0 +1,83 @@
+"""session.create with seeded messages: the seed is durable before the first prompt, and durable once."""
+
+from hermes_state import SessionDB
+from tui_gateway import server
+
+
+def _quiet_create(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)
+
+
+def _create(params: dict) -> dict:
+    resp = server.handle_request({"id": "create", "method": "session.create", "params": params})
+    assert "result" in resp, resp
+    return resp["result"]
+
+
+def test_parentless_seed_survives_a_restart_and_hides_its_runbook(monkeypatch, tmp_path):
+    """A client that opens a chat with its first turns already written (no parent) gets a durable row and
+    transcript at create: a restart before the first prompt resumes it, the hidden runbook stays out of the
+    transcript on the wire, and the seed is not written a second time by the first-submit path."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    _quiet_create(monkeypatch, db)
+    sids = []
+    try:
+        result = _create({
+            "cols": 96, "source": "desktop", "title": "Welcome to Hermes",
+            "messages": [
+                {"role": "user", "content": "Private setup runbook", "display_kind": "hidden"},
+                {"role": "assistant", "content": "Welcome to Hermes"},
+                # Only "hidden" is accepted from the wire; other kinds are stamped by the gateway itself.
+                {"role": "user", "content": "Second question", "display_kind": "steer"},
+            ]})
+        sids.append(result["session_id"])
+        key = result["stored_session_id"]
+        assert [m["role"] for m in result["messages"]] == ["assistant", "user"]
+        assert result["message_count"] == 2  # counts what is on the wire, as session.resume does
+
+        assert db.get_session(key)["title"] == "Welcome to Hermes"
+        rows = db.get_messages_as_conversation(key)
+        assert [r["content"] for r in rows] == ["Private setup runbook", "Welcome to Hermes", "Second question"]
+        assert rows[0]["display_kind"] == "hidden"
+        assert rows[2].get("display_kind") is None
+        listed = server.handle_request({"id": "list", "method": "session.list", "params": {}})["result"]["sessions"]
+        assert next(s for s in listed if s["id"] == key)["preview"].startswith("Second question")  # the hidden row is not the preview
+
+        server._sessions.pop(sids.pop())  # the gateway restarts; only state.db remains
+        resumed = server.handle_request({"id": "resume", "method": "session.resume", "params": {"session_id": key, "cols": 96}})
+        assert "result" in resumed, resumed
+        sids.append(resumed["result"]["session_id"])
+        assert [m["role"] for m in resumed["result"]["messages"]] == ["assistant", "user"]
+
+        server._persist_branch_seed(server._sessions[sids[-1]])  # what the first prompt.submit calls
+        assert len(db.get_messages_as_conversation(key)) == 3
+    finally:
+        for sid in sids:
+            server._sessions.pop(sid, None)
+        db.close()
+
+
+def test_branch_child_seed_is_written_once(monkeypatch, tmp_path):
+    """A seeded branch child persists its copied transcript at create (#93959); the first prompt's seed
+    persist is the fallback for a failed create-time copy, not a second copy."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    _quiet_create(monkeypatch, db)
+    seed = [{"role": "user", "content": "hello from parent"}, {"role": "assistant", "content": "parent reply"}]
+    db.create_session("parent-1", source="desktop")
+    db.append_messages_batch("parent-1", seed)
+    db.set_session_title("parent-1", "Parent chat")
+    sid = None
+    try:
+        result = _create({"cols": 96, "source": "desktop", "parent_session_id": "parent-1", "messages": seed})
+        sid, key = result["session_id"], result["stored_session_id"]
+        assert [r["content"] for r in db.get_messages_as_conversation(key)] == ["hello from parent", "parent reply"]
+
+        server._persist_branch_seed(server._sessions[sid])
+        assert [r["content"] for r in db.get_messages_as_conversation(key)] == ["hello from parent", "parent reply"]
+    finally:
+        if sid:
+            server._sessions.pop(sid, None)
+        db.close()

tui_gateway/methods_session.py

@@ -270,11 +270,31 @@ def _seed_branch_row(record: dict, key: str, parent_session_id: str, history: li
                             source=source, cwd=record["cwd"],
                             profile_name=profile_name_for_home(profile_home) or _current_profile_name(), compensate=True)
             record["pending_title"] = None
+            # The first submit's _persist_branch_seed is the fallback for a failed seed, not a second copy.
+            record["_branch_seed_persisted"] = True
     except Exception:
         logger.warning("seeded-branch persistence failed for %s; falling back to lazy row creation", key,
                        exc_info=True)
 
 
+def _seed_row(record: dict) -> None:
+    """Persist a parentless seeded session NOW, for the reason ``_seed_branch_row`` gives: seeded content is
+    intent, not an abandoned draft, and the renderer's post-create hydration reads the DB. The client's title
+    lands with the row so a restart before the first prompt keeps it. Best-effort — the first-prompt path is
+    the fallback."""
+    try:
+        if _ensure_session_db_row(record) is False:
+            return
+        _persist_branch_seed(record)
+        if title := record.get("pending_title"):
+            with _session_db(record) as db:
+                if db is not None and db.set_session_title(record["session_key"], title):
+                    record["pending_title"] = None
+    except Exception:
+        logger.warning("seeded-session persistence failed for %s; falling back to lazy row creation",
+                       record.get("session_key"), exc_info=True)
+
+
 def _create_overrides(params: dict) -> tuple:
     """PER-SESSION (model, reasoning, service_tier) overrides from the composer — never a global config
     write. ``fast`` presence is the contract: omitted inherits, true pins priority, false pins normal ("")."""
@@ -317,6 +337,7 @@ def _(rid, params: dict) -> dict:
             "cols": int(params.get("cols", 80)), "created_at": now, "edit_snapshots": {},
             "explicit_cwd": explicit_cwd,
             "history": history, "history_lock": threading.Lock(), "history_version": 0, "image_counter": 0,
+            "seeded": bool(history),  # gates _persist_branch_seed: only create-time history is unpersisted
             "cwd": _completion_cwd(params), "inflight_turn": None, "last_active": now,
             "model_override": session_model_override,
             "create_reasoning_override": create_reasoning_override,
@@ -329,7 +350,7 @@ def _(rid, params: dict) -> dict:
             "slash_worker": None, "tool_progress_mode": _load_tool_progress_mode(), "tool_started_at": {},
             "transport": current_transport() or _stdio_transport}
         _register_session_cwd(_sessions[sid])
-    # No DB row here (drafts left "Untitled" litter): created on the first prompt — except seeded branch children.
+    # No DB row here (drafts left "Untitled" litter): created on the first prompt — except seeded sessions.
     # NOTE: we intentionally do NOT persist a DB row here. Every TUI/desktop launch (and every "New agent" /
     # draft) opens a session here just to paint the composer, so eagerly creating a row left an "Untitled"
     # empty session behind for every launch the user never typed into. The row is now created lazily on the
@@ -342,16 +363,21 @@ def _(rid, params: dict) -> dict:
     # optimistic row vanishes on restart. Persisting up front also means a restart keeps the branch (both
     # reports lost it) and the title lands in the parent's lineage instead of falling back to a
     # message-preview name. Title mirrors the TUI /branch naming.
+    # The same holds for a seeded session WITHOUT a parent (a client opening a chat with its first turns
+    # already written): the transcript exists only in memory, so a restart before the first prompt lost it
+    # and the post-create resume 404'd. Persist it up front too; only empty drafts stay lazy.
     if parent_session_id and history:
         _seed_branch_row(_sessions[sid], key, parent_session_id, history, source, profile_home)
+    elif history:
+        _seed_row(_sessions[sid])
     # Return immediately so Ink can paint; the AIAgent builds right after the flush.
     _schedule_agent_build(sid)
     _schedule_session_cap_enforcement()  # trim detached idle sessions over the cap
     cwd = _sessions[sid]["cwd"]
     override = session_model_override or {}
+    messages = _history_to_messages(history)  # hidden seed rows are not on the wire; count what is (as resume does)
     return _ok(rid, {
-        "session_id": sid, "stored_session_id": key, "message_count": len(history),
-        "messages": _history_to_messages(history),
+        "session_id": sid, "stored_session_id": key, "message_count": len(messages), "messages": messages,
         # Reflect the override now so the client doesn't clobber its sticky pick.
         "info": {"model": override.get("model") if override else _resolve_model(),
                  **({"provider": override["provider"]} if override.get("provider") else {}),
@@ -532,10 +558,10 @@ def _resume_live_unpersisted(ctx: _Resume, live_sid: str, live: dict) -> dict:
                 _rebind_live_transport(live_sid, live, transport)
         else:
             _cancel_ws_orphan_reap(live_sid)
-    history = live.get("history") or []
+    messages = ctx.messages(live.get("history") or [])  # count the wire, as every other resume path does
     return _ok(ctx.rid, _attach_todo_state({
         "session_id": live_sid, "stored_session_id": str(live.get("session_key") or ""),
-        "message_count": len(history), "messages": ctx.messages(history),
+        "message_count": len(messages), "messages": messages,
         "info": {"model": _resolve_model(), "lazy": True, "profile_name": profile_name_for_home(live.get("profile_home")) or _response_profile_name(ctx.profile)}}, live))
 
 

tui_gateway/session_history.py

@@ -250,7 +250,13 @@ def _coerce_seed_history(value: Any) -> list[dict]:
             continue
         content = item.get("text") if item.get("content") is None else item.get("content")
         if isinstance(content, str) and content.strip():
-            history.append({"role": item["role"], "content": content})
+            row = {"role": item["role"], "content": content}
+            # "hidden" is the one display_kind a seeding client may author: model-facing scaffolding the
+            # renderer must not paint (a guided-chat runbook). Every other kind is stamped by the gateway
+            # at turn time, so it is not accepted from the wire.
+            if item.get("display_kind") == "hidden":
+                row["display_kind"] = "hidden"
+            history.append(row)
     return history
 
 

tui_gateway/session_workdir.py

@@ -305,11 +305,13 @@ _WORKDIR_SEED_FIELDS = (
 
 
 def _persist_branch_seed(session: dict) -> None:
-    """First-turn persist of a branch's copied transcript. A branch is a draft until its first submit: the parent's
-    messages live only in ``session["history"]`` (ridden into the agent as ``conversation_history``, which
-    ``_flush_messages_to_session_db`` skips by identity), so the row would otherwise resume missing its pre-branch
-    context. Runs once, after ``_ensure_session_db_row`` wrote the row + parent link."""
-    if not (key := session.get("session_key")) or not session.get("parent_session_id") or session.get("_branch_seed_persisted"):
+    """Persist a seeded transcript once its row exists. Seeded messages (a branch's copied parent, a client's
+    opening turns) live only in ``session["history"]`` (ridden into the agent as ``conversation_history``, which
+    ``_flush_messages_to_session_db`` skips by identity), so the row would otherwise resume without them. Runs
+    once: at create for a seeded session, else at the first submit after ``_ensure_session_db_row`` wrote the
+    row. ``seeded`` is stamped by session.create; a resumed session carries its stored transcript in
+    ``history`` and must never re-append it."""
+    if not (key := session.get("session_key")) or not session.get("seeded") or session.get("_branch_seed_persisted"):
         return
     with session["history_lock"]:
         seed = [dict(msg) for msg in (session.get("history") or [])]