diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 867f1356c..809f26868 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -83,6 +83,7 @@ struct Write { id: u64, entry: outbox::Entry, sent: Option, + adopted: bool, } #[derive(Default)] @@ -214,6 +215,7 @@ impl HostLink { } } if let HostMessage::Ack { result: Ok(_), .. } = &message + && !write.adopted && matches!( write.entry.command, Command::SendTurn { .. } @@ -270,6 +272,7 @@ impl HostLink { id: self.next_id(), entry, sent: None, + adopted: false, }); } Ok(()) @@ -302,12 +305,15 @@ impl HostLink { } pub fn retire_acknowledged_message(&self, key: &str) { - self.inner - .delivery - .lock() - .unwrap() - .acknowledged - .retain(|entry| entry.key != key); + let mut delivery = self.inner.delivery.lock().unwrap(); + delivery.acknowledged.retain(|entry| entry.key != key); + // A replica can arrive before its command receipt. Remember adoption + // on the pending write so its later receipt cannot revive a placeholder. + for write in &mut delivery.writes { + if write.entry.key == key { + write.adopted = true; + } + } } pub fn failed_commands(&self) -> Vec<(outbox::Entry, ProtocolError)> { @@ -364,6 +370,7 @@ impl HostLink { id, entry, sent: None, + adopted: false, }); let mut evicted = Vec::new(); while delivery.writes.len() > outbox::MAX_ITEMS diff --git a/crates/core/src/project.rs b/crates/core/src/project.rs index b726f1816..74ef128ee 100644 --- a/crates/core/src/project.rs +++ b/crates/core/src/project.rs @@ -139,6 +139,49 @@ pub struct SessionMeta { pub updated_at: u64, } +/// An ephemeral index of the stored and live sessions using each path or branch. +/// Two distinct owners suffice to answer sharing queries excluding the owner itself. +pub struct WorktreeSharing<'a> { + paths: HashMap>, + branches: HashMap<&'a str, Vec<&'a SessionMeta>>, +} + +impl<'a> WorktreeSharing<'a> { + pub fn new(sessions: impl IntoIterator) -> Self { + let mut sharing = Self { + paths: HashMap::new(), + branches: HashMap::new(), + }; + for meta in sessions { + for ancestor in meta.cwd.ancestors() { + Self::add_owner(sharing.paths.entry(ancestor.to_owned()).or_default(), meta); + } + if let Some(worktree) = &meta.worktree { + Self::add_owner(sharing.branches.entry(&worktree.branch).or_default(), meta); + } + } + sharing + } + + fn add_owner(owners: &mut Vec<&'a SessionMeta>, meta: &'a SessionMeta) { + if owners.len() < 2 && owners.iter().all(|other| other.id != meta.id) { + owners.push(meta); + } + } + + pub fn is_shared(&self, meta: &SessionMeta) -> bool { + let Some(worktree) = &meta.worktree else { + return false; + }; + self.paths + .get(&meta.cwd) + .into_iter() + .chain(self.branches.get(worktree.branch.as_str())) + .flatten() + .any(|other| meta.shares_worktree_with(other)) + } +} + impl SessionMeta { /// Whether `other` works in the worktree this session owns. A fork keeps /// the source's cwd without the `worktree` ownership marker, so the cwd @@ -569,13 +612,16 @@ mod tests { let sibling = SessionMeta::new(ProviderKind::Codex, PathBuf::from("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/wt/source-2"), None); let checkout = SessionMeta::new(ProviderKind::Codex, PathBuf::from("/repo"), None); - assert!(owner.shares_worktree_with(&fork)); - assert!(owner.shares_worktree_with(&nested)); - assert!(!owner.shares_worktree_with(&sibling)); - assert!(!owner.shares_worktree_with(&checkout)); - assert!(!owner.shares_worktree_with(&owner)); + assert!(WorktreeSharing::new([&owner, &fork]).is_shared(&owner)); + assert!(WorktreeSharing::new([&owner, &nested]).is_shared(&owner)); + assert!(!WorktreeSharing::new([&owner, &sibling]).is_shared(&owner)); + assert!(!WorktreeSharing::new([&owner, &checkout]).is_shared(&owner)); + assert!(!WorktreeSharing::new([&owner, &owner]).is_shared(&owner)); + let mut same_branch = sibling.clone(); + same_branch.worktree = owner.worktree.clone(); + assert!(WorktreeSharing::new([&owner, &same_branch]).is_shared(&owner)); assert!( - !fork.shares_worktree_with(&owner), + !WorktreeSharing::new([&owner, &fork]).is_shared(&fork), "a fork owns no worktree" ); } diff --git a/crates/core/src/session.rs b/crates/core/src/session.rs index bbd2b62c8..f543ca53b 100644 --- a/crates/core/src/session.rs +++ b/crates/core/src/session.rs @@ -608,9 +608,14 @@ impl Timeline { /// cannot tell a live turn from one whose provider stopped without a /// record, and a window cut inside the turn misses its start. Nothing the /// records built is discarded, so a later settle can revive the turn. - pub fn settle_running_turn(&mut self, running: Option, started_at: Option) { - let running = running.filter(|turn| *turn < self.turns.len()); - self.turn_running = running.is_some(); + pub fn settle_running_turn( + &mut self, + turn_running: bool, + running: Option, + started_at: Option, + ) { + let running = running.filter(|turn| turn_running && *turn < self.turns.len()); + self.turn_running = turn_running; for (index, turn) in self.turns.iter_mut().enumerate() { turn.running = running == Some(index); } @@ -635,6 +640,10 @@ impl Timeline { .filter(|plan| plan.ready && !self.plan_resolutions.contains_key(plan.item_id.as_str())) } + pub fn plan_resolved(&self, item_id: &str) -> bool { + self.plan_resolutions.contains_key(item_id) + } + /// First user message in the timeline, if any (used for session titles). pub fn first_user_message(&self) -> Option<&str> { self.entries.iter().find_map(|entry| match &entry.content { @@ -2559,9 +2568,9 @@ mod tests { assert!(interrupted.proposed_plan.is_none()); // A client settled idle hides the streamed plan but can revive it. let mut settled = timeline.clone(); - settled.settle_running_turn(None, None); + settled.settle_running_turn(false, None, None); assert!(settled.shown_proposed_plan().is_none()); - settled.settle_running_turn(Some(0), None); + settled.settle_running_turn(true, Some(0), None); assert_eq!( settled.shown_proposed_plan().unwrap().markdown, "# Plan\nstep one" diff --git a/crates/protocol/src/event.rs b/crates/protocol/src/event.rs index 0693ea74d..aa4f450a6 100644 --- a/crates/protocol/src/event.rs +++ b/crates/protocol/src/event.rs @@ -2,8 +2,8 @@ use std::collections::{HashMap, HashSet}; use std::path::PathBuf; use agent::{ - ApprovalMode, ApprovalRequest, InteractionMode, OptionDescriptor, OptionSelection, - ProviderCommand, ProviderKind, RewindMode, + ApprovalMode, ApprovalRequest, InteractionMode, OptionDescriptor, OptionSelection, PlanStep, + ProviderCommand, ProviderKind, RewindMode, TokenUsage, }; use serde::{Deserialize, Serialize}; use tcode_core::{ @@ -24,6 +24,7 @@ pub type SessionEventRecord = StoredEvent; pub enum Topic { SessionEvents { session_id: String }, SessionStatus { session_id: String }, + SessionPlan { session_id: String }, Index, Settings, Providers, @@ -63,7 +64,8 @@ pub enum ServerEvent { request: crate::PreviewRequest, }, SessionEvent(StoredEvent), - SessionStatusReplaced(SessionStatus), + SessionStatusReplaced(Box), + SessionPlanReplaced(SessionPlan), ProvidersReplaced(ProvidersStatus), GitStatusReplaced(GitStatusStatus), /// An unarchived thread was added or changed. Archiving a thread removes @@ -350,11 +352,44 @@ pub struct GitStatusStatus { pub busy: bool, } -/// Full, ephemeral runtime status for one session. -/// -/// Unlike [`StoredEvent`], these values are not derivable by folding the -/// persisted agent event stream. Hosts replace the whole value whenever one of -/// its fields changes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ForkAvailability { + Available, + Unsupported, + Empty, + Running, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct SessionActivity { + pub working: bool, + pub turn_running: bool, + pub background_only: bool, + pub waiting_for_approval: bool, + pub waiting_for_input: bool, + pub unread: bool, + pub fork: ForkAvailability, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct SessionPlan { + pub session_id: String, + pub proposed: Option, + pub steps: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProposedPlanStatus { + pub item_id: String, + /// Absolute turn index in the full session history. + pub turn: usize, + pub markdown: String, + pub ready: bool, + pub resolved: bool, +} + +/// Host-owned runtime and full-history facts for one resident session. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct SessionStatus { pub session_id: String, @@ -367,6 +402,8 @@ pub struct SessionStatus { pub acp_agent_id: Option, pub project_id: Option, pub approval_mode: ApprovalMode, + pub effective_approval_mode: ApprovalMode, + pub native_approval_modes_enabled: bool, pub interaction_mode: InteractionMode, pub queued_messages: Vec, /// Backend-owned drafts used by provider-bound send-path assembly. @@ -378,11 +415,18 @@ pub struct SessionStatus { pub terminal_open: bool, pub terminal_height: f32, pub delivery_in_flight: Option, - pub turn_running: bool, + pub activity: SessionActivity, /// An interrupt was accepted by the provider and its turn has not ended yet. #[serde(default)] pub stopping: bool, - pub working: bool, + pub native_rewind_blocked: bool, + pub checkout_blocked: bool, + pub conversation_read_only: bool, + pub terminal_limit_reached: bool, + pub terminal_split_available: bool, + pub usage: Option, + /// Effective capacity the context meter should use, resolved by the host. + pub context_window: Option, /// The live turn and the requests it waits on. A client's fold of a /// history window can place them but cannot decide them: the window may /// begin after they opened, and a provider can stop without a record. @@ -437,6 +481,7 @@ pub struct TerminalContextStatus { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct QueuedMessageStatus { pub id: u64, + pub editable: bool, #[serde(default, skip_serializing_if = "Option::is_none")] pub delivery_key: Option, pub text: String, @@ -459,18 +504,24 @@ pub struct IndexSnapshot { /// Index facts that are not one thread's metadata. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct IndexSummary { - /// Working, approval, user-input and background-only flags for sidebar rows. + /// Host-authored activity for every unarchived stored session. #[serde(default)] - pub activity: HashMap, + pub activity: HashMap, /// Background title requests, including threads with no live provider. #[serde(default)] pub title_generating: HashSet, /// Archived threads per project id. #[serde(default)] pub archived_counts: HashMap, - /// Worktree branches archived threads still use. - #[serde(default)] - pub archived_worktree_branches: HashSet, + pub worktree_shared: HashSet, + pub archived_revision: u64, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ArchivedSessions { + pub sessions: Vec, + pub worktree_shared: HashSet, + pub revision: u64, } /// A transient runtime notification delivered to clients. diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index d846aeb03..4b7c75665 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -14,14 +14,15 @@ mod wire; pub use command::{Command, CommandResponse, SettingsPatch, TerminalSelection, ThreadExportFormat}; pub use event::{ - AcpMarketplaceItem, EventEnvelope, ExternalImportState, ExternalImportStatus, GitActionRequest, - GitStatusStatus, IndexSnapshot, IndexSummary, MergeWorktreeFailure, NoticeSeverity, - PluginCatalogState, PluginChallenge, PluginChallengeKind, PluginOperationTarget, - PluginStaleReason, ProviderPluginCatalog, ProviderUpdateAvailable, ProviderUpdateRun, - ProviderVersionStatus, ProvidersStatus, QueuedMessageStatus, RuntimeEffect, RuntimeError, - RuntimeNotice, RuntimeNotification, RuntimeOperationId, RuntimeToast, ServerEvent, - SessionEventRecord, SessionStatus, TcodeUpdateStatus, TerminalContextStatus, - TerminalSplitStatus, TerminalStatus, Topic, + AcpMarketplaceItem, ArchivedSessions, EventEnvelope, ExternalImportState, ExternalImportStatus, + ForkAvailability, GitActionRequest, GitStatusStatus, IndexSnapshot, IndexSummary, + MergeWorktreeFailure, NoticeSeverity, PluginCatalogState, PluginChallenge, PluginChallengeKind, + PluginOperationTarget, PluginStaleReason, ProposedPlanStatus, ProviderPluginCatalog, + ProviderUpdateAvailable, ProviderUpdateRun, ProviderVersionStatus, ProvidersStatus, + QueuedMessageStatus, RuntimeEffect, RuntimeError, RuntimeNotice, RuntimeNotification, + RuntimeOperationId, RuntimeToast, ServerEvent, SessionActivity, SessionEventRecord, + SessionPlan, SessionStatus, TcodeUpdateStatus, TerminalContextStatus, TerminalSplitStatus, + TerminalStatus, Topic, }; pub use query::{ ExternalThread, GitDiffResult, GitDiffScope, GitFileText, HostedDevice, HostingAction, @@ -55,6 +56,9 @@ pub use wire::{ // by UpdateProviders, and sequential update run status with toast progress; // MarkSessionRead, the client's acknowledgement that a thread's conversation // loaded, replaces marking a thread read when it is subscribed. +// Unreleased: host-authored session activity, action availability, full usage and +// meter capacity; a separate session plan topic; shared worktree facts and +// archived-list revisions. SessionStatus working/turn_running move into activity. pub const PROTOCOL_VERSION: u32 = 9; #[cfg(test)] diff --git a/crates/protocol/src/query.rs b/crates/protocol/src/query.rs index 0de9a50f1..ec2a0a67f 100644 --- a/crates/protocol/src/query.rs +++ b/crates/protocol/src/query.rs @@ -144,7 +144,7 @@ pub enum QueryResponse { /// screen, not a session. TerminalFrame(Box), ItemOutput(String), - ArchivedSessions(Vec), + ArchivedSessions(crate::ArchivedSessions), } /// One content match in a stored session, addressed by the folded timeline diff --git a/crates/protocol/src/tests.rs b/crates/protocol/src/tests.rs index 344424166..b8f7b2bb4 100644 --- a/crates/protocol/src/tests.rs +++ b/crates/protocol/src/tests.rs @@ -221,7 +221,7 @@ fn binary_payloads_use_base64_and_reject_corrupt_input() { } #[test] -fn older_messages_default_new_optional_fields() { +fn omitted_optional_subscription_index_and_queue_fields_default() { let subscription: Subscription = serde_json::from_value(json!({"topic": {"type": "index"}})).unwrap(); assert_eq!( @@ -231,11 +231,13 @@ fn older_messages_default_new_optional_fields() { after: None } ); - let index: IndexSnapshot = - serde_json::from_value(json!({"sessions": [], "projects": []})).unwrap(); + let index: IndexSnapshot = serde_json::from_value( + json!({"sessions": [], "projects": [], "worktree_shared": [], "archived_revision": 0}), + ) + .unwrap(); assert_eq!(index.summary, IndexSummary::default()); let queued: QueuedMessageStatus = - serde_json::from_value(json!({"id": 1, "text": "next"})).unwrap(); + serde_json::from_value(json!({"id": 1, "text": "next", "editable": true})).unwrap(); assert_eq!(queued.fire_at_unix_secs, None); let command = decode_client_line(r#"{"id":1,"payload":{"type":"command","content":{"type":"capture_terminal_selection","content":{"session_id":"s","terminal_id":9}}}}"#).unwrap(); assert!(matches!( @@ -919,6 +921,20 @@ fn version_six_index_visits_output_and_elision_literal_json() { let index = IndexSnapshot { summary: IndexSummary { archived_counts: [("p".to_string(), 2)].into(), + activity: [( + "cold".into(), + SessionActivity { + working: false, + turn_running: false, + background_only: false, + waiting_for_approval: false, + waiting_for_input: false, + unread: true, + fork: ForkAvailability::Available, + }, + )] + .into(), + archived_revision: 7, ..IndexSummary::default() }, sessions: vec![], @@ -926,8 +942,10 @@ fn version_six_index_visits_output_and_elision_literal_json() { }; assert_eq!( serde_json::to_value(&index).unwrap(), - json!({"activity": {}, "title_generating": [], "archived_counts": {"p": 2}, - "archived_worktree_branches": [], "sessions": [], "projects": []}) + json!({"activity": {"cold": {"working":false,"turn_running":false,"background_only":false, + "waiting_for_approval":false,"waiting_for_input":false,"unread":true,"fork":"available"}}, + "title_generating": [], "archived_counts": {"p": 2}, + "worktree_shared": [], "archived_revision": 7, "sessions": [], "projects": []}) ); assert_eq!( serde_json::to_value(ServerEvent::LastVisitedChanged( @@ -950,6 +968,64 @@ fn version_six_index_visits_output_and_elision_literal_json() { serde_json::from_str::(r#"{"type":"archived_sessions"}"#).unwrap(), Query::ArchivedSessions ); + let status_wire: serde_json::Value = serde_json::from_str(r#"{"type":"session_status_replaced", "content": { + "session_id":"s", "title":"Thread", "cwd":"/workspace", "attachments_dir":"/attachments", + "provider":"codex", "requested_model":null, "requested_profile_id":null, + "acp_agent_id":null, "project_id":null, "approval_mode":"supervised", + "effective_approval_mode":"supervised", "native_approval_modes_enabled":true, "interaction_mode":"build", + "queued_messages":[{"id":3,"delivery_key":"send-key","text":"Next","fire_at_unix_secs":null,"editable":false}], + "review_comment_drafts":[], "terminals":[], "active_terminal_id":null, "terminal_splits":[], + "terminal_contexts":[], "terminal_open":false, "terminal_height":240.0, "delivery_in_flight":3, + "activity":{"working":true,"turn_running":false,"background_only":false,"waiting_for_approval":false, + "waiting_for_input":false,"unread":false,"fork":"available"}, + "stopping":false,"native_rewind_blocked":true,"checkout_blocked":false,"conversation_read_only":false, + "terminal_limit_reached":false,"terminal_split_available":false,"usage":null,"context_window":200000, + "running_turn":null,"pending_approvals":[],"pending_user_input":null,"supports_steering":true, + "provider_option_descriptors":[],"provider_option_selections":[],"provider_commands":[], + "git_branch":null,"branches":[],"draft":false,"draft_workspace":{"kind":"local_checkout"},"worktree":null, + "preparing_worktree":false,"relay_confirmation":null,"native_rewind_pending":false, + "native_rewind_prefill_available":false,"model_pending_restart":false,"options_pending_restart":false, + "approval_pending_restart":false,"ultrathink_armed":false + }}"#).unwrap(); + let status = serde_json::from_value::(status_wire.clone()).unwrap(); + assert_eq!(serde_json::to_value(status).unwrap(), status_wire); + let archive_wire = json!({"type":"archived_sessions", "content": {"sessions":[], "worktree_shared":["owner"], "revision":7}}); + let archived = QueryResponse::ArchivedSessions(ArchivedSessions { + sessions: Vec::new(), + worktree_shared: ["owner".into()].into(), + revision: 7, + }); + assert_eq!(serde_json::to_value(&archived).unwrap(), archive_wire); + assert_eq!( + serde_json::from_value::(archive_wire).unwrap(), + archived + ); + let plan_wire = json!({"type":"session_plan_replaced", "content": {"session_id":"s", "proposed": { + "item_id":"plan", "turn":42, "markdown":"# Plan", "ready":true, "resolved":false + }, "steps":[]}}); + let plan = ServerEvent::SessionPlanReplaced(SessionPlan { + session_id: "s".into(), + proposed: Some(ProposedPlanStatus { + item_id: "plan".into(), + turn: 42, + markdown: "# Plan".into(), + ready: true, + resolved: false, + }), + steps: Vec::new(), + }); + assert_eq!(serde_json::to_value(&plan).unwrap(), plan_wire); + assert_eq!( + serde_json::from_value::(plan_wire).unwrap(), + plan + ); + assert_eq!( + serde_json::to_value(Topic::SessionPlan { + session_id: "s".into() + }) + .unwrap(), + json!({"type":"session_plan", "content":{"session_id":"s"}}) + ); let turn_started = AgentEvent::TurnStarted { turn_id: "t".into(), }; diff --git a/crates/runtime/src/app/command_validation.rs b/crates/runtime/src/app/command_validation.rs index f0d955521..3f0538111 100644 --- a/crates/runtime/src/app/command_validation.rs +++ b/crates/runtime/src/app/command_validation.rs @@ -1,6 +1,30 @@ use super::*; impl AppState { + pub(super) fn native_rewind_blocked(&self, active: &ActiveSession) -> bool { + !active.meta.provider.caps().native_rewind + || active.turn_in_flight + || active.delivery_in_flight.is_some() + || active.background_task_count > 0 + || active.timeline.turn_running + || !active.queue.is_empty() + || self.pending_native_rewinds.contains_key(&active.meta.id) + } + + pub(super) fn checkout_blocked(active: &ActiveSession) -> bool { + // A provider that dies without closing its timeline turn must not block + // checkout forever: the promise is no foreground turn. + active.turn_in_flight + } + + pub(super) fn conversation_read_only(meta: &SessionMeta) -> bool { + meta.native_subagent.is_some() + } + + pub(super) fn queued_message_editable(active: &ActiveSession, id: u64) -> bool { + active.delivery_in_flight != Some(id) + } + /// Validate retained conversation writes before the pipe acknowledges ownership. pub(crate) fn validate_command_target( &self, @@ -155,14 +179,47 @@ impl AppState { ) })?; match command { - Command::SetDraftWorkspace { .. } if !active.draft => return Err(error("not_a_draft", "Only a draft can change workspace.")), - Command::CheckoutBranch { .. } if active.timeline.turn_running => return Err(error("turn_running", "Wait for the running turn before changing branches.")), - Command::CaptureTerminalSelection { selection: None, .. } => return Err(error("no_selection", "Select terminal text first.")), + Command::SetDraftWorkspace { .. } if !active.draft => { + return Err(error("not_a_draft", "Only a draft can change workspace.")); + } + Command::CheckoutBranch { .. } if Self::checkout_blocked(active) => { + return Err(error( + "turn_running", + "Wait for the running turn before changing branches.", + )); + } + Command::CaptureTerminalSelection { + selection: None, .. + } => return Err(error("no_selection", "Select terminal text first.")), Command::NewTerminal { .. } | Command::SplitTerminal { .. } - if active.terminal_workspace.terminals.len() + self.pending_terminal_spawns.get(session_id).map_or(0, HashMap::len) >= MAX_TERMINALS_PER_SESSION => return Err(error("terminal_limit", "This thread has reached its terminal limit.")), - Command::SplitTerminal { .. } if active.terminal_workspace.active_id.is_none_or(|first| active.terminal_workspace.split_for(first).is_some() || self.pending_terminal_spawns.get(session_id).is_some_and(|spawns| spawns.values().any(|action| matches!(action, TerminalSpawnAction::Split { first: pending, .. } if *pending == first)))) => return Err(error("terminal_split_unavailable", "There is no terminal available to split.")), + if self.terminal_limit_reached(active) => + { + return Err(error( + "terminal_limit", + "This thread has reached its terminal limit.", + )); + } + Command::SplitTerminal { .. } if !self.terminal_split_available(active) => { + return Err(error( + "terminal_split_unavailable", + "There is no terminal available to split.", + )); + } Command::RewindTurn { turn, mode, .. } - if !active.meta.provider.caps().native_rewind || active.turn_in_flight || active.delivery_in_flight.is_some() || active.background_task_count > 0 || active.timeline.turn_running || !active.queue.is_empty() || active.timeline.turns.get(*turn).and_then(|turn| turn.provider_checkpoint_id.as_ref()).is_none() || (*turn == 0 && mode.includes_conversation()) || self.pending_native_rewinds.contains_key(session_id) => return Err(error("rewind_unavailable", "This turn cannot be rewound now.")), + if self.native_rewind_blocked(active) + || active + .timeline + .turns + .get(*turn) + .and_then(|turn| turn.provider_checkpoint_id.as_ref()) + .is_none() + || (*turn == 0 && mode.includes_conversation()) => + { + return Err(error( + "rewind_unavailable", + "This turn cannot be rewound now.", + )); + } Command::ActivateTerminal { terminal_id, .. } | Command::CloseTerminal { terminal_id, .. } | Command::CaptureTerminalSelection { terminal_id, .. } @@ -228,7 +285,7 @@ impl AppState { } Command::SteerQueued { id, .. } | Command::DropQueued { id, .. } if !active.queue.iter().any(|message| message.id == *id) - || active.delivery_in_flight == Some(*id) => + || !Self::queued_message_editable(active, *id) => { return Err(error( "unknown_queued_message", @@ -272,7 +329,7 @@ impl AppState { | Command::Steer { .. } | Command::ImplementPlan { .. } ) { - if active.meta.native_subagent.is_some() { + if Self::conversation_read_only(&active.meta) { return Err(error( "read_only_session", "This thread is a read-only mirror.", @@ -366,7 +423,7 @@ mod tests { } } 4 => { - state.resident_mut(&id).unwrap().timeline.turn_running = true; + state.resident_mut(&id).unwrap().turn_in_flight = true; Command::CheckoutBranch { session_id: id.clone(), branch: "main".into(), @@ -416,6 +473,45 @@ mod tests { assert!(state.resident(&id).unwrap().queue.is_empty()); } } + #[test] + fn checkout_is_admitted_after_provider_death_with_an_unclosed_timeline_turn() { + let store = TestStore::new("checkout-after-death"); + let mut state = AppState::new((*store).clone()).unwrap(); + let mut context = TestAppContext::default(); + let mut cx = context.host_cx(); + let id = state.start_draft("fixture".into(), std::env::temp_dir(), &mut cx); + let active = state.resident_mut(&id).unwrap(); + active.timeline.apply_at( + None, + &AgentEvent::TurnStarted { + turn_id: "unclosed".into(), + }, + ); + active.turn_in_flight = true; + active.mark_dead(); + assert!(active.timeline.turn_running); + assert!(!state.session_status_snapshot(&id).unwrap().checkout_blocked); + crate::pipe::handle_client_message( + &mut state, + &mut cx, + ClientMessage { + id: 43, + key: Some("checkout".into()), + payload: ClientPayload::Command(Command::CheckoutBranch { + session_id: id, + branch: "main".into(), + }), + }, + ); + assert!(context.drain_outgoing().iter().any(|message| matches!( + message, + HostMessage::Ack { + id: 43, + result: Ok(_) + } + ))); + } + #[test] fn provider_channel_rejection_is_not_acknowledged_as_success() { let store = TestStore::new("provider-write-rejection"); diff --git a/crates/runtime/src/app/lifecycle.rs b/crates/runtime/src/app/lifecycle.rs index b20267044..7355890d6 100644 --- a/crates/runtime/src/app/lifecycle.rs +++ b/crates/runtime/src/app/lifecycle.rs @@ -9,13 +9,7 @@ impl AppState { let Some(active) = self.resident(target_id) else { return; }; - if !active.meta.provider.caps().native_rewind - || active.turn_in_flight - || active.delivery_in_flight.is_some() - || active.background_task_count > 0 - || active.timeline.turn_running - || !active.queue.is_empty() - { + if self.native_rewind_blocked(active) { self.report_error(RuntimeError::NativeRewindBlocked, cx); return; } @@ -35,10 +29,6 @@ impl AppState { return; } let session_id = active.meta.id.clone(); - if self.pending_native_rewinds.contains_key(&session_id) { - self.report_error(RuntimeError::NativeRewindBlocked, cx); - return; - } let live_commands = match &active.runtime { Runtime::Live(commands) => Some(commands.clone()), Runtime::Idle | Runtime::Starting { .. } => None, @@ -332,10 +322,13 @@ impl AppState { self.clear_approvals(&session_id); self.pending_native_rewinds.remove(&session_id); } - if let Some(active) = self.residents.live.remove(target_id) - && let Runtime::Live(commands) = active.runtime - { - let _ = commands.try_send(SessionCommand::Shutdown); + if let Some(active) = self.residents.live.remove(target_id) { + if active.draft && self.archived_sharing_affected(&active.meta) { + self.archived_revision += 1; + } + if let Runtime::Live(commands) = active.runtime { + let _ = commands.try_send(SessionCommand::Shutdown); + } } self.release_stale_session_logs(cx); } @@ -372,6 +365,9 @@ impl AppState { let Some(mut active) = self.residents.live.remove(target_id) else { return; }; + if active.draft && self.archived_sharing_affected(&active.meta) { + self.archived_revision += 1; + } self.park_terminal_workspace(&mut active); let native_rewind_pending = self.pending_native_rewinds.contains_key(&active.meta.id); let has_work = active.turn_in_flight diff --git a/crates/runtime/src/app/mod.rs b/crates/runtime/src/app/mod.rs index 5584fac4a..5f27b2dc1 100644 --- a/crates/runtime/src/app/mod.rs +++ b/crates/runtime/src/app/mod.rs @@ -26,7 +26,7 @@ use tcode_core::acp::{AcpAgentPatch, InstalledAcpAgent as InstalledAgent}; use tcode_core::attachments::mime_from_path; use tcode_core::git::{GitAction, GitStatus, build_commit_prompt, sanitize_commit_message}; use tcode_core::project::{ - AutoArchiveConfig, AutoArchiveExemptions, Project, SessionMeta, WorktreeInfo, + AutoArchiveConfig, AutoArchiveExemptions, Project, SessionMeta, WorktreeInfo, WorktreeSharing, auto_archive_candidates, }; use tcode_core::provider_status::ProviderSnapshot; @@ -46,13 +46,14 @@ use tcode_core::ui::{ ConversationDestination, MAX_TERMINALS_PER_SESSION, TerminalSplitDirection, WorkspaceMode, }; use tcode_protocol::{ - AcpMarketplaceItem, EventEnvelope, ExternalImportState, ExternalImportStatus, ExternalThread, - GitActionRequest, GitStatusStatus, IndexSnapshot, IndexSummary, MergeWorktreeFailure, - PathEntry, ProtocolError, ProviderVersionStatus as ProtocolProviderVersionStatus, - ProvidersStatus, QueryResponse, QueuedMessageStatus, RecentDir, RuntimeEffect, RuntimeError, - RuntimeNotice, RuntimeNotification as RuntimeEvent, RuntimeOperationId, RuntimeToast, - ServerEvent, SessionEventRecord, SessionSearchHit, SessionStatus, TcodeUpdateStatus, - TerminalStatus, ThreadExportFormat, Topic, + AcpMarketplaceItem, ArchivedSessions, EventEnvelope, ExternalImportState, ExternalImportStatus, + ExternalThread, ForkAvailability, GitActionRequest, GitStatusStatus, IndexSnapshot, + IndexSummary, MergeWorktreeFailure, PathEntry, ProposedPlanStatus, ProtocolError, + ProviderVersionStatus as ProtocolProviderVersionStatus, ProvidersStatus, QueryResponse, + QueuedMessageStatus, RecentDir, RuntimeEffect, RuntimeError, RuntimeNotice, + RuntimeNotification as RuntimeEvent, RuntimeOperationId, RuntimeToast, ServerEvent, + SessionActivity, SessionEventRecord, SessionPlan, SessionSearchHit, SessionStatus, + TcodeUpdateStatus, TerminalStatus, ThreadExportFormat, Topic, }; use tcode_services::acp_registry::{ Registry, RegistryAgent, cached, install, load, platform_key, resolve_recipe, uninstall, @@ -257,7 +258,8 @@ use orchestrate::McpWiring; pub use providers::ProviderCatalog; use providers::{ computer_use_attaches, effort_selection, launch_env_for_profile, normalized_selections, - provider_secret_names, session_launch_env, session_options, + provider_secret_names, session_approval_policy, session_launch_env, session_options, + session_provider_settings, }; pub use sessions::ResidentSessions; pub(crate) use snapshots::DomainDiff; @@ -318,6 +320,7 @@ pub struct AppState { /// Set once the store failed and every thread was stopped. store_failed: bool, pub sessions: Vec, + archived_revision: u64, pub projects: Vec, pub residents: ResidentSessions, /// Terminal resources parked by conversation destination. Drawer chrome is @@ -508,6 +511,7 @@ impl AppState { store_failed: false, sessions, projects, + archived_revision: 0, residents: ResidentSessions::default(), terminal_workspaces: HashMap::new(), terminal_registry: TerminalRegistry::default(), diff --git a/crates/runtime/src/app/options.rs b/crates/runtime/src/app/options.rs index d38305221..2d8a04022 100644 --- a/crates/runtime/src/app/options.rs +++ b/crates/runtime/src/app/options.rs @@ -404,7 +404,7 @@ impl AppState { let Some(active) = self.resident(target_id) else { return; }; - if active.timeline.turn_running { + if Self::checkout_blocked(active) { return; } let cwd = active.meta.cwd.clone(); diff --git a/crates/runtime/src/app/providers.rs b/crates/runtime/src/app/providers.rs index afadb48cc..018835859 100644 --- a/crates/runtime/src/app/providers.rs +++ b/crates/runtime/src/app/providers.rs @@ -1009,6 +1009,38 @@ pub(super) fn computer_use_attaches(provider: ProviderKind, settings: &Settings) settings.computer_use.enabled && !provider.caps().native_computer_use } +pub(super) fn session_provider_settings( + meta: &SessionMeta, + settings: &Settings, +) -> tcode_core::settings::ProviderSettings { + meta.profile_id + .as_deref() + .and_then(|id| settings.resolved_profile(id)) + .map(|profile| profile.settings) + .unwrap_or_else(|| settings.provider(meta.provider)) +} + +pub(super) fn session_approval_policy( + meta: &SessionMeta, + provider_settings: &tcode_core::settings::ProviderSettings, +) -> (ApprovalMode, bool) { + let native_enabled = !meta + .provider + .caps() + .downgrade_approval_without_native_approvals + || provider_settings.pi.native_approvals; + let approval_mode = if !native_enabled { + match meta.approval_mode { + ApprovalMode::Supervised | ApprovalMode::AutoAcceptEdits => ApprovalMode::FullAccess, + ApprovalMode::ReadOnly => ApprovalMode::ReadOnly, + ApprovalMode::FullAccess => ApprovalMode::FullAccess, + } + } else { + meta.approval_mode + }; + (approval_mode, native_enabled) +} + pub(super) fn session_options( meta: &SessionMeta, settings: &Settings, @@ -1021,12 +1053,7 @@ pub(super) fn session_options( // A session's binary / launch-args come from its selected profile (built-in // or user-created), so a third-party profile can point at its own CLI while // sharing the protocol adapter. Falls back to the kind's built-in card. - let provider_settings = meta - .profile_id - .as_deref() - .and_then(|id| settings.resolved_profile(id)) - .map(|profile| profile.settings) - .unwrap_or_else(|| settings.provider(meta.provider)); + let provider_settings = session_provider_settings(meta, settings); // For an ACP session, which agent to launch (and how) comes from the // installed-agent list, keyed by the id the session was created with. let acp_agent: Option = meta @@ -1034,20 +1061,7 @@ pub(super) fn session_options( .as_deref() .and_then(|id| settings.acp_agent(id)) .cloned(); - let approval_mode = if meta - .provider - .caps() - .downgrade_approval_without_native_approvals - && !provider_settings.pi.native_approvals - { - match meta.approval_mode { - ApprovalMode::Supervised | ApprovalMode::AutoAcceptEdits => ApprovalMode::FullAccess, - ApprovalMode::ReadOnly => ApprovalMode::ReadOnly, - ApprovalMode::FullAccess => ApprovalMode::FullAccess, - } - } else { - meta.approval_mode - }; + let (approval_mode, _) = session_approval_policy(meta, &provider_settings); SessionOptions { cwd: meta.cwd.clone(), model: meta.model.clone(), diff --git a/crates/runtime/src/app/send.rs b/crates/runtime/src/app/send.rs index 4c8d3d423..784411577 100644 --- a/crates/runtime/src/app/send.rs +++ b/crates/runtime/src/app/send.rs @@ -188,23 +188,6 @@ impl AppState { self.reschedule_scheduled_wake(cx); } - /// Deterministic queue mutation for cross-crate replica consistency tests. - /// It deliberately skips provider dispatch so GPUI's test scheduler never - /// observes a real adapter thread. - #[doc(hidden)] - #[cfg(any(test, feature = "test-support"))] - pub fn queue_message_for_replica_test( - &mut self, - target_id: &str, - text: String, - _cx: &mut HostCx, - ) { - let Some(active) = self.resident_mut(target_id) else { - return; - }; - active.push_queued(text, Vec::new()); - } - pub(super) fn send_turn_assembled( &mut self, target_id: &str, @@ -214,7 +197,7 @@ impl AppState { ) { if self .resident(target_id) - .is_some_and(|active| active.meta.native_subagent.is_some()) + .is_some_and(|active| Self::conversation_read_only(&active.meta)) { log::warn!("refusing to send to a read-only native subagent mirror session"); return; @@ -642,7 +625,9 @@ impl AppState { // `steer` consumes the session's armed Ultrathink flag, but this // message captured its own at queue time — re-arm so it rides along. active.pending_ultrathink = message.ultrathink; + let command_key = std::mem::replace(&mut cx.delivery_key, message.delivery_key); self.steer_assembled(target_id, message.text, message.attachments, cx); + cx.delivery_key = command_key; self.reschedule_scheduled_wake(cx); } diff --git a/crates/runtime/src/app/sessions.rs b/crates/runtime/src/app/sessions.rs index 37086315d..00f3b1814 100644 --- a/crates/runtime/src/app/sessions.rs +++ b/crates/runtime/src/app/sessions.rs @@ -39,6 +39,7 @@ impl AppState { if self.subscriptions.insert(subscription.topic.clone()) { match &subscription.topic { Topic::SessionStatus { session_id } + | Topic::SessionPlan { session_id } | Topic::SessionEvents { session_id } | Topic::GitStatus { session_id } => self.select_session(session_id, cx), // No projection ran while nobody was attached, so the first @@ -60,12 +61,14 @@ impl AppState { self.subscriptions.remove(&subscription.topic); let session_id = match &subscription.topic { Topic::SessionStatus { session_id } + | Topic::SessionPlan { session_id } | Topic::SessionEvents { session_id } | Topic::GitStatus { session_id } => session_id, _ => return, }; if !self.subscriptions.iter().any(|topic| match topic { Topic::SessionStatus { session_id: id } + | Topic::SessionPlan { session_id: id } | Topic::SessionEvents { session_id: id } | Topic::GitStatus { session_id: id } => id == session_id, _ => false, @@ -403,7 +406,13 @@ impl AppState { /// project group. fn finish_external_import(&mut self, project_id: &str, cx: &mut HostCx) { match self.store.load_index() { - Ok(sessions) => self.sessions = sessions, + Ok(sessions) => { + let archived = self.archived_sessions(); + self.sessions = sessions; + if self.archived_sessions() != archived { + self.archived_revision += 1; + } + } Err(error) => self.report_error( RuntimeError::External(format!("could not reload the imported threads: {error}")), cx, @@ -779,10 +788,11 @@ impl AppState { }; let ids = descendant_session_ids(&self.sessions, session_id); for id in ids { - let Some(meta) = self + let Some(mut meta) = self .sessions - .iter_mut() + .iter() .find(|meta| meta.id == id && meta.archived_at == Some(archived_at)) + .cloned() else { continue; }; @@ -814,7 +824,7 @@ impl AppState { .collect(), unread: sessions .iter() - .filter(|meta| self.session_unread(&meta.id)) + .filter(|meta| self.session_unread(meta)) .map(|meta| meta.id.clone()) .collect(), active: self.residents.live.keys().cloned().collect(), @@ -852,13 +862,16 @@ impl AppState { for id in orchestrators { self.close_orchestrator_children(&id, cx); } - let mut changed = Vec::new(); - for meta in &mut self.sessions { - if ids.contains(meta.id.as_str()) { + let changed: Vec<_> = self + .sessions + .iter() + .filter(|meta| ids.contains(meta.id.as_str())) + .map(|meta| { + let mut meta = meta.clone(); meta.archived_at = Some(archived_at); - changed.push(meta.clone()); - } - } + meta + }) + .collect(); for meta in changed { self.persist_meta(&meta, cx); } @@ -873,7 +886,7 @@ impl AppState { if let Some(session) = self.resident_mut(session_id) { session.meta.title = title.to_string(); } - if let Some(meta) = self.sessions.iter_mut().find(|m| m.id == session_id) { + if let Some(mut meta) = self.sessions.iter().find(|m| m.id == session_id).cloned() { meta.title = title.to_string(); meta.updated_at = now_secs(); let meta = meta.clone(); @@ -881,44 +894,44 @@ impl AppState { } } + pub(super) fn session_fork_availability( + meta: &SessionMeta, + resident: Option<&ActiveSession>, + ) -> ForkAvailability { + if !meta.provider.caps().supports_fork { + ForkAvailability::Unsupported + } else if meta.resume_cursor.is_none() { + ForkAvailability::Empty + } else if resident.is_some_and(|session| session.turn_in_flight) { + ForkAvailability::Running + } else { + ForkAvailability::Available + } + } + /// Duplicate a stored transcript and arrange for its next provider start to /// fork the source's native session. The fork stays idle until its first /// user turn, exactly like a cold-opened stored thread. pub fn fork_thread(&mut self, id: &str, cx: &mut HostCx) -> Option { - let source = self - .resident(id) - .map(|session| (session.meta.clone(), session.turn_in_flight)) - .or_else(|| { - self.sessions - .iter() - .find(|meta| meta.id == id) - .cloned() - .map(|meta| (meta, false)) - }); - let (source, turn_in_flight) = source?; - if !source.provider.caps().supports_fork { - self.report_error( - RuntimeError::External("This provider does not support conversation forks.".into()), - cx, - ); - return None; - } - if source.resume_cursor.is_none() { - self.report_error( - RuntimeError::External("This conversation is empty and cannot be forked.".into()), - cx, - ); - return None; - } - if turn_in_flight { - self.report_error( - RuntimeError::External( - "Wait for the running turn to finish before forking this conversation.".into(), - ), - cx, - ); + let resident = self.resident(id); + let source = resident + .map(|session| &session.meta) + .or_else(|| self.sessions.iter().find(|meta| meta.id == id))?; + let error = match Self::session_fork_availability(source, resident) { + ForkAvailability::Unsupported => { + Some("This provider does not support conversation forks.") + } + ForkAvailability::Empty => Some("This conversation is empty and cannot be forked."), + ForkAvailability::Running => { + Some("Wait for the running turn to finish before forking this conversation.") + } + ForkAvailability::Available => None, + }; + if let Some(error) = error { + self.report_error(RuntimeError::External(error.into()), cx); return None; } + let source = source.clone(); let mut fork = SessionMeta::new(source.provider, source.cwd.clone(), source.model.clone()); fork.title = format!("{} (fork)", source.title); @@ -993,11 +1006,7 @@ impl AppState { self.close_orchestrator_children(session_id, cx); let worktree_remove = meta.as_ref().and_then(|meta| { let worktree = meta.worktree.as_ref().filter(|_| remove_worktree)?; - let shared = self - .sessions - .iter() - .chain(self.residents.live.values().map(|session| &session.meta)) - .any(|other| meta.shares_worktree_with(other)); + let shared = self.worktree_sharing().is_shared(meta); if shared { log::info!( "keeping worktree {} in use by another thread", @@ -1014,6 +1023,12 @@ impl AppState { self.enqueue_store_write(StoreWrite::RemoveSession(session_id.to_string()), cx); // Persist the pruned last-visited map (ignore save errors — cosmetic). self.persist_settings(cx); + if meta + .as_ref() + .is_some_and(|meta| meta.archived_at.is_some() || self.archived_sharing_affected(meta)) + { + self.archived_revision += 1; + } self.sessions.retain(|meta| meta.id != session_id); if let Some((root, cwd)) = worktree_remove { let deleted_id = session_id.to_string(); @@ -1120,17 +1135,13 @@ impl AppState { /// Whether a thread shows an unread dot: it has been visited before, its /// update time is newer than that visit, and it is not the active thread. - pub(crate) fn session_unread(&self, session_id: &str) -> bool { - if self.residents.live.contains_key(session_id) { - return false; - } - let Some(meta) = self.sessions.iter().find(|m| m.id == session_id) else { - return false; - }; - self.settings - .last_visited - .get(session_id) - .is_some_and(|&visited| meta.updated_at > visited) + pub(crate) fn session_unread(&self, meta: &SessionMeta) -> bool { + !self.residents.live.contains_key(&meta.id) + && self + .settings + .last_visited + .get(&meta.id) + .is_some_and(|&visited| meta.updated_at > visited) } /// Remove app-owned worktrees that no session in the loaded store owns or @@ -1196,6 +1207,12 @@ impl AppState { .unblock(move || provision(&root_for_task, &session_id_for_task)) .await; host_cx.enqueue(move |state, cx| { + if state + .resident(&target_id) + .is_some_and(|active| state.archived_sharing_affected(&active.meta)) + { + state.archived_revision += 1; + } let Some(active) = state .resident_mut(&target_id) .filter(|a| a.meta.id == session_id && a.draft) @@ -1362,6 +1379,9 @@ impl AppState { if let Some(mut parked) = self.residents.adopt(&session_id) { parked.idle_since = None; self.restore_terminal_workspace(&mut parked); + if self.archived_sharing_affected(&parked.meta) { + self.archived_revision += 1; + } self.residents.live.insert(session_id.clone(), parked); } self.refresh_git_status(&session_id, cx); @@ -1384,6 +1404,9 @@ impl AppState { let terminal_preferences = self.terminal_preferences_for(&draft); let restored_terminal = self.restore_terminal_workspace(&mut draft); let session_id = draft.meta.id.clone(); + if self.archived_sharing_affected(&draft.meta) { + self.archived_revision += 1; + } self.residents.live.insert(session_id.clone(), draft); if let Some(active) = self.resident(&session_id) { self.refresh_session_git_branch(active.meta.id.clone(), active.meta.cwd.clone(), cx); @@ -1581,6 +1604,9 @@ impl AppState { let terminal_preferences = self.terminal_preferences_for(&parked); let restored_terminal = self.restore_terminal_workspace(&mut parked); let needs_restart = matches!(parked.runtime, Runtime::Idle) && !parked.queue.is_empty(); + if parked.draft && self.archived_sharing_affected(&parked.meta) { + self.archived_revision += 1; + } self.residents.live.insert(session_id.to_string(), parked); self.schedule_timeline_load( session_id.to_string(), diff --git a/crates/runtime/src/app/snapshots.rs b/crates/runtime/src/app/snapshots.rs index e0b7dc438..d38c3b940 100644 --- a/crates/runtime/src/app/snapshots.rs +++ b/crates/runtime/src/app/snapshots.rs @@ -10,6 +10,7 @@ pub(crate) struct DomainDiff { providers: ProvidersStatus, git_status: HashMap, session_statuses: HashMap, + session_plans: HashMap, } impl DomainDiff { @@ -20,6 +21,7 @@ impl DomainDiff { providers: state.providers_status_snapshot(), git_status: HashMap::new(), session_statuses: state.resident_session_status_snapshots(), + session_plans: state.resident_session_plan_snapshots(), } } @@ -84,11 +86,29 @@ impl DomainDiff { Topic::SessionStatus { session_id: id.to_string(), }, - ServerEvent::SessionStatusReplaced(status), + ServerEvent::SessionStatusReplaced(Box::new(status)), cx, ); } self.session_statuses = session_statuses; + let session_plans = state.resident_session_plan_snapshots(); + let mut changed_ids: Vec<_> = session_plans + .iter() + .filter_map(|(id, plan)| { + (self.session_plans.get(id) != Some(plan)).then_some(id.as_str()) + }) + .collect(); + changed_ids.sort_unstable(); + for id in changed_ids { + emit_replacement( + Topic::SessionPlan { + session_id: id.to_string(), + }, + ServerEvent::SessionPlanReplaced(session_plans[id].clone()), + cx, + ); + } + self.session_plans = session_plans; } } @@ -165,31 +185,22 @@ impl AppState { pub fn index_snapshot(&self) -> IndexSnapshot { let mut summary = IndexSummary { title_generating: self.title_generating.clone(), - activity: self - .residents - .ids() - .filter_map(|id| { - let session = self.resident(id)?; - let (approvals, user_input) = self.open_requests(id, session); - Some(( - id.to_string(), - ( - session.has_work(), - !approvals.is_empty(), - user_input.is_some(), - session.background_task_count > 0 - && !session.turn_in_flight - && session.queue.is_empty(), - ), - )) - }) - .collect(), + archived_revision: self.archived_revision, ..IndexSummary::default() }; + let sharing = self.worktree_sharing(); let mut sessions = Vec::new(); for meta in &self.sessions { if meta.archived_at.is_none() { sessions.push(meta.clone()); + let resident = self.resident(&meta.id); + summary.activity.insert( + meta.id.clone(), + self.session_activity(resident.map_or(meta, |session| &session.meta), resident), + ); + if sharing.is_shared(meta) { + summary.worktree_shared.insert(meta.id.clone()); + } continue; } if let Some(project_id) = &meta.project_id { @@ -198,10 +209,19 @@ impl AppState { .entry(project_id.clone()) .or_default() += 1; } - if let Some(worktree) = &meta.worktree { - summary - .archived_worktree_branches - .insert(worktree.branch.clone()); + } + for id in self.residents.ids() { + let Some(session) = self.resident(id) else { + continue; + }; + if session.meta.archived_at.is_none() && !summary.activity.contains_key(id) { + summary.activity.insert( + id.to_string(), + self.session_activity(&session.meta, Some(session)), + ); + if sharing.is_shared(&session.meta) { + summary.worktree_shared.insert(id.to_string()); + } } } IndexSnapshot { @@ -211,15 +231,39 @@ impl AppState { } } - pub fn archived_sessions(&self) -> Vec { - let mut archived: Vec<_> = self + pub fn archived_sessions(&self) -> ArchivedSessions { + let sharing = self.worktree_sharing(); + let mut sessions: Vec<_> = self .sessions .iter() .filter(|meta| meta.archived_at.is_some()) .cloned() .collect(); - archived.sort_by_key(|meta| std::cmp::Reverse(meta.archived_at)); - archived + sessions.sort_by_key(|meta| std::cmp::Reverse(meta.archived_at)); + let worktree_shared = sessions + .iter() + .filter(|meta| sharing.is_shared(meta)) + .map(|meta| meta.id.clone()) + .collect(); + ArchivedSessions { + sessions, + worktree_shared, + revision: self.archived_revision, + } + } + + pub(super) fn archived_sharing_affected(&self, other: &SessionMeta) -> bool { + self.sessions + .iter() + .any(|meta| meta.archived_at.is_some() && meta.shares_worktree_with(other)) + } + + pub(super) fn worktree_sharing(&self) -> WorktreeSharing<'_> { + WorktreeSharing::new( + self.sessions + .iter() + .chain(self.residents.live.values().map(|session| &session.meta)), + ) } pub fn settings_snapshot(&self) -> Settings { @@ -278,8 +322,11 @@ impl AppState { Topic::GitStatus { session_id } => { ServerEvent::GitStatusReplaced(self.git_status_snapshot(session_id)) } - Topic::SessionStatus { session_id } => { - ServerEvent::SessionStatusReplaced(self.session_status_snapshot(session_id)?) + Topic::SessionStatus { session_id } => ServerEvent::SessionStatusReplaced(Box::new( + self.session_status_snapshot(session_id)?, + )), + Topic::SessionPlan { session_id } => { + ServerEvent::SessionPlanReplaced(self.session_plan_snapshot(session_id)?) } Topic::SessionEvents { .. } => self.session_events_snapshot(subscription), Topic::RuntimeEvents => return None, @@ -322,8 +369,54 @@ impl AppState { ) } - /// Build the complete non-event-stream status projection for one resident - /// session. This is the sole constructor for the replicated status domain. + pub(super) fn session_activity( + &self, + meta: &SessionMeta, + resident: Option<&ActiveSession>, + ) -> SessionActivity { + let (approvals, input) = resident.map_or((&[][..], None), |session| { + self.open_requests(&meta.id, session) + }); + SessionActivity { + working: resident.is_some_and(ActiveSession::has_work), + turn_running: resident.is_some_and(|session| session.turn_in_flight), + background_only: resident.is_some_and(|session| { + session.background_task_count > 0 + && !session.turn_in_flight + && session.delivery_in_flight.is_none() + && session.queue.is_empty() + }), + waiting_for_approval: !approvals.is_empty(), + waiting_for_input: input.is_some(), + unread: self.session_unread(meta), + fork: Self::session_fork_availability(meta, resident), + } + } + + pub fn session_plan_snapshot(&self, session_id: &str) -> Option { + let timeline = &self.resident(session_id)?.timeline; + Some(SessionPlan { + session_id: session_id.to_string(), + proposed: timeline + .shown_proposed_plan() + .map(|plan| ProposedPlanStatus { + item_id: plan.item_id.clone(), + turn: plan.turn, + markdown: plan.markdown.clone(), + ready: plan.ready, + resolved: timeline.plan_resolved(&plan.item_id), + }), + steps: timeline.plan_steps.clone(), + }) + } + + fn resident_session_plan_snapshots(&self) -> HashMap { + self.residents + .ids() + .filter_map(|id| Some((id.to_string(), self.session_plan_snapshot(id)?))) + .collect() + } + pub fn session_status_snapshot(&self, session_id: &str) -> Option { let session = self.resident(session_id)?; let meta = &session.meta; @@ -353,6 +446,24 @@ impl AppState { }); let terminal_preferences = self.terminal_preferences_for(session); let (approvals, user_input) = self.open_requests(session_id, session); + let (effective_approval_mode, native_approval_modes_enabled) = + session_approval_policy(meta, &session_provider_settings(meta, &self.settings)); + let context_window = session + .timeline + .usage + .and_then(|usage| usage.context_window) + .map(|reported| { + if meta.provider == ProviderKind::ClaudeCode + && let Some(model) = meta.model.as_deref() + { + reported.min(agent::claude::resolved_context_window( + model, + &meta.option_selections, + )) + } else { + reported + } + }); Some(SessionStatus { session_id: session_id.to_string(), title: meta.title.clone(), @@ -364,6 +475,8 @@ impl AppState { acp_agent_id: meta.acp_agent_id.clone(), project_id: meta.project_id.clone(), approval_mode: meta.approval_mode, + effective_approval_mode, + native_approval_modes_enabled, interaction_mode: meta.interaction_mode, queued_messages: session .queue @@ -371,6 +484,7 @@ impl AppState { .map(|message| QueuedMessageStatus { delivery_key: message.delivery_key.clone(), id: message.id, + editable: Self::queued_message_editable(session, message.id), text: message.text.clone(), fire_at_unix_secs: message.not_before.and_then(|time| { time.duration_since(UNIX_EPOCH) @@ -402,9 +516,15 @@ impl AppState { .map(|preferences| preferences.height.clamp(120., 600.)) .unwrap_or(240.), delivery_in_flight: session.delivery_in_flight, - turn_running: session.turn_in_flight, + activity: self.session_activity(meta, Some(session)), stopping: session.interrupt_requested, - working: session.has_work(), + native_rewind_blocked: self.native_rewind_blocked(session), + checkout_blocked: Self::checkout_blocked(session), + conversation_read_only: Self::conversation_read_only(meta), + terminal_limit_reached: self.terminal_limit_reached(session), + terminal_split_available: self.terminal_split_available(session), + usage: session.timeline.usage, + context_window, // A provider shut down without a closing record leaves its turn // running in the timeline; only an in-flight turn is live. running_turn: session @@ -447,6 +567,17 @@ impl AppState { } pub(super) fn upsert_session_in_memory(&mut self, meta: SessionMeta) { + let existing = self.sessions.iter().find(|existing| existing.id == meta.id); + let archived_metadata_changed = existing != Some(&meta) + && (meta.archived_at.is_some() + || existing.is_some_and(|old| old.archived_at.is_some())); + let sharing_may_change = existing + .is_none_or(|old| old.cwd != meta.cwd || old.worktree != meta.worktree) + && (self.archived_sharing_affected(&meta) + || existing.is_some_and(|old| self.archived_sharing_affected(old))); + if archived_metadata_changed || sharing_may_change { + self.archived_revision += 1; + } match self .sessions .iter_mut() diff --git a/crates/runtime/src/app/terminals.rs b/crates/runtime/src/app/terminals.rs index c86245bb2..fede5f593 100644 --- a/crates/runtime/src/app/terminals.rs +++ b/crates/runtime/src/app/terminals.rs @@ -508,15 +508,31 @@ impl AppState { ); } - pub fn new_terminal(&mut self, target_id: &str, cx: &mut HostCx) { - let Some(active) = self.resident(target_id) else { - return; - }; + pub(super) fn terminal_limit_reached(&self, active: &ActiveSession) -> bool { let pending = self .pending_terminal_spawns .get(&active.meta.id) .map_or(0, HashMap::len); - if active.terminal_workspace.terminals.len() + pending >= MAX_TERMINALS_PER_SESSION { + active.terminal_workspace.terminals.len() + pending >= MAX_TERMINALS_PER_SESSION + } + + pub(super) fn terminal_split_available(&self, active: &ActiveSession) -> bool { + !self.terminal_limit_reached(active) + && active.terminal_workspace.active_id.is_some_and(|first| { + active.terminal_workspace.split_for(first).is_none() + && !self.pending_terminal_spawns.get(&active.meta.id).is_some_and(|spawns| { + spawns.values().any(|action| { + matches!(action, TerminalSpawnAction::Split { first: pending, .. } if *pending == first) + }) + }) + }) + } + + pub fn new_terminal(&mut self, target_id: &str, cx: &mut HostCx) { + let Some(active) = self.resident(target_id) else { + return; + }; + if self.terminal_limit_reached(active) { return; } let cwd = self.spawn_cwd(target_id); @@ -592,27 +608,7 @@ impl AppState { let Some(first) = workspace.active_id else { return; }; - let pending = self - .pending_terminal_spawns - .get(&active.meta.id) - .map_or(0, HashMap::len); - if workspace.terminals.len() + pending >= MAX_TERMINALS_PER_SESSION - || workspace.split_for(first).is_some() - || self - .pending_terminal_spawns - .get(&active.meta.id) - .is_some_and(|spawns| { - spawns.values().any(|action| { - matches!( - action, - TerminalSpawnAction::Split { - first: pending_first, - .. - } if *pending_first == first - ) - }) - }) - { + if !self.terminal_split_available(active) { return; } let session_id = active.meta.id.clone(); diff --git a/crates/runtime/src/app/tests.rs b/crates/runtime/src/app/tests.rs index f7ac0dae0..2af36f2fe 100644 --- a/crates/runtime/src/app/tests.rs +++ b/crates/runtime/src/app/tests.rs @@ -3,7 +3,7 @@ use super::*; use super::{active_session::*, events::*, orchestrate::*, providers::*}; use tcode_core::settings::{SettingsPatch, ThemeMode}; -use tcode_protocol::{Command, CommandResponse, HostMessage}; +use tcode_protocol::{Command, CommandResponse, HostMessage, Subscription}; #[test] fn permission_relaunch_marker_requires_screen_access_only_for_computer_use() { for screen_recording in [false, true] { @@ -362,7 +362,13 @@ fn assert_native_mirror_turn_lifecycle(evict: bool, late: bool, end: MirrorEnd, assert_eq!(turn.status, Some(expected_status)); assert!(!mirror.turn_in_flight); assert!(!mirror.has_work()); - assert!(!state.session_status_snapshot(&mirror_id).unwrap().working); + assert!( + !state + .session_status_snapshot(&mirror_id) + .unwrap() + .activity + .working + ); assert_eq!( mirror .timeline @@ -2764,8 +2770,22 @@ fn only_a_loaded_conversation_marks_a_thread_read() { state.update(cx, |state, _| { state.settings.last_visited.insert(id.clone(), 50); }); - let unread = |state: &TestEntity| state.read(|state| state.session_unread(&id)); + // Read through the index projection every client shows. + let unread = |state: &TestEntity| { + state.read(|state| state.index_snapshot().summary.activity[&id].unread) + }; assert!(unread(&state), "updated after the last visit"); + state.read(|state| { + let activity = &state.index_snapshot().summary.activity[&id]; + assert!( + !activity.working + && !activity.turn_running + && !activity.background_only + && !activity.waiting_for_approval + && !activity.waiting_for_input, + "a thread with no live provider has no work" + ); + }); let subscription = Subscription { topic: Topic::SessionEvents { @@ -4304,10 +4324,12 @@ fn steering_user_and_queue_paths_send_the_same_id_they_record() { let sentinel = active.push_queued("queued sentinel".into(), Vec::new()); state.install_selected(active); + cx.delivery_key = Some("steer-send".into()); state.steer("active", "redirect".into(), Vec::new(), cx); let SessionCommand::Steer { request_id, .. } = receiver.try_recv().unwrap() else { panic!("user steer command missing") }; + assert_eq!(request_id, "local-steer-steer-send"); let active = state.selected_session().unwrap(); assert!(active.turn_in_flight); assert_eq!(active.queue.len(), 1); @@ -4327,10 +4349,19 @@ fn steering_user_and_queue_paths_send_the_same_id_they_record() { let first = active.push_queued("first".into(), Vec::new()); let queued_id = active.push_queued("queued redirect".into(), Vec::new()); let third = active.push_queued("third".into(), Vec::new()); + active + .queue + .iter_mut() + .find(|message| message.id == queued_id) + .unwrap() + .delivery_key = Some("queue-send".into()); + cx.delivery_key = Some("queue-steer-click".into()); state.steer_queued("active", queued_id, cx); let SessionCommand::Steer { request_id, .. } = receiver.try_recv().unwrap() else { panic!("queue-to-steer command missing") }; + assert_eq!(request_id, "local-steer-queue-send"); + assert_eq!(cx.delivery_key.as_deref(), Some("queue-steer-click")); let active = state.selected_session().unwrap(); assert_eq!( active @@ -4508,12 +4539,18 @@ fn schedule_status_and_queue_actions_preserve_or_remove_deadlines() { let status = state.session_status_snapshot("scheduled-active").unwrap(); assert_eq!(status.queued_messages.len(), 1); assert_eq!(status.queued_messages[0].fire_at_unix_secs, Some(fire_at)); + assert!(status.queued_messages[0].editable); let scheduled_id = status.queued_messages[0].id; state.steer_queued("scheduled-active", scheduled_id, cx); let status = state.session_status_snapshot("scheduled-active").unwrap(); assert_eq!(status.queued_messages.len(), 1); assert_eq!(status.queued_messages[0].fire_at_unix_secs, None); + assert!(!status.queued_messages[0].editable); + assert_eq!( + status.delivery_in_flight, + Some(status.queued_messages[0].id) + ); assert!(matches!( receiver.try_recv(), Ok(SessionCommand::SendTurn { .. }) @@ -5790,6 +5827,26 @@ fn fork_thread_clones_timeline_and_provider_cursor() { assert!(store.read_event_log(&empty_source.id).unwrap().is_empty()); let state = cx.new_entity(TestClientState::new(store)); + state.update(cx, |state, cx| { + assert_eq!( + state.index_snapshot().summary.activity[&source.id].fork, + ForkAvailability::Available + ); + let mut resident = ActiveSession::new(source.clone(), false, Vec::new()); + resident.turn_in_flight = true; + state.install_selected(resident); + assert_eq!( + state.index_snapshot().summary.activity[&source.id].fork, + ForkAvailability::Running + ); + assert!(state.host.fork_thread(&source.id, cx).is_none()); + state.selected_session_mut().unwrap().turn_in_flight = false; + state.selected_session_mut().unwrap().background_task_count = 1; + assert_eq!( + state.index_snapshot().summary.activity[&source.id].fork, + ForkAvailability::Available + ); + }); state.update(cx, |state, cx| state.fork_thread(&source.id, cx)); cx.run_until(|state| { state @@ -5808,6 +5865,13 @@ fn fork_thread_clones_timeline_and_provider_cursor() { ); assert_eq!(fork.cwd, source.cwd); assert_eq!(fork.worktree, None); + assert!( + state + .index_snapshot() + .summary + .worktree_shared + .contains(&source.id) + ); assert!(!active.timeline.turn_running); assert_eq!(state.store.read_events(&fork.id).unwrap().len(), 1); assert_eq!( @@ -6560,6 +6624,7 @@ fn turn_running_for_is_independent_of_active_or_parked_location() { let mut delivery = live_session(ProviderKind::ClaudeCode, commands.clone()); delivery.meta.id = "delivery".into(); delivery.delivery_in_flight = Some(7); + delivery.background_task_count = 1; let mut queued = live_session(ProviderKind::ClaudeCode, commands.clone()); queued.meta.id = "queued".into(); @@ -6588,10 +6653,17 @@ fn turn_running_for_is_independent_of_active_or_parked_location() { ("stale timeline", stale_timeline, false), ] { let id = session.meta.id.clone(); + let expected_background = label == "background"; + state.sessions.push(session.meta.clone()); state.install_selected(session); + let index = state.index_snapshot(); + assert_eq!( + index.summary.activity[&id].background_only, expected_background, + "{label}" + ); let active_answer = state.turn_running_for(&id); assert_eq!( - state.session_status_snapshot(&id).unwrap().working, + state.session_status_snapshot(&id).unwrap().activity.working, expected, "{label}" ); @@ -6600,7 +6672,7 @@ fn turn_running_for_is_independent_of_active_or_parked_location() { state.residents.parked.insert(id.clone(), parked); let parked_answer = state.turn_running_for(&id); assert_eq!( - state.session_status_snapshot(&id).unwrap().working, + state.session_status_snapshot(&id).unwrap().activity.working, expected, "{label}" ); @@ -7397,6 +7469,188 @@ fn mux_clients_target_independent_drafts_and_receive_only_their_session_tail() { host.stopped.recv_blocking().unwrap(); } +#[test] +fn archived_revision_invalidates_sharing_facts_when_another_thread_or_draft_leaves() { + let cx = &mut TestAppContext::default(); + let store = TestStore::new("archived-worktree-sharing"); + let project = Project::from_root(store.root().join("project")); + store.upsert_project(&project).unwrap(); + let mut owner = SessionMeta::new(ProviderKind::Codex, store.root().join("owner"), None); + owner.id = "owner".into(); + owner.archived_at = Some(1); + owner.worktree = Some(WorktreeInfo { + root_project_path: project.root.clone(), + base: "main".into(), + branch: "tcode/owner".into(), + }); + store.upsert_meta(&owner).unwrap(); + let mut fork = SessionMeta::new(ProviderKind::Codex, owner.cwd.join("nested"), None); + fork.id = "fork".into(); + store.upsert_meta(&fork).unwrap(); + let state = cx.new_entity(TestClientState::new((*store).clone())); + state.update(cx, |state, cx| { + let before = state.archived_sessions(); + assert!(before.worktree_shared.contains("owner")); + state.delete_session("fork", false, cx); + let after = state.archived_sessions(); + assert!(after.revision > before.revision); + assert!(!after.worktree_shared.contains("owner")); + let draft = state + .host + .start_draft(project.id.clone(), owner.cwd.clone(), cx); + let shared = state.archived_sessions(); + assert!(shared.revision > after.revision); + assert!(shared.worktree_shared.contains("owner")); + state.shutdown_active(&draft, cx); + let unshared = state.archived_sessions(); + assert!(unshared.revision > shared.revision); + assert!(!unshared.worktree_shared.contains("owner")); + }); +} + +#[test] +fn plan_and_usage_projections_survive_a_partial_history_window_and_emit_only_changes() { + let cx = &mut TestAppContext::default(); + let store = TestStore::new("session-plan-usage-window"); + let state = cx.new_entity(TestClientState::new((*store).clone())); + let id = "plan-usage"; + state.update(cx, |state, cx| { + let mut meta = SessionMeta::new( + ProviderKind::ClaudeCode, + store.root().clone(), + Some("claude-sonnet-4-6".into()), + ); + meta.id = id.into(); + meta.option_selections = vec![OptionSelection { + id: "contextWindow".into(), + value: serde_json::json!(500_000), + }]; + state.install_selected(ActiveSession::new(meta, false, Vec::new())); + state.on_event( + id, + AgentEvent::TurnStarted { + turn_id: "first".into(), + }, + cx, + ); + state.on_event( + id, + AgentEvent::TokenUsage(agent::TokenUsage { + context_window: Some(1_000_000), + used_tokens: Some(1234), + total_processed_tokens: Some(9876), + freshness: agent::ContextFreshness::Current, + ..Default::default() + }), + cx, + ); + state.on_event( + id, + AgentEvent::ProposedPlan { + item_id: "plan-item".into(), + markdown: "# Keep the plan".into(), + }, + cx, + ); + state.on_event( + id, + AgentEvent::TurnCompleted { + turn_id: "first".into(), + status: TurnStatus::Completed, + usage: None, + }, + cx, + ); + state.on_event( + id, + AgentEvent::TurnStarted { + turn_id: "second".into(), + }, + cx, + ); + for index in 0..401 { + state.on_event( + id, + AgentEvent::Warning { + message: format!("later record {index}"), + }, + cx, + ); + } + let snapshot = state + .subscription_snapshot(&Subscription { + topic: Topic::SessionEvents { + session_id: id.into(), + }, + after: None, + }) + .unwrap(); + let ServerEvent::SessionSnapshot { from, records, .. } = snapshot.event else { + panic!("history snapshot") + }; + assert!(from > 0); + let partial = Timeline::fold_events(records); + assert!(partial.usage.is_none()); + assert!(partial.proposed_plan.is_none()); + let status = state.session_status_snapshot(id).unwrap(); + let usage = status.usage.unwrap(); + assert_eq!(usage.used_tokens, Some(1234)); + assert_eq!(usage.total_processed_tokens, Some(9876)); + assert_eq!(usage.context_window, Some(1_000_000)); + assert_eq!(status.context_window, Some(500_000)); + let plan = state + .subscription_snapshot(&Subscription { + topic: Topic::SessionPlan { + session_id: id.into(), + }, + after: None, + }) + .unwrap(); + let ServerEvent::SessionPlanReplaced(plan) = plan.event else { + panic!("plan baseline") + }; + let proposed = plan.proposed.unwrap(); + assert_eq!(proposed.markdown, "# Keep the plan"); + assert_eq!(proposed.turn, 0); + assert!(proposed.ready && !proposed.resolved); + }); + cx.run_until_parked(); + let plans = |messages: Vec| { + messages + .into_iter() + .filter_map(|message| match message { + HostMessage::Event(EventEnvelope { + event: ServerEvent::SessionPlanReplaced(plan), + .. + }) => Some(plan), + _ => None, + }) + .collect::>() + }; + assert_eq!(plans(cx.drain_outgoing()).len(), 1); + state.dispatch_command( + cx, + 88, + Command::RenameSession { + session_id: id.into(), + title: "Unrelated metadata".into(), + }, + ); + cx.run_until_parked(); + assert!(plans(cx.drain_outgoing()).is_empty()); + state.dispatch_command( + cx, + 89, + Command::DismissPlan { + session_id: id.into(), + }, + ); + cx.run_until_parked(); + let changed = plans(cx.drain_outgoing()); + assert_eq!(changed.len(), 1); + assert!(changed[0].proposed.as_ref().unwrap().resolved); +} + #[test] fn session_history_snapshot_pages_and_absolute_tail_cursors() { let cx = &mut TestAppContext::default(); @@ -8165,7 +8419,7 @@ fn interrupt_reports_stopping_until_the_turn_completes() { ); let status = state.session_status_snapshot(&id).unwrap(); assert!(!status.stopping); - assert!(!status.turn_running); + assert!(!status.activity.turn_running); } #[test] @@ -8368,6 +8622,7 @@ fn index_and_visit_changes_cross_the_wire_one_thread_at_a_time() { assert_eq!( state .archived_sessions() + .sessions .iter() .map(|m| m.id.as_str()) .collect::>(), @@ -8375,9 +8630,59 @@ fn index_and_visit_changes_cross_the_wire_one_thread_at_a_time() { ); }); + let revision = state.read(|state| state.index_snapshot().summary.archived_revision); state.dispatch_command( cx, 2, + Command::RenameSession { + session_id: "archived".into(), + title: "Archived rename".into(), + }, + ); + cx.run_until_parked(); + let summary_revision = cx + .drain_outgoing() + .into_iter() + .find_map(|message| match message { + HostMessage::Event(EventEnvelope { + event: ServerEvent::IndexSummaryReplaced(summary), + .. + }) => Some(summary.archived_revision), + _ => None, + }) + .expect("archived rename must invalidate the held list"); + assert!(summary_revision > revision); + state.update(cx, |state, cx| { + crate::pipe::handle_client_message( + state, + cx, + tcode_protocol::ClientMessage { + id: 3, + key: None, + payload: tcode_protocol::ClientPayload::Query( + tcode_protocol::Query::ArchivedSessions, + ), + }, + ) + }); + cx.run_until_parked(); + let archived = cx + .drain_outgoing() + .into_iter() + .find_map(|message| match message { + HostMessage::QueryResult { + id: 3, + result: Ok(QueryResponse::ArchivedSessions(archived)), + } => Some(archived), + _ => None, + }) + .expect("archived list reply"); + assert_eq!(archived.revision, summary_revision); + assert_eq!(archived.sessions[0].title, "Archived rename"); + + state.dispatch_command( + cx, + 4, Command::MarkSessionRead { session_id: "kept".into(), through: 1, diff --git a/crates/runtime/src/pipe.rs b/crates/runtime/src/pipe.rs index 17a63ac6a..32bb3a6b5 100644 --- a/crates/runtime/src/pipe.rs +++ b/crates/runtime/src/pipe.rs @@ -833,8 +833,8 @@ fn dispatch_query( cx.spawn_background(async move { result }) } Query::ArchivedSessions => { - let sessions = app.archived_sessions(); - cx.spawn_background(async move { Ok(QueryResponse::ArchivedSessions(sessions)) }) + let archived = app.archived_sessions(); + cx.spawn_background(async move { Ok(QueryResponse::ArchivedSessions(archived)) }) } Query::RenderStoredOutput { session_id, diff --git a/crates/runtime/src/pipe_p4b_tests.rs b/crates/runtime/src/pipe_p4b_tests.rs index 5a7afeec8..43b08baa9 100644 --- a/crates/runtime/src/pipe_p4b_tests.rs +++ b/crates/runtime/src/pipe_p4b_tests.rs @@ -402,7 +402,33 @@ fn import_status_survives_clients_and_gates_new_runs_by_lifecycle() { #[test] fn import_finalizes_the_index_before_finished_even_after_the_initiator_disconnects() { - let (_host, mux, watcher, root, project_id) = project_fixture(); + let (host, mux, watcher, root, project_id) = project_fixture(); + let mut owner = tcode_core::project::SessionMeta::new( + agent::ProviderKind::ClaudeCode, + root.join("project"), + None, + ); + owner.id = "archived-owner".into(); + owner.archived_at = Some(1); + owner.worktree = Some(tcode_core::project::WorktreeInfo { + root_project_path: root.join("repo"), + base: "main".into(), + branch: "tcode/owner".into(), + }); + SessionStore::open_at(root.clone()) + .unwrap() + .upsert_meta(&owner) + .unwrap(); + smol::block_on(host.update_state_for_test(move |state, _cx| { + state.sessions.push(owner); + })) + .unwrap(); + let QueryResponse::ArchivedSessions(before) = + smol::block_on(watcher.query(Query::ArchivedSessions)).unwrap() + else { + panic!("expected archived sessions"); + }; + assert!(!before.worktree_shared.contains("archived-owner")); let history = root.join("history"); let threads = vec![ claude_thread(&history, "thread-a", "first imported conversation"), @@ -486,6 +512,13 @@ fn import_finalizes_the_index_before_finished_even_after_the_initiator_disconnec _ => {} } } + let QueryResponse::ArchivedSessions(after) = + smol::block_on(watcher.query(Query::ArchivedSessions)).unwrap() + else { + panic!("expected archived sessions"); + }; + assert!(after.revision > before.revision); + assert!(after.worktree_shared.contains("archived-owner")); watcher .command_blocking(Command::ShutdownAllAndFlush) .unwrap(); diff --git a/crates/ui/src/chat/components/bubble.rs b/crates/ui/src/chat/components/bubble.rs index d4133aa0d..0699fd1d7 100644 --- a/crates/ui/src/chat/components/bubble.rs +++ b/crates/ui/src/chat/components/bubble.rs @@ -57,6 +57,7 @@ pub(crate) struct BubbleData<'a> { pub(crate) struct RewindMenu { pub(crate) turn: usize, pub(crate) disabled: bool, + pub(crate) conversation_available: bool, } /// The bubble's context-menu items: copy the message, then the rewind modes @@ -69,11 +70,16 @@ fn bubble_menu_items(copy_text: Arc, rewind: Option) -> MenuExt } else { assistant::copy_message_items(copy_text.clone())(menu, window, cx) }; - let Some(RewindMenu { turn, disabled }) = rewind else { + let Some(RewindMenu { + turn, + disabled, + conversation_available, + }) = rewind + else { return menu; }; let mut modes = Vec::new(); - if turn > 0 { + if conversation_available { modes.push(( crate::tr!("chat.rewind_all").into_owned(), RewindMode::FilesAndConversation, @@ -108,6 +114,7 @@ pub(crate) struct RewindHandlers { pub(crate) fn native_rewind_button( turn: usize, + conversation_available: bool, (state, compact): (Option<(bool, bool)>, bool), handlers: RewindHandlers, cx: &App, @@ -146,7 +153,7 @@ pub(crate) fn native_rewind_button( files, } = &handlers; let mut modes = Vec::new(); - if turn > 0 { + if conversation_available { modes.push(( crate::tr!("chat.rewind_all").into_owned(), files_and_conversation.clone(), diff --git a/crates/ui/src/chat/mod.rs b/crates/ui/src/chat/mod.rs index 57c38eb37..36aec071d 100644 --- a/crates/ui/src/chat/mod.rs +++ b/crates/ui/src/chat/mod.rs @@ -737,7 +737,8 @@ impl ChatView { } else { TimelineContinuity::Complete }; - let (running, list_sync, activity_snapshot_keys) = self + let running = self.workspace_store.read(cx).composer_state().turn_running; + let (list_sync, activity_snapshot_keys) = self .workspace_store .read(cx) .with_active_timeline(|timeline| { @@ -753,7 +754,7 @@ impl ChatView { ); let activity_snapshot_keys = awaiting_activity_snapshot .then(|| auto_activity_snapshot_keys(&timeline.entries)); - (timeline.turn_running, list_sync, activity_snapshot_keys) + (list_sync, activity_snapshot_keys) }) .unwrap_or_else(|| { let list_sync = self.turn_index_cache.sync( @@ -764,7 +765,7 @@ impl ChatView { &self.expanded, continuity, ); - (false, list_sync, None) + (list_sync, None) }); if let Some(session_key) = session_key.as_deref() @@ -962,6 +963,7 @@ impl ChatView { one_shot_row_target: Option, cx: &mut Context, ) { + let running = self.workspace_store.read(cx).composer_state().turn_running; let row_count = self.rows.len(); // Auto-scroll can move many rows during a drag. Do not retire any // participant until mouse-up; completed-selection participants remain @@ -988,7 +990,7 @@ impl ChatView { row_count, self.markdown_visible_rows.clone(), one_shot_row_target, - timeline.turn_running, + running, ); let entries = markdown_entries_for_residency(timeline, &self.rows, &scope); let decisions = decide(ResidencyInput { @@ -996,7 +998,7 @@ impl ChatView { visible_rows: self.markdown_visible_rows.clone(), one_shot_row_target, entries: &entries, - stream_running: timeline.turn_running, + stream_running: running, resident_ids: &resident_ids, selection_participants: &selection_participants, selection_drag_active, @@ -1731,7 +1733,11 @@ impl ChatView { .then(|| self.workspace_store.read(cx).chat_native_rewind_state(turn)) .flatten() .filter(|(available, _)| *available) - .map(|(_, disabled)| components::bubble::RewindMenu { turn, disabled }); + .map(|(_, disabled)| components::bubble::RewindMenu { + turn, + disabled, + conversation_available: self.workspace_store.read(cx).absolute_turn(turn) > 0, + }); let rewind = steering .is_none() .then(|| { @@ -1743,6 +1749,7 @@ impl ChatView { }; components::bubble::native_rewind_button( turn, + self.workspace_store.read(cx).absolute_turn(turn) > 0, ( self.workspace_store.read(cx).chat_native_rewind_state(turn), self.window_state.read(cx).compact, diff --git a/crates/ui/src/composer/components/checkout.rs b/crates/ui/src/composer/components/checkout.rs index 132189143..89b7ee643 100644 --- a/crates/ui/src/composer/components/checkout.rs +++ b/crates/ui/src/composer/components/checkout.rs @@ -8,7 +8,7 @@ impl Composer { let checkout = self.workspace_store.read(cx).composer_state().checkout?; let branch = checkout.branch; let branches: Rc<[String]> = checkout.branches.into(); - let turn_running = checkout.turn_running; + let checkout_blocked = checkout.checkout_blocked; let is_draft = checkout.is_draft; let worktree_base = checkout.worktree_base; let worktree = checkout.worktree; @@ -18,13 +18,12 @@ impl Composer { let picker_current = worktree_base.clone().unwrap_or_else(|| branch.clone()); let worktree_mode = worktree_base.is_some(); - // The branch chip: a popover listing local branches. While a turn runs - // the selector is disabled (it just shows a "wait" tooltip). - let right: AnyElement = if turn_running { + let right: AnyElement = if checkout_blocked { Button::new("branch-picker") .ghost() .outline() .compact() + .disabled(true) .tooltip(crate::tr!("composer.wait_turn")) .child( h_flex() diff --git a/crates/ui/src/composer/components/fallback.rs b/crates/ui/src/composer/components/fallback.rs index 8e37b09bf..e2b27f41f 100644 --- a/crates/ui/src/composer/components/fallback.rs +++ b/crates/ui/src/composer/components/fallback.rs @@ -53,10 +53,12 @@ impl Composer { // rewind path additionally needs its turn to have a provider checkpoint. let refused = self.workspace_store.read(cx).last_user_message(); let can_rewind = refused.as_ref().is_some_and(|(turn, _)| { - self.workspace_store - .read(cx) - .chat_native_rewind_state(*turn) - .is_some_and(|(available, disabled)| available && !disabled) + self.workspace_store.read(cx).absolute_turn(*turn) > 0 + && self + .workspace_store + .read(cx) + .chat_native_rewind_state(*turn) + .is_some_and(|(available, disabled)| available && !disabled) }); let header = h_flex() diff --git a/crates/ui/src/composer/components/pickers.rs b/crates/ui/src/composer/components/pickers.rs index f9d8c3b30..a8a03223c 100644 --- a/crates/ui/src/composer/components/pickers.rs +++ b/crates/ui/src/composer/components/pickers.rs @@ -201,9 +201,10 @@ impl Composer { let trigger = Button::new("model-picker") .debug_selector(|| "model-picker".into()) - .when(self.compact || store.native_subagent_readonly(), |button| { - button.w_full().max_w(px(160.)).min_w_0().overflow_hidden() - }) + .when( + self.compact || composer_state.conversation_read_only, + |button| button.w_full().max_w(px(160.)).min_w_0().overflow_hidden(), + ) .ghost() .compact() .h(px(28.)) @@ -212,18 +213,20 @@ impl Composer { .rounded(crate::material::radius_input(cx)) .child( h_flex() - .when(self.compact || store.native_subagent_readonly(), |el| { - el.w_full().min_w_0().overflow_hidden() - }) + .when( + self.compact || composer_state.conversation_read_only, + |el| el.w_full().min_w_0().overflow_hidden(), + ) .gap_1p5() .items_center() .text_size(px(13.)) .child(tinted_provider_glyph(provider, store).small()) .child( div() - .when(self.compact || store.native_subagent_readonly(), |el| { - el.min_w_0().truncate() - }) + .when( + self.compact || composer_state.conversation_read_only, + |el| el.min_w_0().truncate(), + ) .font_medium() .child(display), ) @@ -280,11 +283,8 @@ impl Composer { pub(in super::super) fn render_traits_picker(&self, cx: &mut Context) -> AnyElement { let store = self.workspace_store.read(cx); let composer = store.composer_state(); - // Native providers describe their options through the model catalog; ACP - // agents push theirs over the wire (`AgentEvent::ProviderOptions`). Both - // arrive as `OptionDescriptor`s and render through this one picker. let descriptors = composer.active_option_descriptors.clone(); - if store.native_subagent_readonly() { + if composer.conversation_read_only { let effort = option_selection_str(&composer.active_option_selections, "reasoningEffort"); let descriptor = descriptors.iter().find(|option| { @@ -320,14 +320,14 @@ impl Composer { if descriptors.is_empty() { return div().into_any_element(); } - let spec = match composer.active_model_spec.clone() { - Some(spec) => spec, - None => ModelSpec { + let spec = ModelSpec { + options: descriptors, + ..composer.active_model_spec.clone().unwrap_or(ModelSpec { id: String::new(), display_name: String::new(), is_default: false, - options: descriptors, - }, + options: Vec::new(), + }) }; let selections = composer.active_option_selections; let ultrathink_armed = composer.ultrathink_armed; @@ -505,7 +505,7 @@ impl Composer { /// description, ✓ on the current one). pub(in super::super) fn render_permission_picker(&self, cx: &mut Context) -> AnyElement { let composer = self.workspace_store.read(cx).composer_state(); - let current = composer.approval_mode; + let current = composer.effective_approval_mode; let native_approval_modes_enabled = composer.native_approval_modes_enabled; let (label, icon_path) = approval_mode_meta(current); let muted = cx.theme().muted_foreground; @@ -618,7 +618,7 @@ impl Composer { let composer = self.workspace_store.read(cx).composer_state(); let usage = composer.token_usage; let muted = cx.theme().muted_foreground; - let mode = composer.approval_mode; + let mode = composer.effective_approval_mode; let interaction = composer.interaction_mode; let store_entity = self.workspace_store.clone(); @@ -993,7 +993,7 @@ fn render_compact_model_footer( footer = footer.child(group(label.into(), track, cx)); } - let current = composer_state.approval_mode; + let current = composer_state.effective_approval_mode; let enabled = composer_state.native_approval_modes_enabled; let segments = APPROVAL_MODES .iter() @@ -1405,7 +1405,6 @@ fn render_traits_pane( } let options = &options; let is_reasoning = id == "reasoningEffort"; - let is_context_window = id == "contextWindow"; pane = pane.child(section_header(label, cx)); if is_reasoning && locked { pane = pane.child( @@ -1419,17 +1418,30 @@ fn render_traits_pane( ); continue; } - let resolved = (!is_context_window) - .then(|| resolved_select_value(id, options, default_value, selections)) + let resolved = resolved_select_value(id, options, default_value, selections); + let resolved_window = (id == "contextWindow") + .then(|| { + selections + .iter() + .find(|selection| selection.id == *id) + .and_then(|selection| { + agent::claude::parse_context_window_tokens(&selection.value) + }) + .or_else(|| { + default_value.as_ref().and_then(|value| { + agent::claude::parse_context_window_tokens(&serde_json::json!( + value + )) + }) + }) + }) .flatten(); - let resolved_window = is_context_window - .then(|| agent::claude::resolved_context_window(&spec.id, selections)); for (index, opt) in options.iter().enumerate() { let is_default = default_value.as_deref() == Some(opt.value.as_str()); let is_ultra = is_reasoning && opt.value == "ultrathink"; - let is_selected = if let Some(resolved_window) = resolved_window { + let is_selected = if let Some(window) = resolved_window { agent::claude::parse_context_window_tokens(&serde_json::json!(opt.value)) - == Some(resolved_window) + == Some(window) } else if is_reasoning && ultrathink_armed { is_ultra } else if is_ultra { @@ -1480,17 +1492,19 @@ fn render_traits_pane( }), ); } - if let Some(resolved_window) = resolved_window { - let preset_selected = options.iter().any(|opt| { - agent::claude::parse_context_window_tokens(&serde_json::json!(opt.value)) - == Some(resolved_window) + if id == "contextWindow" { + let custom_selected = resolved_window.is_some_and(|window| { + !options.iter().any(|opt| { + agent::claude::parse_context_window_tokens(&serde_json::json!( + opt.value + )) == Some(window) + }) }); - let custom_selected = !preset_selected; let mut label = crate::tr!("composer.context_window_custom").into_owned(); if custom_selected { label.push_str(&format!( " ({})", - agent::claude::format_context_window(resolved_window) + agent::claude::format_context_window(resolved_window.unwrap()) )); } let input = context_window_custom.clone(); diff --git a/crates/ui/src/composer/components/plan.rs b/crates/ui/src/composer/components/plan.rs index d14e17f65..f500c7c76 100644 --- a/crates/ui/src/composer/components/plan.rs +++ b/crates/ui/src/composer/components/plan.rs @@ -82,6 +82,7 @@ impl Composer { .child( h_flex() .id("implement-main") + .debug_selector(|| "implement-main".into()) .h_full() .px_3() .items_center() diff --git a/crates/ui/src/composer/components/queue_strip.rs b/crates/ui/src/composer/components/queue_strip.rs index 8ce06d615..3a09cd76d 100644 --- a/crates/ui/src/composer/components/queue_strip.rs +++ b/crates/ui/src/composer/components/queue_strip.rs @@ -34,11 +34,10 @@ impl Composer { .messages .into_iter() .filter(|message| { - !deliveries.iter().any(|(key, text, failure, acknowledged)| { + !deliveries.iter().any(|(key, _, failure, acknowledged)| { !acknowledged && failure.is_none() - && (message.delivery_key.as_deref() == Some(key.as_str()) - || (message.delivery_key.is_none() && message.text == *text)) + && message.delivery_key.as_deref() == Some(key.as_str()) }) }) .collect(); @@ -77,6 +76,7 @@ impl Composer { for (index, message) in queued.into_iter().enumerate() { let id = message.id; + let editable = message.editable; let text = message.text.clone(); let scheduled = message.fire_at_unix_secs.is_some(); let steer_tooltip = if scheduled { @@ -123,6 +123,7 @@ impl Composer { }) .child( Button::new(("queue-steer", id as usize)) + .debug_selector(move || format!("queue-steer-{id}")) .ghost() .xsmall() .when(self.compact, |button| button.min_w(px(44.)).min_h(px(44.))) @@ -130,7 +131,9 @@ impl Composer { // Scheduled rows always support send-now: the // runtime removes the deadline and uses the normal // send/queue path even when native steering is absent. - .disabled(!self.interactive(cx) || (!scheduled && !can_steer)) + .disabled( + !editable || !self.interactive(cx) || (!scheduled && !can_steer), + ) .tooltip(steer_tooltip) .on_click(cx.listener(move |this, _, _, cx| { this.workspace_store @@ -139,11 +142,12 @@ impl Composer { ) .child( Button::new(("queue-drop", id as usize)) + .debug_selector(move || format!("queue-drop-{id}")) .ghost() .xsmall() .when(self.compact, |button| button.min_w(px(44.)).min_h(px(44.))) .icon(IconName::Close) - .disabled(!self.interactive(cx)) + .disabled(!editable || !self.interactive(cx)) .tooltip(crate::tr!("composer.drop_queued")) .on_click(cx.listener(move |this, _, window, cx| { this.drop_queued_and_refill(id, text.clone(), window, cx); @@ -151,7 +155,7 @@ impl Composer { ) .context_menu({ let text = message.text.clone(); - let interactive = self.interactive(cx); + let interactive = editable && self.interactive(cx); move |menu, _, _| { menu.menu( crate::tr!("chat.copy_text").into_owned(), diff --git a/crates/ui/src/composer/mod.rs b/crates/ui/src/composer/mod.rs index 182b6ef7e..6bfe6c700 100644 --- a/crates/ui/src/composer/mod.rs +++ b/crates/ui/src/composer/mod.rs @@ -202,6 +202,7 @@ pub struct Composer { /// One cancellable one-second repaint loop, present only while the queue /// strip contains at least one scheduled row. scheduled_countdown_tick: Option>, + queued_refill: Option<(String, u64, String)>, /// Mic button + live dictation session (see `components::voice`). #[cfg(feature = "voice")] voice: Voice, @@ -316,6 +317,7 @@ impl Composer { TopicKind::ActiveSession, TopicKind::Index, TopicKind::SessionStatus, + TopicKind::SessionPlan, TopicKind::SessionEvents, TopicKind::Settings, TopicKind::Providers, @@ -346,6 +348,7 @@ impl Composer { // Recompute the active `@`/`/`/`$` trigger and re-render (also // refreshes the send button's has-text state). InputEvent::Change => { + this.queued_refill = None; // An edit that did not come from the transcript writer // ends dictation (see `components::voice`). #[cfg(feature = "voice")] @@ -453,6 +456,7 @@ impl Composer { image_load_generation: 0, pending_image_loads: 0, scheduled_countdown_tick: None, + queued_refill: None, #[cfg(feature = "voice")] voice, _subscriptions: subscriptions, @@ -540,11 +544,48 @@ impl Composer { &mut self, id: u64, text: String, - window: &mut Window, + _window: &mut Window, cx: &mut Context, ) { + let store = self.workspace_store.read(cx); + if !self.interactive(cx) + || !store.composer_state().queue.is_some_and(|queue| { + queue + .messages + .iter() + .any(|message| message.id == id && message.editable) + }) + { + return; + } + let Some(session_id) = store.active_session_id() else { + return; + }; + self.queued_refill = Some((session_id, id, text)); self.workspace_store .update(cx, |store, _cx| store.drop_queued(id)); + cx.notify(); + } + + fn sync_queued_refill(&mut self, window: &mut Window, cx: &mut Context) { + let Some((session_id, id, _)) = &self.queued_refill else { + return; + }; + let store = self.workspace_store.read(cx); + if store.active_session_id().as_ref() != Some(session_id) { + self.queued_refill = None; + return; + } + let Some(queue) = store.composer_state().queue else { + return; + }; + if let Some(message) = queue.messages.iter().find(|message| message.id == *id) { + if !message.editable { + self.queued_refill = None; + } + return; + } + let (_, _, text) = self.queued_refill.take().unwrap(); let selection = 0..text.len(); self.input.update(cx, |state, cx| { state.set_value(text, window, cx); @@ -582,7 +623,12 @@ impl Composer { window: &mut Window, cx: &mut Context, ) { - if self.workspace_store.read(cx).native_subagent_readonly() { + if self + .workspace_store + .read(cx) + .composer_state() + .conversation_read_only + { return; } if self.compact @@ -998,7 +1044,12 @@ impl Composer { /// The composer's primary control: the stop button while a turn runs, the /// Refine / Implement (split) controls in the plan-ready state, else send. fn render_primary_action(&self, turn_running: bool, cx: &mut Context) -> AnyElement { - if self.workspace_store.read(cx).native_subagent_readonly() { + if self + .workspace_store + .read(cx) + .composer_state() + .conversation_read_only + { return Button::new("send-message") .debug_selector(|| "send-message".into()) .ghost() @@ -1044,10 +1095,11 @@ impl Render for Composer { self.sync_user_input_state(window, cx); self.sync_images_session(cx); self.sync_text_destination(window, cx); + self.sync_queued_refill(window, cx); self.sync_native_rewind_prefill(window, cx); self.sync_fallback_review_draft(window, cx); let composer_state = self.workspace_store.read(cx).composer_state(); - let readonly = self.workspace_store.read(cx).native_subagent_readonly(); + let readonly = composer_state.conversation_read_only; let turn_running = composer_state.turn_running; let approval = composer_state.pending_approval; let approval_count = composer_state.pending_approval_count; @@ -1442,3 +1494,299 @@ impl Render for Composer { ) } } + +#[cfg(test)] +mod replica_tests { + use super::*; + use gpui::{Modifiers, TestAppContext}; + use tcode_protocol::{ + EventEnvelope, HostMessage, ServerEvent, SessionStatus, Topic, encode_line, + }; + + fn attach( + cx: &mut TestAppContext, + ) -> ( + Entity, + async_channel::Sender, + async_channel::Receiver, + SessionStatus, + ) { + cx.update(crate::theme::init); + cx.update(crate::markdown::init); + let root = std::env::temp_dir().join(format!( + "tcode-composer-replica-{}-{}", + std::process::id(), + tcode_services::store::now_millis() + )); + let host = tcode_runtime::pipe::spawn_host( + tcode_services::store::SessionStore::open_at(root.clone()).unwrap(), + tcode_runtime::pipe::HostServices::default(), + ) + .unwrap(); + let status = smol::block_on(host.update_state_for_test(|state, cx| { + let id = state.start_draft("replica".into(), std::env::temp_dir(), cx); + state.session_status_snapshot(&id).unwrap() + })) + .unwrap(); + host.shutdown_blocking().unwrap(); + std::fs::remove_dir_all(root).unwrap(); + let (to_host, outgoing) = async_channel::unbounded(); + let (incoming, from_host) = async_channel::unbounded(); + let link = tcode_client::HostLink::new(to_host, from_host); + let pump = link.clone(); + let executor = cx.background_executor.clone(); + cx.background_executor + .spawn(async move { + pump.pump_with_timer(|| executor.timer(Duration::from_millis(25))) + .await; + }) + .detach(); + let store = cx.new(|cx| { + WorkspaceStore::new_attached( + link, + crate::store::WorkspaceAttachment::Local, + None, + None, + false, + cx, + ) + }); + store.update(cx, |store, _| { + store.select_session(status.session_id.clone()) + }); + (store, incoming, outgoing, status) + } + + fn replace(incoming: &async_channel::Sender, topic: Topic, event: ServerEvent) { + incoming + .try_send( + encode_line(&HostMessage::Event(EventEnvelope { + request_id: None, + topic, + event, + })) + .unwrap(), + ) + .unwrap(); + } + + #[gpui::test] + fn queue_edit_waits_for_host_removal_and_in_flight_actions_are_disabled( + cx: &mut TestAppContext, + ) { + let (store, incoming, outgoing, mut status) = attach(cx); + status.queued_messages = vec![tcode_protocol::QueuedMessageStatus { + id: 7, + editable: false, + delivery_key: None, + text: "Keep this queued until the host accepts the edit".into(), + fire_at_unix_secs: Some(u64::MAX), + }]; + let topic = Topic::SessionStatus { + session_id: status.session_id.clone(), + }; + replace( + &incoming, + topic.clone(), + ServerEvent::SessionStatusReplaced(Box::new(status.clone())), + ); + cx.run_until_parked(); + store.update(cx, |store, cx| store.drain_host_events_for_test(cx)); + let (composer, cx) = + cx.add_window_view(|window, cx| Composer::new(store.clone(), window, cx)); + cx.simulate_resize(gpui::size(px(1024.), px(768.))); + cx.update(|window, cx| _ = window.draw(cx)); + for action in ["queue-drop-7", "queue-steer-7"] { + let button = cx.debug_bounds(action).expect("queue action visible"); + cx.simulate_click(button.center(), Modifiers::default()); + cx.update(|window, cx| _ = window.draw(cx)); + assert!( + composer + .read_with(cx, |composer, cx| composer.draft(cx)) + .is_empty() + ); + } + while let Ok(line) = outgoing.try_recv() { + let message = tcode_protocol::decode_client_line(&line).unwrap(); + assert!( + !matches!( + message.payload, + tcode_protocol::ClientPayload::Command( + tcode_protocol::Command::DropQueued { .. } + | tcode_protocol::Command::SteerQueued { .. } + ) + ), + "disabled actions must not send queue commands" + ); + } + status.queued_messages[0].editable = true; + replace( + &incoming, + topic.clone(), + ServerEvent::SessionStatusReplaced(Box::new(status.clone())), + ); + cx.run_until_parked(); + store.update(cx, |store, cx| store.drain_host_events_for_test(cx)); + cx.update(|window, cx| _ = window.draw(cx)); + let drop_button = cx.debug_bounds("queue-drop-7").unwrap(); + cx.simulate_click(drop_button.center(), Modifiers::default()); + cx.update(|window, cx| _ = window.draw(cx)); + assert!( + composer + .read_with(cx, |composer, cx| composer.draft(cx)) + .is_empty(), + "the pending host drop is still queued" + ); + assert!( + std::iter::from_fn(|| outgoing.try_recv().ok()).any(|line| matches!( + tcode_protocol::decode_client_line(&line).unwrap().payload, + tcode_protocol::ClientPayload::Command(tcode_protocol::Command::DropQueued { + id: 7, + .. + }) + )), + "the enabled drop must reach the host" + ); + status.queued_messages.clear(); + replace( + &incoming, + topic.clone(), + ServerEvent::SessionStatusReplaced(Box::new(status.clone())), + ); + cx.run_until_parked(); + store.update(cx, |store, cx| store.drain_host_events_for_test(cx)); + cx.update(|window, cx| _ = window.draw(cx)); + assert!(cx.debug_bounds("queue-drop-7").is_none()); + assert_eq!( + composer.read_with(cx, |composer, cx| composer.draft(cx)), + "Keep this queued until the host accepts the edit" + ); + cx.update(|window, cx| { + composer.update(cx, |composer, cx| composer.set_draft("", window, cx)) + }); + status + .queued_messages + .push(tcode_protocol::QueuedMessageStatus { + id: 8, + editable: true, + delivery_key: None, + text: "Queued text".into(), + fire_at_unix_secs: None, + }); + replace( + &incoming, + topic.clone(), + ServerEvent::SessionStatusReplaced(Box::new(status.clone())), + ); + cx.run_until_parked(); + store.update(cx, |store, cx| store.drain_host_events_for_test(cx)); + cx.update(|window, cx| _ = window.draw(cx)); + let drop_button = cx.debug_bounds("queue-drop-8").unwrap(); + cx.simulate_click(drop_button.center(), Modifiers::default()); + cx.update(|window, cx| composer.update(cx, |composer, cx| composer.focus(window, cx))); + cx.simulate_input("New draft while the drop is pending"); + cx.run_until_parked(); + status.queued_messages.clear(); + replace( + &incoming, + topic, + ServerEvent::SessionStatusReplaced(Box::new(status)), + ); + cx.run_until_parked(); + store.update(cx, |store, cx| store.drain_host_events_for_test(cx)); + cx.update(|window, cx| _ = window.draw(cx)); + assert_eq!( + composer.read_with(cx, |composer, cx| composer.draft(cx)), + "New draft while the drop is pending" + ); + } + + struct PlanAndComposer { + plan: Entity, + composer: Entity, + } + + impl Render for PlanAndComposer { + fn render(&mut self, _: &mut Window, _: &mut Context) -> impl IntoElement { + v_flex() + .size_full() + .child(div().flex_1().child(self.plan.clone())) + .child(self.composer.clone()) + } + } + + #[gpui::test] + fn ready_plan_replica_renders_without_a_plan_in_the_history_window(cx: &mut TestAppContext) { + let (store, incoming, _outgoing, mut status) = attach(cx); + status.draft = false; + status.interaction_mode = InteractionMode::Plan; + let session_id = status.session_id.clone(); + replace( + &incoming, + Topic::SessionStatus { + session_id: session_id.clone(), + }, + ServerEvent::SessionStatusReplaced(Box::new(status)), + ); + replace( + &incoming, + Topic::SessionEvents { + session_id: session_id.clone(), + }, + ServerEvent::SessionSnapshot { + from: 0, + end: 1, + total: 1, + total_turns: 10, + truncated: false, + records: vec![tcode_core::session::StoredEvent { + ts: Some(1), + elided: None, + event: agent::AgentEvent::TurnStarted { + turn_id: "recent-turn".into(), + }, + }], + }, + ); + replace( + &incoming, + Topic::SessionPlan { + session_id: session_id.clone(), + }, + ServerEvent::SessionPlanReplaced(tcode_protocol::SessionPlan { + session_id, + proposed: Some(tcode_protocol::ProposedPlanStatus { + item_id: "older-plan".into(), + turn: 3, + markdown: "# Ready across devices\n\nImplement the shared plan.".into(), + ready: true, + resolved: false, + }), + steps: vec![agent::PlanStep { + step: "First shared task".into(), + status: agent::PlanStepStatus::Pending, + }], + }), + ); + cx.run_until_parked(); + store.update(cx, |store, cx| store.drain_host_events_for_test(cx)); + assert_eq!( + store.read_with(cx, |store, _| store.with_active_timeline(|timeline| { + timeline.shown_proposed_plan().is_none() + })), + Some(true) + ); + let (_, cx) = cx.add_window_view(|window, cx| PlanAndComposer { + plan: cx.new(|cx| crate::plan_panel::PlanPanel::new(store.clone(), cx)), + composer: cx.new(|cx| Composer::new_with_layout(store.clone(), true, window, cx)), + }); + cx.simulate_resize(gpui::size(px(393.), px(852.))); + cx.update(|window, cx| _ = window.draw(cx)); + assert!(cx.debug_bounds("panel-proposed-plan").is_some()); + assert!(cx.debug_bounds("plan-step-0").is_some()); + assert!( + cx.debug_bounds("implement-main").is_some(), + "Implement must use the ready replica" + ); + } +} diff --git a/crates/ui/src/composer/model.rs b/crates/ui/src/composer/model.rs index ae3fc4f47..7f2c263f2 100644 --- a/crates/ui/src/composer/model.rs +++ b/crates/ui/src/composer/model.rs @@ -365,10 +365,26 @@ pub(super) fn traits_chip_label( continue; } if id == "contextWindow" { - parts.push(agent::claude::format_context_window( - agent::claude::resolved_context_window(&spec.id, selections), - )); - continue; + let selected = selections + .iter() + .find(|selection| selection.id == *id) + .and_then(|selection| { + agent::claude::parse_context_window_tokens(&selection.value) + }); + if let Some(selected) = selected { + parts.push( + options + .iter() + .find(|option| { + agent::claude::parse_context_window_tokens(&serde_json::json!( + option.value + )) == Some(selected) + }) + .map(|option| option.label.clone()) + .unwrap_or_else(|| agent::claude::format_context_window(selected)), + ); + continue; + } } let part = resolved_select_value(id, options, default_value, selections) .and_then(|value| options.iter().find(|o| o.value == value)) @@ -707,70 +723,6 @@ mod tests { #[test] fn traits_chip_joins_descriptor_labels() { let _locale_guard = crate::settings::TestLocaleGuard::acquire(); - let spec = agent::ModelSpec { - id: "claude-fable-5".into(), - display_name: "Claude Fable 5".into(), - is_default: false, - options: vec![ - agent::OptionDescriptor::Select { - id: "reasoningEffort".into(), - label: "Reasoning".into(), - options: vec![ - agent::SelectOption { - value: "high".into(), - label: "High".into(), - description: None, - }, - agent::SelectOption { - value: "max".into(), - label: "Max".into(), - description: None, - }, - ], - default_value: Some("high".into()), - }, - agent::OptionDescriptor::Select { - id: "contextWindow".into(), - label: "Context Window".into(), - options: vec![ - agent::SelectOption { - value: "200k".into(), - label: "200k".into(), - description: None, - }, - agent::SelectOption { - value: "1m".into(), - label: "1M".into(), - description: None, - }, - ], - default_value: Some("200k".into()), - }, - ], - }; - // Context windows resolve from the model's native default, not the - // descriptor's stale fallback. - assert_eq!( - traits_chip_label(&spec, &[], false), - Some("High · 1M".into()) - ); - let sel = vec![agent::OptionSelection { - id: "contextWindow".into(), - value: serde_json::Value::String("1m".into()), - }]; - assert_eq!( - traits_chip_label(&spec, &sel, false), - Some("High · 1M".into()) - ); - let custom = vec![agent::OptionSelection { - id: "contextWindow".into(), - value: serde_json::json!(500_000), - }]; - assert_eq!( - traits_chip_label(&spec, &custom, false), - Some("High · 500k".into()) - ); - // Fast Mode boolean → Fast/Normal; a plain boolean → "