agent_engine/session/

handle.rs

//! `SessionHandle` — the in-process address of a session (PLAN-phase2 §2.6).
//!
//! `EngineHost::create_session` (A1) hands one out per session; every
//! `ClientTransport` is built on it. `echo_for_test` ships a stand-in actor
//! so package B can build the wire before the real `SessionActor` exists.

use std::sync::atomic::{AtomicU8, Ordering};
use std::sync::Arc;

use tokio::sync::{broadcast, mpsc};

use super::transport::TransportError;
use super::types::{Addressed, ClientId, Envelope, SessionCommand, SessionConfig, SessionId, SessionLifecycle, SessionMeta};

/// Live client accounting published by the actor (B4 `sessions()` listing).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Presence {
    pub clients: usize,
    pub input_owner: Option<ClientId>,
    pub awaiting_input: usize,
}
use super::view::RuntimeView;

/// Bounded command queue depth (rpc `WRITER_CHAN_CAP` precedent).
pub const CMD_CHAN_CAP: usize = 256;
/// Default broadcast capacity; `SYNAPS_SESSION_EVENTS_CAP` overrides.
pub const DEFAULT_EVENTS_CAP: usize = 1024;

/// Broadcast capacity from `SYNAPS_SESSION_EVENTS_CAP` (min 16) or the default.
pub fn events_cap() -> usize {
    std::env::var("SYNAPS_SESSION_EVENTS_CAP")
        .ok()
        .and_then(|s| s.parse::<usize>().ok())
        .map(|n| n.max(16))
        .unwrap_or(DEFAULT_EVENTS_CAP)
}

#[derive(Clone)]
pub struct SessionHandle {
    pub id: SessionId,
    cmd_tx: mpsc::Sender<Addressed>,
    events: broadcast::Sender<Envelope>,
    meta: Arc<SessionMeta>,
    /// Sync, lock-free read of the last published RuntimeView (§2.7).
    view: Arc<arc_swap::ArcSwap<RuntimeView>>,
    /// Written by the actor (Live/Parking/Parked/Ending).
    lifecycle: Arc<AtomicU8>,
    /// `conv.session.id` — changes on LinkedSuccessor compaction.
    journal_id: Arc<arc_swap::ArcSwap<String>>,
    /// Clients / input owner / pending prompts, written by the actor.
    presence: Arc<arc_swap::ArcSwap<Presence>>,
    /// Session name — `--name` at create / `/cmd saveas` later. Written by
    /// the actor so listings and `--continue <name>` see renames live.
    name: Arc<arc_swap::ArcSwap<Option<String>>>,
    /// F23: only on a lock-held placeholder — the config to retry the real
    /// `create` with on the next attach. `None` for real actors.
    placeholder_config: Option<Arc<SessionConfig>>,
}

impl std::fmt::Debug for SessionHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SessionHandle")
            .field("id", &self.id)
            .field("model", &self.meta.model)
            .finish()
    }
}

/// Actor-side ends of a session's channels, minted with the handle.
pub struct SessionEndpoints {
    pub cmd_rx: mpsc::Receiver<Addressed>,
    pub events: broadcast::Sender<Envelope>,
    pub view: Arc<arc_swap::ArcSwap<RuntimeView>>,
    pub lifecycle: Arc<AtomicU8>,
    pub journal_id: Arc<arc_swap::ArcSwap<String>>,
    pub presence: Arc<arc_swap::ArcSwap<Presence>>,
    pub name: Arc<arc_swap::ArcSwap<Option<String>>>,
}

impl SessionHandle {
    /// Mint a handle + the actor's ends. Used by `SessionActor` (A1) and the
    /// test `EchoActor`.
    pub fn new(meta: SessionMeta, view: RuntimeView) -> (Self, SessionEndpoints) {
        let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHAN_CAP);
        let (events, _) = broadcast::channel(events_cap());
        let view = Arc::new(arc_swap::ArcSwap::from_pointee(view));
        let lifecycle = Arc::new(AtomicU8::new(SessionLifecycle::Live as u8));
        let journal_id = Arc::new(arc_swap::ArcSwap::from_pointee(meta.id.0.clone()));
        let presence = Arc::new(arc_swap::ArcSwap::from_pointee(Presence::default()));
        let name = Arc::new(arc_swap::ArcSwap::from_pointee(meta.name.clone()));
        let handle = Self {
            id: meta.id.clone(),
            cmd_tx,
            events: events.clone(),
            meta: Arc::new(meta),
            view: Arc::clone(&view),
            lifecycle: Arc::clone(&lifecycle),
            journal_id: Arc::clone(&journal_id),
            presence: Arc::clone(&presence),
            name: Arc::clone(&name),
            placeholder_config: None,
        };
        (
            handle,
            SessionEndpoints {
                cmd_rx,
                events,
                view,
                lifecycle,
                journal_id,
                presence,
                name,
            },
        )
    }

    /// Send a host-originated command (`from: None`). `Backpressure` when
    /// the queue is full (never drop silently); `Closed` once the actor is
    /// gone.
    pub async fn send(&self, cmd: SessionCommand) -> Result<(), TransportError> {
        self.send_addressed(Addressed { from: None, cmd })
    }

    /// Send on behalf of an attached client (ownership is enforced by the
    /// actor, B1).
    pub async fn send_from(&self, from: ClientId, cmd: SessionCommand) -> Result<(), TransportError> {
        self.send_addressed(Addressed { from: Some(from), cmd })
    }

    fn send_addressed(&self, cmd: Addressed) -> Result<(), TransportError> {
        match self.cmd_tx.try_send(cmd) {
            Ok(()) => Ok(()),
            Err(mpsc::error::TrySendError::Full(_)) => Err(TransportError::Backpressure),
            Err(mpsc::error::TrySendError::Closed(_)) => Err(TransportError::Closed),
        }
    }

    /// Park/unpark state as last written by the actor.
    pub fn lifecycle(&self) -> SessionLifecycle {
        SessionLifecycle::from_u8(self.lifecycle.load(Ordering::Acquire))
    }

    /// Current `conv.session.id` (= `id` until a LinkedSuccessor compaction).
    pub fn journal_id(&self) -> String {
        (**self.journal_id.load()).clone()
    }

    /// Clients / input owner / pending prompts as last published.
    pub fn presence(&self) -> Presence {
        **self.presence.load()
    }

    /// Current session name (create `--name` or a later `saveas`).
    pub fn name(&self) -> Option<String> {
        (**self.name.load()).clone()
    }

    /// `meta()` with the live cells filled in (lifecycle, clients, owner,
    /// awaiting_input, journal_id) — what listings show.
    pub fn meta_live(&self) -> SessionMeta {
        let mut m = (*self.meta).clone();
        let p = self.presence();
        m.lifecycle = self.lifecycle();
        m.clients = p.clients;
        m.input_owner = p.input_owner;
        m.awaiting_input = p.awaiting_input;
        m.journal_id = self.journal_id();
        m.name = self.name();
        m
    }

    pub fn subscribe(&self) -> broadcast::Receiver<Envelope> {
        self.events.subscribe()
    }

    /// F23: the retry config carried by a lock-held placeholder.
    pub fn placeholder_config(&self) -> Option<&SessionConfig> {
        self.placeholder_config.as_deref()
    }

    pub(crate) fn with_placeholder_config(mut self, cfg: SessionConfig) -> Self {
        self.placeholder_config = Some(Arc::new(cfg));
        self
    }

    pub fn meta(&self) -> &SessionMeta {
        &self.meta
    }

    pub fn view(&self) -> arc_swap::Guard<Arc<RuntimeView>> {
        self.view.load()
    }

    /// Whether the actor is still alive (its command receiver exists).
    pub fn is_alive(&self) -> bool {
        !self.cmd_tx.is_closed()
    }

    /// Resolves once the actor is gone (command receiver dropped).
    pub async fn closed(&self) {
        self.cmd_tx.closed().await
    }

    /// EchoActor: `Submit{text}` → `TurnStarted`, `Stream(Text(text))`,
    /// `Stream(Done)`, `Conversation` — for B's socket tests before A1 lands.
    #[cfg(any(test, feature = "testing"))]
    pub fn echo_for_test(id: SessionId) -> (Self, tokio::task::JoinHandle<()>) {
        echo::spawn(id)
    }
}

#[cfg(any(test, feature = "testing"))]
pub mod echo {
    //! Minimal stand-in actor. Honours the envelope/seq contract, `Attach`,
    //! `Detach`, `Submit`, `Steer`, `Cancel`, `Query{View}`, `End`;
    //! everything else is acknowledged with a `SystemNotice`.

    use std::collections::HashMap;

    use agent_core::reasoning::ReasoningLevel;

    use super::*;
    use crate::session::types::*;
    use crate::{LlmEvent, SessionEvent, StreamEvent};

    pub fn meta_for(id: &SessionId) -> SessionMeta {
        SessionMeta {
            id: id.clone(),
            name: None,
            model: "echo".into(),
            cwd: None,
            created_at: chrono::Utc::now(),
            continued: false,
            continue_info: None,
            host_pid: std::process::id(),
            lifecycle: SessionLifecycle::Live,
            clients: 0,
            input_owner: None,
            awaiting_input: 0,
            journal_id: id.0.clone(),
            locked_by: None,
        }
    }

    pub fn view_for() -> RuntimeView {
        RuntimeView {
            model: "echo".into(),
            thinking_level: "off".into(),
            reasoning_level: ReasoningLevel::Off,
            is_reasoning_explicit: false,
            thinking_budget: 0,
            context_window: 200_000,
            system_prompt: None,
            compaction_model: "echo".into(),
            api_retries: 0,
            subagent_timeout: 0,
            max_tool_output: 4096,
            bash_timeout: 30,
            bash_max_timeout: 300,
            prompt_generation: 0,
            hook_handler_count: 0,
            prompt_inspection: None,
        }
    }

    pub fn spawn(id: SessionId) -> (SessionHandle, tokio::task::JoinHandle<()>) {
        let (handle, ep) = SessionHandle::new(meta_for(&id), view_for());
        let task = tokio::spawn(run(id, ep));
        (handle, task)
    }

    struct Echo {
        id: SessionId,
        events: broadcast::Sender<Envelope>,
        view: Arc<arc_swap::ArcSwap<RuntimeView>>,
        seq: u64,
        next_client: u64,
        clients: HashMap<ClientId, ClientMeta>,
        conv: ConversationSnapshot,
    }

    impl Echo {
        fn emit(&mut self, event: SessionEventWire) {
            let env = Envelope {
                session_id: self.id.clone(),
                seq: self.seq,
                ts: chrono::Utc::now(),
                event,
            };
            self.seq += 1;
            // No receivers is not an error: streams are not tied to clients.
            let _ = self.events.send(env);
        }

        fn conv_snapshot(&self) -> ConversationSnapshot {
            let mut c = self.conv.clone();
            c.messages_len = c.api_messages.len();
            c
        }

        /// Mirrors `SessionActor::snapshot_for`: Digest clients get the
        /// tail, not the history.
        fn snapshot_for(&self, client: &ClientMeta) -> AttachSnapshot {
            let mut conversation = self.conv_snapshot();
            let display_tail = (client.history == HistoryMode::Digest).then(|| {
                let t = crate::session::display::display_tail(&conversation.api_messages, client.tail_items);
                conversation.api_messages = Vec::new();
                t
            });
            AttachSnapshot {
                meta: meta_for(&self.id),
                view: (**self.view.load()).clone(),
                conversation,
                streaming: false,
                replay: Vec::new(),
                pending_prompts: Vec::new(),
                clients: self.clients.iter().map(|(c, m)| (*c, m.kind)).collect(),
                input_owner: None,
                display_tail,
            }
        }

        fn handle(&mut self, cmd: SessionCommand) -> bool {
            match cmd {
                SessionCommand::Attach { client, mode } => {
                    let cid = ClientId(self.next_client);
                    self.next_client += 1;
                    let kind = client.kind;
                    let snapshot_meta = client.clone();
                    self.clients.insert(cid, client);
                    if mode != AttachMode::Mirror {
                        self.emit(SessionEventWire::SystemNotice(format!(
                            "attach mode {mode:?} not supported yet; using mirror"
                        )));
                    }
                    let snapshot = self.snapshot_for(&snapshot_meta);
                    self.emit(SessionEventWire::Attached {
                        client: cid,
                        snapshot,
                    });
                    self.emit(SessionEventWire::ClientJoined { client: cid, kind });
                }
                SessionCommand::Detach { client } => {
                    if self.clients.remove(&client).is_some() {
                        self.emit(SessionEventWire::ClientLeft { client });
                    }
                }
                SessionCommand::Submit { text, .. } => {
                    let baseline = self.conv.api_messages.len();
                    self.conv.api_messages.push(Arc::new(serde_json::json!({
                        "role": "user", "content": text
                    })));
                    self.emit(SessionEventWire::TurnStarted {
                        turn_baseline: baseline,
                        trigger: TurnTrigger::User,
                        user_text: None,
                    });
                    self.emit(SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::Text(
                        text.clone(),
                    ))));
                    self.conv.api_messages.push(Arc::new(serde_json::json!({
                        "role": "assistant", "content": text
                    })));
                    self.emit(SessionEventWire::Stream(StreamEvent::Session(
                        SessionEvent::MessageHistory(self.conv.api_messages.clone()),
                    )));
                    self.emit(SessionEventWire::Stream(StreamEvent::Session(
                        SessionEvent::Done,
                    )));
                    let snap = self.conv_snapshot();
                    self.emit(SessionEventWire::Conversation(snap));
                }
                SessionCommand::Steer { text } => {
                    self.conv.queued_message = Some(text.clone());
                    self.emit(SessionEventWire::Steered {
                        text,
                        delivered: false,
                    });
                }
                SessionCommand::Cancel => {
                    let snap = self.conv_snapshot();
                    self.emit(SessionEventWire::Conversation(snap));
                }
                SessionCommand::Query { id, query } => {
                    let value = match query {
                        SessionQuery::View => {
                            serde_json::to_value(&**self.view.load()).unwrap_or_default()
                        }
                        SessionQuery::Messages => {
                            serde_json::to_value(&self.conv.api_messages).unwrap_or_default()
                        }
                        SessionQuery::DisplayTail { items } => serde_json::to_value(
                            super::super::display::display_tail(&self.conv.api_messages, items),
                        )
                        .unwrap_or_default(),
                        other => serde_json::json!({ "unsupported": format!("{other:?}") }),
                    };
                    self.emit(SessionEventWire::QueryResult { id, value });
                }
                SessionCommand::Set { id, setting } => {
                    let mut view = (**self.view.load()).clone();
                    if let SessionSetting::Model { model } = &setting {
                        view.model = model.clone();
                    }
                    self.view.store(Arc::new(view.clone()));
                    self.emit(SessionEventWire::SettingChanged(SettingApplied {
                        id,
                        setting: "echo".into(),
                        ok: true,
                        message: None,
                        view,
                        clamp: None,
                    }));
                }
                SessionCommand::End { reason } => {
                    self.emit(SessionEventWire::Ended { reason });
                    return false;
                }
                other => {
                    self.emit(SessionEventWire::SystemNotice(format!("echo: {other:?}")));
                }
            }
            true
        }
    }

    async fn run(id: SessionId, mut ep: SessionEndpoints) {
        let mut echo = Echo {
            id,
            events: ep.events,
            view: ep.view,
            seq: 0,
            next_client: 1,
            clients: HashMap::new(),
            conv: ConversationSnapshot::default(),
        };
        while let Some(Addressed { cmd, .. }) = ep.cmd_rx.recv().await {
            if !echo.handle(cmd) {
                break;
            }
        }
    }
}

/// F23: a Parked placeholder for sessions whose journal lock is held by
/// another process at reload time. Refuses Attach while held; retries
/// on every Attach so that once the holder exits, the next attach works.
pub mod locked_placeholder {
    use std::sync::atomic::Ordering;

    use super::*;
    use crate::session::types::*;
    

    /// Spawn a placeholder actor for session `id` whose journal is locked
    /// by `holder_desc` (e.g. `"pid 1234 (tui)"`). The handle starts
    /// Parked; Attach emits `AttachRefused` naming the holder. The caller
    /// registers it in the session map under the SAME id — no alias.
    pub fn spawn(
        id: SessionId,
        config: SessionConfig,
        holder_desc: String,
    ) -> (SessionHandle, tokio::task::JoinHandle<()>) {
        let meta = SessionMeta {
            id: id.clone(),
            name: config.name.clone(),
            model: config.model_override.clone().unwrap_or_else(|| "unknown".into()),
            cwd: config.cwd.clone(),
            created_at: chrono::Utc::now(),
            continued: true,
            continue_info: None,
            host_pid: std::process::id(),
            lifecycle: SessionLifecycle::Parked,
            clients: 0,
            input_owner: None,
            awaiting_input: 0,
            journal_id: id.0.clone(),
            locked_by: Some(holder_desc.clone()),
        };
        let view = crate::session::view::RuntimeView {
            model: config.model_override.clone().unwrap_or_else(|| "unknown".into()),
            thinking_level: "off".into(),
            reasoning_level: agent_core::reasoning::ReasoningLevel::Off,
            is_reasoning_explicit: false,
            thinking_budget: 0,
            context_window: 200_000,
            system_prompt: None,
            compaction_model: "unknown".into(),
            api_retries: 0,
            subagent_timeout: 0,
            max_tool_output: 4096,
            bash_timeout: 30,
            bash_max_timeout: 300,
            prompt_generation: 0,
            hook_handler_count: 0,
            prompt_inspection: None,
        };
        let (handle, ep) = SessionHandle::new(meta, view);
        let handle = handle.with_placeholder_config(config.clone());
        // Start as Parked.
        ep.lifecycle.store(SessionLifecycle::Parked as u8, Ordering::Release);
        let task = tokio::spawn(run(id, ep, config, holder_desc));
        (handle, task)
    }

    async fn run(
        id: SessionId,
        mut ep: SessionEndpoints,
        _config: SessionConfig,
        holder_desc: String,
    ) {
        let mut seq: u64 = 0;
        let emit = |seq: &mut u64, events: &broadcast::Sender<Envelope>, event: SessionEventWire| {
            let env = Envelope {
                session_id: id.clone(),
                seq: *seq,
                ts: chrono::Utc::now(),
                event,
            };
            *seq += 1;
            let _ = events.send(env);
        };

        while let Some(Addressed { cmd, .. }) = ep.cmd_rx.recv().await {
            match cmd {
                SessionCommand::Attach { .. } => {
                    let msg = format!(
                        "session {}: journal locked by {} \
                         — close it or use synaps --attach from that process",
                        id.as_str(),
                        holder_desc
                    );
                    emit(&mut seq, &ep.events, SessionEventWire::AttachRefused { message: msg });
                }
                SessionCommand::Park => {
                    // Already parked, no-op.
                }
                SessionCommand::End { reason } => {
                    emit(&mut seq, &ep.events, SessionEventWire::Ended { reason });
                    break;
                }
                SessionCommand::Detach { client } => {
                    emit(&mut seq, &ep.events, SessionEventWire::ClientLeft { client });
                }
                SessionCommand::Query { id: qid, .. } => {
                    emit(
                        &mut seq,
                        &ep.events,
                        SessionEventWire::QueryResult {
                            id: qid,
                            value: serde_json::json!({
                                "locked_by": holder_desc,
                                "streaming": false,
                                "pending_prompts": 0,
                            }),
                        },
                    );
                }
                _ => {
                    emit(
                        &mut seq,
                        &ep.events,
                        SessionEventWire::SystemNotice(format!(
                            "session {}: journal locked by {} — command ignored",
                            id.as_str(),
                            holder_desc
                        )),
                    );
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::session::types::*;
    use crate::{LlmEvent, SessionEvent, StreamEvent};

    async fn next(rx: &mut broadcast::Receiver<Envelope>) -> Envelope {
        tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
            .await
            .expect("timely")
            .expect("open")
    }

    #[tokio::test]
    async fn echo_actor_submit_emits_turn_in_order() {
        let (handle, task) = SessionHandle::echo_for_test(SessionId::from("echo-1"));
        let mut rx = handle.subscribe();
        handle
            .send(SessionCommand::Submit {
                text: "hello".into(),
                attachments: vec![],
            })
            .await
            .unwrap();

        let e0 = next(&mut rx).await;
        assert!(matches!(
            e0.event,
            SessionEventWire::TurnStarted {
                turn_baseline: 0,
                trigger: TurnTrigger::User,
                user_text: None,
            }
        ));
        let e1 = next(&mut rx).await;
        match e1.event {
            SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::Text(t))) => assert_eq!(t, "hello"),
            other => panic!("unexpected {other:?}"),
        }
        let e2 = next(&mut rx).await;
        assert!(matches!(
            e2.event,
            SessionEventWire::Stream(StreamEvent::Session(SessionEvent::MessageHistory(_)))
        ));
        let e3 = next(&mut rx).await;
        assert!(matches!(
            e3.event,
            SessionEventWire::Stream(StreamEvent::Session(SessionEvent::Done))
        ));
        let e4 = next(&mut rx).await;
        match e4.event {
            SessionEventWire::Conversation(c) => assert_eq!(c.api_messages.len(), 2),
            other => panic!("unexpected {other:?}"),
        }
        // gapless seq, same session id on every envelope
        assert_eq!(
            [e0.seq, e1.seq, e2.seq, e3.seq, e4.seq],
            [0, 1, 2, 3, 4]
        );
        assert!(e4.session_id == SessionId::from("echo-1"));

        handle
            .send(SessionCommand::End {
                reason: EndReason::ClientQuit,
            })
            .await
            .unwrap();
        let e5 = next(&mut rx).await;
        assert!(matches!(
            e5.event,
            SessionEventWire::Ended {
                reason: EndReason::ClientQuit
            }
        ));
        task.await.unwrap();
        assert!(!handle.is_alive());
        assert!(matches!(
            handle.send(SessionCommand::Save).await,
            Err(TransportError::Closed)
        ));
    }

    #[tokio::test]
    async fn handle_view_swaps_on_set() {
        let (handle, task) = SessionHandle::echo_for_test(SessionId::from("echo-2"));
        let mut rx = handle.subscribe();
        assert_eq!(handle.view().model, "echo");
        handle
            .send(SessionCommand::Set {
                id: 0,
                setting: SessionSetting::Model { model: "other".into() },
            })
            .await
            .unwrap();
        let e = next(&mut rx).await;
        assert!(matches!(e.event, SessionEventWire::SettingChanged(_)));
        assert_eq!(handle.view().model, "other");
        drop(handle);
        task.await.unwrap();
    }
}