Run imports and deletes on the server, and say so on an event

Leaving the import screen used to cancel the batch it had started: the
request was the work, so the coroutine that owned it died with the screen
and coming back showed no sign anything had happened. A half-imported
session is the expensive kind of missing -- the row is back looking
untouched, and taking it again is the second `--resume` the import path
exists to prevent.

So the work runs on the server now. Delete and a new per-session import both
answer 202 and spawn the work, and `session::pending` is the record of it:
what is running, and how the last attempt failed. The phone reads that two
ways and needs both. Every row of the listing carries `pending` and `error`,
which is what a phone that was asleep, out of range or freshly opened has to
go on; `GET /setups/{id}/importable/events` streams the changes, which is
what makes a screen somebody is watching change by itself.

Neither alone is enough, and that is not theoretical. A broadcast has no
memory, so an operation that started and finished while the stream was still
connecting was one nothing would ever be said about -- with responses held
back far enough to make it visible, one row of a pair of deletes cleared and
the other sat on "waiting" for good. The screen now asks again after a
handover when anything still looks outstanding, and takes its row states
from that answer rather than from what it remembers.

The single tap still waits, because "take me to it" needs the session that
was made and 202 does not carry one. Both paths go through the same `spawn`
so they cannot drift about what importing means.

Resolving one importable session no longer lists every one of them:
`import::find` is the same script with one glob narrower, which takes the
import seed off the 3.7-second full scan that `delete` came off earlier.

The SSE connection and its framing are now `Sse`, shared with the session
transcript stream rather than written a second time.

server/src/session/pending.rs

@@ -0,0 +1,282 @@
+//! 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
+//! whole 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 `GET
+//! /setups/{id}/importable` carries what is happening to it, which is what
+//! a phone that was asleep, out of range, or freshly opened has to go on;
+//! and [`Registry::subscribe`] is the live stream, which is what makes a
+//! screen somebody is looking at change by itself. Neither is sufficient
+//! alone -- 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 that
+/// are easy to leave out: it can still be running, it can have finished,
+/// and it can have failed. There is deliberately no "unknown" -- this is
+/// the server's own work, so not knowing would be a bug rather than a
+/// state.
+#[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 rather than at the
+    /// filter 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 -- an error about a
+    /// transcript that is gone has nothing left to be about.
+    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 than one that says it did not finish.
+    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. A deleted session's failure would otherwise outlive
+    /// everything it referred to.
+    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());
+    }
+}