merge: embed canonical hanzo-engine in hanzo-node (default-off engine feature)

One engine everywhere — hanzo-node can serve the same hanzo-server-core
(/v1/responses + /v1/messages + /v1/chat/completions) in-process alongside
gRPC. Default-off so the no-features build is unchanged.
This commit is contained in:
2026-06-18 16:45:44 -07:00
6 changed files with 5397 additions and 106 deletions
+1
View File
@@ -52,6 +52,7 @@ hanzo-dev/bin/
# Rust build artifacts
/target/
/codex-rs/target/
/hanzo-node/rust/target/
**/*.rs.bk
# Local Cargo caches (never commit)
+5
View File
@@ -0,0 +1,5 @@
# The optional `engine` feature pulls in hanzo-engine's candle fork (hanzo-ml),
# whose SIMD kernels require `-C target-cpu=native` to compile. This is a no-op
# for the default (engine-off) build, so it is safe to set unconditionally.
[build]
rustflags = ["-C", "target-cpu=native"]
+5163 -106
View File
File diff suppressed because it is too large Load Diff
+23
View File
@@ -81,6 +81,11 @@ async-stream = "0.3"
tokio-stream = "0.1"
futures = "0.3"
# Canonical inference engine (optional, default-off behind the `engine` feature).
# Path-deps into the sibling `engine` workspace; compiled only when `engine` is on.
hanzo-server-core = { path = "../../../engine/hanzo-server-core", optional = true, default-features = false }
hanzo-engine = { path = "../../../engine/hanzo-engine", optional = true }
[build-dependencies]
tonic-build = "0.12"
@@ -93,6 +98,24 @@ default = []
full = ["mlx"]
mlx = [] # Optional MLX acceleration support
# Embed the canonical Hanzo inference engine in-process (serves the OpenAI/
# Anthropic/Responses HTTP API alongside gRPC). Default-off and purely additive:
# with `engine` disabled the node builds and behaves exactly as before.
engine = ["dep:hanzo-server-core", "dep:hanzo-engine"]
# Hardware acceleration backends. Each forwards to the matching engine feature.
# Mirror of hanzo-server/Cargo.toml. All imply `engine`.
cuda = ["engine", "hanzo-server-core/cuda", "hanzo-engine/cuda"]
cudnn = ["engine", "hanzo-server-core/cudnn", "hanzo-engine/cudnn"]
metal = ["engine", "hanzo-server-core/metal", "hanzo-engine/metal"]
rocm = ["engine", "hanzo-server-core/rocm", "hanzo-engine/rocm"]
vulkan = ["engine", "hanzo-server-core/vulkan", "hanzo-engine/vulkan"]
accelerate = ["engine", "hanzo-server-core/accelerate", "hanzo-engine/accelerate"]
mkl = ["engine", "hanzo-server-core/mkl", "hanzo-engine/mkl"]
nccl = ["engine", "hanzo-server-core/nccl", "hanzo-engine/nccl"]
flash-attn = ["cuda", "hanzo-server-core/flash-attn", "hanzo-engine/flash-attn"]
flash-attn-v3 = ["cuda", "hanzo-server-core/flash-attn-v3", "hanzo-engine/flash-attn-v3"]
[lints.rust]
# Match hanzo-dev linting style
+84
View File
@@ -25,6 +25,9 @@ fn print_help() {
println!(" --bootstrap <ADDR> Bootstrap peer address (can be repeated)");
println!(" --operator <ADDR> Operator wallet address");
println!(" --mlx Enable MLX acceleration (macOS only)");
println!(" --engine-port <ADDR> Embedded engine HTTP API address or port (default: 0.0.0.0:36900)");
println!(" --model <ID> Load this model into the embedded engine (requires `engine` build feature)");
println!(" --engine-cpu Force CPU-only inference for the embedded engine");
println!(" -h, --help Print help");
println!(" -V, --version Print version");
println!();
@@ -37,9 +40,58 @@ fn print_help() {
println!(" HANZO_NETWORK_ID Network ID");
println!(" HANZO_OPERATOR Operator wallet address");
println!(" HANZO_MLX Enable MLX acceleration (1/true)");
println!(" HANZO_ENGINE_ADDR Embedded engine HTTP API address or port");
println!(" HANZO_ENGINE_MODEL Model ID to load into the embedded engine");
println!(" RUST_LOG Log level (e.g., info, debug, hanzo_node=debug)");
}
/// Normalize an engine API spec: a bare port becomes `0.0.0.0:<port>`, an
/// `addr:port` is taken as-is.
fn normalize_engine_addr(spec: &str) -> String {
if spec.parse::<u16>().is_ok() {
format!("0.0.0.0:{spec}")
} else {
spec.to_string()
}
}
/// Apply a model id to the engine config. With the `engine` feature this builds
/// a `ModelSelected::Run` auto-loader selector and enables the engine; without
/// it, the model id is recorded only to emit a warning.
fn apply_engine_model(config: &mut NodeConfig, model_id: String) {
#[cfg(feature = "engine")]
{
config.engine.enabled = true;
config.engine.model = Some(hanzo_engine::ModelSelected::Run {
model_id,
tokenizer_json: None,
dtype: hanzo_engine::ModelDType::Auto,
topology: None,
organization: None,
write_uqff: None,
from_uqff: None,
imatrix: None,
calibration_file: None,
max_edge: None,
max_seq_len: hanzo_engine::AutoDeviceMapParams::DEFAULT_MAX_SEQ_LEN,
max_batch_size: hanzo_engine::AutoDeviceMapParams::DEFAULT_MAX_BATCH_SIZE,
max_num_images: None,
max_image_length: None,
hf_cache_path: None,
matformer_config_path: None,
matformer_slice_name: None,
});
}
#[cfg(not(feature = "engine"))]
{
let _ = (&mut *config, model_id);
tracing::warn!(
"--model/HANZO_ENGINE_MODEL was set but this binary was built without the \
`engine` feature; the embedded engine is unavailable and the flag is ignored."
);
}
}
fn parse_args() -> Result<Option<NodeConfig>> {
let args: Vec<String> = env::args().collect();
let mut config = NodeConfig::default();
@@ -69,6 +121,12 @@ fn parse_args() -> Result<Option<NodeConfig>> {
if let Ok(v) = env::var("HANZO_MLX") {
config.mlx_enabled = v == "1" || v.to_lowercase() == "true";
}
if let Ok(v) = env::var("HANZO_ENGINE_ADDR") {
config.engine.engine_api_addr = normalize_engine_addr(&v);
}
if let Ok(v) = env::var("HANZO_ENGINE_MODEL") {
apply_engine_model(&mut config, v);
}
let mut i = 1;
while i < args.len() {
@@ -140,6 +198,25 @@ fn parse_args() -> Result<Option<NodeConfig>> {
"--mlx" => {
config.mlx_enabled = true;
}
"--engine-port" | "--engine-addr" => {
i += 1;
if i >= args.len() {
return Err(hanzo_node::Error::Config(
"--engine-port requires a value".into(),
));
}
config.engine.engine_api_addr = normalize_engine_addr(&args[i]);
}
"--model" => {
i += 1;
if i >= args.len() {
return Err(hanzo_node::Error::Config("--model requires a value".into()));
}
apply_engine_model(&mut config, args[i].clone());
}
"--engine-cpu" => {
config.engine.cpu = true;
}
arg if arg.starts_with('-') => {
return Err(hanzo_node::Error::Config(format!("Unknown option: {arg}")));
}
@@ -175,6 +252,13 @@ async fn main() -> Result<()> {
"Starting Hanzo Node"
);
if config.engine.enabled {
tracing::info!(
engine_api = %format!("http://{}", config.engine.engine_api_addr),
"Embedded engine enabled; HTTP API (OpenAI/Anthropic/Responses) will serve here"
);
}
// Create and start the node
let node = HanzoNode::new(config).await?;
+121
View File
@@ -25,6 +25,46 @@ pub use iam::{IamConfig, IamService};
use std::sync::Arc;
use tokio::sync::RwLock;
/// Default bind address for the embedded canonical engine HTTP API.
/// Matches the `node` provider contract used by hanzo-dev.
pub const DEFAULT_ENGINE_API_ADDR: &str = "0.0.0.0:36900";
/// Configuration for the optional in-process canonical inference engine.
///
/// When `enabled` and a model is configured (only meaningful with the `engine`
/// feature compiled in), the node serves the engine's OpenAI/Anthropic/Responses
/// HTTP API on `engine_api_addr`, concurrently with the gRPC server. When the
/// `engine` feature is off, or `enabled` is false, this is inert and the node
/// behaves identically to before.
#[derive(Debug, Clone)]
pub struct EngineConfig {
/// Serve the embedded engine HTTP API. No-op unless the `engine` feature is built.
pub enabled: bool,
/// Force CPU-only inference.
pub cpu: bool,
/// Bind address for the engine HTTP API.
pub engine_api_addr: String,
/// Model to load via the auto loader. Required for the engine to actually start.
#[cfg(feature = "engine")]
pub model: Option<hanzo_engine::ModelSelected>,
}
impl Default for EngineConfig {
fn default() -> Self {
Self {
enabled: false,
cpu: false,
engine_api_addr: DEFAULT_ENGINE_API_ADDR.to_string(),
#[cfg(feature = "engine")]
model: None,
}
}
}
/// Shared canonical-engine state, re-exported for downstream callers.
#[cfg(feature = "engine")]
pub use hanzo_server_core::types::SharedHanzoState;
/// Node configuration
#[derive(Debug, Clone)]
pub struct NodeConfig {
@@ -50,6 +90,8 @@ pub struct NodeConfig {
pub iam: IamConfig,
/// Lux node endpoint for consensus replication (e.g., "http://localhost:9650")
pub lux_endpoint: Option<String>,
/// Embedded canonical inference engine configuration (default-off).
pub engine: EngineConfig,
}
impl Default for NodeConfig {
@@ -66,6 +108,7 @@ impl Default for NodeConfig {
mlx_enabled: false,
iam: IamConfig::default(),
lux_endpoint: None,
engine: EngineConfig::default(),
}
}
}
@@ -88,6 +131,8 @@ pub struct HanzoNode {
compute: Arc<compute::ComputeManager>,
iam: Arc<IamService>,
start_time: std::time::Instant,
#[cfg(feature = "engine")]
hanzo_state: Option<SharedHanzoState>,
}
impl HanzoNode {
@@ -117,6 +162,9 @@ impl HanzoNode {
tracing::warn!("IAM authentication disabled - all requests will be allowed");
}
#[cfg(feature = "engine")]
let hanzo_state = Self::build_engine_state(&config.engine).await?;
Ok(Self {
config,
state: Arc::new(RwLock::new(NodeState::Starting)),
@@ -125,9 +173,39 @@ impl HanzoNode {
compute,
iam,
start_time: std::time::Instant::now(),
#[cfg(feature = "engine")]
hanzo_state,
})
}
/// Build the embedded canonical-engine state when enabled and a model is set.
/// Returns `Ok(None)` (and the node serves gRPC only) when the engine is
/// disabled or no model is configured.
#[cfg(feature = "engine")]
async fn build_engine_state(cfg: &EngineConfig) -> Result<Option<SharedHanzoState>> {
use hanzo_server_core::hanzo_for_server_builder::HanzoForServerBuilder;
if !cfg.enabled {
return Ok(None);
}
let Some(model) = cfg.model.clone() else {
tracing::warn!(
"Embedded engine enabled but no model configured; engine HTTP API will not start. \
Pass --model/HANZO_ENGINE_MODEL to load one."
);
return Ok(None);
};
tracing::info!(cpu = cfg.cpu, "Loading embedded canonical engine model");
let state = HanzoForServerBuilder::new()
.with_model(model)
.with_cpu(cfg.cpu)
.build()
.await
.map_err(|e| Error::Config(format!("Failed to build embedded engine: {e}")))?;
Ok(Some(state))
}
/// Start the node
pub async fn start(&self) -> Result<()> {
tracing::info!("Starting Hanzo Node");
@@ -162,12 +240,55 @@ impl HanzoNode {
"Hanzo Node is ready"
);
// When the embedded engine is built, serve its HTTP API concurrently with
// gRPC. Otherwise keep the original behavior: gRPC blocks until shutdown.
#[cfg(feature = "engine")]
if let Some(state) = self.hanzo_state.clone() {
return self.run_with_engine(rpc_server, state).await;
}
// Run the gRPC server (this blocks until shutdown)
rpc_server.run().await?;
Ok(())
}
/// Serve the gRPC server and the embedded engine HTTP API concurrently.
#[cfg(feature = "engine")]
async fn run_with_engine(
&self,
rpc_server: rpc::RpcServer,
state: SharedHanzoState,
) -> Result<()> {
use hanzo_server_core::hanzo_server_router_builder::HanzoServerRouterBuilder;
// hanzo-server-core is pulled with default-features = false, so its
// `swagger-ui` feature is off and the /docs + /api-doc routes are compiled
// out. The router therefore exposes only the API surface (chat/completions,
// responses, messages, models, ...), which is what we want for the node.
let app = HanzoServerRouterBuilder::new()
.with_hanzo(state)
.build()
.await
.map_err(|e| Error::Config(format!("Failed to build engine router: {e}")))?;
let addr = self.config.engine.engine_api_addr.clone();
let listener = tokio::net::TcpListener::bind(&addr).await?;
tracing::info!(engine_api = %addr, "Embedded engine HTTP API listening");
let grpc = tokio::spawn(async move { rpc_server.run().await });
let engine = tokio::spawn(async move {
axum::serve(listener, app)
.await
.map_err(Error::from)
});
let (grpc_res, engine_res) = tokio::join!(grpc, engine);
grpc_res.map_err(|e| Error::Internal(format!("gRPC task panicked: {e}")))??;
engine_res.map_err(|e| Error::Internal(format!("engine task panicked: {e}")))??;
Ok(())
}
/// Graceful shutdown
pub async fn shutdown(&self) -> Result<()> {
tracing::info!("Shutting down Hanzo Node");