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    fn auth(hex: char, now: u64) -> Value {
88        json!({
89            "operation_id": format!("hop_{}", hex.to_string().repeat(24)),
90            "idempotency_ref": format!("hidem_{}", hex.to_string().repeat(24)),
91            "actor_id": "operator",
92            "now_unix_ms": now,
93        })
94    }
95
96    fn task<'a>(snap: &'a Value, id: &str) -> &'a Value {
97        snap["engine"]["tasks"].as_array().unwrap().iter().find(|t| t["task_id"] == id).expect("task in snapshot")
98    }
99
100    #[test]
101    fn task_create_move_cancel_over_the_api() {
102        let dir = tempfile::tempdir().unwrap();
103        let mut k = Kernel::open(dir.path()).unwrap();
104        let id = format!("htask_{}", "b".repeat(24));
105        let create = json!({
106            "authority": auth('1', 10), "task_id": id, "title": "Port the kernel",
107            "body": "hq-core owns tasks", "parent_task_id": null, "dependencies": [], "initial_state": "backlog",
108        });
109        k.call("POST", &["tasks"], &create).unwrap();
110        // Same operation replays idempotently.
111        k.call("POST", &["tasks"], &create).unwrap();
112        let snap = k.call("GET", &["snapshot"], &Value::Null).unwrap();
113        assert_eq!(task(&snap, &id)["state"], "backlog");
114        let rev = task(&snap, &id)["revision"].clone();
115
116        k.call("POST", &["tasks", "move"], &json!({
117            "authority": auth('2', 20), "task_id": id, "expected_revision": rev, "state": "ready",
118        })).unwrap();
119        let snap = k.call("GET", &["snapshot"], &Value::Null).unwrap();
120        assert_eq!(task(&snap, &id)["state"], "ready");
121        let rev2 = task(&snap, &id)["revision"].clone();
122        assert_ne!(rev, rev2);
123
124        // Stale revision is refused.
125        let stale = k.call("POST", &["tasks", "cancel"], &json!({
126            "authority": auth('3', 30), "task_id": id, "expected_revision": rev,
127        }));
128        assert_eq!(stale.unwrap_err().status, 409);
129        k.call("POST", &["tasks", "cancel"], &json!({
130            "authority": auth('4', 40), "task_id": id, "expected_revision": rev2,
131        })).unwrap();
132        drop(k);
133        // Durable across reopen.
134        let mut k = Kernel::open(dir.path()).unwrap();
135        let snap = k.call("GET", &["snapshot"], &Value::Null).unwrap();
136        assert_eq!(task(&snap, &id)["state"], "cancelled");
137    }
138
139    #[test]
140    fn opens_empty_kernel_and_reads_snapshot() {
141        let dir = tempfile::tempdir().unwrap();
142        let mut k = Kernel::open(dir.path()).unwrap();
143        let snap = k.call("GET", &["snapshot"], &Value::Null).unwrap();
144        assert_eq!(snap["poisoned"], false);
145        assert!(snap["engine"].is_object());
146        assert_eq!(k.call("GET", &["observation"], &Value::Null).unwrap()["poisoned"], false);
147        assert_eq!(k.call("GET", &["nope"], &Value::Null).unwrap_err().status, 404);
148        assert_eq!(k.call("POST", &["tasks"], &json!({"x":1})).unwrap_err().status, 400);
149        drop(k);
150        // Reopens the same durable files.
151        Kernel::open(dir.path()).unwrap();
152    }
153}
hatchery_server/kernel.rs

hatchery_server/
kernel.rs
