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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
//! The session facade: the single owner of [`RuntimeHandles`] and the only
//! place that drives `everruns_core` runtime APIs (`run_turn`, `messages`,
//! `events`, `execute_command`, the live event broadcast).
//!
//! The TUI (`crate::app`) talks to a [`Session`] and consumes the
//! [`crate::transcript::TurnEvent`] stream it produces; it never subscribes to
//! the raw runtime event bus or reaches into runtime internals. This keeps the
//! `everruns_core` event/message model from leaking across the UI surface and
//! makes the turn lifecycle independently testable.
use std::collections::HashSet;
use std::sync::Arc;
use anyhow::Result;
use everruns_core::command::ExecuteCommandRequest;
use everruns_core::message::ContentPart;
use everruns_core::tools::Tool;
use everruns_core::typed_id::SessionId;
use tokio::sync::{broadcast, mpsc, oneshot};
use crate::runtime::{ModelState, RuntimeHandles};
use crate::tools::{BashTool, Workspace};
use crate::transcript::{
Author, ChatLine, DeltaRouter, TurnEvent, assistant_lines_since, handle_live_event,
lines_for_event_with_router, lines_for_replayed_event, remember_write_todos_args,
shell_result_lines, status_for_event, tokens_for_event,
};
/// Outcome of a capability-provided slash command, reduced to what the host UI
/// needs so the `everruns_core` command result type does not leak past the
/// facade.
pub(crate) struct CommandOutcome {
pub success: bool,
pub message: String,
}
/// A running turn (or `!shell` command): the event stream the host drains and a
/// one-shot cancel trigger.
pub(crate) struct TurnHandle {
pub events: mpsc::UnboundedReceiver<TurnEvent>,
pub cancel: oneshot::Sender<()>,
}
/// Facade over the runtime for a single session. Cheap to construct; holds
/// shared handles (`Arc`-backed) plus a [`ModelState`] clone (also `Arc`-shared,
/// so model changes made elsewhere are observed here).
#[derive(Clone)]
pub(crate) struct Session {
handles: RuntimeHandles,
model: ModelState,
}
impl Session {
pub fn new(handles: RuntimeHandles, model: ModelState) -> Self {
Self { handles, model }
}
pub fn session_id(&self) -> SessionId {
self.handles.session_id
}
/// Translate the prefix of persisted events that were replayed from disk
/// into transcript lines (used to seed the transcript on resume).
pub async fn replayed_lines(&self, count: usize) -> Result<Vec<ChatLine>> {
let events = self.handles.runtime.events().await?;
Ok(events
.iter()
.take(count)
.flat_map(lines_for_replayed_event)
.collect())
}
/// Execute a capability-provided command through the runtime, returning a
/// host-facing [`CommandOutcome`].
pub async fn execute_command(
&self,
name: &str,
arguments: Option<String>,
) -> Result<CommandOutcome> {
let request = ExecuteCommandRequest {
name: name.to_string(),
arguments,
controls: None,
};
let result = self
.handles
.runtime
.execute_command(self.handles.session_id, request)
.await?;
Ok(CommandOutcome {
success: result.success,
message: result.message,
})
}
/// Run an agent turn for `prompt` (with any pending `images`). Spawns the
/// turn task and returns a [`TurnHandle`]; the task emits `TurnEvent`s and
/// finishes with `Done` (or `Failed`).
pub fn run_turn(&self, prompt: String, images: Vec<ContentPart>) -> TurnHandle {
let handles = self.handles.clone();
let model = self.model.clone();
let (tx, rx) = mpsc::unbounded_channel::<TurnEvent>();
let (cancel_tx, mut cancel_rx) = oneshot::channel::<()>();
// Subscribe BEFORE spawning the turn so we don't miss the first
// few events (turn.started, reason.started). The broadcast only
// delivers events emitted after subscribe().
let mut live = handles.events.subscribe();
tokio::spawn(async move {
let session_id = handles.session_id;
let before = match handles.runtime.messages(session_id).await {
Ok(m) => m.len(),
Err(e) => {
let _ = tx.send(TurnEvent::Failed(format!("load history: {e}")));
let _ = tx.send(TurnEvent::Done);
return;
}
};
let events_before = match handles.runtime.events().await {
Ok(e) => e.len(),
Err(_) => 0,
};
let input = model.input_message_with_images(prompt, images);
let runtime = handles.runtime.clone();
let mut turn = tokio::spawn(async move { runtime.run_turn(session_id, input).await });
let mut emitted_events = HashSet::new();
let mut delta_router = DeltaRouter::default();
// High-water mark into the persisted event vec. Each catch-up
// advances it to `events.len()` so a later catch-up only scans the
// newly persisted suffix — repeated `Lagged` recoveries on a long
// session stay O(total new events), not O(n²) re-scans.
let mut events_cursor = events_before;
let mut cancelled = false;
// Set once the turn task joins from within the loop, so we don't
// await the `JoinHandle` twice.
let mut joined = None;
loop {
tokio::select! {
biased;
_ = &mut cancel_rx => {
cancelled = true;
turn.abort();
break;
}
recv = live.recv() => match recv {
Ok(event) => {
if event.session_id != session_id {
continue;
}
handle_live_event(
&event,
&mut emitted_events,
&mut delta_router,
&tx,
);
}
Err(broadcast::error::RecvError::Lagged(_)) => {
// Receiver overflow: catch up from the canonical
// event vec so we don't lose persistent events.
// Resubscribe to restart from the current head.
live = handles.events.subscribe();
catch_up_events(
&handles,
&mut events_cursor,
&mut emitted_events,
&mut delta_router,
&tx,
)
.await;
}
Err(broadcast::error::RecvError::Closed) => break,
},
// Turn finished: no poll/latency. Live events buffered
// before this point are preferred (biased order); the tail
// is drained by the catch-up below.
res = &mut turn => {
joined = Some(res);
break;
}
}
}
if cancelled {
let _ = tx.send(TurnEvent::Stream(None));
let _ = tx.send(TurnEvent::Lines(vec![ChatLine {
author: Author::System,
text: "turn cancelled".into(),
}]));
let _ = tx.send(TurnEvent::Done);
return;
}
// Drain any tail events emitted between the last broadcast
// poll and the turn's actual completion.
catch_up_events(
&handles,
&mut events_cursor,
&mut emitted_events,
&mut delta_router,
&tx,
)
.await;
// Clear any in-flight streaming preview before we finalize.
let _ = tx.send(TurnEvent::Stream(None));
// `joined` is set unless the loop broke on a closed broadcast
// before the turn finished; await the handle in that rare case.
let result = match joined {
Some(res) => res,
None => turn.await,
};
let result = match result {
Ok(result) => result,
Err(e) => {
let _ = tx.send(TurnEvent::Failed(format!("turn task: {e}")));
let _ = tx.send(TurnEvent::Done);
return;
}
};
let response = match result {
Ok(r) => r,
Err(e) => {
let _ = tx.send(TurnEvent::Failed(format!("{e}")));
let _ = tx.send(TurnEvent::Done);
return;
}
};
let messages = handles
.runtime
.messages(session_id)
.await
.unwrap_or_default();
// Assistant text from the turn.
let mut out = assistant_lines_since(&messages, before);
if out.is_empty() && !response.response.is_empty() {
out.push(ChatLine {
author: Author::Assistant,
text: response.response,
});
}
if !response.success
&& let Some(err) = response.error
{
out.push(ChatLine {
author: Author::System,
text: format!("turn error: {err}"),
});
}
let _ = tx.send(TurnEvent::Lines(out));
let _ = tx.send(TurnEvent::Done);
});
TurnHandle {
events: rx,
cancel: cancel_tx,
}
}
/// Run a host-local `!shell` command (not part of a tool-call lifecycle).
/// Output is rendered inline; nothing is persisted to the session.
pub fn run_shell(
&self,
command: String,
workspace: Arc<crate::workspace_host::WorkspaceHost>,
) -> TurnHandle {
let (tx, rx) = mpsc::unbounded_channel::<TurnEvent>();
let (cancel_tx, mut cancel_rx) = oneshot::channel::<()>();
tokio::spawn(async move {
let tool = BashTool::new(Workspace::new(workspace));
let run = tool.execute(serde_json::json!({
"command": command,
// Direct shell output is not persisted through a tool-call
// lifecycle, so render a useful bounded window inline.
"output": "normal",
}));
tokio::select! {
result = run => {
let _ = tx.send(TurnEvent::Lines(shell_result_lines(result)));
}
_ = &mut cancel_rx => {
let _ = tx.send(TurnEvent::Lines(vec![ChatLine {
author: Author::System,
text: "turn cancelled".into(),
}]));
}
}
let _ = tx.send(TurnEvent::Done);
});
TurnHandle {
events: rx,
cancel: cancel_tx,
}
}
}
/// Drain any persisted events (from `runtime.events()`) that the broadcast
/// receiver may have missed — used after a `Lagged` recv error and once more at
/// end-of-turn so the transcript is never missing tool/reason completion lines.
async fn catch_up_events(
handles: &RuntimeHandles,
cursor: &mut usize,
emitted_events: &mut HashSet<String>,
router: &mut DeltaRouter,
tx: &mpsc::UnboundedSender<TurnEvent>,
) {
let events = handles.runtime.events().await.unwrap_or_default();
let mut lines = Vec::new();
// Only scan the suffix persisted since the last catch-up; `emitted_events`
// still de-dupes overlap with events already delivered live.
for event in events.iter().skip(*cursor) {
let event_id = event.id.to_string();
if !emitted_events.insert(event_id) {
continue;
}
if let Some(tokens) = tokens_for_event(event) {
let _ = tx.send(TurnEvent::Tokens(tokens));
}
remember_write_todos_args(event, router);
if let Some(activity) = status_for_event(event) {
let _ = tx.send(TurnEvent::Activity(activity));
}
lines.extend(lines_for_event_with_router(event, router));
}
*cursor = events.len();
if !lines.is_empty() {
let _ = tx.send(TurnEvent::Lines(lines));
}
}