mod builder;
pub use builder::OrcsAppBuilder;
use crate::AppError;
use crate::SharedPrinterSlot;
use orcs_event::Signal;
use orcs_runtime::auth::{DefaultGrantStore, GrantPolicy};
use orcs_runtime::io::{ConsoleRenderer, InputParser};
use orcs_runtime::session::SessionStore;
use orcs_runtime::{
ConfigResolver, IOInput, IOInputHandle, IOOutput, IOOutputHandle, InputCommand, InputContext,
LocalFileStore, OrcsConfig, OrcsEngine, SessionAsset,
};
use orcs_types::{ChannelId, Principal, PrincipalId};
use rustyline::highlight::{CmdKind, Highlighter};
use rustyline::history::DefaultHistory;
use rustyline::{
Completer, Editor, ExternalPrinter as RustylineExternalPrinter, Helper, Hinter, Validator,
};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
#[derive(Debug, Clone)]
struct PendingApproval {
id: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LoopControl {
Continue,
Exit,
Pause,
}
#[derive(Debug)]
enum ReadlineEvent {
Line(String),
Eof,
}
#[derive(Completer, Helper, Hinter, Validator)]
struct OrcsHelper;
impl Highlighter for OrcsHelper {
fn highlight_char(&self, _line: &str, _pos: usize, kind: CmdKind) -> bool {
kind == CmdKind::Other
}
}
pub struct OrcsApp {
pub(super) config: OrcsConfig,
pub(super) resolver: Box<dyn ConfigResolver>,
pub(super) engine: OrcsEngine,
pub(super) principal: Principal,
pub(super) renderer: ConsoleRenderer,
pub(super) io_input: IOInputHandle,
pub(super) io_output: IOOutputHandle,
pending_approval: Option<PendingApproval>,
pub(super) store: LocalFileStore,
pub(super) session: SessionAsset,
pub(super) is_resumed: bool,
pub(super) restored_count: usize,
pub(super) shared_grants: Arc<DefaultGrantStore>,
pub(super) component_routes: HashMap<String, ChannelId>,
printer_slot: SharedPrinterSlot,
}
impl OrcsApp {
#[must_use]
pub fn builder(resolver: impl ConfigResolver + 'static) -> OrcsAppBuilder {
OrcsAppBuilder::new(resolver)
}
#[must_use]
pub fn engine(&self) -> &OrcsEngine {
&self.engine
}
pub fn engine_mut(&mut self) -> &mut OrcsEngine {
&mut self.engine
}
#[must_use]
pub fn config(&self) -> &OrcsConfig {
&self.config
}
pub fn reload_config(&mut self) -> Result<(), AppError> {
self.config = self
.resolver
.resolve()
.map_err(|e| AppError::Config(e.to_string()))?;
Ok(())
}
#[must_use]
pub fn renderer(&self) -> &ConsoleRenderer {
&self.renderer
}
#[must_use]
pub fn pending_approval_id(&self) -> Option<&str> {
self.pending_approval.as_ref().map(|p| p.id.as_str())
}
#[must_use]
pub fn session_id(&self) -> &str {
&self.session.id
}
#[must_use]
pub fn is_resumed(&self) -> bool {
self.is_resumed
}
#[must_use]
pub fn restored_count(&self) -> usize {
self.restored_count
}
pub async fn save_session(&mut self) -> Result<(), AppError> {
for (fqn, snapshot) in self.engine.collected_snapshots() {
self.session
.component_snapshots
.insert(fqn.clone(), snapshot.clone());
}
match self.shared_grants.list_grants() {
Ok(grants) => {
tracing::debug!("Saving {} grant(s) to session", grants.len());
self.session.save_grants(grants);
}
Err(e) => {
tracing::error!("Failed to list grants for session save: {e}");
}
}
self.session.touch();
self.store.save(&self.session).await?;
Ok(())
}
pub async fn load_session(&mut self, session_id: &str) -> Result<(), AppError> {
self.session = self.store.load(session_id).await?;
Ok(())
}
fn history_path(&self) -> PathBuf {
self.config.paths.history_file_or_default()
}
fn spawn_readline_thread(
history_path: PathBuf,
) -> (
tokio::sync::mpsc::UnboundedReceiver<ReadlineEvent>,
Option<Box<dyn RustylineExternalPrinter + Send>>,
) {
let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel();
let (printer_tx, printer_rx) =
std::sync::mpsc::sync_channel::<Option<Box<dyn RustylineExternalPrinter + Send>>>(1);
if let Err(e) = std::thread::Builder::new()
.name("orcs-readline".into())
.spawn(move || {
let config = rustyline::Config::builder().auto_add_history(true).build();
let mut rl: Editor<OrcsHelper, DefaultHistory> = match Editor::with_config(config) {
Ok(editor) => editor,
Err(e) => {
tracing::error!("Failed to create readline editor: {e}");
let _ = printer_tx.send(None);
return;
}
};
rl.set_helper(Some(OrcsHelper));
if let Err(e) = rl.load_history(&history_path) {
tracing::debug!("History load: {e} (expected on first run)");
}
let printer = rl
.create_external_printer()
.ok()
.map(|p| Box::new(p) as Box<dyn RustylineExternalPrinter + Send>);
let _ = printer_tx.send(printer);
loop {
match rl.readline("orcs> ") {
Ok(line) => {
let _ = rl.save_history(&history_path);
if event_tx.send(ReadlineEvent::Line(line)).is_err() {
break; }
}
Err(rustyline::error::ReadlineError::Interrupted) => {
continue;
}
Err(rustyline::error::ReadlineError::Eof) => {
let _ = event_tx.send(ReadlineEvent::Eof);
break;
}
Err(e) => {
tracing::error!("Readline error: {e}");
let _ = event_tx.send(ReadlineEvent::Eof);
break;
}
}
}
let _ = rl.save_history(&history_path);
})
{
tracing::error!("failed to spawn readline thread: {e}");
}
let printer = printer_rx.recv().ok().flatten();
(event_rx, printer)
}
fn render_safe(&self, output: &IOOutput) {
if let Some(formatted) = Self::format_io_output(output, self.renderer.is_verbose()) {
match self.printer_slot.try_print(formatted) {
crate::PrintResult::Sent | crate::PrintResult::Dropped => return,
crate::PrintResult::NoPrinter => {}
}
}
self.renderer.render_output(output);
}
fn format_io_output(output: &IOOutput, verbose: bool) -> Option<String> {
match output {
IOOutput::Print { text, style } => match style {
orcs_runtime::OutputStyle::Normal | orcs_runtime::OutputStyle::Info => {
Some(format!("{text}\n"))
}
orcs_runtime::OutputStyle::Warn => Some(format!("[WARN] {text}\n")),
orcs_runtime::OutputStyle::Error => Some(format!("[ERROR] {text}\n")),
orcs_runtime::OutputStyle::Success => Some(format!("\x1B[32m{text}\x1B[0m\n")),
orcs_runtime::OutputStyle::Debug => {
if verbose {
Some(format!("[DEBUG] {text}\n"))
} else {
None
}
}
},
IOOutput::Prompt { message } => Some(format!("{message} ")),
IOOutput::ShowApprovalRequest {
id,
operation,
description,
} => Some(format!(
"\n [{operation}] {id} - {description}\n Enter 'y' to approve, 'n' to reject:\n"
)),
IOOutput::ShowApproved { approval_id } => {
Some(format!(" \u{2713} Approved: {approval_id}\n"))
}
IOOutput::ShowRejected {
approval_id,
reason,
} => {
if let Some(reason) = reason {
Some(format!(" \u{2717} Rejected: {approval_id} ({reason})\n"))
} else {
Some(format!(" \u{2717} Rejected: {approval_id}\n"))
}
}
IOOutput::ShowProcessing {
component,
operation,
} => Some(format!(" [{component}] Processing ({operation})...\n")),
IOOutput::Clear => Some("\x1B[2J\x1B[1;1H".to_string()),
}
}
pub async fn run_command(&mut self, command: &str) -> Result<i32, AppError> {
self.engine.start();
let trimmed = command.trim();
if trimmed.is_empty() {
self.engine.stop();
self.engine.shutdown().await;
return Ok(0);
}
if trimmed.starts_with('@') {
let cmd = InputParser.parse(trimmed);
match cmd {
InputCommand::ComponentMessage {
ref target,
ref message,
} => {
self.handle_component_message(target, message);
}
_ => {
eprintln!("Invalid component target: {trimmed}");
self.engine.stop();
self.engine.shutdown().await;
return Ok(1);
}
}
} else {
let io_input = IOInput::line(command);
if let Err(e) = self.io_input.try_send(io_input) {
eprintln!("Failed to send command: {e:?}");
self.engine.stop();
self.engine.shutdown().await;
return Ok(1);
}
}
const IDLE: tokio::time::Duration = tokio::time::Duration::from_secs(3);
const MAX_WAIT: tokio::time::Duration = tokio::time::Duration::from_secs(120);
let start = tokio::time::Instant::now();
let mut received_output = false;
let mut exit_code: i32 = 0;
loop {
if start.elapsed() > MAX_WAIT {
tracing::warn!(
"Command mode: max timeout reached ({}s)",
MAX_WAIT.as_secs()
);
break;
}
let timeout = if received_output { IDLE } else { MAX_WAIT };
tokio::select! {
Some(io_out) = self.io_output.recv() => {
match &io_out {
IOOutput::Print { text, style } => {
received_output = true;
if style.is_warning_or_error() {
exit_code = 1;
eprintln!("{text}");
} else {
println!("{text}");
}
}
IOOutput::Prompt { .. } => {
break;
}
IOOutput::ShowApprovalRequest { .. } => {
eprintln!("Command requires interactive approval. Use interactive mode.");
exit_code = 1;
break;
}
_ => {}
}
}
() = tokio::time::sleep(timeout) => {
tracing::debug!(
"Command mode: timeout (received_output={received_output})",
);
break;
}
}
}
self.engine.stop();
self.engine.shutdown().await;
Ok(exit_code)
}
pub async fn run_interactive(&mut self) -> Result<(), AppError> {
let io_id = self.engine.io_channel();
tracing::info!("Starting interactive mode (IO channel: {})", io_id);
self.engine.start();
self.renderer.render_output(&IOOutput::info(
"Interactive mode started. Type 'q' to quit, 'help' for commands.",
));
if self.is_resumed {
self.renderer.render_output(&IOOutput::info(format!(
"Resumed session: {} ({} component(s) restored)",
self.session.id, self.restored_count
)));
} else {
self.renderer
.render_output(&IOOutput::info(format!("Session ID: {}", self.session.id)));
}
let (mut readline_rx, printer) = Self::spawn_readline_thread(self.history_path());
if let Some(p) = printer {
self.printer_slot.set(p);
}
let mut should_save = false;
tokio::task::yield_now().await;
loop {
if !self.engine.is_running() {
break;
}
tokio::select! {
biased;
Some(io_out) = self.io_output.recv() => {
self.handle_io_output(io_out);
}
event = readline_rx.recv() => {
match event {
Some(ReadlineEvent::Line(line)) => {
match self.handle_input(&line) {
LoopControl::Continue => {}
LoopControl::Exit => break,
LoopControl::Pause => {
should_save = true;
break;
}
}
}
Some(ReadlineEvent::Eof) | None => {
tracing::debug!("readline: EOF or channel closed");
break;
}
}
}
}
}
self.printer_slot.clear();
self.engine.stop();
self.engine.shutdown().await;
if should_save {
self.renderer
.render_output(&IOOutput::info("Saving session..."));
self.save_session().await?;
self.renderer.render_output(&IOOutput::info(format!(
"Session saved: {}",
self.session.id
)));
self.renderer.render_output(&IOOutput::info(format!(
"Resume with: orcs --resume {}",
self.session.id
)));
}
self.renderer
.render_output(&IOOutput::info("Shutting down..."));
Ok(())
}
fn handle_input(&mut self, line: &str) -> LoopControl {
let cmd = InputParser.parse(line);
match &cmd {
InputCommand::Quit => {
tracing::info!("Quit requested");
return LoopControl::Exit;
}
InputCommand::Veto => {
tracing::warn!("Veto signal sent");
self.send_io_input(line);
return LoopControl::Exit;
}
InputCommand::Approve { approval_id } => {
self.handle_approve_via_io(line, approval_id.as_deref());
}
InputCommand::Reject {
approval_id,
reason: _,
} => {
self.handle_reject_via_io(line, approval_id.as_deref());
}
InputCommand::Pause => {
tracing::info!("Pause requested");
return LoopControl::Pause;
}
InputCommand::Resume => {
self.render_safe(&IOOutput::info(
"Resume is handled at startup. Use: orcs --resume",
));
}
InputCommand::Steer { message } => {
let io_id = self.engine.io_channel();
self.engine.signal(Signal::steer(
io_id,
message.as_str(),
Principal::User(PrincipalId::new()),
));
self.render_safe(&IOOutput::info(format!("Steer: {message}")));
}
InputCommand::Empty => {
}
InputCommand::ComponentMessage { target, message } => {
self.handle_component_message(target, message);
}
InputCommand::Unknown { input } => {
if input == "help" {
self.show_help();
} else if !input.is_empty() {
self.handle_echo_input(input.as_str());
}
}
}
LoopControl::Continue
}
fn send_io_input(&self, line: &str) {
let context = self.build_input_context();
let io_input = IOInput::line_with_context(line, context);
if let Err(e) = self.io_input.try_send(io_input) {
tracing::warn!("Failed to send IOInput: {:?}", e);
}
}
fn build_input_context(&self) -> InputContext {
match &self.pending_approval {
Some(pending) => InputContext::with_approval_id(&pending.id),
None => InputContext::empty(),
}
}
fn handle_echo_input(&mut self, input: &str) {
let io_input = IOInput::line(input);
if let Err(e) = self.io_input.try_send(io_input) {
self.render_safe(&IOOutput::error(format!("Failed to send message: {e:?}")));
}
}
fn handle_component_message(&mut self, target: &str, message: &str) {
let channel_id = match self.component_routes.get(target) {
Some(id) => *id,
None => {
let available: Vec<&str> =
self.component_routes.keys().map(|k| k.as_str()).collect();
self.render_safe(&IOOutput::error(format!(
"Unknown component: @{}. Available: {}",
target,
available.join(", ")
)));
return;
}
};
let event = orcs_runtime::Event {
category: orcs_event::EventCategory::UserInput,
operation: "input".to_string(),
source: orcs_types::ComponentId::builtin("app"),
payload: serde_json::json!({ "message": message }),
};
match self.engine.inject_event(channel_id, event) {
Ok(()) => {
tracing::debug!("Routed @{} to channel {}", target, channel_id);
}
Err(e) => {
self.render_safe(&IOOutput::error(format!(
"Failed to route to @{}: {}",
target, e
)));
}
}
}
fn handle_approve_via_io(&mut self, _line: &str, explicit_id: Option<&str>) {
let approval_id = explicit_id
.map(String::from)
.or_else(|| self.pending_approval.as_ref().map(|p| p.id.clone()));
if let Some(id) = approval_id {
let signal = Signal::approve(&id, self.principal.clone());
self.engine.signal(signal);
tracing::info!(approval_id = %id, "Approved");
if self.pending_approval.as_ref().map(|p| p.id.as_str()) == Some(id.as_str()) {
self.pending_approval = None;
}
} else {
self.render_safe(&IOOutput::warn("No pending approval. Use: y <id>"));
}
}
fn handle_reject_via_io(&mut self, _line: &str, explicit_id: Option<&str>) {
let approval_id = explicit_id
.map(String::from)
.or_else(|| self.pending_approval.as_ref().map(|p| p.id.clone()));
if let Some(id) = approval_id {
let signal = Signal::reject(&id, None, self.principal.clone());
self.engine.signal(signal);
tracing::info!(approval_id = %id, "Rejected");
if self.pending_approval.as_ref().map(|p| p.id.as_str()) == Some(id.as_str()) {
self.pending_approval = None;
}
} else {
self.render_safe(&IOOutput::warn("No pending approval. Use: n <id> [reason]"));
}
}
fn show_help(&self) {
self.render_safe(&IOOutput::info(
"Commands:\n\
\x20 y [id] - Approve pending request\n\
\x20 n [id] [reason] - Reject pending request\n\
\x20 p / pause - Pause execution\n\
\x20 r / resume - Resume execution\n\
\x20 s <message> - Steer with instruction\n\
\x20 q / quit - Quit application\n\
\x20 veto / stop - Emergency stop",
));
}
fn handle_io_output(&mut self, io_out: IOOutput) {
if let IOOutput::ShowApprovalRequest { ref id, .. } = io_out {
self.pending_approval = Some(PendingApproval { id: id.clone() });
}
self.render_safe(&io_out);
}
}
#[cfg(test)]
mod tests {
use super::*;
use orcs_runtime::sandbox::ProjectSandbox;
use orcs_runtime::NoOpResolver;
use orcs_runtime::WorkDir;
use std::path::PathBuf;
fn test_policy() -> Arc<dyn orcs_runtime::sandbox::SandboxPolicy> {
Arc::new(ProjectSandbox::new(".").expect("test sandbox"))
}
#[test]
fn orcs_helper_forces_refresh_on_char_insert() {
let helper = OrcsHelper;
assert!(
helper.highlight_char("", 0, CmdKind::Other),
"CmdKind::Other (edit_insert) should force full line refresh"
);
}
#[test]
fn orcs_helper_skips_refresh_on_cursor_move() {
let helper = OrcsHelper;
assert!(
!helper.highlight_char("abc", 1, CmdKind::MoveCursor),
"CmdKind::MoveCursor should not trigger refresh"
);
}
#[test]
fn orcs_helper_skips_refresh_on_forced_refresh() {
let helper = OrcsHelper;
assert!(
!helper.highlight_char("abc", 0, CmdKind::ForcedRefresh),
"CmdKind::ForcedRefresh should not trigger extra refresh from helper"
);
}
#[tokio::test]
async fn builder_default() {
let app = OrcsApp::builder(NoOpResolver)
.with_sandbox(test_policy())
.build()
.await
.expect("builder should succeed with NoOpResolver");
assert!(app.pending_approval_id().is_none());
}
#[tokio::test]
async fn builder_with_custom_resolver() {
struct VerboseResolver;
impl ConfigResolver for VerboseResolver {
fn resolve(&self) -> Result<OrcsConfig, orcs_runtime::ConfigError> {
let mut config = OrcsConfig::default();
config.ui.verbose = true;
Ok(config)
}
}
let app = OrcsApp::builder(VerboseResolver)
.with_sandbox(test_policy())
.build()
.await
.expect("builder should succeed with VerboseResolver");
assert!(app.config().ui.verbose);
}
#[tokio::test]
async fn app_engine_access() {
let app = OrcsApp::builder(NoOpResolver)
.with_sandbox(test_policy())
.build()
.await
.expect("builder should succeed");
assert!(!app.engine().is_running());
}
#[tokio::test]
async fn app_engine_parallel_access() {
let app = OrcsApp::builder(NoOpResolver)
.with_sandbox(test_policy())
.build()
.await
.expect("builder should succeed");
let io = app.engine().io_channel();
let world = app.engine().world_read();
let w = world.read().await;
assert!(w.get(&io).is_some());
}
#[tokio::test]
async fn app_session_id() {
let app = OrcsApp::builder(NoOpResolver)
.with_sandbox(test_policy())
.build()
.await
.expect("builder should succeed");
assert!(!app.session_id().is_empty());
}
#[tokio::test]
async fn app_new_session_not_resumed() {
let app = OrcsApp::builder(NoOpResolver)
.with_sandbox(test_policy())
.build()
.await
.expect("builder should succeed");
assert!(!app.is_resumed());
assert_eq!(app.restored_count(), 0);
}
#[tokio::test]
async fn app_save_and_load_session() {
let wd = WorkDir::temporary().expect("should create temp WorkDir for session test");
let temp_dir = wd.path().to_path_buf();
struct SessionResolver {
session_dir: PathBuf,
}
impl ConfigResolver for SessionResolver {
fn resolve(&self) -> Result<OrcsConfig, orcs_runtime::ConfigError> {
let mut config = OrcsConfig::default();
config.paths.session_dir = Some(self.session_dir.clone());
Ok(config)
}
}
let session_id = {
let mut app = OrcsApp::builder(SessionResolver {
session_dir: temp_dir.clone(),
})
.with_sandbox(test_policy())
.build()
.await
.expect("builder should succeed with SessionResolver");
let id = app.session_id().to_string();
app.save_session()
.await
.expect("save session should succeed");
id
};
let app = OrcsApp::builder(SessionResolver {
session_dir: temp_dir.clone(),
})
.with_sandbox(test_policy())
.resume(&session_id)
.build()
.await
.expect("resume should succeed for saved session");
assert!(app.is_resumed());
assert_eq!(app.session_id(), session_id);
}
#[tokio::test]
async fn app_resume_nonexistent_session_fails() {
let wd = WorkDir::temporary().expect("should create temp WorkDir for resume test");
let temp_dir = wd.path().to_path_buf();
struct SessionResolver {
session_dir: PathBuf,
}
impl ConfigResolver for SessionResolver {
fn resolve(&self) -> Result<OrcsConfig, orcs_runtime::ConfigError> {
let mut config = OrcsConfig::default();
config.paths.session_dir = Some(self.session_dir.clone());
Ok(config)
}
}
let result = OrcsApp::builder(SessionResolver {
session_dir: temp_dir.clone(),
})
.with_sandbox(test_policy())
.resume("nonexistent-session-id")
.build()
.await;
assert!(result.is_err());
}
#[tokio::test]
async fn app_reload_config() {
use std::sync::atomic::{AtomicBool, Ordering};
struct ToggleResolver {
debug: Arc<AtomicBool>,
}
impl ConfigResolver for ToggleResolver {
fn resolve(&self) -> Result<OrcsConfig, orcs_runtime::ConfigError> {
Ok(OrcsConfig {
debug: self.debug.load(Ordering::Relaxed),
..OrcsConfig::default()
})
}
}
let flag = Arc::new(AtomicBool::new(false));
let mut app = OrcsApp::builder(ToggleResolver {
debug: Arc::clone(&flag),
})
.with_sandbox(test_policy())
.build()
.await
.expect("builder should succeed with ToggleResolver");
assert!(!app.config().debug);
flag.store(true, Ordering::Relaxed);
app.reload_config().expect("reload config should succeed");
assert!(app.config().debug);
}
}