Skip to content

perf: compact thread logs at turn end under a layout epoch #587

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Tryanks wants to merge 1 commit into from
Closed

perf: compact thread logs at turn end under a layout epoch #587

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions crates/app/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,7 @@ fn start_local(store: SessionStore) -> (SpawnedHost, HostMux) {
let mut host_services = HostServices {
background_startup_probes: true,
ai_title_generation: true,
compact_existing_logs: true,
..HostServices::default()
};
match mcp_host::Host::bind() {
Expand Down Expand Up @@ -250,6 +251,7 @@ impl LocalKernel {
if let Err(error) = self.control_link.subscribe(tcode_protocol::Subscription {
topic: topic.clone(),
after: None,
epoch: None,
}) {
log::error!("could not request local settings: {}", error.message);
return settings::Settings::default();
Expand All @@ -271,9 +273,11 @@ impl LocalKernel {
Err(_) => break settings::Settings::default(),
}
};
let _ = self
.control_link
.unsubscribe(tcode_protocol::Subscription { topic, after: None });
let _ = self.control_link.unsubscribe(tcode_protocol::Subscription {
topic,
after: None,
epoch: None,
});
result
}
}
Expand Down
43 changes: 30 additions & 13 deletions crates/client/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -703,18 +703,20 @@ impl HostLink {
self.send_payload(self.next_id(), ClientPayload::Unsubscribe(subscription))
}

/// Advance replay memory only after the store has applied these records. Sending
/// the latest subscription also updates native/browser transport replay caches.
/// The host returns an empty tail when the store is already up to date.
pub fn update_after(&self, topic: &Topic, after: u64) -> Result<(), ProtocolError> {
/// Advance replay memory to position `after` of layout `epoch` only after
/// the store has applied these records. Sending the latest subscription also
/// updates native/browser transport replay caches. The host returns an empty
/// tail when the store is already up to date.
pub fn update_after(&self, topic: &Topic, epoch: u64, after: u64) -> Result<(), ProtocolError> {
let subscription = {
let mut topics = self.inner.subscribed_topics.lock().unwrap();
let Some(subscription) = topics.get_mut(topic) else {
return Ok(());
};
if subscription.after == Some(after) {
if (subscription.epoch, subscription.after) == (Some(epoch), Some(after)) {
return Ok(());
}
subscription.epoch = Some(epoch);
subscription.after = Some(after);
subscription.clone()
};
Expand Down Expand Up @@ -804,6 +806,7 @@ impl HostLink {
ClientPayload::Unsubscribe(Subscription {
topic: topic.clone(),
after: None,
epoch: None,
}),
);
}
Expand Down Expand Up @@ -1305,6 +1308,7 @@ mod tests {
session_id: "one".into(),
},
after: None,
epoch: None,
};
link.subscribe(subscription.clone()).unwrap();
outgoing.try_recv().unwrap();
Expand All @@ -1328,14 +1332,15 @@ mod tests {
let reply = EventEnvelope {
request_id: Some(request.id),
topic: subscription.topic,
event: ServerEvent::SessionSnapshot {
event: ServerEvent::SessionSnapshot(tcode_protocol::SessionWindow {
epoch: 0,
total: 0,
total_turns: 0,
truncated: false,
from: 0,
end: 0,
records: vec![],
},
}),
};
assert!(link.subscription_reply_is_current(&reply));
}
Expand Down Expand Up @@ -1385,14 +1390,19 @@ mod tests {
link.subscribe(Subscription {
topic: topic.clone(),
after: None,
epoch: None,
})
.unwrap();
outgoing.try_recv().unwrap();
link.update_after(&topic, 7).unwrap();
link.update_after(&topic, 2, 7).unwrap();
let updated = tcode_protocol::decode_client_line(&outgoing.try_recv().unwrap()).unwrap();
assert!(matches!(
updated.payload,
ClientPayload::Subscribe(Subscription { after: Some(7), .. })
ClientPayload::Subscribe(Subscription {
after: Some(7),
epoch: Some(2),
..
})
));
link.set_connection_state(ConnectionState::Reconnecting {
attempt: 1,
Expand All @@ -1404,18 +1414,23 @@ mod tests {
let queued_reply = EventEnvelope {
request_id: Some(replay.id),
topic: topic.clone(),
event: ServerEvent::SessionSnapshot {
event: ServerEvent::SessionSnapshot(tcode_protocol::SessionWindow {
epoch: 0,
total: 0,
total_turns: 0,
truncated: false,
from: 0,
end: 0,
records: vec![],
},
}),
};
assert!(link.subscription_reply_is_current(&queued_reply));
link.unsubscribe(Subscription { topic, after: None })
.unwrap();
link.unsubscribe(Subscription {
topic,
after: None,
epoch: None,
})
.unwrap();
assert!(!link.subscription_reply_is_current(&queued_reply));
assert!(link.inner.subscription_requests.lock().unwrap().is_empty());
outgoing.try_recv().unwrap();
Expand Down Expand Up @@ -1487,11 +1502,13 @@ mod tests {

link.subscribe(Subscription {
after: None,
epoch: None,
topic: Topic::Index,
})
.unwrap();
link.subscribe(Subscription {
after: None,
epoch: None,
topic: Topic::Index,
})
.unwrap();
Expand Down
58 changes: 36 additions & 22 deletions crates/core/src/session.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,10 @@ use serde::{Deserialize, Serialize};

use crate::git::merge_file_changes_by_path;

mod compaction;

pub use compaction::{CompactedRecord, compact_records};

/// Claude Code's own prompt for resuming work after a usage-window reset.
pub const RESUME_PROMPT: &str = "Continue from where you left off.";

Expand Down Expand Up @@ -169,7 +173,7 @@ impl From<AgentEvent> for StoredEvent {
}

/// One renderable row in the chat timeline.
#[derive(Debug, Clone)]
#[derive(Debug, Clone, PartialEq)]
pub struct TimelineEntry {
/// Provider item id (or a synthetic id for errors).
pub id: String,
Expand All @@ -181,7 +185,7 @@ pub struct TimelineEntry {
}

/// Per-turn ("Work Log" section) metadata folded from turn lifecycle events.
#[derive(Debug, Clone, Default)]
#[derive(Debug, Clone, Default, PartialEq)]
pub struct TurnMeta {
/// Provider-native id for this turn, used to attach replacement turn-diff
/// snapshots without relying on ambient "current turn" state during replay.
Expand Down Expand Up @@ -243,7 +247,7 @@ impl TurnTiming {
/// turn, intersected with the turn's own `TurnStarted`..`TurnCompleted` bounds.
/// Only the current turn can have items in flight, so one accumulator is
/// enough; it resets whenever a turn opens.
#[derive(Debug, Clone, Default)]
#[derive(Debug, Clone, Default, PartialEq)]
struct ToolClock {
/// Item ids currently known to be in progress.
open: HashSet<String>,
Expand Down Expand Up @@ -408,13 +412,13 @@ fn tool_is_active(lifecycle: ToolLifecycle, state: ToolState) -> bool {
}
}

#[derive(Debug, Clone)]
#[derive(Debug, Clone, PartialEq)]
pub struct TurnChangeSet {
pub changes: Vec<FileChange>,
pub completeness: ChangeCompleteness,
}

#[derive(Debug, Clone)]
#[derive(Debug, Clone, PartialEq)]
pub enum EntryContent {
Item(ItemContent),
/// A user message injected into an already-open turn. Provider-originated
Expand Down Expand Up @@ -472,7 +476,7 @@ pub enum SteeringStatus {
/// A proposed plan captured this session (Codex plan item / Claude
/// `ExitPlanMode`). Streaming deltas accumulate into `markdown`; a `ProposedPlan`
/// event replaces it with the final text.
#[derive(Debug, Clone, Default)]
#[derive(Debug, Clone, Default, PartialEq)]
pub struct ProposedPlan {
pub item_id: String,
pub markdown: String,
Expand Down Expand Up @@ -502,8 +506,10 @@ pub struct RunningTurn {
pub started_at: Option<u64>,
}

/// Folded view of a session's event history.
#[derive(Debug, Clone, Default)]
/// Folded view of a session's event history. Two timelines are equal only
/// when every later event folds the same onto both, so equality covers the
/// private state the fold continues from, not just what renders.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct Timeline {
/// Top-level entries are shared so virtualized UI snapshots can retain a
/// turn without cloning its potentially large message, command-output, and
Expand Down Expand Up @@ -646,12 +652,7 @@ impl Timeline {

/// Apply one event recorded at `ts` (unix ms). Mutates in place.
pub fn apply_at(&mut self, ts: Option<u64>, event: &AgentEvent) {
// Every timestamped event inside the open turn raises the turn's
// watermark, not just tool lifecycle ones: a completion stamped before
// any of them means the wall clock regressed, and the breakdown is
// withheld rather than guessed at. The completion itself is excluded —
// it is what gets compared against the watermark.
if self.turn_is_open() && !matches!(event, AgentEvent::TurnCompleted { .. }) {
if self.watermarks(event) {
self.tool_clock.observe(ts);
}
match event {
Expand Down Expand Up @@ -754,10 +755,7 @@ impl Timeline {
completeness,
} => {
let turn = self
.turns
.iter()
.position(|turn| turn.provider_turn_id.as_deref() == Some(turn_id.as_str()))
.or(self.current_turn)
.provider_turn(turn_id)
.unwrap_or_else(|| self.push_turn(ts));
if !turn_id.is_empty() {
self.turns[turn].provider_turn_id = Some(turn_id.clone());
Expand All @@ -772,10 +770,7 @@ impl Timeline {
checkpoint_id,
} => {
let turn = self
.turns
.iter()
.position(|turn| turn.provider_turn_id.as_deref() == Some(turn_id.as_str()))
.or(self.current_turn)
.provider_turn(turn_id)
.unwrap_or_else(|| self.push_turn(ts));
self.turns[turn].provider_turn_id = Some(turn_id.clone());
self.turns[turn].provider_checkpoint_id = Some(checkpoint_id.clone());
Expand Down Expand Up @@ -1069,6 +1064,25 @@ impl Timeline {
}
}

/// Whether `event`'s time raises the open turn's clock watermark. Every
/// timestamped event inside the open turn does, not just tool lifecycle
/// ones: a completion stamped before any of them means the wall clock
/// regressed, and the breakdown is withheld rather than guessed at. The
/// completion itself is excluded — it is what gets compared against the
/// watermark.
fn watermarks(&self, event: &AgentEvent) -> bool {
self.turn_is_open() && !matches!(event, AgentEvent::TurnCompleted { .. })
}

/// The turn a provider-addressed record lands on: the first turn carrying
/// `turn_id`, else the current one. `None` means the record opens a turn.
fn provider_turn(&self, turn_id: &str) -> Option<usize> {
self.turns
.iter()
.position(|turn| turn.provider_turn_id.as_deref() == Some(turn_id))
.or(self.current_turn)
}

/// Whether the current turn is still accumulating. A turn is finished once
/// a `TurnCompleted` has been folded, which records a status even when the
/// event carried no timestamp to store as `end_ts`; both must be checked or
Expand Down
Loading
Loading