robonix_executor/
main.rs

1// SPDX-License-Identifier: MulanPSL-2.0
2// Author: wheatfox <wheatfox17@icloud.com>
3//
4// robonix-executor — capability-call dispatch runtime.
5// On startup executor:
6//   1. Connects to atlas, registers as `com.robonix.system.executor`.
7//   2. Declares its gRPC Execute and CancelAllPlans capabilities.
8//   3. Declares built-in capabilities under `robonix/system/executor/builtin/<op>`
9//      so pilot's atlas-driven discovery surfaces them to the LLM as plain
10//      capabilities. Calls hitting these contracts short-circuit to in-process
11//      handlers in `dispatch::builtin` — no MCP loopback.
12//   4. Serves Execute on `listen`. Per-call dispatch resolves provider via
13//      `ConnectCapability(provider_id, contract_id, MCP)` on atlas.
14
15mod config;
16mod dispatch;
17mod pb;
18mod plan_runtime;
19mod rtdl_wire;
20mod service;
21mod verification;
22
23use anyhow::{Context, Result};
24use clap::Parser;
25use config::{Args, EXECUTOR_NAMESPACE, ExecutorConfig};
26use dispatch::builtin::BUILTINS;
27use pb::contracts::robonix_system_executor_cancel_all_plans_server::RobonixSystemExecutorCancelAllPlansServer;
28use pb::contracts::robonix_system_executor_control_plan_server::RobonixSystemExecutorControlPlanServer;
29use pb::contracts::robonix_system_executor_execute_server::RobonixSystemExecutorExecuteServer;
30use pb::contracts::robonix_system_executor_get_health_server::RobonixSystemExecutorGetHealthServer;
31use pb::contracts::robonix_system_executor_list_active_plans_server::RobonixSystemExecutorListActivePlansServer;
32use robonix_atlas::client::{self as atlas_client, AtlasClient};
33use robonix_atlas::pb as atlas_pb;
34use robonix_scribe::{info, warn};
35use service::ExecutorServiceImpl;
36use std::sync::Arc;
37use std::time::Duration;
38
39#[tokio::main]
40async fn main() -> Result<()> {
41    let parsed = Args::parse();
42    // Apply the manifest's per-component `log:` level (delivered inside
43    // --config-json) to scribe's file sink before the first log line.
44    robonix_scribe::init_from_config("executor", parsed.config_json.as_deref());
45    info!("robonix-executor starting");
46
47    let cfg = ExecutorConfig::resolve(parsed)?;
48
49    info!("connecting to atlas at {}", cfg.atlas_endpoint);
50    let mut atlas =
51        AtlasClient::connect_with_retry(&cfg.atlas_endpoint, 10, Duration::from_secs(2))
52            .await
53            .context("connect to atlas")?;
54
55    atlas
56        .register_service(&cfg.id, EXECUTOR_NAMESPACE, "")
57        .await?;
58    info!("registered as '{}' under '{EXECUTOR_NAMESPACE}'", cfg.id);
59
60    let listen_addr: std::net::SocketAddr = cfg
61        .listen
62        .parse()
63        .with_context(|| format!("invalid executor listen address '{}'", cfg.listen))?;
64    let advertised = match listen_addr.ip() {
65        std::net::IpAddr::V4(ip) if ip.is_unspecified() => {
66            format!("127.0.0.1:{}", listen_addr.port())
67        }
68        _ => listen_addr.to_string(),
69    };
70
71    // Execute RPC: pilot → executor for plan dispatch.
72    atlas
73        .declare_capability(
74            &cfg.id,
75            "robonix/system/executor/execute",
76            atlas_pb::Transport::Grpc,
77            &advertised,
78            atlas_client::grpc_params(
79                "capabilities/system/executor/execute.v1.toml",
80                "robonix.contracts.RobonixSystemExecutorExecute",
81                "/robonix.contracts.RobonixSystemExecutorExecute/Execute",
82            ),
83        )
84        .await?;
85
86    // Out-of-band RTDL meta operations. These never enter PlanRuntime as a
87    // new plan, so canceling work cannot create a self-referential cancel tree.
88    atlas
89        .declare_capability(
90            &cfg.id,
91            "robonix/system/executor/control_plan",
92            atlas_pb::Transport::Grpc,
93            &advertised,
94            atlas_client::grpc_params(
95                "capabilities/system/executor/control_plan.v1.toml",
96                "robonix.contracts.RobonixSystemExecutorControlPlan",
97                "/robonix.contracts.RobonixSystemExecutorControlPlan/ControlPlan",
98            ),
99        )
100        .await?;
101
102    // Read-only control path for clients and observability. Polling it must not
103    // create an RTDL query plan of its own.
104    atlas
105        .declare_capability(
106            &cfg.id,
107            "robonix/system/executor/list_active_plans",
108            atlas_pb::Transport::Grpc,
109            &advertised,
110            atlas_client::grpc_params(
111                "capabilities/system/executor/list_active_plans.v1.toml",
112                "robonix.contracts.RobonixSystemExecutorListActivePlans",
113                "/robonix.contracts.RobonixSystemExecutorListActivePlans/ListActivePlans",
114            ),
115        )
116        .await?;
117
118    // CancelAllPlans RPC: control path for cancelling every active RTDL plan.
119    atlas
120        .declare_capability(
121            &cfg.id,
122            "robonix/system/executor/cancel_all_plans",
123            atlas_pb::Transport::Grpc,
124            &advertised,
125            atlas_client::grpc_params(
126                "capabilities/system/executor/cancel_all_plans.v1.toml",
127                "robonix.contracts.RobonixSystemExecutorCancelAllPlans",
128                "/robonix.contracts.RobonixSystemExecutorCancelAllPlans/CancelAll",
129            ),
130        )
131        .await?;
132
133    // Module health RPC: Vitals polls this for system-module health.
134    atlas
135        .declare_capability(
136            &cfg.id,
137            "robonix/system/executor/get_health",
138            atlas_pb::Transport::Grpc,
139            &advertised,
140            atlas_client::grpc_params(
141                "capabilities/system/executor/get_health.toml",
142                "robonix.contracts.RobonixSystemExecutorGetHealth",
143                "/robonix.contracts.RobonixSystemExecutorGetHealth/GetModuleHealth",
144            ),
145        )
146        .await?;
147
148    // Built-in capabilities: declared as MCP-transport capabilities so pilot's
149    // catalog discovery sees them like any user MCP provider. The endpoint is a
150    // sentinel — dispatch never dials it; calls hitting these contracts hit
151    // the provider_id == self short-circuit in `dispatch::dispatch`.
152    let builtin_endpoint = format!("internal://{}/builtin", cfg.id);
153    for spec in BUILTINS {
154        let contract_id = format!("{EXECUTOR_NAMESPACE}/builtin/{}", spec.op);
155        atlas
156            .declare_capability_with_description(
157                &cfg.id,
158                &contract_id,
159                atlas_pb::Transport::Mcp,
160                &builtin_endpoint,
161                atlas_client::mcp_params(spec.input_schema_json),
162                spec.description,
163            )
164            .await
165            .with_context(|| format!("declare builtin '{}'", contract_id))?;
166    }
167    info!(
168        "declared executor gRPC capabilities + {} builtin capabilities at {advertised}",
169        BUILTINS.len()
170    );
171
172    // Executor has no Driver lifecycle handshake — it's ready as soon as
173    // the gRPC server is up. Push ACTIVE so `rbnx caps` doesn't show the
174    // legacy-fallback INACTIVE forever.
175    if let Err(e) = atlas
176        .set_lifecycle_state(&cfg.id, atlas_pb::LifecycleState::StateActive, "")
177        .await
178    {
179        warn!("SetLifecycleState(ACTIVE) failed: {e:#}");
180    }
181
182    // Atlas evicts providers after ~60s without a heartbeat. Send one every
183    // 20s so we stay registered for the lifetime of the process.
184    {
185        let mut hb = atlas.clone();
186        let provider_id = cfg.id.clone();
187        tokio::spawn(async move {
188            let mut tick = tokio::time::interval(Duration::from_secs(20));
189            tick.tick().await;
190            loop {
191                tick.tick().await;
192                if let Err(e) = hb.heartbeat(&provider_id).await {
193                    warn!("heartbeat failed: {e:#}");
194                }
195            }
196        });
197    }
198
199    let verification = Arc::new(verification::VerificationPolicy::new(cfg.verification));
200    info!(
201        "loaded {} executor verification rule(s)",
202        verification.len()
203    );
204    let svc = ExecutorServiceImpl::new(atlas, cfg.id.clone(), verification);
205    info!("executor gRPC on {listen_addr}");
206    info!("robonix-executor ready on {listen_addr}");
207
208    tonic::transport::Server::builder()
209        .add_service(RobonixSystemExecutorExecuteServer::new(svc.clone()))
210        .add_service(RobonixSystemExecutorCancelAllPlansServer::new(svc.clone()))
211        .add_service(RobonixSystemExecutorControlPlanServer::new(svc.clone()))
212        .add_service(RobonixSystemExecutorListActivePlansServer::new(svc.clone()))
213        .add_service(RobonixSystemExecutorGetHealthServer::new(svc))
214        .serve(listen_addr)
215        .await
216        .context("executor gRPC server failed")?;
217
218    Ok(())
219}
robonix_executor/main.rs

robonix_executor/
main.rs
