From 36ccb2c08fb4f2169b0f986fdbe5c3e2cd711b87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Sun, 19 Apr 2026 15:14:10 -0400 Subject: [PATCH] Wire native runtime flows into GPUI runs --- AgentCompany-Rust/Cargo.lock | 5 + AgentCompany-Rust/crates/app_ui/Cargo.toml | 5 + .../crates/app_ui/src/app_root.rs | 18 +- .../crates/app_ui/src/app_root/runs.rs | 37 ++ .../crates/app_ui/src/app_root/runtime.rs | 547 ++++++++++++++++++ .../crates/pi_bridge/src/client.rs | 141 +++-- AgentCompany-Rust/crates/pi_bridge/src/lib.rs | 2 +- AgentCompany-Rust/crates/storage/src/store.rs | 95 ++- .../storage/tests/runtime_live_flow_e2e.rs | 121 ++++ AgentCompany-Rust/crates/support/src/paths.rs | 3 + AgentCompany-Rust/prd.json | 4 +- AgentCompany-Rust/progress.txt | 1 + 12 files changed, 930 insertions(+), 49 deletions(-) create mode 100644 AgentCompany-Rust/crates/app_ui/src/app_root/runtime.rs create mode 100644 AgentCompany-Rust/crates/storage/tests/runtime_live_flow_e2e.rs diff --git a/AgentCompany-Rust/Cargo.lock b/AgentCompany-Rust/Cargo.lock index 6739f821..e034182e 100644 --- a/AgentCompany-Rust/Cargo.lock +++ b/AgentCompany-Rust/Cargo.lock @@ -88,11 +88,16 @@ dependencies = [ name = "app_ui" version = "0.1.0" dependencies = [ + "anyhow", "app_models", "domain", "gpui", "parking_lot", + "pi_bridge", + "runtime", + "serde_json", "storage", + "support", "unicode-segmentation", ] diff --git a/AgentCompany-Rust/crates/app_ui/Cargo.toml b/AgentCompany-Rust/crates/app_ui/Cargo.toml index e3802834..e5946314 100644 --- a/AgentCompany-Rust/crates/app_ui/Cargo.toml +++ b/AgentCompany-Rust/crates/app_ui/Cargo.toml @@ -6,9 +6,14 @@ publish.workspace = true version.workspace = true [dependencies] +anyhow.workspace = true app_models.workspace = true domain.workspace = true gpui.workspace = true parking_lot.workspace = true +pi_bridge.workspace = true +runtime.workspace = true +serde_json.workspace = true storage.workspace = true +support.workspace = true unicode-segmentation.workspace = true diff --git a/AgentCompany-Rust/crates/app_ui/src/app_root.rs b/AgentCompany-Rust/crates/app_ui/src/app_root.rs index a9ee9dce..99e8f494 100644 --- a/AgentCompany-Rust/crates/app_ui/src/app_root.rs +++ b/AgentCompany-Rust/crates/app_ui/src/app_root.rs @@ -18,6 +18,7 @@ mod messages_parts; mod onboarding; mod overview; mod runs; +mod runtime; mod settings; mod sidebar; mod workspaces; @@ -32,6 +33,9 @@ pub struct AppRoot { message_input: Entity, document_title_input: Entity, document_body_input: Entity, + runtime_steer_input: Entity, + runtime_follow_up_input: Entity, + runtime_session: Option, } impl AppRoot { @@ -46,6 +50,9 @@ impl AppRoot { message_input: TextInput::new(cx, "Board message"), document_title_input: TextInput::new(cx, "Document title"), document_body_input: TextInput::new(cx, "Document body"), + runtime_steer_input: TextInput::new(cx, "Steer the active run"), + runtime_follow_up_input: TextInput::new(cx, "Queue a follow-up after the current run"), + runtime_session: None, }; root.sync_selection(); root @@ -206,12 +213,21 @@ impl AppRoot { self.set_status("Choose an approval first.", cx); return; }; + let selected_approval = self + .snapshot + .approvals + .iter() + .find(|approval| approval.id == approval_id) + .cloned(); let result = { let mut store = self.store.lock(); store.decide_approval(&company_id, &approval_id, state, note) }; match result { - Ok(_) => self.reload(format!("Approval moved to {}.", state.as_str()), cx), + Ok(_) => { + self.maybe_dispatch_runtime_approval(selected_approval.as_ref(), state, cx); + self.reload(format!("Approval moved to {}.", state.as_str()), cx); + } Err(error) => self.set_status(error.to_string(), cx), } } diff --git a/AgentCompany-Rust/crates/app_ui/src/app_root/runs.rs b/AgentCompany-Rust/crates/app_ui/src/app_root/runs.rs index 13c385d1..d81b1d64 100644 --- a/AgentCompany-Rust/crates/app_ui/src/app_root/runs.rs +++ b/AgentCompany-Rust/crates/app_ui/src/app_root/runs.rs @@ -150,6 +150,43 @@ impl AppRoot { .text_color(theme::text_muted()) .child(run.status.clone()), ) + .child( + div() + .mt(px(18.0)) + .gap_3() + .child( + div() + .flex() + .gap_3() + .child(primary_button("Launch native runtime", cx, |this, cx| { + this.launch_selected_run(cx); + })) + .child(secondary_button("Approve selected gate", cx, |this, cx| { + this.decide_selected_approval(domain::ApprovalState::Approved, "Approved from native runtime controls.", cx); + })) + .child(secondary_button("Reject selected gate", cx, |this, cx| { + this.decide_selected_approval(domain::ApprovalState::Rejected, "Rejected from native runtime controls.", cx); + })), + ) + .child(self.runtime_steer_input.clone()) + .child( + div() + .flex() + .justify_end() + .child(primary_button("Send steer", cx, |this, cx| { + this.send_runtime_steer(cx); + })), + ) + .child(self.runtime_follow_up_input.clone()) + .child( + div() + .flex() + .justify_end() + .child(secondary_button("Queue follow-up", cx, |this, cx| { + this.send_runtime_follow_up(cx); + })), + ), + ) .child( div() .mt(px(16.0)) diff --git a/AgentCompany-Rust/crates/app_ui/src/app_root/runtime.rs b/AgentCompany-Rust/crates/app_ui/src/app_root/runtime.rs new file mode 100644 index 00000000..4a1faf54 --- /dev/null +++ b/AgentCompany-Rust/crates/app_ui/src/app_root/runtime.rs @@ -0,0 +1,547 @@ +use super::*; +use ::runtime::RuntimeConfig; +use anyhow::{Context as _, Result, anyhow}; +use pi_bridge::{ + ClientRequest, FollowUpRequest, RunBudget, RunRequest, RuntimeEvent, RuntimeHandshake, + SteerRequest, ToolApprovalDecision, +}; +use std::{ + env, + path::PathBuf, + sync::mpsc::{self, Receiver, Sender, TryRecvError}, + time::Duration, +}; +use support::paths::{RUNTIME_ARGS_ENV, RUNTIME_COMMAND_ENV, RUNTIME_WORKDIR_ENV}; + +pub(super) struct LiveRuntimeSession { + pub run_id: String, + pub commands: Sender, + pub updates: Receiver, +} + +pub(super) enum RuntimeCommand { + ToolApproval { + run_id: String, + tool_call_id: String, + approved: bool, + note: String, + }, + Steer { + run_id: String, + body: String, + }, + FollowUp { + run_id: String, + body: String, + }, +} + +pub(super) enum RuntimeUpdate { + Event(RuntimeEvent), + Failed(String), + Closed, +} + +struct RuntimeApprovalPayload { + run_id: String, + tool_call_id: String, +} + +impl AppRoot { + pub(super) fn launch_selected_run(&mut self, cx: &mut Context) { + if self.runtime_session.is_some() { + self.set_status("A native runtime session is already active.", cx); + return; + } + + let selected_run_id = match self.state.selected_run_id.clone() { + Some(run_id) => run_id, + None => { + self.set_status("Choose a run first.", cx); + return; + } + }; + + let request = match self.build_run_request(selected_run_id.as_str()) { + Ok(request) => request, + Err(error) => { + self.set_status(error.to_string(), cx); + return; + } + }; + let config = match resolve_runtime_config() { + Ok(config) => config, + Err(error) => { + self.set_status(error.to_string(), cx); + return; + } + }; + + let result = { + let mut store = self.store.lock(); + store.mark_run_started(selected_run_id.as_str()) + }; + if let Err(error) = result { + self.set_status(error.to_string(), cx); + return; + } + + let (command_tx, command_rx) = mpsc::channel(); + let (update_tx, update_rx) = mpsc::channel(); + spawn_runtime_worker(config, request, command_rx, update_tx); + self.runtime_session = Some(LiveRuntimeSession { + run_id: selected_run_id, + commands: command_tx, + updates: update_rx, + }); + self.reload("Native runtime connected.", cx); + self.start_runtime_poll_loop(cx); + } + + pub(super) fn send_runtime_steer(&mut self, cx: &mut Context) { + let Some(run_id) = self.active_runtime_run_id() else { + self.set_status("Launch a native runtime first.", cx); + return; + }; + let body = self.runtime_steer_input.read(cx).value(); + let trimmed = body.trim().to_string(); + if trimmed.is_empty() { + self.set_status("Steer message cannot be empty.", cx); + return; + } + if let Err(error) = self.send_runtime_command(RuntimeCommand::Steer { + run_id, + body: trimmed, + }) { + self.runtime_session = None; + self.set_status(error.to_string(), cx); + return; + } + self.runtime_steer_input + .update(cx, |input, _| input.clear()); + self.set_status("Steer queued into the native runtime.", cx); + } + + pub(super) fn send_runtime_follow_up(&mut self, cx: &mut Context) { + let Some(run_id) = self.active_runtime_run_id() else { + self.set_status("Launch a native runtime first.", cx); + return; + }; + let body = self.runtime_follow_up_input.read(cx).value(); + let trimmed = body.trim().to_string(); + if trimmed.is_empty() { + self.set_status("Follow-up message cannot be empty.", cx); + return; + } + if let Err(error) = self.send_runtime_command(RuntimeCommand::FollowUp { + run_id, + body: trimmed, + }) { + self.runtime_session = None; + self.set_status(error.to_string(), cx); + return; + } + self.runtime_follow_up_input + .update(cx, |input, _| input.clear()); + self.set_status("Follow-up queued into the native runtime.", cx); + } + + pub(super) fn maybe_dispatch_runtime_approval( + &mut self, + approval: Option<&domain::Approval>, + state: ApprovalState, + cx: &mut Context, + ) { + let Some(approval) = approval else { + return; + }; + if approval.approval_type != "runtime_tool_call" { + return; + } + let payload = match parse_runtime_approval_payload(approval) { + Ok(payload) => payload, + Err(error) => { + self.set_status(error.to_string(), cx); + return; + } + }; + let approved = matches!(state, ApprovalState::Approved); + let note = match state { + ApprovalState::Approved => "Approved from native review.", + ApprovalState::Rejected => "Rejected from native review.", + ApprovalState::Pending => "Pending native review.", + ApprovalState::RevisionRequested => "Revision requested from native review.", + } + .to_string(); + if let Err(error) = self.send_runtime_command(RuntimeCommand::ToolApproval { + run_id: payload.run_id, + tool_call_id: payload.tool_call_id, + approved, + note, + }) { + self.runtime_session = None; + self.set_status(error.to_string(), cx); + } + } + + fn active_runtime_run_id(&self) -> Option { + let session = self.runtime_session.as_ref()?; + let selected_matches = + self.state.selected_run_id.as_deref() == Some(session.run_id.as_str()); + if selected_matches { + Some(session.run_id.clone()) + } else { + None + } + } + + fn send_runtime_command(&self, command: RuntimeCommand) -> Result<()> { + let session = self + .runtime_session + .as_ref() + .ok_or_else(|| anyhow!("native runtime session is unavailable"))?; + session + .commands + .send(command) + .context("native runtime command channel is closed") + } + + fn start_runtime_poll_loop(&self, cx: &mut Context) { + cx.spawn( + |entity: gpui::WeakEntity, async_cx: &mut gpui::AsyncApp| { + let entity = entity.clone(); + let mut async_cx = async_cx.clone(); + async move { + loop { + async_cx + .background_executor() + .timer(Duration::from_millis(120)) + .await; + let keep_running = entity + .update(&mut async_cx, |this, cx| this.drain_runtime_updates(cx)) + .unwrap_or(false); + if !keep_running { + break; + } + } + } + }, + ) + .detach(); + } + + fn drain_runtime_updates(&mut self, cx: &mut Context) -> bool { + let mut keep_running = self.runtime_session.is_some(); + loop { + let next_update = match self.runtime_session.as_mut() { + Some(session) => match session.updates.try_recv() { + Ok(update) => Some(update), + Err(TryRecvError::Empty) => None, + Err(TryRecvError::Disconnected) => { + self.runtime_session = None; + self.reload("Native runtime disconnected.", cx); + keep_running = false; + None + } + }, + None => None, + }; + let Some(update) = next_update else { + break; + }; + self.apply_runtime_update(update, cx); + keep_running = self.runtime_session.is_some(); + } + keep_running + } + + fn apply_runtime_update(&mut self, update: RuntimeUpdate, cx: &mut Context) { + match update { + RuntimeUpdate::Event(event) => { + let selected_run_id = self.state.selected_run_id.clone(); + let result = { + let mut store = self.store.lock(); + store.ingest_runtime_event(event.clone()) + }; + match result { + Ok(()) => { + self.reload("Native runtime event applied.", cx); + match event { + RuntimeEvent::ToolCall { + run_id, + tool_call_id, + .. + } => { + self.state.selected_run_id = Some(run_id.clone()); + self.state.selected_approval_id = self + .snapshot + .approvals + .iter() + .find(|approval| { + approval.approval_type == "runtime_tool_call" + && approval + .payload_json + .as_deref() + .map(|payload| { + payload.contains(run_id.as_str()) + && payload.contains(tool_call_id.as_str()) + }) + .unwrap_or(false) + }) + .map(|approval| approval.id.clone()); + cx.notify(); + } + RuntimeEvent::RunFinished { run_id, .. } => { + self.state.selected_run_id = Some(run_id); + cx.notify(); + } + RuntimeEvent::Message { run_id, .. } + | RuntimeEvent::ToolResult { run_id, .. } + | RuntimeEvent::Artifact { run_id, .. } => { + if selected_run_id.as_deref() != Some(run_id.as_str()) { + self.state.selected_run_id = Some(run_id); + cx.notify(); + } + } + RuntimeEvent::HandshakeAck { .. } => {} + } + } + Err(error) => { + self.runtime_session = None; + self.set_status(error.to_string(), cx); + } + } + } + RuntimeUpdate::Failed(message) => { + self.runtime_session = None; + self.set_status(message, cx); + } + RuntimeUpdate::Closed => { + self.runtime_session = None; + self.set_status("Native runtime session closed.", cx); + } + } + } + + fn build_run_request(&self, run_id: &str) -> Result { + let run = self + .snapshot + .runs + .iter() + .find(|run| run.id == run_id) + .cloned() + .ok_or_else(|| anyhow!("run {run_id} was not found in the active snapshot"))?; + let task = run + .task_id + .as_ref() + .and_then(|task_id| self.snapshot.tasks.iter().find(|task| &task.id == task_id)) + .cloned() + .ok_or_else(|| anyhow!("selected run must be linked to a task before native launch"))?; + let company_id = self + .current_company_id() + .map(str::to_string) + .ok_or_else(|| anyhow!("create a company first"))?; + let workspace_path = run.workspace_id.as_ref().and_then(|workspace_id| { + self.snapshot + .workspaces + .iter() + .find(|workspace| &workspace.id == workspace_id) + .map(|workspace| workspace.local_path.clone()) + }); + let prompt = if task.description.trim().is_empty() { + task.title.clone() + } else { + format!("{}\n\n{}", task.title, task.description) + }; + Ok(RunRequest { + run_id: run.id.clone(), + session_id: Some(run.id.clone()), + company_id, + task_id: Some(task.id.clone()), + agent_id: run.agent_id.clone(), + workspace_path, + connector_id: run.connector_id.clone(), + model: run.model.clone(), + prompt, + context_json: serde_json::json!({ + "task_title": task.title, + "task_description": task.description, + "source": "native_gpui", + }), + budget: RunBudget { + thinking_level: Some("high".to_string()), + token_budget: None, + }, + }) + } +} + +fn resolve_runtime_config() -> Result { + let command = env::var(RUNTIME_COMMAND_ENV) + .ok() + .filter(|value| !value.trim().is_empty()) + .map(PathBuf::from) + .ok_or_else(|| { + anyhow!("set {RUNTIME_COMMAND_ENV} to the Pi Mono sidecar executable path") + })?; + let args = match env::var(RUNTIME_ARGS_ENV) { + Ok(value) if !value.trim().is_empty() => serde_json::from_str::>(&value) + .with_context(|| format!("{RUNTIME_ARGS_ENV} must be a JSON string array"))?, + _ => Vec::new(), + }; + let working_dir = env::var(RUNTIME_WORKDIR_ENV) + .ok() + .filter(|value| !value.trim().is_empty()) + .map(PathBuf::from); + Ok(RuntimeConfig { + command, + args, + working_dir, + }) +} + +fn parse_runtime_approval_payload(approval: &domain::Approval) -> Result { + let payload_json = approval + .payload_json + .as_deref() + .ok_or_else(|| anyhow!("runtime approval payload is missing"))?; + let payload = serde_json::from_str::(payload_json) + .context("failed to decode runtime approval payload")?; + Ok(RuntimeApprovalPayload { + run_id: payload + .get("run_id") + .and_then(serde_json::Value::as_str) + .map(str::to_string) + .ok_or_else(|| anyhow!("runtime approval payload is missing run_id"))?, + tool_call_id: payload + .get("tool_call_id") + .and_then(serde_json::Value::as_str) + .map(str::to_string) + .ok_or_else(|| anyhow!("runtime approval payload is missing tool_call_id"))?, + }) +} + +fn spawn_runtime_worker( + config: RuntimeConfig, + request: RunRequest, + command_rx: Receiver, + update_tx: Sender, +) { + std::thread::spawn(move || { + if let Err(error) = run_runtime_worker(config, request, command_rx, &update_tx) { + let _ = update_tx.send(RuntimeUpdate::Failed(error.to_string())); + } + let _ = update_tx.send(RuntimeUpdate::Closed); + }); +} + +fn run_runtime_worker( + config: RuntimeConfig, + request: RunRequest, + command_rx: Receiver, + update_tx: &Sender, +) -> Result<()> { + let run_id = request.run_id.clone(); + let (mut writer, mut events) = + pi_bridge::PiBridgeClient::spawn_split(config.command, &config.args, config.working_dir)?; + writer.send(&ClientRequest::Handshake(RuntimeHandshake { + protocol_version: 1, + app_name: "AgentCompany-Rust".to_string(), + app_version: env!("CARGO_PKG_VERSION").to_string(), + }))?; + update_tx + .send(RuntimeUpdate::Event(events.read_event()?)) + .context("failed to deliver runtime handshake event")?; + writer.send(&ClientRequest::StartRun(Box::new(request)))?; + + loop { + let event = events.read_event()?; + let pending_tool_call = match &event { + RuntimeEvent::ToolCall { tool_call_id, .. } => Some(tool_call_id.clone()), + _ => None, + }; + let run_finished = matches!(event, RuntimeEvent::RunFinished { .. }); + update_tx + .send(RuntimeUpdate::Event(event)) + .context("failed to deliver runtime event")?; + + if let Some(expected_tool_call_id) = pending_tool_call { + loop { + match command_rx + .recv() + .context("runtime command channel closed")? + { + RuntimeCommand::ToolApproval { + run_id: command_run_id, + tool_call_id, + approved, + note, + } if command_run_id == run_id && tool_call_id == expected_tool_call_id => { + writer.send(&ClientRequest::ToolApproval(ToolApprovalDecision { + run_id: command_run_id, + tool_call_id, + approved, + note, + }))?; + break; + } + RuntimeCommand::Steer { + run_id: command_run_id, + body, + } if command_run_id == run_id => { + writer.send(&ClientRequest::Steer(SteerRequest { + run_id: command_run_id, + body, + }))?; + } + RuntimeCommand::FollowUp { + run_id: command_run_id, + body, + } if command_run_id == run_id => { + writer.send(&ClientRequest::FollowUp(FollowUpRequest { + run_id: command_run_id, + body, + }))?; + } + _ => {} + } + } + } + + if run_finished { + loop { + match command_rx + .recv() + .context("runtime command channel closed")? + { + RuntimeCommand::Steer { + run_id: command_run_id, + body, + } if command_run_id == run_id => { + writer.send(&ClientRequest::Steer(SteerRequest { + run_id: command_run_id, + body, + }))?; + update_tx + .send(RuntimeUpdate::Event(events.read_event()?)) + .context("failed to deliver steer runtime event")?; + } + RuntimeCommand::FollowUp { + run_id: command_run_id, + body, + } if command_run_id == run_id => { + writer.send(&ClientRequest::FollowUp(FollowUpRequest { + run_id: command_run_id, + body, + }))?; + update_tx + .send(RuntimeUpdate::Event(events.read_event()?)) + .context("failed to deliver follow-up runtime event")?; + } + RuntimeCommand::ToolApproval { .. } => {} + _ => {} + } + } + } + } +} diff --git a/AgentCompany-Rust/crates/pi_bridge/src/client.rs b/AgentCompany-Rust/crates/pi_bridge/src/client.rs index 32369621..66255d9b 100644 --- a/AgentCompany-Rust/crates/pi_bridge/src/client.rs +++ b/AgentCompany-Rust/crates/pi_bridge/src/client.rs @@ -6,6 +6,15 @@ use std::{ process::{Child, ChildStdin, ChildStdout, Command, Stdio}, }; +pub struct PiBridgeWriter { + stdin: ChildStdin, +} + +pub struct PiBridgeEventStream { + child: Child, + stdout: BufReader, +} + pub struct PiBridgeClient { child: Child, stdin: ChildStdin, @@ -14,60 +23,32 @@ pub struct PiBridgeClient { impl PiBridgeClient { pub fn spawn(command: PathBuf, args: &[String], working_dir: Option) -> Result { - let mut process = Command::new(&command); - process - .args(args) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::inherit()); - - if let Some(dir) = working_dir { - process.current_dir(dir); - } - - let mut child = process - .spawn() - .with_context(|| format!("failed to spawn Pi Mono sidecar at {}", command.display()))?; - let stdin = child - .stdin - .take() - .context("Pi Mono sidecar stdin is unavailable")?; - let stdout = child - .stdout - .take() - .context("Pi Mono sidecar stdout is unavailable")?; + let (child, stdin, stdout) = spawn_process(command, args, working_dir)?; Ok(Self { child, stdin, - stdout: BufReader::new(stdout), + stdout, }) } + pub fn spawn_split( + command: PathBuf, + args: &[String], + working_dir: Option, + ) -> Result<(PiBridgeWriter, PiBridgeEventStream)> { + let (child, stdin, stdout) = spawn_process(command, args, working_dir)?; + Ok(( + PiBridgeWriter { stdin }, + PiBridgeEventStream { child, stdout }, + )) + } + pub fn send(&mut self, message: &ClientRequest) -> Result<()> { - let payload = - serde_json::to_vec(message).context("failed to serialize Pi bridge request")?; - self.stdin - .write_all(&payload) - .context("failed to write Pi bridge request")?; - self.stdin - .write_all(b"\n") - .context("failed to terminate Pi bridge request")?; - self.stdin - .flush() - .context("failed to flush Pi bridge request")?; - Ok(()) + write_request(&mut self.stdin, message) } pub fn read_event(&mut self) -> Result { - let mut line = String::new(); - let bytes = self - .stdout - .read_line(&mut line) - .context("failed to read Pi bridge event")?; - if bytes == 0 { - return Err(anyhow::anyhow!("Pi Mono sidecar closed stdout")); - } - serde_json::from_str(line.trim()).context("failed to decode Pi bridge event") + read_event(&mut self.stdout) } pub fn wait(&mut self) -> Result<()> { @@ -77,3 +58,75 @@ impl PiBridgeClient { Ok(()) } } + +impl PiBridgeWriter { + pub fn send(&mut self, message: &ClientRequest) -> Result<()> { + write_request(&mut self.stdin, message) + } +} + +impl PiBridgeEventStream { + pub fn read_event(&mut self) -> Result { + read_event(&mut self.stdout) + } + + pub fn wait(&mut self) -> Result<()> { + self.child + .wait() + .context("failed while waiting for Pi Mono sidecar")?; + Ok(()) + } +} + +fn spawn_process( + command: PathBuf, + args: &[String], + working_dir: Option, +) -> Result<(Child, ChildStdin, BufReader)> { + let mut process = Command::new(&command); + process + .args(args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()); + + if let Some(dir) = working_dir { + process.current_dir(dir); + } + + let mut child = process + .spawn() + .with_context(|| format!("failed to spawn Pi Mono sidecar at {}", command.display()))?; + let stdin = child + .stdin + .take() + .context("Pi Mono sidecar stdin is unavailable")?; + let stdout = child + .stdout + .take() + .context("Pi Mono sidecar stdout is unavailable")?; + Ok((child, stdin, BufReader::new(stdout))) +} + +fn write_request(stdin: &mut ChildStdin, message: &ClientRequest) -> Result<()> { + let payload = serde_json::to_vec(message).context("failed to serialize Pi bridge request")?; + stdin + .write_all(&payload) + .context("failed to write Pi bridge request")?; + stdin + .write_all(b"\n") + .context("failed to terminate Pi bridge request")?; + stdin.flush().context("failed to flush Pi bridge request")?; + Ok(()) +} + +fn read_event(stdout: &mut BufReader) -> Result { + let mut line = String::new(); + let bytes = stdout + .read_line(&mut line) + .context("failed to read Pi bridge event")?; + if bytes == 0 { + return Err(anyhow::anyhow!("Pi Mono sidecar closed stdout")); + } + serde_json::from_str(line.trim()).context("failed to decode Pi bridge event") +} diff --git a/AgentCompany-Rust/crates/pi_bridge/src/lib.rs b/AgentCompany-Rust/crates/pi_bridge/src/lib.rs index f1931788..e39246d6 100644 --- a/AgentCompany-Rust/crates/pi_bridge/src/lib.rs +++ b/AgentCompany-Rust/crates/pi_bridge/src/lib.rs @@ -1,5 +1,5 @@ pub mod client; pub mod protocol; -pub use client::PiBridgeClient; +pub use client::{PiBridgeClient, PiBridgeEventStream, PiBridgeWriter}; pub use protocol::*; diff --git a/AgentCompany-Rust/crates/storage/src/store.rs b/AgentCompany-Rust/crates/storage/src/store.rs index 1413ecf7..572d553b 100644 --- a/AgentCompany-Rust/crates/storage/src/store.rs +++ b/AgentCompany-Rust/crates/storage/src/store.rs @@ -271,13 +271,77 @@ impl ProfileStore { Ok(artifact_id) } + pub fn mark_run_started(&mut self, run_id: &str) -> Result<()> { + let (company_id, task_id) = self.run_context(run_id)?; + let ts = now_iso()?; + self.conn.execute( + "update runs set status = 'running', started_at = coalesce(started_at, ?1), finished_at = null, error_message = null, updated_at = ?1 where id = ?2", + params![ts, run_id], + )?; + if let Some(task_id) = task_id { + self.conn.execute( + "update tasks set active_run_id = ?1, status = case when status = 'todo' then 'running' else status end, started_at = coalesce(started_at, ?2), updated_at = ?2 where id = ?3", + params![run_id, ts, task_id], + )?; + } + self.add_activity( + &company_id, + BOARD_READER_ID, + "started_run", + "runs", + run_id, + "native runtime launch", + )?; + Ok(()) + } + + pub fn create_runtime_tool_approval( + &mut self, + run_id: &str, + tool_call_id: &str, + tool_name: &str, + input_json: &serde_json::Value, + ) -> Result { + let (company_id, task_id) = self.run_context(run_id)?; + let agent_id = self.agent_id_for_run(run_id)?; + let approval_id = Uuid::new_v4().to_string(); + let ts = now_iso()?; + let payload_json = serde_json::json!({ + "run_id": run_id, + "tool_call_id": tool_call_id, + "tool_name": tool_name, + "input_json": input_json, + }) + .to_string(); + let impact_summary = serde_json::to_string(input_json) + .context("failed to encode runtime approval payload")?; + self.conn.execute( + "insert into approvals (id, company_id, related_task_id, requested_by_agent_id, related_agent_id, type, payload_summary, impact_summary, payload_json, decision_note, state, created_at, updated_at) values (?1, ?2, ?3, ?4, ?4, 'runtime_tool_call', ?5, ?6, ?7, '', 'pending', ?8, ?8)", + params![approval_id, company_id, task_id, agent_id, tool_name.trim(), impact_summary, payload_json, ts], + )?; + self.add_activity( + &company_id, + agent_id.as_str(), + "requested_runtime_tool_approval", + "approvals", + &approval_id, + tool_name, + )?; + Ok(approval_id) + } + pub fn ingest_runtime_event(&mut self, event: RuntimeEvent) -> Result<()> { match event { RuntimeEvent::HandshakeAck { .. } => Ok(()), RuntimeEvent::Message { run_id, role, body } => { let company_id = self.company_id_for_run(&run_id)?; + let from_agent_id = self.agent_id_for_run(&run_id)?; let title = format!("message:{role}"); self.append_run_event(&company_id, &run_id, "message", &title, &body)?; + self.conn.execute( + "insert into agent_messages (id, company_id, from_agent_id, to_agent_id, sender_kind, channel, channel_target_id, subject, body, priority, read_at, parent_message_id, attachments_json, created_at) values (?1, ?2, ?3, null, 'agent', 'run', ?4, ?5, ?6, 'normal', null, null, '[]', ?7)", + params![Uuid::new_v4().to_string(), company_id, from_agent_id, run_id, title, body, now_iso()?], + )?; Ok(()) } RuntimeEvent::ToolCall { @@ -291,6 +355,7 @@ impl ProfileStore { .context("failed to encode tool call payload")?; let title = format!("{tool_name}:{tool_call_id}"); self.append_run_event(&company_id, &run_id, "tool_call", &title, &detail)?; + self.create_runtime_tool_approval(&run_id, &tool_call_id, &tool_name, &input_json)?; Ok(()) } RuntimeEvent::ToolResult { @@ -331,12 +396,18 @@ impl ProfileStore { session_id: _, cost_usd, } => { - let company_id = self.company_id_for_run(&run_id)?; + let (company_id, task_id) = self.run_context(&run_id)?; let ts = now_iso()?; self.conn.execute( "update runs set status = ?1, summary = ?2, cost_usd = ?3, finished_at = ?4, updated_at = ?4 where id = ?5", params![status, summary, cost_usd, ts, run_id], )?; + if let Some(task_id) = task_id { + self.conn.execute( + "update tasks set active_run_id = null, updated_at = ?1 where id = ?2", + params![ts, task_id], + )?; + } self.append_run_event( &company_id, &run_id, @@ -918,6 +989,28 @@ impl ProfileStore { .optional()? .ok_or_else(|| anyhow!("run {run_id} was not found")) } + + fn agent_id_for_run(&self, run_id: &str) -> Result { + self.conn + .query_row( + "select agent_id from runs where id = ?1", + params![run_id], + |row| row.get(0), + ) + .optional()? + .ok_or_else(|| anyhow!("run {run_id} was not found")) + } + + fn run_context(&self, run_id: &str) -> Result<(String, Option)> { + self.conn + .query_row( + "select company_id, task_id from runs where id = ?1", + params![run_id], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional()? + .ok_or_else(|| anyhow!("run {run_id} was not found")) + } } fn query_many(conn: &Connection, sql: &str, params: P, mut mapper: F) -> Result> diff --git a/AgentCompany-Rust/crates/storage/tests/runtime_live_flow_e2e.rs b/AgentCompany-Rust/crates/storage/tests/runtime_live_flow_e2e.rs new file mode 100644 index 00000000..4f049ac0 --- /dev/null +++ b/AgentCompany-Rust/crates/storage/tests/runtime_live_flow_e2e.rs @@ -0,0 +1,121 @@ +use pi_bridge::RuntimeEvent; +use storage::{NewCompanyInput, ProfileStore}; +use tempfile::tempdir; + +#[test] +fn live_runtime_events_drive_approvals_messages_and_run_state() { + let sandbox = tempdir().expect("temp dir"); + let mut store = ProfileStore::open(sandbox.path()).expect("open store"); + + let company_id = store + .create_company(NewCompanyInput { + name: "Orbit Live Runtime".to_string(), + description: "Verifies live runtime wiring".to_string(), + workspace_path: "/tmp/orbit-live-runtime".to_string(), + execution_mode: "ai_ceo".to_string(), + }) + .expect("create company"); + + let ts = support::time::now_iso().expect("timestamp"); + let task_id = uuid::Uuid::new_v4().to_string(); + let run_id = uuid::Uuid::new_v4().to_string(); + let agent_id = uuid::Uuid::new_v4().to_string(); + let workspace_id = uuid::Uuid::new_v4().to_string(); + + let conn = rusqlite::Connection::open(store.db_path()).expect("open raw db"); + conn.execute( + "insert into workspaces (id, company_id, name, local_path, repo_url, repo_ref, project_id, state, is_primary, created_at, updated_at) values (?1, ?2, 'Primary Workspace', '/tmp/orbit-live-runtime', '', '', null, 'ready', 1, ?3, ?3)", + rusqlite::params![workspace_id, company_id, ts], + ) + .expect("insert workspace"); + conn.execute( + "insert into agents (id, company_id, name, role, title, department, status, reports_to, connector_id, workspace_id, model, capabilities, budget_monthly_usd, spent_monthly_usd, created_at, updated_at) values (?1, ?2, 'Alex Runtime', 'engineer', 'Runtime Engineer', null, 'idle', null, 'anthropic', ?3, 'claude-sonnet', '[]', 0, 0, ?4, ?4)", + rusqlite::params![agent_id, company_id, workspace_id, ts], + ) + .expect("insert agent"); + conn.execute( + "insert into tasks (id, company_id, project_id, goal_id, parent_id, title, description, assignee_agent_id, workspace_id, status, priority, active_run_id, cost_usd, task_type, requires_user_review, started_at, completed_at, cancelled_at, created_at, updated_at) values (?1, ?2, null, null, null, 'Ship runtime rewrite', 'Keep the native app in sync with Pi Mono.', ?3, ?4, 'todo', 'high', null, 0, 'execution', 1, null, null, null, ?5, ?5)", + rusqlite::params![task_id, company_id, agent_id, workspace_id, ts], + ) + .expect("insert task"); + conn.execute( + "insert into runs (id, company_id, task_id, agent_id, workspace_id, connector_id, status, summary, error_message, model, cost_usd, started_at, finished_at, created_at, updated_at) values (?1, ?2, ?3, ?4, ?5, 'anthropic', 'queued', 'Queued from native UI', null, 'claude-sonnet', null, null, null, ?6, ?6)", + rusqlite::params![run_id, company_id, task_id, agent_id, workspace_id, ts], + ) + .expect("insert run"); + + store.mark_run_started(&run_id).expect("mark run started"); + store + .ingest_runtime_event(RuntimeEvent::ToolCall { + run_id: run_id.clone(), + tool_call_id: "tool-1".to_string(), + tool_name: "write_file".to_string(), + input_json: serde_json::json!({"path": "README.md"}), + }) + .expect("ingest tool call"); + store + .ingest_runtime_event(RuntimeEvent::Message { + run_id: run_id.clone(), + role: "assistant".to_string(), + body: "Draft ready for review".to_string(), + }) + .expect("ingest message"); + store + .ingest_runtime_event(RuntimeEvent::RunFinished { + run_id: run_id.clone(), + status: "completed".to_string(), + summary: "Native runtime finished cleanly".to_string(), + session_id: Some(run_id.clone()), + cost_usd: Some(2.5), + }) + .expect("ingest finished event"); + + let snapshot = store.snapshot().expect("snapshot"); + let task = snapshot + .tasks + .iter() + .find(|task| task.id == task_id) + .expect("task present"); + assert_eq!(task.active_run_id, None); + + let approval = snapshot + .approvals + .iter() + .find(|approval| approval.approval_type == "runtime_tool_call") + .expect("runtime approval present"); + assert_eq!(approval.related_task_id.as_deref(), Some(task_id.as_str())); + assert_eq!( + approval.related_agent_id.as_deref(), + Some(agent_id.as_str()) + ); + assert_eq!(approval.payload_summary, "write_file"); + assert!( + approval + .payload_json + .as_deref() + .expect("payload json") + .contains("tool-1") + ); + + let runtime_message = snapshot + .agent_messages + .iter() + .find(|message| { + message.channel == "run" + && message.channel_target_id.as_deref() == Some(run_id.as_str()) + }) + .expect("runtime message projected"); + assert_eq!(runtime_message.from_agent_id, agent_id); + assert_eq!(runtime_message.body, "Draft ready for review"); + + let run = snapshot + .runs + .iter() + .find(|run| run.id == run_id) + .expect("run present"); + assert_eq!(run.status, "completed"); + assert_eq!(run.summary, "Native runtime finished cleanly"); + assert_eq!(run.cost_usd, Some(2.5)); + assert!(run.started_at.is_some()); + assert!(run.finished_at.is_some()); +} diff --git a/AgentCompany-Rust/crates/support/src/paths.rs b/AgentCompany-Rust/crates/support/src/paths.rs index 80f15a13..98b1d06a 100644 --- a/AgentCompany-Rust/crates/support/src/paths.rs +++ b/AgentCompany-Rust/crates/support/src/paths.rs @@ -6,6 +6,9 @@ use std::{ pub const APP_NAME: &str = "AgentCompany"; pub const PROFILE_ENV: &str = "AGENTCOMPANY_PROFILE_DIR"; +pub const RUNTIME_COMMAND_ENV: &str = "AGENTCOMPANY_PI_COMMAND"; +pub const RUNTIME_ARGS_ENV: &str = "AGENTCOMPANY_PI_ARGS_JSON"; +pub const RUNTIME_WORKDIR_ENV: &str = "AGENTCOMPANY_PI_WORKDIR"; pub fn default_profile_dir() -> Result { dirs::data_dir() diff --git a/AgentCompany-Rust/prd.json b/AgentCompany-Rust/prd.json index 01c53c8b..47684aa4 100644 --- a/AgentCompany-Rust/prd.json +++ b/AgentCompany-Rust/prd.json @@ -113,8 +113,8 @@ "cargo test --workspace passes" ], "priority": 8, - "passes": false, - "notes": "In progress: runtime events and artifacts now persist into SQLite, surface in ProfileSnapshot, render in Runs UI, and pass storage-backed e2e verification." + "passes": true, + "notes": "Completed with live native runtime session wiring: selected runs now launch Pi Mono via env-configured sidecar, tool-call approvals round-trip through native approvals, runtime messages project into agent_messages, and artifacts/events continue to persist into Runs UI. Verified by cargo clippy/test and runtime_live_flow_e2e.rs." }, { "id": "US-009", diff --git a/AgentCompany-Rust/progress.txt b/AgentCompany-Rust/progress.txt index 969eb016..172ef545 100644 --- a/AgentCompany-Rust/progress.txt +++ b/AgentCompany-Rust/progress.txt @@ -13,3 +13,4 @@ Started: 2026-04-19 - Added runtime contract persistence entities for run events and artifacts, exposed them in ProfileSnapshot, and rendered them in the native runs inspector. - Added runtime_contract_snapshot_e2e.rs to verify runtime event and artifact persistence across reopen. - Runtime contract integration advanced: RuntimeEvent ingestion now persists message/tool/artifact/run-finished events into storage and exposes them in the runs inspector. +- Wired a live native runtime session into the GPUI Runs surface. Selected runs now launch Pi Mono through env-configured stdio sidecar settings, runtime tool calls create approval rows, approval decisions send ToolApproval back into the live session, runtime messages project into agent_messages, and run lifecycle still persists into run_events/artifacts/runs.\n \ No newline at end of file