1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
//! Session state mirrored to ACP clients alongside the event stream.
//!
//! Context fill and cumulative cost go out as `usage_update` and the todo
//! checklist as `plan`. None is a run event: each is read from session-owned
//! state, or the usage ledger, when an event says it may have moved.
use agent_client_protocol::schema::v1::{
Cost, Plan, PlanEntry, PlanEntryPriority, PlanEntryStatus, SessionId, SessionNotification,
SessionUpdate, UsageUpdate,
};
use rho_sdk::RunEvent;
use super::AcpClientPort;
use crate::{
tools::todo::{TodoItem, TodoState, TodoStatus},
usage::report::Valuation,
};
/// Tracks one prompt's context fill and checklist for a session.
///
/// Build one per prompt, since the session handle changes when the runtime is
/// rebound. The host keeps the last checklist it sent across prompts and hands
/// it in, so a checklist restored by `session/load` goes out with the first
/// prompt. Nothing is sent during load itself: hosts such as T3 Code discard
/// every update for a session while its load is in flight.
pub(super) struct SessionSignals {
session: rho_sdk::Session,
todo: TodoState,
/// Configured window for the current model; the provider-reported window
/// is the fallback, matching the TUI context meter.
context_window: Option<u64>,
sent_plan: Vec<PlanEntry>,
}
impl SessionSignals {
pub(super) fn new(
session: rho_sdk::Session,
todo: TodoState,
context_window: Option<u64>,
sent_plan: Vec<PlanEntry>,
) -> Self {
Self {
session,
todo,
context_window,
sent_plan,
}
}
/// The update a run event triggers, if any. A new step or a compaction
/// moves context fill, and the requests before it moved cost; a finished
/// tool may have replaced the checklist, whether it ran directly or inside
/// code mode.
pub(super) async fn observe(&mut self, event: &RunEvent) -> Option<SessionUpdate> {
match event {
RunEvent::StepStarted { .. } | RunEvent::CompactionCompleted { .. } => {
Some(self.usage().await)
}
RunEvent::ToolFinished { .. } => self.plan_change(),
_ => None,
}
}
/// Current context fill and cumulative cost. ACP's `size` is required; 0
/// means the window is unknown, which clients render as no maximum.
///
/// `cost` is the session's whole ledger total, so it already includes
/// turns from before a restart or `session/load` and never double counts
/// them. It is omitted, never sent as 0, while no request has a price;
/// an all-local session omits it too, as the TUI statusline does.
pub(super) async fn usage(&self) -> SessionUpdate {
let estimate = self.session.context_estimate();
let size = self
.context_window
.or(estimate.reported_context_window())
.unwrap_or(0);
let cost = match crate::usage::session_valuation(self.session.id()).await {
Valuation::Usd(micros) => Some(Cost::new(micros as f64 / 1_000_000.0, "USD")),
Valuation::Local | Valuation::Unpriced => None,
};
SessionUpdate::UsageUpdate(UsageUpdate::new(estimate.tokens(), size).cost(cost))
}
/// Sends the checklist before the run starts if it differs from what the
/// client last saw, as after a load restored one.
pub(super) async fn send_plan_change(
&mut self,
session_id: &SessionId,
client: &dyn AcpClientPort,
) {
if let Some(update) = self.plan_change() {
send(session_id, client, update).await;
}
}
/// Sends current context fill at turn end, outside the event stream.
pub(super) async fn send_usage(&self, session_id: &SessionId, client: &dyn AcpClientPort) {
send(session_id, client, self.usage().await).await;
}
/// The checklist the client last saw, for the next prompt's tracker.
pub(super) fn into_sent_plan(self) -> Vec<PlanEntry> {
self.sent_plan
}
fn plan_change(&mut self) -> Option<SessionUpdate> {
let entries = plan_entries(&self.todo);
if entries == self.sent_plan {
return None;
}
self.sent_plan.clone_from(&entries);
Some(SessionUpdate::Plan(Plan::new(entries)))
}
}
/// Best effort: a closed connection fails the next request instead.
async fn send(session_id: &SessionId, client: &dyn AcpClientPort, update: SessionUpdate) {
let notification = SessionNotification::new(session_id.clone(), update);
if let Err(error) = client.send_session_notification(notification).await {
tracing::debug!(%error, "could not send ACP session update");
}
}
fn plan_entries(todo: &TodoState) -> Vec<PlanEntry> {
todo.list()
.map(|list| list.todos.iter().map(plan_entry).collect())
.unwrap_or_default()
}
fn plan_entry(item: &TodoItem) -> PlanEntry {
let status = match item.status {
TodoStatus::Pending => PlanEntryStatus::Pending,
TodoStatus::InProgress => PlanEntryStatus::InProgress,
TodoStatus::Completed => PlanEntryStatus::Completed,
};
// The todo tool has no priorities; Medium is the neutral ACP value.
PlanEntry::new(item.content.clone(), PlanEntryPriority::Medium, status)
}
#[cfg(test)]
#[path = "signals_tests.rs"]
mod tests;