hatchery_server/
kernel.rs

1//! hq-core task kernel: the former harness task/run/delivery kernel and the observation
2//! store, hosted in-process as libraries behind the hub (no separate `hatchery-harness`
3//! process, no legacy C2 control socket).
4//!
5//! Increment 1 (this module): durable task kernel + observation store, operator task verbs
6//! and read snapshots on the hub operator API under `/v1/kernel/...`. Run dispatch to nodes
7//! (worktrees, scheduling, deliveries) goes over the core drain in later increments.
8
9use hatchery_harness_protocol::{
10    HarnessCancelTaskRequestV1, HarnessCreateTaskRequestV1, HarnessMoveTaskRequestV1,
11    HarnessReplaceTaskRequestV1, HarnessRetryTaskRequestV1,
12};
13use hatchery_harness_service::HarnessService;
14use hatchery_observation_service::ObservationService;
15use serde::de::DeserializeOwned;
16use serde_json::{json, Value};
17use std::path::Path;
18
19pub const KERNEL_DB: &str = "harness.db";
20pub const OBSERVATION_DB: &str = "observation.db";
21
22pub struct Kernel {
23    harness: HarnessService,
24    observation: ObservationService,
25}
26
27#[derive(Debug)]
28pub struct KernelError {
29    pub status: u16,
30    pub message: String,
31}
32
33impl KernelError {
34    fn new(status: u16, message: impl Into<String>) -> Self {
35        Self { status, message: message.into() }
36    }
37}
38
39fn parse<T: DeserializeOwned>(body: &Value) -> Result<T, KernelError> {
40    serde_json::from_value(body.clone()).map_err(|e| KernelError::new(400, format!("bad request: {e}")))
41}
42
43impl Kernel {
44    /// Open (or create) the kernel databases under the server data dir.
45    pub fn open(data_dir: &Path) -> Result<Self, String> {
46        let harness = HarnessService::open(data_dir.join(KERNEL_DB)).map_err(|e| format!("kernel db: {e}"))?;
47        let observation =
48            ObservationService::open(data_dir.join(OBSERVATION_DB)).map_err(|e| format!("observation db: {e}"))?;
49        Ok(Self { harness, observation })
50    }
51
52    /// Operator verbs and reads. `parts` is the path after `/v1/kernel`.
53    pub fn call(&mut self, method: &str, parts: &[&str], body: &Value) -> Result<Value, KernelError> {
54        let applied = |r: Result<hatchery_harness_engine::HarnessApplyOutcome, hatchery_harness_service::HarnessServiceError>| {
55            r.map(|o| json!({ "outcome": format!("{o:?}") })).map_err(|e| KernelError::new(409, e.to_string()))
56        };
57        match (method, parts) {
58            ("GET", ["snapshot"]) => {
59                let engine = self.harness.engine().checkpoint();
60                let snap = self.harness.committed_snapshot();
61                Ok(json!({
62                    "engine": serde_json::to_value(&engine).map_err(|e| KernelError::new(500, e.to_string()))?,
63                    "dispatch_contexts": serde_json::to_value(&snap.dispatch_contexts).map_err(|e| KernelError::new(500, e.to_string()))?,
64                    "poisoned": self.harness.is_poisoned(),
65                }))
66            }
67            ("GET", ["observation"]) => Ok(json!({
68                "cursors": self.observation.durable_resume_cursors().iter()
69                    .map(|(n, c)| json!({ "node": format!("{n:?}"), "cursor": format!("{c:?}") }))
70                    .collect::<Vec<_>>(),
71                "poisoned": self.observation.is_poisoned(),
72            })),
73            ("POST", ["tasks"]) => applied(self.harness.operator_create_task(parse::<HarnessCreateTaskRequestV1>(body)?)),
74            ("POST", ["tasks", "replace"]) => applied(self.harness.operator_replace_task(parse::<HarnessReplaceTaskRequestV1>(body)?)),
75            ("POST", ["tasks", "move"]) => applied(self.harness.operator_move_task(parse::<HarnessMoveTaskRequestV1>(body)?)),
76            ("POST", ["tasks", "cancel"]) => applied(self.harness.operator_cancel_task(parse::<HarnessCancelTaskRequestV1>(body)?)),
77            ("POST", ["tasks", "retry"]) => applied(self.harness.operator_retry_task(parse::<HarnessRetryTaskRequestV1>(body)?)),
78            _ => Err(KernelError::new(404, "unknown kernel route")),
79        }
80    }
81}
82
83#[cfg(test)]
84mod tests {
85    use super::*;
86
87    #[test]
88    fn opens_empty_kernel_and_reads_snapshot() {
89        let dir = tempfile::tempdir().unwrap();
90        let mut k = Kernel::open(dir.path()).unwrap();
91        let snap = k.call("GET", &["snapshot"], &Value::Null).unwrap();
92        assert_eq!(snap["poisoned"], false);
93        assert!(snap["engine"].is_object());
94        assert_eq!(k.call("GET", &["observation"], &Value::Null).unwrap()["poisoned"], false);
95        assert_eq!(k.call("GET", &["nope"], &Value::Null).unwrap_err().status, 404);
96        assert_eq!(k.call("POST", &["tasks"], &json!({"x":1})).unwrap_err().status, 400);
97        drop(k);
98        // Reopens the same durable files.
99        Kernel::open(dir.path()).unwrap();
100    }
101}
hatchery_server/kernel.rs

hatchery_server/
kernel.rs
