fix(session): port conservative input latch and quit stamping from main (0046 M3.4)

crates/runner-app/src/session/manager/lifecycle.rs

@@ -66,6 +66,8 @@ impl SessionManager {
                 Some(mut h) => {
                     state.activity = None;
                     state.suppress_local_input_busy = false;
+                    state.local_input_pending = false;
+                    state.last_local_input_at = None;
                     state.mission_status_sink = None;
                     state.completion_armed = false;
                     state.pending_resize = None;
@@ -290,6 +292,8 @@ impl SessionManager {
                 state.handle = None;
                 state.activity = None;
                 state.suppress_local_input_busy = false;
+                state.local_input_pending = false;
+                state.last_local_input_at = None;
                 state.mission_status_sink = None;
                 state.completion_armed = false;
             }

crates/runner-app/src/session/manager/mod.rs

@@ -556,6 +556,8 @@ struct SessionState {
     handle: Option<SessionHandle>,
     activity: Option<SessionActivityState>,
     suppress_local_input_busy: bool,
+    local_input_pending: bool,
+    last_local_input_at: Option<Instant>,
     mission_status_sink: Option<ForwarderEmitCtx>,
     completion_armed: bool,
     output_buffer: VecDeque<OutputEvent>,
@@ -589,6 +591,8 @@ impl SessionState {
         self.handle.is_none()
             && self.activity.is_none()
             && !self.suppress_local_input_busy
+            && !self.local_input_pending
+            && self.last_local_input_at.is_none()
             && self.mission_status_sink.is_none()
             && !self.completion_armed
             && self.output_buffer.is_empty()
@@ -784,6 +788,8 @@ impl SessionManager {
     ) {
         let state = self.session_state_or_insert(session_id);
         let mut state = state.lock().unwrap();
+        state.local_input_pending = false;
+        state.last_local_input_at = None;
         state.handle = Some(handle);
         state.mission_status_sink = mission_status_sink;
         state.killed = false;

crates/runner-app/src/session/manager/output.rs

@@ -4,6 +4,60 @@ pub(super) const PURGE_RESUME_RESET: &[u8] = b"\x1bc";
 pub(super) const KEEP_RESUME_SEAM: &[u8] =
     b"\x1b[0m\x1b[?2004l\x1b[?1000l\x1b[?1002l\x1b[?1003l\x1b[?1006l\r\n";
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum LocalInputClass {
+    SetPending,
+    ClearPending,
+    ActivityOnly,
+}
+
+pub(super) fn classify_local_input(bytes: &[u8]) -> Option<LocalInputClass> {
+    if bytes.is_empty() {
+        return None;
+    }
+    if bytes == b"\r" || bytes == b"\x03" {
+        return Some(LocalInputClass::ClearPending);
+    }
+    if bytes == b"\x16" || bytes.starts_with(b"\x1b[200~") {
+        return Some(LocalInputClass::SetPending);
+    }
+    if bytes.starts_with(b"\x1b") {
+        return Some(LocalInputClass::ActivityOnly);
+    }
+    if bytes
+        .iter()
+        .any(|byte| matches!(byte, 0x20..=0x7e | 0x80..=0xff))
+    {
+        Some(LocalInputClass::SetPending)
+    } else {
+        Some(LocalInputClass::ActivityOnly)
+    }
+}
+
+pub(super) fn update_local_input_state(
+    state: &mut SessionState,
+    input_class: Option<LocalInputClass>,
+    now: Instant,
+) -> bool {
+    match input_class {
+        Some(LocalInputClass::SetPending) => {
+            state.local_input_pending = true;
+            state.last_local_input_at = Some(now);
+            false
+        }
+        Some(LocalInputClass::ClearPending) => {
+            state.local_input_pending = false;
+            state.last_local_input_at = None;
+            true
+        }
+        Some(LocalInputClass::ActivityOnly) => {
+            state.last_local_input_at = Some(now);
+            false
+        }
+        None => false,
+    }
+}
+
 impl SessionManager {
     /// Forwarder thread shared by `spawn`, `spawn_direct`, and `resume`.
     /// Drains the runtime's `OutputStream` into `session/output`
@@ -216,12 +270,15 @@ impl SessionManager {
     ) -> Result<()> {
         let rt_session = self.live_runtime_session(session_id)?;
         let submitted = bytes == b"\r";
+        let input_class = classify_local_input(bytes);
         let session = self.session_state(session_id);
         let (transition, mission_status_sink, mission_scoped) =
             if let Some(session) = session.as_ref() {
                 let mut session = session.lock().unwrap();
                 let previous_activity = session.activity;
                 let previous_suppression = session.suppress_local_input_busy;
+                let previous_input_pending = session.local_input_pending;
+                let previous_input_at = session.last_local_input_at;
                 let mission_status_sink = session.mission_status_sink.clone();
                 let mission_scoped = session
                     .handle
@@ -245,9 +302,15 @@ impl SessionManager {
                     }
                     None
                 };
+                // The deferred router slice will consume the clear transition and
+                // recent-input timestamp when it ports delivery gating.
+                let _input_cleared =
+                    update_local_input_state(&mut session, input_class, Instant::now());
                 if let Err(error) = self.write_stdin_bytes(&rt_session, bytes) {
                     session.activity = previous_activity;
                     session.suppress_local_input_busy = previous_suppression;
+                    session.local_input_pending = previous_input_pending;
+                    session.last_local_input_at = previous_input_at;
                     return Err(error);
                 }
                 if submitted {

crates/runner-app/src/session/manager/tests.rs

@@ -1103,6 +1103,66 @@ fn inject_stdin_on_unknown_session_errors_cleanly() {
     assert!(format!("{err}").contains("session not found"));
 }
 
+#[test]
+fn local_input_byte_classes_drive_pending_state() {
+    use super::output::{classify_local_input, update_local_input_state, LocalInputClass};
+
+    assert_eq!(
+        classify_local_input(b"x"),
+        Some(LocalInputClass::SetPending)
+    );
+    assert_eq!(
+        classify_local_input("界".as_bytes()),
+        Some(LocalInputClass::SetPending)
+    );
+    for protocol in [
+        b"\x1b[A".as_slice(),
+        b"\x1b]10;rgb:dcdc/dcdc/e0e0\x1b\\",
+        b"\x1b]11;rgb:1515/1616/1b1b\x1b\\",
+    ] {
+        assert_eq!(
+            classify_local_input(protocol),
+            Some(LocalInputClass::ActivityOnly),
+            "terminal protocol traffic must not mark local input pending"
+        );
+    }
+    assert_eq!(
+        classify_local_input(b"\x1b[200~pasted text\x1b[201~"),
+        Some(LocalInputClass::SetPending)
+    );
+    assert_eq!(
+        classify_local_input(b"\x16"),
+        Some(LocalInputClass::SetPending)
+    );
+    assert_eq!(
+        classify_local_input(b"\r"),
+        Some(LocalInputClass::ClearPending)
+    );
+    assert_eq!(
+        classify_local_input(b"\x03"),
+        Some(LocalInputClass::ClearPending)
+    );
+
+    let now = Instant::now();
+    let mut state = SessionState::default();
+    update_local_input_state(&mut state, classify_local_input(b"draft"), now);
+    assert!(state.local_input_pending);
+    assert_eq!(state.last_local_input_at, Some(now));
+
+    update_local_input_state(&mut state, classify_local_input(b"\r"), now);
+    assert!(!state.local_input_pending);
+    assert!(state.last_local_input_at.is_none());
+
+    update_local_input_state(&mut state, classify_local_input(b"\x1b[D"), now);
+    assert!(!state.local_input_pending);
+    assert_eq!(state.last_local_input_at, Some(now));
+
+    state.local_input_pending = true;
+    update_local_input_state(&mut state, classify_local_input(b"\x03"), now);
+    assert!(!state.local_input_pending);
+    assert!(state.last_local_input_at.is_none());
+}
+
 // `await_pty_output` was deleted in the Step 9 cutover. Tests
 // that previously observed echoed bytes from /bin/cat through
 // a portable-pty master now assert on FakeRuntime's captured
@@ -2115,6 +2175,14 @@ fn direct_chat_typing_stays_idle_until_submit() {
 
     mgr.inject_direct_stdin(&spawned.id, b"x", cap.as_ref())
         .unwrap();
+    assert!(
+        mgr.session_state(&spawned.id)
+            .unwrap()
+            .lock()
+            .unwrap()
+            .local_input_pending,
+        "printable terminal input must latch pending input",
+    );
     assert!(
         !mgr.take_completion_armed(std::slice::from_ref(&spawned.id)),
         "typing without submit must not arm completion",
@@ -2132,6 +2200,12 @@ fn direct_chat_typing_stays_idle_until_submit() {
 
     mgr.inject_direct_stdin(&spawned.id, b"\r", cap.as_ref())
         .unwrap();
+    {
+        let state = mgr.session_state(&spawned.id).unwrap();
+        let state = state.lock().unwrap();
+        assert!(!state.local_input_pending, "Enter must clear the latch");
+        assert!(state.last_local_input_at.is_none());
+    }
     let submitted = wait_for_session_status_event(&cap, &spawned.id, SessionActivityState::Busy);
     assert_eq!(submitted.source, "input-submit");
     assert_eq!(

crates/runner-native/src/bootstrap.rs

@@ -1,8 +1,8 @@
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
-use anyhow::{Context as _, Result};
-use runner_app::{db, event_bus, events, mcp, session, shell_path, windows, AppCore};
+use anyhow::{bail, Context as _, Result};
+use runner_app::{db, event_bus, events, mcp, repo, session, shell_path, windows, AppCore};
 
 pub const APP_IDENTIFIER: &str = "com.wycstudios.runner";
 
@@ -76,24 +76,20 @@ pub fn boot_core(paths: &NativePaths) -> Result<AppCore> {
     Ok(core)
 }
 
-pub fn stop_running_direct_sessions(core: &AppCore) -> Result<()> {
+pub fn stop_running_sessions_on_quit(core: &AppCore) -> Result<()> {
     let ids = {
-        let conn = core.db.get().context("get database connection")?;
-        let mut stmt = conn.prepare(
-            "SELECT id FROM sessions
-             WHERE status = 'running' AND mission_id IS NULL",
-        )?;
-        let rows = stmt.query_map([], |row| row.get::<_, String>(0))?;
-        let mut ids = Vec::new();
-        for row in rows {
-            ids.push(row?);
-        }
-        ids
+        let mut conn = core.db.get().context("get database connection")?;
+        repo::session::mark_running_for_resume_on_launch(&mut conn)
+            .context("stamp sessions for resume on launch")?
     };
+    let mut failures = Vec::new();
     for id in ids {
-        core.sessions
-            .kill(&id)
-            .with_context(|| format!("stop direct session {id}"))?;
+        if let Err(error) = core.sessions.kill(&id) {
+            failures.push(format!("{id}: {error}"));
+        }
+    }
+    if !failures.is_empty() {
+        bail!("failed to stop sessions on quit: {}", failures.join("; "));
     }
     Ok(())
 }
@@ -118,4 +114,59 @@ mod tests {
         assert!(debug.app_data_dir.ends_with("com.wycstudios.runner-dev"));
         assert!(debug.log_dir.ends_with("com.wycstudios.runner-dev"));
     }
+
+    #[test]
+    fn quit_preparation_stamps_running_sessions_and_requeues_claims() {
+        let temp = tempfile::tempdir().unwrap();
+        let pool = Arc::new(db::open_pool(&temp.path().join("runner.db")).unwrap());
+        {
+            let conn = pool.get().unwrap();
+            conn.execute(
+                "INSERT INTO runners
+                    (id, handle, display_name, runtime, command,
+                     args_json, created_at, updated_at)
+                 VALUES
+                    ('r1', 'alpha', 'Alpha', 'shell', '/bin/cat',
+                     '[]', '2026-08-18T00:00:00Z', '2026-08-18T00:00:00Z')",
+                [],
+            )
+            .unwrap();
+            conn.execute(
+                "INSERT INTO sessions
+                    (id, runner_id, status, started_at, resume_on_launch)
+                 VALUES
+                    ('s1', 'r1', 'running', '2026-08-18T00:00:00Z', 0),
+                    ('claimed', 'r1', 'stopped', '2026-08-18T00:00:01Z', 2)",
+                [],
+            )
+            .unwrap();
+        }
+        let runtime: Arc<dyn session::runtime::SessionRuntime> =
+            Arc::new(session::pty_runtime::PtyRuntime::new());
+        let core = AppCore {
+            db: Arc::clone(&pool),
+            app_data_dir: PathBuf::new(),
+            sessions: session::SessionManager::new(shell_path::LoginShellEnv::default(), runtime),
+            buses: event_bus::BusRegistry::new(),
+            routers: runner_app::router::RouterRegistry::new(),
+            mcp: Arc::new(mcp::McpHandle::new()),
+            windows: Arc::new(windows::WindowRegistry::new()),
+            events: events::EventChannel::new(),
+            app_version: "0.0.0-test".into(),
+        };
+
+        stop_running_sessions_on_quit(&core).unwrap();
+
+        let conn = pool.get().unwrap();
+        for id in ["s1", "claimed"] {
+            let stamp: i64 = conn
+                .query_row(
+                    "SELECT resume_on_launch FROM sessions WHERE id = ?1",
+                    [id],
+                    |row| row.get(0),
+                )
+                .unwrap();
+            assert_eq!(stamp, 1, "{id} must be pending for the next launch");
+        }
+    }
 }

crates/runner-native/src/main.rs

@@ -16,7 +16,7 @@ use runner_app::model::{Runner, SessionStatus};
 use runner_app::ops::session::DirectSessionEntry;
 use runner_app::AppCore;
 use runner_native::bootstrap::{
-    boot_core, native_paths, stop_running_direct_sessions, NativePaths,
+    boot_core, native_paths, stop_running_sessions_on_quit, NativePaths,
 };
 use runner_native::pane_layout::{
     PaneLayout, PaneLeaf, PaneNode, PresetKind, SplitOrientation, TabSet,
@@ -215,13 +215,17 @@ fn run() -> Result<()> {
     Application::new().run(move |cx: &mut App| {
         let quit_core = core.clone();
         cx.on_action(move |_: &Quit, cx| {
-            let _ = stop_running_direct_sessions(&quit_core);
+            if let Err(error) = stop_running_sessions_on_quit(&quit_core) {
+                eprintln!("Runner Native quit session teardown failed: {error:#}");
+            }
             cx.quit();
         });
         let close_core = core.clone();
         cx.on_window_closed(move |cx| {
-            let _ = stop_running_direct_sessions(&close_core);
             if cx.windows().is_empty() {
+                if let Err(error) = stop_running_sessions_on_quit(&close_core) {
+                    eprintln!("Runner Native quit session teardown failed: {error:#}");
+                }
                 cx.quit();
             }
         })
@@ -255,7 +259,7 @@ fn run() -> Result<()> {
         cx.activate(true);
     });
 
-    stop_running_direct_sessions(&shutdown_core)?;
+    stop_running_sessions_on_quit(&shutdown_core)?;
     Ok(())
 }