oino_core/session/
mod.rs

1//! `AgentSession`: agent lifecycle, persistence, prompts, queues,
2//! models, tools, compaction, retry, bash, tree navigation and the extension bindings of one
3//! session.
4//!
5//! The session is cheap to clone (`Arc` inside). State is split into small mutexes that are
6//! never held across an `.await` on provider, tool or extension calls (architecture §7).
7
8mod bash;
9mod boundary;
10pub mod checkpoints;
11mod compaction_flow;
12mod feeds;
13mod frontend;
14mod hooks;
15mod host;
16mod listener;
17mod model;
18mod nested;
19mod prompt;
20mod queue;
21mod retry;
22mod stats;
23mod tools;
24mod tree;
25
26use std::any::Any;
27use std::future::Future;
28use std::panic::AssertUnwindSafe;
29use std::path::{Path, PathBuf};
30use std::sync::{Arc, Mutex, MutexGuard, PoisonError, RwLock};
31
32use futures::FutureExt;
33use oino_agent::{Agent, AgentOptions, StreamFn, Subscription};
34use oino_ai::transcript::SystemState;
35use oino_ai::{
36    AgentMessage, AssistantMessage, BashExecutionMessage, CustomMessage, Model, SystemMessage,
37    ThinkingLevel,
38};
39use oino_models::ModelRegistry;
40use oino_resources::{BuildSystemPromptOptions, ResourceLoader};
41use oino_session::{EntryId, SessionEntry, SessionManager};
42use oino_settings::SettingsManager;
43use serde::{Deserialize, Serialize};
44use tokio::sync::watch;
45use tokio_util::sync::CancellationToken;
46
47use crate::events::{AgentSessionEvent, ClientEvent, ExtensionErrorEvent, OinoEvent, RunState};
48use crate::extensions::api::{CallContext, CallKind, ExtensionMode};
49use crate::extensions::runner::{ErrorSink, ExtensionRunner, ExtensionSet};
50use crate::hub::{EventHub, EventStream};
51use crate::ui::{RegistrySink, UiBridge, UiOutput, UiRegistry, UiSink, is_dialog};
52
53pub use bash::BashResult;
54pub use checkpoints::{Checkpoint, CheckpointFile, RewindFailure, RewindResult};
55pub use feeds::{DetachedBash, ExtensionInfo, RunningBash};
56pub use frontend::{
57    AuthProviderInfo, CompletionEntry, CompletionGroup, EditorEvent, EditorIntent, FRONTEND_CALLER,
58    FrontendError, RenderDensity, RenderRequest, RenderTarget, RenderWidth,
59};
60pub use host::SlashCommandInfo;
61pub(crate) use host::check_render_override;
62pub(crate) use model::branch_refs as model_branch_refs;
63pub use model::{CycleDirection, CycleModelResult};
64pub use prompt::{InputSource, PromptDisposition, PromptOptions, StreamingBehavior};
65pub use queue::{QueueEdit, QueueKind};
66pub use stats::{ContextUsage, SessionStats, TokenTotals};
67pub use tools::{BUILTIN_PATH_PREFIX, ToolEntry, ToolInfo, ToolSource};
68pub use tree::{ForkMessage, NavigateTreeOptions, NavigateTreeResult};
69
70pub(crate) fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
71    mutex.lock().unwrap_or_else(PoisonError::into_inner)
72}
73
74/// Panic payload → message (`Box<dyn Any + Send>`).
75#[must_use]
76pub(crate) fn panic_message(panic: &(dyn Any + Send)) -> String {
77    if let Some(text) = panic.downcast_ref::<&str>() {
78        (*text).to_owned()
79    } else if let Some(text) = panic.downcast_ref::<String>() {
80        text.clone()
81    } else {
82        "unknown panic".to_owned()
83    }
84}
85
86/// Kinds of session-spawned tasks covered by panic isolation
87/// ([`AgentSession::spawn_isolated`], X-PANIC).
88#[derive(Clone, Copy, Debug, PartialEq, Eq)]
89pub enum IsolatedTask {
90    /// The agent drive task (`drive_run`) and its deferred follow-up starts: a panic
91    /// resets the run book so the session accepts a new prompt.
92    Drive,
93    /// Child event forwarding (`forward_child_events`).
94    ChildEvents,
95    /// The arbiter wait-source poll timer.
96    ArbiterPoll,
97}
98
99impl IsolatedTask {
100    /// Task name used in logs.
101    #[must_use]
102    pub const fn name(self) -> &'static str {
103        match self {
104            Self::Drive => "drive_run",
105            Self::ChildEvents => "forward_child_events",
106            Self::ArbiterPoll => "arbiter_poll",
107        }
108    }
109}
110
111/// `getCurrentSystemMessage` over agent messages.
112#[must_use]
113pub fn current_system_message(messages: &[AgentMessage]) -> Option<SystemMessage> {
114    SystemState::fold(messages.iter().filter_map(|message| match message {
115        AgentMessage::System(system) => Some(system),
116        _ => None,
117    }))
118    .into_message()
119}
120
121/// Failure of a session operation (fixed error texts).
122#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
123#[non_exhaustive]
124pub enum SessionOpError {
125    /// `prompt` while streaming without `streaming_behavior`.
126    #[error(
127        "Agent is already processing. Specify streamingBehavior ('steer' or 'followUp') to queue the message."
128    )]
129    StreamingWithoutBehavior,
130    /// No model selected.
131    #[error("{0}")]
132    NoModel(String),
133    /// No credentials for the provider.
134    #[error("{message}")]
135    NoApiKey {
136        /// Provider id.
137        provider: String,
138        /// Message.
139        message: String,
140    },
141    /// An extension command cannot be queued.
142    #[error(
143        "Extension command \"/{0}\" cannot be queued. Use prompt() or execute the command when not streaming."
144    )]
145    CommandNotAllowed(String),
146    /// Prompt during manual compaction.
147    #[error(
148        "Cannot submit a prompt while compaction is in progress. Wait for compaction to finish and retry."
149    )]
150    CompactionInProgress,
151    /// Another runner holds the single-writer lease and did not respond to the takeover
152    /// request (D-LOCK / D-TAKEOVER); the text names the holder.
153    #[error("{0}")]
154    SessionLocked(String),
155    /// Session-replacing operations need a command context.
156    #[error("This operation is only available in command and shortcut contexts")]
157    CommandContextRequired,
158    /// Any other failure.
159    #[error("{0}")]
160    Failed(String),
161}
162
163/// What an `agent_before_settle` / `turn_end` run ended with.
164#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
165#[serde(rename_all = "lowercase")]
166pub enum ActivityOutcome {
167    /// Normal end.
168    #[default]
169    Completed,
170    /// Aborted.
171    Aborted,
172    /// Error.
173    Error,
174}
175
176/// Mutable run bookkeeping (private fields).
177#[allow(
178    clippy::struct_excessive_bools,
179    reason = "mirrors the private AgentSession flags one-to-one"
180)]
181#[derive(Default)]
182pub(crate) struct RunBook {
183    pub(crate) steering: Vec<String>,
184    pub(crate) follow_up: Vec<String>,
185    pub(crate) pending_next_turn: Vec<CustomMessage>,
186    pub(crate) pending_custom: Vec<CustomMessage>,
187    pub(crate) pending_bash: Vec<BashExecutionMessage>,
188    pub(crate) run_active: bool,
189    pub(crate) abort_requested: bool,
190    pub(crate) overflow_recovery_attempted: bool,
191    pub(crate) retry_attempt: u32,
192    pub(crate) failed_response: Option<AssistantMessage>,
193    pub(crate) last_assistant: Option<AssistantMessage>,
194    pub(crate) last_outcome: ActivityOutcome,
195    pub(crate) turn_index: u64,
196    pub(crate) is_before_settle: bool,
197    pub(crate) abort_during_before_settle: bool,
198    pub(crate) emitting_settled: bool,
199    pub(crate) boundary_dispatched: bool,
200    pub(crate) run_prompt_options: Option<BuildSystemPromptOptions>,
201    /// Entry id of the latest persisted assistant message.
202    pub(crate) last_assistant_entry: Option<EntryId>,
203    /// Entry ids of persisted tool results by tool-call id (current run).
204    pub(crate) tool_result_entries: Vec<(String, EntryId)>,
205}
206
207/// Cancellation handles of long operations.
208#[derive(Default)]
209pub(crate) struct Cancels {
210    pub(crate) compaction: Option<CancellationToken>,
211    pub(crate) auto_compaction: Option<CancellationToken>,
212    pub(crate) branch_summary: Option<CancellationToken>,
213    pub(crate) retry: Option<CancellationToken>,
214    pub(crate) bash: Vec<(u64, CancellationToken)>,
215    pub(crate) next_bash: u64,
216}
217
218/// Inputs of [`AgentSession::new`].
219pub struct SessionConfig {
220    /// Working directory.
221    pub cwd: PathBuf,
222    /// Agent dir (`~/.oino`).
223    pub agent_dir: PathBuf,
224    /// Root of the bundled Oino docs (system prompt docs section).
225    pub docs_root: PathBuf,
226    /// Session tree.
227    pub manager: SessionManager,
228    /// Settings (shared per cwd).
229    pub settings: Arc<Mutex<SettingsManager>>,
230    /// Resources (shared per cwd).
231    pub resources: Arc<RwLock<ResourceLoader>>,
232    /// Models and auth.
233    pub models: ModelRegistry,
234    /// Loaded extensions.
235    pub extensions: Arc<ExtensionSet>,
236    /// Provider dispatch.
237    pub stream: Arc<dyn StreamFn>,
238    /// Run mode.
239    pub mode: ExtensionMode,
240    /// Dialog-capable UI attached.
241    pub has_ui: bool,
242    /// Built-in tools (with prompt metadata).
243    pub builtin_tools: Vec<ToolEntry>,
244    /// SDK custom tools.
245    pub custom_tools: Vec<ToolEntry>,
246    /// Initially active tool names.
247    pub initial_active_tools: Vec<String>,
248    /// `--tools` allowlist patterns.
249    pub allowed_tools: Option<Vec<String>>,
250    /// `--exclude-tools` patterns.
251    pub excluded_tools: Vec<String>,
252    /// Initial model.
253    pub model: Option<Model>,
254    /// Initial thinking level.
255    pub thinking_level: ThinkingLevel,
256    /// `--models` scope.
257    pub scoped_models: Vec<oino_models::resolver::ScopedModel>,
258    /// CLI flag values for extension flags.
259    pub flag_values: indexmap::IndexMap<String, oino_ai::JsonValue>,
260    /// Nested-agent inputs of a child session (`None` for root sessions).
261    pub child: Option<crate::agents::ChildConfig>,
262}
263
264pub(crate) struct Inner {
265    pub(crate) agent: Agent,
266    pub(crate) stream: Arc<dyn StreamFn>,
267    pub(crate) manager: Mutex<SessionManager>,
268    pub(crate) settings: Arc<Mutex<SettingsManager>>,
269    pub(crate) resources: Arc<RwLock<ResourceLoader>>,
270    pub(crate) models: ModelRegistry,
271    pub(crate) runner: RwLock<ExtensionRunner>,
272    pub(crate) hub: EventHub,
273    pub(crate) ui: UiBridge,
274    /// Retained UI state (§6): entries, meta, dialogs, patches.
275    pub(crate) registry: Arc<Mutex<UiRegistry>>,
276    pub(crate) book: Mutex<RunBook>,
277    pub(crate) cancels: Mutex<Cancels>,
278    pub(crate) tools: Mutex<tools::ToolState>,
279    pub(crate) nested: Mutex<nested::NestedState>,
280    pub(crate) scoped_models: Mutex<Vec<oino_models::resolver::ScopedModel>>,
281    pub(crate) idle: watch::Sender<bool>,
282    pub(crate) cwd: PathBuf,
283    pub(crate) agent_dir: PathBuf,
284    pub(crate) docs_root: PathBuf,
285    pub(crate) mode: ExtensionMode,
286    pub(crate) subscription: Mutex<Option<Subscription>>,
287    pub(crate) agents: crate::agents::AgentsState,
288    pub(crate) arbiter: Mutex<boundary::ArbiterState>,
289    pub(crate) runtime: std::sync::OnceLock<std::sync::Weak<crate::runtime::RuntimeCore>>,
290    /// Composer text last reported through `editor_event` (`ui.editor-text`).
291    pub(crate) editor_text: Mutex<String>,
292    /// Topics, frontends, running bash calls, checkpoints, runtime handle.
293    pub(crate) frontend: feeds::FrontendState,
294}
295
296/// One live session. Cheap to clone.
297#[derive(Clone)]
298pub struct AgentSession {
299    pub(crate) inner: Arc<Inner>,
300}
301
302impl std::fmt::Debug for AgentSession {
303    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
304        f.debug_struct("AgentSession")
305            .field("session_id", &self.session_id())
306            .finish_non_exhaustive()
307    }
308}
309
310struct HubErrors {
311    hub: EventHub,
312    registry: Arc<Mutex<UiRegistry>>,
313}
314
315impl ErrorSink for HubErrors {
316    fn report(&self, error: ExtensionErrorEvent) {
317        tracing::warn!(extension = %error.extension_path, event = %error.event, "{}", error.error);
318        self.hub.publish(error);
319    }
320
321    fn disabled(&self, event: OinoEvent) {
322        if let OinoEvent::ExtensionDisabled { extension_id, .. } = &event {
323            tracing::error!(extension = %extension_id, "extension disabled for this session");
324            lock(&self.registry).remove_owner(extension_id);
325            lock(&self.registry).flush();
326        }
327        self.hub.publish(event);
328    }
329}
330
331/// The session's UI plumbing: legacy records go to the hub for v1 frontends,
332/// registry ops go out as `oino_ui_patch` events (§9.2).
333struct UiHub {
334    hub: EventHub,
335    registry: Arc<Mutex<UiRegistry>>,
336}
337
338impl UiHub {
339    fn flush(&self) {
340        lock(&self.registry).flush();
341    }
342}
343
344impl UiSink for UiHub {
345    fn publish(&self, output: UiOutput) {
346        match output {
347            UiOutput::Request(request) => {
348                if is_dialog(&request.method) {
349                    lock(&self.registry).dialog_opened(request.to_registry());
350                    self.flush();
351                }
352                self.hub.publish(ClientEvent::Ui(request));
353            }
354            UiOutput::Dismiss { id, reason } => {
355                lock(&self.registry).dialog_dismissed(&id, reason.into());
356                self.flush();
357                self.hub
358                    .publish(OinoEvent::ExtensionUiDismiss { id, reason });
359            }
360        }
361    }
362}
363
364impl RegistrySink for UiHub {
365    fn publish(&self, patch: oino_ui::slot::UiPatch) {
366        self.hub.publish(OinoEvent::from(patch));
367    }
368}
369
370impl AgentSession {
371    /// Creates a session (`new AgentSession` + `createAgentSession` agent wiring). Call
372    /// [`AgentSession::bind_extensions`] afterwards to emit `session_start`.
373    #[must_use]
374    #[allow(
375        clippy::too_many_lines,
376        reason = "one constructor for the whole session state, mirroring createAgentSession"
377    )]
378    pub fn new(config: SessionConfig) -> Self {
379        let hub = EventHub::new();
380        let session_id = config.manager.session_id().to_owned();
381        let has_ui = config.has_ui;
382        let cx = CallContext {
383            session_id: session_id.clone(),
384            mode: config.mode,
385            has_ui,
386            cwd: config.cwd.clone(),
387            is_child: config.child.is_some(),
388            child_role: config.child.as_ref().and_then(|c| c.link.role.clone()),
389            call_kind: CallKind::Event,
390        };
391        let registry = Arc::new(Mutex::new(UiRegistry::new(
392            oino_ui::slot::Scope::Session,
393            Some(session_id.clone()),
394        )));
395        let ui_hub = Arc::new(UiHub {
396            hub: hub.clone(),
397            registry: Arc::clone(&registry),
398        });
399        lock(&registry).set_sink(ui_hub.clone());
400        let ui = UiBridge::new(has_ui, session_id.clone(), ui_hub);
401        let runner = ExtensionRunner::new(
402            Arc::clone(&config.extensions),
403            cx,
404            Arc::new(HubErrors {
405                hub: hub.clone(),
406                registry: Arc::clone(&registry),
407            }),
408            config.flag_values.clone(),
409        )
410        .with_dialogs(ui.clone());
411        let context = config.manager.build_session_context();
412        let settings = lock(&config.settings).snapshot();
413        let session_stream = Arc::new(hooks::SessionStream {
414            inner: std::sync::OnceLock::new(),
415            base: Arc::clone(&config.stream),
416        });
417        let mut options = AgentOptions::new(Arc::clone(&session_stream) as Arc<dyn StreamFn>);
418        options.model = config
419            .model
420            .clone()
421            .unwrap_or_else(oino_agent::agent::unknown_model);
422        options.thinking_level = config.thinking_level;
423        options.messages = context.messages;
424        options.steering_mode = map_queue_mode(settings.steering_mode());
425        options.follow_up_mode = map_queue_mode(settings.follow_up_mode());
426        options.stream_options.session_id = Some(session_id);
427        options.stream_options.transport = Some(settings.transport());
428        options.stream_options.thinking_budgets = settings.thinking_budgets();
429        options.stream_options.max_retry_delay_ms =
430            Some(settings.provider_retry_settings().max_retry_delay_ms);
431        options.convert_to_llm = convert_with_block_images(Arc::clone(&config.settings));
432        let agent = Agent::new(options);
433        let (idle, _) = watch::channel(true);
434        let agents = match config.child {
435            Some(child) => crate::agents::AgentsState::child(child),
436            None => crate::agents::AgentsState::root(
437                Arc::clone(&config.extensions),
438                Arc::clone(&config.stream),
439            ),
440        };
441        let tool_state = tools::ToolState::new(
442            config.builtin_tools,
443            config.custom_tools,
444            config.allowed_tools.as_deref(),
445            &config.excluded_tools,
446        );
447        let inner = Arc::new(Inner {
448            agent,
449            stream: Arc::clone(&session_stream) as Arc<dyn StreamFn>,
450            manager: Mutex::new(config.manager),
451            settings: config.settings,
452            resources: config.resources,
453            models: config.models,
454            runner: RwLock::new(runner),
455            hub,
456            ui,
457            registry,
458            book: Mutex::new(RunBook::default()),
459            cancels: Mutex::new(Cancels::default()),
460            tools: Mutex::new(tool_state),
461            nested: Mutex::new(nested::NestedState::default()),
462            scoped_models: Mutex::new(config.scoped_models),
463            idle,
464            cwd: config.cwd,
465            agent_dir: config.agent_dir,
466            docs_root: config.docs_root,
467            mode: config.mode,
468            subscription: Mutex::new(None),
469            agents,
470            arbiter: Mutex::new(boundary::ArbiterState::default()),
471            runtime: std::sync::OnceLock::new(),
472            editor_text: Mutex::new(String::new()),
473            frontend: feeds::FrontendState::new(),
474        });
475        let _ = session_stream.inner.set(Arc::downgrade(&inner));
476        let session = Self { inner };
477        let subscription = session
478            .inner
479            .agent
480            .subscribe(Arc::new(listener::Listener::new(&session.inner)));
481        *lock(&session.inner.subscription) = Some(subscription);
482        session
483            .inner
484            .agent
485            .set_hooks(hooks::install(&session.inner));
486        session.refresh_tool_registry(Some(config.initial_active_tools), true);
487        session
488    }
489
490    /// `bindExtensions` tail: emits `session_start` and merges `resources_discover`
491    /// results into the resources.
492    pub async fn bind_extensions(&self, reason: &str, previous_session_file: Option<PathBuf>) {
493        let runner = self.runner();
494        if runner.has_handlers("session_start") {
495            let mut payload = serde_json::json!({"type": "session_start", "reason": reason});
496            if let Some(previous) = previous_session_file {
497                payload["previousSessionFile"] = serde_json::json!(previous.display().to_string());
498            }
499            runner.emit(payload).await;
500        }
501        let discover_reason = if reason == "reload" {
502            "reload"
503        } else {
504            "startup"
505        };
506        self.extend_resources_from_extensions(discover_reason).await;
507        // Static `register-renderer{target:"turn-footer"}` / manifest renderers get their
508        // registry entry when the instance is bound (§6.7).
509        for (extension, spec) in runner.renderers() {
510            if spec.get("target").and_then(oino_ai::JsonValue::as_str) == Some("turn-footer")
511                && let Some(key) = spec.get("key").and_then(oino_ai::JsonValue::as_str)
512            {
513                self.ui_register_turn_footer(&extension, key);
514            }
515        }
516        if !runner.shortcuts().is_empty() {
517            self.publish_shortcuts();
518        }
519    }
520
521    /// Package resources (manifest `resources` of every loaded extension, package
522    /// resources) followed by `resources_discover` results.
523    async fn extend_resources_from_extensions(&self, reason: &str) {
524        use oino_resources::source::{PathMetadata, SourceOrigin, SourceScope};
525        let runner = self.runner();
526        let entry =
527            |path: PathBuf, extension: &str, origin: SourceOrigin, base: Option<PathBuf>| {
528                oino_resources::ExtraResourcePath {
529                    path,
530                    metadata: PathMetadata {
531                        source: format!("extension:{extension}"),
532                        scope: SourceScope::Temporary,
533                        origin,
534                        base_dir: base,
535                    },
536                }
537            };
538        let mut paths = oino_resources::ResourceExtensionPaths::default();
539        for extension in runner.set().extensions() {
540            let id = extension.instance.id();
541            let dirs = extension.instance.resource_dirs();
542            let package = |list: Vec<PathBuf>| {
543                list.into_iter()
544                    .map(|dir| {
545                        let base = dir.parent().map(Path::to_path_buf);
546                        entry(dir, id, SourceOrigin::Package, base)
547                    })
548                    .collect::<Vec<_>>()
549            };
550            paths.skill_paths.extend(package(dirs.skills));
551            paths.prompt_paths.extend(package(dirs.prompts));
552            paths.theme_paths.extend(package(dirs.themes));
553        }
554        if runner.has_handlers("resources_discover") {
555            let found = runner
556                .emit_resources_discover(&self.inner.cwd.display().to_string(), reason)
557                .await;
558            let discovered = |list: Vec<(String, String)>| {
559                list.into_iter()
560                    .map(|(path, extension)| {
561                        entry(
562                            PathBuf::from(path),
563                            &extension,
564                            SourceOrigin::TopLevel,
565                            None,
566                        )
567                    })
568                    .collect::<Vec<_>>()
569            };
570            paths.skill_paths.extend(discovered(found.skill_paths));
571            paths.prompt_paths.extend(discovered(found.prompt_paths));
572            paths.theme_paths.extend(discovered(found.theme_paths));
573        }
574        if paths.skill_paths.is_empty()
575            && paths.prompt_paths.is_empty()
576            && paths.theme_paths.is_empty()
577        {
578            return;
579        }
580        self.inner
581            .resources
582            .write()
583            .unwrap_or_else(PoisonError::into_inner)
584            .extend(&paths);
585        let active = self.active_tool_names();
586        self.rebuild_system_prompt(&active);
587    }
588
589    /// `session_shutdown{reason:"reload"}`, settings and resources reload, tool
590    /// registry rebuild, then `session_start{reason:"reload"}` and `resources_discover`.
591    ///
592    /// # Errors
593    ///
594    /// Never fails today; kept fallible for the package reload path.
595    pub async fn reload(&self) -> Result<(), SessionOpError> {
596        // Reload drops every extension instance: its session entries go with it (§6.6).
597        {
598            let mut registry = lock(&self.inner.registry);
599            registry.clear();
600            registry.flush();
601        }
602        let runner = self.runner();
603        if runner.has_handlers("session_shutdown") {
604            runner
605                .emit(serde_json::json!({"type": "session_shutdown", "reason": "reload"}))
606                .await;
607        }
608        let snapshot = {
609            let mut settings = lock(&self.inner.settings);
610            settings.reload();
611            settings.snapshot()
612        };
613        self.inner
614            .agent
615            .set_steering_mode(map_queue_mode(snapshot.steering_mode()));
616        self.inner
617            .agent
618            .set_follow_up_mode(map_queue_mode(snapshot.follow_up_mode()));
619        let trusted = lock(&self.inner.settings).is_project_trusted();
620        self.inner
621            .resources
622            .write()
623            .unwrap_or_else(PoisonError::into_inner)
624            .reload(&snapshot, trusted);
625        self.refresh_tool_registry(None, true);
626        self.bind_extensions("reload", None).await;
627        Ok(())
628    }
629
630    /// Routes host calls of nested agents through `router` (set once by the runtime).
631    pub(crate) fn link_router(&self, router: &Arc<crate::services::HostRouter>) {
632        let _ = self.inner.agents.router.set(Arc::downgrade(router));
633    }
634
635    /// Lets command contexts replace the session through its runtime (set once).
636    pub(crate) fn link_runtime(&self, runtime: std::sync::Weak<crate::runtime::RuntimeCore>) {
637        let _ = self.inner.runtime.set(runtime);
638    }
639
640    pub(crate) fn runtime_core(&self) -> Option<Arc<crate::runtime::RuntimeCore>> {
641        self.inner.runtime.get().and_then(std::sync::Weak::upgrade)
642    }
643
644    pub(crate) fn from_inner(inner: Arc<Inner>) -> Self {
645        Self { inner }
646    }
647
648    /// The extension runner of this session.
649    #[must_use]
650    pub fn runner(&self) -> ExtensionRunner {
651        self.inner
652            .runner
653            .read()
654            .unwrap_or_else(PoisonError::into_inner)
655            .clone()
656    }
657
658    /// Ordered client event stream (bounded, coalescing).
659    #[must_use]
660    pub fn subscribe(&self) -> EventStream {
661        self.inner.hub.subscribe()
662    }
663
664    /// The event hub (runtime rebinding).
665    #[must_use]
666    pub fn hub(&self) -> &EventHub {
667        &self.inner.hub
668    }
669
670    pub(crate) fn emit(&self, event: AgentSessionEvent) {
671        self.inner.hub.publish(event);
672    }
673
674    /// Frontend → extension UI plumbing.
675    #[must_use]
676    pub fn ui(&self) -> UiBridge {
677        self.inner.ui.clone()
678    }
679
680    /// The underlying agent.
681    #[must_use]
682    pub fn agent(&self) -> &Agent {
683        &self.inner.agent
684    }
685
686    /// Runs `f` with the session manager locked.
687    pub fn with_manager<R>(&self, f: impl FnOnce(&mut SessionManager) -> R) -> R {
688        f(&mut lock(&self.inner.manager))
689    }
690
691    /// Runs `f` with the settings manager locked.
692    pub fn with_settings<R>(&self, f: impl FnOnce(&mut SettingsManager) -> R) -> R {
693        f(&mut lock(&self.inner.settings))
694    }
695
696    /// Model registry.
697    #[must_use]
698    pub fn models(&self) -> &ModelRegistry {
699        &self.inner.models
700    }
701
702    /// Session cwd.
703    #[must_use]
704    pub fn cwd(&self) -> &Path {
705        &self.inner.cwd
706    }
707
708    /// Agent dir.
709    #[must_use]
710    pub fn agent_dir(&self) -> &Path {
711        &self.inner.agent_dir
712    }
713
714    /// Run mode.
715    #[must_use]
716    pub fn mode(&self) -> ExtensionMode {
717        self.inner.mode
718    }
719
720    /// Current model (`None` until one is selected).
721    #[must_use]
722    pub fn model(&self) -> Option<Model> {
723        let model = self.inner.agent.model();
724        (model.provider != "unknown" || model.id != "unknown").then_some(model)
725    }
726
727    /// Current thinking level.
728    #[must_use]
729    pub fn thinking_level(&self) -> ThinkingLevel {
730        self.inner.agent.thinking_level()
731    }
732
733    /// Whether an agent run (or post-run continuation) is active.
734    #[must_use]
735    pub fn is_streaming(&self) -> bool {
736        lock(&self.inner.book).run_active
737    }
738
739    /// Whether a compaction or branch summary is running.
740    #[must_use]
741    pub fn is_compacting(&self) -> bool {
742        let cancels = lock(&self.inner.cancels);
743        cancels.compaction.is_some()
744            || cancels.auto_compaction.is_some()
745            || cancels.branch_summary.is_some()
746    }
747
748    /// Returns whether the session is idle.
749    #[must_use]
750    pub fn is_idle(&self) -> bool {
751        !self.is_streaming() && !self.is_compacting()
752    }
753
754    /// Whether an automatic retry is waiting.
755    #[must_use]
756    pub fn is_retrying(&self) -> bool {
757        lock(&self.inner.cancels).retry.is_some()
758    }
759
760    /// Whether a `!` command runs.
761    #[must_use]
762    pub fn is_bash_running(&self) -> bool {
763        !lock(&self.inner.cancels).bash.is_empty()
764    }
765
766    /// Current retry attempt (0 when not retrying).
767    #[must_use]
768    pub fn retry_attempt(&self) -> u32 {
769        lock(&self.inner.book).retry_attempt
770    }
771
772    /// The finalized transcript (session projection).
773    #[must_use]
774    pub fn messages(&self) -> Vec<AgentMessage> {
775        self.inner.agent.messages()
776    }
777
778    /// Session file, when persisted.
779    #[must_use]
780    pub fn session_file(&self) -> Option<PathBuf> {
781        lock(&self.inner.manager)
782            .session_file()
783            .map(Path::to_path_buf)
784    }
785
786    /// Session id.
787    #[must_use]
788    pub fn session_id(&self) -> String {
789        lock(&self.inner.manager).session_id().to_owned()
790    }
791
792    /// Session display name.
793    #[must_use]
794    pub fn session_name(&self) -> Option<String> {
795        lock(&self.inner.manager).session_name()
796    }
797
798    /// Steering texts queued.
799    #[must_use]
800    pub fn steering_messages(&self) -> Vec<String> {
801        lock(&self.inner.book).steering.clone()
802    }
803
804    /// Follow-up texts queued.
805    #[must_use]
806    pub fn follow_up_messages(&self) -> Vec<String> {
807        lock(&self.inner.book).follow_up.clone()
808    }
809
810    /// Number of queued texts.
811    #[must_use]
812    pub fn pending_message_count(&self) -> usize {
813        let book = lock(&self.inner.book);
814        book.steering.len() + book.follow_up.len()
815    }
816
817    /// Clears both queues and returns their texts.
818    #[must_use]
819    pub fn clear_queue(&self) -> (Vec<String>, Vec<String>) {
820        let (steering, follow_up) = {
821            let mut book = lock(&self.inner.book);
822            (
823                std::mem::take(&mut book.steering),
824                std::mem::take(&mut book.follow_up),
825            )
826        };
827        self.inner.agent.clear_all_queues();
828        self.emit_queue_update();
829        (steering, follow_up)
830    }
831
832    pub(crate) fn emit_queue_update(&self) {
833        let (steering, follow_up) = {
834            let book = lock(&self.inner.book);
835            (book.steering.clone(), book.follow_up.clone())
836        };
837        self.emit(AgentSessionEvent::QueueUpdate {
838            steering,
839            follow_up,
840        });
841    }
842
843    /// Text of the last assistant message.
844    #[must_use]
845    pub fn last_assistant_text(&self) -> Option<String> {
846        let messages = self.messages();
847        let last = messages.iter().rev().find_map(|message| match message {
848            AgentMessage::Assistant(assistant)
849                if !(assistant.stop_reason == oino_ai::StopReason::Aborted
850                    && assistant.content.is_empty()) =>
851            {
852                Some(assistant)
853            }
854            _ => None,
855        })?;
856        let text: String = last
857            .content
858            .iter()
859            .filter_map(|block| match block {
860                oino_ai::AssistantContent::Text(text) => Some(text.text.as_str()),
861                _ => None,
862            })
863            .collect();
864        let trimmed = text.trim();
865        (!trimmed.is_empty()).then(|| trimmed.to_owned())
866    }
867
868    /// Steering drain mode.
869    #[must_use]
870    pub fn steering_mode(&self) -> oino_agent::QueueMode {
871        self.inner.agent.steering_mode()
872    }
873
874    /// Follow-up drain mode.
875    #[must_use]
876    pub fn follow_up_mode(&self) -> oino_agent::QueueMode {
877        self.inner.agent.follow_up_mode()
878    }
879
880    /// Sets the steering mode and saves it to settings.
881    pub fn set_steering_mode(&self, mode: oino_agent::QueueMode) {
882        self.inner.agent.set_steering_mode(mode);
883        let result = self.with_settings(|s| s.set_steering_mode(unmap_queue_mode(mode)));
884        if let Err(error) = result {
885            tracing::warn!("saving steeringMode failed: {error}");
886        }
887    }
888
889    /// Sets the follow-up mode and saves it to settings.
890    pub fn set_follow_up_mode(&self, mode: oino_agent::QueueMode) {
891        self.inner.agent.set_follow_up_mode(mode);
892        let result = self.with_settings(|s| s.set_follow_up_mode(unmap_queue_mode(mode)));
893        if let Err(error) = result {
894            tracing::warn!("saving followUpMode failed: {error}");
895        }
896    }
897
898    /// Replaces the `--models` scope.
899    pub fn set_scoped_models(&self, scope: Vec<oino_models::resolver::ScopedModel>) {
900        *lock(&self.inner.scoped_models) = scope;
901    }
902
903    /// The `--models` scope.
904    #[must_use]
905    pub fn scoped_models(&self) -> Vec<oino_models::resolver::ScopedModel> {
906        lock(&self.inner.scoped_models).clone()
907    }
908
909    /// Rebuilds the finalized transcript from the session projection (the
910    /// `_refreshFinalizedContext`).
911    pub fn refresh_context(&self) {
912        let messages = lock(&self.inner.manager)
913            .build_session_projection()
914            .messages;
915        self.inner.agent.set_messages(messages);
916    }
917
918    /// Copy of the session entries (`get_entries`).
919    #[must_use]
920    pub fn entries(&self) -> Vec<SessionEntry> {
921        lock(&self.inner.manager).entries().to_vec()
922    }
923
924    /// Coarse run state.
925    #[must_use]
926    pub fn run_state(&self) -> RunState {
927        if self.is_compacting() {
928            RunState::Compacting
929        } else if self.is_retrying() {
930            RunState::Retrying
931        } else if self.is_streaming() {
932            RunState::Running
933        } else if !lock(&self.inner.arbiter).waiting_on.is_empty() {
934            RunState::Waiting
935        } else {
936            RunState::Idle
937        }
938    }
939
940    pub(crate) fn set_idle_if_idle(&self) {
941        if self.is_idle() {
942            self.inner.idle.send_replace(true);
943        }
944    }
945
946    pub(crate) fn mark_busy(&self) {
947        self.inner.idle.send_replace(false);
948    }
949
950    /// Resolves once the session is idle.
951    pub async fn wait_for_idle(&self) {
952        if self.is_idle() {
953            return;
954        }
955        let mut idle = self.inner.idle.subscribe();
956        loop {
957            if self.is_idle() && *idle.borrow() {
958                return;
959            }
960            if idle.changed().await.is_err() {
961                return;
962            }
963        }
964    }
965
966    /// Aborts the current operation and waits for idle.
967    pub async fn abort(&self) {
968        {
969            let mut book = lock(&self.inner.book);
970            if book.run_active {
971                book.abort_requested = true;
972            }
973            if book.is_before_settle {
974                book.abort_during_before_settle = true;
975            }
976        }
977        self.abort_retry();
978        self.abort_compaction();
979        self.abort_branch_summary();
980        self.inner.agent.abort();
981        self.wait_for_idle().await;
982    }
983
984    /// Cancels the retry wait.
985    pub fn abort_retry(&self) {
986        if let Some(token) = &lock(&self.inner.cancels).retry {
987            token.cancel();
988        }
989    }
990
991    /// Cancels manual or automatic compaction.
992    pub fn abort_compaction(&self) {
993        let cancels = lock(&self.inner.cancels);
994        for token in [&cancels.compaction, &cancels.auto_compaction]
995            .into_iter()
996            .flatten()
997        {
998            token.cancel();
999        }
1000    }
1001
1002    /// Cancels branch summarisation.
1003    pub fn abort_branch_summary(&self) {
1004        if let Some(token) = &lock(&self.inner.cancels).branch_summary {
1005            token.cancel();
1006        }
1007    }
1008
1009    /// Cancels running `!` commands.
1010    pub fn abort_bash(&self) {
1011        for (_, token) in &lock(&self.inner.cancels).bash {
1012            token.cancel();
1013        }
1014    }
1015
1016    /// Disposes the session: aborts everything and invalidates extension contexts.
1017    pub async fn dispose(&self) {
1018        // Boxed: closing a child disposes its session, which closes its own children.
1019        Box::pin(self.close_children()).await;
1020        self.abort_retry();
1021        self.abort_compaction();
1022        self.abort_branch_summary();
1023        self.abort_bash();
1024        self.inner.agent.abort();
1025        self.runner().invalidate(None);
1026        self.inner.ui.cancel_all();
1027        {
1028            let mut registry = lock(&self.inner.registry);
1029            registry.clear();
1030            registry.flush();
1031        }
1032        *lock(&self.inner.subscription) = None;
1033    }
1034
1035    /// Spawns a session task with per-session panic isolation (X-PANIC, core spec §2.3):
1036    /// a panic is caught, logged with the session id and published as
1037    /// [`OinoEvent::SessionCrashed`]. A crashed [`IsolatedTask::Drive`] task additionally
1038    /// resets the run book and repairs dangling tool calls, so the session stays open and
1039    /// the next prompt works — state is rebuilt from the log. Other sessions run as
1040    /// separate tasks and are unaffected.
1041    ///
1042    /// Returns the wrapper task's handle; aborting it aborts `fut`.
1043    pub fn spawn_isolated(
1044        &self,
1045        task: IsolatedTask,
1046        fut: impl Future<Output = ()> + Send + 'static,
1047    ) -> tokio::task::JoinHandle<()> {
1048        let session = self.clone();
1049        tokio::spawn(async move {
1050            if let Err(panic) = AssertUnwindSafe(fut).catch_unwind().await {
1051                session.on_isolated_panic(task, &*panic);
1052            }
1053        })
1054    }
1055
1056    /// Panic recovery (X-PANIC): report, reset, publish.
1057    fn on_isolated_panic(&self, task: IsolatedTask, panic: &(dyn Any + Send)) {
1058        let error = panic_message(panic);
1059        tracing::error!(
1060            session = %self.session_id(),
1061            task = task.name(),
1062            "session task panicked: {error}"
1063        );
1064        if task == IsolatedTask::Drive {
1065            self.recover_crashed_run();
1066        }
1067        self.inner.hub.publish(OinoEvent::SessionCrashed { error });
1068        // The run is over however it ended: tell clients the session is idle again.
1069        self.inner.hub.publish(OinoEvent::RunState {
1070            state: RunState::Idle,
1071            waiting_on: None,
1072        });
1073        self.set_idle_if_idle();
1074    }
1075
1076    /// Resets a run crashed mid-flight (X-PANIC): closes dangling tool calls of the
1077    /// trailing turn (unknown outcome, D-REPAIR semantics — a no-op for in-memory sessions
1078    /// and balanced transcripts), clears the run book and cancellation state, and unsticks
1079    /// the agent so the next `prompt` is accepted.
1080    fn recover_crashed_run(&self) {
1081        match self.with_manager(SessionManager::repair_interrupted_turn) {
1082            Ok(report) if !report.closed_tool_calls.is_empty() || report.torn_tail => {
1083                tracing::warn!(
1084                    session = %self.session_id(),
1085                    "crashed run repaired: {} dangling tool call(s) closed",
1086                    report.closed_tool_calls.len()
1087                );
1088                self.refresh_context();
1089            }
1090            Ok(_) => {}
1091            Err(error) => {
1092                tracing::warn!(session = %self.session_id(), "repair after crash failed: {error}");
1093            }
1094        }
1095        {
1096            let mut book = lock(&self.inner.book);
1097            book.run_active = false;
1098            book.abort_requested = false;
1099            book.overflow_recovery_attempted = false;
1100            book.retry_attempt = 0;
1101            book.failed_response = None;
1102            book.is_before_settle = false;
1103            book.abort_during_before_settle = false;
1104            book.emitting_settled = false;
1105            book.boundary_dispatched = false;
1106            book.run_prompt_options = None;
1107            book.tool_result_entries.clear();
1108        }
1109        {
1110            let mut cancels = lock(&self.inner.cancels);
1111            for token in [
1112                &cancels.retry,
1113                &cancels.compaction,
1114                &cancels.auto_compaction,
1115                &cancels.branch_summary,
1116            ]
1117            .into_iter()
1118            .flatten()
1119            {
1120                token.cancel();
1121            }
1122            cancels.retry = None;
1123            cancels.compaction = None;
1124            cancels.auto_compaction = None;
1125            cancels.branch_summary = None;
1126        }
1127        lock(&self.inner.arbiter).waiting_on.clear();
1128        self.inner.agent.recover_orphaned_run();
1129    }
1130
1131    /// Sets the session name.
1132    ///
1133    /// # Errors
1134    ///
1135    /// Persisting the `session_info` entry failed.
1136    pub async fn set_session_name(&self, name: Option<String>) -> Result<(), SessionOpError> {
1137        let current = {
1138            let mut manager = lock(&self.inner.manager);
1139            manager
1140                .append_session_info(name.as_deref())
1141                .map_err(|e| SessionOpError::Failed(e.to_string()))?;
1142            manager.session_name()
1143        };
1144        self.emit(AgentSessionEvent::SessionInfoChanged {
1145            name: current.clone(),
1146        });
1147        let mut payload = serde_json::json!({"type": "session_info_changed"});
1148        if let Some(name) = current {
1149            payload["name"] = serde_json::json!(name);
1150        }
1151        self.runner().emit(payload).await;
1152        Ok(())
1153    }
1154
1155    /// Sets or clears a label.
1156    ///
1157    /// # Errors
1158    ///
1159    /// Unknown entry or persistence failure.
1160    #[allow(
1161        clippy::unused_async,
1162        clippy::unused_async_trait_impl,
1163        reason = "async per the runtime.md AgentSession contract"
1164    )]
1165    pub async fn set_label(
1166        &self,
1167        entry: &EntryId,
1168        label: Option<String>,
1169    ) -> Result<(), SessionOpError> {
1170        lock(&self.inner.manager)
1171            .append_label_change(entry, label.as_deref())
1172            .map(|_| ())
1173            .map_err(|e| SessionOpError::Failed(e.to_string()))
1174    }
1175
1176    pub(crate) fn publish_entry(&self, entry_id: &EntryId) {
1177        let entry = lock(&self.inner.manager).entry(entry_id).cloned();
1178        if let Some(entry) = entry {
1179            self.emit(AgentSessionEvent::EntryAppended { entry });
1180        }
1181    }
1182
1183    pub(crate) fn emit_model_changed(&self, source: crate::events::ModelChangeSource) {
1184        if let Some(model) = self.model() {
1185            self.inner.hub.publish(OinoEvent::ModelChanged {
1186                model,
1187                routed_model: self.routed_model(),
1188                source,
1189            });
1190        }
1191    }
1192}
1193
1194fn map_queue_mode(mode: oino_settings::QueueMode) -> oino_agent::QueueMode {
1195    match mode {
1196        oino_settings::QueueMode::All => oino_agent::QueueMode::All,
1197        oino_settings::QueueMode::OneAtATime => oino_agent::QueueMode::OneAtATime,
1198    }
1199}
1200
1201fn unmap_queue_mode(mode: oino_agent::QueueMode) -> oino_settings::QueueMode {
1202    match mode {
1203        oino_agent::QueueMode::All => oino_settings::QueueMode::All,
1204        oino_agent::QueueMode::OneAtATime => oino_settings::QueueMode::OneAtATime,
1205    }
1206}
1207
1208/// Converts an entry to LLM records, blocking images.
1209fn convert_with_block_images(settings: Arc<Mutex<SettingsManager>>) -> oino_agent::ConvertToLlm {
1210    Arc::new(move |messages: &[AgentMessage]| {
1211        let converted = oino_session::convert_to_llm(messages);
1212        if !lock(&settings).settings().image_settings().block_images {
1213            return converted;
1214        }
1215        converted.into_iter().map(block_images).collect()
1216    })
1217}
1218
1219const IMAGE_DISABLED: &str = "Image reading is disabled.";
1220
1221fn block_images(message: oino_ai::Message) -> oino_ai::Message {
1222    use oino_ai::{Message, UserContent, UserContentField};
1223    let strip = |content: Vec<UserContent>| -> Vec<UserContent> {
1224        let mut out: Vec<UserContent> = Vec::new();
1225        for block in content {
1226            let block = match block {
1227                UserContent::Image(_) => UserContent::text(IMAGE_DISABLED),
1228                text @ UserContent::Text(_) => text,
1229            };
1230            let duplicate = matches!((&block, out.last()), (UserContent::Text(a), Some(UserContent::Text(b))) if a.text == IMAGE_DISABLED && b.text == IMAGE_DISABLED);
1231            if !duplicate {
1232                out.push(block);
1233            }
1234        }
1235        out
1236    };
1237    match message {
1238        Message::User(mut user) => {
1239            if let UserContentField::Blocks(blocks) = user.content {
1240                user.content = UserContentField::Blocks(strip(blocks));
1241            }
1242            Message::User(user)
1243        }
1244        Message::ToolResult(mut result) => {
1245            result.content = strip(result.content);
1246            Message::ToolResult(result)
1247        }
1248        other => other,
1249    }
1250}
oino_core/session/mod.rs

oino_core/session/
mod.rs
