diff --git a/AGENTS.md b/AGENTS.md index 8f864cd..3b74a36 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -5,14 +5,18 @@ t3term is a Rust terminal client for the T3 Code server. README.md documents the ## Who does what - The user sets scope and merges every pull request. No agent merges a pull request, whether it changes the UI or not. -- GPT-6.1 Sol (xhigh) orchestrates the sprint. It runs all testing and computer use itself, reviews each pull request, prepares it for merge and writes its walkthrough. +- GPT-6.1 Sol (xhigh) orchestrates the sprint. It runs all testing and computer use itself, reviews each pull request and prepares it for merge. - Claude Opus 5.5 (xhigh) writes the implementation, the tests and the fixes. It is the only agent that writes code. ## Pull requests - Keep them small and frequent. Open a draft early, push at each checkpoint and mark it ready when the work is done. -- Keep a decision log in the description, in the format of pr-walkthrough's `references/decision-log.md`: what you noticed, how you confirmed it, the options you chose and rejected, what review changed and the known limits. -- Every pull request gets a short walkthrough made with the pr-walkthrough skill (https://github.com/krishhgg/pr-walkthrough) that teaches the Rust it uses. The pages stay on the reader's machine, so never commit them. +- Keep the description short and lead with what the pull request changes. End it with a decision log under these five headings: + - What I noticed: the problem or gap that started the work. + - How I confirmed it: the code, contract or run that showed it. + - Options: the approach chosen and the ones rejected, with why. + - What review changed: what Greptile, Sol or the user asked for and what changed as a result. + - Known limits: what the pull request leaves out or doesn't handle. - Changes to how the TUI looks carry screenshots or recordings. Keep UI and non-UI changes in separate pull requests where practical. - Greptile reviews every pull request. A pull request is ready for the user when CI passes and every Greptile finding is fixed or answered. diff --git a/README.md b/README.md index 69efb14..0023a0b 100644 --- a/README.md +++ b/README.md @@ -108,6 +108,8 @@ A menu choice turns its chip blue and goes to T3 with the thread's next message, ![Choosing a model, effort and mode in the TUI, then sending](docs/screenshots/picker.gif) +The chips and menus follow T3's provider list as it changes. The TUI subscribes to T3's server config, so when T3 adds or drops a provider, renames a model or gives a provider a status such as `error`, the chips and any open menu show it without reading the list again. The model menu shows a status after the provider's name. An open menu keeps its highlight on the same model as rows move. When a draft's provider goes away, its chip shows the thread's model, and the draft keeps the choice until the provider comes back. If T3 refuses the subscription, each menu reads the list with `server.getConfig` when it opens. + Ultrathink applies to that one message and comes back only if that message fails to send. T3 refuses a mode change while a run is active, so the TUI leaves the message in the composer to send once the run ends. That includes the switch back to Build below. A message that fails after you have typed something else or opened another thread is kept: the status line says so, Ctrl+R swaps it with the composer's text, and opening its thread with an empty composer brings it back. A send that times out can still reach T3, so if the thread later shows it, the TUI drops its kept copy rather than send it twice. ### Plan mode @@ -306,6 +308,7 @@ The choices on `send` and `settings` are `--model`, `--effort`, `--option ID=VAL - **Auth.** On macOS, t3term keeps one login per T3 server in the login Keychain under the service `t3term`. It lasts 30 days, the same as T3's own default, and holds only `orchestration:read` and `orchestration:operate`. Each run checks it against `/api/auth/session`, and if it expires within a day or the server rejects it, t3term issues a new one and revokes the old one, so you never log in by hand. `t3term logout` revokes and deletes it. - **RPC.** Effect RPC over one `/ws` connection. The client acks every stream chunk and pings every 10 seconds. If no frame arrives for 30 seconds it treats the socket as dead. - **State.** Each V2 event carries the whole updated entity, so the reducer upserts it by id. After a dropped connection the client resubscribes with `afterSequence` and skips any replayed event it already applied. +- **Server config.** The TUI subscribes to `subscribeServerConfig` with an empty payload. Each subscription opens with a snapshot of the whole config and goes on with changes to the provider list, the settings or the keybindings. The stream has no cursor, so after a dropped connection the TUI keeps the config it has until the new subscription's snapshot replaces it. Retries wait 250 ms at first, doubling up to 10 seconds, and a snapshot brings the wait back to 250 ms. A server that refuses the subscription, because it lacks the method or the session lacks `orchestration:read`, gets a `server.getConfig` read each time a menu opens. The CLI reads `server.getConfig` when a command needs the model list. - **Rendering.** The TUI draws only after input or a server event, at most 30 times a second. The open thread keeps its transcript blocks between frames and describes an item again only after T3 changes it, so a frame drawn for a key or the clock describes nothing. Each block keeps its wrapped lines until it changes or the width does, and only the visible rows are copied into a frame. While the open thread has a run going or a card on screen reads Working or Goal, a once-a-second tick advances the spinners and clocks, and that tick stops when none does. While a thread is snoozed, one timer waits for the soonest snooze to end, because no server event marks that moment. The same redraw moves the card back and shows Woke. Otherwise an idle TUI wakes only for input or server events.
@@ -323,7 +326,7 @@ Set `T3TERM_NO_SAVED_LOGIN=1` to use a session that lasts one run instead. t3ter cargo test ``` -The unit tests cover the reducers, Markdown wrapping, the composer, the model menus, the transcript and the lines under a file change, the tasks drawer, the thread error banner, hiding and resizing the sidebar and auth command parsing. The TUI tests draw frames on ratatui's test backend with a client that never connects, and a test build keeps saved settings in memory, so no test writes `~/.config/t3term/settings.json`. `tests/fake_server.rs` drives the real RPC client against a fake Effect RPC server, dropping the socket mid-stream to check the resume cursor, chunk acks, batched frames, duplicate suppression and error decoding. `tests/cli.rs` runs the built binary with a temporary home and checks the `--json` error and exit code when the server is gone, when it isn't on protocol 2 and when the TUI has no terminal. It also runs `t3term read` against a fake server and a fake `t3` that issues a made-up session. It checks that the plain output drops the control characters of a checklist, a compaction summary and an edit's file name, and that `--json` writes them as escapes that decode to the text T3 sent. Each edit prints as one row, without its operations, error or patch. Another test runs `t3term send --wait` against a fake server that also plays T3's side of the WebSocket. It checks that each finished handoff in the turn prints its marker and endpoints on stderr once, even when T3 sends the item again, and that `--json` prints only the result. A test of `t3term wait` checks that each edit in the turn prints its row on stderr once, when it finishes, however often T3 sends it again. None of them reach a real T3 server or the Keychain, and CI runs them on macOS for every pull request. +The unit tests cover the reducers, Markdown wrapping, the composer, the model menus, the transcript and the lines under a file change, the tasks drawer, the thread error banner, hiding and resizing the sidebar and auth command parsing. The TUI tests draw frames on ratatui's test backend with a client that never connects, and a test build keeps saved settings in memory, so no test writes `~/.config/t3term/settings.json`. `tests/fake_server.rs` drives the real RPC client against a fake Effect RPC server, dropping the socket mid-stream to check the resume cursor, chunk acks, batched frames, duplicate suppression and error decoding. Its config tests check the subscription's exact request, a fresh snapshot with no cursor after each drop, the backoff, stopping when the TUI goes while a watch connects or waits, and both ways T3 refuses the subscription. The TUI's config tests run on a runtime that never polls its tasks, so they can count each model-list read the TUI starts and show which changes reach the chips and menus without one. `tests/cli.rs` runs the built binary with a temporary home and checks the `--json` error and exit code when the server is gone, when it isn't on protocol 2 and when the TUI has no terminal. It also runs `t3term read` against a fake server and a fake `t3` that issues a made-up session. It checks that the plain output drops the control characters of a checklist, a compaction summary and an edit's file name, and that `--json` writes them as escapes that decode to the text T3 sent. Each edit prints as one row, without its operations, error or patch. Another test runs `t3term send --wait` against a fake server that also plays T3's side of the WebSocket. It checks that each finished handoff in the turn prints its marker and endpoints on stderr once, even when T3 sends the item again, and that `--json` prints only the result. A test of `t3term wait` checks that each edit in the turn prints its row on stderr once, when it finishes, however often T3 sends it again. None of them reach a real T3 server or the Keychain, and CI runs them on macOS for every pull request. `docs/screenshots/` also holds `before-tui.png`, `after-tui.png`, `approval.png`, `streaming.gif`, `picker-model.png` and `tool-calls-failed.png`. diff --git a/docs/sprint.md b/docs/sprint.md index 345f1f2..bc1b067 100644 --- a/docs/sprint.md +++ b/docs/sprint.md @@ -34,6 +34,7 @@ Rows are tracked by URL, because GitHub numbers issues and pull requests from on | 12 | https://github.com/krishhgg/t3term/pull/19 | Resize the main sidebar by dragging its edge or with [ and ], remembered between runs | | | 13 | https://github.com/krishhgg/t3term/pull/20 | File changes in tool output: each edit's operations and a failed edit's error, with a bounded patch preview when a server sends one | | | 14 | https://github.com/krishhgg/t3term/pull/21 | A dismissible banner over the conversation for the thread's error, from a failed send or from T3 | | +| 15 | https://github.com/krishhgg/t3term/pull/22 | Live server config in the TUI, so provider and model changes reach the chips and menus without a reread | | ## Maintenance review PRs diff --git a/src/client.rs b/src/client.rs index 8e0cd09..584b54d 100644 --- a/src/client.rs +++ b/src/client.rs @@ -439,6 +439,15 @@ impl Client { after_sequence, ) } + + /// Streams the server's configuration, as `spawn_config_watch` describes. + pub fn watch_config(self: &Arc) -> mpsc::UnboundedReceiver { + let client = self.clone(); + spawn_config_watch(move || { + let client = client.clone(); + Box::pin(async move { client.rpc().await }) + }) + } } /// T3 can restart the agent's session to apply these settings, so t3term changes them only @@ -457,7 +466,8 @@ fn busy_for_settings(run: &Value) -> anyhow::Error { #[derive(Debug, Clone)] pub enum WatchEvent { Item(Value), - /// The connection dropped; the watcher is reconnecting and will resume after the last sequence. + /// The connection dropped. The watcher is reconnecting, and resumes after the last sequence + /// or starts again from a snapshot, as its stream does. Reconnecting { reason: String, retry_in: Duration, @@ -466,6 +476,33 @@ pub enum WatchEvent { Failed(String), } +/// Opens a connection for a watcher. Each reconnect calls it again. +type Connect = dyn Fn() -> Pin> + Send>> + Send + Sync; + +/// Where a watcher's stream starts again after a dropped connection. +#[derive(Clone, Copy)] +enum Resume { + /// After the highest event sequence delivered, from this one at first. The orchestration + /// streams work this way, and their `{"kind": "synchronized"}` marker says a subscription + /// has caught up. + AfterSequence(Option), + /// From a new snapshot. A stream with no cursor, such as the server config, sends the whole + /// state first, as `{"type": "snapshot"}`. + Snapshot, +} + +impl Resume { + /// Whether `item` shows the subscription working, so the next drop retries quickly. + fn healthy(self, item: &Value) -> bool { + match self { + Resume::AfterSequence(_) => { + item.get("kind").and_then(Value::as_str) == Some("synchronized") + } + Resume::Snapshot => item.get("type").and_then(Value::as_str) == Some("snapshot"), + } + } +} + /// Runs a subscription until its receiver is dropped. Tracks the highest sequence it delivered so a /// resubscribe replays only what the consumer has not seen. pub fn spawn_watch( @@ -477,12 +514,50 @@ pub fn spawn_watch( where F: Fn() -> Pin> + Send>> + Send + Sync + 'static, { + watch( + Box::new(connect), + tag, + payload, + Resume::AfterSequence(after_sequence), + ) +} + +/// Runs T3's config subscription, `subscribeServerConfig`, until its receiver is dropped. Each +/// subscription opens with a snapshot of the whole configuration and goes on with changes to +/// parts of it. The stream has no cursor, so a resubscribe asks for nothing but a new snapshot. +/// The payload is empty, which leaves out the theme and usage-limit events t3term doesn't read. +pub fn spawn_config_watch(connect: F) -> mpsc::UnboundedReceiver +where + F: Fn() -> Pin> + Send>> + Send + Sync + 'static, +{ + watch( + Box::new(connect), + "subscribeServerConfig", + json!({}), + Resume::Snapshot, + ) +} + +fn watch( + connect: Box, + tag: &'static str, + payload: Value, + resume: Resume, +) -> mpsc::UnboundedReceiver { let (events, receiver) = mpsc::unbounded_channel(); tokio::spawn(async move { - let mut last_sequence = after_sequence; + let mut last_sequence = match resume { + Resume::AfterSequence(sequence) => sequence, + Resume::Snapshot => None, + }; let mut backoff = Duration::from_millis(250); loop { - let reason = match connect().await { + let connected = tokio::select! { + connected = connect() => connected, + // The consumer went away, such as a TUI that quit. + _ = events.closed() => return, + }; + let reason = match connected { Err(e) => e.to_string(), Ok(rpc) => { let mut request = payload.clone(); @@ -499,26 +574,12 @@ where }; match next { Some(Ok(item)) => { - if item.get("kind").and_then(Value::as_str) - == Some("synchronized") - { + if resume.healthy(&item) { backoff = Duration::from_millis(250); } - // Enrichment snapshots do not move the resume cursor. - let enrichment = - item.get("resolvedRepositoryIdentityRoots").is_some(); - let sequence = if enrichment { - None - } else { - item.get("sequence") - .or_else(|| item.get("snapshotSequence")) - .or_else(|| { - item.get("snapshot") - .and_then(|s| s.get("snapshotSequence")) - }) - .and_then(Value::as_u64) - }; - if let Some(sequence) = sequence { + if let Resume::AfterSequence(_) = resume + && let Some(sequence) = delivered_sequence(&item) + { last_sequence = Some( last_sequence.map_or(sequence, |l| l.max(sequence)), ); @@ -547,9 +608,24 @@ where { return; } - tokio::time::sleep(backoff).await; + tokio::select! { + _ = tokio::time::sleep(backoff) => {} + _ = events.closed() => return, + } backoff = (backoff * 2).min(MAX_BACKOFF); } }); receiver } + +/// The event sequence an orchestration item carries, which a resubscribe resumes after. +fn delivered_sequence(item: &Value) -> Option { + // Enrichment snapshots do not move the resume cursor. + if item.get("resolvedRepositoryIdentityRoots").is_some() { + return None; + } + item.get("sequence") + .or_else(|| item.get("snapshotSequence")) + .or_else(|| item.get("snapshot").and_then(|s| s.get("snapshotSequence"))) + .and_then(Value::as_u64) +} diff --git a/src/projection.rs b/src/projection.rs index 9ba0299..999b6f0 100644 --- a/src/projection.rs +++ b/src/projection.rs @@ -468,6 +468,51 @@ impl ShellState { } } +/// Applies one item from T3's config subscription, `subscribeServerConfig`, to `config`, as T3's +/// own client does in packages/client-runtime/src/state/serverConfigProjection.ts. +/// +/// A snapshot replaces the whole configuration, so a provider it leaves out is gone. T3 sends +/// one when a subscription opens and again whenever the configuration changes as a whole. +/// `providerStatuses` replaces the whole provider list, never one provider. `settingsUpdated` +/// replaces the settings, and `keybindingsUpdated` the keybindings and their issues. Each +/// leaves every other field as it was, fields this build doesn't know included. +/// +/// A change that comes before any configuration has nothing to apply to, and is ignored. So is +/// an item of another version or type, such as the theme and usage-limit events t3term doesn't +/// ask for, and one that lacks the part it changes. None of them touches the configuration held. +pub fn apply_config(config: &mut Option, item: &Value) -> Applied { + if item.get("version").and_then(Value::as_u64) != Some(1) { + return Applied::Ignored; + } + let kind = item.get("type").and_then(Value::as_str).unwrap_or_default(); + if kind == "snapshot" { + let Some(snapshot) = item.get("config").filter(|c| c["providers"].is_array()) else { + return Applied::Ignored; + }; + *config = Some(snapshot.clone()); + return Applied::Snapshot; + } + let Some(held) = config.as_mut().and_then(Value::as_object_mut) else { + return Applied::Ignored; + }; + let payload = &item["payload"]; + let keybindings = payload["keybindings"].is_array() && payload["issues"].is_array(); + match kind { + "providerStatuses" if payload["providers"].is_array() => { + held.insert("providers".into(), payload["providers"].clone()); + } + "settingsUpdated" if payload["settings"].is_object() => { + held.insert("settings".into(), payload["settings"].clone()); + } + "keybindingsUpdated" if keybindings => { + held.insert("keybindings".into(), payload["keybindings"].clone()); + held.insert("issues".into(), payload["issues"].clone()); + } + _ => return Applied::Ignored, + } + Applied::Event(kind.to_string()) +} + #[cfg(test)] mod tests { use super::*; @@ -728,4 +773,142 @@ mod tests { assert_eq!(shell.sequence, 5); assert_eq!(shell.projects[0]["repositoryIdentity"]["name"], "r"); } + + /// A snapshot from T3's config subscription. + fn config_snapshot(config: Value) -> Value { + json!({"version": 1, "type": "snapshot", "config": config}) + } + + /// A change from T3's config subscription. + fn config_change(kind: &str, payload: Value) -> Value { + json!({"version": 1, "type": kind, "payload": payload}) + } + + fn provider(id: &str, status: &str) -> Value { + let model = format!("{id}-model"); + json!({"instanceId": id, "status": status, "enabled": true, "models": [{"slug": model}]}) + } + + /// A configuration with every field the TUI reads and some it doesn't, including one no + /// build knows yet. + fn full_config(providers: Value) -> Value { + json!({ + "environment": {"capabilities": {"threadSnooze": true, "threadSettlement": true}}, + "providers": providers, + "settings": {"enableAssistantStreaming": true, "providers": {"codex": {}}}, + "keybindings": [{"key": "mod+k", "command": "commandPalette.toggle"}], + "issues": [], + "availableEditors": ["cursor"], + "futureField": {"kept": true}, + }) + } + + #[test] + fn a_config_snapshot_replaces_the_whole_config_and_drops_providers_it_leaves_out() { + let mut config = None; + let providers = json!([provider("codex", "ready"), provider("claude", "ready")]); + let first = full_config(providers); + assert_eq!( + apply_config(&mut config, &config_snapshot(first.clone())), + Applied::Snapshot + ); + assert_eq!(config.as_ref(), Some(&first)); + + // A later snapshot is the whole configuration, so what it leaves out is gone, fields + // and providers alike. + let providers = json!([provider("codex", "error")]); + let second = json!({"environment": {"capabilities": {}}, "providers": providers}); + assert_eq!( + apply_config(&mut config, &config_snapshot(second.clone())), + Applied::Snapshot + ); + assert_eq!(config.as_ref(), Some(&second)); + } + + #[test] + fn config_changes_replace_their_own_fields_and_keep_the_rest() { + let providers = json!([provider("codex", "ready"), provider("claude", "ready")]); + let before = full_config(providers); + let mut config = None; + apply_config(&mut config, &config_snapshot(before.clone())); + + // The new list is the whole list: Codex is gone though the change doesn't name it. + let providers = json!([provider("claude", "error"), provider("cursor", "ready")]); + let change = config_change("providerStatuses", json!({"providers": providers})); + assert_eq!( + apply_config(&mut config, &change), + Applied::Event("providerStatuses".into()) + ); + let mut expected = before.clone(); + expected["providers"] = providers; + assert_eq!(config.as_ref(), Some(&expected)); + + let settings = json!({"enableAssistantStreaming": false}); + let change = config_change("settingsUpdated", json!({"settings": settings})); + assert_eq!( + apply_config(&mut config, &change), + Applied::Event("settingsUpdated".into()) + ); + expected["settings"] = settings; + assert_eq!(config.as_ref(), Some(&expected)); + + let keybindings = json!([]); + let issues = json!([{"kind": "keybindings.malformed-config", "message": "bad"}]); + let payload = json!({"keybindings": keybindings, "issues": issues}); + let change = config_change("keybindingsUpdated", payload); + assert_eq!( + apply_config(&mut config, &change), + Applied::Event("keybindingsUpdated".into()) + ); + expected["keybindings"] = keybindings; + expected["issues"] = issues; + assert_eq!(config.as_ref(), Some(&expected)); + } + + #[test] + fn config_items_that_change_nothing_leave_the_config_as_it_was() { + // A change before any snapshot has no configuration to apply to. + let mut config = None; + let providers = json!([provider("codex", "ready")]); + let early = config_change("providerStatuses", json!({"providers": providers})); + assert_eq!(apply_config(&mut config, &early), Applied::Ignored); + assert_eq!(config, None); + let malformed = config_snapshot(json!({"environment": {}})); + assert_eq!(apply_config(&mut config, &malformed), Applied::Ignored); + assert_eq!(config, None); + + let good = full_config(providers); + apply_config(&mut config, &config_snapshot(good.clone())); + for item in [ + // Another version, or none. + json!({"version": 2, "type": "snapshot", "config": {"providers": []}}), + json!({"type": "providerStatuses", "payload": {"providers": []}}), + // Snapshots without a provider list. + config_snapshot(json!({"environment": {}})), + config_snapshot(json!({"providers": {"codex": {}}})), + config_snapshot(Value::Null), + json!({"version": 1, "type": "snapshot"}), + // Changes without the part they change. + config_change("providerStatuses", json!({})), + config_change("providerStatuses", json!({"providers": null})), + json!({"version": 1, "type": "providerStatuses"}), + config_change("settingsUpdated", json!({"settings": []})), + config_change("keybindingsUpdated", json!({"keybindings": []})), + config_change( + "keybindingsUpdated", + json!({"keybindings": {}, "issues": []}), + ), + // Events t3term doesn't ask for, or that no build knows yet. + config_change("environmentThemesUpdated", json!({"themes": []})), + config_change("usageLimitSourcesUpdated", json!({"sources": []})), + config_change("somethingNew", json!({"providers": []})), + // Not a config item at all. + json!({"kind": "synchronized"}), + json!("snapshot"), + ] { + let applied = apply_config(&mut config, &item); + assert_eq!(applied, Applied::Ignored, "{item}"); + assert_eq!(config.as_ref(), Some(&good), "{item}"); + } + } } diff --git a/src/tui/mod.rs b/src/tui/mod.rs index 4538ac8..2831e4f 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -37,7 +37,7 @@ use unicode_width::UnicodeWidthStr; use crate::client::{Client, IfBusy, WatchEvent}; use crate::models::Choice; -use crate::projection::{Applied, ShellState, ThreadState, is_active_status, status}; +use crate::projection::{Applied, ShellState, ThreadState, apply_config, is_active_status, status}; use crate::settings::Settings; use crate::transcript::{self, BlockKind, ChangeLine}; use composer::Composer; @@ -268,8 +268,12 @@ impl Prepared { enum ActionResult { Info(String), Error(String), - /// T3's provider and model list, from `server.getConfig`. - Config(Value), + /// T3's configuration from `server.getConfig`, with the count of changes from the config + /// subscription when the read was asked for. + Config { + config: Value, + revision: u64, + }, /// What one tool printed, for the row that asked. `None` when T3 couldn't be reached. Output { item_id: String, @@ -377,7 +381,15 @@ impl OpenThread { struct App { client: Arc, theme: Theme, + /// T3's configuration, with its providers and models. The config subscription keeps it + /// current, and it stays as it was while the subscription reconnects. config: Option, + /// False once T3 refuses the config subscription. Each menu then reads the configuration + /// with `server.getConfig`, as t3term did before it subscribed. + config_subscribed: bool, + /// How many changes the config subscription has made, so a `server.getConfig` answer asked + /// for before the latest of them can't put older providers back. + config_revision: u64, /// Model, effort and mode choices per thread, sent with the thread's next message. drafts: HashMap, /// Messages that failed to send, until they go out or reach T3 late. @@ -488,9 +500,10 @@ impl Drop for Screen { async fn event_loop(terminal: &mut ratatui::DefaultTerminal, client: Arc) -> Result<()> { let (actions, mut action_results) = mpsc::unbounded_channel(); let mut shell_events = client.watch_shell(None); + // Dropped when the loop ends, which ends the subscription. + let mut config_events = client.watch_config(); let settings = Settings::load().await; let mut app = App::new(client, Theme::detect(), &settings, actions); - app.load_config(); let mut input = EventStream::new(); let mut dirty = true; let mut last_draw = Instant::now() - FRAME; @@ -533,6 +546,10 @@ async fn event_loop(terminal: &mut ratatui::DefaultTerminal, client: Arc app.on_shell_event(event); dirty = true; } + Some(event) = config_events.recv() => { + app.on_config_event(event); + dirty = true; + } Some(event) = thread_event => { app.on_thread_event(event); dirty = true; @@ -649,6 +666,8 @@ impl App { client, theme, config: None, + config_subscribed: true, + config_revision: 0, drafts: HashMap::new(), unsent: Unsent::default(), picker: None, @@ -718,6 +737,25 @@ impl App { } } + /// Applies an item from T3's config subscription. A reconnect keeps the configuration as it + /// was until the new subscription's snapshot replaces it. + fn on_config_event(&mut self, event: WatchEvent) { + match event { + WatchEvent::Item(item) => { + if self.update_config(|config| apply_config(config, &item) != Applied::Ignored) { + self.config_revision += 1; + } + } + WatchEvent::Reconnecting { .. } => {} + // Such as a server without the subscription. Read the configuration once now, and + // again each time a menu opens. + WatchEvent::Failed(_) => { + self.config_subscribed = false; + self.load_config(); + } + } + } + fn on_thread_event(&mut self, event: WatchEvent) { let Some(open) = self.open.as_mut() else { return; @@ -1409,16 +1447,15 @@ impl App { match result { ActionResult::Info(text) => self.message = Some((text, false)), ActionResult::Error(text) => self.message = Some((text, true)), - ActionResult::Config(config) => { - // A refresh can add or drop rows, so the highlight follows its entry, not - // its row number. - let previous = self.selected_pick(); - self.config = Some(config); - if self.picker.is_some() { - self.select_pick(previous); + ActionResult::Config { config, revision } => { + // A change from the subscription since the read was asked for is newer than + // the answer. + if revision == self.config_revision { + self.update_config(|held| { + *held = Some(config); + true + }); } - // Its capabilities decide whether the Snoozed and Settled shelves apply. - self.rebuild_rows(); } ActionResult::Output { item_id, @@ -1535,17 +1572,37 @@ impl App { // ---- model, effort and mode menus ---- - /// Fetches T3's provider list in the background. Menus use the last copy until it arrives. + /// Reads T3's configuration once in the background. Menus use the last copy until it + /// arrives. fn load_config(&self) { let (client, results) = (self.client.clone(), self.actions.clone()); + let revision = self.config_revision; tokio::spawn(async move { let _ = results.send(match client.server_config().await { - Ok(config) => ActionResult::Config(config), + Ok(config) => ActionResult::Config { config, revision }, Err(e) => ActionResult::Error(format!("Couldn't read T3's model list: {e}")), }); }); } + /// Changes the configuration with `change`, which says whether it changed anything, and + /// keeps what depends on it in step. A change can add or drop menu rows, so an open menu's + /// highlight follows its entry, not its row number. The sidebar is laid out again when the + /// change turns snooze or settlement on or off, which decide whether those shelves apply. + fn update_config(&mut self, change: impl FnOnce(&mut Option) -> bool) -> bool { + let (pick, capabilities) = (self.selected_pick(), self.capabilities()); + if !change(&mut self.config) { + return false; + } + if self.picker.is_some() { + self.select_pick(pick); + } + if self.capabilities() != capabilities { + self.rebuild_rows(); + } + true + } + /// The open thread's settings with its draft applied. fn settings_view(&self) -> Option { let open = self.open.as_ref()?; @@ -1579,8 +1636,10 @@ impl App { )); return; } - // Provider status changes while T3 runs, so each menu refreshes the list. - if self.picker.is_none() { + // Provider status changes while T3 runs. The config subscription brings each change, + // so a menu reads the configuration only while the TUI has none or after T3 refused + // the subscription. + if self.picker.is_none() && (self.config.is_none() || !self.config_subscribed) { self.load_config(); } self.focus = Focus::Composer; @@ -6514,4 +6573,349 @@ mod tests { assert!(banner_shows(&mut app)); assert_eq!(banner_work(&app), before); } + + // ---- live server configuration ---- + + /// A runtime that is never driven. A read of T3's configuration that the TUI spawns on it + /// waits without running, so nothing connects and `reads` can count it. + fn idle_runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .build() + .unwrap() + } + + /// How many reads of T3's configuration the TUI has started on `runtime`. Nothing else in + /// these tests spawns a task. + fn reads(runtime: &tokio::runtime::Runtime) -> usize { + runtime.metrics().num_alive_tasks() + } + + /// A provider in T3's configuration with `models`, each a slug and a name. + fn provider(id: &str, name: &str, status: &str, models: &[(&str, &str)]) -> Value { + let models: Vec = models + .iter() + .map(|&(slug, title)| json!({"slug": slug, "name": title})) + .collect(); + json!({"instanceId": id, "displayName": name, "enabled": true, "status": status, + "models": models}) + } + + /// Codex, with GPT-6 and GPT-6 Mini. + fn codex(status: &str) -> Value { + let models = [("gpt-6", "GPT-6"), ("gpt-6-mini", "GPT-6 Mini")]; + provider("codex", "Codex", status, &models) + } + + /// Claude, with Opus 5.5 and Haiku 4.5. + fn claude(status: &str) -> Value { + let models = [ + ("claude-opus-5-5", "Opus 5.5"), + ("claude-haiku-4-5", "Haiku 4.5"), + ]; + provider("claudeAgent", "Claude", status, &models) + } + + /// A snapshot from the config subscription that lists `providers`, on a server with + /// `capabilities`. + fn config_snapshot(providers: Vec, capabilities: Value) -> WatchEvent { + let environment = json!({"capabilities": capabilities}); + let config = json!({"providers": providers, "settings": {}, "environment": environment}); + let item = json!({"version": 1, "type": "snapshot", "config": config}); + WatchEvent::Item(item) + } + + /// A change from the config subscription that lists every provider again. + fn provider_statuses(providers: Vec) -> WatchEvent { + let payload = json!({"providers": providers}); + let item = json!({"version": 1, "type": "providerStatuses", "payload": payload}); + WatchEvent::Item(item) + } + + /// Thread `t`, open and live on Codex's GPT-6. + fn thread_on_codex() -> OpenThread { + let thread = json!({ + "id": "t", + "modelSelection": {"instanceId": "codex", "model": "gpt-6"}, + "runtimeMode": "full-access", + "interactionMode": "default", + }); + let projection = json!({"thread": thread, "turnItems": []}); + let item = json!({"kind": "snapshot", "snapshotSequence": 1, "projection": projection}); + let mut open = OpenThread::new("t".into(), mpsc::unbounded_channel().1); + assert_eq!(open.apply(&item), Applied::Snapshot); + open.connection = "live".into(); + open + } + + /// The model chip's text in a frame 120 columns by 30 rows. + fn model_chip(app: &mut App) -> String { + let buffer = cells(app, 120, 30); + let chip = app.chips.iter().find(|(_, kind)| *kind == Kind::Model); + text_in(&buffer, chip.expect("a model chip").0) + } + + /// The open menu's text in a frame 120 columns by 30 rows. + fn menu_text(app: &mut App) -> String { + let buffer = cells(app, 120, 30); + assert!(app.picker.is_some(), "no menu is open"); + text_in(&buffer, app.picker_area) + } + + #[test] + fn provider_changes_reach_the_chips_and_the_next_menu_without_a_read() { + let runtime = idle_runtime(); + let _entered = runtime.enter(); + let mut app = tui(&Settings::default()); + app.open_thread(thread_on_codex()); + let snapshot = config_snapshot(vec![codex("ready"), claude("ready")], json!({})); + app.on_config_event(snapshot); + let chip = model_chip(&mut app); + assert!(chip.contains("GPT-6"), "{chip}"); + assert!(!chip.contains("Turbo"), "{chip}"); + + // With no menu open, T3 renames the thread's model and finds Claude broken. + let models = [("gpt-6", "GPT-6 Turbo"), ("gpt-6-mini", "GPT-6 Mini")]; + let renamed = provider("codex", "Codex", "ready", &models); + app.on_config_event(provider_statuses(vec![renamed, claude("error")])); + let chip = model_chip(&mut app); + assert!(chip.contains("GPT-6 Turbo"), "{chip}"); + + // The next menu shows both changes from the copy the TUI holds, and reads nothing. + alt(&mut app, 'm'); + let menu = menu_text(&mut app); + assert!(menu.contains("GPT-6 Turbo"), "{menu}"); + assert!(menu.contains("Claude (error)"), "{menu}"); + assert_eq!(reads(&runtime), 0); + } + + #[test] + fn an_open_menu_keeps_its_highlight_on_the_same_model_as_providers_change() { + let runtime = idle_runtime(); + let _entered = runtime.enter(); + let mut app = tui(&Settings::default()); + app.open_thread(thread_on_codex()); + let snapshot = config_snapshot(vec![codex("ready"), claude("ready")], json!({})); + app.on_config_event(snapshot); + alt(&mut app, 'm'); + for _ in 0..3 { + press(&mut app, KeyCode::Down); + } + let haiku = Some(Pick::Model("claudeAgent/claude-haiku-4-5".into())); + assert_eq!(app.selected_pick(), haiku); + let row = app.picker.as_ref().unwrap().selected; + + // A provider listed first moves Haiku two rows down, and the highlight moves with it. + let models = [("composer-2", "Composer 2")]; + let cursor = provider("cursor", "Cursor", "ready", &models); + let providers = vec![cursor.clone(), codex("ready"), claude("ready")]; + app.on_config_event(provider_statuses(providers)); + let menu = menu_text(&mut app); + assert!(menu.contains("Composer 2"), "{menu}"); + assert_eq!(app.selected_pick(), haiku); + assert_eq!(app.picker.as_ref().unwrap().selected, row + 2); + + // With Claude gone, the highlight goes to the thread's model and the menu stays open. + app.on_config_event(provider_statuses(vec![cursor, codex("ready")])); + let gpt = Some(Pick::Model("codex/gpt-6".into())); + assert_eq!(app.selected_pick(), gpt); + let menu = menu_text(&mut app); + assert!(!menu.contains("Haiku"), "{menu}"); + assert_eq!(reads(&runtime), 0); + } + + #[test] + fn provider_changes_leave_the_draft_and_the_composer_alone() { + let runtime = idle_runtime(); + let _entered = runtime.enter(); + let mut app = tui(&Settings::default()); + app.open_thread(thread_on_codex()); + let snapshot = config_snapshot(vec![codex("ready"), claude("ready")], json!({})); + app.on_config_event(snapshot); + app.composer.insert_str("half a reply"); + alt(&mut app, 'm'); + for _ in 0..3 { + press(&mut app, KeyCode::Down); + } + press(&mut app, KeyCode::Enter); + let haiku = Some("claudeAgent/claude-haiku-4-5".to_string()); + assert_eq!(app.drafts["t"].model, haiku); + let chip = model_chip(&mut app); + assert!(chip.contains("Haiku 4.5"), "{chip}"); + + // While Claude is gone, the chip falls back to the thread's model. The draft waits for + // Claude to come back. + app.on_config_event(provider_statuses(vec![codex("ready")])); + let chip = model_chip(&mut app); + assert!(chip.contains("GPT-6"), "{chip}"); + assert_eq!(app.drafts["t"].model, haiku); + assert_eq!(app.composer.text(), "half a reply"); + + app.on_config_event(provider_statuses(vec![codex("ready"), claude("ready")])); + let chip = model_chip(&mut app); + assert!(chip.contains("Haiku 4.5"), "{chip}"); + assert_eq!(app.composer.text(), "half a reply"); + assert_eq!(reads(&runtime), 0); + } + + #[test] + fn a_model_list_read_that_answers_after_live_config_puts_nothing_back() { + let runtime = idle_runtime(); + let _entered = runtime.enter(); + let mut app = tui(&Settings::default()); + app.open_thread(thread_on_codex()); + // With no configuration yet, the menu reads it. + alt(&mut app, 'm'); + assert_eq!(reads(&runtime), 1); + let menu = menu_text(&mut app); + assert!(menu.contains("Loading T3's model list…"), "{menu}"); + let asked_at = app.config_revision; + + // The subscription's snapshot arrives first and finds Claude broken. + let snapshot = config_snapshot(vec![codex("ready"), claude("error")], json!({})); + app.on_config_event(snapshot); + let menu = menu_text(&mut app); + assert!(menu.contains("Claude (error)"), "{menu}"); + + // The read answers late with an older list, which changes nothing. + let pi = provider("pi", "Pi", "ready", &[("pi-1", "Pi One")]); + let older = json!({"providers": [codex("ready"), claude("ready"), pi]}); + let answer = ActionResult::Config { + config: older, + revision: asked_at, + }; + app.on_action_result(answer); + let menu = menu_text(&mut app); + assert!(menu.contains("Claude (error)"), "{menu}"); + assert!(!menu.contains("Pi One"), "{menu}"); + + // Later changes still apply, and the next menu reads nothing. + app.on_config_event(provider_statuses(vec![codex("ready"), claude("ready")])); + press(&mut app, KeyCode::Esc); + alt(&mut app, 'm'); + let menu = menu_text(&mut app); + assert!(menu.contains("Opus 5.5"), "{menu}"); + assert!(!menu.contains("(error)"), "{menu}"); + assert_eq!(reads(&runtime), 1); + } + + #[test] + fn a_model_list_read_fills_the_menu_until_the_first_snapshot() { + let runtime = idle_runtime(); + let _entered = runtime.enter(); + let mut app = tui(&Settings::default()); + app.open_thread(thread_on_codex()); + alt(&mut app, 'm'); + assert_eq!(reads(&runtime), 1); + let asked_at = app.config_revision; + + // A change before any snapshot has nothing to apply to, so the read's answer still + // counts. + app.on_config_event(provider_statuses(vec![codex("error")])); + let menu = menu_text(&mut app); + assert!(menu.contains("Loading T3's model list…"), "{menu}"); + let config = json!({"providers": [codex("ready"), claude("ready")]}); + let answer = ActionResult::Config { + config, + revision: asked_at, + }; + app.on_action_result(answer); + let menu = menu_text(&mut app); + assert!(menu.contains("Opus 5.5"), "{menu}"); + assert!(!menu.contains("(error)"), "{menu}"); + + // The first snapshot replaces what the read brought. + app.on_config_event(config_snapshot(vec![codex("ready")], json!({}))); + let menu = menu_text(&mut app); + assert!(menu.contains("GPT-6"), "{menu}"); + assert!(!menu.contains("Opus 5.5"), "{menu}"); + assert_eq!(reads(&runtime), 1); + } + + #[test] + fn a_reconnect_keeps_the_config_until_a_fresh_snapshot_replaces_it() { + let runtime = idle_runtime(); + let _entered = runtime.enter(); + let mut app = tui(&Settings::default()); + let mut zulu = listed("z", "Zulu", "idle"); + zulu["snoozedUntil"] = json!("2999-01-01T00:00:00.000Z"); + app.on_shell_event(shell_snapshot(vec![zulu])); + app.open_thread(thread_on_codex()); + let rows = screen(&mut app, 120, 30); + assert!(!rows.iter().any(|row| row.contains("Snoozed")), "{rows:#?}"); + + // The first snapshot turns snooze on, and the sidebar moves Zulu to its shelf. + let providers = vec![codex("ready"), claude("ready")]; + let snapshot = config_snapshot(providers, json!({"threadSnooze": true})); + app.on_config_event(snapshot); + let rows = screen(&mut app, 120, 30); + let snoozed = row_showing(&rows, "Snoozed ─"); + assert!(snoozed < row_showing(&rows, "Zulu")); + alt(&mut app, 'm'); + for _ in 0..3 { + press(&mut app, KeyCode::Down); + } + let haiku = Some(Pick::Model("claudeAgent/claude-haiku-4-5".into())); + assert_eq!(app.selected_pick(), haiku); + + // While the subscription reconnects, everything stays as the last snapshot left it. + let reason = "T3 connection lost: socket ended".to_string(); + let retry_in = Duration::from_millis(250); + app.on_config_event(WatchEvent::Reconnecting { reason, retry_in }); + assert_eq!(app.selected_pick(), haiku); + let menu = menu_text(&mut app); + assert!(menu.contains("Haiku 4.5"), "{menu}"); + let rows = screen(&mut app, 120, 30); + let snoozed = row_showing(&rows, "Snoozed ─"); + assert!(snoozed < row_showing(&rows, "Zulu")); + + // The new subscription's snapshot replaces all of it, so Claude and snooze go. + app.on_config_event(config_snapshot(vec![codex("ready")], json!({}))); + let gpt = Some(Pick::Model("codex/gpt-6".into())); + assert_eq!(app.selected_pick(), gpt); + let rows = screen(&mut app, 120, 30); + assert!(!rows.iter().any(|row| row.contains("Snoozed")), "{rows:#?}"); + assert!(row_showing(&rows, "Active ─") < row_showing(&rows, "Zulu")); + assert_eq!(reads(&runtime), 0); + } + + #[test] + fn after_t3_refuses_the_config_subscription_each_menu_reads_the_model_list() { + let runtime = idle_runtime(); + let _entered = runtime.enter(); + let mut app = tui(&Settings::default()); + app.open_thread(thread_on_codex()); + // T3 refuses the subscription, and the TUI reads the configuration once at once. + let refusal = "Unknown request tag: subscribeServerConfig".to_string(); + app.on_config_event(WatchEvent::Failed(refusal)); + assert_eq!(reads(&runtime), 1); + let asked_at = app.config_revision; + let config = json!({"providers": [codex("ready"), claude("ready")]}); + let answer = ActionResult::Config { + config, + revision: asked_at, + }; + app.on_action_result(answer); + + // Each menu now reads it again, and shows the last copy until the answer comes. + alt(&mut app, 'm'); + assert_eq!(reads(&runtime), 2); + let menu = menu_text(&mut app); + assert!(menu.contains("Opus 5.5"), "{menu}"); + assert!(!menu.contains("(error)"), "{menu}"); + let config = json!({"providers": [codex("ready"), claude("error")]}); + let answer = ActionResult::Config { + config, + revision: asked_at, + }; + app.on_action_result(answer); + let menu = menu_text(&mut app); + assert!(menu.contains("Claude (error)"), "{menu}"); + + // Moving to another menu reads nothing more. Opening one again reads once more. + alt(&mut app, 'e'); + assert_eq!(reads(&runtime), 2); + press(&mut app, KeyCode::Esc); + alt(&mut app, 'm'); + assert_eq!(reads(&runtime), 3); + } } diff --git a/tests/fake_server.rs b/tests/fake_server.rs index 03502aa..1bdfde6 100644 --- a/tests/fake_server.rs +++ b/tests/fake_server.rs @@ -1,18 +1,27 @@ //! Drives the RPC client and watcher against a fake Effect RPC server. +use std::collections::HashMap; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; use serde_json::{Value, json}; use tokio::net::TcpListener; +use tokio::sync::{mpsc, oneshot}; use tokio_tungstenite::tungstenite::Message; -use t3term::client::{WatchEvent, spawn_watch}; -use t3term::projection::ThreadState; +use t3term::client::{WatchEvent, spawn_config_watch, spawn_watch}; +use t3term::projection::{Applied, ThreadState, apply_config}; use t3term::rpc::{RpcClient, RpcError}; type Socket = tokio_tungstenite::WebSocketStream; +/// What a watcher's connect function returns. +type Connecting = Pin> + Send>>; + async fn recv(socket: &mut Socket) -> Value { loop { match socket.next().await.expect("client message").expect("frame") { @@ -34,6 +43,69 @@ async fn send(socket: &mut Socket, value: Value) { .unwrap(); } +/// Accepts the next connection to `listener` as a WebSocket. +async fn accept(listener: &TcpListener) -> Socket { + let (stream, _) = listener.accept().await.unwrap(); + tokio_tungstenite::accept_async(stream).await.unwrap() +} + +/// The next frame other than an interrupt of request `ended`. t3term interrupts each stream it +/// drops, even one the server has already ended. +async fn recv_after(socket: &mut Socket, ended: &Value) -> Value { + loop { + let frame = recv(socket).await; + if frame != interrupt(ended) { + return frame; + } + } +} + +/// A chunk of `values` on the stream of request `id`. +fn chunk(id: &Value, values: Value) -> Value { + json!({"_tag": "Chunk", "requestId": id, "values": values}) +} + +/// What t3term sends when a chunk on the stream of request `id` arrives. +fn ack(id: &Value) -> Value { + json!({"_tag": "Ack", "requestId": id}) +} + +/// What t3term sends when it drops the stream of request `id`. +fn interrupt(id: &Value) -> Value { + json!({"_tag": "Interrupt", "requestId": id, "interruptors": []}) +} + +/// Opens a new connection to `url` each time a watcher connects. +fn dial(url: String) -> impl Fn() -> Connecting + Send + Sync + 'static { + move || -> Connecting { + let url = url.clone(); + Box::pin(async move { RpcClient::connect_url(&url).await }) + } +} + +/// Hands a watcher `rpc` each time it connects, as `Client::rpc` does while the socket is open. +fn reuse(rpc: RpcClient) -> impl Fn() -> Connecting + Send + Sync + 'static { + move || -> Connecting { + let rpc = rpc.clone(); + Box::pin(async move { anyhow::Ok(rpc) }) + } +} + +/// The watcher's next event, or `None` once it has stopped. Fails the test after five seconds. +async fn next_event(events: &mut mpsc::UnboundedReceiver) -> Option { + tokio::time::timeout(Duration::from_secs(5), events.recv()) + .await + .expect("a watch event within five seconds") +} + +/// The item a watch event carries. Any other event fails the test. +fn item(event: Option) -> Value { + match event { + Some(WatchEvent::Item(item)) => item, + other => panic!("expected an item, got {other:?}"), + } +} + fn delta(sequence: u64, text: &str) -> Value { json!({"kind": "event", "sequence": sequence, "event": {"type": "turn-item.updated", "payload": { "id": "a1", "type": "assistant_message", "ordinal": 1, "runId": "r1", "text": text, "streaming": true @@ -152,3 +224,306 @@ async fn unary_call_reports_the_server_failure_message() { other => panic!("unexpected {other:?}"), } } + +// ---- the config subscription ---- + +/// How Effect's RPC server fails a request for a method the server doesn't have. +const UNKNOWN_METHOD: &str = "Unknown request tag: subscribeServerConfig"; + +/// How T3 fails a request from a session without the scope the method needs. +const MISSING_SCOPE: &str = + "The authenticated token is missing required scope: orchestration:read."; + +/// Reads a config subscription's request and returns its id. The payload is always empty: no +/// cursor, and none of the optional flags for the theme and usage-limit events. +async fn config_request(socket: &mut Socket) -> Value { + let request = recv(socket).await; + let id = request["id"].clone(); + assert!(id.is_string(), "{request}"); + let expected = json!({"_tag": "Request", "id": id, "tag": "subscribeServerConfig", + "payload": {}, "headers": []}); + assert_eq!(request, expected); + id +} + +/// A provider in T3's configuration. +fn provider(id: &str) -> Value { + json!({"instanceId": id, "status": "ready", "enabled": true, "models": []}) +} + +/// A config snapshot that lists `providers`. +fn config_snapshot(providers: Value) -> Value { + let config = json!({"providers": providers, "settings": {}}); + json!({"version": 1, "type": "snapshot", "config": config}) +} + +/// A new list of every provider. +fn provider_statuses(providers: Value) -> Value { + let payload = json!({"providers": providers}); + json!({"version": 1, "type": "providerStatuses", "payload": payload}) +} + +/// An `Exit` that fails request `id` for `reason`, one reason of an Effect `Cause`. +fn refusal(id: &Value, reason: Value) -> Value { + let exit = json!({"_tag": "Failure", "cause": [reason]}); + json!({"_tag": "Exit", "requestId": id, "exit": exit}) +} + +#[tokio::test] +async fn config_watch_starts_again_from_a_snapshot_and_never_sends_a_cursor() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + + let server = tokio::spawn(async move { + // First connection: a snapshot and a change, each with fields an orchestration stream + // would resume after, then a drop without an Exit. + let mut socket = accept(&listener).await; + let id = config_request(&mut socket).await; + let mut snapshot = config_snapshot(json!([provider("codex"), provider("claude")])); + snapshot["sequence"] = json!(41); + snapshot["snapshotSequence"] = json!(40); + let mut change = provider_statuses(json!([provider("claude")])); + change["sequence"] = json!(42); + change["snapshot"] = json!({"snapshotSequence": 43}); + send(&mut socket, chunk(&id, json!([snapshot, change]))).await; + assert_eq!(recv(&mut socket).await, ack(&id)); + drop(socket); + + // Second connection: the same request, so T3 sends the whole configuration again. + let mut socket = accept(&listener).await; + let id = config_request(&mut socket).await; + let snapshot = config_snapshot(json!([provider("cursor")])); + send(&mut socket, chunk(&id, json!([snapshot]))).await; + assert_eq!(recv(&mut socket).await, ack(&id)); + // Dropping the receiver interrupts the stream. + assert_eq!(recv(&mut socket).await, interrupt(&id)); + }); + + let mut events = spawn_config_watch(dial(url)); + let mut config = None; + let first = item(next_event(&mut events).await); + assert_eq!(apply_config(&mut config, &first), Applied::Snapshot); + let change = item(next_event(&mut events).await); + let applied = apply_config(&mut config, &change); + assert_eq!(applied, Applied::Event("providerStatuses".into())); + let held = config.clone().unwrap(); + assert_eq!(held["providers"], json!([provider("claude")])); + match next_event(&mut events).await { + Some(WatchEvent::Reconnecting { .. }) => {} + other => panic!("expected a reconnect, got {other:?}"), + } + // The new subscription's snapshot replaces the whole configuration. + let fresh = item(next_event(&mut events).await); + assert_eq!(apply_config(&mut config, &fresh), Applied::Snapshot); + let held = config.clone().unwrap(); + assert_eq!(held["providers"], json!([provider("cursor")])); + drop(events); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .expect("the server saw the interrupt") + .unwrap(); +} + +#[tokio::test] +async fn config_watch_backs_off_until_a_snapshot_shows_the_subscription_works() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + + let server = tokio::spawn(async move { + let snapshot = config_snapshot(json!([provider("codex")])); + let change = provider_statuses(json!([provider("codex")])); + // Four connections that drop after these items: a snapshot, nothing, a snapshot again + // and a change alone. + let drops = [ + json!([snapshot]), + json!([]), + json!([snapshot]), + json!([change]), + ]; + for items in drops { + let mut socket = accept(&listener).await; + let id = config_request(&mut socket).await; + if items != json!([]) { + send(&mut socket, chunk(&id, items)).await; + assert_eq!(recv(&mut socket).await, ack(&id)); + } + } + // A fifth that stays up until the receiver goes. + let mut socket = accept(&listener).await; + let id = config_request(&mut socket).await; + send(&mut socket, chunk(&id, json!([snapshot]))).await; + assert_eq!(recv(&mut socket).await, ack(&id)); + assert_eq!(recv(&mut socket).await, interrupt(&id)); + }); + + let mut events = spawn_config_watch(dial(url)); + let mut waits = Vec::new(); + loop { + match next_event(&mut events).await { + Some(WatchEvent::Reconnecting { retry_in, .. }) => waits.push(retry_in.as_millis()), + // The fifth connection's snapshot. + Some(WatchEvent::Item(_)) if waits.len() == 4 => break, + Some(WatchEvent::Item(_)) => {} + other => panic!("unexpected {other:?}"), + } + } + // A snapshot brings the wait back to the shortest. A subscription that sends nothing + // doubles it, and so does one that sends only a change. + assert_eq!(waits, [250, 500, 250, 500]); + drop(events); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .expect("the server saw the interrupt") + .unwrap(); +} + +#[tokio::test] +async fn dropping_the_receiver_stops_a_config_watch_that_is_still_connecting() { + let calls = Arc::new(AtomicUsize::new(0)); + let (alive, stopped) = oneshot::channel::<()>(); + let counted = calls.clone(); + // A connection that never opens. + let events = spawn_config_watch(move || -> Connecting { + // The watcher holds this function, and so `alive`, until it stops. + let _alive = &alive; + counted.fetch_add(1, Ordering::SeqCst); + Box::pin(std::future::pending::>()) + }); + let connecting = async { + while calls.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }; + tokio::time::timeout(Duration::from_secs(5), connecting) + .await + .expect("the watcher connects"); + + drop(events); + let stopped = tokio::time::timeout(Duration::from_secs(5), stopped).await; + assert!(stopped.expect("the watcher stops").is_err()); + assert_eq!(calls.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn dropping_the_receiver_stops_a_config_watch_that_waits_to_retry() { + let calls = Arc::new(AtomicUsize::new(0)); + let (alive, stopped) = oneshot::channel::<()>(); + let counted = calls.clone(); + // A connection refused at once, so the watcher waits longer before each try. + let mut events = spawn_config_watch(move || -> Connecting { + let _alive = &alive; + counted.fetch_add(1, Ordering::SeqCst); + Box::pin(async { Err(anyhow::anyhow!("connection refused")) }) + }); + let mut waits = Vec::new(); + while waits.len() < 3 { + match next_event(&mut events).await { + Some(WatchEvent::Reconnecting { reason, retry_in }) => { + assert_eq!(reason, "connection refused"); + waits.push(retry_in.as_millis()); + } + other => panic!("expected a reconnect, got {other:?}"), + } + } + assert_eq!(waits, [250, 500, 1000]); + + // The watcher now waits a second before its fourth try. Dropping the receiver ends the + // wait, so the fourth try never starts. + drop(events); + let stopped = tokio::time::timeout(Duration::from_secs(5), stopped).await; + assert!(stopped.expect("the watcher stops").is_err()); + assert_eq!(calls.load(Ordering::SeqCst), 3); +} + +#[tokio::test] +async fn an_unknown_config_method_fails_that_watch_and_leaves_the_shell_watch_alone() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + // Both subscriptions share the connection, and either can arrive first. + let mut ids = HashMap::new(); + for _ in 0..2 { + let request = recv(&mut socket).await; + let tag = request["tag"].as_str().unwrap().to_string(); + ids.insert(tag, request["id"].clone()); + } + let config = ids["subscribeServerConfig"].clone(); + let shell = ids["orchestration.subscribeShell"].clone(); + // A server without the method fails that request alone, as Effect does. + let unknown = json!({"_tag": "Die", "defect": UNKNOWN_METHOD}); + send(&mut socket, refusal(&config, unknown)).await; + + let snapshot = json!({"kind": "snapshot", "snapshot": {"snapshotSequence": 1}}); + let values = json!([snapshot, {"kind": "synchronized"}]); + send(&mut socket, chunk(&shell, values)).await; + assert_eq!(recv_after(&mut socket, &config).await, ack(&shell)); + let event = json!({"kind": "thread.updated", "sequence": 2, "thread": {"id": "t1"}}); + send(&mut socket, chunk(&shell, json!([event]))).await; + assert_eq!(recv_after(&mut socket, &config).await, ack(&shell)); + // Nothing asks for the config again, so the next frame interrupts the shell watch. + assert_eq!(recv_after(&mut socket, &config).await, interrupt(&shell)); + }); + + let rpc = RpcClient::connect_url(&url).await.unwrap(); + let mut config_events = spawn_config_watch(reuse(rpc.clone())); + let tag = "orchestration.subscribeShell"; + let payload = json!({"requestCompletionMarker": true}); + let mut shell_events = spawn_watch(reuse(rpc), tag, payload, None); + + match next_event(&mut config_events).await { + Some(WatchEvent::Failed(message)) => assert_eq!(message, UNKNOWN_METHOD), + other => panic!("expected the refusal, got {other:?}"), + } + assert!(next_event(&mut config_events).await.is_none(), "it retried"); + + // The shell watch on the same connection saw each item and never reconnected. + let mut kinds = Vec::new(); + while kinds.len() < 3 { + let next = item(next_event(&mut shell_events).await); + kinds.push(next["kind"].as_str().unwrap().to_string()); + } + assert_eq!(kinds, ["snapshot", "synchronized", "thread.updated"]); + drop(shell_events); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .expect("the server saw the interrupt") + .unwrap(); +} + +#[tokio::test] +async fn a_refused_config_subscription_ends_the_watch_without_a_retry() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + // The two ways T3 refuses it: Effect's answer for a method the server doesn't have, and + // T3's own for a session without the scope it needs. + let unknown = json!({"_tag": "Die", "defect": UNKNOWN_METHOD}); + let scope = json!({"_tag": "Fail", "error": {"_tag": "EnvironmentAuthorizationError", + "message": MISSING_SCOPE, "requiredScope": "orchestration:read"}}); + let refusals = [(unknown, UNKNOWN_METHOD), (scope, MISSING_SCOPE)]; + + let answers = refusals.clone(); + let server = tokio::spawn(async move { + for (reason, _) in answers { + let mut socket = accept(&listener).await; + let id = config_request(&mut socket).await; + send(&mut socket, refusal(&id, reason)).await; + // t3term interrupts the stream it dropped, and asks for nothing more. + assert_eq!(recv(&mut socket).await, interrupt(&id)); + } + }); + + for (_, message) in refusals { + let mut events = spawn_config_watch(dial(url.clone())); + match next_event(&mut events).await { + Some(WatchEvent::Failed(refused)) => assert_eq!(refused, message), + other => panic!("expected the refusal, got {other:?}"), + } + assert!(next_event(&mut events).await.is_none(), "it retried"); + } + tokio::time::timeout(Duration::from_secs(5), server) + .await + .expect("the server saw both interrupts") + .unwrap(); +}