Merge branch 'main' of git.arirex.me:iris/ai-app

server/src/config.rs

@@ -246,6 +246,26 @@ pub struct SessionConfig {
     /// never arrived is not diagnosable at all.
     #[serde(default = "notify_default")]
     pub notify: bool,
+    /// Whether this session's process is stopped when the server exits,
+    /// instead of being left running for the next start to adopt.
+    ///
+    /// A fact about the session rather than about the run that spawned it,
+    /// which is why it is persisted: whichever server is running when the
+    /// time comes is the one that has to act on it, and a session nobody
+    /// meant to keep should not depend on the same server still being up
+    /// to clean it away.
+    ///
+    /// Written by a server started with `--throwaway-sessions`, which is
+    /// the default in a debug build. A session spawned while testing is
+    /// one nobody means to keep, and under the ordinary rule its `claude`
+    /// outlives every server that ever knew about it -- twelve of them
+    /// accumulated on this machine in a day, each holding a conversation
+    /// open.
+    ///
+    /// Absent means false: every session written before this existed, and
+    /// every one spawned by a release build.
+    #[serde(default, skip_serializing_if = "not_set")]
+    pub throwaway: bool,
     /// Epoch seconds when the session was spawned.
     pub created: f64,
 }
@@ -254,6 +274,13 @@ fn notify_default() -> bool {
     true
 }
 
+/// Keeps the ordinary case out of the file entirely -- see
+/// [`SessionConfig::throwaway`], which is false for every session a
+/// production build writes.
+fn not_set(flag: &bool) -> bool {
+    !*flag
+}
+
 /// The name of the echo provider, and of the setup this machine gets on
 /// first run.
 ///
@@ -421,6 +448,7 @@ mod tests {
                 permission_mode: None,
                 params: BTreeMap::new(),
                 notify: true,
+                throwaway: false,
                 created: 1234.5,
             }],
         };

server/src/main.rs

@@ -93,6 +93,34 @@ struct Args {
     /// This reopens them on demand rather than by unplugging something.
     #[arg(long, default_value_t = 0, value_name = "MS")]
     delay: u64,
+
+    /// Mark every session spawned here as throwaway: its process is
+    /// stopped when this server exits, instead of being left running for
+    /// the next start to adopt. On by default in a debug build.
+    ///
+    /// Sessions outlive the backend on purpose, which is right for the
+    /// ones somebody is using and wrong for the ones a test made: a
+    /// session spawned to check something leaves a `claude` behind that
+    /// every later server adopts, and they accumulate silently -- twelve
+    /// of them on this machine in a day, each holding a conversation open.
+    /// So a development build cleans up after itself unless told not to
+    /// (`--throwaway-sessions=false`), and a release build never does
+    /// unless asked.
+    ///
+    /// The flag decides only what *new* sessions are marked as. What
+    /// happens on the way out is decided by the mark, which is written
+    /// into the session and outlives the server that made it -- so
+    /// sessions spawned without it keep running, whichever server is up
+    /// when one exits.
+    #[arg(
+        long,
+        default_value_t = cfg!(debug_assertions),
+        action = clap::ArgAction::Set,
+        num_args = 0..=1,
+        default_missing_value = "true",
+        value_name = "BOOL",
+    )]
+    throwaway_sessions: bool,
 }
 
 #[tokio::main]
@@ -122,8 +150,15 @@ async fn main() -> Result<()> {
     let models = Arc::new(models::ModelStore::new(models_dir.clone()));
     let manager = Arc::new(
         SessionManager::new(config_path.clone(), data_dir, models_dir.clone())
-            .with_context(|| format!("failed to load {}", config_path.display()))?,
+            .with_context(|| format!("failed to load {}", config_path.display()))?
+            .marking_new_sessions_throwaway(args.throwaway_sessions),
     );
+    if args.throwaway_sessions {
+        tracing::warn!(
+            "sessions spawned here are marked throwaway -- their processes are stopped when this \
+             server exits rather than left running (--throwaway-sessions=false to keep them)"
+        );
+    }
     // After construction rather than inside it: seeding asks this machine
     // what it has, which is I/O, and a constructor that quietly runs a
     // subprocess is a surprise to every caller including the tests.
@@ -244,16 +279,21 @@ async fn main() -> Result<()> {
     // their processes are meant to outlive this one, so restarting the
     // backend does not end a turn somebody is waiting on. Each is recorded
     // in its session directory and adopted again on the way back up (see
-    // `session::process`). Both signals, because systemd and OpenRC send
-    // TERM while a terminal sends INT.
+    // `session::process`). The exception is the sessions marked throwaway,
+    // which are stopped first -- see `--throwaway-sessions`. Both signals,
+    // because systemd and OpenRC send TERM while a terminal sends INT.
     let serving = axum_server::bind_rustls(addr, tls_config)
         .serve(app.into_make_service_with_connect_info::<SocketAddr>());
     let mut terminate = signal(SignalKind::terminate()).context("listening for SIGTERM")?;
     tokio::select! {
         served = serving => served.context("TLS listener failed")?,
-        _ = terminate.recv() => tracing::info!("SIGTERM -- detaching from sessions"),
-        _ = tokio::signal::ctrl_c() => tracing::info!("interrupted -- detaching from sessions"),
+        _ = terminate.recv() => tracing::info!("SIGTERM -- letting go of sessions"),
+        _ = tokio::signal::ctrl_c() => tracing::info!("interrupted -- letting go of sessions"),
     }
+    // Stopped before the rest are let go of, and on every way out of the
+    // select above: a throwaway session is one nobody meant to keep, and
+    // the whole point is that nothing has to remember to clean it up.
+    manager.stop_throwaway_sessions();
     manager.detach_all();
 
     Ok(())

server/src/session/driver.rs

@@ -556,10 +556,16 @@ pub trait Driver: Send + Sync {
     }
 
     fn detach(&self);
-    /// End the process for good, because the session it belongs to is
-    /// being deleted. The path out for everything [`detach`] preserves.
+    /// End the process for good, because it must not survive this. The
+    /// path out for everything [`detach`] preserves.
+    ///
+    /// Two callers, and the difference between them is only what is being
+    /// ended: a session being deleted, whose conversation goes with it, and
+    /// a throwaway session at a server's exit, whose transcript stays and
+    /// whose process does not (see [`SessionConfig::throwaway`]).
     ///
     /// [`detach`]: Driver::detach
+    /// [`SessionConfig::throwaway`]: crate::config::SessionConfig::throwaway
     fn stop(&self);
 }
 

server/src/session/mod.rs

@@ -530,6 +530,10 @@ pub struct SessionManager {
     /// Held here rather than per session for the reason
     /// [`SessionManager::subscribe_notifications`] gives.
     notifications: broadcast::Sender<Notification>,
+    /// What to mark sessions spawned here as -- see
+    /// [`SessionManager::marking_new_sessions_throwaway`] and
+    /// [`SessionConfig::throwaway`].
+    spawn_throwaway: bool,
     inner: RwLock<Inner>,
 }
 
@@ -581,11 +585,30 @@ impl SessionManager {
             data_dir,
             models_dir,
             notifications,
+            spawn_throwaway: false,
             inner: RwLock::new(Inner { config, live }),
         };
         Ok(manager)
     }
 
+    /// Marks every session spawned from here on as one whose process is
+    /// stopped when this server exits -- see [`SessionConfig::throwaway`]
+    /// and [`SessionManager::stop_throwaway_sessions`].
+    ///
+    /// Set from `--throwaway-sessions`, which a debug build defaults to on.
+    /// It decides only what a *new* session is marked as; what happens on
+    /// the way out is decided by the mark, which is the session's own and
+    /// outlives the server that made it.
+    ///
+    /// Consuming rather than a fourth constructor parameter: it is one
+    /// caller's business, and every test and every other caller would
+    /// otherwise have to say "no, not that" at a constructor that is
+    /// already about three paths.
+    pub fn marking_new_sessions_throwaway(mut self, throwaway: bool) -> Self {
+        self.spawn_throwaway = throwaway;
+        self
+    }
+
     /// Writes this machine into a config that has no setups, with the
     /// providers actually found on it.
     ///
@@ -783,9 +806,81 @@ impl SessionManager {
         for session in inner.live.values() {
             session.detach();
         }
+        // Counted from the records rather than from the sessions: the
+        // throwaway ones have just been stopped and their records cleared
+        // (see `stop_throwaway_sessions`), so the number of *sessions*
+        // would promise the next start processes that are not there.
+        let left = inner
+            .live
+            .values()
+            .filter(|session| process::live(session.dir()).is_some())
+            .count();
+        tracing::info!("left {left} session process(es) running to be reattached to");
+    }
+
+    /// Ends the process of every session marked throwaway, and waits for
+    /// them to actually go.
+    ///
+    /// The counterpart to [`SessionManager::detach_all`], and the two are
+    /// called in that order on the way out: this one deals with the
+    /// sessions nobody meant to keep, and everything else is let go of
+    /// still running, as it always was.
+    ///
+    /// Which sessions those are is read from the *mark*, never from what
+    /// this server was told at startup -- see
+    /// [`SessionConfig::throwaway`]. A session spawned by a test run is
+    /// something to clean away whichever server happens to be up when it
+    /// ends, and a server started without the flag must not adopt a pile
+    /// of test sessions and then be the one thing keeping them alive.
+    ///
+    /// Waiting is the part that cannot be skipped. `process::stop` leaves
+    /// its SIGKILL on a tokio timer, and a runtime that is shutting down
+    /// never runs it -- so without [`process::wait_gone`] this would report
+    /// stopping processes that go on running, which is how the original
+    /// `shutdown_all` leaked them.
+    pub fn stop_throwaway_sessions(&self) {
+        let inner = self.inner.read().unwrap();
+        let throwaway: Vec<&SessionConfig> = inner
+            .config
+            .sessions
+            .iter()
+            .filter(|meta| meta.throwaway)
+            .collect();
+        // Taken before anything is asked to stop: `Driver::stop` forgets
+        // the record, and what has to be waited for is exactly what was
+        // signalled.
+        let records: Vec<process::Record> = throwaway
+            .iter()
+            .filter_map(|meta| process::live(&self.data_dir.join(&meta.id)))
+            .collect();
+        for meta in &throwaway {
+            let dir = self.data_dir.join(&meta.id);
+            match inner
+                .live
+                .get(&meta.id)
+                .and_then(|session| session.driver())
+            {
+                Some(driver) => driver.stop(),
+                // No driver is either a session with no process -- nothing
+                // to do -- or one whose launch failed with a process still
+                // running, which is the case worth covering: the record is
+                // the session's rather than any dialect's, which is the
+                // same reason `stop_session` signals it directly.
+                None => {
+                    if let Some(record) = process::live(&dir) {
+                        process::stop(&record, process::STOP_GRACE);
+                        process::clear(&dir);
+                    }
+                }
+            }
+        }
+        if records.is_empty() {
+            return;
+        }
+        process::wait_gone(&records, process::STOP_GRACE);
         tracing::info!(
-            "left {} session process(es) running to be reattached to",
-            inner.live.len()
+            "stopped {} throwaway session process(es) rather than leaving them running",
+            records.len()
         );
     }
 
@@ -955,6 +1050,11 @@ impl SessionManager {
             // waiting for, and a switch on the spawn screen would be a
             // decision asked before there is anything to decide about.
             notify: true,
+            // What this server was told to mark new sessions as. Recorded
+            // on the session rather than remembered here, so whichever
+            // server is running when the time comes knows what to do with
+            // it -- see `SessionConfig::throwaway`.
+            throwaway: self.spawn_throwaway,
             created: now(),
         };
 
@@ -2870,6 +2970,108 @@ mod tests {
         assert_eq!(manager.sessions()[0].status, SessionStatus::Idle);
     }
 
+    /// Seeds a config with a stand-in for the Claude CLI, and returns the
+    /// provider's name.
+    ///
+    /// A shell script that holds its stdin open and writes nothing, so it
+    /// lives exactly as long as nobody signals it. What the throwaway rule
+    /// is about is a process's lifetime rather than any dialect, and the
+    /// real CLI would cost tokens to say the same thing.
+    fn seed_stand_in_cli(config_path: &Path, dir: &Path) -> String {
+        use std::os::unix::fs::PermissionsExt;
+
+        let command = dir.join("stand-in-cli");
+        std::fs::write(&command, "#!/bin/sh\ncat > /dev/null\n").expect("write stand-in");
+        std::fs::set_permissions(&command, std::fs::Permissions::from_mode(0o755)).expect("chmod");
+        Config {
+            setups: vec![Config::seed(vec![
+                Config::echo_provider(),
+                ProviderConfig {
+                    name: "stand-in".to_string(),
+                    kind: DriverKind::ClaudeCli,
+                    command: Some(command.to_string_lossy().into_owned()),
+                    models: Vec::new(),
+                },
+            ])],
+            ..Config::default()
+        }
+        .save(config_path)
+        .expect("seed config");
+        "stand-in".to_string()
+    }
+
+    fn stand_in_spec(provider: &str) -> SpawnSpec {
+        SpawnSpec {
+            provider: provider.to_string(),
+            ..echo_spec()
+        }
+    }
+
+    /// A session spawned while testing is cleaned away on the way out, and
+    /// the sessions beside it are not.
+    ///
+    /// The two halves are one rule. Leaving processes running is the whole
+    /// design -- a rebuild must not end a turn -- and it is exactly wrong
+    /// for a session nobody meant to keep: those leave a `claude` behind
+    /// that every later server adopts, and they accumulate unnoticed. So
+    /// the mark decides, and it is the session's own rather than the
+    /// running server's, which is what this asks: the manager that stops
+    /// them is not the one that spawned the session it must not touch.
+    #[tokio::test]
+    async fn only_sessions_marked_throwaway_are_stopped_on_the_way_out() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let config_path = dir.path().join("config.ron");
+        let data_dir = dir.path().join("sessions");
+        let provider = seed_stand_in_cli(&config_path, dir.path());
+
+        // Spawned by a server that marks nothing: this one is somebody's.
+        let manager = SessionManager::new(
+            config_path.clone(),
+            data_dir.clone(),
+            data_dir.join("models"),
+        )
+        .expect("manager");
+        let keeper = manager
+            .spawn_session(stand_in_spec(&provider))
+            .expect("spawn keeper");
+        let keeper_process =
+            process::live(&data_dir.join(&keeper.id)).expect("the keeper has a process");
+        drop(manager);
+
+        // And a second server, marking what it spawns, which adopts the
+        // first one's session.
+        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
+            .expect("manager restart")
+            .marking_new_sessions_throwaway(true);
+        let throwaway = manager
+            .spawn_session(stand_in_spec(&provider))
+            .expect("spawn throwaway");
+        let throwaway_process =
+            process::live(&data_dir.join(&throwaway.id)).expect("the throwaway has a process");
+
+        manager.stop_throwaway_sessions();
+
+        // Both answers taken before anything is asserted, and the keeper
+        // ended here: a failing assertion must not be what decides whether
+        // this test leaves a process behind.
+        let throwaway_after = throwaway_process.liveness();
+        let keeper_after = keeper_process.liveness();
+        manager.delete_session(&keeper.id).expect("delete keeper");
+        process::wait_gone(&[keeper_process], process::STOP_GRACE);
+
+        assert_eq!(
+            throwaway_after,
+            process::Liveness::Dead,
+            "a throwaway session's process outlived the server that spawned it",
+        );
+        assert_eq!(
+            keeper_after,
+            process::Liveness::Alive,
+            "a session nobody marked was stopped along with the throwaway ones -- restarting the \
+             backend is not allowed to end a turn",
+        );
+    }
+
     /// Backdates every line in a transcript, so a restart has something to
     /// report that the clock could not have produced.
     fn rewrite_transcript_times(path: &Path, ts: f64) {

server/src/session/process.rs

@@ -84,7 +84,7 @@ impl Record {
     pub fn of(pid: u32, detail: Detail) -> Option<Self> {
         Some(Self {
             pid,
-            started: started_at(pid).ok().flatten()?,
+            started: stat_of(pid).ok().flatten()?.started,
             detail,
         })
     }
@@ -92,10 +92,16 @@ impl Record {
     /// Whether the process this describes is still the one running under
     /// that pid.
     pub fn liveness(&self) -> Liveness {
-        match started_at(self.pid) {
+        match stat_of(self.pid) {
             // A different start time is a reused pid, which is a different
             // process and so definitely not ours.
-            Ok(Some(started)) if started == self.started => Liveness::Alive,
+            Ok(Some(stat)) if stat.started == self.started => {
+                if stat.exited {
+                    Liveness::Dead
+                } else {
+                    Liveness::Alive
+                }
+            }
             Ok(Some(_)) | Ok(None) => Liveness::Dead,
             Err(_) => Liveness::Unknown,
         }
@@ -226,17 +232,54 @@ pub fn stop(record: &Record, grace: std::time::Duration) {
     let record = record.clone();
     tokio::spawn(async move {
         tokio::time::sleep(grace).await;
-        if record.liveness() == Liveness::Alive {
-            tracing::warn!(
-                "session process {} did not stop within {:?}; killing it",
-                record.pid,
-                grace
-            );
-            signal(record.pid, libc::SIGKILL);
-        }
+        kill_if_still_there(&record, grace);
     });
 }
 
+/// Waits for processes already asked to stop, and kills whichever have
+/// not, for a caller that is about to exit.
+///
+/// The waiting cannot be [`stop`]'s here, and that is the whole reason
+/// this exists: the kill it leaves behind is a timer inside the tokio
+/// runtime, and a runtime that is shutting down never runs it. That is
+/// how the backend's original `shutdown_all` leaked the processes it had
+/// just asked to stop -- it reported them stopped, too, which is worse
+/// than not asking.
+///
+/// One deadline for all of them rather than one each: they were signalled
+/// together, so waiting is bounded by the grace period however many there
+/// are, and a server does not sit for a minute on the way out.
+pub fn wait_gone(records: &[Record], grace: std::time::Duration) {
+    /// How often to look. Short enough that the ordinary case -- a
+    /// process that goes at once -- costs nothing noticeable, and long
+    /// enough not to spin.
+    const LOOK: std::time::Duration = std::time::Duration::from_millis(20);
+
+    let deadline = std::time::Instant::now() + grace;
+    for record in records {
+        while record.liveness() == Liveness::Alive && std::time::Instant::now() < deadline {
+            std::thread::sleep(LOOK);
+        }
+        kill_if_still_there(record, grace);
+    }
+}
+
+/// The end of both paths above: a process that was asked to stop and did
+/// not is killed. Written once because the two callers differ only in how
+/// they wait, and a grace period that means one thing in one of them and
+/// something else in the other is exactly the drift `STOP_GRACE` was
+/// gathered here to prevent.
+fn kill_if_still_there(record: &Record, grace: std::time::Duration) {
+    if record.liveness() == Liveness::Alive {
+        tracing::warn!(
+            "session process {} did not stop within {:?}; killing it",
+            record.pid,
+            grace
+        );
+        signal(record.pid, libc::SIGKILL);
+    }
+}
+
 fn signal(pid: u32, signal: libc::c_int) {
     // SAFETY: `kill` with a positive pid touches only that process, and
     // the pid came from a record whose start time was just confirmed to
@@ -260,7 +303,26 @@ fn signal(pid: u32, signal: libc::c_int) {
 /// first is what would let a machine without a readable `/proc` look like
 /// a machine with nothing running on it. Linux-specific, like `import`'s
 /// use of GNU `stat`.
-fn started_at(pid: u32) -> std::io::Result<Option<u64>> {
+/// What `/proc` says about a pid.
+struct Stat {
+    /// The kernel's start time in clock ticks since boot -- see
+    /// [`Record::started`].
+    started: u64,
+    /// State `Z`: the process has ended, and the kernel is keeping its
+    /// entry only until somebody collects the exit status.
+    ///
+    /// Read rather than ignored, because the entry it leaves behind has
+    /// the same pid *and* the same start time, so a process that has
+    /// plainly finished goes on answering "still there" for as long as
+    /// nothing reaps it. None of this module's callers want that answer: a
+    /// session whose CLI has exited is over whether or not the status has
+    /// been collected, and reporting it alive makes `Exited` unsayable --
+    /// the session shows `unknown`, its Start button never appears, and
+    /// stopping it says there is nothing to stop.
+    exited: bool,
+}
+
+fn stat_of(pid: u32) -> std::io::Result<Option<Stat>> {
     let stat = match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
         Ok(stat) => stat,
         Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
@@ -272,14 +334,16 @@ fn started_at(pid: u32) -> std::io::Result<Option<u64>> {
     let unreadable =
         || std::io::Error::new(std::io::ErrorKind::InvalidData, "unreadable /proc stat");
     let after_name = stat.rsplit_once(')').ok_or_else(unreadable)?.1;
-    // Field 3 is the first after the name, so field 22 is the 20th here.
-    after_name
-        .split_whitespace()
-        .nth(19)
+    // Field 3 is the first after the name, so the state is the first here
+    // and field 22 is the 20th.
+    let mut fields = after_name.split_whitespace();
+    let exited = fields.next().ok_or_else(unreadable)? == "Z";
+    let started = fields
+        .nth(18)
         .ok_or_else(unreadable)?
         .parse()
-        .map(Some)
-        .map_err(|_| unreadable())
+        .map_err(|_| unreadable())?;
+    Ok(Some(Stat { started, exited }))
 }
 
 /// Reads `path` from `from`, returning what is there and where reading