From 2b8fa55ab1877ad359b414dc43aa0b2b875d524e Mon Sep 17 00:00:00 2001 From: darmie Date: Fri, 19 Jun 2026 14:43:46 +0100 Subject: [PATCH 1/4] feat(cli): unified `reflow` command-line tool A single `reflow` binary (crates/reflow_cli) over the existing runtime APIs, so graph execution, workspaces, distributed peers, and tracing share one entry point instead of scattered binaries. Commands: - run load a graph, resolve actors (bundled reflow_components catalog + --pack packs, mirroring rfl_template_actor_new), build the Network (Network::with_graph), start, and run until Ctrl-C. --trace / --trace-server / --trace-tail wire tracing + stream live events via the local tap. - graph validate|inspect parse a graph; report nodes/connections/components and which are resolvable. - workspace discover|list|namespaces|analyze|compose|run over WorkspaceDiscovery + GraphComposer (multi_graph). - peer spawn [--send] distributed peer; logic extracted into a new reflow_distributed::peer::run_peer reused by both the `reflow-peer` bin (now a thin shim) and this subcommand. - discovery serve the registry server (reflow_distributed::serve). - trace serve|tail|query|get collector (reflow_tracing::TraceServer) + consumer (TracingClient subscribe/query/get). - serve execs the reflow_server binary (--port; Zeal omitted). The server crate carries an out-of-tree zeal-sdk path so it isn't linked; the CLI spawns it instead. clap 4 derive + tracing_subscriber, matching the house style. Integration tests (graph validate/inspect + a run smoke that loads/resolves/starts) pass. Follow-ups: script-actor resolution in `run` (needs runtime setup), distributed graph composition across peers beyond `peer spawn`, and decoupling reflow_server from zeal so `serve` can link it directly. --- Cargo.lock | 21 ++ crates/reflow_cli/Cargo.toml | 35 ++++ crates/reflow_cli/src/commands/discovery.rs | 39 ++++ crates/reflow_cli/src/commands/graph.rs | 74 +++++++ crates/reflow_cli/src/commands/mod.rs | 7 + crates/reflow_cli/src/commands/peer.rs | 25 +++ crates/reflow_cli/src/commands/run.rs | 45 +++++ crates/reflow_cli/src/commands/serve.rs | 57 ++++++ crates/reflow_cli/src/commands/trace.rs | 160 ++++++++++++++++ crates/reflow_cli/src/commands/workspace.rs | 151 +++++++++++++++ crates/reflow_cli/src/main.rs | 64 +++++++ crates/reflow_cli/src/runtime.rs | 150 +++++++++++++++ crates/reflow_cli/tests/cli.rs | 58 ++++++ .../tests/fixtures/smoke.graph.json | 21 ++ .../reflow_distributed/src/bin/reflow_peer.rs | 181 +----------------- crates/reflow_distributed/src/lib.rs | 1 + crates/reflow_distributed/src/peer.rs | 165 ++++++++++++++++ 17 files changed, 1082 insertions(+), 172 deletions(-) create mode 100644 crates/reflow_cli/Cargo.toml create mode 100644 crates/reflow_cli/src/commands/discovery.rs create mode 100644 crates/reflow_cli/src/commands/graph.rs create mode 100644 crates/reflow_cli/src/commands/mod.rs create mode 100644 crates/reflow_cli/src/commands/peer.rs create mode 100644 crates/reflow_cli/src/commands/run.rs create mode 100644 crates/reflow_cli/src/commands/serve.rs create mode 100644 crates/reflow_cli/src/commands/trace.rs create mode 100644 crates/reflow_cli/src/commands/workspace.rs create mode 100644 crates/reflow_cli/src/main.rs create mode 100644 crates/reflow_cli/src/runtime.rs create mode 100644 crates/reflow_cli/tests/cli.rs create mode 100644 crates/reflow_cli/tests/fixtures/smoke.graph.json create mode 100644 crates/reflow_distributed/src/peer.rs diff --git a/Cargo.lock b/Cargo.lock index c68617d..2fc0192 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6522,6 +6522,27 @@ dependencies = [ "web-sys", ] +[[package]] +name = "reflow_cli" +version = "0.2.0" +dependencies = [ + "anyhow", + "chrono", + "clap", + "flume", + "reflow_distributed", + "reflow_rt", + "reflow_tracing", + "reflow_tracing_protocol", + "serde", + "serde_json", + "tokio", + "toml", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "reflow_components" version = "0.2.1" diff --git a/crates/reflow_cli/Cargo.toml b/crates/reflow_cli/Cargo.toml new file mode 100644 index 0000000..441ef81 --- /dev/null +++ b/crates/reflow_cli/Cargo.toml @@ -0,0 +1,35 @@ +[package] +name = "reflow_cli" +version = "0.2.0" +edition = "2024" +authors = ["Damilare Akinlaja"] +description = "The unified `reflow` command-line tool — run graphs, manage multi-graph workspaces, spawn distributed peers, and drive tracing." +license = "MIT OR Apache-2.0" +repository = "https://github.com/offbit-ai/reflow" +rust-version = "1.85" + +[[bin]] +name = "reflow" +path = "src/main.rs" + +[dependencies] +# Runtime umbrella: network, graph, bundled component catalog, pack loader. +reflow_rt = { path = "../reflow_rt" } +# Distributed peer + discovery server. +reflow_distributed = { path = "../reflow_distributed" } +# Tracing collector (lib surface) + client. +reflow_tracing = { path = "../reflow_tracing" } +reflow_tracing_protocol = { path = "../reflow_tracing_protocol" } + +clap = { version = "4.5", features = ["derive"] } +tokio = { version = "1", features = ["full"] } +anyhow = "1" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +toml = "0.8" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +flume = "0.11" +chrono = "0.4" +uuid = { version = "1", features = ["v4"] } + diff --git a/crates/reflow_cli/src/commands/discovery.rs b/crates/reflow_cli/src/commands/discovery.rs new file mode 100644 index 0000000..5382168 --- /dev/null +++ b/crates/reflow_cli/src/commands/discovery.rs @@ -0,0 +1,39 @@ +//! `reflow discovery serve` — the peer-discovery registry server. + +use anyhow::{Context, Result}; +use clap::Subcommand; +use std::net::SocketAddr; +use std::time::Duration; + +#[derive(Subcommand)] +pub enum DiscoveryCmd { + /// Run the discovery registry HTTP server. + Serve { + /// Address to bind, e.g. 0.0.0.0:9000. + #[arg(long, default_value = "0.0.0.0:9000")] + bind: String, + /// Seconds before an unrefreshed entry expires. + #[arg(long, default_value_t = 60)] + entry_ttl_secs: u64, + /// Seconds between prune sweeps. + #[arg(long, default_value_t = 15)] + prune_interval_secs: u64, + }, +} + +pub async fn run(cmd: DiscoveryCmd) -> Result<()> { + let DiscoveryCmd::Serve { + bind, + entry_ttl_secs, + prune_interval_secs, + } = cmd; + let addr: SocketAddr = bind + .parse() + .with_context(|| format!("invalid --bind address {bind:?}"))?; + reflow_distributed::serve(reflow_distributed::ServerConfig { + bind: addr, + entry_ttl: Duration::from_secs(entry_ttl_secs), + prune_interval: Duration::from_secs(prune_interval_secs), + }) + .await +} diff --git a/crates/reflow_cli/src/commands/graph.rs b/crates/reflow_cli/src/commands/graph.rs new file mode 100644 index 0000000..a2c1c1a --- /dev/null +++ b/crates/reflow_cli/src/commands/graph.rs @@ -0,0 +1,74 @@ +//! `reflow graph validate|inspect `. + +use crate::runtime; +use anyhow::Result; +use clap::Subcommand; +use std::path::PathBuf; + +#[derive(Subcommand)] +pub enum GraphCmd { + /// Parse and validate a graph file; report unresolved components. + Validate { + /// Path to the graph file. + graph: PathBuf, + }, + /// Print a summary of a graph (nodes, connections, components). + Inspect { + /// Path to the graph file. + graph: PathBuf, + }, +} + +fn resolvable(component: &str) -> bool { + reflow_rt::pack_loader::instantiate(component).is_some() + || reflow_rt::components::get_actor_for_template(component).is_some() +} + +pub async fn run(cmd: GraphCmd) -> Result<()> { + match cmd { + GraphCmd::Validate { graph } => { + let export = runtime::load_graph_export(&graph)?; + let components = runtime::component_ids(&export); + println!( + "✓ {} — {} nodes, {} connections, {} components", + graph.display(), + export.processes.len(), + export.connections.len(), + components.len() + ); + let unresolved: Vec<&String> = + components.iter().filter(|c| !resolvable(c)).collect(); + if !unresolved.is_empty() { + println!( + "⚠ {} unresolved component(s) (load with --pack at run time): {}", + unresolved.len(), + unresolved + .iter() + .map(|s| s.as_str()) + .collect::>() + .join(", ") + ); + } + Ok(()) + } + GraphCmd::Inspect { graph } => { + let export = runtime::load_graph_export(&graph)?; + let name = export + .properties + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("(unnamed)"); + println!("graph: {name}"); + println!("nodes ({}):", export.processes.len()); + let mut nodes: Vec<_> = export.processes.iter().collect(); + nodes.sort_by(|a, b| a.0.cmp(b.0)); + for (id, node) in nodes { + let mark = if resolvable(&node.component) { "" } else { " (unresolved)" }; + println!(" {id} [{}]{mark}", node.component); + } + println!("connections: {}", export.connections.len()); + println!("inports: {} outports: {}", export.inports.len(), export.outports.len()); + Ok(()) + } + } +} diff --git a/crates/reflow_cli/src/commands/mod.rs b/crates/reflow_cli/src/commands/mod.rs new file mode 100644 index 0000000..89e60af --- /dev/null +++ b/crates/reflow_cli/src/commands/mod.rs @@ -0,0 +1,7 @@ +pub mod discovery; +pub mod graph; +pub mod peer; +pub mod run; +pub mod serve; +pub mod trace; +pub mod workspace; diff --git a/crates/reflow_cli/src/commands/peer.rs b/crates/reflow_cli/src/commands/peer.rs new file mode 100644 index 0000000..a38ee06 --- /dev/null +++ b/crates/reflow_cli/src/commands/peer.rs @@ -0,0 +1,25 @@ +//! `reflow peer …` — spawn and drive distributed network peers. + +use anyhow::Result; +use clap::Subcommand; +use reflow_distributed::peer::run_peer; +use reflow_distributed::peer_config::PeerConfig; +use std::path::PathBuf; + +#[derive(Subcommand)] +pub enum PeerCmd { + /// Spawn a peer from a TOML config; join its network and stay running. + Spawn { + /// Path to the peer TOML config. + config: PathBuf, + /// One-shot message to send on startup: `network:actor:port:text`. + #[arg(long)] + send: Option, + }, +} + +pub async fn run(cmd: PeerCmd) -> Result<()> { + let PeerCmd::Spawn { config, send } = cmd; + let peer_config = PeerConfig::from_path(&config)?; + run_peer(peer_config, send).await +} diff --git a/crates/reflow_cli/src/commands/run.rs b/crates/reflow_cli/src/commands/run.rs new file mode 100644 index 0000000..c8630ea --- /dev/null +++ b/crates/reflow_cli/src/commands/run.rs @@ -0,0 +1,45 @@ +//! `reflow run ` — load a graph and execute it in-process. + +use crate::runtime::{self, TraceOpts}; +use anyhow::Result; +use clap::Args; +use std::path::PathBuf; + +#[derive(Args)] +pub struct RunArgs { + /// Path to the graph file (GraphExport JSON). + pub graph: PathBuf, + + /// Load an actor pack (.rflpack or dylib). Repeatable. + #[arg(long = "pack", value_name = "PATH")] + pub packs: Vec, + + /// Enable tracing for this run. + #[arg(long)] + pub trace: bool, + + /// Tracing collector URL (implies --trace). Default ws://localhost:8080. + #[arg(long = "trace-server", value_name = "WS_URL")] + pub trace_server: Option, + + /// Stream live trace events to stdout as JSON (implies --trace; uses the + /// local tap, so no collector is required). + #[arg(long = "trace-tail")] + pub trace_tail: bool, +} + +pub async fn run(args: RunArgs) -> Result<()> { + let export = runtime::load_graph_export(&args.graph)?; + runtime::load_packs(&args.packs)?; + let label = args.graph.display().to_string(); + runtime::run_graph_export( + export, + TraceOpts { + enabled: args.trace, + server: args.trace_server, + tail: args.trace_tail, + }, + &label, + ) + .await +} diff --git a/crates/reflow_cli/src/commands/serve.rs b/crates/reflow_cli/src/commands/serve.rs new file mode 100644 index 0000000..a407abc --- /dev/null +++ b/crates/reflow_cli/src/commands/serve.rs @@ -0,0 +1,57 @@ +//! `reflow serve` — run the Reflow HTTP server daemon. +//! +//! The `reflow_server` crate currently carries an out-of-tree `zeal-sdk` path +//! dependency, so the CLI doesn't link it. Instead `serve` execs the +//! `reflow_server` binary (built separately) with Zeal disabled — a thin +//! convenience wrapper. + +use anyhow::{bail, Context, Result}; +use clap::Args; +use std::path::PathBuf; + +#[derive(Args)] +pub struct ServeArgs { + /// Port to listen on. + #[arg(long, default_value_t = 8080)] + pub port: u16, +} + +pub async fn run(args: ServeArgs) -> Result<()> { + let bin = locate_reflow_server(); + tracing::info!("starting {} on port {} (Zeal disabled)", bin.display(), args.port); + let status = tokio::process::Command::new(&bin) + .arg("--port") + .arg(args.port.to_string()) + .status() + .await + .with_context(|| { + format!( + "could not run `{}`. Build it first (cargo build -p reflow_server) \ + or put it on PATH.", + bin.display() + ) + })?; + if !status.success() { + bail!("reflow_server exited unsuccessfully ({status})"); + } + Ok(()) +} + +/// Prefer a `reflow_server` next to the current `reflow` executable; otherwise +/// fall back to the name on `PATH`. +fn locate_reflow_server() -> PathBuf { + let name = if cfg!(windows) { + "reflow_server.exe" + } else { + "reflow_server" + }; + if let Ok(exe) = std::env::current_exe() + && let Some(dir) = exe.parent() + { + let candidate = dir.join(name); + if candidate.exists() { + return candidate; + } + } + PathBuf::from(name) +} diff --git a/crates/reflow_cli/src/commands/trace.rs b/crates/reflow_cli/src/commands/trace.rs new file mode 100644 index 0000000..f44c4bb --- /dev/null +++ b/crates/reflow_cli/src/commands/trace.rs @@ -0,0 +1,160 @@ +//! `reflow trace …` — run the collector or consume traces. + +use anyhow::{Context, Result}; +use clap::Subcommand; +use reflow_tracing_protocol::client::{TracingClient, TracingConfig}; +use reflow_tracing_protocol::{FlowId, SubscriptionFilters, TraceId, TraceQuery}; +use std::path::PathBuf; + +const DEFAULT_SERVER: &str = "ws://localhost:8080"; + +#[derive(Subcommand)] +pub enum TraceCmd { + /// Run the tracing collector server. + Serve { + /// Optional config file (TOML); otherwise uses the standard search path. + #[arg(long)] + config: Option, + }, + /// Stream live trace events from a collector. + Tail { + #[arg(long, default_value = DEFAULT_SERVER)] + server: String, + /// Filter by flow id (repeatable). + #[arg(long = "flow-id")] + flow_ids: Vec, + /// Filter by actor id (repeatable). + #[arg(long = "actor-id")] + actor_ids: Vec, + }, + /// Query historical traces. + Query { + #[arg(long, default_value = DEFAULT_SERVER)] + server: String, + #[arg(long = "flow-id")] + flow_id: Option, + #[arg(long, default_value_t = 100)] + limit: usize, + }, + /// Fetch a single trace by id. + Get { + trace_id: String, + #[arg(long, default_value = DEFAULT_SERVER)] + server: String, + }, +} + +pub async fn run(cmd: TraceCmd) -> Result<()> { + match cmd { + TraceCmd::Serve { config } => serve(config).await, + TraceCmd::Tail { + server, + flow_ids, + actor_ids, + } => tail(server, flow_ids, actor_ids).await, + TraceCmd::Query { + server, + flow_id, + limit, + } => query(server, flow_id, limit).await, + TraceCmd::Get { trace_id, server } => get(server, trace_id).await, + } +} + +async fn serve(config: Option) -> Result<()> { + use reflow_tracing::config::Config; + use reflow_tracing::server::TraceServer; + + let cfg = match config { + Some(path) => { + let text = std::fs::read_to_string(&path) + .with_context(|| format!("reading {}", path.display()))?; + toml::from_str::(&text) + .with_context(|| format!("parsing {}", path.display()))? + } + None => Config::load().unwrap_or_else(|_| { + tracing::warn!("no tracing config found; using defaults"); + Config::default() + }), + }; + + let addr: std::net::SocketAddr = format!("{}:{}", cfg.server.host, cfg.server.port) + .parse() + .context("invalid server host/port")?; + let server = TraceServer::new(cfg).await?; + let listener = tokio::net::TcpListener::bind(addr).await?; + tracing::info!("tracing collector listening on {}", addr); + server.run(listener).await +} + +fn client(server: &str) -> TracingClient { + TracingClient::new(TracingConfig { + server_url: server.to_string(), + ..TracingConfig::default() + }) +} + +async fn tail(server: String, flow_ids: Vec, actor_ids: Vec) -> Result<()> { + let c = client(&server); + c.connect().await.context("connecting to collector")?; + + let (tx, rx) = flume::unbounded(); + c.set_notification_tap(tx); + c.subscribe(SubscriptionFilters { + flow_ids: (!flow_ids.is_empty()).then(|| flow_ids.into_iter().map(FlowId::new).collect()), + actor_ids: (!actor_ids.is_empty()).then_some(actor_ids), + event_types: None, + status_filter: None, + }) + .await + .context("subscribing")?; + + eprintln!("reflow: tailing {server} (Ctrl-C to stop)"); + loop { + tokio::select! { + evt = rx.recv_async() => match evt { + Ok(evt) => { + if let Ok(json) = serde_json::to_string(&evt) { println!("{json}"); } + } + Err(_) => break, + }, + _ = tokio::signal::ctrl_c() => break, + } + } + Ok(()) +} + +async fn query(server: String, flow_id: Option, limit: usize) -> Result<()> { + let c = client(&server); + c.connect().await.context("connecting to collector")?; + let traces = c + .query_traces(TraceQuery { + flow_id: flow_id.map(FlowId::new), + execution_id: None, + time_range: None, + status: None, + actor_filter: None, + limit: Some(limit), + offset: None, + }) + .await + .context("querying traces")?; + for t in &traces { + println!("{}", serde_json::to_string(t)?); + } + eprintln!("reflow: {} trace(s)", traces.len()); + Ok(()) +} + +async fn get(server: String, trace_id: String) -> Result<()> { + let id = TraceId( + uuid::Uuid::parse_str(&trace_id).with_context(|| format!("invalid trace id {trace_id:?}"))?, + ); + let c = client(&server); + c.connect().await.context("connecting to collector")?; + match c.get_trace(id).await.context("get_trace")? { + Some(t) => println!("{}", serde_json::to_string_pretty(&t)?), + None => eprintln!("reflow: no trace {trace_id}"), + } + Ok(()) +} diff --git a/crates/reflow_cli/src/commands/workspace.rs b/crates/reflow_cli/src/commands/workspace.rs new file mode 100644 index 0000000..98464b2 --- /dev/null +++ b/crates/reflow_cli/src/commands/workspace.rs @@ -0,0 +1,151 @@ +//! `reflow workspace …` — discover, analyze, compose, and run multi-graph workspaces. + +use crate::runtime::{self, TraceOpts}; +use anyhow::{anyhow, Result}; +use clap::Subcommand; +use reflow_rt::network::multi_graph::GraphComposer; +use reflow_rt::network::multi_graph::workspace::{ + WorkspaceComposition, WorkspaceConfig, WorkspaceDiscovery, +}; +use std::path::PathBuf; + +#[derive(Subcommand)] +pub enum WorkspaceCmd { + /// Discover graphs in a directory and print graphs + namespaces + analysis. + Discover { dir: PathBuf }, + /// List the graphs discovered in a workspace. + List { dir: PathBuf }, + /// List the namespaces discovered in a workspace. + Namespaces { dir: PathBuf }, + /// Analyze cross-graph dependencies and interfaces. + Analyze { dir: PathBuf }, + /// Compose all discovered graphs into one graph. + Compose { + dir: PathBuf, + /// Write the composed graph JSON here instead of stdout. + #[arg(short, long)] + out: Option, + }, + /// Compose the workspace and run it in-process. + Run { + dir: PathBuf, + /// Enable tracing for the run. + #[arg(long)] + trace: bool, + }, +} + +async fn discover(dir: &PathBuf) -> Result { + let config = WorkspaceConfig { + root_path: dir.clone(), + ..Default::default() + }; + WorkspaceDiscovery::new(config) + .discover_workspace() + .await + .map_err(|e| anyhow!("workspace discovery failed: {e:?}")) +} + +fn graph_name(g: &reflow_rt::graph::types::GraphExport) -> String { + g.properties + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("(unnamed)") + .to_string() +} + +async fn compose(dir: &PathBuf) -> Result { + let wc = discover(dir).await?; + let mut composer = GraphComposer::new(); + let graph = composer + .compose_graphs(wc.composition) + .await + .map_err(|e| anyhow!("composition failed: {e:?}"))?; + Ok(graph.export()) +} + +pub async fn run(cmd: WorkspaceCmd) -> Result<()> { + match cmd { + WorkspaceCmd::Discover { dir } => { + let wc = discover(&dir).await?; + println!("workspace: {}", wc.workspace_root.display()); + println!( + "{} graph(s), {} namespace(s)", + wc.discovered_graphs.len(), + wc.discovered_namespaces.len() + ); + let mut nss: Vec<_> = wc.discovered_namespaces.values().collect(); + nss.sort_by(|a, b| a.namespace.cmp(&b.namespace)); + for info in nss { + println!(" {} — {} graph(s)", info.namespace, info.graph_count); + } + println!( + "dependencies: {}, provided interfaces: {}, required interfaces: {}, auto-connections: {}", + wc.analysis.dependencies.len(), + wc.analysis.exposed_interfaces.len(), + wc.analysis.required_interfaces.len(), + wc.analysis.auto_connections.len() + ); + Ok(()) + } + WorkspaceCmd::List { dir } => { + let wc = discover(&dir).await?; + let mut graphs: Vec<_> = wc.discovered_graphs.iter().collect(); + graphs.sort_by_key(|g| graph_name(&g.graph)); + for g in graphs { + println!( + "{}\t[{}]", + graph_name(&g.graph), + g.workspace_metadata.discovered_namespace + ); + } + Ok(()) + } + WorkspaceCmd::Namespaces { dir } => { + let wc = discover(&dir).await?; + let mut nss: Vec<_> = wc.discovered_namespaces.values().collect(); + nss.sort_by(|a, b| a.namespace.cmp(&b.namespace)); + for info in nss { + println!("{}\t{} graph(s)\t{}", info.namespace, info.graph_count, info.path.display()); + } + Ok(()) + } + WorkspaceCmd::Analyze { dir } => { + let wc = discover(&dir).await?; + let a = &wc.analysis; + println!("dependencies ({}):", a.dependencies.len()); + for d in &a.dependencies { + println!(" {d:?}"); + } + println!("provided interfaces ({}):", a.exposed_interfaces.len()); + println!("required interfaces ({}):", a.required_interfaces.len()); + println!("auto-connections ({}):", a.auto_connections.len()); + Ok(()) + } + WorkspaceCmd::Compose { dir, out } => { + let export = compose(&dir).await?; + let json = serde_json::to_string_pretty(&export)?; + match out { + Some(path) => { + std::fs::write(&path, json)?; + eprintln!("reflow: wrote composed graph to {}", path.display()); + } + None => println!("{json}"), + } + Ok(()) + } + WorkspaceCmd::Run { dir, trace } => { + let export = compose(&dir).await?; + let label = format!("workspace {}", dir.display()); + runtime::run_graph_export( + export, + TraceOpts { + enabled: trace, + ..Default::default() + }, + &label, + ) + .await + } + } +} diff --git a/crates/reflow_cli/src/main.rs b/crates/reflow_cli/src/main.rs new file mode 100644 index 0000000..d3f4e30 --- /dev/null +++ b/crates/reflow_cli/src/main.rs @@ -0,0 +1,64 @@ +//! `reflow` — the unified Reflow command-line tool. +//! +//! Run graphs in-process, manage multi-graph workspaces, spawn distributed +//! peers, drive tracing, and serve the HTTP API — one entry point over the +//! existing runtime APIs. + +mod commands; +mod runtime; + +use clap::{Parser, Subcommand}; + +#[derive(Parser)] +#[command(name = "reflow", version, about = "Run and orchestrate Reflow graphs", long_about = None)] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + /// Load a graph file and execute it in-process. + Run(commands::run::RunArgs), + /// Inspect or validate a graph file. + #[command(subcommand)] + Graph(commands::graph::GraphCmd), + /// Discover and run multi-graph workspaces. + #[command(subcommand)] + Workspace(commands::workspace::WorkspaceCmd), + /// Spawn and drive distributed network peers. + #[command(subcommand)] + Peer(commands::peer::PeerCmd), + /// Run the peer-discovery registry server. + #[command(subcommand)] + Discovery(commands::discovery::DiscoveryCmd), + /// Run the tracing collector or consume traces. + #[command(subcommand)] + Trace(commands::trace::TraceCmd), + /// Run the Reflow HTTP server daemon. + Serve(commands::serve::ServeArgs), +} + +fn init_logging() { + use tracing_subscriber::EnvFilter; + let _ = tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")), + ) + .try_init(); +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + init_logging(); + let cli = Cli::parse(); + match cli.command { + Command::Run(args) => commands::run::run(args).await, + Command::Graph(cmd) => commands::graph::run(cmd).await, + Command::Workspace(cmd) => commands::workspace::run(cmd).await, + Command::Peer(cmd) => commands::peer::run(cmd).await, + Command::Discovery(cmd) => commands::discovery::run(cmd).await, + Command::Trace(cmd) => commands::trace::run(cmd).await, + Command::Serve(args) => commands::serve::run(args).await, + } +} diff --git a/crates/reflow_cli/src/runtime.rs b/crates/reflow_cli/src/runtime.rs new file mode 100644 index 0000000..f3dd8a7 --- /dev/null +++ b/crates/reflow_cli/src/runtime.rs @@ -0,0 +1,150 @@ +//! Shared helpers: load a graph file, resolve its actors against the bundled +//! catalog + loaded packs, and wait for a shutdown signal. + +use anyhow::{bail, Context, Result}; +use std::collections::BTreeSet; +use std::path::Path; + +use reflow_rt::graph::types::GraphExport; +use reflow_rt::graph::Graph; +use reflow_rt::network::network::{Network, NetworkConfig}; + +/// Tracing options for a run. +#[derive(Default)] +pub struct TraceOpts { + pub enabled: bool, + pub server: Option, + pub tail: bool, +} + +impl TraceOpts { + fn any(&self) -> bool { + self.enabled || self.tail || self.server.is_some() + } +} + +/// Read and parse a graph JSON file into a `GraphExport`. Ensures a `name` +/// property exists (defaulted from the file name) since `Graph::load` requires +/// it. +pub fn load_graph_export(path: &Path) -> Result { + let text = std::fs::read_to_string(path) + .with_context(|| format!("reading graph file {}", path.display()))?; + let mut export: GraphExport = serde_json::from_str(&text) + .with_context(|| format!("parsing {} as a Reflow graph (GraphExport JSON)", path.display()))?; + if !export.properties.contains_key("name") { + let name = path + .file_stem() + .and_then(|s| s.to_str()) + .unwrap_or("graph") + .to_string(); + export + .properties + .insert("name".into(), serde_json::Value::String(name)); + } + Ok(export) +} + +/// Distinct component (template) ids referenced by a graph's nodes. +pub fn component_ids(export: &GraphExport) -> Vec { + export + .processes + .values() + .map(|n| n.component.clone()) + .collect::>() + .into_iter() + .collect() +} + +/// Load actor packs from the given paths into the process-wide registry. +pub fn load_packs(packs: &[std::path::PathBuf]) -> Result<()> { + for path in packs { + let ids = reflow_rt::pack_loader::load_pack(path) + .with_context(|| format!("loading actor pack {}", path.display()))?; + tracing::info!("loaded pack {} ({} templates)", path.display(), ids.len()); + } + Ok(()) +} + +/// Resolve every component the graph references and register it into the +/// network: loaded packs first, then the bundled `reflow_components` catalog. +/// Returns an error naming any component that couldn't be resolved. +pub fn resolve_and_register(net: &mut Network, export: &GraphExport) -> Result<()> { + let mut unresolved = Vec::new(); + for comp in component_ids(export) { + let actor = reflow_rt::pack_loader::instantiate(&comp) + .or_else(|| reflow_rt::components::get_actor_for_template(&comp)); + match actor { + // `register_actor_arc` errors if a template is already registered + // (e.g. by a pack); that's fine — ignore the duplicate. + Some(actor) => { + let _ = net.register_actor_arc(&comp, actor); + } + None => unresolved.push(comp), + } + } + if !unresolved.is_empty() { + let available = reflow_rt::components::get_template_mapping(); + let mut sample: Vec<&String> = available.keys().take(20).collect(); + sample.sort(); + bail!( + "could not resolve {} component(s): {}\n\ + Load the actor pack that provides them with --pack , or check the ids.\n\ + {} bundled templates are available (e.g. {}).", + unresolved.len(), + unresolved.join(", "), + available.len(), + sample + .iter() + .map(|s| s.as_str()) + .collect::>() + .join(", ") + ); + } + Ok(()) +} + +/// Block until Ctrl-C. +pub async fn wait_for_ctrl_c() { + let _ = tokio::signal::ctrl_c().await; +} + +/// Build a network from a graph export, resolve its actors, start it, and run +/// until Ctrl-C. Shared by `reflow run` and `reflow workspace run`. +pub async fn run_graph_export(export: GraphExport, trace: TraceOpts, label: &str) -> Result<()> { + let mut config = NetworkConfig::default(); + config.tracing.enabled = trace.any(); + if let Some(url) = &trace.server { + config.tracing.server_url = url.clone(); + } + + let graph = Graph::load(export.clone(), None); + let network = Network::with_graph(config, &graph); + + if trace.tail { + let rx = network.lock().unwrap().get_trace_receiver(); + tokio::spawn(async move { + while let Ok(evt) = rx.recv_async().await { + if let Ok(json) = serde_json::to_string(&evt) { + println!("{json}"); + } + } + }); + } + + { + let mut net = network.lock().unwrap(); + resolve_and_register(&mut net, &export)?; + net.start()?; + } + tracing::info!( + "started {label} — {} nodes, {} connections", + export.processes.len(), + export.connections.len() + ); + eprintln!("reflow: running — press Ctrl-C to stop"); + + wait_for_ctrl_c().await; + eprintln!("reflow: shutting down…"); + network.lock().unwrap().shutdown(); + Ok(()) +} diff --git a/crates/reflow_cli/tests/cli.rs b/crates/reflow_cli/tests/cli.rs new file mode 100644 index 0000000..7ea9992 --- /dev/null +++ b/crates/reflow_cli/tests/cli.rs @@ -0,0 +1,58 @@ +//! Integration tests for the `reflow` binary, driven as a subprocess. + +use std::process::{Command, Stdio}; +use std::time::Duration; + +const BIN: &str = env!("CARGO_BIN_EXE_reflow"); + +fn fixture() -> String { + format!("{}/tests/fixtures/smoke.graph.json", env!("CARGO_MANIFEST_DIR")) +} + +#[test] +fn graph_validate_reports_nodes_and_components() { + let out = Command::new(BIN) + .args(["graph", "validate", &fixture()]) + .output() + .expect("run reflow graph validate"); + assert!(out.status.success(), "validate failed: {:?}", out); + let stdout = String::from_utf8_lossy(&out.stdout); + assert!(stdout.contains("1 nodes"), "unexpected output: {stdout}"); + // tpl_signal resolves against the bundled catalog, so no "unresolved" warning. + assert!(!stdout.contains("unresolved"), "unexpected unresolved: {stdout}"); +} + +#[test] +fn graph_inspect_lists_the_component() { + let out = Command::new(BIN) + .args(["graph", "inspect", &fixture()]) + .output() + .expect("run reflow graph inspect"); + assert!(out.status.success()); + let stdout = String::from_utf8_lossy(&out.stdout); + assert!(stdout.contains("tpl_signal"), "unexpected output: {stdout}"); +} + +#[test] +fn run_loads_resolves_and_starts() { + let mut child = Command::new(BIN) + .args(["run", &fixture()]) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn reflow run"); + + // Give it a beat to load, resolve actors against the catalog, and start. + std::thread::sleep(Duration::from_millis(1500)); + let still_running = child.try_wait().expect("try_wait").is_none(); + + let _ = child.kill(); + let out = child.wait_with_output().expect("collect output"); + let stderr = String::from_utf8_lossy(&out.stderr); + + assert!(still_running, "process exited early; stderr:\n{stderr}"); + assert!( + stderr.contains("running") || stderr.contains("started"), + "did not see a start message; stderr:\n{stderr}" + ); +} diff --git a/crates/reflow_cli/tests/fixtures/smoke.graph.json b/crates/reflow_cli/tests/fixtures/smoke.graph.json new file mode 100644 index 0000000..58f4dc9 --- /dev/null +++ b/crates/reflow_cli/tests/fixtures/smoke.graph.json @@ -0,0 +1,21 @@ +{ + "caseSensitive": false, + "properties": { + "name": "cli-smoke" + }, + "processes": { + "s": { + "id": "s", + "component": "tpl_signal", + "metadata": null + } + }, + "connections": [], + "inports": {}, + "outports": {}, + "groups": [], + "graph_dependencies": [], + "external_connections": [], + "provided_interfaces": {}, + "required_interfaces": {} +} diff --git a/crates/reflow_distributed/src/bin/reflow_peer.rs b/crates/reflow_distributed/src/bin/reflow_peer.rs index a532763..5a2f84a 100644 --- a/crates/reflow_distributed/src/bin/reflow_peer.rs +++ b/crates/reflow_distributed/src/bin/reflow_peer.rs @@ -1,35 +1,13 @@ -//! `reflow-peer` — runs a single member of a distributed Reflow -//! network from a TOML config file. -//! -//! What it does on startup: -//! 1. Loads `peer.toml` — bind address, discovery endpoints, -//! auth token, peer connect list, remote-actor proxies. -//! 2. Builds a `DistributedNetwork` and starts the bridge. -//! 3. Registers a built-in `recorder` actor on the local -//! network. Every message landing on `recorder.in` is logged -//! via `tracing` — useful as a smoke target without writing -//! any glue code. -//! 4. Dials each `[[connect]]` entry. -//! 5. Registers each `[[remote_actor]]` proxy. -//! 6. Idles until Ctrl-C. -//! -//! With two peers + one `reflow-discovery` server, this binary -//! alone is enough to demonstrate federation end-to-end. +//! `reflow-peer` — runs a single member of a distributed Reflow network from a +//! TOML config file. A thin shim over [`reflow_distributed::peer::run_peer`]; +//! the same logic is reachable via `reflow peer spawn`. -use std::collections::HashMap; use std::path::PathBuf; -use std::sync::Arc; -use anyhow::{Context, Result}; +use anyhow::Result; use clap::Parser; -use futures::StreamExt; -use reflow_actor::{ - Actor, ActorBehavior, ActorConfig, ActorContext, ActorLoad, MemoryState, Port, - message::Message, -}; +use reflow_distributed::peer::run_peer; use reflow_distributed::peer_config::PeerConfig; -use reflow_network::distributed_network::DistributedNetwork; -use reflow_network::tracing::TracingIntegration; use tracing_subscriber::EnvFilter; #[derive(Parser, Debug)] @@ -39,9 +17,8 @@ struct Args { #[arg(short, long)] config: PathBuf, - /// One-shot test send: `:::`. - /// Fires after federation is up; the peer keeps running - /// afterward. + /// One-shot test send: `:::`. Fires after + /// federation is up; the peer keeps running afterward. #[arg(long)] send: Option, } @@ -55,146 +32,6 @@ async fn main() -> Result<()> { .init(); let args = Args::parse(); - let cfg = PeerConfig::from_path(&args.config)?; - - tracing::info!( - "starting peer: network_id={}, instance_id={}, bind={}:{}", - cfg.network_id, - cfg.instance_id, - cfg.bind_address, - cfg.bind_port, - ); - - let dist_cfg = cfg.to_distributed_config(); - let mut net = DistributedNetwork::new(dist_cfg).await?; - - // Built-in recorder actor — logs every inbound message. Useful - // as a smoke-test target without anyone having to write code. - net.register_local_actor("recorder", RecorderActor::new(), None) - .context("register recorder")?; - - net.start().await?; - - // Brief settle window — bridge needs a beat to start accepting - // before we dial outwards. - tokio::time::sleep(std::time::Duration::from_millis(150)).await; - - for connect in &cfg.connect { - match net.connect_to_network(&connect.endpoint).await { - Ok(()) => tracing::info!("connected to peer at {}", connect.endpoint), - Err(e) => tracing::warn!("failed to connect to {}: {}", connect.endpoint, e), - } - } - - for ra in &cfg.remote_actor { - match net.register_remote_actor(&ra.actor_id, &ra.network_id).await { - Ok(()) => tracing::info!( - "registered remote actor proxy: {}@{}", - ra.actor_id, - ra.network_id - ), - Err(e) => tracing::warn!( - "failed to register {}@{}: {}", - ra.actor_id, - ra.network_id, - e - ), - } - } - - if let Some(spec) = args.send { - send_one_shot(&net, &spec).await?; - } - - tracing::info!("peer ready, awaiting messages (Ctrl-C to stop)"); - tokio::signal::ctrl_c().await?; - tracing::info!("shutting down"); - net.shutdown().await?; - Ok(()) -} - -async fn send_one_shot(net: &DistributedNetwork, spec: &str) -> Result<()> { - // Format: ::: - let mut parts = spec.splitn(4, ':'); - let network = parts.next().context("--send: missing network_id")?; - let actor = parts.next().context("--send: missing actor_id")?; - let port = parts.next().context("--send: missing port")?; - let text = parts - .next() - .context("--send: missing text payload (after the third ':')")?; - - let payload = Message::String(Arc::new(text.to_string())); - net.send_to_remote_actor(network, actor, port, payload, None) - .await - .with_context(|| format!("send_to_remote_actor: {network}:{actor}:{port}"))?; - tracing::info!("sent test message to {network}/{actor}.{port}: {text:?}"); - Ok(()) -} - -// ─── built-in recorder actor ────────────────────────────────────── - -struct RecorderActor { - inports: Port, - outports: Port, - load: Arc, -} - -impl RecorderActor { - fn new() -> Self { - Self { - inports: flume::unbounded(), - outports: flume::unbounded(), - load: Arc::new(ActorLoad::new(0)), - } - } -} - -impl Actor for RecorderActor { - fn get_behavior(&self) -> ActorBehavior { - Box::new(move |context: ActorContext| { - Box::pin(async move { - let payload = context.get_payload(); - for (port, msg) in payload.iter() { - tracing::info!("recorder.{port}: {msg:?}"); - } - Ok(HashMap::new()) - }) - }) - } - fn get_inports(&self) -> Port { - self.inports.clone() - } - fn get_outports(&self) -> Port { - self.outports.clone() - } - fn load_count(&self) -> Arc { - self.load.clone() - } - fn create_instance(&self) -> Arc { - Arc::new(Self::new()) - } - fn create_process( - &self, - actor_config: ActorConfig, - _tracing: Option, - ) -> std::pin::Pin + Send + 'static>> { - let behavior = self.get_behavior(); - let (_, receiver) = self.get_inports(); - let outports = self.get_outports(); - let load = self.load_count(); - - Box::pin(async move { - while let Some(packet) = receiver.stream().next().await { - let context = ActorContext::new( - packet, - outports.clone(), - Arc::new(parking_lot::Mutex::new(MemoryState::default())), - actor_config.clone(), - load.clone(), - ); - let _ = behavior(context).await; - load.reset(); - } - }) - } + let config = PeerConfig::from_path(&args.config)?; + run_peer(config, args.send).await } diff --git a/crates/reflow_distributed/src/lib.rs b/crates/reflow_distributed/src/lib.rs index f3258fb..d9a4465 100644 --- a/crates/reflow_distributed/src/lib.rs +++ b/crates/reflow_distributed/src/lib.rs @@ -30,6 +30,7 @@ use axum::{ use parking_lot::RwLock; use reflow_network::discovery::{NetworkInfo, RegistrationRequest}; +pub mod peer; pub mod peer_config; /// In-memory registry. Entries decay if the peer hasn't pinged in diff --git a/crates/reflow_distributed/src/peer.rs b/crates/reflow_distributed/src/peer.rs new file mode 100644 index 0000000..b8c7626 --- /dev/null +++ b/crates/reflow_distributed/src/peer.rs @@ -0,0 +1,165 @@ +//! Runs a single member of a distributed Reflow network from a [`PeerConfig`]. +//! +//! Extracted so both the `reflow-peer` binary and the unified `reflow peer +//! spawn` CLI drive the exact same logic. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, Result}; +use futures::StreamExt; +use reflow_actor::{ + Actor, ActorBehavior, ActorConfig, ActorContext, ActorLoad, MemoryState, Port, + message::Message, +}; +use reflow_network::distributed_network::DistributedNetwork; +use reflow_network::tracing::TracingIntegration; + +use crate::peer_config::PeerConfig; + +/// Bring up a peer: build the `DistributedNetwork`, register a built-in +/// `recorder` actor, start the bridge, dial `[[connect]]` entries, register +/// `[[remote_actor]]` proxies, optionally fire a one-shot `send`, then idle +/// until Ctrl-C. +pub async fn run_peer(config: PeerConfig, send: Option) -> Result<()> { + tracing::info!( + "starting peer: network_id={}, instance_id={}, bind={}:{}", + config.network_id, + config.instance_id, + config.bind_address, + config.bind_port, + ); + + let mut net = DistributedNetwork::new(config.to_distributed_config()).await?; + + // Built-in recorder actor — logs every inbound message. A smoke-test target + // without anyone having to write code. + net.register_local_actor("recorder", RecorderActor::new(), None) + .context("register recorder")?; + + net.start().await?; + + // Brief settle window — the bridge needs a beat to start accepting before + // we dial outwards. + tokio::time::sleep(Duration::from_millis(150)).await; + + for connect in &config.connect { + match net.connect_to_network(&connect.endpoint).await { + Ok(()) => tracing::info!("connected to peer at {}", connect.endpoint), + Err(e) => tracing::warn!("failed to connect to {}: {}", connect.endpoint, e), + } + } + + for ra in &config.remote_actor { + match net.register_remote_actor(&ra.actor_id, &ra.network_id).await { + Ok(()) => tracing::info!( + "registered remote actor proxy: {}@{}", + ra.actor_id, + ra.network_id + ), + Err(e) => tracing::warn!( + "failed to register {}@{}: {}", + ra.actor_id, + ra.network_id, + e + ), + } + } + + if let Some(spec) = send { + send_one_shot(&net, &spec).await?; + } + + tracing::info!("peer ready, awaiting messages (Ctrl-C to stop)"); + tokio::signal::ctrl_c().await?; + tracing::info!("shutting down"); + net.shutdown().await?; + Ok(()) +} + +/// Fire a one-shot message: `:::`. +pub async fn send_one_shot(net: &DistributedNetwork, spec: &str) -> Result<()> { + let mut parts = spec.splitn(4, ':'); + let network = parts.next().context("--send: missing network_id")?; + let actor = parts.next().context("--send: missing actor_id")?; + let port = parts.next().context("--send: missing port")?; + let text = parts + .next() + .context("--send: missing text payload (after the third ':')")?; + + let payload = Message::String(Arc::new(text.to_string())); + net.send_to_remote_actor(network, actor, port, payload, None) + .await + .with_context(|| format!("send_to_remote_actor: {network}:{actor}:{port}"))?; + tracing::info!("sent test message to {network}/{actor}.{port}: {text:?}"); + Ok(()) +} + +// ─── built-in recorder actor ────────────────────────────────────── + +struct RecorderActor { + inports: Port, + outports: Port, + load: Arc, +} + +impl RecorderActor { + fn new() -> Self { + Self { + inports: flume::unbounded(), + outports: flume::unbounded(), + load: Arc::new(ActorLoad::new(0)), + } + } +} + +impl Actor for RecorderActor { + fn get_behavior(&self) -> ActorBehavior { + Box::new(move |context: ActorContext| { + Box::pin(async move { + let payload = context.get_payload(); + for (port, msg) in payload.iter() { + tracing::info!("recorder.{port}: {msg:?}"); + } + Ok(HashMap::new()) + }) + }) + } + fn get_inports(&self) -> Port { + self.inports.clone() + } + fn get_outports(&self) -> Port { + self.outports.clone() + } + fn load_count(&self) -> Arc { + self.load.clone() + } + fn create_instance(&self) -> Arc { + Arc::new(Self::new()) + } + fn create_process( + &self, + actor_config: ActorConfig, + _tracing: Option, + ) -> std::pin::Pin + Send + 'static>> { + let behavior = self.get_behavior(); + let (_, receiver) = self.get_inports(); + let outports = self.get_outports(); + let load = self.load_count(); + + Box::pin(async move { + while let Some(packet) = receiver.stream().next().await { + let context = ActorContext::new( + packet, + outports.clone(), + Arc::new(parking_lot::Mutex::new(MemoryState::default())), + actor_config.clone(), + load.clone(), + ); + let _ = behavior(context).await; + load.reset(); + } + }) + } +} From 2ba69ed351e8d03035039d4c6005efd00085abb2 Mon Sep 17 00:00:00 2001 From: darmie Date: Fri, 19 Jun 2026 21:30:46 +0100 Subject: [PATCH 2/4] feat(cli,server): run `reflow serve` in-process; make Zeal optional The CLI was the only thing left fragmented: `serve` shelled out to a separate `reflow_server` binary because the server crate couldn't be linked. The blocker was its out-of-tree `zeal-sdk` path dependency. This makes Zeal an opt-in feature so the server core links cleanly into the unified `reflow` binary. reflow_server: - `zeal-sdk` is now `optional = true` behind a `zeal` feature, `default = ["zeal"]` so the standalone `reflow_server` binary and existing Zeal users are unchanged. - Feature-gate the Zeal-only modules (zeal_converter, zip_session, trace_collector, template_adapter), `start_zeal_execution`, the `/zeal/*` routes + handlers, and the Zeal branch of `start_server`. `EventBridge` stays in both configs: without `zeal`, `attach` just drains the event channel so the engine never blocks. - Self-contained manifest (concrete serde/serde_json/anyhow versions instead of `workspace = true`): the crate is excluded from the workspace, so it must resolve standalone when linked as a path dep. - Fix a startup panic present in both configs: the `/webhook/{id}/{path:.*}` route used actix/regex syntax axum 0.7 rejects ("only one parameter per path segment"), so the server panicked on boot. matchit 0.7 also won't accept a catch-all sibling of a param route, and the route never functioned, so it's removed; `/webhook/{id}` is unaffected. reflow_cli: - Depend on `reflow_server` with `default-features = false` (Zeal omitted). - `reflow serve` now calls `reflow_server::start_server` in-process, mapping `--port/--bind/--redis` onto `ServerConfig`. No subprocess, no separate binary. Verified: `reflow serve` answers GET /health 200; reflow_server builds with and without `zeal`; reflow_cli tests pass; `cargo metadata` resolves. Note: linking reflow_server pulls its `zeal-sdk` path into the resolve graph, so `cargo metadata`/maturin now need the sibling Zeal checkout. Follow-up if the Python-SDK CI must build without it: vendor zeal-sdk or make it a git dep. --- Cargo.lock | 159 ++++++++++++++++++++++- crates/reflow_cli/Cargo.toml | 3 + crates/reflow_cli/src/commands/serve.rs | 71 +++++----- crates/reflow_server/Cargo.toml | 17 ++- crates/reflow_server/src/engine.rs | 2 + crates/reflow_server/src/event_bridge.rs | 35 ++++- crates/reflow_server/src/lib.rs | 20 ++- crates/reflow_server/src/rest_api.rs | 52 ++++++-- 8 files changed, 287 insertions(+), 72 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2fc0192..e5985ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -483,6 +483,7 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", "axum-core", + "base64 0.22.1", "bytes 1.10.1", "futures-util", "http 1.3.1", @@ -501,8 +502,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper 1.0.2", "tokio", + "tokio-tungstenite 0.24.0", "tower 0.5.2", "tower-layer", "tower-service", @@ -3180,6 +3183,25 @@ dependencies = [ "tracing", ] +[[package]] +name = "h2" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +dependencies = [ + "atomic-waker", + "bytes 1.10.1", + "fnv", + "futures-core", + "futures-sink", + "http 1.3.1", + "indexmap 2.10.0", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "half" version = "2.6.0" @@ -3493,7 +3515,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "h2", + "h2 0.3.26", "http 0.2.12", "http-body 0.4.6", "httparse", @@ -3516,6 +3538,7 @@ dependencies = [ "bytes 1.10.1", "futures-channel", "futures-util", + "h2 0.4.15", "http 1.3.1", "http-body 1.0.1", "httparse", @@ -3557,6 +3580,22 @@ dependencies = [ "tokio-native-tls", ] +[[package]] +name = "hyper-tls" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0" +dependencies = [ + "bytes 1.10.1", + "http-body-util", + "hyper 1.6.0", + "hyper-util", + "native-tls", + "tokio", + "tokio-native-tls", + "tower-service", +] + [[package]] name = "hyper-tungstenite" version = "0.11.1" @@ -3589,9 +3628,11 @@ dependencies = [ "percent-encoding", "pin-project-lite", "socket2 0.5.10", + "system-configuration 0.6.1", "tokio", "tower-service", "tracing", + "windows-registry 0.5.3", ] [[package]] @@ -3919,7 +3960,7 @@ checksum = "4d40460c0ce33d6ce4b0630ad68ff63d6661961c48b6dba35e5a4d81cfb48222" dependencies = [ "socket2 0.6.4", "widestring", - "windows-registry", + "windows-registry 0.6.1", "windows-result 0.4.1", "windows-sys 0.61.2", ] @@ -6154,7 +6195,7 @@ dependencies = [ "quinn-udp", "rustc-hash 2.1.1", "rustls 0.23.28", - "socket2 0.5.10", + "socket2 0.6.4", "thiserror 2.0.12", "tokio", "tracing", @@ -6191,7 +6232,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2 0.6.4", "tracing", "windows-sys 0.60.2", ] @@ -6532,6 +6573,7 @@ dependencies = [ "flume", "reflow_distributed", "reflow_rt", + "reflow_server", "reflow_tracing", "reflow_tracing_protocol", "serde", @@ -7110,6 +7152,34 @@ dependencies = [ "serde_json", ] +[[package]] +name = "reflow_server" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum", + "chrono", + "dashmap 6.1.0", + "env_logger", + "flume", + "futures 0.3.31", + "futures-util", + "log", + "redis", + "reflow_actor_macro", + "reflow_components", + "reflow_graph", + "reflow_network", + "reqwest 0.12.22", + "serde", + "serde_json", + "tokio", + "tokio-tungstenite 0.24.0", + "tower 0.4.13", + "tower-http 0.5.2", + "uuid", +] + [[package]] name = "reflow_shader" version = "0.2.1" @@ -7327,11 +7397,11 @@ dependencies = [ "encoding_rs", "futures-core", "futures-util", - "h2", + "h2 0.3.26", "http 0.2.12", "http-body 0.4.6", "hyper 0.14.32", - "hyper-tls", + "hyper-tls 0.5.0", "ipnet", "js-sys", "log", @@ -7366,16 +7436,21 @@ checksum = "cbc931937e6ca3a06e3b6c0aa7841849b160a90351d6ab467a8b9b9959767531" dependencies = [ "base64 0.22.1", "bytes 1.10.1", + "encoding_rs", "futures-core", "futures-util", + "h2 0.4.15", "http 1.3.1", "http-body 1.0.1", "http-body-util", "hyper 1.6.0", "hyper-rustls", + "hyper-tls 0.6.0", "hyper-util", "js-sys", "log", + "mime", + "native-tls", "percent-encoding", "pin-project-lite", "quinn", @@ -7386,6 +7461,7 @@ dependencies = [ "serde_urlencoded", "sync_wrapper 1.0.2", "tokio", + "tokio-native-tls", "tokio-rustls 0.26.2", "tokio-util", "tower 0.5.2", @@ -8960,6 +9036,17 @@ dependencies = [ "system-configuration-sys 0.5.0", ] +[[package]] +name = "system-configuration" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b" +dependencies = [ + "bitflags 2.9.1", + "core-foundation 0.9.4", + "system-configuration-sys 0.6.0", +] + [[package]] name = "system-configuration" version = "0.7.0" @@ -9466,6 +9553,20 @@ dependencies = [ "tungstenite 0.20.1", ] +[[package]] +name = "tokio-tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" +dependencies = [ + "futures-util", + "log", + "native-tls", + "tokio", + "tokio-native-tls", + "tungstenite 0.24.0", +] + [[package]] name = "tokio-tungstenite" version = "0.26.2" @@ -9606,6 +9707,22 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e9cd434a998747dd2c4276bc96ee2e0c7a2eadf3cae88e52be55a05fa9053f5" +dependencies = [ + "bitflags 2.9.1", + "bytes 1.10.1", + "http 1.3.1", + "http-body 1.0.1", + "http-body-util", + "pin-project-lite", + "tower-layer", + "tower-service", +] + [[package]] name = "tower-http" version = "0.6.6" @@ -9804,6 +9921,25 @@ dependencies = [ "utf-8", ] +[[package]] +name = "tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a" +dependencies = [ + "byteorder", + "bytes 1.10.1", + "data-encoding", + "http 1.3.1", + "httparse", + "log", + "native-tls", + "rand 0.8.5", + "sha1", + "thiserror 1.0.69", + "utf-8", +] + [[package]] name = "tungstenite" version = "0.26.2" @@ -11102,6 +11238,17 @@ dependencies = [ "windows-link 0.2.1", ] +[[package]] +name = "windows-registry" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b8a9ed28765efc97bbc954883f4e6796c33a06546ebafacbabee9696967499e" +dependencies = [ + "windows-link 0.1.3", + "windows-result 0.3.4", + "windows-strings 0.4.2", +] + [[package]] name = "windows-registry" version = "0.6.1" diff --git a/crates/reflow_cli/Cargo.toml b/crates/reflow_cli/Cargo.toml index 441ef81..dedd2e3 100644 --- a/crates/reflow_cli/Cargo.toml +++ b/crates/reflow_cli/Cargo.toml @@ -20,6 +20,9 @@ reflow_distributed = { path = "../reflow_distributed" } # Tracing collector (lib surface) + client. reflow_tracing = { path = "../reflow_tracing" } reflow_tracing_protocol = { path = "../reflow_tracing_protocol" } +# HTTP server core, linked in-process. Zeal is omitted (default-features = false) +# so the out-of-tree zeal-sdk isn't pulled in. +reflow_server = { path = "../reflow_server", default-features = false } clap = { version = "4.5", features = ["derive"] } tokio = { version = "1", features = ["full"] } diff --git a/crates/reflow_cli/src/commands/serve.rs b/crates/reflow_cli/src/commands/serve.rs index a407abc..b26f021 100644 --- a/crates/reflow_cli/src/commands/serve.rs +++ b/crates/reflow_cli/src/commands/serve.rs @@ -1,57 +1,44 @@ -//! `reflow serve` — run the Reflow HTTP server daemon. +//! `reflow serve` — run the Reflow HTTP server in-process. //! -//! The `reflow_server` crate currently carries an out-of-tree `zeal-sdk` path -//! dependency, so the CLI doesn't link it. Instead `serve` execs the -//! `reflow_server` binary (built separately) with Zeal disabled — a thin -//! convenience wrapper. +//! Links `reflow_server` directly (with `default-features = false`, so the +//! out-of-tree Zeal SDK is omitted) and calls `start_server`. No separate +//! binary, no subprocess — one self-contained `reflow` tool. -use anyhow::{bail, Context, Result}; +use anyhow::Result; use clap::Args; -use std::path::PathBuf; +use reflow_server::{ServerConfig, start_server}; #[derive(Args)] pub struct ServeArgs { /// Port to listen on. #[arg(long, default_value_t = 8080)] pub port: u16, + /// Address to bind. + #[arg(long, default_value = "0.0.0.0")] + pub bind: String, + /// Optional Redis URL for workflow persistence (survives restarts). + #[arg(long)] + pub redis: Option, } pub async fn run(args: ServeArgs) -> Result<()> { - let bin = locate_reflow_server(); - tracing::info!("starting {} on port {} (Zeal disabled)", bin.display(), args.port); - let status = tokio::process::Command::new(&bin) - .arg("--port") - .arg(args.port.to_string()) - .status() - .await - .with_context(|| { - format!( - "could not run `{}`. Build it first (cargo build -p reflow_server) \ - or put it on PATH.", - bin.display() - ) - })?; - if !status.success() { - bail!("reflow_server exited unsuccessfully ({status})"); - } - Ok(()) -} - -/// Prefer a `reflow_server` next to the current `reflow` executable; otherwise -/// fall back to the name on `PATH`. -fn locate_reflow_server() -> PathBuf { - let name = if cfg!(windows) { - "reflow_server.exe" - } else { - "reflow_server" + let config = ServerConfig { + port: args.port, + bind_address: args.bind, + redis_url: args.redis, + // Zeal is omitted in the CLI build; leave the session unconfigured. + zeal_url: None, + ..ServerConfig::default() }; - if let Ok(exe) = std::env::current_exe() - && let Some(dir) = exe.parent() - { - let candidate = dir.join(name); - if candidate.exists() { - return candidate; + tracing::info!( + "starting reflow server on {}:{} (Zeal disabled){}", + config.bind_address, + config.port, + if config.redis_url.is_some() { + ", Redis persistence on" + } else { + "" } - } - PathBuf::from(name) + ); + start_server(Some(config)).await } diff --git a/crates/reflow_server/Cargo.toml b/crates/reflow_server/Cargo.toml index a0bb278..50fe4a8 100644 --- a/crates/reflow_server/Cargo.toml +++ b/crates/reflow_server/Cargo.toml @@ -9,7 +9,11 @@ name = "reflow_server" path = "src/main.rs" [features] -default = [] +default = ["zeal"] +# Zeal IDE integration (ZIP session, trace submission, workflow conversion). +# Pulls in the out-of-tree `zeal-sdk`; turn off (`default-features = false`) to +# build the server core without it — e.g. when linked into the `reflow` CLI. +zeal = ["dep:zeal-sdk"] api_services = ["dep:reflow_api_services"] gpu = ["reflow_components/gpu"] @@ -19,10 +23,13 @@ reflow_graph = { path = "../reflow_graph" } reflow_components = { path = "../reflow_components", default-features = false } reflow_api_services = { path = "../reflow_api_services", optional = true } reflow_actor_macro = { path = "../reflow_actor_macro" } -zeal-sdk = { path = "../../../zeal/packages/zeal-rust-sdk" } -serde = { workspace = true } -serde_json = { workspace = true } -anyhow = { workspace = true } +zeal-sdk = { path = "../../../zeal/packages/zeal-rust-sdk", optional = true } +# Concrete versions (not `workspace = true`): reflow_server is excluded from the +# workspace, so it must resolve standalone when linked as a path dep (e.g. by the +# `reflow` CLI). Keep these in step with the workspace root's versions. +serde = { version = "1.0.215", features = ["derive", "rc"] } +serde_json = "1.0.133" +anyhow = "1.0.94" dashmap = "6.1.0" tokio = { version = "1.32", features = ["full"] } axum = { version = "0.7", features = ["ws"] } diff --git a/crates/reflow_server/src/engine.rs b/crates/reflow_server/src/engine.rs index f883a1b..09f503b 100644 --- a/crates/reflow_server/src/engine.rs +++ b/crates/reflow_server/src/engine.rs @@ -19,6 +19,7 @@ use reflow_network::graph::types::GraphExport; use reflow_network::network::{Network, NetworkConfig, NetworkEvent}; use serde::{Deserialize, Serialize}; +#[cfg(feature = "zeal")] use crate::zeal_converter::{ZealWorkflow, convert_zeal_to_graph_export}; /// Extracted StreamHandle metadata from actor outputs. @@ -263,6 +264,7 @@ impl ExecutionEngine { } /// Start a Zeal workflow — converts from Zeal format then executes. + #[cfg(feature = "zeal")] pub async fn start_zeal_execution( &self, zeal_workflow: ZealWorkflow, diff --git a/crates/reflow_server/src/event_bridge.rs b/crates/reflow_server/src/event_bridge.rs index d7167fd..8dd39af 100644 --- a/crates/reflow_server/src/event_bridge.rs +++ b/crates/reflow_server/src/event_bridge.rs @@ -7,17 +7,24 @@ //! One bridge is spawned per execution. It runs until the channel closes //! (execution finishes) or the bridge is explicitly dropped. -use std::sync::Arc; +use crate::engine::EngineEvent; +#[cfg(feature = "zeal")] +use std::sync::Arc; +#[cfg(feature = "zeal")] use log::{debug, error, info, warn}; +#[cfg(feature = "zeal")] use reflow_network::stream::STREAM_REGISTRY; - -use crate::engine::{EngineEvent, EngineEventType}; +#[cfg(feature = "zeal")] +use crate::engine::EngineEventType; +#[cfg(feature = "zeal")] use crate::trace_collector::TraceCollector; +#[cfg(feature = "zeal")] use crate::zip_session::ZipSession; /// Buffer size for observer taps attached to display-forwarded streams. /// Uses `try_send` so dropping frames is acceptable for display. +#[cfg(feature = "zeal")] const OBSERVER_BUFFER_SIZE: usize = 32; // ============================================================================ @@ -26,12 +33,19 @@ const OBSERVER_BUFFER_SIZE: usize = 32; /// Shared observability consumers that live for the lifetime of the server. /// Cloned into each per-execution bridge task. +/// +/// Without the `zeal` feature there are no consumers: [`EventBridge::attach`] +/// just drains the event channel so the engine's sender never blocks. +#[derive(Default)] pub struct EventBridge { + #[cfg(feature = "zeal")] trace_collector: Option>, + #[cfg(feature = "zeal")] zip_session: Option>, } impl EventBridge { + #[cfg(feature = "zeal")] pub fn new( trace_collector: Option>, zip_session: Option>, @@ -42,10 +56,24 @@ impl EventBridge { } } + /// Drain `event_rx` for a single execution without forwarding anywhere. + /// Present only without the `zeal` feature; keeps the engine's sender from + /// blocking when no observability consumers are wired in. + #[cfg(not(feature = "zeal"))] + pub fn attach( + &self, + _workflow_id: String, + _execution_id: String, + event_rx: flume::Receiver, + ) { + tokio::spawn(async move { while event_rx.recv_async().await.is_ok() {} }); + } + /// Spawn a background task that drains `event_rx` for a single execution, /// forwarding events to the TraceCollector and ZipSession. /// /// The task exits when the channel closes (sender dropped). + #[cfg(feature = "zeal")] pub fn attach( &self, workflow_id: String, @@ -182,6 +210,7 @@ impl EventBridge { /// Drain an observer receiver and forward each frame to Zeal as binary /// WebSocket messages. Emits `StreamClosed` / `StreamError` engine events /// when the stream terminates. + #[cfg(feature = "zeal")] async fn forward_stream_to_zeal( obs_rx: flume::Receiver, stream_id: u64, diff --git a/crates/reflow_server/src/lib.rs b/crates/reflow_server/src/lib.rs index 120740e..0f88889 100644 --- a/crates/reflow_server/src/lib.rs +++ b/crates/reflow_server/src/lib.rs @@ -29,10 +29,16 @@ pub mod engine; pub mod event_bridge; pub mod peer_mesh; pub mod rest_api; +pub mod workflow_store; + +// Zeal IDE integration — only built with the `zeal` feature. +#[cfg(feature = "zeal")] pub mod template_adapter; +#[cfg(feature = "zeal")] pub mod trace_collector; -pub mod workflow_store; +#[cfg(feature = "zeal")] pub mod zeal_converter; +#[cfg(feature = "zeal")] pub mod zip_session; use std::sync::Arc; @@ -109,7 +115,17 @@ pub async fn start_server(config: Option) -> Result<()> { // 1. Create the shared execution engine (with Redis persistence if configured) let engine = Arc::new(ExecutionEngine::new_with_redis(config.redis_url.clone())); - // 2. Optionally create the observability pipeline (TraceCollector + ZipSession + EventBridge) + // 2. Optionally create the observability pipeline (TraceCollector + ZipSession + EventBridge). + // Built only with the `zeal` feature; without it the server runs core execution + REST only. + #[cfg(not(feature = "zeal"))] + let event_bridge: Option> = { + if config.zeal_url.is_some() { + info!("zeal_url is set but the server was built without the `zeal` feature — ignoring"); + } + None + }; + + #[cfg(feature = "zeal")] let event_bridge = if let Some(zeal_url) = &config.zeal_url { // Trace collector submits per-node data via HTTP let trace_collector = Arc::new(trace_collector::TraceCollector::new(zeal_url)); diff --git a/crates/reflow_server/src/rest_api.rs b/crates/reflow_server/src/rest_api.rs index 3201742..0dfbb79 100644 --- a/crates/reflow_server/src/rest_api.rs +++ b/crates/reflow_server/src/rest_api.rs @@ -19,6 +19,7 @@ use tower_http::cors::CorsLayer; use crate::engine::{ExecutionEngine, ExecutionState}; use crate::event_bridge::EventBridge; +#[cfg(feature = "zeal")] use crate::zeal_converter::{ZealWorkflow, convert_zeal_to_graph_export}; // ============================================================================ @@ -155,6 +156,7 @@ async fn cancel_workflow( } } +#[cfg(feature = "zeal")] async fn execute_zeal_workflow( State(state): State, Json(request): Json, @@ -198,6 +200,7 @@ async fn execute_zeal_workflow( } } +#[cfg(feature = "zeal")] async fn convert_zeal_workflow( Json(zeal_workflow): Json, ) -> Result>, StatusCode> { @@ -229,6 +232,24 @@ async fn convert_zeal_workflow( } } +/// Convert a `{ "workflow": ZealWorkflow }` payload into a graph JSON value. +/// Without the `zeal` feature there's no converter, so it rejects the request. +fn graph_json_from_zeal_workflow(workflow: &serde_json::Value) -> Result { + #[cfg(feature = "zeal")] + { + let zeal_workflow: ZealWorkflow = + serde_json::from_value(workflow.clone()).map_err(|_| StatusCode::BAD_REQUEST)?; + let graph_export = + convert_zeal_to_graph_export(&zeal_workflow).map_err(|_| StatusCode::BAD_REQUEST)?; + serde_json::to_value(graph_export).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR) + } + #[cfg(not(feature = "zeal"))] + { + let _ = workflow; + Err(StatusCode::BAD_REQUEST) + } +} + /// Publish a workflow graph to Reflow. /// /// Zeal pushes a workflow graph here. Reflow stores it and returns @@ -241,13 +262,9 @@ async fn publish_workflow( Path(workflow_id): Path, Json(request): Json, ) -> Result>, StatusCode> { - // Accept either a Zeal workflow or a raw graph + // Accept either a Zeal workflow (only with the `zeal` feature) or a raw graph. let graph_json = if let Some(workflow) = request.get("workflow") { - let zeal_workflow: ZealWorkflow = - serde_json::from_value(workflow.clone()).map_err(|_| StatusCode::BAD_REQUEST)?; - let graph_export = - convert_zeal_to_graph_export(&zeal_workflow).map_err(|_| StatusCode::BAD_REQUEST)?; - serde_json::to_value(graph_export).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + graph_json_from_zeal_workflow(workflow)? } else if let Some(graph) = request.get("graph") { graph.clone() } else { @@ -288,6 +305,7 @@ async fn handle_webhook( headers: axum::http::HeaderMap, body: axum::body::Bytes, ) -> Result>, StatusCode> { + let request_path = format!("/webhook/{}", workflow_id); // Parse body as JSON (fall back to string) let body_json = serde_json::from_slice::(&body) .unwrap_or_else(|_| serde_json::json!(String::from_utf8_lossy(&body).to_string())); @@ -307,7 +325,7 @@ async fn handle_webhook( "body": body_json, "headers": headers_json, "method": "POST", - "path": format!("/webhook/{}", workflow_id), + "path": request_path, }); // Execute the workflow with the request data as input. @@ -560,17 +578,23 @@ pub fn build_router( event_bridge, }; - Router::new() + #[cfg_attr(not(feature = "zeal"), allow(unused_mut))] + let mut router = Router::new() .route("/health", get(health_check)) .route("/workflows", post(start_workflow)) .route("/workflows/{execution_id}", get(get_execution_status)) .route("/workflows/{execution_id}/cancel", post(cancel_workflow)) - .route("/zeal/workflows", post(execute_zeal_workflow)) - .route("/zeal/convert", post(convert_zeal_workflow)) .route("/workflows/{workflow_id}/publish", post(publish_workflow)) .route("/webhook/{workflow_id}", post(handle_webhook)) - .route("/webhook/{workflow_id}/{path:.*}", post(handle_webhook)) - .route("/ws", get(websocket_handler)) - .layer(CorsLayer::permissive()) - .with_state(state) + .route("/ws", get(websocket_handler)); + + // Zeal-specific conversion + execution routes — only with the `zeal` feature. + #[cfg(feature = "zeal")] + { + router = router + .route("/zeal/workflows", post(execute_zeal_workflow)) + .route("/zeal/convert", post(convert_zeal_workflow)); + } + + router.layer(CorsLayer::permissive()).with_state(state) } From 16f3ffd881f8458cc7d905e9d0bac2410097f7b2 Mon Sep 17 00:00:00 2001 From: darmie Date: Sat, 20 Jun 2026 09:44:56 +0100 Subject: [PATCH 3/4] build(server): make zeal-sdk a pinned git dependency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the out-of-tree `path = "../../../zeal/..."` with a git dependency (offbit-ai/zeal, pinned to a rev). Since `zeal-sdk` is optional and not enabled by any workspace member (the CLI links reflow_server with default-features = false), it never enters the workspace resolve graph: it's absent from Cargo.lock and `cargo metadata` only lists it as a declared optional dep. So workspace builds and maturin no longer need a sibling Zeal checkout — the git dep is fetched only when the `zeal` feature is explicitly enabled (standalone server). --- crates/reflow_server/Cargo.toml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/reflow_server/Cargo.toml b/crates/reflow_server/Cargo.toml index 50fe4a8..8dfe850 100644 --- a/crates/reflow_server/Cargo.toml +++ b/crates/reflow_server/Cargo.toml @@ -23,7 +23,9 @@ reflow_graph = { path = "../reflow_graph" } reflow_components = { path = "../reflow_components", default-features = false } reflow_api_services = { path = "../reflow_api_services", optional = true } reflow_actor_macro = { path = "../reflow_actor_macro" } -zeal-sdk = { path = "../../../zeal/packages/zeal-rust-sdk", optional = true } +# Git dep (not a local path) so building doesn't require a sibling Zeal checkout. +# Pinned to a rev for reproducibility; bump deliberately or switch to branch = "main". +zeal-sdk = { git = "https://github.com/offbit-ai/zeal.git", rev = "5aa5cd0f5626264351e7a110cc87c005eb92290a", optional = true } # Concrete versions (not `workspace = true`): reflow_server is excluded from the # workspace, so it must resolve standalone when linked as a path dep (e.g. by the # `reflow` CLI). Keep these in step with the workspace root's versions. From 933b9252cd0d67264a5a24779d04290adc7fe901 Mon Sep 17 00:00:00 2001 From: darmie Date: Sat, 20 Jun 2026 10:00:36 +0100 Subject: [PATCH 4/4] ci: build + release the `reflow` CLI binary per platform MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `.github/workflows/release-cli.yml`. On a `cli-v*` tag it builds the unified `reflow` binary for five targets and attaches per-target archives to a GitHub Release; `workflow_dispatch` builds all targets for verification without releasing. Each target builds on its NATIVE runner (incl. an arm64 Linux runner) rather than cross-compiling: the CLI links `reflow_server`, which pulls `native-tls` (OpenSSL on Linux). Native runners avoid cross-compiling OpenSSL for aarch64. Targets: {aarch64,x86_64}-apple-darwin, {x86_64,aarch64}-unknown-linux-gnu, x86_64-pc-windows-msvc. Unix → .tar.gz, Windows → .zip. Matches the existing publish-* workflows (dtolnay/rust-toolchain, sccache, action-gh-release). --- .github/workflows/release-cli.yml | 115 ++++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) create mode 100644 .github/workflows/release-cli.yml diff --git a/.github/workflows/release-cli.yml b/.github/workflows/release-cli.yml new file mode 100644 index 0000000..8155834 --- /dev/null +++ b/.github/workflows/release-cli.yml @@ -0,0 +1,115 @@ +# Build the unified `reflow` CLI for every supported triple and attach a +# per-triple archive to a GitHub Release. +# +# Each target is built on its NATIVE runner (including an arm64 Linux runner) +# rather than cross-compiled: the CLI links `reflow_server`, which pulls +# `native-tls` (OpenSSL on Linux, SChannel on Windows, Secure Transport on +# macOS). Native runners sidestep cross-compiling OpenSSL for aarch64. +# +# Triggers +# -------- +# - Tag push matching `cli-v*` (e.g. `cli-v0.2.0`) — builds every target and +# publishes a Release. +# - Manual `workflow_dispatch` — builds every target for verification, no +# Release. + +name: release-cli + +on: + push: + tags: + - 'cli-v*' + workflow_dispatch: + +permissions: + contents: write + +env: + SCCACHE_GHA_ENABLED: "true" + RUSTC_WRAPPER: "sccache" + +jobs: + build: + name: build / ${{ matrix.target }} + runs-on: ${{ matrix.host }} + strategy: + fail-fast: false + matrix: + include: + - host: macos-14 + target: aarch64-apple-darwin + - host: macos-14 + target: x86_64-apple-darwin + - host: ubuntu-latest + target: x86_64-unknown-linux-gnu + - host: ubuntu-24.04-arm + target: aarch64-unknown-linux-gnu + - host: windows-latest + target: x86_64-pc-windows-msvc + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + with: + targets: ${{ matrix.target }} + - uses: mozilla-actions/sccache-action@v0.0.5 + + - name: Build reflow CLI (release) + shell: bash + run: cargo build --release --locked --target "${{ matrix.target }}" -p reflow_cli + + - name: Package (unix) + if: runner.os != 'Windows' + shell: bash + run: | + set -eux + stage="reflow-${{ matrix.target }}" + mkdir -p "dist/$stage" + cp "target/${{ matrix.target }}/release/reflow" "dist/$stage/" + cp LICENSE* README* "dist/$stage/" 2>/dev/null || true + tar -C dist -czf "dist/$stage.tar.gz" "$stage" + + - name: Package (windows) + if: runner.os == 'Windows' + shell: bash + run: | + set -eux + stage="reflow-${{ matrix.target }}" + mkdir -p "dist/$stage" + cp "target/${{ matrix.target }}/release/reflow.exe" "dist/$stage/" + cp LICENSE* README* "dist/$stage/" 2>/dev/null || true + 7z a "dist/$stage.zip" "./dist/$stage" + + - uses: actions/upload-artifact@v4 + with: + name: reflow-cli-${{ matrix.target }} + path: | + dist/*.tar.gz + dist/*.zip + if-no-files-found: error + + release: + name: release + needs: build + runs-on: ubuntu-latest + if: startsWith(github.ref, 'refs/tags/cli-v') + steps: + - name: Download per-target archives + uses: actions/download-artifact@v4 + with: + path: artifacts + pattern: reflow-cli-* + merge-multiple: true + + - name: List archives + run: ls -laR artifacts + + - name: Create GitHub Release + uses: softprops/action-gh-release@v2 + with: + tag_name: ${{ github.ref_name }} + name: reflow CLI ${{ github.ref_name }} + generate_release_notes: true + files: | + artifacts/*.tar.gz + artifacts/*.zip + fail_on_unmatched_files: true