ai-app/server/src/session/pending.rs

//! What is being done to a machine's Claude Code sessions right now.
//!
//! Importing and deleting used to be whatever the phone was in the middle of:
//! the request was the work, so leaving the screen cancelled it and coming back
//! showed no sign it had ever started. Sessions half-imported that way are the
//! expensive kind of missing -- the row is back in the list looking untouched,
//! and taking it again is the second `--resume` the import path exists to
//! prevent.
//!
//! So the work runs here, on the server, and this is the record of it. The
//! phone reads that record two ways and needs both: every row of the importable
//! listing carries what is happening to it, which is what a phone that was
//! asleep has to go on; and [`Registry::subscribe`] is the live stream, which is
//! what makes a screen change by itself. A broadcast has no memory, and a
//! listing is only true when it was fetched.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use serde::Serialize;
use tokio::sync::broadcast;

/// What is being done to an importable session.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum Operation {
    Importing,
    Deleting,
}

impl Operation {
    /// The word a row shows while this runs. Fixed here rather than in the app
    /// so the two ends cannot disagree about what a state is called.
    pub fn label(self) -> &'static str {
        match self {
            Self::Importing => "importing",
            Self::Deleting => "deleting",
        }
    }
}

/// One change to what is in flight, as it goes out on the stream.
///
/// The three states are every way an operation ends, including the two easy to
/// leave out: still running, finished, and failed. There is deliberately no
/// "unknown" -- this is the server's own work, so not knowing would be a bug.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase", tag = "state")]
pub enum Change {
    Started {
        setup: String,
        session: String,
        operation: Operation,
    },
    Finished {
        setup: String,
        session: String,
    },
    Failed {
        setup: String,
        session: String,
        message: String,
    },
}

impl Change {
    /// Which machine this is about, so a stream scoped to one can drop the rest.
    /// Every variant carries it; matching here keeps that fact in one place.
    pub fn setup(&self) -> &str {
        match self {
            Self::Started { setup, .. }
            | Self::Finished { setup, .. }
            | Self::Failed { setup, .. } => setup,
        }
    }
}

/// Everything in flight, and the last failure against each session.
#[derive(Debug)]
pub struct Registry {
    running: Mutex<HashMap<(String, String), Operation>>,
    /// Kept after the operation ends, because a phone that was not looking when
    /// it failed has no other way to find out. Replaced when the next operation
    /// on that session starts, and dropped by [`Registry::prune`] when the
    /// session is no longer on the machine.
    failures: Mutex<HashMap<(String, String), String>>,
    changes: broadcast::Sender<Change>,
}

impl Default for Registry {
    fn default() -> Self {
        Self {
            running: Mutex::new(HashMap::new()),
            failures: Mutex::new(HashMap::new()),
            // Enough that a phone watching one screen cannot lag behind a batch
            // of any size somebody would start by hand.
            changes: broadcast::channel(256).0,
        }
    }
}

impl Registry {
    /// Marks an operation as running and announces it.
    ///
    /// The returned guard is how it stops being marked: settle it with
    /// [`InFlight::succeeded`] or [`InFlight::failed`], or drop it and it reports
    /// a failure. Dropping without settling means the task was cancelled or
    /// panicked, and a row stuck on "importing" for ever is a worse answer.
    pub fn begin(self: &Arc<Self>, setup: &str, session: &str, operation: Operation) -> InFlight {
        let key = (setup.to_string(), session.to_string());
        self.running.lock().unwrap().insert(key.clone(), operation);
        self.failures.lock().unwrap().remove(&key);
        let _ = self.changes.send(Change::Started {
            setup: key.0.clone(),
            session: key.1.clone(),
            operation,
        });
        InFlight {
            registry: Arc::clone(self),
            key,
            settled: false,
        }
    }

    /// What is happening to this session, if anything is.
    pub fn running(&self, setup: &str, session: &str) -> Option<Operation> {
        let key = (setup.to_string(), session.to_string());
        self.running.lock().unwrap().get(&key).copied()
    }

    /// How the last operation on this session failed, if it did.
    pub fn failure(&self, setup: &str, session: &str) -> Option<String> {
        let key = (setup.to_string(), session.to_string());
        self.failures.lock().unwrap().get(&key).cloned()
    }

    /// Forgets failures against sessions the machine no longer has. Called from
    /// the listing, which is the only place that knows what is still there.
    pub fn prune(&self, setup: &str, present: &[String]) {
        self.failures
            .lock()
            .unwrap()
            .retain(|(kept_setup, session), _| {
                kept_setup != setup || present.iter().any(|id| id == session)
            });
    }

    /// Every change as it happens. See the module note on why this is not the
    /// only way the phone finds out.
    pub fn subscribe(&self) -> broadcast::Receiver<Change> {
        self.changes.subscribe()
    }
}

/// An operation that is running, and its way back out of the registry.
pub struct InFlight {
    registry: Arc<Registry>,
    key: (String, String),
    settled: bool,
}

impl InFlight {
    pub fn succeeded(mut self) {
        self.settle(None);
    }

    pub fn failed(mut self, message: String) {
        self.settle(Some(message));
    }

    fn settle(&mut self, failure: Option<String>) {
        if self.settled {
            return;
        }
        self.settled = true;
        self.registry.running.lock().unwrap().remove(&self.key);
        let (setup, session) = (self.key.0.clone(), self.key.1.clone());
        let change = match failure {
            Some(message) => {
                self.registry
                    .failures
                    .lock()
                    .unwrap()
                    .insert(self.key.clone(), message.clone());
                Change::Failed {
                    setup,
                    session,
                    message,
                }
            }
            None => Change::Finished { setup, session },
        };
        let _ = self.registry.changes.send(change);
    }
}

impl Drop for InFlight {
    fn drop(&mut self) {
        self.settle(Some("the server stopped before it finished".to_string()));
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn an_operation_is_visible_while_it_runs_and_gone_after() {
        let registry = Arc::new(Registry::default());
        let mut changes = registry.subscribe();

        let running = registry.begin("local", "abc", Operation::Importing);
        assert_eq!(registry.running("local", "abc"), Some(Operation::Importing));
        assert!(matches!(changes.try_recv(), Ok(Change::Started { .. })));

        running.succeeded();
        assert_eq!(registry.running("local", "abc"), None);
        assert_eq!(registry.failure("local", "abc"), None);
        assert!(matches!(changes.try_recv(), Ok(Change::Finished { .. })));
    }

    /// A failure outlives the operation, because the phone that needs it may not
    /// have been listening when it happened.
    #[test]
    fn a_failure_is_kept_until_something_replaces_or_prunes_it() {
        let registry = Arc::new(Registry::default());

        registry
            .begin("local", "abc", Operation::Deleting)
            .failed("no such session".to_string());
        assert_eq!(registry.running("local", "abc"), None);
        assert_eq!(
            registry.failure("local", "abc").as_deref(),
            Some("no such session")
        );

        // Still on the machine, so the failure is still about something.
        registry.prune("local", &["abc".to_string()]);
        assert!(registry.failure("local", "abc").is_some());

        // Another machine's listing says nothing about this one's.
        registry.prune("other", &[]);
        assert!(registry.failure("local", "abc").is_some());

        registry.prune("local", &[]);
        assert!(registry.failure("local", "abc").is_none());
    }

    /// Trying again clears the last failure, so a row cannot show an error from
    /// before the attempt somebody is currently watching.
    #[test]
    fn starting_again_clears_the_previous_failure() {
        let registry = Arc::new(Registry::default());
        registry
            .begin("local", "abc", Operation::Importing)
            .failed("first go".to_string());

        let second = registry.begin("local", "abc", Operation::Importing);
        assert_eq!(registry.failure("local", "abc"), None);
        second.succeeded();
    }

    /// A task that is cancelled or panics must not leave a row saying something
    /// is still happening to it.
    #[test]
    fn dropping_an_unsettled_operation_reports_a_failure() {
        let registry = Arc::new(Registry::default());
        drop(registry.begin("local", "abc", Operation::Deleting));
        assert_eq!(registry.running("local", "abc"), None);
        assert!(registry.failure("local", "abc").is_some());
    }
}