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

agent_engine/session/
handle.rs
