From fb8b7d910c46b4b10fe758caddf6c9099d6271af Mon Sep 17 00:00:00 2001 From: limityan Date: Fri, 3 Jul 2026 11:03:27 +0800 Subject: [PATCH] refactor(core): route remote exec through runtime port --- docs/plans/core-decomposition-completed.md | 4 +- docs/plans/core-decomposition-plan.md | 4 +- .../rules/source/forbidden-rules.mjs | 50 +++ scripts/core-boundaries/self-test.mjs | 10 + src/apps/cli/src/agent/agentic_system.rs | 3 + src/apps/desktop/src/api/agentic_api.rs | 12 + src/apps/desktop/src/api/tool_api.rs | 12 +- src/apps/desktop/src/lib.rs | 3 + src/apps/server/src/bootstrap.rs | 3 + .../src/agentic/coordination/coordinator.rs | 24 +- .../src/agentic/execution/execution_engine.rs | 3 + .../src/agentic/execution/round_executor.rs | 2 + .../core/src/agentic/execution/types.rs | 5 +- .../implementations/exec_command/command.rs | 155 +++++--- .../exec_command/completion.rs | 9 +- .../implementations/exec_command/control.rs | 138 ++++++- .../exec_command/env_snapshot.rs | 24 +- .../implementations/exec_command/input.rs | 17 +- .../implementations/exec_command/stdin.rs | 23 +- .../agentic/tools/pipeline/state_manager.rs | 1 + .../agentic/tools/pipeline/tool_pipeline.rs | 1 + .../core/src/agentic/tools/pipeline/types.rs | 3 +- .../src/agentic/tools/tool_context_runtime.rs | 27 +- .../src/product_runtime/runtime_services.rs | 43 +++ src/crates/contracts/runtime-ports/AGENTS.md | 3 + src/crates/contracts/runtime-ports/src/lib.rs | 177 ++++++++- .../execution/runtime-services/src/lib.rs | 22 +- .../runtime-services/src/test_support.rs | 80 ++++- .../tests/runtime_services_contracts.rs | 2 + .../services/services-integrations/AGENTS.md | 4 +- .../services/services-integrations/Cargo.toml | 1 + .../src/remote_ssh/mod.rs | 4 + .../remote_ssh/remote_exec_runtime_port.rs | 339 ++++++++++++++++++ 33 files changed, 1099 insertions(+), 109 deletions(-) create mode 100644 src/crates/services/services-integrations/src/remote_ssh/remote_exec_runtime_port.rs diff --git a/docs/plans/core-decomposition-completed.md b/docs/plans/core-decomposition-completed.md index 4606314a55..6bfa33727b 100644 --- a/docs/plans/core-decomposition-completed.md +++ b/docs/plans/core-decomposition-completed.md @@ -21,8 +21,8 @@ - `bitfun-events` 已承接 backend event DTO、agentic event DTO、framework-neutral Agentic frontend event projection 和 platform-neutral `EventEmitter` trait；Tauri/WebSocket transport 只负责 delivery。 - `services-integrations` 已承接 remote-connect primitives、wire command routing / response assembly、LAN IP/URL 探测、ngrok 进程/tunnel lifecycle、mobile-web relay upload manifest / incremental upload / fallback upload、IM bot provider-neutral config / persistence / file auto-push / locale / menu / state / command parsing、Weixin provider client、workspace search concrete owner、remote SSH/SFTP/PTY owner、Remote SSH disabled runtime surface、Remote SSH workspace/session identity helper、remote workspace-search disabled surface、DeepResearch report IO / display-map sidecar、MiniApp host dispatch / storage / worker / import IO、announcement remote fetch/cache、browser CDP endpoint HTTP probing / page creation、WebFetch / WebSearch concrete HTTP provider、debug-log HTTP ingest posting、review-platform HTTP transport，以及 MCP server registry、connection pool、catalog cache、reconnect retry state、runtime-only config overlay、local command resolution helper 和 lifecycle status policy。 - `tool-contracts` 已承接 provider-neutral tool DTO、manifest/catalog/admission/result presentation、Computer Use DTO/input parser/screenshot payload、confirmation facts、truncation recovery presentation、runtime restriction policy 和 provider-entry materialization；core 只保留 Computer Use 旧 public path re-export / compatibility shim 与产品执行入口。 -- `tool-execution` 已承接 local / remote IO helper、Bash shell helper、batching plan、retry policy、state counting、tool state event payload shaping / result redaction、cancellation-state/token-store policy、background exec output capture、ExecCommand provider-neutral 呈现 / 输入默认值 / 结果 shape / shell metadata / shell argv / remote shell probe / remote env snapshot 解析、cache 与 capture policy / lifecycle facts / control facts / completion shape、prompt-safe tool context facts / custom-data materialization、Computer Use loop detection / screenshot hash / verification / retry policy，以及 File tool 的 provider-neutral 结果展示、写入 mode/status/line-count 规则、Edit guardrail 分类和 Delete success 文本；core 只保留 ToolResult 包装、权限、checkpoint、runtime handles、process manager / host adapter 调用、read-state adapter、remote shell/FS 调用和旧工具入口。 -- `runtime-ports` / `terminal-core` 已承接本地 ExecCommand 会话执行端口和 concrete provider：`TerminalPort` 暴露本地命令执行、stdin 写入、会话控制和生命周期事件边界，`TerminalRuntimePort` 复用原 `ExecProcessManager` 行为，desktop/server/CLI 等产品入口通过 `CoreRuntimeServicesProvider` 构造 terminal provider 并显式注入 `ConversationCoordinator` / 执行上下文 / `ToolRuntimeHandles`；core 本地 `ExecCommand` / `WriteStdin` / `ExecControl` 只消费端口，不再直接调用全局本地进程 manager。远端 SSH ExecCommand 仍是单独 remote/product command adapter 待迁移项。 +- `tool-execution` 已承接 local / remote IO helper、Bash shell helper、batching plan、retry policy、state counting、tool state event payload shaping / result redaction、cancellation-state/token-store policy、background exec output capture、ExecCommand provider-neutral 呈现 / 输入默认值 / 结果 shape / shell metadata / shell argv / remote shell probe / remote env snapshot 解析、cache 与 capture policy / lifecycle facts / control facts / completion shape、prompt-safe tool context facts / custom-data materialization、Computer Use loop detection / screenshot hash / verification / retry policy，以及 File tool 的 provider-neutral 结果展示、写入 mode/status/line-count 规则、Edit guardrail 分类和 Delete success 文本；core 只保留 ToolResult 包装、权限、checkpoint、runtime handles、host adapter 调用、read-state adapter、remote FS 调用和旧工具入口。 +- `runtime-ports` / `terminal-core` / `services-integrations` 已承接 ExecCommand 会话执行端口和 concrete provider：`TerminalPort` 暴露本地命令执行、stdin 写入、会话控制和生命周期事件边界，`RemoteExecPort` 暴露远端 SSH 命令执行、bounded one-shot command、stdin、会话控制和生命周期事件边界；`TerminalRuntimePort` 复用原本地 `ExecProcessManager` 行为，`RemoteExecRuntimePort` 复用原 remote exec manager 与旧 SSH one-shot 行为，当前 desktop / CLI 产品入口和保留 server bootstrap 初始化路径通过 `CoreRuntimeServicesProvider` 构造 provider 并显式注入 `ConversationCoordinator` / 执行上下文 / `ToolRuntimeHandles`；core `ExecCommand` / `WriteStdin` / `ExecControl` 只消费端口，不再直接调用全局本地或远端进程 manager。 - `agent-runtime` 已承接 scheduler/background delivery 纯决策、dialog lifecycle port contracts、runtime event queue/router、session management/cancellation port contracts、session/config/summary facts、persisted session state sidecar / processing-state sanitization、session state facts / event-label projection、session state manager / event emission owner、dialog-turn id / stats facts、side-question runtime-only tracking、thread-goal facts、context profile / model capability policy、prompt markup / prompt / prompt-cache facts 与持久化写入决策、remote file delivery prompt facts、turn skill/agent snapshot DTO/diff/render/store、file-read session state / prior-read guardrail / freshness 决策、session evidence ledger 与 compression-contract projection、dialog-turn cancellation token store、tool confirmation gate / wait channel state、user-question wait channel state、custom agent / mode / subagent schema、默认值、discovery/loading、markdown IO、validation、review 工具过滤、skill catalog/root specs、mode policy、selection/shadow/mode-info 规则、assistant payload rendering、post-call hook routing、DeepReview provider-neutral policy/queue/retry/diagnostics shaping 与 queue event payload shaping、DeepResearch citation renumber 与 report post-process gate，并建立不暴露 `bitfun-core` / `product-full` / concrete manager 的内部 SDK facade。SDK facade 已支持注入 fake runtime services、tool registry、harness registry、hook registry 和 agent registry。 - `harness` 已建立 descriptor、route plan 和 legacy provider registry。 - `product-domains` 已承接 MiniApp state/workflow planning、built-in seed orchestration / host adapter contract、compile / permission adaptation、import lifecycle、AI / Agent permission、rate-limit、model/message/session/workspace/turn-text bridge rules、AI / Agent 请求计划、stream / runtime event payload、worker restart / draft key / workspace input 规则、function-agent prompt/parser/response policy 和部分 Git snapshot/fallback 逻辑。 diff --git a/docs/plans/core-decomposition-plan.md b/docs/plans/core-decomposition-plan.md index 80e74e4a4a..cea147698b 100644 --- a/docs/plans/core-decomposition-plan.md +++ b/docs/plans/core-decomposition-plan.md @@ -87,8 +87,8 @@ 当前大型 PR 批次： - Platform Provider Closure 已继续收口具备服务层 owner 的 concrete provider：remote-connect LAN IP/URL 探测、ngrok 进程/tunnel lifecycle、mobile-web relay 上传、announcement remote fetch/cache、browser CDP endpoint HTTP probing / page creation、WebFetch / WebSearch concrete HTTP provider、debug-log HTTP ingest posting 和 review-platform HTTP transport。core 保留兼容 facade、配置读取、产品编排、provider DTO 映射和工具结果 envelope。 -- Terminal / ExecCommand 路径已迁出 provider-neutral 输入默认值、结果 shape、shell metadata、shell argv、PowerShell UTF-8 policy、remote shell probe、remote env snapshot 解析/cache/capture policy、remote non-TTY control wrapper 和 lifecycle status facts；本地 concrete execution 已通过 `TerminalPort` / `TerminalRuntimePort` 收口到 desktop/server/CLI 等产品入口显式注入的 terminal provider，并由执行上下文传递到工具 runtime handles。core 本地 `ExecCommand` / `WriteStdin` / `ExecControl` 不再直接调用全局本地进程 manager，只保留工具入口、权限/checkpoint 和结果 envelope。 -- 后续仍需迁移 remote ExecCommand/product 命令路径、Computer Use OS action、部分 Git/process/session host adapter、MCP auth URL helper 和 OpenCode/UI extension 的真实消费路径；这些路径必须在具备等价测试、host port 和交付形态矩阵后再迁移。 +- Terminal / ExecCommand 路径已迁出 provider-neutral 输入默认值、结果 shape、shell metadata、shell argv、PowerShell UTF-8 policy、remote shell probe、remote env snapshot 解析/cache/capture policy、remote non-TTY control wrapper 和 lifecycle status facts；本地 concrete execution 已通过 `TerminalPort` / `TerminalRuntimePort` 收口到当前 desktop / CLI 产品入口显式注入的 terminal provider，远端 SSH concrete execution 已通过 `RemoteExecPort` / `RemoteExecRuntimePort` 收口到 services-integrations remote owner，bounded shell probe 通过端口保留旧 SSH one-shot stdout/stderr/timeout 语义，并由执行上下文传递到工具 runtime handles。core `ExecCommand` / `WriteStdin` / `ExecControl` 不再直接调用全局本地或远端进程 manager，只保留工具入口、权限/checkpoint 和结果 envelope。 +- 后续仍需迁移 Computer Use OS action、部分 Git/process/session host adapter、MCP auth URL helper 和 OpenCode/UI extension 的真实消费路径；这些路径必须在具备等价测试、host port 和交付形态矩阵后再迁移。 保护： diff --git a/scripts/core-boundaries/rules/source/forbidden-rules.mjs b/scripts/core-boundaries/rules/source/forbidden-rules.mjs index 20a414c31b..85b21a8fd7 100644 --- a/scripts/core-boundaries/rules/source/forbidden-rules.mjs +++ b/scripts/core-boundaries/rules/source/forbidden-rules.mjs @@ -1570,6 +1570,21 @@ export const forbiddenContentRules = [ message: 'core local ExecCommand adapter must not call the global local process manager; use injected TerminalPort', }, + { + regex: /\bget_global_remote_exec_process_manager\b/, + message: + 'core remote ExecCommand adapter must not call the global remote process manager; use injected RemoteExecPort', + }, + { + regex: /\bSSHConnectionManager\b/, + message: + 'core remote ExecCommand adapter must not depend on concrete SSH managers; use injected RemoteExecPort', + }, + { + regex: /\bSSHCommandOptions\b/, + message: + 'core remote ExecCommand adapter must not depend on concrete SSH command options; use injected RemoteExecPort', + }, { regex: /\bLocalExecCommandRequest\b/, message: @@ -1620,6 +1635,16 @@ export const forbiddenContentRules = [ message: 'core local WriteStdin adapter must not call the global local process manager; use injected TerminalPort', }, + { + regex: /\bget_global_remote_exec_process_manager\b/, + message: + 'core remote WriteStdin adapter must not call the global remote process manager; use injected RemoteExecPort', + }, + { + regex: /\bRemoteExecError\b/, + message: + 'core remote WriteStdin adapter must not match concrete remote exec errors; use PortErrorKind', + }, { regex: /\bLocalWriteStdinRequest\b/, message: @@ -1655,6 +1680,16 @@ export const forbiddenContentRules = [ message: 'core local ExecControl adapter must not call the global local process manager; use injected TerminalPort', }, + { + regex: /\bget_global_remote_exec_process_manager\b/, + message: + 'core remote ExecControl adapter must not call the global remote process manager; use injected RemoteExecPort', + }, + { + regex: /\bRemoteExecError\b/, + message: + 'core remote ExecControl adapter must not match concrete remote exec errors; use PortErrorKind', + }, { regex: /\bLocalExecControlRequest\b/, message: @@ -1680,6 +1715,11 @@ export const forbiddenContentRules = [ message: 'core local ExecCommand input adapter must not call the global local process manager; use injected TerminalPort', }, + { + regex: /\bget_global_remote_exec_process_manager\b/, + message: + 'core remote ExecCommand input adapter must not call the global remote process manager; use injected RemoteExecPort', + }, { regex: /\bLocalSendStdinRequest\b/, message: @@ -1745,6 +1785,16 @@ export const forbiddenContentRules = [ message: 'core exec_command env snapshot adapter must not own snapshot cache ttl; use tool-runtime exec_command', }, + { + regex: /\bget_global_remote_exec_process_manager\b/, + message: + 'core exec_command env snapshot adapter must not call the global remote process manager; use injected RemoteExecPort', + }, + { + regex: /\bSSHConnectionManager\b/, + message: + 'core exec_command env snapshot adapter must not depend on concrete SSH managers; use injected RemoteExecPort', + }, ], }, { diff --git a/scripts/core-boundaries/self-test.mjs b/scripts/core-boundaries/self-test.mjs index 3ad445b31a..6726bd2c95 100644 --- a/scripts/core-boundaries/self-test.mjs +++ b/scripts/core-boundaries/self-test.mjs @@ -989,6 +989,9 @@ export function runManifestParserSelfTest({ 'command\\s+-v\\s+bash', 'REMOTE_SHELL_PROBE_TIMEOUT_MS', 'get_global_exec_process_manager', + 'get_global_remote_exec_process_manager', + 'SSHConnectionManager', + 'SSHCommandOptions', 'LocalExecCommandRequest', 'CoreRuntimeServicesProvider::terminal_port', 'TerminalRuntimePort', @@ -1007,6 +1010,8 @@ export function runManifestParserSelfTest({ '"status"\\s*:\\s*"session_not_found"', 'No input was sent', 'get_global_exec_process_manager', + 'get_global_remote_exec_process_manager', + 'RemoteExecError', 'LocalWriteStdinRequest', 'CoreRuntimeServicesProvider::terminal_port', 'TerminalRuntimePort', @@ -1026,6 +1031,8 @@ export function runManifestParserSelfTest({ } for (const contract of [ 'get_global_exec_process_manager', + 'get_global_remote_exec_process_manager', + 'RemoteExecError', 'LocalExecControlRequest', 'CoreRuntimeServicesProvider::terminal_port', 'TerminalRuntimePort', @@ -1039,6 +1046,7 @@ export function runManifestParserSelfTest({ ); for (const contract of [ 'get_global_exec_process_manager', + 'get_global_remote_exec_process_manager', 'LocalSendStdinRequest', 'CoreRuntimeServicesProvider::terminal_port', 'TerminalRuntimePort', @@ -1060,6 +1068,8 @@ export function runManifestParserSelfTest({ 'ENV_SNAPSHOT_TIMEOUT_MS', 'ENV_SNAPSHOT_MAX_OUTPUT_CHARS', 'ENV_SNAPSHOT_TTL', + 'get_global_remote_exec_process_manager', + 'SSHConnectionManager', ]) { if (!coreExecCommandEnvSnapshotRuleText.includes(contract)) { throw new Error(`core exec_command env_snapshot boundary rule must forbid ${contract}`); diff --git a/src/apps/cli/src/agent/agentic_system.rs b/src/apps/cli/src/agent/agentic_system.rs index ea67a82286..d4be14d334 100644 --- a/src/apps/cli/src/agent/agentic_system.rs +++ b/src/apps/cli/src/agent/agentic_system.rs @@ -13,6 +13,9 @@ pub async fn init_agentic_system() -> Result { system .coordinator .set_terminal_port(CoreRuntimeServicesProvider::terminal_port()); + system + .coordinator + .set_remote_exec_port(CoreRuntimeServicesProvider::remote_exec_port()); Ok(system) } diff --git a/src/apps/desktop/src/api/agentic_api.rs b/src/apps/desktop/src/api/agentic_api.rs index 36a7079915..631b1dbd4b 100644 --- a/src/apps/desktop/src/api/agentic_api.rs +++ b/src/apps/desktop/src/api/agentic_api.rs @@ -1534,6 +1534,11 @@ pub async fn control_background_command( } else { coordinator.inner().terminal_port() }; + let remote_exec_port = if remote { + coordinator.inner().remote_exec_port() + } else { + None + }; control_exec_command_session( ExecCommandControlRequest { @@ -1544,6 +1549,7 @@ pub async fn control_background_command( yield_time_ms: Some(250), }, terminal_port.as_ref(), + remote_exec_port.as_ref(), ) .await .map(|response| { @@ -1606,6 +1612,11 @@ pub async fn send_background_command_input( } else { coordinator.inner().terminal_port() }; + let remote_exec_port = if remote { + coordinator.inner().remote_exec_port() + } else { + None + }; send_exec_command_input( ExecCommandInputRequest { session_id, @@ -1614,6 +1625,7 @@ pub async fn send_background_command_input( remote, }, terminal_port.as_ref(), + remote_exec_port.as_ref(), ) .await .map_err(|e| { diff --git a/src/apps/desktop/src/api/tool_api.rs b/src/apps/desktop/src/api/tool_api.rs index 171d06bf05..05f0e6402a 100644 --- a/src/apps/desktop/src/api/tool_api.rs +++ b/src/apps/desktop/src/api/tool_api.rs @@ -12,6 +12,7 @@ use bitfun_core::agentic::{ workspace::{local_workspace_services, remote_workspace_services}, WorkspaceBinding, }; +use bitfun_core::product_runtime::CoreRuntimeServicesProvider; use bitfun_core::service::remote_ssh::workspace_state::{ get_remote_workspace_manager, lookup_remote_connection, workspace_session_identity, }; @@ -172,7 +173,16 @@ async fn build_tool_context(workspace_path: Option<&str>) -> ToolUseContext { None => None, }; - ToolUseContext::for_tool_listing(workspace, workspace_services) + let remote_exec_port = workspace + .as_ref() + .is_some_and(WorkspaceBinding::is_remote) + .then(CoreRuntimeServicesProvider::remote_exec_port); + + ToolUseContext::for_tool_listing_with_remote_exec_port( + workspace, + workspace_services, + remote_exec_port, + ) } fn to_dynamic_mcp_tool_info( diff --git a/src/apps/desktop/src/lib.rs b/src/apps/desktop/src/lib.rs index d4e13a7e3d..a285aee455 100644 --- a/src/apps/desktop/src/lib.rs +++ b/src/apps/desktop/src/lib.rs @@ -1387,6 +1387,9 @@ async fn init_agentic_system() -> anyhow::Result<( coordinator.set_terminal_port( bitfun_core::product_runtime::CoreRuntimeServicesProvider::terminal_port(), ); + coordinator.set_remote_exec_port( + bitfun_core::product_runtime::CoreRuntimeServicesProvider::remote_exec_port(), + ); coordination::ConversationCoordinator::set_global(coordinator.clone()); diff --git a/src/apps/server/src/bootstrap.rs b/src/apps/server/src/bootstrap.rs index 6db7f800cb..379727cfaf 100644 --- a/src/apps/server/src/bootstrap.rs +++ b/src/apps/server/src/bootstrap.rs @@ -104,6 +104,9 @@ pub async fn initialize(workspace: Option) -> anyhow::Result>>, thread_goal_runtime: Arc, terminal_port: OnceLock>, + remote_exec_port: OnceLock>, } impl ConversationCoordinator { @@ -1063,6 +1065,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet active_turns_per_session: Arc::new(DashMap::new()), thread_goal_runtime: Arc::new(ThreadGoalRuntime::new()), terminal_port: OnceLock::new(), + remote_exec_port: OnceLock::new(), } } @@ -1080,6 +1083,16 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet self.terminal_port.get().map(Arc::clone) } + pub fn set_remote_exec_port(&self, remote_exec_port: Arc) { + if self.remote_exec_port.set(remote_exec_port).is_err() { + log::warn!("Remote exec port is already configured; ignoring duplicate injection"); + } + } + + pub fn remote_exec_port(&self) -> Option> { + self.remote_exec_port.get().map(Arc::clone) + } + /// Inject the DialogScheduler notification channel after construction. /// Called once during app initialization after the scheduler is created. pub fn set_scheduler_notifier(&self, tx: mpsc::Sender<(String, TurnOutcome)>) { @@ -2753,6 +2766,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet runtime_tool_restrictions: ToolRuntimeRestrictions::default(), workspace_services: manual_workspace_services, terminal_port: self.terminal_port(), + remote_exec_port: self.remote_exec_port(), round_injection: None, recover_partial_on_cancel: false, }; @@ -3377,6 +3391,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet runtime_tool_restrictions, workspace_services, terminal_port: self.terminal_port(), + remote_exec_port: self.remote_exec_port(), round_injection: self.round_injection_source.get().cloned(), recover_partial_on_cancel: false, }; @@ -4787,6 +4802,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet runtime_tool_restrictions, workspace_services: subagent_services, terminal_port: self.terminal_port(), + remote_exec_port: self.remote_exec_port(), // Subagents are autonomous; user steering is targeted at top-level // dialog turns only. Leave None so we don't intercept buffer entries // that belong to a different (parent) session/turn. @@ -6528,6 +6544,9 @@ mod tests { coordinator.set_terminal_port( bitfun_runtime_services::test_support::FakeRuntimeServicesProvider::terminal_port(), ); + coordinator.set_remote_exec_port( + bitfun_runtime_services::test_support::FakeRuntimeServicesProvider::remote_exec_port(), + ); (coordinator, session_manager) } @@ -6546,6 +6565,7 @@ mod tests { let (coordinator, _) = test_coordinator(); assert!(coordinator.terminal_port().is_some()); + assert!(coordinator.remote_exec_port().is_some()); } #[test] diff --git a/src/crates/assembly/core/src/agentic/execution/execution_engine.rs b/src/crates/assembly/core/src/agentic/execution/execution_engine.rs index 8768cefcb2..a86b0b5318 100644 --- a/src/crates/assembly/core/src/agentic/execution/execution_engine.rs +++ b/src/crates/assembly/core/src/agentic/execution/execution_engine.rs @@ -1167,6 +1167,7 @@ impl ExecutionEngine { cancellation_token: CancellationToken::new(), workspace_services: context.workspace_services.clone(), terminal_port: context.terminal_port.clone(), + remote_exec_port: context.remote_exec_port.clone(), recover_partial_on_cancel: context.recover_partial_on_cancel, }; @@ -2659,6 +2660,7 @@ impl ExecutionEngine { cancellation_token: CancellationToken::new(), workspace_services: context.workspace_services.clone(), terminal_port: context.terminal_port.clone(), + remote_exec_port: context.remote_exec_port.clone(), recover_partial_on_cancel: context.recover_partial_on_cancel, }; @@ -3615,6 +3617,7 @@ mod tests { runtime_tool_restrictions: ToolRuntimeRestrictions::default(), workspace_services: None, terminal_port: None, + remote_exec_port: None, round_injection: None, recover_partial_on_cancel: false, }; diff --git a/src/crates/assembly/core/src/agentic/execution/round_executor.rs b/src/crates/assembly/core/src/agentic/execution/round_executor.rs index eb7b3e7dd4..9ad21cc7b1 100644 --- a/src/crates/assembly/core/src/agentic/execution/round_executor.rs +++ b/src/crates/assembly/core/src/agentic/execution/round_executor.rs @@ -734,6 +734,7 @@ impl RoundExecutor { steering_interrupt: context.steering_interrupt.clone(), workspace_services: context.workspace_services.clone(), terminal_port: context.terminal_port.clone(), + remote_exec_port: context.remote_exec_port.clone(), }; // Read tool execution related configuration from global config @@ -1357,6 +1358,7 @@ mod tests { cancellation_token: CancellationToken::new(), workspace_services: None, terminal_port: None, + remote_exec_port: None, recover_partial_on_cancel: false, } } diff --git a/src/crates/assembly/core/src/agentic/execution/types.rs b/src/crates/assembly/core/src/agentic/execution/types.rs index fa1468f75e..ea98c36c3f 100644 --- a/src/crates/assembly/core/src/agentic/execution/types.rs +++ b/src/crates/assembly/core/src/agentic/execution/types.rs @@ -7,7 +7,7 @@ use crate::agentic::tools::ToolRuntimeRestrictions; use crate::agentic::workspace::WorkspaceServices; use crate::agentic::WorkspaceBinding; pub use bitfun_agent_runtime::events::FinishReason; -use bitfun_runtime_ports::{DelegationPolicy, TerminalPort}; +use bitfun_runtime_ports::{DelegationPolicy, RemoteExecPort, TerminalPort}; use serde_json::Value; use std::collections::HashMap; use std::sync::Arc; @@ -30,6 +30,8 @@ pub struct ExecutionContext { pub workspace_services: Option, /// Terminal execution provider injected by product assembly. pub terminal_port: Option>, + /// Remote execution provider injected by product assembly. + pub remote_exec_port: Option>, /// When set, engine drains pending round injections at each round boundary /// and injects them into the dialog history without ending the turn. pub round_injection: Option>, @@ -63,6 +65,7 @@ pub struct RoundContext { pub cancellation_token: CancellationToken, pub workspace_services: Option, pub terminal_port: Option>, + pub remote_exec_port: Option>, pub recover_partial_on_cancel: bool, } diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/command.rs b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/command.rs index b1e9cd67eb..5ed5b47894 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/command.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/command.rs @@ -11,21 +11,18 @@ use crate::agentic::tools::framework::{Tool, ToolResult, ToolUseContext, Validat use crate::infrastructure::events::event_system::{ get_global_event_system, BackendEvent::BackgroundCommandLifecycle, }; -use crate::service::remote_ssh::{ - get_global_remote_exec_process_manager, get_remote_workspace_manager, RemoteExecCommandRequest, - RemoteExecProcessLifecycleEvent, RemoteExecProcessLifecycleStatus, SSHCommandOptions, - SSHConnectionManager, -}; use crate::util::errors::{BitFunError, BitFunResult}; use crate::util::types::event::BackgroundCommandLifecycleInfo; use async_trait::async_trait; use bitfun_runtime_ports::{ - TerminalExecCommandRequest, TerminalExecProcessLifecycleEvent, - TerminalExecProcessLifecycleStatus, + RemoteExecCommandRequest, RemoteExecOneShotCommandRequest, RemoteExecPort, + RemoteExecProcessLifecycleEvent, RemoteExecProcessLifecycleStatus, TerminalExecCommandRequest, + TerminalExecProcessLifecycleEvent, TerminalExecProcessLifecycleStatus, }; use serde_json::{json, Value}; use std::collections::HashMap; use std::path::{Path, PathBuf}; +use std::sync::Arc; use terminal_core::ShellType; use tokio::sync::mpsc; use tool_runtime::exec_command::{ @@ -169,18 +166,15 @@ impl ExecCommandTool { } async fn resolve_remote_shell( - ssh_manager: &SSHConnectionManager, + remote_exec_port: &Arc, connection_id: &str, ) -> RemoteShell { - let result = ssh_manager - .execute_command_with_options( - connection_id, - remote_exec_shell_probe_command(), - SSHCommandOptions { - timeout_ms: Some(REMOTE_EXEC_SHELL_PROBE_TIMEOUT_MS), - cancellation_token: None, - }, - ) + let result = remote_exec_port + .exec_command_once(RemoteExecOneShotCommandRequest { + connection_id: connection_id.to_string(), + command: remote_exec_shell_probe_command().to_string(), + timeout_ms: Some(REMOTE_EXEC_SHELL_PROBE_TIMEOUT_MS), + }) .await; if let Ok(result) = result { @@ -397,23 +391,13 @@ impl ExecCommandTool { BitFunError::tool("remote connection id is required for ExecCommand".to_string()) })? .to_string(); - let ssh_manager = get_remote_workspace_manager() - .ok_or_else(|| { - BitFunError::tool( - "remote workspace manager is not initialized for ExecCommand".to_string(), - ) - })? - .get_ssh_manager() - .await - .ok_or_else(|| { - BitFunError::tool( - "remote SSH manager is not initialized for ExecCommand".to_string(), - ) - })?; + let remote_exec_port = context.remote_exec_port().ok_or_else(|| { + BitFunError::tool("remote exec runtime service is required for ExecCommand".to_string()) + })?; let yield_time_ms = parsed_input.yield_time_ms; - let shell = Self::resolve_remote_shell(&ssh_manager, &connection_id).await; + let shell = Self::resolve_remote_shell(remote_exec_port, &connection_id).await; let env_snapshot = remote_env_snapshot_for( - ssh_manager.clone(), + remote_exec_port, &connection_id, &shell.path, &shell.shell_type, @@ -448,24 +432,21 @@ impl ExecCommandTool { }; let request = RemoteExecCommandRequest { - ssh_manager, connection_id, command, tty, yield_time_ms: Some(yield_time_ms), max_output_chars: None, - lifecycle_tx: Self::start_remote_lifecycle_bridge(context, self.name()), - output_capture_tx, + lifecycle_sink: Self::start_remote_lifecycle_bridge(context, self.name()), + output_sink: output_capture_tx, }; let progress_bridge = ExecOutputProgressBridge::start(context, self.name()); let response_result = if let Some(bridge) = progress_bridge.as_ref() { - get_global_remote_exec_process_manager() + remote_exec_port .exec_command_streaming(request, bridge.sender()) .await } else { - get_global_remote_exec_process_manager() - .exec_command(request) - .await + remote_exec_port.exec_command(request).await }; if let Some(bridge) = progress_bridge { bridge.finish().await; @@ -478,7 +459,10 @@ impl ExecCommandTool { .finish(capture_id, BackgroundCommandOutputStatus::Failed, None) .await; } - return Err(BitFunError::tool(format!("ExecCommand failed: {error}"))); + return Err(BitFunError::tool(format!( + "ExecCommand failed: {}", + error.message + ))); } }; let completion = response.completion.map(exec_command_remote_completion); @@ -768,11 +752,83 @@ mod tests { use std::collections::HashMap; use std::path::Path; use std::path::PathBuf; + use std::sync::Arc; use terminal_core::ShellType; use tool_runtime::exec_command::{ remote_exec_shell_login_args, EXEC_COMMAND_POWERSHELL_UTF8_OUTPUT_PREFIX, }; + #[derive(Debug)] + struct ShellProbeRemoteExecPort { + response: bitfun_runtime_ports::RemoteExecOneShotCommandResponse, + } + + impl bitfun_runtime_ports::RuntimeServicePort for ShellProbeRemoteExecPort { + fn capability(&self) -> bitfun_runtime_ports::RuntimeServiceCapability { + bitfun_runtime_ports::RuntimeServiceCapability::RemoteExec + } + } + + #[async_trait::async_trait] + impl bitfun_runtime_ports::RemoteExecPort for ShellProbeRemoteExecPort { + async fn exec_command_once( + &self, + _request: bitfun_runtime_ports::RemoteExecOneShotCommandRequest, + ) -> bitfun_runtime_ports::PortResult + { + Ok(self.response.clone()) + } + + async fn exec_command( + &self, + _request: bitfun_runtime_ports::RemoteExecCommandRequest, + ) -> bitfun_runtime_ports::PortResult + { + panic!("shell probe must not use managed remote exec sessions"); + } + + async fn exec_command_streaming( + &self, + _request: bitfun_runtime_ports::RemoteExecCommandRequest, + _output_sink: bitfun_runtime_ports::RemoteExecStreamingOutputSink, + ) -> bitfun_runtime_ports::PortResult + { + panic!("shell probe must not use managed remote exec sessions"); + } + + async fn write_stdin( + &self, + _request: bitfun_runtime_ports::RemoteWriteStdinRequest, + ) -> bitfun_runtime_ports::PortResult + { + panic!("shell probe must not write stdin"); + } + + async fn write_stdin_streaming( + &self, + _request: bitfun_runtime_ports::RemoteWriteStdinRequest, + _output_sink: bitfun_runtime_ports::RemoteExecStreamingOutputSink, + ) -> bitfun_runtime_ports::PortResult + { + panic!("shell probe must not write stdin"); + } + + async fn send_stdin( + &self, + _request: bitfun_runtime_ports::RemoteSendStdinRequest, + ) -> bitfun_runtime_ports::PortResult<()> { + panic!("shell probe must not send stdin"); + } + + async fn control_session( + &self, + _request: bitfun_runtime_ports::RemoteExecControlRequest, + ) -> bitfun_runtime_ports::PortResult + { + panic!("shell probe must not control managed sessions"); + } + } + #[test] fn powershell_commands_force_utf8_output() { let argv = ExecCommandTool::argv_for_shell( @@ -925,6 +981,25 @@ mod tests { assert_eq!(metadata.path, "/usr/local/bin/xonsh"); } + #[tokio::test] + async fn remote_shell_probe_uses_stdout_only() { + let remote_exec_port: Arc = + Arc::new(ShellProbeRemoteExecPort { + response: bitfun_runtime_ports::RemoteExecOneShotCommandResponse { + stdout: "/bin/bash\n".to_string(), + stderr: "/tmp/not-a-shell-from-stderr\n".to_string(), + exit_code: 0, + interrupted: false, + timed_out: false, + }, + }); + + let shell = ExecCommandTool::resolve_remote_shell(&remote_exec_port, "conn-1").await; + + assert_eq!(shell.path, "/bin/bash"); + assert_eq!(shell.shell_type, ShellType::Bash); + } + #[test] fn remote_shell_login_args_use_login_without_interactive_startup() { assert_eq!(remote_exec_shell_login_args(), &["-lc"]); diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/completion.rs b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/completion.rs index 0226325920..d75d547200 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/completion.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/completion.rs @@ -1,10 +1,7 @@ -use crate::service::remote_ssh::{ - RemoteExecSessionCompletion, RemoteExecSessionCompletionSource, - RemoteExecSessionCompletionStatus, -}; use bitfun_runtime_ports::{ - TerminalExecSessionCompletion, TerminalExecSessionCompletionSource, - TerminalExecSessionCompletionStatus, + RemoteExecSessionCompletion, RemoteExecSessionCompletionSource, + RemoteExecSessionCompletionStatus, TerminalExecSessionCompletion, + TerminalExecSessionCompletionSource, TerminalExecSessionCompletionStatus, }; use tool_runtime::exec_command::{ ExecCommandCompletion, ExecCommandCompletionSource, ExecCommandCompletionStatus, diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/control.rs b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/control.rs index c5cbdcf4f1..9bcd79be1c 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/control.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/control.rs @@ -1,13 +1,10 @@ use super::completion::{exec_command_local_completion, exec_command_remote_completion}; use crate::agentic::tools::framework::{Tool, ToolResult, ToolUseContext, ValidationResult}; -use crate::service::remote_ssh::{ - get_global_remote_exec_process_manager, RemoteExecControlAction, RemoteExecControlOrigin, - RemoteExecControlRequest, RemoteExecError, -}; use crate::util::errors::{BitFunError, BitFunResult}; use async_trait::async_trait; use bitfun_runtime_ports::{ - PortErrorKind, TerminalExecControlAction, TerminalExecControlOrigin, + PortErrorKind, RemoteExecControlAction, RemoteExecControlOrigin, RemoteExecControlRequest, + RemoteExecPort, TerminalExecControlAction, TerminalExecControlOrigin, TerminalExecControlRequest, TerminalPort, }; use serde_json::{json, Value}; @@ -53,9 +50,15 @@ pub enum ExecCommandControlError { pub async fn control_exec_command_session( request: ExecCommandControlRequest, terminal_port: Option<&Arc>, + remote_exec_port: Option<&Arc>, ) -> Result { if request.remote { - let response = get_global_remote_exec_process_manager() + let remote_exec_port = remote_exec_port.ok_or_else(|| { + ExecCommandControlError::Tool(BitFunError::tool( + "remote exec runtime service is required for ExecControl".to_string(), + )) + })?; + let response = remote_exec_port .control_session(RemoteExecControlRequest { session_id: request.session_id, action: ExecControlTool::remote_action(request.action), @@ -65,11 +68,12 @@ pub async fn control_exec_command_session( }) .await .map_err(|error| match error { - RemoteExecError::SessionNotFound(session_id) => { - ExecCommandControlError::SessionNotFound(session_id) + error if error.kind == PortErrorKind::NotFound => { + ExecCommandControlError::SessionNotFound(request.session_id) } error => ExecCommandControlError::Tool(BitFunError::tool(format!( - "ExecControl failed: {error}" + "ExecControl failed: {}", + error.message ))), })?; @@ -180,7 +184,11 @@ impl ExecControlTool { } } - async fn call_remote_pipe(&self, input: &Value) -> BitFunResult> { + async fn call_remote_pipe( + &self, + input: &Value, + context: &ToolUseContext, + ) -> BitFunResult> { if let Some(message) = exec_command_control_tool_input_validation_message(input) { return Err(BitFunError::tool(message.to_string())); } @@ -197,6 +205,7 @@ impl ExecControlTool { yield_time_ms: parsed_input.yield_time_ms, }, None, + context.remote_exec_port(), ) .await { @@ -312,7 +321,7 @@ Output is only what was produced during this tool call's wait window."# context: &ToolUseContext, ) -> BitFunResult> { if context.is_remote() { - return self.call_remote_pipe(input).await; + return self.call_remote_pipe(input, context).await; } if let Some(message) = exec_command_control_tool_input_validation_message(input) { @@ -332,6 +341,7 @@ Output is only what was produced during this tool call's wait window."# yield_time_ms: parsed_input.yield_time_ms, }, terminal_port, + None, ) .await { @@ -373,15 +383,21 @@ mod tests { }; use crate::agentic::tools::framework::ToolResult; use bitfun_runtime_ports::{ - PortError, PortErrorKind, PortResult, RuntimeServiceCapability, RuntimeServicePort, - TerminalExecCommandRequest, TerminalExecCommandResponse, TerminalExecControlRequest, - TerminalExecStreamingOutputSink, TerminalPort, TerminalSendStdinRequest, - TerminalWriteStdinRequest, + PortError, PortErrorKind, PortResult, RemoteExecCommandRequest, RemoteExecCommandResponse, + RemoteExecControlRequest, RemoteExecOneShotCommandRequest, + RemoteExecOneShotCommandResponse, RemoteExecPort, RemoteExecStreamingOutputSink, + RemoteSendStdinRequest, RemoteWriteStdinRequest, RuntimeServiceCapability, + RuntimeServicePort, TerminalExecCommandRequest, TerminalExecCommandResponse, + TerminalExecControlRequest, TerminalExecStreamingOutputSink, TerminalPort, + TerminalSendStdinRequest, TerminalWriteStdinRequest, }; #[derive(Debug)] struct MissingSessionTerminalPort; + #[derive(Debug)] + struct MissingSessionRemoteExecPort; + impl RuntimeServicePort for MissingSessionTerminalPort { fn capability(&self) -> RuntimeServiceCapability { RuntimeServiceCapability::Terminal @@ -435,6 +451,66 @@ mod tests { } } + impl RuntimeServicePort for MissingSessionRemoteExecPort { + fn capability(&self) -> RuntimeServiceCapability { + RuntimeServiceCapability::RemoteExec + } + } + + #[async_trait::async_trait] + impl RemoteExecPort for MissingSessionRemoteExecPort { + async fn exec_command_once( + &self, + _request: RemoteExecOneShotCommandRequest, + ) -> PortResult { + unused_remote_exec_response() + } + + async fn exec_command( + &self, + _request: RemoteExecCommandRequest, + ) -> PortResult { + unused_remote_exec_response() + } + + async fn exec_command_streaming( + &self, + _request: RemoteExecCommandRequest, + _output_sink: RemoteExecStreamingOutputSink, + ) -> PortResult { + unused_remote_exec_response() + } + + async fn write_stdin( + &self, + _request: RemoteWriteStdinRequest, + ) -> PortResult { + unused_remote_exec_response() + } + + async fn write_stdin_streaming( + &self, + _request: RemoteWriteStdinRequest, + _output_sink: RemoteExecStreamingOutputSink, + ) -> PortResult { + unused_remote_exec_response() + } + + async fn send_stdin(&self, _request: RemoteSendStdinRequest) -> PortResult<()> { + Err(PortError::new( + PortErrorKind::Backend, + "unused remote exec test method", + )) + } + + async fn control_session( + &self, + _request: RemoteExecControlRequest, + ) -> PortResult { + Err(PortError::new(PortErrorKind::NotFound, "session not found")) + } + } + fn unused_terminal_response() -> PortResult { Err(PortError::new( PortErrorKind::Backend, @@ -442,6 +518,13 @@ mod tests { )) } + fn unused_remote_exec_response() -> PortResult { + Err(PortError::new( + PortErrorKind::Backend, + "unused remote exec test method", + )) + } + #[test] fn session_not_found_result_uses_plain_assistant_message() { let results = ExecControlTool::session_not_found_result( @@ -487,6 +570,7 @@ mod tests { yield_time_ms: Some(0), }, Some(&terminal_port), + None, ) .await .expect_err("missing session should be structured"); @@ -496,4 +580,28 @@ mod tests { ExecCommandControlError::SessionNotFound(987_654) )); } + + #[tokio::test] + async fn control_exec_command_session_returns_structured_remote_session_not_found() { + let remote_exec_port: std::sync::Arc = + std::sync::Arc::new(MissingSessionRemoteExecPort); + let error = control_exec_command_session( + ExecCommandControlRequest { + session_id: 987_655, + action: ExecCommandControlAction::Kill, + origin: ExecCommandControlOrigin::ModelTool, + remote: true, + yield_time_ms: Some(0), + }, + None, + Some(&remote_exec_port), + ) + .await + .expect_err("missing remote session should be structured"); + + assert!(matches!( + error, + ExecCommandControlError::SessionNotFound(987_655) + )); + } } diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/env_snapshot.rs b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/env_snapshot.rs index 44bed25478..c4643f715a 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/env_snapshot.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/env_snapshot.rs @@ -1,8 +1,9 @@ use super::shell_kind::exec_command_shell_kind; -use crate::service::remote_ssh::{ - get_global_remote_exec_process_manager, RemoteExecCommandRequest, RemoteExecControlAction, - RemoteExecControlOrigin, RemoteExecControlRequest, SSHConnectionManager, +use bitfun_runtime_ports::{ + RemoteExecCommandRequest, RemoteExecControlAction, RemoteExecControlOrigin, + RemoteExecControlRequest, RemoteExecPort, }; +use std::sync::Arc; use std::sync::OnceLock; use terminal_core::ShellType; use tool_runtime::exec_command::{ @@ -16,7 +17,7 @@ static REMOTE_ENV_SNAPSHOT_CACHE: OnceLock = pub(super) type RemoteEnvSnapshot = ExecCommandRemoteEnvSnapshot; pub(super) async fn remote_env_snapshot_for( - ssh_manager: SSHConnectionManager, + remote_exec_port: &Arc, connection_id: &str, shell_path: &str, shell_type: &ShellType, @@ -33,7 +34,8 @@ pub(super) async fn remote_env_snapshot_for( } let snapshot = - match capture_remote_env_snapshot(ssh_manager, connection_id, shell_path, shell_type).await + match capture_remote_env_snapshot(remote_exec_port, connection_id, shell_path, shell_type) + .await { Ok(snapshot) => snapshot, Err(_) => return None, @@ -43,29 +45,27 @@ pub(super) async fn remote_env_snapshot_for( } async fn capture_remote_env_snapshot( - ssh_manager: SSHConnectionManager, + remote_exec_port: &Arc, connection_id: &str, shell_path: &str, shell_type: &ShellType, ) -> anyhow::Result { let command = remote_env_snapshot_command(shell_path, shell_type); - let manager = get_global_remote_exec_process_manager(); let policy = remote_exec_env_snapshot_capture_policy(); - let response = manager + let response = remote_exec_port .exec_command(RemoteExecCommandRequest { - ssh_manager, connection_id: connection_id.to_string(), command, tty: true, yield_time_ms: Some(policy.timeout_ms), max_output_chars: Some(policy.max_output_chars), - lifecycle_tx: None, - output_capture_tx: None, + lifecycle_sink: None, + output_sink: None, }) .await?; if let Some(session_id) = response.session_id { - let _ = manager + let _ = remote_exec_port .control_session(RemoteExecControlRequest { session_id, action: RemoteExecControlAction::Kill, diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/input.rs b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/input.rs index 0671cb5c8e..cb6fa9c44e 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/input.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/input.rs @@ -1,6 +1,7 @@ -use crate::service::remote_ssh::{get_global_remote_exec_process_manager, RemoteSendStdinRequest}; use crate::util::errors::{BitFunError, BitFunResult}; -use bitfun_runtime_ports::{TerminalPort, TerminalSendStdinRequest}; +use bitfun_runtime_ports::{ + RemoteExecPort, RemoteSendStdinRequest, TerminalPort, TerminalSendStdinRequest, +}; use std::sync::Arc; #[derive(Debug, Clone)] @@ -14,16 +15,24 @@ pub struct ExecCommandInputRequest { pub async fn send_exec_command_input( request: ExecCommandInputRequest, terminal_port: Option<&Arc>, + remote_exec_port: Option<&Arc>, ) -> BitFunResult<()> { if request.remote { - get_global_remote_exec_process_manager() + let remote_exec_port = remote_exec_port.ok_or_else(|| { + BitFunError::tool( + "remote exec runtime service is required for ExecCommand input".to_string(), + ) + })?; + remote_exec_port .send_stdin(RemoteSendStdinRequest { session_id: request.session_id, chars: request.chars, append_enter: request.append_enter, }) .await - .map_err(|error| BitFunError::tool(format!("ExecCommand input failed: {error}")))?; + .map_err(|error| { + BitFunError::tool(format!("ExecCommand input failed: {}", error.message)) + })?; return Ok(()); } diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/stdin.rs b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/stdin.rs index 088b204689..8efaee5ef2 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/stdin.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/exec_command/stdin.rs @@ -1,12 +1,9 @@ use super::completion::{exec_command_local_completion, exec_command_remote_completion}; use super::progress::ExecOutputProgressBridge; use crate::agentic::tools::framework::{Tool, ToolResult, ToolUseContext, ValidationResult}; -use crate::service::remote_ssh::{ - get_global_remote_exec_process_manager, RemoteExecError, RemoteWriteStdinRequest, -}; use crate::util::errors::{BitFunError, BitFunResult}; use async_trait::async_trait; -use bitfun_runtime_ports::{PortErrorKind, TerminalWriteStdinRequest}; +use bitfun_runtime_ports::{PortErrorKind, RemoteWriteStdinRequest, TerminalWriteStdinRequest}; use serde_json::{json, Value}; use tool_runtime::exec_command::{ render_write_stdin_response_for_assistant, write_stdin_input_from_input, @@ -57,25 +54,31 @@ impl WriteStdinTool { yield_time_ms: Some(parsed_input.yield_time_ms), max_output_chars: None, }; + let remote_exec_port = context.remote_exec_port().ok_or_else(|| { + BitFunError::tool("remote exec runtime service is required for WriteStdin".to_string()) + })?; let progress_bridge = ExecOutputProgressBridge::start(context, self.name()); let response_result = if let Some(bridge) = progress_bridge.as_ref() { - get_global_remote_exec_process_manager() + remote_exec_port .write_stdin_streaming(request, bridge.sender()) .await } else { - get_global_remote_exec_process_manager() - .write_stdin(request) - .await + remote_exec_port.write_stdin(request).await }; if let Some(bridge) = progress_bridge { bridge.finish().await; } let response = match response_result { Ok(response) => response, - Err(RemoteExecError::SessionNotFound(session_id)) => { + Err(error) if error.kind == PortErrorKind::NotFound => { return Ok(Self::session_not_found_result(session_id, true)); } - Err(error) => return Err(BitFunError::tool(format!("WriteStdin failed: {error}"))), + Err(error) => { + return Err(BitFunError::tool(format!( + "WriteStdin failed: {}", + error.message + ))); + } }; let data = write_stdin_result_value(ExecCommandResultFields { diff --git a/src/crates/assembly/core/src/agentic/tools/pipeline/state_manager.rs b/src/crates/assembly/core/src/agentic/tools/pipeline/state_manager.rs index ce79795259..a198ce3d44 100644 --- a/src/crates/assembly/core/src/agentic/tools/pipeline/state_manager.rs +++ b/src/crates/assembly/core/src/agentic/tools/pipeline/state_manager.rs @@ -323,6 +323,7 @@ mod tests { steering_interrupt: None, workspace_services: None, terminal_port: None, + remote_exec_port: None, }, ToolExecutionOptions::default(), ) diff --git a/src/crates/assembly/core/src/agentic/tools/pipeline/tool_pipeline.rs b/src/crates/assembly/core/src/agentic/tools/pipeline/tool_pipeline.rs index 7e5b7ee333..0cba3d820f 100644 --- a/src/crates/assembly/core/src/agentic/tools/pipeline/tool_pipeline.rs +++ b/src/crates/assembly/core/src/agentic/tools/pipeline/tool_pipeline.rs @@ -1572,6 +1572,7 @@ mod tests { steering_interrupt: None, workspace_services: None, terminal_port: None, + remote_exec_port: None, } } diff --git a/src/crates/assembly/core/src/agentic/tools/pipeline/types.rs b/src/crates/assembly/core/src/agentic/tools/pipeline/types.rs index 77b3f72619..51d9c382dd 100644 --- a/src/crates/assembly/core/src/agentic/tools/pipeline/types.rs +++ b/src/crates/assembly/core/src/agentic/tools/pipeline/types.rs @@ -6,7 +6,7 @@ use crate::agentic::round_preempt::DialogRoundInjectionInterrupt; use crate::agentic::tools::ToolRuntimeRestrictions; use crate::agentic::workspace::WorkspaceServices; use crate::agentic::WorkspaceBinding; -use bitfun_runtime_ports::{DelegationPolicy, TerminalPort}; +use bitfun_runtime_ports::{DelegationPolicy, RemoteExecPort, TerminalPort}; use std::collections::HashMap; use std::sync::Arc; use std::time::SystemTime; @@ -80,6 +80,7 @@ pub struct ToolExecutionContext { pub steering_interrupt: Option, pub workspace_services: Option, pub terminal_port: Option>, + pub remote_exec_port: Option>, } /// Tool execution task diff --git a/src/crates/assembly/core/src/agentic/tools/tool_context_runtime.rs b/src/crates/assembly/core/src/agentic/tools/tool_context_runtime.rs index 56e4398dca..80a0ce8638 100644 --- a/src/crates/assembly/core/src/agentic/tools/tool_context_runtime.rs +++ b/src/crates/assembly/core/src/agentic/tools/tool_context_runtime.rs @@ -37,7 +37,7 @@ use bitfun_agent_runtime::checkpoint::{ }; use bitfun_agent_runtime::remote_file_delivery::TOOL_CONTEXT_REMOTE_FILE_DELIVERY_KEY; use bitfun_agent_tools::{PortableToolContextProvider, ToolContextFacts, ToolWorkspaceKind}; -use bitfun_runtime_ports::{DelegationPolicy, TerminalPort, ToolRuntimeHandles}; +use bitfun_runtime_ports::{DelegationPolicy, RemoteExecPort, TerminalPort, ToolRuntimeHandles}; use log::warn; use serde_json::Value; use sha2::{Digest, Sha256}; @@ -129,9 +129,21 @@ impl ToolUseContext { self.runtime_handles.terminal_port() } + pub fn remote_exec_port(&self) -> Option<&Arc> { + self.runtime_handles.remote_exec_port() + } + pub fn for_tool_listing( workspace: Option, workspace_services: Option, + ) -> Self { + Self::for_tool_listing_with_remote_exec_port(workspace, workspace_services, None) + } + + pub fn for_tool_listing_with_remote_exec_port( + workspace: Option, + workspace_services: Option, + remote_exec_port: Option>, ) -> Self { Self { tool_call_id: None, @@ -143,7 +155,12 @@ impl ToolUseContext { custom_data: HashMap::new(), computer_use_host: None, runtime_tool_restrictions: ToolRuntimeRestrictions::default(), - runtime_handles: core_tool_runtime_handles(workspace_services, None, None), + runtime_handles: core_tool_runtime_handles( + workspace_services, + None, + None, + remote_exec_port, + ), } } } @@ -221,6 +238,7 @@ pub(crate) fn build_tool_use_context_for_execution_context( context.workspace_services.clone(), Some(cancellation_token), context.terminal_port.clone(), + context.remote_exec_port.clone(), ), runtime_tool_restrictions: context.runtime_tool_restrictions.clone(), } @@ -273,7 +291,7 @@ pub(crate) fn build_tool_description_context( custom_data, computer_use_host: None, runtime_tool_restrictions: ToolRuntimeRestrictions::default(), - runtime_handles: core_tool_runtime_handles(workspace_services.cloned(), None, None), + runtime_handles: core_tool_runtime_handles(workspace_services.cloned(), None, None, None), } } @@ -281,9 +299,11 @@ fn core_tool_runtime_handles( workspace_services: Option, cancellation_token: Option, terminal_port: Option>, + remote_exec_port: Option>, ) -> ToolRuntimeHandles { ToolRuntimeHandles::new(workspace_services, cancellation_token) .with_terminal_port(terminal_port) + .with_remote_exec_port(remote_exec_port) } fn build_tool_context_custom_data(context: &ToolExecutionContext) -> HashMap { @@ -1350,6 +1370,7 @@ mod task_context_tests { steering_interrupt: None, workspace_services: None, terminal_port: None, + remote_exec_port: None, }, ToolExecutionOptions::default(), ) diff --git a/src/crates/assembly/core/src/product_runtime/runtime_services.rs b/src/crates/assembly/core/src/product_runtime/runtime_services.rs index 985b7c11c6..b6fb081420 100644 --- a/src/crates/assembly/core/src/product_runtime/runtime_services.rs +++ b/src/crates/assembly/core/src/product_runtime/runtime_services.rs @@ -5,6 +5,8 @@ use std::sync::Arc; +#[cfg(feature = "ssh-remote")] +use bitfun_runtime_ports::{PortError, PortErrorKind, PortResult, RemoteExecPort}; use bitfun_runtime_ports::{ RemoteProjectionPort, RemoteWorkspacePort, SessionStorePort, TerminalPort, }; @@ -20,6 +22,35 @@ use crate::service_agent_runtime::{ CoreRemoteWorkspaceFileRuntimeHost, CoreRemoteWorkspaceRuntimeHost, }; +#[cfg(feature = "ssh-remote")] +#[derive(Debug, Clone, Copy, Default)] +struct CoreRemoteExecSshManagerProvider; + +#[cfg(feature = "ssh-remote")] +#[async_trait::async_trait] +impl bitfun_services_integrations::remote_ssh::RemoteExecSshManagerProvider + for CoreRemoteExecSshManagerProvider +{ + async fn ssh_manager( + &self, + ) -> PortResult { + let manager = + crate::service::remote_ssh::get_remote_workspace_manager().ok_or_else(|| { + PortError::new( + PortErrorKind::NotAvailable, + "remote workspace manager is not initialized", + ) + })?; + + manager.get_ssh_manager().await.ok_or_else(|| { + PortError::new( + PortErrorKind::NotAvailable, + "remote SSH manager is not initialized", + ) + }) + } +} + #[derive(Debug, Clone, Copy, Default)] pub struct CoreRuntimeServicesProvider; @@ -31,6 +62,15 @@ impl CoreRuntimeServicesProvider { pub fn terminal_port() -> Arc { Arc::new(TerminalRuntimePort::default()) } + + #[cfg(feature = "ssh-remote")] + pub fn remote_exec_port() -> Arc { + Arc::new( + bitfun_services_integrations::remote_ssh::RemoteExecRuntimePort::new(Arc::new( + CoreRemoteExecSshManagerProvider, + )), + ) + } } impl RuntimeServicesProvider for CoreRuntimeServicesProvider { @@ -44,6 +84,9 @@ impl RuntimeServicesProvider for CoreRuntimeServicesProvider { .with_optional_git(Some(RuntimeServiceMarkerPort::git_port())) .with_optional_mcp_catalog(Some(RuntimeServiceMarkerPort::mcp_catalog_port())); + #[cfg(feature = "ssh-remote")] + let builder = builder.with_optional_remote_exec(Some(Self::remote_exec_port())); + #[cfg(feature = "service-integrations")] { let remote_workspace: Arc = diff --git a/src/crates/contracts/runtime-ports/AGENTS.md b/src/crates/contracts/runtime-ports/AGENTS.md index 142b084e03..89f7fb56cc 100644 --- a/src/crates/contracts/runtime-ports/AGENTS.md +++ b/src/crates/contracts/runtime-ports/AGENTS.md @@ -13,6 +13,9 @@ facts. It is an interface crate, not a runtime implementation crate. or catch-all context structs. - This crate may define portable request/response DTOs, runtime handles, capability facts, cancellation surfaces, and service traits. +- `RemoteExecPort` owns only remote command/stdin/control DTOs, bounded + one-shot command results, and lifecycle event shapes; SSH managers, channels, + process storage, and workspace lookup do not belong here. - `SessionStorePort` owns typed session storage-path resolution plus restore / load request and timing facts only. Concrete session persistence, file IO, session lifecycle, context restore, and prompt assembly do not belong here. diff --git a/src/crates/contracts/runtime-ports/src/lib.rs b/src/crates/contracts/runtime-ports/src/lib.rs index ed9193b034..152d56d11e 100644 --- a/src/crates/contracts/runtime-ports/src/lib.rs +++ b/src/crates/contracts/runtime-ports/src/lib.rs @@ -59,6 +59,7 @@ pub enum RuntimeServiceCapability { Events, Clock, Terminal, + RemoteExec, Network, Git, McpCatalog, @@ -78,6 +79,7 @@ impl RuntimeServiceCapability { Self::Events => "events", Self::Clock => "clock", Self::Terminal => "terminal", + Self::RemoteExec => "remote_exec", Self::Network => "network", Self::Git => "git", Self::McpCatalog => "mcp_catalog", @@ -437,6 +439,121 @@ pub type TerminalExecLifecycleSink = mpsc::UnboundedSender; pub type TerminalExecStreamingOutputSink = mpsc::Sender; +#[derive(Debug, Clone)] +pub struct RemoteExecCommandRequest { + pub connection_id: String, + pub command: String, + pub tty: bool, + pub yield_time_ms: Option, + pub max_output_chars: Option, + pub lifecycle_sink: Option, + pub output_sink: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RemoteExecOneShotCommandRequest { + pub connection_id: String, + pub command: String, + pub timeout_ms: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RemoteExecOneShotCommandResponse { + pub stdout: String, + pub stderr: String, + pub exit_code: i32, + pub interrupted: bool, + pub timed_out: bool, +} + +#[derive(Debug, Clone)] +pub struct RemoteWriteStdinRequest { + pub session_id: i32, + pub chars: String, + pub append_enter: bool, + pub yield_time_ms: Option, + pub max_output_chars: Option, +} + +#[derive(Debug, Clone)] +pub struct RemoteSendStdinRequest { + pub session_id: i32, + pub chars: String, + pub append_enter: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RemoteExecControlAction { + Interrupt, + Kill, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RemoteExecControlOrigin { + ModelTool, + OutOfBand, +} + +#[derive(Debug, Clone)] +pub struct RemoteExecControlRequest { + pub session_id: i32, + pub action: RemoteExecControlAction, + pub origin: RemoteExecControlOrigin, + pub yield_time_ms: Option, + pub max_output_chars: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RemoteExecSessionCompletionStatus { + Exited, + Interrupted, + Killed, + Pruned, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RemoteExecSessionCompletionSource { + Process, + OutOfBandControl, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RemoteExecSessionCompletion { + pub status: RemoteExecSessionCompletionStatus, + pub source: RemoteExecSessionCompletionSource, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct RemoteExecCommandResponse { + pub chunk_id: String, + pub wall_time_seconds: f64, + pub output: String, + pub session_id: Option, + pub exit_code: Option, + pub original_output_chars: usize, + pub completion: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RemoteExecProcessLifecycleStatus { + Running, + Exited, + Interrupted, + Killed, + Pruned, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RemoteExecProcessLifecycleEvent { + pub session_id: i32, + pub status: RemoteExecProcessLifecycleStatus, + pub exit_code: Option, +} + +pub type RemoteExecLifecycleSink = mpsc::UnboundedSender; +pub type RemoteExecOutputSink = mpsc::UnboundedSender; +pub type RemoteExecStreamingOutputSink = mpsc::Sender; + /// Runtime handles injected into tool execution contexts. /// /// This bundle is intentionally handle-only. Concrete local or remote @@ -447,6 +564,7 @@ pub struct ToolRuntimeHandles { workspace_services: Option, cancellation_token: Option, terminal_port: Option>, + remote_exec_port: Option>, } impl ToolRuntimeHandles { @@ -458,6 +576,7 @@ impl ToolRuntimeHandles { workspace_services, cancellation_token, terminal_port: None, + remote_exec_port: None, } } @@ -466,6 +585,14 @@ impl ToolRuntimeHandles { self } + pub fn with_remote_exec_port( + mut self, + remote_exec_port: Option>, + ) -> Self { + self.remote_exec_port = remote_exec_port; + self + } + pub fn workspace_services(&self) -> Option<&WorkspaceServices> { self.workspace_services.as_ref() } @@ -477,6 +604,10 @@ impl ToolRuntimeHandles { pub fn terminal_port(&self) -> Option<&Arc> { self.terminal_port.as_ref() } + + pub fn remote_exec_port(&self) -> Option<&Arc> { + self.remote_exec_port.as_ref() + } } impl std::fmt::Debug for ToolRuntimeHandles { @@ -500,6 +631,13 @@ impl std::fmt::Debug for ToolRuntimeHandles { "terminal_port", &self.terminal_port.as_ref().map(|_| ""), ) + .field( + "remote_exec_port", + &self + .remote_exec_port + .as_ref() + .map(|_| ""), + ) .finish() } } @@ -564,6 +702,43 @@ pub trait TerminalPort: RuntimeServicePort + std::fmt::Debug { ) -> PortResult; } +#[async_trait::async_trait] +pub trait RemoteExecPort: RuntimeServicePort + std::fmt::Debug { + async fn exec_command_once( + &self, + request: RemoteExecOneShotCommandRequest, + ) -> PortResult; + + async fn exec_command( + &self, + request: RemoteExecCommandRequest, + ) -> PortResult; + + async fn exec_command_streaming( + &self, + request: RemoteExecCommandRequest, + output_sink: RemoteExecStreamingOutputSink, + ) -> PortResult; + + async fn write_stdin( + &self, + request: RemoteWriteStdinRequest, + ) -> PortResult; + + async fn write_stdin_streaming( + &self, + request: RemoteWriteStdinRequest, + output_sink: RemoteExecStreamingOutputSink, + ) -> PortResult; + + async fn send_stdin(&self, request: RemoteSendStdinRequest) -> PortResult<()>; + + async fn control_session( + &self, + request: RemoteExecControlRequest, + ) -> PortResult; +} + pub trait NetworkPort: RuntimeServicePort {} pub trait GitPort: RuntimeServicePort {} @@ -2605,7 +2780,7 @@ mod tests { )); assert_eq!( format!("{:?}", handles), - "ToolRuntimeHandles { workspace_services: Some(\"\"), cancellation_token: Some(\"\"), terminal_port: None }" + "ToolRuntimeHandles { workspace_services: Some(\"\"), cancellation_token: Some(\"\"), terminal_port: None, remote_exec_port: None }" ); } } diff --git a/src/crates/execution/runtime-services/src/lib.rs b/src/crates/execution/runtime-services/src/lib.rs index 05f68b1c28..7ae64f6ac9 100644 --- a/src/crates/execution/runtime-services/src/lib.rs +++ b/src/crates/execution/runtime-services/src/lib.rs @@ -4,9 +4,9 @@ use std::sync::Arc; use bitfun_runtime_ports::{ ClockPort, FileSystemPort, GitPort, McpCatalogPort, NetworkPort, PermissionPort, - RemoteCapabilityPort, RemoteConnectionPort, RemoteProjectionPort, RemoteWorkspacePort, - RuntimeEventSink, RuntimeServiceCapability, RuntimeServicePort, SessionStorePort, TerminalPort, - WorkspacePort, + RemoteCapabilityPort, RemoteConnectionPort, RemoteExecPort, RemoteProjectionPort, + RemoteWorkspacePort, RuntimeEventSink, RuntimeServiceCapability, RuntimeServicePort, + SessionStorePort, TerminalPort, WorkspacePort, }; pub mod backend_events; @@ -77,6 +77,7 @@ pub struct RuntimeServices { pub events: Arc, pub clock: Arc, pub terminal: Option>, + pub remote_exec: Option>, pub network: Option>, pub git: Option>, pub mcp_catalog: Option>, @@ -99,6 +100,10 @@ impl std::fmt::Debug for RuntimeServices { "terminal", &self.terminal.as_ref().map(|port| port.capability()), ) + .field( + "remote_exec", + &self.remote_exec.as_ref().map(|port| port.capability()), + ) .field( "network", &self.network.as_ref().map(|port| port.capability()), @@ -147,6 +152,7 @@ impl RuntimeServices { | RuntimeServiceCapability::Events | RuntimeServiceCapability::Clock => true, RuntimeServiceCapability::Terminal => self.terminal.is_some(), + RuntimeServiceCapability::RemoteExec => self.remote_exec.is_some(), RuntimeServiceCapability::Network => self.network.is_some(), RuntimeServiceCapability::Git => self.git.is_some(), RuntimeServiceCapability::McpCatalog => self.mcp_catalog.is_some(), @@ -188,6 +194,7 @@ pub struct RuntimeServicesBuilder { events: Option>, clock: Option>, terminal: Option>, + remote_exec: Option>, network: Option>, git: Option>, mcp_catalog: Option>, @@ -237,6 +244,11 @@ impl RuntimeServicesBuilder { self } + pub fn with_optional_remote_exec(mut self, port: Option>) -> Self { + self.remote_exec = port; + self + } + pub fn with_optional_network(mut self, port: Option>) -> Self { self.network = port; self @@ -302,6 +314,10 @@ impl RuntimeServicesBuilder { events: Self::required(self.events, RuntimeServiceCapability::Events)?, clock: Self::required_service(self.clock, RuntimeServiceCapability::Clock)?, terminal: Self::optional_service(self.terminal, RuntimeServiceCapability::Terminal)?, + remote_exec: Self::optional_service( + self.remote_exec, + RuntimeServiceCapability::RemoteExec, + )?, network: Self::optional_service(self.network, RuntimeServiceCapability::Network)?, git: Self::optional_service(self.git, RuntimeServiceCapability::Git)?, mcp_catalog: Self::optional_service( diff --git a/src/crates/execution/runtime-services/src/test_support.rs b/src/crates/execution/runtime-services/src/test_support.rs index 17b273c2fb..cb53b503a3 100644 --- a/src/crates/execution/runtime-services/src/test_support.rs +++ b/src/crates/execution/runtime-services/src/test_support.rs @@ -4,10 +4,13 @@ use bitfun_runtime_ports::{ ClockPort, FileSystemPort, GitPort, McpCatalogPort, NetworkPort, PermissionDecision, PermissionPort, PermissionRequest, PortError, PortErrorKind, PortResult, RemoteAssistantWorkspaceFacts, RemoteCapabilityPort, RemoteConnectionPort, - RemoteProjectionPort, RemoteRecentWorkspaceFacts, RemoteWorkspaceFacts, - RemoteWorkspaceFileRuntimeHost, RemoteWorkspaceKind, RemoteWorkspacePort, - RemoteWorkspaceRuntimeHost, RemoteWorkspaceUpdate, RuntimeEventEnvelope, RuntimeEventSink, - RuntimeServiceCapability, RuntimeServicePort, SessionStorageKind, SessionStoragePathRequest, + RemoteExecCommandRequest, RemoteExecCommandResponse, RemoteExecControlRequest, + RemoteExecOneShotCommandRequest, RemoteExecOneShotCommandResponse, RemoteExecPort, + RemoteExecStreamingOutputSink, RemoteProjectionPort, RemoteRecentWorkspaceFacts, + RemoteSendStdinRequest, RemoteWorkspaceFacts, RemoteWorkspaceFileRuntimeHost, + RemoteWorkspaceKind, RemoteWorkspacePort, RemoteWorkspaceRuntimeHost, RemoteWorkspaceUpdate, + RemoteWriteStdinRequest, RuntimeEventEnvelope, RuntimeEventSink, RuntimeServiceCapability, + RuntimeServicePort, SessionStorageKind, SessionStoragePathRequest, SessionStoragePathResolution, SessionStorePort, TerminalExecCommandRequest, TerminalExecCommandResponse, TerminalExecControlRequest, TerminalExecStreamingOutputSink, TerminalPort, TerminalSendStdinRequest, TerminalWriteStdinRequest, WorkspacePort, @@ -111,6 +114,68 @@ impl McpCatalogPort for FakeRuntimePort {} impl RemoteConnectionPort for FakeRuntimePort {} impl RemoteCapabilityPort for FakeRuntimePort {} +#[async_trait::async_trait] +impl RemoteExecPort for FakeRuntimePort { + async fn exec_command_once( + &self, + _request: RemoteExecOneShotCommandRequest, + ) -> PortResult { + fake_remote_exec_not_available() + } + + async fn exec_command( + &self, + _request: RemoteExecCommandRequest, + ) -> PortResult { + fake_remote_exec_not_available() + } + + async fn exec_command_streaming( + &self, + _request: RemoteExecCommandRequest, + _output_sink: RemoteExecStreamingOutputSink, + ) -> PortResult { + fake_remote_exec_not_available() + } + + async fn write_stdin( + &self, + _request: RemoteWriteStdinRequest, + ) -> PortResult { + fake_remote_exec_not_available() + } + + async fn write_stdin_streaming( + &self, + _request: RemoteWriteStdinRequest, + _output_sink: RemoteExecStreamingOutputSink, + ) -> PortResult { + fake_remote_exec_not_available() + } + + async fn send_stdin(&self, _request: RemoteSendStdinRequest) -> PortResult<()> { + Err(fake_remote_exec_error()) + } + + async fn control_session( + &self, + _request: RemoteExecControlRequest, + ) -> PortResult { + fake_remote_exec_not_available() + } +} + +fn fake_remote_exec_not_available() -> PortResult { + Err(fake_remote_exec_error()) +} + +fn fake_remote_exec_error() -> PortError { + PortError::new( + PortErrorKind::NotAvailable, + "fake remote exec port does not implement remote execution", + ) +} + #[async_trait::async_trait] impl RemoteWorkspaceRuntimeHost for FakeRuntimePort { async fn current_workspace(&self) -> Option { @@ -200,6 +265,10 @@ impl FakeRuntimeServicesProvider { Arc::new(FakeRuntimePort::new(RuntimeServiceCapability::Terminal)) } + pub fn remote_exec_port() -> Arc { + Arc::new(FakeRuntimePort::new(RuntimeServiceCapability::RemoteExec)) + } + pub fn with_all_remote(mut self) -> Self { self.include_remote = true; self @@ -239,6 +308,8 @@ impl RuntimeServicesProvider for FakeRuntimeServicesProvider { let remote_connection: Arc = Arc::new(FakeRuntimePort::new( RuntimeServiceCapability::RemoteConnection, )); + let remote_exec: Arc = + Arc::new(FakeRuntimePort::new(RuntimeServiceCapability::RemoteExec)); let remote_workspace: Arc = Arc::new(FakeRuntimePort::new( RuntimeServiceCapability::RemoteWorkspace, )); @@ -251,6 +322,7 @@ impl RuntimeServicesProvider for FakeRuntimeServicesProvider { builder .with_optional_remote_connection(Some(remote_connection)) + .with_optional_remote_exec(Some(remote_exec)) .with_optional_remote_workspace(Some(remote_workspace)) .with_optional_remote_projection(Some(remote_projection)) .with_optional_remote_capabilities(Some(remote_capabilities)) diff --git a/src/crates/execution/runtime-services/tests/runtime_services_contracts.rs b/src/crates/execution/runtime-services/tests/runtime_services_contracts.rs index 5f4a6bb2e7..1f514684d5 100644 --- a/src/crates/execution/runtime-services/tests/runtime_services_contracts.rs +++ b/src/crates/execution/runtime-services/tests/runtime_services_contracts.rs @@ -37,6 +37,7 @@ fn fake_provider_registers_required_and_remote_services_through_registry() { assert!(services.has_capability(RuntimeServiceCapability::Events)); assert!(services.has_capability(RuntimeServiceCapability::Clock)); assert!(services.has_capability(RuntimeServiceCapability::RemoteConnection)); + assert!(services.has_capability(RuntimeServiceCapability::RemoteExec)); assert!(services.has_capability(RuntimeServiceCapability::RemoteWorkspace)); assert!(services.has_capability(RuntimeServiceCapability::RemoteProjection)); assert!(services.has_capability(RuntimeServiceCapability::RemoteCapabilities)); @@ -94,6 +95,7 @@ fn marker_ports_register_optional_service_availability_without_core_dependency() .expect("optional ports should satisfy matching capabilities"); assert!(services.has_capability(RuntimeServiceCapability::Terminal)); + assert!(!services.has_capability(RuntimeServiceCapability::RemoteExec)); assert!(services.has_capability(RuntimeServiceCapability::Network)); assert!(services.has_capability(RuntimeServiceCapability::Git)); assert!(services.has_capability(RuntimeServiceCapability::McpCatalog)); diff --git a/src/crates/services/services-integrations/AGENTS.md b/src/crates/services/services-integrations/AGENTS.md index b950c4f117..616e3cf444 100644 --- a/src/crates/services/services-integrations/AGENTS.md +++ b/src/crates/services/services-integrations/AGENTS.md @@ -32,8 +32,8 @@ slices that are outside pure product logic but still platform-neutral. execution remain core-owned unless a reviewed port/provider moves them with equivalence tests. - Remote-SSH path/session identity helpers, disabled surfaces, SSH channels, - SFTP, remote FS, remote terminal, and manager assembly live here behind - explicit remote SSH features. + SFTP, remote FS, remote terminal, remote ExecCommand runtime-port adapter, and + manager assembly live here behind explicit remote SSH features. - Workspace search owns the local flashgrep daemon/session lifecycle and indexed-search result conversion behind `workspace-search`; product config and workspace bootstrap stay in the core facade as injected hooks. diff --git a/src/crates/services/services-integrations/Cargo.toml b/src/crates/services/services-integrations/Cargo.toml index ea886598fd..4a0500dd3d 100644 --- a/src/crates/services/services-integrations/Cargo.toml +++ b/src/crates/services/services-integrations/Cargo.toml @@ -139,6 +139,7 @@ remote-ssh-concrete = [ "anyhow", "async-trait", "base64", + "bitfun-runtime-ports", "chrono", "dirs", "rand", diff --git a/src/crates/services/services-integrations/src/remote_ssh/mod.rs b/src/crates/services/services-integrations/src/remote_ssh/mod.rs index 758cfc2492..ac15ad39d6 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/mod.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/mod.rs @@ -18,6 +18,8 @@ mod password_vault; #[cfg(feature = "remote-ssh-concrete")] mod remote_exec; #[cfg(feature = "remote-ssh-concrete")] +mod remote_exec_runtime_port; +#[cfg(feature = "remote-ssh-concrete")] pub mod remote_fs; #[cfg(feature = "remote-ssh-concrete")] pub mod remote_terminal; @@ -51,6 +53,8 @@ pub use remote_exec::{ RemoteExecSessionCompletionStatus, RemoteSendStdinRequest, RemoteWriteStdinRequest, }; #[cfg(feature = "remote-ssh-concrete")] +pub use remote_exec_runtime_port::{RemoteExecRuntimePort, RemoteExecSshManagerProvider}; +#[cfg(feature = "remote-ssh-concrete")] pub use remote_fs::RemoteFileService; #[cfg(feature = "remote-ssh-concrete")] pub use remote_terminal::{RemoteTerminalManager, RemoteTerminalSession, SessionStatus}; diff --git a/src/crates/services/services-integrations/src/remote_ssh/remote_exec_runtime_port.rs b/src/crates/services/services-integrations/src/remote_ssh/remote_exec_runtime_port.rs new file mode 100644 index 0000000000..093fe0b11a --- /dev/null +++ b/src/crates/services/services-integrations/src/remote_ssh/remote_exec_runtime_port.rs @@ -0,0 +1,339 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use bitfun_runtime_ports as ports; +use tokio::sync::mpsc; + +use super::{ + get_global_remote_exec_process_manager, RemoteExecCommandRequest as ServiceCommandRequest, + RemoteExecCommandResponse as ServiceCommandResponse, + RemoteExecControlAction as ServiceControlAction, + RemoteExecControlOrigin as ServiceControlOrigin, + RemoteExecControlRequest as ServiceControlRequest, RemoteExecError, + RemoteExecProcessLifecycleEvent as ServiceLifecycleEvent, + RemoteExecProcessLifecycleStatus as ServiceLifecycleStatus, RemoteExecProcessManager, + RemoteExecSessionCompletion as ServiceCompletion, + RemoteExecSessionCompletionSource as ServiceCompletionSource, + RemoteExecSessionCompletionStatus as ServiceCompletionStatus, + RemoteSendStdinRequest as ServiceSendStdinRequest, + RemoteWriteStdinRequest as ServiceStdinRequest, SSHCommandOptions, SSHCommandResult, + SSHConnectionManager, +}; + +#[async_trait] +pub trait RemoteExecSshManagerProvider: Send + Sync + std::fmt::Debug { + async fn ssh_manager(&self) -> ports::PortResult; +} + +pub struct RemoteExecRuntimePort { + ssh_manager_provider: Arc, + manager: Arc, +} + +impl RemoteExecRuntimePort { + pub fn new(ssh_manager_provider: Arc) -> Self { + Self { + ssh_manager_provider, + manager: get_global_remote_exec_process_manager(), + } + } + + async fn command_request( + &self, + request: ports::RemoteExecCommandRequest, + ) -> ports::PortResult { + Ok(ServiceCommandRequest { + ssh_manager: self.ssh_manager_provider.ssh_manager().await?, + connection_id: request.connection_id, + command: request.command, + tty: request.tty, + yield_time_ms: request.yield_time_ms, + max_output_chars: request.max_output_chars, + lifecycle_tx: bridge_lifecycle_sink(request.lifecycle_sink), + output_capture_tx: request.output_sink, + }) + } +} + +impl std::fmt::Debug for RemoteExecRuntimePort { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RemoteExecRuntimePort") + .field("ssh_manager_provider", &self.ssh_manager_provider) + .field("manager", &"") + .finish() + } +} + +impl ports::RuntimeServicePort for RemoteExecRuntimePort { + fn capability(&self) -> ports::RuntimeServiceCapability { + ports::RuntimeServiceCapability::RemoteExec + } +} + +#[async_trait] +impl ports::RemoteExecPort for RemoteExecRuntimePort { + async fn exec_command_once( + &self, + request: ports::RemoteExecOneShotCommandRequest, + ) -> ports::PortResult { + let ssh_manager = self.ssh_manager_provider.ssh_manager().await?; + ssh_manager + .execute_command_with_options( + &request.connection_id, + &request.command, + SSHCommandOptions { + timeout_ms: request.timeout_ms, + cancellation_token: None, + }, + ) + .await + .map(one_shot_response_from_service) + .map_err(|error| { + ports::PortError::new(ports::PortErrorKind::Backend, error.to_string()) + }) + } + + async fn exec_command( + &self, + request: ports::RemoteExecCommandRequest, + ) -> ports::PortResult { + let request = self.command_request(request).await?; + self.manager + .exec_command(request) + .await + .map(response_from_service) + .map_err(error_from_service) + } + + async fn exec_command_streaming( + &self, + request: ports::RemoteExecCommandRequest, + output_sink: ports::RemoteExecStreamingOutputSink, + ) -> ports::PortResult { + let request = self.command_request(request).await?; + self.manager + .exec_command_streaming(request, output_sink) + .await + .map(response_from_service) + .map_err(error_from_service) + } + + async fn write_stdin( + &self, + request: ports::RemoteWriteStdinRequest, + ) -> ports::PortResult { + self.manager + .write_stdin(stdin_request_to_service(request)) + .await + .map(response_from_service) + .map_err(error_from_service) + } + + async fn write_stdin_streaming( + &self, + request: ports::RemoteWriteStdinRequest, + output_sink: ports::RemoteExecStreamingOutputSink, + ) -> ports::PortResult { + self.manager + .write_stdin_streaming(stdin_request_to_service(request), output_sink) + .await + .map(response_from_service) + .map_err(error_from_service) + } + + async fn send_stdin(&self, request: ports::RemoteSendStdinRequest) -> ports::PortResult<()> { + self.manager + .send_stdin(ServiceSendStdinRequest { + session_id: request.session_id, + chars: request.chars, + append_enter: request.append_enter, + }) + .await + .map_err(error_from_service) + } + + async fn control_session( + &self, + request: ports::RemoteExecControlRequest, + ) -> ports::PortResult { + self.manager + .control_session(ServiceControlRequest { + session_id: request.session_id, + action: control_action_to_service(request.action), + origin: control_origin_to_service(request.origin), + yield_time_ms: request.yield_time_ms, + max_output_chars: request.max_output_chars, + }) + .await + .map(response_from_service) + .map_err(error_from_service) + } +} + +fn stdin_request_to_service(request: ports::RemoteWriteStdinRequest) -> ServiceStdinRequest { + ServiceStdinRequest { + session_id: request.session_id, + chars: request.chars, + append_enter: request.append_enter, + yield_time_ms: request.yield_time_ms, + max_output_chars: request.max_output_chars, + } +} + +fn one_shot_response_from_service( + response: SSHCommandResult, +) -> ports::RemoteExecOneShotCommandResponse { + ports::RemoteExecOneShotCommandResponse { + stdout: response.stdout, + stderr: response.stderr, + exit_code: response.exit_code, + interrupted: response.interrupted, + timed_out: response.timed_out, + } +} + +fn response_from_service(response: ServiceCommandResponse) -> ports::RemoteExecCommandResponse { + ports::RemoteExecCommandResponse { + chunk_id: response.chunk_id, + wall_time_seconds: response.wall_time_seconds, + output: response.output, + session_id: response.session_id, + exit_code: response.exit_code, + original_output_chars: response.original_output_chars, + completion: response.completion.map(completion_from_service), + } +} + +fn completion_from_service(completion: ServiceCompletion) -> ports::RemoteExecSessionCompletion { + ports::RemoteExecSessionCompletion { + status: match completion.status { + ServiceCompletionStatus::Exited => ports::RemoteExecSessionCompletionStatus::Exited, + ServiceCompletionStatus::Interrupted => { + ports::RemoteExecSessionCompletionStatus::Interrupted + } + ServiceCompletionStatus::Killed => ports::RemoteExecSessionCompletionStatus::Killed, + ServiceCompletionStatus::Pruned => ports::RemoteExecSessionCompletionStatus::Pruned, + }, + source: match completion.source { + ServiceCompletionSource::Process => ports::RemoteExecSessionCompletionSource::Process, + ServiceCompletionSource::OutOfBandControl => { + ports::RemoteExecSessionCompletionSource::OutOfBandControl + } + }, + } +} + +fn control_action_to_service(action: ports::RemoteExecControlAction) -> ServiceControlAction { + match action { + ports::RemoteExecControlAction::Interrupt => ServiceControlAction::Interrupt, + ports::RemoteExecControlAction::Kill => ServiceControlAction::Kill, + } +} + +fn control_origin_to_service(origin: ports::RemoteExecControlOrigin) -> ServiceControlOrigin { + match origin { + ports::RemoteExecControlOrigin::ModelTool => ServiceControlOrigin::ModelTool, + ports::RemoteExecControlOrigin::OutOfBand => ServiceControlOrigin::OutOfBand, + } +} + +fn bridge_lifecycle_sink( + lifecycle_sink: Option, +) -> Option> { + let lifecycle_sink = lifecycle_sink?; + let (tx, mut rx) = mpsc::unbounded_channel::(); + tokio::spawn(async move { + while let Some(event) = rx.recv().await { + let _ = lifecycle_sink.send(lifecycle_event_from_service(event)); + } + }); + Some(tx) +} + +fn lifecycle_event_from_service( + event: ServiceLifecycleEvent, +) -> ports::RemoteExecProcessLifecycleEvent { + ports::RemoteExecProcessLifecycleEvent { + session_id: event.session_id, + status: match event.status { + ServiceLifecycleStatus::Running => ports::RemoteExecProcessLifecycleStatus::Running, + ServiceLifecycleStatus::Exited => ports::RemoteExecProcessLifecycleStatus::Exited, + ServiceLifecycleStatus::Interrupted => { + ports::RemoteExecProcessLifecycleStatus::Interrupted + } + ServiceLifecycleStatus::Killed => ports::RemoteExecProcessLifecycleStatus::Killed, + ServiceLifecycleStatus::Pruned => ports::RemoteExecProcessLifecycleStatus::Pruned, + }, + exit_code: event.exit_code, + } +} + +fn error_from_service(error: RemoteExecError) -> ports::PortError { + match error { + RemoteExecError::SessionNotFound(session_id) => ports::PortError::new( + ports::PortErrorKind::NotFound, + format!("remote exec session not found: {session_id}"), + ), + RemoteExecError::Other(error) => { + ports::PortError::new(ports::PortErrorKind::Backend, error.to_string()) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn one_shot_response_preserves_stdout_stderr_and_timeout_flags() { + let response = one_shot_response_from_service(SSHCommandResult { + stdout: "/bin/bash\n".to_string(), + stderr: "/tmp/noise\n".to_string(), + exit_code: 0, + interrupted: false, + timed_out: true, + }); + + assert_eq!(response.stdout, "/bin/bash\n"); + assert_eq!(response.stderr, "/tmp/noise\n"); + assert_eq!(response.exit_code, 0); + assert!(!response.interrupted); + assert!(response.timed_out); + } + + #[test] + fn completion_and_lifecycle_mapping_preserve_remote_status() { + let completion = completion_from_service(ServiceCompletion { + status: ServiceCompletionStatus::Killed, + source: ServiceCompletionSource::OutOfBandControl, + }); + assert_eq!( + completion.status, + ports::RemoteExecSessionCompletionStatus::Killed + ); + assert_eq!( + completion.source, + ports::RemoteExecSessionCompletionSource::OutOfBandControl + ); + + let lifecycle = lifecycle_event_from_service(ServiceLifecycleEvent { + session_id: 42, + status: ServiceLifecycleStatus::Interrupted, + exit_code: Some(130), + }); + assert_eq!(lifecycle.session_id, 42); + assert_eq!( + lifecycle.status, + ports::RemoteExecProcessLifecycleStatus::Interrupted + ); + assert_eq!(lifecycle.exit_code, Some(130)); + } + + #[test] + fn session_not_found_maps_to_port_not_found() { + let error = error_from_service(RemoteExecError::SessionNotFound(77)); + + assert_eq!(error.kind, ports::PortErrorKind::NotFound); + assert!(error.message.contains("77")); + } +}