use std::fs;
use std::io::{self, BufRead, IsTerminal, Write};
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;
use async_trait::async_trait;
use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyModifiers};
use tokio::sync::Mutex;
use super::args::{CliArgs, CliCommand, ConfigureCommand, OutputMode};
use super::pairing::{
ApprovedBrowsersState, PairingCommandOptions, StdinPairingConfirmation, pairing_store_error,
run_client_command,
};
use super::state::{RuntimeScreenModel, StatusViewModel, WizardAction, WizardState};
use super::tui::{TerminalControl, TerminalSession, render_running, render_status, render_wizard};
use crate::config::app::{
ConfigLayer, IrohRelayConfig, ProviderId, StoredConfig, resolve_layers,
validate_frontend_origin,
};
use crate::config::paths::{AppPaths, PathEnvironment};
use crate::config::store::FileConfigStore;
use crate::dependencies::command::{CommandRunner, TokioCommandRunner};
use crate::dependencies::npm::{CLAUDE_PACKAGE, ManagedNodeRuntime, ManagedNpm};
use crate::dependencies::pi::{ManagedPi, ManagedPiRuntime};
use crate::dependencies::platform::Platform;
use crate::domain::errors::{AgentError, AgentResult, ErrorCode};
use crate::domain::policy::CommandPolicy;
use crate::infrastructure;
use crate::operational::pi_rpc_manager::PiRpcManager;
use crate::operational::session_manager::SessionManager;
use crate::operational::skill_service::SkillService;
use crate::pairing::{ClientStore, FileClientStore, ServerIdentity};
use crate::providers::{
descriptor::descriptor,
login::{LoginHandoff, TokioInheritedSpawner, TokioInterruptSignal},
probe::{PiAuthMetadata, ProviderProbe, ProviderProbeResult, ProviderProbeRuntime, ToolLaunch},
};
use crate::runtime::server;
use crate::runtime::{
DaemonReadyWriter, IrohEndpointStarter, RuntimeDependencies, RuntimeOptions, RuntimePhase,
RuntimeStatus, RuntimeSupervisor,
};
use crate::setup::flow::{
DelegatedReadiness, SetupConfigStore, SetupConsent, SetupConsentKind, SetupFlow, SetupOptions,
};
use crate::setup::probes::{
FilePiAuthMetadata, LocalPiProbe, LocalPiSetupOperations, LocalProvidersProbe,
LocalSystemOperations, LocalSystemProbe, PiSetupOperations, ProviderExtensionInstaller,
ProviderLogin, ProviderRuntimeSource, check_pairing_readiness,
};
use crate::setup::readiness::{CheckId, CheckResult, CheckStatus, ReadinessReport};
use crate::transport::{iroh_endpoint::IrohEndpointFactory, iroh_identity::IrohIdentityStore};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum LaunchMode {
Interactive,
Status,
Setup,
Configure,
Doctor,
Start,
Stop,
NonInteractiveDiagnostic,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RuntimeSelection {
Integrated,
LegacyOutbound,
}
pub(crate) fn select_runtime(
command: Option<&CliCommand>,
legacy_credentials_available: bool,
) -> RuntimeSelection {
if command.is_none() && legacy_credentials_available {
RuntimeSelection::LegacyOutbound
} else {
RuntimeSelection::Integrated
}
}
pub(crate) enum CliExit {
Success,
ActionRequired,
}
#[async_trait]
pub(crate) trait SetupCommand: Send + Sync {
async fn execute(&self) -> AgentResult<ReadinessReport>;
}
#[async_trait]
impl SetupCommand for SetupFlow<'_> {
async fn execute(&self) -> AgentResult<ReadinessReport> {
let _ = self.run_pi_stage().await?;
let _ = self.run_connectivity_stage()?;
self.check().await
}
}
pub(crate) async fn run_setup_command<W: Write>(
command: &dyn SetupCommand,
output_mode: OutputMode,
output: &mut W,
) -> AgentResult<CliExit> {
let report = command.execute().await?;
write_report(&report, output_mode, output)?;
Ok(if report.start_allowed {
CliExit::Success
} else {
CliExit::ActionRequired
})
}
impl CliExit {
pub(crate) fn code(&self) -> ExitCode {
match self {
Self::Success => ExitCode::SUCCESS,
Self::ActionRequired => ExitCode::from(2),
}
}
}
pub(crate) fn decide_launch_mode(
command: Option<&CliCommand>,
is_tty: bool,
is_complete: bool,
) -> LaunchMode {
match command {
Some(CliCommand::Status) => LaunchMode::Status,
Some(CliCommand::Setup) => LaunchMode::Setup,
Some(CliCommand::Configure(_)) => LaunchMode::Configure,
Some(CliCommand::Doctor) => LaunchMode::Doctor,
Some(CliCommand::Start(_)) => LaunchMode::Start,
Some(CliCommand::Stop) => LaunchMode::Stop,
None if is_tty => LaunchMode::Interactive,
None if is_complete => LaunchMode::Start,
None => LaunchMode::NonInteractiveDiagnostic,
}
}
pub(crate) async fn run(args: CliArgs) -> AgentResult<CliExit> {
let path_environment = PathEnvironment::from_env();
let mut stdout = io::stdout();
run_with_output(args, &path_environment, &mut stdout).await
}
pub(crate) async fn run_with_output<W: Write>(
args: CliArgs,
path_environment: &PathEnvironment,
output: &mut W,
) -> AgentResult<CliExit> {
let output_mode = args.output_mode();
let (detach, daemon_child) = match args.command.as_ref() {
Some(CliCommand::Start(start)) => (start.detach(), start.daemon_child()),
_ => (false, false),
};
let daemon_notifier = daemon_child
.then(DaemonReadyWriter::from_environment)
.transpose()?;
let can_use_tui = output_mode == OutputMode::Human
&& !args.non_interactive
&& io::stdin().is_terminal()
&& io::stdout().is_terminal();
let paths = AppPaths::resolve(args.config.clone(), path_environment)?;
if matches!(args.command, Some(CliCommand::Stop)) {
let outcome = crate::runtime::stop_server(&paths.server_lock_file()).await?;
write_stop_outcome(outcome, output_mode, output)?;
return Ok(CliExit::Success);
}
let legacy_ready = legacy_environment_complete();
let environment = prepare_environment_layer(ConfigLayer::from_env()?, legacy_ready);
if let Some(CliCommand::Configure(configure)) = args.command.as_ref()
&& let Some(ConfigureCommand::Clients(clients)) = configure.command.as_ref()
{
let identity =
ServerIdentity::load_or_create(&paths.server_key_file).map_err(pairing_store_error)?;
let server_id = identity.server_id().as_str().to_owned();
let mut store = FileClientStore::new(&paths.clients_file, identity);
let mut confirmation = StdinPairingConfirmation;
run_client_command(
&mut store,
&clients.command,
PairingCommandOptions {
server_id: &server_id,
json: output_mode == OutputMode::Json,
non_interactive: pairing_mutation_is_non_interactive(
args.non_interactive,
can_use_tui,
),
assume_yes: args.yes,
},
&mut confirmation,
output,
)?;
return Ok(CliExit::Success);
}
let store = FileConfigStore::new(&paths.config_file);
let stored = store.load()?;
let cli = ConfigLayer::default();
let effective = resolve_layers(stored, environment, cli)?;
let runtime_selection = select_runtime(args.command.as_ref(), legacy_ready);
let platform = Platform::detect()?;
let command_runner = TokioCommandRunner;
let setup_store = EffectiveSetupStore::new(store.clone(), effective.clone());
let system_operations = LocalSystemOperations;
let system = LocalSystemProbe::with_operations(platform, paths.clone(), &system_operations);
let pi_operations = LocalPiSetupOperations::new(paths.clone(), platform, &command_runner);
let pi = LocalPiProbe::new(platform.supports_managed_install(), &pi_operations);
let pi_agent_dir = path_environment
.home
.as_ref()
.map(|home| home.join(".pi/agent"))
.or_else(|| legacy_ready.then(|| paths.data_dir.join("legacy-unused-pi-agent")))
.ok_or_else(|| setup_environment_failed("HOME must be set for Pi provider setup"))?;
let provider_runtime = StoredProviderRuntimeSource {
store: &setup_store,
neutral_cwd: paths.data_dir.clone(),
};
let pi_auth = FilePiAuthMetadata::new(pi_agent_dir.join("auth.json"));
let extension_installer = ManagedProviderExtensions {
store: &setup_store,
runner: &command_runner,
pi_setup: &pi_operations,
paths: paths.clone(),
pi_agent_dir: pi_agent_dir.clone(),
};
let provider_login = CliProviderLogin::new(&provider_runtime, &command_runner, &pi_auth);
let providers = LocalProvidersProbe::new(
&provider_runtime,
&command_runner,
&pi_auth,
&extension_installer,
&provider_login,
);
let consent = CliSetupConsent;
let pairing = check_pairing_readiness(&paths);
let mut delegated = DelegatedReadiness::foundation(legacy_ready);
delegated.pairing = pairing.clone();
if runtime_selection == RuntimeSelection::Integrated {
delegated.connectivity = connectivity_readiness(&effective);
}
let setup = SetupFlow::new(
&setup_store,
&system,
&pi,
&providers,
&consent,
SetupOptions {
interactive: can_use_tui,
assume_yes: args.yes,
selected_providers: None,
},
delegated,
);
let mut report = if legacy_ready {
foundation_readiness_for_runtime(&effective, true, pairing, runtime_selection)
} else {
setup.check().await?
};
let mode = decide_launch_mode(args.command.as_ref(), can_use_tui, report.start_allowed);
if mode == LaunchMode::Status {
report.iroh_endpoint_id = status_iroh_endpoint_id(&paths)?;
}
tracing::debug!(
config_file = %store.path().display(),
data_dir = %paths.data_dir.display(),
cache_dir = %paths.cache_dir.display(),
assume_yes = args.yes,
"resolved CLI configuration"
);
match mode {
LaunchMode::Status | LaunchMode::Doctor => {
write_report(&report, output_mode, output)?;
Ok(CliExit::Success)
}
LaunchMode::Setup | LaunchMode::Configure => {
if can_use_tui {
run_wizard_tui(&setup, effective.selected_providers.clone(), &paths).await?;
} else {
return run_setup_command(&setup, output_mode, output).await;
}
let refreshed = setup.check().await?;
if refreshed.start_allowed {
Ok(CliExit::Success)
} else {
Ok(CliExit::ActionRequired)
}
}
LaunchMode::Interactive => {
if report.start_allowed {
run_status_tui(&report)?;
} else {
run_wizard_tui(&setup, effective.selected_providers.clone(), &paths).await?;
}
Ok(CliExit::Success)
}
LaunchMode::Start if !report.start_allowed => {
if can_use_tui {
run_status_tui(&report)?;
} else {
write_report(&report, output_mode, output)?;
}
Ok(CliExit::ActionRequired)
}
LaunchMode::Start => {
if detach {
let link = crate::runtime::launch_detached(&paths).await?;
write_detached_start(&link, output_mode, output)?;
return Ok(CliExit::Success);
}
match runtime_selection {
RuntimeSelection::LegacyOutbound => {
run_existing_runtime(paths, effective).await?;
}
RuntimeSelection::Integrated => {
run_integrated_runtime(
paths,
effective,
output_mode,
can_use_tui,
daemon_notifier,
output,
)
.await?;
}
}
Ok(CliExit::Success)
}
LaunchMode::NonInteractiveDiagnostic => Err(AgentError::new(
ErrorCode::InvalidMessage,
"PC Agent setup is incomplete; run `regy-pc-agent setup` in a terminal or inspect `regy-pc-agent --json status`",
)),
LaunchMode::Stop => unreachable!("stop is dispatched before setup"),
}
}
pub(crate) fn write_detached_start<W: Write>(
frontend_deep_link: &str,
output_mode: OutputMode,
output: &mut W,
) -> AgentResult<()> {
match output_mode {
OutputMode::Json => {
serde_json::to_writer(
&mut *output,
&serde_json::json!({ "frontend_deep_link": frontend_deep_link }),
)
.map_err(|error| output_error(format!("failed to write detached status: {error}")))?;
writeln!(output)
.map_err(|error| output_error(format!("failed to write detached status: {error}")))
}
OutputMode::Human => writeln!(output, "{frontend_deep_link}")
.map_err(|error| output_error(format!("failed to write detached status: {error}"))),
}
}
fn write_stop_outcome<W: Write>(
outcome: crate::runtime::StopOutcome,
output_mode: OutputMode,
output: &mut W,
) -> AgentResult<()> {
match output_mode {
OutputMode::Json => {
serde_json::to_writer(&mut *output, &serde_json::json!({ "state": outcome }))
.map_err(|error| output_error(format!("failed to write stop status: {error}")))?;
writeln!(output)
.map_err(|error| output_error(format!("failed to write stop status: {error}")))
}
OutputMode::Human => writeln!(
output,
"{}",
match outcome {
crate::runtime::StopOutcome::Stopped => "PC Agent server stopped.",
crate::runtime::StopOutcome::NotRunning => {
"PC Agent server is not running."
}
}
)
.map_err(|error| output_error(format!("failed to write stop status: {error}"))),
}
}
pub(crate) fn pairing_mutation_is_non_interactive(
explicitly_non_interactive: bool,
can_use_tui: bool,
) -> bool {
explicitly_non_interactive || !can_use_tui
}
#[cfg(test)]
pub(crate) fn foundation_readiness(
config: &StoredConfig,
legacy_ready: bool,
pairing: CheckResult,
) -> ReadinessReport {
foundation_readiness_for_runtime(config, legacy_ready, pairing, RuntimeSelection::Integrated)
}
fn foundation_readiness_for_runtime(
config: &StoredConfig,
legacy_ready: bool,
pairing: CheckResult,
runtime_selection: RuntimeSelection,
) -> ReadinessReport {
let pi_ready = legacy_ready || config.tools.pi_command.is_some();
let connectivity = match runtime_selection {
RuntimeSelection::Integrated => connectivity_readiness(config),
RuntimeSelection::LegacyOutbound => check(CheckId::Tunnel, CheckStatus::Ready, false, None),
};
ReadinessReport::from_checks(vec![
check(CheckId::System, CheckStatus::Ready, true, None),
check(
CheckId::Pi,
required_status(pi_ready),
true,
(!pi_ready).then_some("run setup to locate or install Pi"),
),
pairing,
connectivity,
])
}
fn required_status(ready: bool) -> CheckStatus {
if ready {
CheckStatus::Ready
} else {
CheckStatus::ActionRequired
}
}
pub(crate) fn connectivity_readiness(config: &StoredConfig) -> CheckResult {
if config
.frontend_origin
.as_deref()
.is_none_or(|origin| validate_frontend_origin(origin).is_err())
{
check(
CheckId::Tunnel,
CheckStatus::ActionRequired,
true,
Some("frontend_origin must be an exact HTTPS origin"),
)
} else {
check(CheckId::Tunnel, CheckStatus::Ready, true, None)
}
}
fn check(id: CheckId, status: CheckStatus, required: bool, action: Option<&str>) -> CheckResult {
CheckResult {
id,
status,
required,
detail: None,
action: action.map(str::to_string),
actions: Vec::new(),
}
}
fn legacy_environment_complete() -> bool {
["PC_AGENT_BACKEND_URL", "PC_AGENT_TOKEN"]
.into_iter()
.all(|key| {
std::env::var_os(key).is_some_and(|value| !value.to_string_lossy().trim().is_empty())
})
}
pub(crate) fn prepare_environment_layer(
mut environment: ConfigLayer,
legacy_ready: bool,
) -> ConfigLayer {
if !legacy_ready {
return environment;
}
if let Some(tools) = &mut environment.tools {
clear_relative_command(&mut tools.pi_command);
}
if let Some(server) = &mut environment.server {
clear_relative_command(&mut server.default_command);
server.allowed_commands = None;
server.allowed_workdirs = None;
}
environment
}
fn clear_relative_command(command: &mut Option<Vec<String>>) {
if command
.as_ref()
.and_then(|arguments| arguments.first())
.is_some_and(|program| !Path::new(program).is_absolute())
{
*command = None;
}
}
struct EffectiveSetupStore {
file: FileConfigStore,
current: StdMutex<StoredConfig>,
}
impl EffectiveSetupStore {
fn new(file: FileConfigStore, current: StoredConfig) -> Self {
Self {
file,
current: StdMutex::new(current),
}
}
}
impl SetupConfigStore for EffectiveSetupStore {
fn load(&self) -> AgentResult<StoredConfig> {
self.current
.lock()
.map(|config| config.clone())
.map_err(|_| setup_environment_failed("setup configuration state is unavailable"))
}
fn save(&self, config: &StoredConfig) -> AgentResult<()> {
self.file.save(config)?;
*self
.current
.lock()
.map_err(|_| setup_environment_failed("setup configuration state is unavailable"))? =
config.clone();
Ok(())
}
}
struct StoredProviderRuntimeSource<'a> {
store: &'a dyn SetupConfigStore,
neutral_cwd: PathBuf,
}
impl ProviderRuntimeSource for StoredProviderRuntimeSource<'_> {
fn runtime(&self) -> AgentResult<ProviderProbeRuntime> {
let tools = self.store.load()?.tools;
let pi = tools
.pi_command
.as_deref()
.and_then(tool_launch)
.ok_or_else(|| setup_environment_failed("Pi must be ready before provider setup"))?;
let claude = tools.claude_command.as_deref().and_then(tool_launch);
Ok(ProviderProbeRuntime::new(
pi,
claude,
self.neutral_cwd.clone(),
))
}
}
fn tool_launch(command: &[String]) -> Option<ToolLaunch> {
match command {
[program] => Some(ToolLaunch::direct(program)),
[node, script] => Some(ToolLaunch::managed_node(node, script)),
_ => None,
}
}
struct ManagedProviderExtensions<'a> {
store: &'a dyn SetupConfigStore,
runner: &'a dyn CommandRunner,
pi_setup: &'a dyn PiSetupOperations,
paths: AppPaths,
pi_agent_dir: PathBuf,
}
#[async_trait]
impl ProviderExtensionInstaller for ManagedProviderExtensions<'_> {
async fn install(&self, provider: ProviderId) -> AgentResult<()> {
let Some(extension) = descriptor(provider).extension else {
return Ok(());
};
prepare_private_directory(&self.pi_agent_dir)?;
let tools_path = self.paths.data_dir.join("tools");
let mut config =
ensure_managed_provider_toolchain(self.store, self.pi_setup, &tools_path).await?;
let runtime = managed_extension_runtime(&config, &tools_path)
.ok_or_else(|| setup_environment_failed("managed provider toolchain is unavailable"))?;
if provider == ProviderId::ClaudeCode && config.tools.claude_command.is_none() {
let installed = ManagedNpm::new(
ManagedNodeRuntime::new(&runtime.node, &runtime.npm, &runtime.tools),
self.runner,
)
.install(&CLAUDE_PACKAGE)
.await?;
let executable = fs::canonicalize(installed.executable).map_err(|_| {
setup_environment_failed("managed Claude executable is unavailable")
})?;
config.tools.claude_command =
Some(paths_to_command(&[runtime.node.clone(), executable])?);
self.store.save(&config)?;
}
let settings = self.pi_agent_dir.join("settings.json");
ManagedPi::new(
ManagedPiRuntime::new(
runtime.pi,
runtime.node,
runtime.npm,
runtime.tools,
settings,
),
self.runner,
)
.install_extension(&extension.package)
.await
}
}
pub(crate) async fn ensure_managed_provider_toolchain(
store: &dyn SetupConfigStore,
pi_setup: &dyn PiSetupOperations,
tools: &Path,
) -> AgentResult<StoredConfig> {
let mut config = store.load()?;
if managed_extension_runtime(&config, tools).is_some() {
return Ok(config);
}
let installed = pi_setup.install().await?;
config.tools.node_command = Some(installed.node);
config.tools.pi_command = Some(installed.pi);
config.tools.claude_command = installed.claude;
if managed_extension_runtime(&config, tools).is_none() {
return Err(setup_environment_failed(
"managed provider toolchain could not be verified",
));
}
store.save(&config)?;
Ok(config)
}
struct ManagedExtensionRuntime {
tools: PathBuf,
node: PathBuf,
npm: PathBuf,
pi: PathBuf,
}
fn managed_extension_runtime(
config: &StoredConfig,
tools: &Path,
) -> Option<ManagedExtensionRuntime> {
let tools = fs::canonicalize(tools).ok()?;
let node = single_command_path(config.tools.node_command.as_deref(), "Node").ok()?;
let pi = managed_script_path(config.tools.pi_command.as_deref(), &node, "Pi").ok()?;
let npm = fs::canonicalize(node.parent()?.join("npm")).ok()?;
if [node.as_path(), npm.as_path(), pi.as_path()]
.iter()
.any(|path| !path.starts_with(&tools))
{
return None;
}
Some(ManagedExtensionRuntime {
tools,
node,
npm,
pi,
})
}
fn prepare_private_directory(path: &Path) -> AgentResult<()> {
match fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_dir() => {
return Err(setup_environment_failed(
"Pi agent directory must be a private directory",
));
}
Ok(_) => {}
Err(error) if error.kind() == io::ErrorKind::NotFound => fs::create_dir_all(path)
.map_err(|_| setup_environment_failed("Pi agent directory could not be created"))?,
Err(_) => {
return Err(setup_environment_failed(
"Pi agent directory could not be inspected",
));
}
}
fs::set_permissions(path, fs::Permissions::from_mode(0o700))
.map_err(|_| setup_environment_failed("Pi agent directory could not be secured"))
}
fn single_command_path(command: Option<&[String]>, name: &'static str) -> AgentResult<PathBuf> {
let Some([program]) = command else {
return Err(setup_environment_failed(match name {
"Node" => "managed Node command must contain exactly one path",
_ => "managed tool command must contain exactly one path",
}));
};
fs::canonicalize(program)
.map_err(|_| setup_environment_failed("managed tool executable is unavailable"))
}
fn managed_script_path(
command: Option<&[String]>,
node: &Path,
name: &'static str,
) -> AgentResult<PathBuf> {
let Some([command_node, script]) = command else {
return Err(setup_environment_failed(match name {
"Pi" => "managed Pi command must contain Node and script paths",
_ => "managed tool command must contain Node and script paths",
}));
};
let command_node = fs::canonicalize(command_node)
.map_err(|_| setup_environment_failed("managed Node executable is unavailable"))?;
if command_node != node {
return Err(setup_environment_failed(
"managed Pi and Node commands must use the same runtime",
));
}
fs::canonicalize(script)
.map_err(|_| setup_environment_failed("managed Pi executable is unavailable"))
}
fn paths_to_command(paths: &[PathBuf]) -> AgentResult<Vec<String>> {
paths
.iter()
.map(|path| {
path.to_str()
.map(str::to_owned)
.ok_or_else(|| setup_environment_failed("managed tool path is not valid UTF-8"))
})
.collect()
}
struct CliProviderLogin<'a> {
runtime: &'a dyn ProviderRuntimeSource,
runner: &'a dyn CommandRunner,
pi_auth: &'a dyn PiAuthMetadata,
}
impl<'a> CliProviderLogin<'a> {
fn new(
runtime: &'a dyn ProviderRuntimeSource,
runner: &'a dyn CommandRunner,
pi_auth: &'a dyn PiAuthMetadata,
) -> Self {
Self {
runtime,
runner,
pi_auth,
}
}
}
#[async_trait]
impl ProviderLogin for CliProviderLogin<'_> {
async fn login(&self, provider: ProviderId) -> AgentResult<ProviderProbeResult> {
let runtime = self.runtime.runtime()?;
let probe = ProviderProbe::new(runtime.clone(), self.runner, self.pi_auth);
let mut terminal = ConsoleTerminal;
LoginHandoff::new(
runtime,
&mut terminal,
&TokioInheritedSpawner,
&TokioInterruptSignal,
&probe,
)
.run(provider)
.await
}
}
struct ConsoleTerminal;
impl TerminalControl for ConsoleTerminal {
fn suspend(&mut self) -> AgentResult<()> {
Ok(())
}
fn resume(&mut self) -> AgentResult<()> {
Ok(())
}
fn show_instruction(&mut self, instruction: &'static str) -> AgentResult<()> {
let mut stdout = io::stdout().lock();
writeln!(stdout, "{instruction}")
.and_then(|()| stdout.flush())
.map_err(|_| setup_environment_failed("provider login instruction could not be shown"))
}
}
struct CliSetupConsent;
impl SetupConsent for CliSetupConsent {
fn confirm(&self, _kind: SetupConsentKind, warning: &'static str) -> AgentResult<bool> {
prompt_for_consent(&mut io::stdin().lock(), &mut io::stdout().lock(), warning)
}
fn notify(&self, _kind: SetupConsentKind, warning: &'static str) -> AgentResult<()> {
let mut stderr = io::stderr().lock();
writeln!(stderr, "{warning}")
.and_then(|()| stderr.flush())
.map_err(|_| setup_environment_failed("setup warning could not be shown"))
}
}
pub(crate) fn prompt_for_consent<R: BufRead, W: Write>(
input: &mut R,
output: &mut W,
warning: &'static str,
) -> AgentResult<bool> {
writeln!(output, "{warning}")
.and_then(|()| write!(output, "Continue? [y/N] "))
.and_then(|()| output.flush())
.map_err(|_| setup_environment_failed("setup consent prompt could not be shown"))?;
let mut answer = String::new();
input
.read_line(&mut answer)
.map_err(|_| setup_environment_failed("setup consent response could not be read"))?;
Ok(matches!(
answer.trim().to_ascii_lowercase().as_str(),
"y" | "yes"
))
}
fn setup_environment_failed(message: &'static str) -> AgentError {
AgentError::new(ErrorCode::InvalidMessage, message)
}
pub(crate) fn write_report<W: Write>(
report: &ReadinessReport,
output_mode: OutputMode,
output: &mut W,
) -> AgentResult<()> {
match output_mode {
OutputMode::Json => {
serde_json::to_writer(&mut *output, report)
.map_err(|error| output_error(format!("failed to write JSON status: {error}")))?;
writeln!(output)
.map_err(|error| output_error(format!("failed to finish JSON status: {error}")))?;
}
OutputMode::Human => {
for item in &report.checks {
writeln!(output, "{:?}: {:?}", item.id, item.status).map_err(|error| {
output_error(format!("failed to write human status: {error}"))
})?;
if let Some(action) = &item.action {
writeln!(output, " {action}").map_err(|error| {
output_error(format!("failed to write human status: {error}"))
})?;
}
for action in &item.actions {
writeln!(output, " {}", action.title).map_err(|error| {
output_error(format!("failed to write human status: {error}"))
})?;
for (index, step) in action.steps.iter().enumerate() {
writeln!(output, " {}. {step}", index + 1).map_err(|error| {
output_error(format!("failed to write human status: {error}"))
})?;
}
}
}
if let Some(endpoint_id) = &report.iroh_endpoint_id {
writeln!(output, "Iroh Endpoint ID: {endpoint_id}").map_err(|error| {
output_error(format!("failed to write human status: {error}"))
})?;
}
writeln!(output, "Start allowed: {}", report.start_allowed)
.map_err(|error| output_error(format!("failed to write human status: {error}")))?;
}
}
Ok(())
}
fn status_iroh_endpoint_id(paths: &AppPaths) -> AgentResult<Option<String>> {
let key_file = paths.iroh_endpoint_key_file();
IrohIdentityStore::new(&key_file)
.load_existing()
.map(|identity| identity.map(|identity| identity.endpoint_id().to_string()))
.map_err(|_| {
AgentError::new(
ErrorCode::PairingStorageFailed,
"Iroh endpoint identity is invalid or unavailable; identity storage requires a parent directory owned by the current user with mode 0700 and a regular key file owned by the current user with mode 0600; remove the invalid key and run `regy-pc-agent start` to create a new identity",
)
})
}
pub(crate) fn write_runtime_status<W: Write>(
status: &RuntimeStatus,
output_mode: OutputMode,
output: &mut W,
) -> AgentResult<()> {
match output_mode {
OutputMode::Json => {
serde_json::to_writer(&mut *output, status)
.map_err(|error| output_error(format!("failed to write JSON runtime: {error}")))?;
writeln!(output)
.map_err(|error| output_error(format!("failed to finish JSON runtime: {error}")))?;
}
OutputMode::Human => {
writeln!(output, "Phase: {}", runtime_phase_label(status.phase)).map_err(|error| {
output_error(format!("failed to write runtime status: {error}"))
})?;
if let Some(endpoint_id) = &status.iroh_endpoint_id {
writeln!(output, "Iroh Endpoint ID: {endpoint_id}").map_err(|error| {
output_error(format!("failed to write runtime status: {error}"))
})?;
}
if let Some(code) = &status.pairing_code {
writeln!(output, "Pairing code: {code}").map_err(|error| {
output_error(format!("failed to write runtime status: {error}"))
})?;
}
if let Some(expires) = &status.pairing_expires_at {
writeln!(output, "Pairing expires: {expires}").map_err(|error| {
output_error(format!("failed to write runtime status: {error}"))
})?;
}
if let Some(link) = &status.frontend_deep_link {
writeln!(output, "Frontend: {link}").map_err(|error| {
output_error(format!("failed to write runtime status: {error}"))
})?;
}
if let Some(message) = &status.message {
writeln!(output, "Status: {message}").map_err(|error| {
output_error(format!("failed to write runtime status: {error}"))
})?;
}
}
}
Ok(())
}
fn runtime_phase_label(phase: RuntimePhase) -> &'static str {
match phase {
RuntimePhase::Starting => "starting",
RuntimePhase::Ready => "ready",
RuntimePhase::Degraded => "degraded",
RuntimePhase::Stopping => "stopping",
RuntimePhase::Stopped => "stopped",
}
}
pub(crate) fn is_quit_key(key: &KeyEvent) -> bool {
key.code == KeyCode::Esc
|| key.code == KeyCode::Char('q')
|| (key.code == KeyCode::Char('c') && key.modifiers.contains(KeyModifiers::CONTROL))
}
pub(crate) fn wizard_navigation_action(key: &KeyEvent) -> Option<WizardAction> {
match key.code {
KeyCode::Up | KeyCode::Left | KeyCode::Esc => Some(WizardAction::Back),
KeyCode::Down | KeyCode::Right => Some(WizardAction::Next),
_ => None,
}
}
async fn run_wizard_tui(
flow: &SetupFlow<'_>,
selected_providers: Vec<ProviderId>,
paths: &AppPaths,
) -> AgentResult<()> {
let mut session = TerminalSession::new().map_err(terminal_error)?;
let mut report = flow.check().await?;
let mut state = WizardState::from_report_with_providers(&report, selected_providers);
state.set_approved_browsers(load_approved_browsers(paths)?);
loop {
session
.terminal_mut()
.draw(|frame| render_wizard(frame, &state))
.map_err(terminal_error)?;
match event::read().map_err(terminal_error)? {
Event::Key(key) => {
if let Some(action) = wizard_navigation_action(&key) {
state.apply(action);
continue;
}
if is_quit_key(&key) {
return Ok(());
}
match key.code {
KeyCode::Char('1') if state.active() == CheckId::Providers => {
state.toggle_provider(ProviderId::OpenaiCodex);
}
KeyCode::Char('2') if state.active() == CheckId::Providers => {
state.toggle_provider(ProviderId::ClaudeCode);
}
KeyCode::Char('3') if state.active() == CheckId::Providers => {
state.toggle_provider(ProviderId::Antigravity);
}
KeyCode::Enter => {
match state.active() {
CheckId::System => {}
CheckId::Pi => {
drop(session);
let _ = flow.run_pi_stage().await?;
session = TerminalSession::new().map_err(terminal_error)?;
}
CheckId::Providers => {
let selected = state.selected_providers().to_vec();
drop(session);
let _ = flow.run_models_stage_with(selected).await?;
session = TerminalSession::new().map_err(terminal_error)?;
}
CheckId::Pairing => {}
CheckId::Tunnel => {
drop(session);
configure_connectivity(flow).await?;
session = TerminalSession::new().map_err(terminal_error)?;
}
}
report = flow.check().await?;
let selected = state.selected_providers().to_vec();
state = WizardState::from_report_with_providers(&report, selected);
state.set_approved_browsers(load_approved_browsers(paths)?);
}
_ => {}
}
}
Event::Resize(_, _) => {}
_ => {}
}
}
}
async fn configure_connectivity(flow: &SetupFlow<'_>) -> AgentResult<()> {
flow.run_connectivity_stage().map(|_| ())
}
fn load_approved_browsers(paths: &AppPaths) -> AgentResult<ApprovedBrowsersState> {
let identity =
ServerIdentity::load_or_create(&paths.server_key_file).map_err(pairing_store_error)?;
let store = FileClientStore::new(&paths.clients_file, identity);
let clients = ClientStore::list(&store).map_err(pairing_store_error)?;
Ok(ApprovedBrowsersState::new(clients))
}
fn run_status_tui(report: &ReadinessReport) -> AgentResult<()> {
let mut session = TerminalSession::new().map_err(terminal_error)?;
let model = StatusViewModel::from(report);
session
.terminal_mut()
.draw(|frame| render_status(frame, &model))
.map_err(terminal_error)?;
loop {
match event::read().map_err(terminal_error)? {
Event::Key(key) if is_quit_key(&key) || key.code == KeyCode::Enter => {
return Ok(());
}
Event::Resize(_, _) => {
session
.terminal_mut()
.draw(|frame| render_status(frame, &model))
.map_err(terminal_error)?;
}
_ => {}
}
}
}
fn terminal_error(error: io::Error) -> AgentError {
AgentError::new(
ErrorCode::InvalidMessage,
format!("terminal error: {error}"),
)
}
fn output_error(message: String) -> AgentError {
AgentError::new(ErrorCode::InvalidMessage, message)
}
pub(crate) fn integrated_frontend_origin(stored: &StoredConfig) -> AgentResult<String> {
let origin = stored.frontend_origin.clone().ok_or_else(|| {
AgentError::new(
ErrorCode::InvalidMessage,
"Iroh start requires a bare HTTPS frontend_origin",
)
})?;
crate::config::app::validate_frontend_origin(&origin).map_err(|_| {
AgentError::new(
ErrorCode::InvalidMessage,
"Iroh start requires a bare HTTPS frontend_origin",
)
})?;
Ok(origin)
}
async fn run_integrated_runtime<W: Write>(
paths: AppPaths,
stored: StoredConfig,
output_mode: OutputMode,
can_use_tui: bool,
daemon_notifier: Option<DaemonReadyWriter>,
output: &mut W,
) -> AgentResult<()> {
let mut config = crate::config::Config::from_env()?;
if let Some(command) = &stored.tools.pi_command {
config.pi_command = command.clone();
}
if let Some(command) = &stored.server.default_command {
config.default_command = command.clone();
}
config.allowed_commands = stored.server.allowed_commands.clone();
config.allowed_workdirs = stored.server.allowed_workdirs.clone();
let policy = CommandPolicy::new(
config.allowed_commands.clone(),
config.allowed_workdirs.clone(),
);
let skill_home_root = config
.home_dir
.as_deref()
.map(PathBuf::from)
.or_else(|| std::env::var_os("HOME").map(PathBuf::from))
.unwrap_or_else(|| PathBuf::from("."));
let pi_manager = Arc::new(PiRpcManager::new(
config.pi_command.clone(),
Duration::from_secs(config.pi_rpc_timeout_seconds),
));
let skill_service = SkillService::new(
Arc::new(infrastructure::skill_catalog::SkillCatalogRegistry::from_config(&config)?),
Arc::new(infrastructure::skill_store::SkillStore::new(
PathBuf::from(&config.pi_global_skills_dir),
skill_home_root.clone(),
policy.clone(),
infrastructure::skill_catalog::build_http_client()?,
)),
infrastructure::pi_skill_runtime::PiSkillRuntime::new(
config.pi_command.clone(),
Duration::from_secs(config.pi_rpc_timeout_seconds),
pi_manager.clone(),
),
skill_home_root,
);
let manager = Arc::new(Mutex::new(SessionManager::new(
policy,
config.default_command.clone(),
)));
let frontend_origin = integrated_frontend_origin(&stored)?;
let (ui_application, pairing) = server::integrated_ui_application(
&config,
manager.clone(),
pi_manager.clone(),
skill_service,
&paths,
)
.await?;
let relays = IrohRelayConfig::from_urls(&stored.server.iroh_relay_urls)?;
let iroh_endpoint_factory = Arc::new(IrohEndpointFactory::public(
IrohIdentityStore::new(paths.iroh_endpoint_key_file()),
relays,
ui_application.hub(),
pairing.clone(),
)) as Arc<dyn IrohEndpointStarter>;
let shutdown = Arc::new(server::AgentRuntimeShutdown::new(manager, pi_manager));
let supervisor = RuntimeSupervisor::with_dependencies(
RuntimeOptions { frontend_origin },
paths,
pairing.clone(),
RuntimeDependencies {
application: ui_application,
iroh_endpoint_factory,
shutdown,
},
);
let running = supervisor.start().await?;
if let Some(notifier) = daemon_notifier {
return run_daemon_runtime(&running, notifier, server::shutdown_signal()).await;
}
if can_use_tui {
return run_runtime_tui(&running, pairing).await;
}
run_runtime_status_stream(&running, output_mode, output, server::shutdown_signal()).await
}
pub(crate) fn daemon_startup_link(status: &RuntimeStatus) -> AgentResult<Option<&str>> {
if let Some(link) = status
.frontend_deep_link
.as_deref()
.filter(|link| !link.is_empty())
{
return Ok(Some(link));
}
if matches!(
status.phase,
RuntimePhase::Degraded | RuntimePhase::Stopping | RuntimePhase::Stopped
) {
return Err(AgentError::new(
ErrorCode::InvalidMessage,
"PC Agent detached startup ended before the frontend link was ready",
));
}
Ok(None)
}
async fn run_daemon_runtime<S>(
running: &crate::runtime::RunningRuntime,
notifier: DaemonReadyWriter,
shutdown: S,
) -> AgentResult<()>
where
S: std::future::Future<Output = ()>,
{
let mut statuses = running.subscribe_statuses();
let mut status = running.status();
let mut notifier = Some(notifier);
let mut ready = false;
tokio::pin!(shutdown);
let run_result = async {
loop {
if !ready {
match daemon_startup_link(&status) {
Ok(Some(link)) => {
notifier
.take()
.expect("daemon notifier is present before readiness")
.ready(link)?;
ready = true;
}
Ok(None) => {}
Err(error) => {
if let Some(notifier) = notifier.take() {
notifier.failed("PC Agent server did not become ready")?;
}
break Err(error);
}
}
}
if status.phase == RuntimePhase::Stopped {
break Ok(());
}
tokio::select! {
biased;
_ = &mut shutdown => {
if let Some(notifier) = notifier.take() {
notifier.failed("PC Agent server stopped before readiness")?;
}
break Ok(());
},
next = statuses.recv() => match next {
Ok(next) => status = next,
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
status = running.status();
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
if let Some(notifier) = notifier.take() {
notifier.failed("PC Agent runtime closed before readiness")?;
}
break Ok(());
}
}
}
}
}
.await;
let stop_result = running.stop().await;
preserve_primary(run_result, stop_result)
}
pub(crate) async fn run_runtime_status_stream<W, S>(
running: &crate::runtime::RunningRuntime,
output_mode: OutputMode,
output: &mut W,
shutdown: S,
) -> AgentResult<()>
where
W: Write,
S: std::future::Future<Output = ()>,
{
let mut statuses = running.subscribe_statuses();
let mut last = None;
tokio::pin!(shutdown);
let run_result = async {
emit_changed_runtime_status(&running.status(), &mut last, output_mode, output)?;
loop {
tokio::select! {
biased;
_ = &mut shutdown => return Ok(()),
status = statuses.recv() => {
match status {
Ok(status) => {
let stopped = status.phase == RuntimePhase::Stopped;
emit_changed_runtime_status(&status, &mut last, output_mode, output)?;
if stopped {
return Ok(());
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
emit_changed_runtime_status(&running.status(), &mut last, output_mode, output)?;
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
}
}
}
}
}
.await;
let stop_result = running.stop().await;
let drain_result = if run_result.is_ok() && stop_result.is_ok() {
let mut result = Ok(());
while let Ok(status) = statuses.try_recv() {
if let Err(error) = emit_changed_runtime_status(&status, &mut last, output_mode, output)
{
result = Err(error);
break;
}
}
result
} else {
Ok(())
};
preserve_primary(run_result, stop_result).and(drain_result)
}
async fn run_runtime_tui(
running: &crate::runtime::RunningRuntime,
pairing: Arc<crate::presentation::ui_handshake::UiPairingService>,
) -> AgentResult<()> {
let mut statuses = running.subscribe_statuses();
let (event_history, mut events) = running.subscribe_events_with_history();
let mut model = RuntimeScreenModel::new(running.status(), 0);
model.replace_events(event_history);
if let Err(error) = refresh_runtime_approved_client_count(&mut model, pairing.clone()).await {
let cleanup = running.stop().await;
return preserve_primary(Err(error), cleanup);
}
let mut session = match TerminalSession::new() {
Ok(session) => session,
Err(error) => {
let primary = Err(terminal_error(error));
let cleanup = running.stop().await;
return preserve_primary(primary, cleanup);
}
};
let run_result = async {
session
.terminal_mut()
.draw(|frame| render_running(frame, &model))
.map_err(terminal_error)?;
let shutdown = server::shutdown_signal();
tokio::pin!(shutdown);
let mut input_tick = tokio::time::interval(Duration::from_millis(50));
input_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let first_count_refresh = tokio::time::Instant::now() + Duration::from_secs(1);
let mut count_tick = tokio::time::interval_at(first_count_refresh, Duration::from_secs(1));
count_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tokio::select! {
biased;
_ = &mut shutdown => return Ok(()),
status = statuses.recv() => match status {
Ok(status) => {
let stopped = status.phase == RuntimePhase::Stopped;
model.update_status(status);
session
.terminal_mut()
.draw(|frame| render_running(frame, &model))
.map_err(terminal_error)?;
if stopped {
return Ok(());
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
model.update_status(running.status());
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
},
runtime_event = events.recv() => match runtime_event {
Ok(runtime_event) => {
model.push_event(runtime_event);
session
.terminal_mut()
.draw(|frame| render_running(frame, &model))
.map_err(terminal_error)?;
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
let (history, replacement) = running.subscribe_events_with_history();
model.replace_events(history);
events = replacement;
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => {}
},
_ = count_tick.tick() => {
if refresh_runtime_approved_client_count(&mut model, pairing.clone()).await? {
session
.terminal_mut()
.draw(|frame| render_running(frame, &model))
.map_err(terminal_error)?;
}
},
_ = input_tick.tick() => {
let mut quit_requested = false;
while event::poll(Duration::ZERO).map_err(terminal_error)? {
match event::read().map_err(terminal_error)? {
Event::Key(key) if is_quit_key(&key) => {
quit_requested = true;
break;
}
Event::Resize(_, _) => {
session
.terminal_mut()
.draw(|frame| render_running(frame, &model))
.map_err(terminal_error)?;
}
_ => {}
}
}
if quit_requested {
return Ok(());
}
}
}
}
}
.await;
let stop_result = running.stop().await;
let final_draw_result = if run_result.is_ok() && stop_result.is_ok() {
while let Ok(status) = statuses.try_recv() {
model.update_status(status);
}
while let Ok(event) = events.try_recv() {
model.push_event(event);
}
session
.terminal_mut()
.draw(|frame| render_running(frame, &model))
.map(|_| ())
.map_err(terminal_error)
} else {
Ok(())
};
preserve_primary(run_result, stop_result).and(final_draw_result)
}
pub(crate) async fn refresh_runtime_approved_client_count(
model: &mut RuntimeScreenModel,
pairing: Arc<crate::presentation::ui_handshake::UiPairingService>,
) -> AgentResult<bool> {
let approved_client_count =
tokio::task::spawn_blocking(move || pairing.approved_client_count())
.await
.map_err(|_| {
AgentError::new(
ErrorCode::PairingStorageFailed,
"approved browser count task failed",
)
})?
.map_err(|_| {
AgentError::new(
ErrorCode::PairingStorageFailed,
"approved browser count is unavailable",
)
})?;
if model.approved_client_count() == approved_client_count {
return Ok(false);
}
model.update_approved_client_count(approved_client_count);
Ok(true)
}
fn preserve_primary(primary: AgentResult<()>, cleanup: AgentResult<()>) -> AgentResult<()> {
match primary {
Err(error) => Err(error),
Ok(()) => cleanup,
}
}
fn emit_changed_runtime_status<W: Write>(
status: &RuntimeStatus,
last: &mut Option<RuntimeStatus>,
output_mode: OutputMode,
output: &mut W,
) -> AgentResult<()> {
if last.as_ref() == Some(status) {
return Ok(());
}
write_runtime_status(status, output_mode, output)?;
output
.flush()
.map_err(|error| output_error(format!("failed to flush runtime status: {error}")))?;
*last = Some(status.clone());
Ok(())
}
async fn run_existing_runtime(_paths: AppPaths, _stored: StoredConfig) -> AgentResult<()> {
let config = crate::config::Config::from_env()?;
let policy = CommandPolicy::new(
config.allowed_commands.clone(),
config.allowed_workdirs.clone(),
);
let skill_home_root = config
.home_dir
.as_deref()
.map(PathBuf::from)
.or_else(|| std::env::var_os("HOME").map(PathBuf::from))
.unwrap_or_else(|| PathBuf::from("."));
let pi_manager = Arc::new(PiRpcManager::new(
config.pi_command.clone(),
Duration::from_secs(config.pi_rpc_timeout_seconds),
));
let skill_service = SkillService::new(
Arc::new(infrastructure::skill_catalog::SkillCatalogRegistry::from_config(&config)?),
Arc::new(infrastructure::skill_store::SkillStore::new(
PathBuf::from(&config.pi_global_skills_dir),
skill_home_root.clone(),
policy.clone(),
infrastructure::skill_catalog::build_http_client()?,
)),
infrastructure::pi_skill_runtime::PiSkillRuntime::new(
config.pi_command.clone(),
Duration::from_secs(config.pi_rpc_timeout_seconds),
pi_manager.clone(),
),
skill_home_root,
);
let manager = Arc::new(Mutex::new(SessionManager::new(
policy,
config.default_command.clone(),
)));
let run = server::run_agent(config, manager.clone(), pi_manager.clone(), skill_service);
let shutdown = server::shutdown_signal();
server::run_until_shutdown(manager, pi_manager, run, shutdown).await
}