use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use futures::StreamExt;
use futures::future::BoxFuture;
use futures::stream::BoxStream;
use oino_ai::JsonValue;
use oino_core::events::{ClientEvent, OinoEvent, ReplaceReason};
use oino_core::runtime::{AgentSessionRuntime, ForkPosition};
use oino_core::session::{AgentSession, CycleDirection, InputSource, PromptOptions, RewindResult};
use oino_core::ui::UiResponse;
use oino_session::EntryId;
use serde::Serialize;
use serde_json::json;
use crate::backend::{ClientError, SessionBackend};
use crate::follow::Followers;
use crate::protocol::{AgentCommand, OinoCommand, RpcCommand, RpcResponseData, RpcSessionState};
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ServerInfo {
pub name: String,
pub version: String,
}
impl Default for ServerInfo {
fn default() -> Self {
Self {
name: "oino".to_owned(),
version: env!("CARGO_PKG_VERSION").to_owned(),
}
}
}
pub struct InProcessBackend {
shared: Arc<Shared>,
handle: Option<tokio::runtime::Handle>,
}
struct Shared {
runtime: Arc<AgentSessionRuntime>,
server: ServerInfo,
closed: AtomicBool,
followers: Followers,
}
impl std::fmt::Debug for InProcessBackend {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("InProcessBackend")
.field("server", &self.shared.server)
.finish_non_exhaustive()
}
}
fn data(value: impl Serialize) -> Result<RpcResponseData, ClientError> {
serde_json::to_value(value)
.map(Some)
.map_err(|e| ClientError::Protocol(e.to_string()))
}
fn failed(error: impl std::fmt::Display) -> ClientError {
ClientError::Command(error.to_string())
}
fn find_agent(
session: &AgentSession,
agent_id: &str,
) -> Result<Arc<oino_core::agents::ChildAgent>, ClientError> {
session
.find_agent(agent_id)
.ok_or_else(|| ClientError::Command(format!("Unknown agent: {agent_id}")))
}
impl InProcessBackend {
#[must_use]
pub fn new(runtime: Arc<AgentSessionRuntime>) -> Self {
Self::with_server_info(runtime, ServerInfo::default())
}
#[must_use]
pub fn with_server_info(runtime: Arc<AgentSessionRuntime>, server: ServerInfo) -> Self {
let handle = tokio::runtime::Handle::try_current()
.ok()
.or_else(|| runtime.session().runtime_handle());
Self {
shared: Arc::new(Shared {
runtime,
server,
closed: AtomicBool::new(false),
followers: Followers::default(),
}),
handle,
}
}
#[must_use]
pub fn runtime(&self) -> &Arc<AgentSessionRuntime> {
&self.shared.runtime
}
#[must_use]
pub fn state(&self) -> RpcSessionState {
self.shared.state()
}
fn on_runtime<T: Send + 'static>(
&self,
task: impl Future<Output = T> + Send + 'static,
closed: T,
) -> BoxFuture<'static, T> {
match (tokio::runtime::Handle::try_current(), &self.handle) {
(Err(_), Some(handle)) => {
let join = handle.spawn(task);
Box::pin(async move { join.await.unwrap_or(closed) })
}
_ => Box::pin(task),
}
}
}
impl Shared {
fn session(&self) -> AgentSession {
self.runtime.session()
}
fn path(&self, path: &str) -> PathBuf {
let path = Path::new(path);
if path.is_absolute() {
path.to_path_buf()
} else {
self.runtime.services().cwd.join(path)
}
}
fn rebind(&self, reason: ReplaceReason, cancelled: bool) {
if cancelled {
return;
}
let session = self.session();
let announce = ClientEvent::Oino(OinoEvent::SessionReplaced {
session_file: session.session_file().map(|p| p.display().to_string()),
session_id: session.session_id(),
reason,
});
self.followers
.rebind(|| self.runtime.session().subscribe(), &announce);
}
fn state(&self) -> RpcSessionState {
let session = self.session();
let auto_compaction_enabled = session.with_settings(|s| s.settings().compaction_enabled());
RpcSessionState {
model: session.model(),
thinking_level: session.thinking_level(),
is_streaming: session.is_streaming(),
is_compacting: session.is_compacting(),
steering_mode: session.steering_mode(),
follow_up_mode: session.follow_up_mode(),
session_file: session.session_file().map(|p| p.display().to_string()),
session_id: session.session_id(),
session_name: session.session_name(),
auto_compaction_enabled,
message_count: session.messages().len(),
pending_message_count: session.pending_message_count(),
}
}
async fn dispatch(
&self,
request_id: Option<String>,
command: RpcCommand,
) -> Result<RpcResponseData, ClientError> {
if self.closed.load(Ordering::Acquire) {
return Err(ClientError::Closed);
}
match command {
RpcCommand::Agent(command) => self.agent(request_id, command).await,
RpcCommand::Oino(command) => self.oino(command).await,
}
}
#[allow(
clippy::too_many_lines,
reason = "one arm per agent command, mirroring handleCommand"
)]
async fn agent(
&self,
request_id: Option<String>,
command: AgentCommand,
) -> Result<RpcResponseData, ClientError> {
let session = self.session();
match command {
AgentCommand::Prompt {
message,
images,
streaming_behavior,
} => {
let disposition = session
.prompt(
message,
PromptOptions {
images,
streaming_behavior,
source: InputSource::Rpc,
..PromptOptions::default()
},
)
.await
.map_err(failed)?;
data(json!({ "disposition": disposition }))
}
AgentCommand::Steer { message, images } => {
let disposition = session
.steer(message, images, InputSource::Rpc)
.await
.map_err(failed)?;
data(json!({ "disposition": disposition }))
}
AgentCommand::FollowUp { message, images } => {
let disposition = session
.follow_up(message, images, InputSource::Rpc)
.await
.map_err(failed)?;
data(json!({ "disposition": disposition }))
}
AgentCommand::Abort => {
session.abort().await;
Ok(None)
}
AgentCommand::ClearQueue => {
let (steering, follow_up) = session.clear_queue();
data(json!({ "steering": steering, "followUp": follow_up }))
}
AgentCommand::NewSession { parent_session } => {
let parent = parent_session.map(|p| self.path(&p));
let result = self.runtime.new_session(parent).await.map_err(failed)?;
self.rebind(ReplaceReason::New, result.cancelled);
data(json!({ "cancelled": result.cancelled }))
}
AgentCommand::GetState => data(self.state()),
AgentCommand::SetModel { provider, model_id } => {
let model = session
.models()
.available()
.into_iter()
.find(|m| m.provider == provider && m.id == model_id)
.ok_or_else(|| {
ClientError::Command(format!("Model not found: {provider}/{model_id}"))
})?;
let model = session.set_model(model, true).await.map_err(failed)?;
data(model)
}
AgentCommand::CycleModel => {
let result = session
.cycle_model(CycleDirection::Forward, true)
.await
.map_err(failed)?;
data(result)
}
AgentCommand::GetAvailableModels => {
data(json!({ "models": session.models().available() }))
}
AgentCommand::SetThinkingLevel { level } => {
session.set_thinking_level(level, true).await;
Ok(None)
}
AgentCommand::CycleThinkingLevel => {
let level = session.cycle_thinking_level(true).await;
Ok(Some(level.map_or(
JsonValue::Null,
|level| json!({ "level": level }),
)))
}
AgentCommand::GetAvailableThinkingLevels => {
data(json!({ "levels": session.available_thinking_levels() }))
}
AgentCommand::SetSteeringMode { mode } => {
session.set_steering_mode(mode);
Ok(None)
}
AgentCommand::SetFollowUpMode { mode } => {
session.set_follow_up_mode(mode);
Ok(None)
}
AgentCommand::Compact {
custom_instructions,
} => data(session.compact(custom_instructions).await.map_err(failed)?),
AgentCommand::SetAutoCompaction { enabled } => {
session.set_auto_compaction_enabled(enabled);
Ok(None)
}
AgentCommand::SetAutoRetry { enabled } => {
session.set_auto_retry_enabled(enabled);
Ok(None)
}
AgentCommand::AbortRetry => {
session.abort_retry();
Ok(None)
}
AgentCommand::Bash {
command,
exclude_from_context,
} => {
let result = session
.execute_bash(command, exclude_from_context.unwrap_or(false), request_id)
.await
.map_err(failed)?;
data(result)
}
AgentCommand::AbortBash => {
session.abort_bash();
Ok(None)
}
AgentCommand::GetSessionStats => data(session.session_stats()),
AgentCommand::ExportHtml { output_path } => {
let path = session
.export_html(output_path.map(|p| self.path(&p)))
.await
.map_err(failed)?;
data(json!({ "path": path.display().to_string() }))
}
AgentCommand::SwitchSession { session_path } => {
let result = self
.runtime
.switch_session(self.path(&session_path), None)
.await
.map_err(failed)?;
self.rebind(ReplaceReason::Resume, result.cancelled);
data(json!({ "cancelled": result.cancelled }))
}
AgentCommand::Fork { entry_id } => {
let result = self
.runtime
.fork(&EntryId::new(entry_id), ForkPosition::Before)
.await
.map_err(failed)?;
self.rebind(ReplaceReason::Fork, result.cancelled);
let mut value = json!({ "cancelled": result.cancelled });
if let Some(text) = result.selected_text {
value = json!({ "text": text, "cancelled": result.cancelled });
}
Ok(Some(value))
}
AgentCommand::Clone => {
let leaf = session.with_manager(|m| m.leaf_id().cloned());
if leaf.is_none() {
return Err(ClientError::Command(
"Cannot clone session: no current entry selected".to_owned(),
));
}
let result = self.runtime.clone_session().await.map_err(failed)?;
self.rebind(ReplaceReason::Clone, result.cancelled);
data(json!({ "cancelled": result.cancelled }))
}
AgentCommand::GetForkMessages => {
data(json!({ "messages": session.user_messages_for_forking() }))
}
AgentCommand::GetEntries { since } => {
let (entries, leaf) =
session.with_manager(|m| (m.entries().to_vec(), m.leaf_id().cloned()));
let entries = match since {
None => entries,
Some(since) => {
let index = entries
.iter()
.position(|e| e.id.as_str() == since)
.ok_or_else(|| {
ClientError::Command(format!("Entry not found: {since}"))
})?;
entries[index + 1..].to_vec()
}
};
data(json!({ "entries": entries, "leafId": leaf }))
}
AgentCommand::GetTree => {
let (tree, leaf) = session.with_manager(|m| (m.tree(), m.leaf_id().cloned()));
data(json!({ "tree": tree, "leafId": leaf }))
}
AgentCommand::GetLastAssistantText => {
data(json!({ "text": session.last_assistant_text() }))
}
AgentCommand::SetSessionName { name } => {
let name = name.trim();
if name.is_empty() {
return Err(ClientError::Command(
"Session name cannot be empty".to_owned(),
));
}
session
.set_session_name(Some(name.to_owned()))
.await
.map_err(failed)?;
Ok(None)
}
AgentCommand::GetMessages => data(json!({ "messages": session.messages() })),
AgentCommand::GetCommands => data(json!({ "commands": session.commands() })),
}
}
async fn oino(&self, command: OinoCommand) -> Result<RpcResponseData, ClientError> {
let session = self.session();
match command {
OinoCommand::Hello { client, .. } => {
session.register_frontend(&client.name);
data(json!({
"server": {"name": self.server.name, "version": self.server.version},
"protocol": 1,
"sessionId": session.session_id(),
"safeMode": self
.runtime
.services()
.safe_mode
.as_ref()
.is_some_and(|plan| plan.disable_extensions),
}))
}
OinoCommand::UiSnapshot => {
let snapshot = session.ui_snapshot();
let mut value = serde_json::to_value(snapshot).unwrap_or(JsonValue::Null);
if let Some(object) = value.as_object_mut() {
object.insert(
"requests".to_owned(),
serde_json::to_value(session.ui().snapshot()).unwrap_or(JsonValue::Null),
);
}
data(value)
}
OinoCommand::UiVisible {
entry_id,
visible,
frontend,
} => {
let frontend = frontend
.as_deref()
.filter(|name| !name.is_empty())
.unwrap_or("rpc");
session.ui_visible(frontend, &entry_id, visible);
Ok(None)
}
OinoCommand::UiResync { .. } => data(session.ui_resync()),
OinoCommand::NavigateTree { target_id, options } => {
let result = session
.navigate_tree(&EntryId::new(target_id), options)
.await
.map_err(failed)?;
data(result)
}
OinoCommand::SetLabel { entry_id, label } => {
session
.set_label(&EntryId::new(entry_id), label)
.await
.map_err(failed)?;
Ok(None)
}
OinoCommand::ImportJsonl { path } => {
let result = self
.runtime
.import_jsonl(self.path(&path), None)
.await
.map_err(failed)?;
self.rebind(ReplaceReason::Import, result.cancelled);
data(json!({ "cancelled": result.cancelled }))
}
OinoCommand::ExportJsonl { output_path } => {
let path = session
.export_jsonl(output_path.map(|p| self.path(&p)))
.await
.map_err(failed)?;
data(json!({ "path": path.display().to_string() }))
}
OinoCommand::GetContextUsage => data(session.context_usage()),
command @ (OinoCommand::Topics { .. }
| OinoCommand::BusGet { .. }
| OinoCommand::Agents { .. }
| OinoCommand::AgentMessages { .. }
| OinoCommand::AgentAbort { .. }
| OinoCommand::QueueEdit(_)
| OinoCommand::Checkpoints
| OinoCommand::Rewind { .. }
| OinoCommand::RunningBash
| OinoCommand::Extensions
| OinoCommand::SetExtensionEnabled { .. }) => self.gui(&session, command).await,
command => self.frontend(&session, command).await,
}
}
async fn gui(
&self,
session: &AgentSession,
command: OinoCommand,
) -> Result<RpcResponseData, ClientError> {
match command {
OinoCommand::Topics {
subscribe,
unsubscribe,
} => {
let topics = session.subscribe_topics(&subscribe, &unsubscribe).await;
data(json!({ "topics": topics, "patterns": session.topic_patterns() }))
}
OinoCommand::BusGet { topic } => Ok(Some(session.bus_get(&topic).map_or_else(
|| json!({ "data": null }),
|v| json!({ "data": v.data, "publisher": v.publisher }),
))),
OinoCommand::Agents { follow } => {
if let Some(follow) = follow {
session.follow_agents(follow);
}
data(json!({ "agents": session.agent_tree(), "follow": session.follows_agents() }))
}
OinoCommand::AgentMessages { agent_id } => {
let agent = find_agent(session, &agent_id)?;
data(json!({ "messages": agent.messages() }))
}
OinoCommand::AgentAbort { agent_id } => {
find_agent(session, &agent_id)?.abort();
Ok(None)
}
OinoCommand::QueueEdit(edit) => {
session.edit_queue(edit).await.map_err(failed)?;
Ok(None)
}
OinoCommand::Checkpoints => data(json!({ "checkpoints": session.checkpoints() })),
OinoCommand::Rewind {
entry_id,
files,
conversation,
} => self.rewind(session, &entry_id, files, conversation).await,
OinoCommand::RunningBash => data(json!({ "calls": session.running_bash().await })),
OinoCommand::Extensions => data(json!({ "extensions": session.extension_infos() })),
OinoCommand::SetExtensionEnabled { id, enabled } => {
session
.set_extension_enabled(&id, enabled)
.map_err(failed)?;
Ok(None)
}
other => Err(ClientError::Protocol(format!(
"{} is not a GUI command",
other.kind()
))),
}
}
async fn rewind(
&self,
session: &AgentSession,
entry_id: &str,
files: bool,
conversation: bool,
) -> Result<RpcResponseData, ClientError> {
if session.is_streaming() {
return Err(ClientError::Command(
"Cannot rewind while the agent is running".to_owned(),
));
}
let mut result = if files && (session.has_checkpoint(entry_id) || !conversation) {
session.rewind_files(entry_id).map_err(failed)?
} else {
RewindResult::default()
};
if conversation {
let forked = self
.runtime
.fork(&EntryId::new(entry_id), ForkPosition::Before)
.await
.map_err(failed)?;
self.rebind(ReplaceReason::Fork, forked.cancelled);
result.cancelled = forked.cancelled;
result.text = forked.selected_text;
}
data(result)
}
fn render_request(
command: &OinoCommand,
) -> Result<oino_core::session::RenderRequest, ClientError> {
let OinoCommand::Render {
target,
key,
entry_id,
turn_entry_id,
turn_index,
running,
density,
expanded,
width,
data: payload,
} = command
else {
return Err(ClientError::Protocol(format!(
"{} is not a render command",
command.kind()
)));
};
let target = oino_core::session::RenderTarget::parse(target)
.ok_or_else(|| ClientError::Protocol(format!("render target `{target}`")))?;
Ok(oino_core::session::RenderRequest {
target,
key: key.clone(),
entry_id: entry_id.clone(),
turn_entry_id: turn_entry_id.clone(),
turn_index: *turn_index,
running: *running,
density: density.as_deref().map_or_else(Default::default, |d| {
oino_core::session::RenderDensity::parse(d).unwrap_or_default()
}),
expanded: *expanded,
width: width.as_deref().map_or_else(Default::default, |w| {
oino_core::session::RenderWidth::parse(w).unwrap_or_default()
}),
data: payload.clone(),
})
}
#[allow(
clippy::too_many_lines,
reason = "one arm per frontend command, mirroring the RPC surface"
)]
async fn frontend(
&self,
session: &AgentSession,
command: OinoCommand,
) -> Result<RpcResponseData, ClientError> {
match command {
OinoCommand::GetCompletions { text, cursor } => {
data(json!({ "groups": session.completions(&text, cursor).await }))
}
OinoCommand::UiAction {
panel_id,
action_id,
value,
values,
path,
source,
} => {
let action = oino_core::ui::UiAction {
panel_id,
action_id,
value,
values: values.and_then(|values| values.as_object().cloned()),
path: path.and_then(|path| serde_json::from_value(path).ok()),
source: source.as_deref().and_then(|name| match name {
"tui" => Some(oino_core::ui::ActionSource::Tui),
"app" => Some(oino_core::ui::ActionSource::App),
"rpc" => Some(oino_core::ui::ActionSource::Rpc),
_ => None,
}),
};
session.ui_action_with(&action).map_err(failed)?;
Ok(None)
}
OinoCommand::EditorEvent(event) => {
let consumed = session.editor_event(event).await;
data(json!({ "consumed": consumed }))
}
OinoCommand::InvokeShortcut { shortcut_id } => {
session.invoke_shortcut(&shortcut_id).map_err(failed)?;
Ok(None)
}
OinoCommand::GetShortcuts => {
let shortcuts: Vec<JsonValue> = session
.runner()
.shortcuts()
.into_iter()
.map(|(extension, spec)| {
json!({
"id": spec.id,
"defaultKeys": spec.default_keys,
"description": spec.description,
"extensionId": extension,
})
})
.collect();
data(json!({ "shortcuts": shortcuts }))
}
command @ (OinoCommand::GetSettingsSchema
| OinoCommand::GetExtensionSettings { .. }
| OinoCommand::SetExtensionSetting { .. }
| OinoCommand::UnsetExtensionSetting { .. }
| OinoCommand::GetSettings { .. }
| OinoCommand::SetSetting { .. }) => self.settings_command(session, command),
OinoCommand::ServiceCall {
target,
name,
payload,
} => data(
session
.service_call(target.as_deref(), &name, payload)
.await
.map_err(failed)?,
),
OinoCommand::AuthProviders => data(json!({ "providers": session.auth_providers() })),
OinoCommand::Login { provider } => {
session.login(&provider).await.map_err(failed)?;
Ok(None)
}
OinoCommand::Logout { provider } => {
session.logout(&provider).map_err(failed)?;
Ok(None)
}
OinoCommand::SetApiKey { provider, key } => {
session.set_api_key(&provider, &key).await.map_err(failed)?;
Ok(None)
}
render @ OinoCommand::Render { .. } => {
let node = session
.render_spec(&Self::render_request(&render)?)
.map_err(failed)?;
data(json!({ "node": node }))
}
OinoCommand::Hello { .. }
| OinoCommand::Topics { .. }
| OinoCommand::BusGet { .. }
| OinoCommand::Agents { .. }
| OinoCommand::AgentMessages { .. }
| OinoCommand::AgentAbort { .. }
| OinoCommand::QueueEdit(_)
| OinoCommand::Checkpoints
| OinoCommand::Rewind { .. }
| OinoCommand::RunningBash
| OinoCommand::Extensions
| OinoCommand::SetExtensionEnabled { .. }
| OinoCommand::UiSnapshot
| OinoCommand::UiVisible { .. }
| OinoCommand::UiResync { .. }
| OinoCommand::NavigateTree { .. }
| OinoCommand::SetLabel { .. }
| OinoCommand::ImportJsonl { .. }
| OinoCommand::ExportJsonl { .. }
| OinoCommand::GetContextUsage => Err(ClientError::Protocol(format!(
"{} is not a frontend command",
command.kind()
))),
}
}
fn settings_command(
&self,
session: &AgentSession,
command: OinoCommand,
) -> Result<RpcResponseData, ClientError> {
let registry = self.runtime.extension_settings();
match command {
OinoCommand::GetSettingsSchema => data(json!({ "namespaces": registry.schema() })),
OinoCommand::GetExtensionSettings { namespace, scope } => {
if registry.namespace(&namespace).is_none() {
return Err(ClientError::Command(format!(
"Unknown settings namespace: {namespace}"
)));
}
let resolved = registry.resolved(&namespace, scope);
data(json!({
"values": resolved.values,
"diagnostics": resolved.diagnostics,
}))
}
OinoCommand::SetExtensionSetting {
namespace,
key,
value,
scope,
} => {
let value = value.filter(|v| !v.is_null());
registry
.set(&namespace, &key, value, scope)
.map_err(failed)?;
Ok(None)
}
OinoCommand::UnsetExtensionSetting {
namespace,
key,
scope,
} => {
registry
.set(&namespace, &key, None, scope)
.map_err(failed)?;
Ok(None)
}
OinoCommand::GetSettings { key } => {
let merged = registry.merged_settings();
match key {
None => data(json!({ "settings": merged })),
Some(key) => {
let path: Vec<&str> = key.split('.').collect();
let value = oino_settings::json::get_path(&merged, &path).cloned();
data(json!({ "value": value }))
}
}
}
OinoCommand::SetSetting { key, value, scope } => {
let path: Vec<&str> = key.split('.').filter(|p| !p.is_empty()).collect();
if path.is_empty() {
return Err(ClientError::Command("Setting key cannot be empty".into()));
}
let value = value.filter(|v| !v.is_null());
session
.with_settings(|settings| settings.set(scope, &path, value))
.map_err(failed)?;
Ok(None)
}
other => Err(ClientError::Protocol(format!(
"{} is not a settings command",
other.kind()
))),
}
}
}
impl SessionBackend for InProcessBackend {
fn send(&self, command: RpcCommand) -> BoxFuture<'_, Result<RpcResponseData, ClientError>> {
self.send_with_id(None, command)
}
fn send_with_id(
&self,
request_id: Option<String>,
command: RpcCommand,
) -> BoxFuture<'_, Result<RpcResponseData, ClientError>> {
let shared = Arc::clone(&self.shared);
self.on_runtime(
async move { shared.dispatch(request_id, command).await },
Err(ClientError::Closed),
)
}
fn events(&self) -> BoxStream<'static, ClientEvent> {
self.shared
.followers
.stream(self.shared.session().subscribe())
.boxed()
}
fn ui_respond(&self, response: UiResponse) {
self.shared.session().ui().respond(response);
}
fn session_id(&self) -> String {
self.shared.session().session_id()
}
fn shutdown(&self) -> BoxFuture<'_, ()> {
let shared = Arc::clone(&self.shared);
self.on_runtime(
async move {
if !shared.closed.swap(true, Ordering::AcqRel) {
shared.runtime.dispose().await;
shared.followers.close();
}
},
(),
)
}
fn shutdown_requested(&self) -> BoxFuture<'static, ()> {
let runtime = Arc::clone(&self.shared.runtime);
self.on_runtime(
async move {
let mut requested = runtime.shutdown_requested();
if requested.wait_for(|on| *on).await.is_err() {
return std::future::pending().await;
}
runtime.session().wait_for_idle().await;
},
(),
)
}
}