fix(cli-rs): find a crashed service between event stream attempts

The event stream found a restarted service inside its 2 s connect timeout, which cancelled ensure before the 5 s spawn delay let it start a replacement, so a crashed managed service was never replaced. The transport now exposes reconnect, Follow implements it, and the store's event source runs it between attempts as TS createClientConnection does. The stream's own request no longer resolves the service inside the timeout.

packages/cli-rs/Cargo.lock

@@ -1167,8 +1167,11 @@ dependencies = [
  "indexmap",
  "log",
  "opencode-api",
+ "opencode-local",
+ "opencode-service",
  "serde",
  "serde_json",
+ "tempfile",
  "tokio",
  "tokio-util",
 ]

packages/cli-rs/crates/api/src/transport.rs

@@ -77,6 +77,13 @@ pub trait Transport: Send + Sync {
     fn endpoint(&self, url: url::Url, auth: Option<&Auth>) -> (url::Url, Option<Auth>) {
         (self.route(url), auth.cloned())
     }
+
+    /// Finds the server again after the event stream disconnected (the TS `service.reconnect`). Returns true when
+    /// a server was resolved. The connection loop runs this between stream attempts, outside the connect timeout,
+    /// so a managed service has time to start a replacement.
+    fn reconnect(&self) -> BoxFuture<'_, bool> {
+        Box::pin(async { false })
+    }
 }
 
 /// The checked response of a server-sent events endpoint: status and content type are verified, and

packages/cli-rs/crates/service/src/follow.rs

@@ -7,7 +7,8 @@
 //! origin the client was built for go to the current service with its current password. A request
 //! that fails to connect or is refused with 401 finds the service again, through
 //! [`connect::reconnect`], before it returns its error. The request is not retried, so a request
-//! that may have reached the old service is never sent twice; the event stream retries itself.
+//! that may have reached the old service is never sent twice. The event stream retries itself and
+//! finds the service through [`Transport::reconnect`], between attempts.
 
 use std::sync::{Arc, RwLock};
 
@@ -67,16 +68,26 @@ impl Transport for Follow {
             if request.url.origin() != self.origin {
                 return self.inner.send(request).await;
             }
+            // The event stream opens under the connection loop's connect timeout, which would cancel a
+            // service start; the loop finds the service through `Transport::reconnect` instead.
+            let stream = request.url.path() == "/api/event";
             let generation = self.reroute(&mut request);
             let result = self.inner.send(request).await;
             // A transport error or 401.
-            if !result.as_ref().is_ok_and(|response| response.status != 401) {
-                self.reconnect(generation).await;
+            if !stream && !result.as_ref().is_ok_and(|response| response.status != 401) {
+                self.resolve(generation).await;
             }
             result
         })
     }
 
+    fn reconnect(&self) -> BoxFuture<'_, bool> {
+        Box::pin(async move {
+            let generation = self.route.read().expect("route lock").generation;
+            self.resolve(generation).await
+        })
+    }
+
     fn route(&self, url: url::Url) -> url::Url {
         self.endpoint(url, None).0
     }
@@ -122,26 +133,27 @@ impl Follow {
         route.generation
     }
 
-    // Finds the service again, unless another request did since `generation`.
-    async fn reconnect(&self, generation: u64) {
+    // Finds the service again, unless another request did since `generation`. True when it is found.
+    async fn resolve(&self, generation: u64) -> bool {
         let _resolving = self.resolving.lock().await;
         if self.route.read().expect("route lock").generation != generation {
-            return;
+            return true;
         }
         let found = connect::reconnect(self.inner.clone(), &self.paths, &self.env).await;
         let endpoint = match found {
             Ok(found) => found.service.endpoint,
             Err(error) => {
                 log::warn!("service.reconnect failed: {error}");
-                return;
+                return false;
             }
         };
         let Ok(url) = url::Url::parse(&endpoint.url) else {
             log::warn!("service.reconnect found an invalid url: {}", endpoint.url);
-            return;
+            return false;
         };
         log::info!("service.reconnect url={url}");
         *self.route.write().expect("route lock") = Route { url, auth: endpoint.auth, generation: generation + 1 };
+        true
     }
 }
 

packages/cli-rs/crates/store/Cargo.toml

@@ -20,5 +20,8 @@ tokio-util.workspace = true
 
 [dev-dependencies]
 bytes.workspace = true
+opencode-local.workspace = true
+opencode-service.workspace = true
 serde.workspace = true
+tempfile.workspace = true
 tokio = { workspace = true, features = ["test-util"] }

packages/cli-rs/crates/store/src/connection.rs

@@ -8,6 +8,7 @@ use futures_util::stream::BoxStream;
 use opencode_api::client::Client;
 use opencode_api::events::{self, Activity, Connection, Source, Status};
 use opencode_api::generated::*;
+use opencode_api::transport::BoxFuture;
 use tokio::sync::{mpsc, watch};
 use tokio::task::JoinHandle;
 
@@ -381,6 +382,10 @@ impl Source<V2Event> for Events {
         })
         .boxed()
     }
+
+    fn reconnect(&mut self) -> BoxFuture<'_, bool> {
+        self.client.transport().reconnect()
+    }
 }
 
 /// Connects the store to its client's event stream on the store's runtime, so the UI thread can

packages/cli-rs/crates/store/tests/reconnect.rs

@@ -0,0 +1,141 @@
+//! The event stream of a managed client recovers when its service crashes (review carbon-1). The crashed
+//! service leaves its registration behind, so `ensure` waits the spawn delay (5 s) before it starts a
+//! replacement: longer than the connect timeout (2 s) a stream attempt may take. The connection loop
+//! must find the service between attempts, as TS `createClientConnection` runs `reconnect`.
+
+use std::ffi::OsString;
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use bytes::Bytes;
+use futures_util::StreamExt;
+use opencode_api::events::{self, Status};
+use opencode_api::transport::{BoxFuture, Request, Response, Transport};
+use opencode_local::paths::Paths;
+use opencode_local::registration;
+use opencode_service::connect::{self, Env};
+use opencode_service::follow;
+use opencode_service::ts_command::write_script;
+use opencode_store::connection::Events;
+use tokio::sync::watch;
+
+const OLD: u16 = 4096;
+const NEW: u16 = 5000;
+
+/// The old service on port 4096 until it dies, and its replacement on port 5000 once the contender
+/// registered it.
+struct Services {
+    registration: PathBuf,
+    alive: watch::Sender<bool>,
+    streams: Mutex<Vec<u16>>,
+}
+
+impl Transport for Services {
+    fn send(&self, request: Request) -> BoxFuture<'_, Result<Response, String>> {
+        let port = request.url.port().unwrap_or_default();
+        let path = request.url.path().to_owned();
+        Box::pin(async move {
+            let pid = match port {
+                OLD if *self.alive.borrow() => 9,
+                NEW => match registration::read(&self.registration) {
+                    Some(info) if info.url.ends_with(":5000") => info.pid,
+                    _ => return Err("connection refused".to_owned()),
+                },
+                _ => return Err("connection refused".to_owned()),
+            };
+            match path.as_str() {
+                "/api/info" => Ok(reply(
+                    200,
+                    "application/json",
+                    format!(r#"{{"version":"2.0.0","pid":{pid},"urls":[],"paths":{{"tmp":"/tmp"}}}}"#),
+                    futures_util::stream::empty().boxed(),
+                )),
+                "/api/event" => {
+                    self.streams.lock().unwrap().push(port);
+                    let mut alive = self.alive.subscribe();
+                    let tail = match port {
+                        OLD => futures_util::stream::once(async move {
+                            let _ = alive.wait_for(|alive| !alive).await;
+                            Err("connection reset".to_owned())
+                        })
+                        .boxed(),
+                        _ => futures_util::stream::pending().boxed(),
+                    };
+                    Ok(reply(
+                        200,
+                        "text/event-stream",
+                        "data: {\"id\":\"evt_1\",\"type\":\"server.connected\",\"data\":{}}\n\n".to_owned(),
+                        tail,
+                    ))
+                }
+                _ => Ok(reply(404, "application/json", "{}".to_owned(), futures_util::stream::empty().boxed())),
+            }
+        })
+    }
+}
+
+fn reply(
+    status: u16,
+    content_type: &str,
+    body: String,
+    tail: futures_util::stream::BoxStream<'static, Result<Bytes, String>>,
+) -> Response {
+    Response {
+        status,
+        headers: vec![("content-type".to_owned(), content_type.to_owned())],
+        body: Box::pin(futures_util::stream::iter([Ok(Bytes::from(body))]).chain(tail)),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn the_event_stream_follows_a_service_that_crashed_and_was_replaced() {
+    let dir = tempfile::tempdir().unwrap();
+    let paths = Paths::from_vars(|name| match name {
+        "HOME" => Some(OsString::from(dir.path())),
+        "XDG_STATE_HOME" => Some(dir.path().join("state").into()),
+        "XDG_CONFIG_HOME" => Some(dir.path().join("config").into()),
+        _ => None,
+    })
+    .unwrap();
+    std::fs::create_dir_all(&paths.state).unwrap();
+    let file = paths.state.join("service-oc-x.json");
+    std::fs::write(&file, format!(r#"{{"url":"http://127.0.0.1:{OLD}","pid":9,"password":"old"}}"#)).unwrap();
+    // The contender registers the replacement under its own pid.
+    let program = dir.path().join("opencode-ts");
+    write_script(
+        &program,
+        &format!(
+            "#!/bin/sh\nprintf '{{\"url\":\"http://127.0.0.1:{NEW}\",\"pid\":%s,\"password\":\"new\"}}' $$ > {0}.tmp\nmv {0}.tmp {0}\nexec sleep 30\n",
+            file.display()
+        ),
+    );
+    let services = Arc::new(Services {
+        registration: file.clone(),
+        alive: watch::channel(true).0,
+        streams: Mutex::new(Vec::new()),
+    });
+    let env = Env {
+        ts_bin: Some(program.to_string_lossy().into_owned()),
+        channel: Some("oc-x".to_owned()),
+        ..Env::default()
+    };
+    let connection = connect::resolve(services.clone(), &paths, &env, &connect::Options::default()).await.unwrap();
+    let client = follow::client(services.clone(), &connection, &paths, &env).unwrap();
+    let (connection, _batches) = events::connect(Events { client }, events::Options::default());
+    let mut state = connection.watch();
+    state.wait_for(|state| state.status == Status::Connected).await.unwrap();
+
+    services.alive.send_replace(false);
+    state.wait_for(|state| state.status != Status::Connected).await.unwrap();
+    let started = Instant::now();
+    let reconnected =
+        tokio::time::timeout(Duration::from_secs(15), state.wait_for(|state| state.status == Status::Connected)).await;
+    let replacement = registration::read(&file).filter(|info| info.pid != 9);
+    if let Some(info) = &replacement {
+        let _ = std::process::Command::new("kill").arg(info.pid.to_string()).status();
+    }
+    assert!(reconnected.is_ok(), "no reconnect after {:?}: {:?}", started.elapsed(), connection.history());
+    assert!(replacement.is_some(), "no replacement service was started");
+    assert_eq!(services.streams.lock().unwrap().last(), Some(&NEW));
+}