hatchery_server/
kernel_runtime.rs

1//! hq-core kernel runtime: the full former-harness runtime (run dispatch, managed
2//! worktrees, scheduling and continuations, delivery bundles, observation feed, operator
3//! and read host) hosted in-process.
4//!
5//! The runtime's node adapter dials core's own operator control socket, so every run
6//! dispatch, spawn and observation read goes core -> drain -> C2 -> node. Nothing dials
7//! a separate C2 control endpoint.
8//!
9//! The runtime owns the kernel databases (single writer); the hub forwards operator
10//! requests to it via [`RemoteKernel`]. Its operator/read host binds loopback only.
11//!
12//! Catalogs: launch-plan overrides and delivery bundles are read as `*.json` files from
13//! `<data dir>/kernel/launch-plans/` and `<data dir>/kernel/delivery-bundles/`. With none,
14//! the runtime derives default launch plans from the live node inventory.
15
16use crate::kernel::{KernelError, KERNEL_DB, OBSERVATION_DB};
17use hatchery_harness_api::{HarnessOperatorCredential, HarnessOperatorEnvelopeV1, HarnessOperatorRequestV1};
18use hatchery_harness_service::c2::HarnessC2Adapter;
19use hatchery_harness_service::dispatch::{compile_delivery_catalog_from_json_arguments, HarnessLaunchCatalog};
20use hatchery_harness_service::runtime::{start_harness_host_with_operator_and_catalogs, HarnessRuntimeCatalogs};
21use hatchery_harness_service::HarnessService;
22use hatchery_observation_service::ObservationService;
23use serde_json::Value;
24use std::io::{BufRead, BufReader, Write};
25use std::net::{Shutdown, SocketAddr, TcpStream};
26use std::path::{Path, PathBuf};
27use std::time::Duration;
28use tokio::sync::watch;
29
30/// Set to `1` to run the kernel runtime (needs the core mesh).
31pub const KERNEL_RUNTIME_ENV: &str = "HATCHERY_KERNEL_RUNTIME";
32/// Loopback address of the runtime's operator/read host.
33pub const KERNEL_BIND_ENV: &str = "HATCHERY_KERNEL_BIND";
34pub const DEFAULT_KERNEL_BIND: &str = "127.0.0.1:18961";
35/// Operator credential the hub presents to the runtime host.
36pub const KERNEL_OPERATOR_TOKEN_ENV: &str = "GATE4AGENT_HARNESS_OPERATOR_TOKEN";
37
38const CALL_DEADLINE: Duration = Duration::from_secs(45);
39
40/// Settings resolved from the environment (token var removed from the process env).
41pub struct KernelRuntimeConfig {
42    pub bind: SocketAddr,
43    pub credential: HarnessOperatorCredential,
44}
45
46impl KernelRuntimeConfig {
47    pub fn from_env() -> Result<Option<Self>, String> {
48        if std::env::var(KERNEL_RUNTIME_ENV).ok().as_deref() != Some("1") {
49            return Ok(None);
50        }
51        let bind: SocketAddr = std::env::var(KERNEL_BIND_ENV)
52            .unwrap_or_else(|_| DEFAULT_KERNEL_BIND.into())
53            .parse()
54            .map_err(|_| format!("{KERNEL_BIND_ENV} must be a socket address"))?;
55        if !bind.ip().is_loopback() {
56            return Err(format!("{KERNEL_BIND_ENV} must be loopback"));
57        }
58        let token = std::env::var(KERNEL_OPERATOR_TOKEN_ENV)
59            .map_err(|_| format!("{KERNEL_OPERATOR_TOKEN_ENV} is required with {KERNEL_RUNTIME_ENV}=1"))?;
60        std::env::remove_var(KERNEL_OPERATOR_TOKEN_ENV);
61        let credential = HarnessOperatorCredential::parse(token)
62            .map_err(|e| format!("{KERNEL_OPERATOR_TOKEN_ENV} is malformed: {e}"))?;
63        Ok(Some(Self { bind, credential }))
64    }
65}
66
67/// Hub-side handle: forwards typed operator requests to the runtime host.
68#[derive(Clone)]
69pub struct RemoteKernel {
70    endpoint: SocketAddr,
71    credential: HarnessOperatorCredential,
72}
73
74impl RemoteKernel {
75    pub fn new(endpoint: SocketAddr, credential: HarnessOperatorCredential) -> Self {
76        Self { endpoint, credential }
77    }
78
79    pub fn endpoint(&self) -> SocketAddr {
80        self.endpoint
81    }
82
83    /// One operator request (`HarnessOperatorRequestV1` JSON) -> the host's reply JSON.
84    pub fn call(&self, request: &Value) -> Result<Value, KernelError> {
85        let request: HarnessOperatorRequestV1 = serde_json::from_value(request.clone())
86            .map_err(|e| KernelError { status: 400, message: format!("bad operator request: {e}") })?;
87        let envelope = HarnessOperatorEnvelopeV1::new(self.credential.clone(), request);
88        let mut line = serde_json::to_vec(&envelope)
89            .map_err(|e| KernelError { status: 500, message: e.to_string() })?;
90        line.push(b'\n');
91        let io = |e: std::io::Error| KernelError { status: 503, message: format!("kernel runtime unreachable: {e}") };
92        let mut stream = TcpStream::connect_timeout(&self.endpoint, Duration::from_secs(3)).map_err(io)?;
93        stream.set_write_timeout(Some(CALL_DEADLINE)).map_err(io)?;
94        stream.set_read_timeout(Some(CALL_DEADLINE)).map_err(io)?;
95        stream.write_all(&line).map_err(io)?;
96        stream.shutdown(Shutdown::Write).map_err(io)?;
97        let mut reply = String::new();
98        BufReader::new(stream).read_line(&mut reply).map_err(io)?;
99        serde_json::from_str(reply.trim_end())
100            .map_err(|e| KernelError { status: 502, message: format!("bad kernel reply: {e}") })
101    }
102}
103
104fn read_json_dir(dir: &Path) -> Vec<String> {
105    let Ok(entries) = std::fs::read_dir(dir) else { return Vec::new() };
106    let mut files: Vec<PathBuf> = entries
107        .filter_map(|e| e.ok().map(|e| e.path()))
108        .filter(|p| p.extension().is_some_and(|x| x == "json"))
109        .collect();
110    files.sort();
111    files.iter().filter_map(|p| std::fs::read_to_string(p).ok()).collect()
112}
113
114/// Launch-plan overrides and delivery bundles from `<data dir>/kernel/`.
115pub fn load_catalogs(data_dir: &Path) -> Result<HarnessRuntimeCatalogs, String> {
116    let base = data_dir.join("kernel");
117    let launch = HarnessLaunchCatalog::from_json_arguments(read_json_dir(&base.join("launch-plans")))
118        .map_err(|e| format!("launch plans: {e}"))?;
119    let delivery = compile_delivery_catalog_from_json_arguments(read_json_dir(&base.join("delivery-bundles")))
120        .map_err(|e| format!("delivery bundles: {e}"))?;
121    HarnessRuntimeCatalogs::new(launch, delivery).map_err(|e| format!("catalogs disagree: {e}"))
122}
123
124async fn run_once(
125    data_dir: &Path,
126    operator_endpoint: &str,
127    operator_token: &str,
128    config_bind: SocketAddr,
129    credential: &HarnessOperatorCredential,
130    shutdown: &mut watch::Receiver<bool>,
131) -> Result<(), String> {
132    let catalogs = load_catalogs(data_dir)?;
133    let harness = HarnessService::open(data_dir.join(KERNEL_DB)).map_err(|e| format!("kernel db: {e}"))?;
134    let observation =
135        ObservationService::open(data_dir.join(OBSERVATION_DB)).map_err(|e| format!("observation db: {e}"))?;
136    let (adapter, events) = HarnessC2Adapter::connect(operator_endpoint.to_owned(), operator_token.to_owned())
137        .await
138        .map_err(|e| format!("operator link: {e}"))?;
139    let (host, mut task) = start_harness_host_with_operator_and_catalogs(
140        harness,
141        observation,
142        adapter,
143        events,
144        config_bind,
145        Some(credential.clone()),
146        catalogs,
147    )
148    .await
149    .map_err(|e| format!("kernel host: {e}"))?;
150    tracing::info!(endpoint = %host.endpoint().socket_addr(), "kernel runtime up (dispatch over core drain)");
151    tokio::select! {
152        r = &mut task => match r {
153            Ok(Ok(())) => Err("kernel runtime stopped".into()),
154            Ok(Err(e)) => Err(format!("kernel runtime: {e}")),
155            Err(e) => Err(format!("kernel runtime panicked: {e}")),
156        },
157        _ = shutdown.changed() => {
158            let _ = host.shutdown().await;
159            let _ = task.await;
160            Ok(())
161        }
162    }
163}
164
165/// Run the kernel runtime until shutdown, restarting it with backoff if it stops.
166pub async fn run(
167    data_dir: PathBuf,
168    operator_endpoint: String,
169    operator_token: String,
170    config: KernelRuntimeConfig,
171    mut shutdown: watch::Receiver<bool>,
172) {
173    let mut backoff = Duration::from_millis(500);
174    while !*shutdown.borrow() {
175        match run_once(&data_dir, &operator_endpoint, &operator_token, config.bind, &config.credential, &mut shutdown).await {
176            Ok(()) => return,
177            Err(err) => tracing::warn!(error = %err, "kernel runtime down; restarting"),
178        }
179        tokio::select! {
180            _ = tokio::time::sleep(backoff) => {}
181            _ = shutdown.changed() => return,
182        }
183        backoff = (backoff * 2).min(Duration::from_secs(30));
184    }
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190
191    #[test]
192    fn empty_catalog_dirs_load() {
193        let dir = tempfile::tempdir().unwrap();
194        load_catalogs(dir.path()).unwrap();
195        std::fs::create_dir_all(dir.path().join("kernel/delivery-bundles")).unwrap();
196        std::fs::write(dir.path().join("kernel/delivery-bundles/bad.json"), "{nope").unwrap();
197        assert!(load_catalogs(dir.path()).is_err());
198    }
199
200    #[test]
201    fn remote_kernel_rejects_unknown_request_before_dialing() {
202        let cred = HarnessOperatorCredential::parse(format!("{}{}", "g4aho_", "a".repeat(64))).unwrap();
203        let r = RemoteKernel::new("127.0.0.1:9".parse().unwrap(), cred);
204        assert_eq!(r.call(&serde_json::json!({"nope": 1})).unwrap_err().status, 400);
205    }
206}
hatchery_server/kernel_runtime.rs

hatchery_server/
kernel_runtime.rs
