Skip to content
19 changes: 13 additions & 6 deletions crates/client/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ struct Write {
id: u64,
entry: outbox::Entry,
sent: Option<web_time::Instant>,
adopted: bool,
}

#[derive(Default)]
Expand Down Expand Up @@ -214,6 +215,7 @@ impl HostLink {
}
}
if let HostMessage::Ack { result: Ok(_), .. } = &message
&& !write.adopted
&& matches!(
write.entry.command,
Command::SendTurn { .. }
Expand Down Expand Up @@ -270,6 +272,7 @@ impl HostLink {
id: self.next_id(),
entry,
sent: None,
adopted: false,
});
}
Ok(())
Expand Down Expand Up @@ -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)> {
Expand Down Expand Up @@ -364,6 +370,7 @@ impl HostLink {
id,
entry,
sent: None,
adopted: false,
});
let mut evicted = Vec::new();
while delivery.writes.len() > outbox::MAX_ITEMS
Expand Down
58 changes: 52 additions & 6 deletions crates/core/src/project.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<PathBuf, Vec<&'a SessionMeta>>,
branches: HashMap<&'a str, Vec<&'a SessionMeta>>,
}

impl<'a> WorktreeSharing<'a> {
pub fn new(sessions: impl IntoIterator<Item = &'a SessionMeta>) -> 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
Expand Down Expand Up @@ -569,13 +612,16 @@ mod tests {
let sibling = SessionMeta::new(ProviderKind::Codex, PathBuf::from("/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"
);
}
Expand Down
19 changes: 14 additions & 5 deletions crates/core/src/session.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<usize>, started_at: Option<u64>) {
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<usize>,
started_at: Option<u64>,
) {
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);
}
Expand All @@ -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 {
Expand Down Expand Up @@ -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"
Expand Down
81 changes: 66 additions & 15 deletions crates/protocol/src/event.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand All @@ -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,
Expand Down Expand Up @@ -63,7 +64,8 @@ pub enum ServerEvent {
request: crate::PreviewRequest,
},
SessionEvent(StoredEvent),
SessionStatusReplaced(SessionStatus),
SessionStatusReplaced(Box<SessionStatus>),
SessionPlanReplaced(SessionPlan),
ProvidersReplaced(ProvidersStatus),
GitStatusReplaced(GitStatusStatus),
/// An unarchived thread was added or changed. Archiving a thread removes
Expand Down Expand Up @@ -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<ProposedPlanStatus>,
pub steps: Vec<PlanStep>,
}

#[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,
Expand All @@ -367,6 +402,8 @@ pub struct SessionStatus {
pub acp_agent_id: Option<String>,
pub project_id: Option<String>,
pub approval_mode: ApprovalMode,
pub effective_approval_mode: ApprovalMode,
pub native_approval_modes_enabled: bool,
pub interaction_mode: InteractionMode,
pub queued_messages: Vec<QueuedMessageStatus>,
/// Backend-owned drafts used by provider-bound send-path assembly.
Expand All @@ -378,11 +415,18 @@ pub struct SessionStatus {
pub terminal_open: bool,
pub terminal_height: f32,
pub delivery_in_flight: Option<u64>,
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<TokenUsage>,
/// Effective capacity the context meter should use, resolved by the host.
pub context_window: Option<u64>,
/// 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.
Expand Down Expand Up @@ -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<String>,
pub text: String,
Expand All @@ -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<String, (bool, bool, bool, bool)>,
pub activity: HashMap<String, SessionActivity>,
/// Background title requests, including threads with no live provider.
#[serde(default)]
pub title_generating: HashSet<String>,
/// Archived threads per project id.
#[serde(default)]
pub archived_counts: HashMap<String, usize>,
/// Worktree branches archived threads still use.
#[serde(default)]
pub archived_worktree_branches: HashSet<String>,
pub worktree_shared: HashSet<String>,
pub archived_revision: u64,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ArchivedSessions {
pub sessions: Vec<SessionMeta>,
pub worktree_shared: HashSet<String>,
pub revision: u64,
}

/// A transient runtime notification delivered to clients.
Expand Down
20 changes: 12 additions & 8 deletions crates/protocol/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)]
Expand Down
2 changes: 1 addition & 1 deletion crates/protocol/src/query.rs
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,7 @@ pub enum QueryResponse {
/// screen, not a session.
TerminalFrame(Box<crate::terminal::TerminalFrame>),
ItemOutput(String),
ArchivedSessions(Vec<tcode_core::project::SessionMeta>),
ArchivedSessions(crate::ArchivedSessions),
}

/// One content match in a stored session, addressed by the folded timeline
Expand Down
Loading
Loading