fix(cli-mode): deliver large prompts via temp file + never lose the prompt on spawn failure

crates/db/src/models/session.rs

@@ -210,12 +210,61 @@ impl Session {
         Ok(())
     }
 
+    /// Park `prompt` ONLY if nothing is parked yet. Loop wake-up re-parks use
+    /// this so continuation boilerplate can never overwrite a parked-but-
+    /// undelivered user prompt — the "never destroy the prompt" invariant has
+    /// to hold on the write side too, not just the CAS-guarded clear. Returns
+    /// whether the park happened; a skipped park is fine (the parked prompt
+    /// is delivered first and the loop re-detects its limit banner).
+    pub async fn set_pending_cli_prompt_if_empty(
+        pool: &SqlitePool,
+        id: Uuid,
+        prompt: &str,
+    ) -> Result<bool, sqlx::Error> {
+        let result = sqlx::query!(
+            r#"UPDATE sessions
+               SET pending_cli_prompt = $1
+               WHERE id = $2 AND pending_cli_prompt IS NULL"#,
+            prompt,
+            id
+        )
+        .execute(pool)
+        .await?;
+        Ok(result.rows_affected() > 0)
+    }
+
+    /// The workspace's parked CLI prompt, wherever it lives: creation parks it
+    /// on the CLI-first session, loop wake-ups re-park on the LATEST session,
+    /// and an attach may resolve a third (frontend-selected) session — so the
+    /// peek must be workspace-scoped or a prompt parked on a sibling session
+    /// row would be stranded forever. Returns the owning session's id (for
+    /// the eventual CAS clear) and the prompt.
+    pub async fn peek_pending_cli_prompt_for_workspace(
+        pool: &SqlitePool,
+        workspace_id: Uuid,
+    ) -> Result<Option<(Uuid, String)>, sqlx::Error> {
+        let row = sqlx::query!(
+            r#"SELECT id AS "id!: Uuid", pending_cli_prompt AS "prompt!: String"
+               FROM sessions
+               WHERE workspace_id = $1 AND pending_cli_prompt IS NOT NULL
+               ORDER BY created_at DESC
+               LIMIT 1"#,
+            workspace_id
+        )
+        .fetch_optional(pool)
+        .await?;
+        Ok(row.map(|r| (r.id, r.prompt)))
+    }
+
     /// Read the parked CLI prompt WITHOUT clearing it. The clear is deferred
-    /// to [`clear_pending_cli_prompt`] after the tmux session is confirmed
-    /// created, so a failure between attach and spawn can't destroy the
-    /// user's only copy of their prompt, and two racing first-attaches that
-    /// both peek the same prompt are harmless — whichever wins `new-session`
-    /// carries it (the loser's `-A` reattach ignores its bootstrap).
+    /// to [`clear_pending_cli_prompt`] until delivery is CONFIRMED — the
+    /// launch bootstrap consumed the staged prompt file with the agent up, or
+    /// the paste into the agent's pane succeeded — so neither a failure
+    /// between attach and spawn nor tmux rejecting the launch command after
+    /// spawn can destroy the user's only copy of the prompt. Racing
+    /// first-attaches are serialized by an in-process delivery claim
+    /// (`CliPromptDelivery` in local-deployment); the losing attach simply
+    /// doesn't carry the prompt.
     pub async fn peek_pending_cli_prompt(
         pool: &SqlitePool,
         id: Uuid,
@@ -229,19 +278,30 @@ impl Session {
         .flatten())
     }
 
-    /// Clear the parked CLI prompt once it has been delivered to a freshly
-    /// created tmux session. Idempotent and atomic (a single guarded UPDATE),
-    /// so a double-call from racing attaches is a no-op.
-    pub async fn clear_pending_cli_prompt(pool: &SqlitePool, id: Uuid) -> Result<(), sqlx::Error> {
-        sqlx::query!(
+    /// Clear the parked CLI prompt once THIS delivery's copy of it has been
+    /// confirmed delivered. Compare-and-swap on the exact delivered value —
+    /// a prompt parked mid-confirmation (e.g. a loop wake-up re-parked while
+    /// an initial prompt's delivery was still being confirmed) is newer,
+    /// undelivered, and must not be destroyed by the older delivery's clear.
+    /// Returns whether the clear happened (`false` = superseded; the newer
+    /// prompt stays parked for its own delivery). Idempotent and atomic (a
+    /// single guarded UPDATE), so a double-call from racing attaches is a
+    /// no-op.
+    pub async fn clear_pending_cli_prompt(
+        pool: &SqlitePool,
+        id: Uuid,
+        delivered: &str,
+    ) -> Result<bool, sqlx::Error> {
+        let result = sqlx::query!(
             r#"UPDATE sessions
                SET pending_cli_prompt = NULL
-               WHERE id = $1 AND pending_cli_prompt IS NOT NULL"#,
-            id
+               WHERE id = $1 AND pending_cli_prompt = $2"#,
+            id,
+            delivered
         )
         .execute(pool)
         .await?;
-        Ok(())
+        Ok(result.rows_affected() > 0)
     }
 
     /// Persist the model + reasoning effort chosen at CLI-first creation so the

crates/local-deployment/src/loop_supervisor.rs

@@ -26,8 +26,11 @@ use db::{
         session::Session,
     },
 };
+use uuid::Uuid;
 
-use crate::pty::{capture_cli_pane, cli_tmux_available, cli_tmux_session_exists, send_cli_keys};
+use crate::pty::{
+    CliPromptDelivery, capture_cli_pane, cli_tmux_available, cli_tmux_session_exists, send_cli_keys,
+};
 
 /// How often to poll panes for limit banners and check for due wake-ups.
 const POLL_INTERVAL: Duration = Duration::from_secs(20);
@@ -413,10 +416,19 @@ async fn deliver_due_wakeups(db: &DBService, _now: DateTime<Utc>) -> Result<(),
         // The tmux session may have been reaped before a long usage-window wake
         // came due — re-park the prompt so the next attach delivers it.
         if !cli_tmux_session_exists(wid).await {
-            if let Ok(Some(session)) = Session::find_latest_by_workspace_id(pool, wid).await {
-                let _ = Session::set_pending_cli_prompt(pool, session.id, &prompt).await;
-                tracing::info!("loop: workspace {wid} session gone; re-parked wake-up prompt");
+            let outcome = repark_prompt(pool, wid, &prompt).await;
+            if wakeup_stays_pending(&outcome, is_limit) {
+                // Not re-parked and not deliverable now (transient DB error, or
+                // a manual wake-up blocked behind an already-parked prompt):
+                // leave it pending so a later tick delivers it, rather than
+                // marking it fired and dropping the user's prompt.
+                tracing::warn!(
+                    "loop: workspace {wid} session gone; wake-up not re-parked ({outcome:?}), \
+                     leaving pending"
+                );
+                continue;
             }
+            tracing::info!("loop: workspace {wid} session gone; re-park outcome {outcome:?}");
             ScheduledWakeup::mark_fired(pool, wakeup.id).await?;
             continue;
         }
@@ -444,6 +456,38 @@ async fn deliver_due_wakeups(db: &DBService, _now: DateTime<Utc>) -> Result<(),
             }
         }
 
+        // Serialize with any in-flight parked-prompt delivery (terminal
+        // attach): two concurrent paste+Enter pairs into the same pane would
+        // interleave into one garbled submission. If a delivery holds the
+        // claim, leave the wake-up pending — the next tick retries after the
+        // delivery window closes.
+        let Some(_claim) = CliPromptDelivery::try_claim(wid) else {
+            tracing::debug!(
+                "loop: prompt delivery in flight for workspace {wid}; deferring wake-up"
+            );
+            continue;
+        };
+
+        // "Parked prompt delivers first" holds on the SEND path too: if the
+        // workspace has an undelivered prompt queued (an earlier delivery went
+        // unconfirmed and left it parked), don't send the wake-up ahead of it —
+        // that would reach the agent out of order. Leave the wake-up pending;
+        // a terminal attach delivers and clears the parked prompt, then a later
+        // tick delivers the wake-up. (Transient DB error: also retry.)
+        match Session::peek_pending_cli_prompt_for_workspace(pool, wid).await {
+            Ok(Some(_)) => {
+                tracing::debug!(
+                    "loop: workspace {wid} has a parked prompt; deferring wake-up until it delivers"
+                );
+                continue;
+            }
+            Ok(None) => {}
+            Err(e) => {
+                tracing::warn!("loop: failed to check parked prompt for workspace {wid}: {e}");
+                continue;
+            }
+        }
+
         if send_cli_keys(wid, &prompt).await {
             if is_limit {
                 let _ = LoopAutomation::increment_attempts(pool, wid).await;
@@ -453,13 +497,114 @@ async fn deliver_due_wakeups(db: &DBService, _now: DateTime<Utc>) -> Result<(),
                 wakeup.kind.as_str()
             );
         } else {
-            tracing::warn!("loop: failed to deliver wake-up to workspace {wid}");
+            // Delivery failed (e.g. tmux rejected an over-long send). Re-park the
+            // prompt so the next terminal attach delivers it rather than dropping
+            // the wake-up — mirrors the session-gone branch above.
+            tracing::warn!("loop: failed to deliver wake-up to workspace {wid}; re-parking");
+            let outcome = repark_prompt(pool, wid, &prompt).await;
+            if wakeup_stays_pending(&outcome, is_limit) {
+                // Transient DB error, or a manual wake-up blocked behind an
+                // already-parked prompt: leave it pending for a later tick
+                // instead of marking it fired and dropping it.
+                tracing::warn!(
+                    "loop: wake-up for workspace {wid} not re-parked ({outcome:?}); leaving pending"
+                );
+                continue;
+            }
         }
         ScheduledWakeup::mark_fired(pool, wakeup.id).await?;
     }
     Ok(())
 }
 
+/// Outcome of a [`repark_prompt`] attempt, so callers can decide whether to
+/// mark the wake-up fired or leave it pending for a retry.
+#[derive(Debug, PartialEq, Eq)]
+enum ReparkOutcome {
+    /// This wake-up's prompt is now parked for the next attach to deliver.
+    Parked,
+    /// The workspace already had an undelivered prompt parked, so this wake-up's
+    /// prompt was NOT parked (the parked one delivers first). Fine to drop a
+    /// limit-kind continuation (the loop re-detects its banner afterwards), but
+    /// a MANUAL wake-up is the user's explicit prompt and must not be dropped —
+    /// the caller leaves it pending until the slot frees.
+    AlreadyParked,
+    /// No session row exists to park on — nothing more the loop can do.
+    NoSession,
+    /// A transient DB error — leave the wake-up pending so the next tick retries
+    /// rather than silently dropping it.
+    Failed,
+}
+
+/// Best-effort: park `prompt` on the workspace's latest session so a terminal
+/// attach delivers it (the attach path pastes a parked prompt into a live
+/// agent pane, or hands it to the next fresh launch). Shared by both wake-up
+/// failure branches so the re-park policy has exactly one owner.
+///
+/// Parks only when the workspace has NO undelivered prompt anywhere: attach
+/// delivery peeks the workspace's newest pending prompt across ALL its session
+/// rows ([`Session::peek_pending_cli_prompt_for_workspace`]), so parking
+/// continuation boilerplate onto a newer, empty session while an OLDER session
+/// still holds an undelivered user prompt would let the continuation be
+/// delivered first. Skipping keeps the user prompt ahead; the loop re-detects
+/// its limit banner afterwards.
+async fn repark_prompt(pool: &sqlx::SqlitePool, wid: Uuid, prompt: &str) -> ReparkOutcome {
+    // Workspace-scoped guard, matching the workspace-scoped delivery peek — not
+    // just the latest session row's slot (which `set_pending_cli_prompt_if_empty`
+    // checks): a prompt parked on ANY row means one is already queued to deliver.
+    match Session::peek_pending_cli_prompt_for_workspace(pool, wid).await {
+        Ok(Some(_)) => {
+            tracing::info!(
+                "loop: workspace {wid} already has a parked prompt; wake-up not re-parked"
+            );
+            return ReparkOutcome::AlreadyParked;
+        }
+        Ok(None) => {}
+        Err(e) => {
+            tracing::warn!("loop: failed to check parked prompt for workspace {wid}: {e}");
+            return ReparkOutcome::Failed;
+        }
+    }
+    match Session::find_latest_by_workspace_id(pool, wid).await {
+        Ok(Some(session)) => {
+            match Session::set_pending_cli_prompt_if_empty(pool, session.id, prompt).await {
+                Ok(true) => ReparkOutcome::Parked,
+                // A prompt was parked between our check and here (raced an
+                // attach/another wake-up); treat it like the already-parked case.
+                Ok(false) => ReparkOutcome::AlreadyParked,
+                Err(e) => {
+                    tracing::warn!(
+                        "loop: failed to re-park wake-up prompt for workspace {wid}: {e}"
+                    );
+                    ReparkOutcome::Failed
+                }
+            }
+        }
+        Ok(None) => {
+            tracing::warn!("loop: no session to re-park wake-up prompt for workspace {wid}");
+            ReparkOutcome::NoSession
+        }
+        Err(e) => {
+            tracing::warn!("loop: failed to re-park wake-up prompt for workspace {wid}: {e}");
+            ReparkOutcome::Failed
+        }
+    }
+}
+
+/// Whether a wake-up whose re-park had this `outcome` must be left PENDING (not
+/// marked fired). A transient DB error always retries. `AlreadyParked` drops a
+/// limit-kind continuation (self-heals — the loop re-detects its banner) but
+/// keeps a MANUAL wake-up pending, since a manual wake-up is the user's explicit
+/// prompt and there is no second parking slot to hold it while the existing
+/// prompt delivers.
+fn wakeup_stays_pending(outcome: &ReparkOutcome, is_limit: bool) -> bool {
+    match outcome {
+        ReparkOutcome::Failed => true,
+        ReparkOutcome::AlreadyParked => !is_limit,
+        ReparkOutcome::Parked | ReparkOutcome::NoSession => false,
+    }
+}
+
 /// Scan each enabled workspace's pane and schedule a wake-up on a fresh limit.
 async fn detect_and_schedule(db: &DBService, now: DateTime<Utc>) -> Result<(), sqlx::Error> {
     let pool = &db.pool;
@@ -527,6 +672,22 @@ mod tests {
         parse_reset_at(text, now, utc())
     }
 
+    #[test]
+    fn wakeup_pending_policy_keeps_manual_but_drops_limit_when_already_parked() {
+        // Transient DB error: always retry, regardless of kind.
+        assert!(wakeup_stays_pending(&ReparkOutcome::Failed, true));
+        assert!(wakeup_stays_pending(&ReparkOutcome::Failed, false));
+        // Already-parked: a limit continuation self-heals (re-detected), so
+        // drop it; a manual wake-up is the user's prompt, so keep it pending.
+        assert!(!wakeup_stays_pending(&ReparkOutcome::AlreadyParked, true));
+        assert!(wakeup_stays_pending(&ReparkOutcome::AlreadyParked, false));
+        // Parked / no-session are final for both kinds.
+        assert!(!wakeup_stays_pending(&ReparkOutcome::Parked, true));
+        assert!(!wakeup_stays_pending(&ReparkOutcome::Parked, false));
+        assert!(!wakeup_stays_pending(&ReparkOutcome::NoSession, true));
+        assert!(!wakeup_stays_pending(&ReparkOutcome::NoSession, false));
+    }
+
     #[test]
     fn rate_limit_banner_is_classified_first() {
         let pane = "API Error: Server is temporarily limiting requests (not your usage limit) · Rate limited";