agent_engine/session/
handle.rs

1//! `SessionHandle` — the in-process address of a session (PLAN-phase2 §2.6).
2//!
3//! `EngineHost::create_session` (A1) hands one out per session; every
4//! `ClientTransport` is built on it. `echo_for_test` ships a stand-in actor
5//! so package B can build the wire before the real `SessionActor` exists.
6
7use std::sync::atomic::{AtomicU8, Ordering};
8use std::sync::Arc;
9
10use tokio::sync::{broadcast, mpsc};
11
12use super::transport::TransportError;
13use super::types::{Addressed, ClientId, Envelope, SessionCommand, SessionConfig, SessionId, SessionLifecycle, SessionMeta};
14
15/// Live client accounting published by the actor (B4 `sessions()` listing).
16#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
17pub struct Presence {
18    pub clients: usize,
19    pub input_owner: Option<ClientId>,
20    pub awaiting_input: usize,
21}
22use super::view::RuntimeView;
23
24/// Bounded command queue depth (rpc `WRITER_CHAN_CAP` precedent).
25pub const CMD_CHAN_CAP: usize = 256;
26/// Default broadcast capacity; `SYNAPS_SESSION_EVENTS_CAP` overrides.
27pub const DEFAULT_EVENTS_CAP: usize = 1024;
28
29/// Broadcast capacity from `SYNAPS_SESSION_EVENTS_CAP` (min 16) or the default.
30pub fn events_cap() -> usize {
31    std::env::var("SYNAPS_SESSION_EVENTS_CAP")
32        .ok()
33        .and_then(|s| s.parse::<usize>().ok())
34        .map(|n| n.max(16))
35        .unwrap_or(DEFAULT_EVENTS_CAP)
36}
37
38#[derive(Clone)]
39pub struct SessionHandle {
40    pub id: SessionId,
41    cmd_tx: mpsc::Sender<Addressed>,
42    events: broadcast::Sender<Envelope>,
43    meta: Arc<SessionMeta>,
44    /// Sync, lock-free read of the last published RuntimeView (§2.7).
45    view: Arc<arc_swap::ArcSwap<RuntimeView>>,
46    /// Written by the actor (Live/Parking/Parked/Ending).
47    lifecycle: Arc<AtomicU8>,
48    /// `conv.session.id` — changes on LinkedSuccessor compaction.
49    journal_id: Arc<arc_swap::ArcSwap<String>>,
50    /// Clients / input owner / pending prompts, written by the actor.
51    presence: Arc<arc_swap::ArcSwap<Presence>>,
52    /// Session name — `--name` at create / `/cmd saveas` later. Written by
53    /// the actor so listings and `--continue <name>` see renames live.
54    name: Arc<arc_swap::ArcSwap<Option<String>>>,
55    /// Per-session `/keepwarm` override (task #444), written by the actor.
56    /// Lives here so a daemon reload whose checkpoint did not confirm still
57    /// carries the user's explicit choice.
58    cache_keepwarm: super::cache_keepwarm::OverrideSlot,
59    /// F23: only on a lock-held placeholder — the config to retry the real
60    /// `create` with on the next attach. `None` for real actors.
61    placeholder_config: Option<Arc<SessionConfig>>,
62}
63
64impl std::fmt::Debug for SessionHandle {
65    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
66        f.debug_struct("SessionHandle")
67            .field("id", &self.id)
68            .field("model", &self.meta.model)
69            .finish()
70    }
71}
72
73/// Actor-side ends of a session's channels, minted with the handle.
74pub struct SessionEndpoints {
75    pub cmd_rx: mpsc::Receiver<Addressed>,
76    pub events: broadcast::Sender<Envelope>,
77    pub view: Arc<arc_swap::ArcSwap<RuntimeView>>,
78    pub lifecycle: Arc<AtomicU8>,
79    pub journal_id: Arc<arc_swap::ArcSwap<String>>,
80    pub presence: Arc<arc_swap::ArcSwap<Presence>>,
81    pub name: Arc<arc_swap::ArcSwap<Option<String>>>,
82    pub cache_keepwarm: super::cache_keepwarm::OverrideSlot,
83}
84
85impl SessionHandle {
86    /// Mint a handle + the actor's ends. Used by `SessionActor` (A1) and the
87    /// test `EchoActor`.
88    pub fn new(meta: SessionMeta, view: RuntimeView) -> (Self, SessionEndpoints) {
89        let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHAN_CAP);
90        let (events, _) = broadcast::channel(events_cap());
91        let view = Arc::new(arc_swap::ArcSwap::from_pointee(view));
92        let lifecycle = Arc::new(AtomicU8::new(SessionLifecycle::Live as u8));
93        let journal_id = Arc::new(arc_swap::ArcSwap::from_pointee(meta.id.0.clone()));
94        let presence = Arc::new(arc_swap::ArcSwap::from_pointee(Presence::default()));
95        let name = Arc::new(arc_swap::ArcSwap::from_pointee(meta.name.clone()));
96        let cache_keepwarm = super::cache_keepwarm::OverrideSlot::default();
97        let handle = Self {
98            id: meta.id.clone(),
99            cmd_tx,
100            events: events.clone(),
101            meta: Arc::new(meta),
102            view: Arc::clone(&view),
103            lifecycle: Arc::clone(&lifecycle),
104            journal_id: Arc::clone(&journal_id),
105            presence: Arc::clone(&presence),
106            name: Arc::clone(&name),
107            cache_keepwarm: Arc::clone(&cache_keepwarm),
108            placeholder_config: None,
109        };
110        (
111            handle,
112            SessionEndpoints {
113                cmd_rx,
114                events,
115                view,
116                lifecycle,
117                journal_id,
118                presence,
119                name,
120                cache_keepwarm,
121            },
122        )
123    }
124
125    /// Send a host-originated command (`from: None`). `Backpressure` when
126    /// the queue is full (never drop silently); `Closed` once the actor is
127    /// gone.
128    pub async fn send(&self, cmd: SessionCommand) -> Result<(), TransportError> {
129        self.send_addressed(Addressed { from: None, cmd })
130    }
131
132    /// Send on behalf of an attached client (ownership is enforced by the
133    /// actor, B1).
134    pub async fn send_from(&self, from: ClientId, cmd: SessionCommand) -> Result<(), TransportError> {
135        self.send_addressed(Addressed { from: Some(from), cmd })
136    }
137
138    fn send_addressed(&self, cmd: Addressed) -> Result<(), TransportError> {
139        match self.cmd_tx.try_send(cmd) {
140            Ok(()) => Ok(()),
141            Err(mpsc::error::TrySendError::Full(_)) => Err(TransportError::Backpressure),
142            Err(mpsc::error::TrySendError::Closed(_)) => Err(TransportError::Closed),
143        }
144    }
145
146    /// Park/unpark state as last written by the actor.
147    pub fn lifecycle(&self) -> SessionLifecycle {
148        SessionLifecycle::from_u8(self.lifecycle.load(Ordering::Acquire))
149    }
150
151    /// Current `conv.session.id` (= `id` until a LinkedSuccessor compaction).
152    pub fn journal_id(&self) -> String {
153        (**self.journal_id.load()).clone()
154    }
155
156    /// Clients / input owner / pending prompts as last published.
157    pub fn presence(&self) -> Presence {
158        **self.presence.load()
159    }
160
161    /// Current session name (create `--name` or a later `saveas`).
162    pub fn name(&self) -> Option<String> {
163        (**self.name.load()).clone()
164    }
165
166    /// The session's `/keepwarm` override (`off`/`auto`/`<N>h`), if any.
167    pub fn cache_keepwarm_override(&self) -> Option<String> {
168        self.cache_keepwarm
169            .lock()
170            .unwrap_or_else(std::sync::PoisonError::into_inner)
171            .map(|m| m.to_string())
172    }
173
174    /// `meta()` with the live cells filled in (lifecycle, clients, owner,
175    /// awaiting_input, journal_id) — what listings show.
176    pub fn meta_live(&self) -> SessionMeta {
177        let mut m = (*self.meta).clone();
178        let p = self.presence();
179        m.lifecycle = self.lifecycle();
180        m.clients = p.clients;
181        m.input_owner = p.input_owner;
182        m.awaiting_input = p.awaiting_input;
183        m.journal_id = self.journal_id();
184        m.name = self.name();
185        m
186    }
187
188    pub fn subscribe(&self) -> broadcast::Receiver<Envelope> {
189        self.events.subscribe()
190    }
191
192    /// F23: the retry config carried by a lock-held placeholder.
193    pub fn placeholder_config(&self) -> Option<&SessionConfig> {
194        self.placeholder_config.as_deref()
195    }
196
197    pub(crate) fn with_placeholder_config(mut self, cfg: SessionConfig) -> Self {
198        self.placeholder_config = Some(Arc::new(cfg));
199        self
200    }
201
202    pub fn meta(&self) -> &SessionMeta {
203        &self.meta
204    }
205
206    pub fn view(&self) -> arc_swap::Guard<Arc<RuntimeView>> {
207        self.view.load()
208    }
209
210    /// Whether the actor is still alive (its command receiver exists).
211    pub fn is_alive(&self) -> bool {
212        !self.cmd_tx.is_closed()
213    }
214
215    /// Resolves once the actor is gone (command receiver dropped).
216    pub async fn closed(&self) {
217        self.cmd_tx.closed().await
218    }
219
220    /// EchoActor: `Submit{text}` → `TurnStarted`, `Stream(Text(text))`,
221    /// `Stream(Done)`, `Conversation` — for B's socket tests before A1 lands.
222    #[cfg(any(test, feature = "testing"))]
223    pub fn echo_for_test(id: SessionId) -> (Self, tokio::task::JoinHandle<()>) {
224        echo::spawn(id)
225    }
226}
227
228#[cfg(any(test, feature = "testing"))]
229pub mod echo {
230    //! Minimal stand-in actor. Honours the envelope/seq contract, `Attach`,
231    //! `Detach`, `Submit`, `Steer`, `Cancel`, `Query{View}`, `End`;
232    //! everything else is acknowledged with a `SystemNotice`.
233
234    use std::collections::HashMap;
235
236    use agent_core::reasoning::ReasoningLevel;
237
238    use super::*;
239    use crate::session::types::*;
240    use crate::{LlmEvent, SessionEvent, StreamEvent};
241
242    pub fn meta_for(id: &SessionId) -> SessionMeta {
243        SessionMeta {
244            id: id.clone(),
245            name: None,
246            model: "echo".into(),
247            cwd: None,
248            created_at: chrono::Utc::now(),
249            continued: false,
250            continue_info: None,
251            host_pid: std::process::id(),
252            lifecycle: SessionLifecycle::Live,
253            clients: 0,
254            input_owner: None,
255            awaiting_input: 0,
256            journal_id: id.0.clone(),
257            locked_by: None,
258        }
259    }
260
261    pub fn view_for() -> RuntimeView {
262        RuntimeView {
263            model: "echo".into(),
264            thinking_level: "off".into(),
265            reasoning_level: ReasoningLevel::Off,
266            is_reasoning_explicit: false,
267            thinking_budget: 0,
268            context_window: 200_000,
269            system_prompt: None,
270            compaction_model: "echo".into(),
271            api_retries: 0,
272            subagent_timeout: 0,
273            max_tool_output: 4096,
274            bash_timeout: 30,
275            bash_max_timeout: 300,
276            prompt_generation: 0,
277            hook_handler_count: 0,
278            prompt_inspection: None,
279        }
280    }
281
282    pub fn spawn(id: SessionId) -> (SessionHandle, tokio::task::JoinHandle<()>) {
283        let (handle, ep) = SessionHandle::new(meta_for(&id), view_for());
284        let task = tokio::spawn(run(id, ep));
285        (handle, task)
286    }
287
288    struct Echo {
289        id: SessionId,
290        events: broadcast::Sender<Envelope>,
291        view: Arc<arc_swap::ArcSwap<RuntimeView>>,
292        seq: u64,
293        next_client: u64,
294        clients: HashMap<ClientId, ClientMeta>,
295        conv: ConversationSnapshot,
296    }
297
298    impl Echo {
299        fn emit(&mut self, event: SessionEventWire) {
300            let env = Envelope {
301                session_id: self.id.clone(),
302                seq: self.seq,
303                ts: chrono::Utc::now(),
304                event,
305            };
306            self.seq += 1;
307            // No receivers is not an error: streams are not tied to clients.
308            let _ = self.events.send(env);
309        }
310
311        fn conv_snapshot(&self) -> ConversationSnapshot {
312            let mut c = self.conv.clone();
313            c.messages_len = c.api_messages.len();
314            c
315        }
316
317        /// Mirrors `SessionActor::snapshot_for`: Digest clients get the
318        /// tail, not the history.
319        fn snapshot_for(&self, client: &ClientMeta) -> AttachSnapshot {
320            let mut conversation = self.conv_snapshot();
321            let display_tail = (client.history == HistoryMode::Digest).then(|| {
322                let t = crate::session::display::display_tail(&conversation.api_messages, client.tail_items);
323                conversation.api_messages = Vec::new();
324                t
325            });
326            AttachSnapshot {
327                meta: meta_for(&self.id),
328                view: (**self.view.load()).clone(),
329                conversation,
330                streaming: false,
331                replay: Vec::new(),
332                pending_prompts: Vec::new(),
333                clients: self.clients.iter().map(|(c, m)| (*c, m.kind)).collect(),
334                input_owner: None,
335                display_tail,
336            }
337        }
338
339        fn handle(&mut self, cmd: SessionCommand) -> bool {
340            match cmd {
341                SessionCommand::Attach { client, mode } => {
342                    let cid = ClientId(self.next_client);
343                    self.next_client += 1;
344                    let kind = client.kind;
345                    let snapshot_meta = client.clone();
346                    self.clients.insert(cid, client);
347                    if mode != AttachMode::Mirror {
348                        self.emit(SessionEventWire::SystemNotice(format!(
349                            "attach mode {mode:?} not supported yet; using mirror"
350                        )));
351                    }
352                    let snapshot = self.snapshot_for(&snapshot_meta);
353                    self.emit(SessionEventWire::Attached {
354                        client: cid,
355                        snapshot,
356                    });
357                    self.emit(SessionEventWire::ClientJoined { client: cid, kind });
358                }
359                SessionCommand::Detach { client } => {
360                    if self.clients.remove(&client).is_some() {
361                        self.emit(SessionEventWire::ClientLeft { client });
362                    }
363                }
364                SessionCommand::Submit { text, .. } => {
365                    let baseline = self.conv.api_messages.len();
366                    self.conv.api_messages.push(Arc::new(serde_json::json!({
367                        "role": "user", "content": text
368                    })));
369                    self.emit(SessionEventWire::TurnStarted {
370                        turn_baseline: baseline,
371                        trigger: TurnTrigger::User,
372                        user_text: None,
373                    });
374                    self.emit(SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::Text(
375                        text.clone(),
376                    ))));
377                    self.conv.api_messages.push(Arc::new(serde_json::json!({
378                        "role": "assistant", "content": text
379                    })));
380                    self.emit(SessionEventWire::Stream(StreamEvent::Session(
381                        SessionEvent::MessageHistory(self.conv.api_messages.clone()),
382                    )));
383                    self.emit(SessionEventWire::Stream(StreamEvent::Session(
384                        SessionEvent::Done,
385                    )));
386                    let snap = self.conv_snapshot();
387                    self.emit(SessionEventWire::Conversation(snap));
388                }
389                SessionCommand::Steer { text } => {
390                    self.conv.queued_message = Some(text.clone());
391                    self.emit(SessionEventWire::Steered {
392                        text,
393                        delivered: false,
394                    });
395                }
396                SessionCommand::Cancel => {
397                    let snap = self.conv_snapshot();
398                    self.emit(SessionEventWire::Conversation(snap));
399                }
400                SessionCommand::Query { id, query } => {
401                    let value = match query {
402                        SessionQuery::View => {
403                            serde_json::to_value(&**self.view.load()).unwrap_or_default()
404                        }
405                        SessionQuery::Messages => {
406                            serde_json::to_value(&self.conv.api_messages).unwrap_or_default()
407                        }
408                        SessionQuery::DisplayTail { items } => serde_json::to_value(
409                            super::super::display::display_tail(&self.conv.api_messages, items),
410                        )
411                        .unwrap_or_default(),
412                        other => serde_json::json!({ "unsupported": format!("{other:?}") }),
413                    };
414                    self.emit(SessionEventWire::QueryResult { id, value });
415                }
416                SessionCommand::Set { id, setting } => {
417                    let mut view = (**self.view.load()).clone();
418                    if let SessionSetting::Model { model } = &setting {
419                        view.model = model.clone();
420                    }
421                    self.view.store(Arc::new(view.clone()));
422                    self.emit(SessionEventWire::SettingChanged(SettingApplied {
423                        id,
424                        setting: "echo".into(),
425                        ok: true,
426                        message: None,
427                        view,
428                        clamp: None,
429                    }));
430                }
431                SessionCommand::End { reason } => {
432                    self.emit(SessionEventWire::Ended { reason });
433                    return false;
434                }
435                other => {
436                    self.emit(SessionEventWire::SystemNotice(format!("echo: {other:?}")));
437                }
438            }
439            true
440        }
441    }
442
443    async fn run(id: SessionId, mut ep: SessionEndpoints) {
444        let mut echo = Echo {
445            id,
446            events: ep.events,
447            view: ep.view,
448            seq: 0,
449            next_client: 1,
450            clients: HashMap::new(),
451            conv: ConversationSnapshot::default(),
452        };
453        while let Some(Addressed { cmd, .. }) = ep.cmd_rx.recv().await {
454            if !echo.handle(cmd) {
455                break;
456            }
457        }
458    }
459}
460
461/// F23: a Parked placeholder for sessions whose journal lock is held by
462/// another process at reload time. Refuses Attach while held; retries
463/// on every Attach so that once the holder exits, the next attach works.
464pub mod locked_placeholder {
465    use std::sync::atomic::Ordering;
466
467    use super::*;
468    use crate::session::types::*;
469    
470
471    /// Spawn a placeholder actor for session `id` whose journal is locked
472    /// by `holder_desc` (e.g. `"pid 1234 (tui)"`). The handle starts
473    /// Parked; Attach emits `AttachRefused` naming the holder. The caller
474    /// registers it in the session map under the SAME id — no alias.
475    pub fn spawn(
476        id: SessionId,
477        config: SessionConfig,
478        holder_desc: String,
479    ) -> (SessionHandle, tokio::task::JoinHandle<()>) {
480        let meta = SessionMeta {
481            id: id.clone(),
482            name: config.name.clone(),
483            model: config.model_override.clone().unwrap_or_else(|| "unknown".into()),
484            cwd: config.cwd.clone(),
485            created_at: chrono::Utc::now(),
486            continued: true,
487            continue_info: None,
488            host_pid: std::process::id(),
489            lifecycle: SessionLifecycle::Parked,
490            clients: 0,
491            input_owner: None,
492            awaiting_input: 0,
493            journal_id: id.0.clone(),
494            locked_by: Some(holder_desc.clone()),
495        };
496        let view = crate::session::view::RuntimeView {
497            model: config.model_override.clone().unwrap_or_else(|| "unknown".into()),
498            thinking_level: "off".into(),
499            reasoning_level: agent_core::reasoning::ReasoningLevel::Off,
500            is_reasoning_explicit: false,
501            thinking_budget: 0,
502            context_window: 200_000,
503            system_prompt: None,
504            compaction_model: "unknown".into(),
505            api_retries: 0,
506            subagent_timeout: 0,
507            max_tool_output: 4096,
508            bash_timeout: 30,
509            bash_max_timeout: 300,
510            prompt_generation: 0,
511            hook_handler_count: 0,
512            prompt_inspection: None,
513        };
514        let (handle, ep) = SessionHandle::new(meta, view);
515        let handle = handle.with_placeholder_config(config.clone());
516        // Start as Parked.
517        ep.lifecycle.store(SessionLifecycle::Parked as u8, Ordering::Release);
518        let task = tokio::spawn(run(id, ep, config, holder_desc));
519        (handle, task)
520    }
521
522    async fn run(
523        id: SessionId,
524        mut ep: SessionEndpoints,
525        _config: SessionConfig,
526        holder_desc: String,
527    ) {
528        let mut seq: u64 = 0;
529        let emit = |seq: &mut u64, events: &broadcast::Sender<Envelope>, event: SessionEventWire| {
530            let env = Envelope {
531                session_id: id.clone(),
532                seq: *seq,
533                ts: chrono::Utc::now(),
534                event,
535            };
536            *seq += 1;
537            let _ = events.send(env);
538        };
539
540        while let Some(Addressed { cmd, .. }) = ep.cmd_rx.recv().await {
541            match cmd {
542                SessionCommand::Attach { .. } => {
543                    let msg = format!(
544                        "session {}: journal locked by {} \
545                         — close it or use synaps --attach from that process",
546                        id.as_str(),
547                        holder_desc
548                    );
549                    emit(&mut seq, &ep.events, SessionEventWire::AttachRefused { message: msg });
550                }
551                SessionCommand::Park => {
552                    // Already parked, no-op.
553                }
554                SessionCommand::End { reason } => {
555                    emit(&mut seq, &ep.events, SessionEventWire::Ended { reason });
556                    break;
557                }
558                SessionCommand::Detach { client } => {
559                    emit(&mut seq, &ep.events, SessionEventWire::ClientLeft { client });
560                }
561                SessionCommand::Query { id: qid, .. } => {
562                    emit(
563                        &mut seq,
564                        &ep.events,
565                        SessionEventWire::QueryResult {
566                            id: qid,
567                            value: serde_json::json!({
568                                "locked_by": holder_desc,
569                                "streaming": false,
570                                "pending_prompts": 0,
571                            }),
572                        },
573                    );
574                }
575                _ => {
576                    emit(
577                        &mut seq,
578                        &ep.events,
579                        SessionEventWire::SystemNotice(format!(
580                            "session {}: journal locked by {} — command ignored",
581                            id.as_str(),
582                            holder_desc
583                        )),
584                    );
585                }
586            }
587        }
588    }
589}
590
591#[cfg(test)]
592mod tests {
593    use super::*;
594    use crate::session::types::*;
595    use crate::{LlmEvent, SessionEvent, StreamEvent};
596
597    async fn next(rx: &mut broadcast::Receiver<Envelope>) -> Envelope {
598        tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
599            .await
600            .expect("timely")
601            .expect("open")
602    }
603
604    #[tokio::test]
605    async fn echo_actor_submit_emits_turn_in_order() {
606        let (handle, task) = SessionHandle::echo_for_test(SessionId::from("echo-1"));
607        let mut rx = handle.subscribe();
608        handle
609            .send(SessionCommand::Submit {
610                text: "hello".into(),
611                attachments: vec![],
612            })
613            .await
614            .unwrap();
615
616        let e0 = next(&mut rx).await;
617        assert!(matches!(
618            e0.event,
619            SessionEventWire::TurnStarted {
620                turn_baseline: 0,
621                trigger: TurnTrigger::User,
622                user_text: None,
623            }
624        ));
625        let e1 = next(&mut rx).await;
626        match e1.event {
627            SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::Text(t))) => assert_eq!(t, "hello"),
628            other => panic!("unexpected {other:?}"),
629        }
630        let e2 = next(&mut rx).await;
631        assert!(matches!(
632            e2.event,
633            SessionEventWire::Stream(StreamEvent::Session(SessionEvent::MessageHistory(_)))
634        ));
635        let e3 = next(&mut rx).await;
636        assert!(matches!(
637            e3.event,
638            SessionEventWire::Stream(StreamEvent::Session(SessionEvent::Done))
639        ));
640        let e4 = next(&mut rx).await;
641        match e4.event {
642            SessionEventWire::Conversation(c) => assert_eq!(c.api_messages.len(), 2),
643            other => panic!("unexpected {other:?}"),
644        }
645        // gapless seq, same session id on every envelope
646        assert_eq!(
647            [e0.seq, e1.seq, e2.seq, e3.seq, e4.seq],
648            [0, 1, 2, 3, 4]
649        );
650        assert!(e4.session_id == SessionId::from("echo-1"));
651
652        handle
653            .send(SessionCommand::End {
654                reason: EndReason::ClientQuit,
655            })
656            .await
657            .unwrap();
658        let e5 = next(&mut rx).await;
659        assert!(matches!(
660            e5.event,
661            SessionEventWire::Ended {
662                reason: EndReason::ClientQuit
663            }
664        ));
665        task.await.unwrap();
666        assert!(!handle.is_alive());
667        assert!(matches!(
668            handle.send(SessionCommand::Save).await,
669            Err(TransportError::Closed)
670        ));
671    }
672
673    #[tokio::test]
674    async fn handle_view_swaps_on_set() {
675        let (handle, task) = SessionHandle::echo_for_test(SessionId::from("echo-2"));
676        let mut rx = handle.subscribe();
677        assert_eq!(handle.view().model, "echo");
678        handle
679            .send(SessionCommand::Set {
680                id: 0,
681                setting: SessionSetting::Model { model: "other".into() },
682            })
683            .await
684            .unwrap();
685        let e = next(&mut rx).await;
686        assert!(matches!(e.event, SessionEventWire::SettingChanged(_)));
687        assert_eq!(handle.view().model, "other");
688        drop(handle);
689        task.await.unwrap();
690    }
691}
agent_engine/session/handle.rs

agent_engine/session/
handle.rs
