hatchery_server/
hub.rs

1//! P6: the hub behind the server control API: node calls forwarded by C2 and operator (HQ)
2//! calls for mail and tasks. All state lives under the data dir; admission is always read from
3//! the P2 ledger, never from the request.
4
5use crate::messenger::{Messenger, MessengerError, OPERATOR};
6use crate::provision::{read_ledger, LedgerEntry, NodeStatus};
7use crate::tasks::{Refused, Session, TaskStore};
8use serde_json::{json, Value};
9use std::path::{Path, PathBuf};
10use std::sync::{Condvar, Mutex};
11use std::time::{Duration, Instant};
12
13#[derive(Debug)]
14pub struct HubError {
15    pub status: u16,
16    pub message: String,
17}
18
19impl HubError {
20    fn new(status: u16, message: impl Into<String>) -> Self {
21        Self { status, message: message.into() }
22    }
23}
24
25impl From<MessengerError> for HubError {
26    fn from(e: MessengerError) -> Self {
27        match e {
28            MessengerError::Forbidden => Self::new(403, "refused"),
29            MessengerError::Bad(m) => Self::new(400, m),
30            MessengerError::Internal(m) => Self::new(500, m),
31        }
32    }
33}
34
35pub struct Hub {
36    data_dir: PathBuf,
37    messenger: Mutex<Messenger>,
38    tasks: Mutex<TaskStore>,
39    /// Bumped on every message sent; lets HQ long-poll for new mail instead of polling blindly.
40    mail_signal: (Mutex<u64>, Condvar),
41}
42
43fn refused(r: Refused) -> HubError {
44    match r {
45        Refused::NotFound => HubError::new(404, "not found"),
46        Refused::Conflict(m) => HubError::new(409, m),
47        Refused::Invalid(m) => HubError::new(400, m),
48    }
49}
50
51fn int(e: crate::error::ServerError) -> HubError {
52    HubError::new(500, e.to_string())
53}
54
55fn opt_str<'a>(v: &'a Value, name: &str) -> Option<&'a str> {
56    v.get(name).and_then(Value::as_str)
57}
58
59fn query_param<'a>(query: &'a str, name: &str) -> Option<&'a str> {
60    query.split('&').find_map(|kv| kv.strip_prefix(name).and_then(|r| r.strip_prefix('='))).filter(|v| !v.is_empty())
61}
62
63fn str_field<'a>(v: &'a Value, name: &str) -> Result<&'a str, HubError> {
64    v.get(name).and_then(Value::as_str).ok_or_else(|| HubError::new(400, format!("missing field: {name}")))
65}
66
67impl Hub {
68    pub fn open(data_dir: &Path) -> Result<Self, crate::error::ServerError> {
69        Ok(Self {
70            data_dir: data_dir.to_owned(),
71            messenger: Mutex::new(Messenger::open(data_dir)?),
72            tasks: Mutex::new(TaskStore::open(data_dir)?),
73            mail_signal: (Mutex::new(0), Condvar::new()),
74        })
75    }
76
77    fn bump_mail(&self) {
78        *self.mail_signal.0.lock().expect("signal") += 1;
79        self.mail_signal.1.notify_all();
80    }
81
82    fn entry(&self, node_id: &str) -> Option<LedgerEntry> {
83        read_ledger(&self.data_dir).ok()?.into_iter().find(|e| e.node_id == node_id)
84    }
85
86    /// Active (member/trusted) node holding `capability`, per the ledger.
87    fn admitted(&self, node_id: &str, capability: &str) -> bool {
88        node_id != OPERATOR
89            && self.entry(node_id).is_some_and(|e| {
90                matches!(e.status, NodeStatus::Member | NodeStatus::Trusted) && e.grants.capabilities.iter().any(|c| c == capability)
91            })
92    }
93
94    /// A call made on behalf of a node. The caller (C2) has authenticated the node and checked its
95    /// policy; the hub checks the ledger again, independently.
96    pub fn node_call(&self, node_id: &str, op: &str, payload: &Value) -> Result<Value, HubError> {
97        let capability = match op.split_once('.').map(|p| p.0) {
98            Some("mail") => "mail",
99            Some("task") | Some("run") => "tasks",
100            _ => return Err(HubError::new(404, "unknown operation")),
101        };
102        if !self.admitted(node_id, capability) {
103            return Err(HubError::new(403, "refused"));
104        }
105        match op {
106            "mail.send" => {
107                let key = payload.get("idempotency_key").and_then(Value::as_str).map(str::to_owned);
108                let id = self.messenger.lock().expect("messenger").send(node_id, str_field(payload, "to")?, str_field(payload, "subject")?, str_field(payload, "body")?, key)?;
109                self.bump_mail();
110                Ok(json!({ "message_id": id }))
111            }
112            "mail.inbox" => {
113                let since = payload.get("since_ms").and_then(Value::as_u64).unwrap_or(0);
114                let limit = payload.get("limit").and_then(Value::as_u64).unwrap_or(50).min(256) as u16;
115                let mail = self.messenger.lock().expect("messenger").inbox(node_id, since, limit)?;
116                Ok(json!({ "messages": mail }))
117            }
118            "mail.ack" => {
119                self.messenger.lock().expect("messenger").ack(node_id, str_field(payload, "message_id")?)?;
120                Ok(json!({ "ok": true }))
121            }
122            "task.list" => {
123                let tasks = self.tasks.lock().expect("tasks").list(Some(node_id), true).map_err(|e| HubError::new(500, e.to_string()))?;
124                Ok(json!({ "tasks": tasks }))
125            }
126            "task.update" => {
127                let id = payload.get("id").and_then(Value::as_i64).ok_or_else(|| HubError::new(400, "missing field: id"))?;
128                let status = str_field(payload, "status")?;
129                if !matches!(status, "in_progress" | "done") {
130                    return Err(HubError::new(400, "status must be in_progress or done"));
131                }
132                let changed = self.tasks.lock().expect("tasks").set_status(id, status, Some(node_id)).map_err(|e| HubError::new(500, e.to_string()))?;
133                if changed { Ok(json!({ "ok": true })) } else { Err(HubError::new(403, "refused")) }
134            }
135            // The run's own session closes it (decision 8); the run must live on this node.
136            "run.close" => {
137                let run_id = payload.get("run_id").and_then(Value::as_i64).ok_or_else(|| HubError::new(400, "missing field: run_id"))?;
138                let tasks = self.tasks.lock().expect("tasks");
139                let run = tasks.run(run_id).map_err(int)?.ok_or_else(|| HubError::new(404, "not found"))?;
140                if run.node_id != node_id {
141                    return Err(HubError::new(403, "refused"));
142                }
143                let run = tasks
144                    .close_run(run_id, str_field(payload, "status")?, "session", opt_str(payload, "session_id"), opt_str(payload, "branch"), opt_str(payload, "commit"), opt_str(payload, "result_message_id"))
145                    .map_err(int)?
146                    .map_err(refused)?;
147                Ok(json!({ "run": run }))
148            }
149            _ => Err(HubError::new(404, "unknown operation")),
150        }
151    }
152
153    /// Operator (HQ) calls, authenticated by the control bearer in the HTTP layer.
154    pub fn operator_call(&self, method: &str, path: &str, body: &Value) -> Result<Value, HubError> {
155        let (path, query) = path.split_once('?').unwrap_or((path, ""));
156        let parts: Vec<&str> = path.trim_matches('/').split('/').collect();
157        match (method, parts.as_slice()) {
158            ("POST", ["v1", "tasks"]) => {
159                let assignee = opt_str(body, "assignee");
160                if let Some(node) = assignee {
161                    if !self.admitted(node, "tasks") {
162                        return Err(HubError::new(403, "assignee does not hold the tasks capability"));
163                    }
164                }
165                let tasks = self.tasks.lock().expect("tasks");
166                let op = opt_str(body, "op_id");
167                if let Some(done) = tasks.replay(op).map_err(int)? {
168                    return Ok(done);
169                }
170                let status = opt_str(body, "status");
171                if let Some(st) = status {
172                    if !matches!(st, "backlog" | "open" | "assigned") {
173                        return Err(HubError::new(400, "initial status must be backlog, open or assigned"));
174                    }
175                }
176                let task = tasks
177                    .create_full(str_field(body, "title")?, opt_str(body, "body").unwrap_or(""), assignee, opt_str(body, "role"), body.get("parent_id").and_then(Value::as_i64), status)
178                    .map_err(int)?;
179                let out = json!({ "task": task });
180                tasks.record(op, &out).map_err(int)?;
181                Ok(out)
182            }
183            ("GET", ["v1", "tasks"]) => {
184                let roles: Option<Vec<String>> = query_param(query, "role").map(|r| r.split(',').map(str::to_owned).collect());
185                let tasks = self.tasks.lock().expect("tasks").list_filtered(query_param(query, "assignee"), roles.as_deref(), false).map_err(int)?;
186                Ok(json!({ "tasks": tasks }))
187            }
188            ("GET", ["v1", "tasks", id]) => {
189                let id: i64 = id.parse().map_err(|_| HubError::new(400, "bad id"))?;
190                let tasks = self.tasks.lock().expect("tasks");
191                let task = tasks.get(id).map_err(int)?.ok_or_else(|| HubError::new(404, "not found"))?;
192                Ok(json!({ "task": task, "runs": tasks.runs(id).map_err(int)? }))
193            }
194            ("POST", ["v1", "tasks", id, "move"]) => {
195                let id: i64 = id.parse().map_err(|_| HubError::new(400, "bad id"))?;
196                let tasks = self.tasks.lock().expect("tasks");
197                let op = opt_str(body, "op_id");
198                if let Some(done) = tasks.replay(op).map_err(int)? {
199                    return Ok(done);
200                }
201                let task = tasks
202                    .move_to(id, str_field(body, "status")?, body.get("expected_revision").and_then(Value::as_i64))
203                    .map_err(int)?
204                    .map_err(refused)?;
205                let out = json!({ "task": task });
206                tasks.record(op, &out).map_err(int)?;
207                Ok(out)
208            }
209            ("POST", ["v1", "tasks", id, "runs"]) => {
210                let id: i64 = id.parse().map_err(|_| HubError::new(400, "bad id"))?;
211                let run = self.tasks.lock().expect("tasks").start_run(id, str_field(body, "session_id")?).map_err(int)?.map_err(refused)?;
212                Ok(json!({ "run": run }))
213            }
214            ("POST", ["v1", "runs", id, "close"]) => {
215                let id: i64 = id.parse().map_err(|_| HubError::new(400, "bad id"))?;
216                let run = self
217                    .tasks
218                    .lock()
219                    .expect("tasks")
220                    .close_run(id, str_field(body, "status")?, "operator", None, opt_str(body, "branch"), opt_str(body, "commit"), opt_str(body, "result_message_id"))
221                    .map_err(int)?
222                    .map_err(refused)?;
223                Ok(json!({ "run": run }))
224            }
225            ("GET", ["v1", "sessions"]) => {
226                let sessions = self.tasks.lock().expect("tasks").sessions(query_param(query, "node")).map_err(int)?;
227                Ok(json!({ "sessions": sessions }))
228            }
229            // Record a real provider session (a provider conversation id); drafts are not recorded.
230            ("POST", ["v1", "sessions"]) => {
231                let session = Session {
232                    session_id: str_field(body, "session_id")?.to_owned(),
233                    node_id: str_field(body, "node_id")?.to_owned(),
234                    provider: str_field(body, "provider")?.to_owned(),
235                    workspace_id: opt_str(body, "workspace_id").map(str::to_owned),
236                    role: opt_str(body, "role").map(str::to_owned),
237                    state: opt_str(body, "state").unwrap_or("live").to_owned(),
238                    created_at_ms: 0,
239                    updated_at_ms: 0,
240                };
241                let session = self.tasks.lock().expect("tasks").upsert_session(&session).map_err(int)?;
242                Ok(json!({ "session": session }))
243            }
244            ("POST", ["v1", "tasks", id, "assign"]) => {
245                let id: i64 = id.parse().map_err(|_| HubError::new(400, "bad id"))?;
246                let node = str_field(body, "assignee")?;
247                if !self.admitted(node, "tasks") {
248                    return Err(HubError::new(403, "assignee does not hold the tasks capability"));
249                }
250                if self.tasks.lock().expect("tasks").assign(id, node).map_err(|e| HubError::new(500, e.to_string()))? { Ok(json!({ "ok": true })) } else { Err(HubError::new(409, "task cannot be assigned")) }
251            }
252            ("POST", ["v1", "tasks", id, "cancel"]) => {
253                let id: i64 = id.parse().map_err(|_| HubError::new(400, "bad id"))?;
254                if self.tasks.lock().expect("tasks").set_status(id, "cancelled", None).map_err(|e| HubError::new(500, e.to_string()))? { Ok(json!({ "ok": true })) } else { Err(HubError::new(409, "task is final")) }
255            }
256            ("POST", ["v1", "mail", "send"]) => {
257                let id = self.messenger.lock().expect("messenger").send(OPERATOR, str_field(body, "to")?, str_field(body, "subject")?, str_field(body, "body")?, None)?;
258                self.bump_mail();
259                Ok(json!({ "message_id": id }))
260            }
261            // Long-poll for HQ: returns as soon as the operator inbox holds a message newer than
262            // `since_ms`, or after `timeout_secs` (max 30) with an empty list.
263            ("POST", ["v1", "mail", "wait"]) => {
264                let since = body.get("since_ms").and_then(Value::as_u64).unwrap_or(0);
265                let deadline = Instant::now() + Duration::from_secs(body.get("timeout_secs").and_then(Value::as_u64).unwrap_or(25).min(30));
266                loop {
267                    let seen = *self.mail_signal.0.lock().expect("signal");
268                    let mail: Vec<_> = self.messenger.lock().expect("messenger").inbox(OPERATOR, since, 100)?.into_iter().filter(|m| m.created_at_ms > since).collect();
269                    if !mail.is_empty() {
270                        return Ok(json!({ "messages": mail }));
271                    }
272                    let now = Instant::now();
273                    if now >= deadline {
274                        return Ok(json!({ "messages": [] }));
275                    }
276                    let guard = self.mail_signal.0.lock().expect("signal");
277                    if *guard == seen {
278                        let _ = self.mail_signal.1.wait_timeout(guard, deadline - now).expect("signal wait");
279                    }
280                }
281            }
282            ("POST", ["v1", "mail", "thread"]) => {
283                let mail = self.messenger.lock().expect("messenger").thread(str_field(body, "node")?, 200)?;
284                Ok(json!({ "messages": mail }))
285            }
286            ("POST", ["v1", "mail", "inbox"]) => {
287                let mail = self.messenger.lock().expect("messenger").inbox(OPERATOR, 0, 100)?;
288                Ok(json!({ "messages": mail }))
289            }
290            _ => Err(HubError::new(404, "not found")),
291        }
292    }
293}
294
295#[cfg(test)]
296mod task_api_tests {
297    use super::*;
298
299    #[test]
300    fn operator_task_create_move_cancel_and_runs() {
301        let dir = tempfile::tempdir().unwrap();
302        let hub = Hub::open(dir.path()).unwrap();
303        let op = |m: &str, p: &str, b: Value| hub.operator_call(m, p, &b);
304        let t = op("POST", "/v1/tasks", json!({"title": "port", "role": "dev", "status": "backlog", "op_id": "c1"})).unwrap();
305        let id = t["task"]["id"].as_i64().unwrap();
306        // Idempotent create.
307        assert_eq!(op("POST", "/v1/tasks", json!({"title": "port", "op_id": "c1"})).unwrap(), t);
308        assert_eq!(op("GET", "/v1/tasks?role=advisor", json!({})).unwrap()["tasks"].as_array().unwrap().len(), 0);
309        assert_eq!(op("GET", "/v1/tasks?role=dev", json!({})).unwrap()["tasks"].as_array().unwrap().len(), 1);
310        let moved = op("POST", &format!("/v1/tasks/{id}/move"), json!({"status": "open", "expected_revision": 1})).unwrap();
311        assert_eq!(moved["task"]["revision"], 2);
312        assert_eq!(op("POST", &format!("/v1/tasks/{id}/move"), json!({"status": "open", "expected_revision": 1})).unwrap_err().status, 409);
313        // Runs need a recorded real session.
314        assert_eq!(op("POST", &format!("/v1/tasks/{id}/runs"), json!({"session_id": "s1"})).unwrap_err().status, 400);
315        op("POST", "/v1/sessions", json!({"session_id": "s1", "node_id": "alpha", "provider": "codex"})).unwrap();
316        let run = op("POST", &format!("/v1/tasks/{id}/runs"), json!({"session_id": "s1"})).unwrap();
317        let rid = run["run"]["id"].as_i64().unwrap();
318        // Node path is gated by the ledger.
319        assert_eq!(hub.node_call("alpha", "run.close", &json!({"run_id": rid, "status": "done", "session_id": "s1"})).unwrap_err().status, 403);
320        let closed = op("POST", &format!("/v1/runs/{rid}/close"), json!({"status": "cancelled"})).unwrap();
321        assert_eq!(closed["run"]["closed_by"], "operator");
322        let got = op("GET", &format!("/v1/tasks/{id}"), json!({})).unwrap();
323        assert_eq!(got["task"]["status"], "cancelled");
324        assert_eq!(got["runs"].as_array().unwrap().len(), 1);
325        assert_eq!(op("POST", &format!("/v1/tasks/{id}/move"), json!({"status": "open"})).unwrap_err().status, 409);
326        assert_eq!(op("GET", "/v1/sessions?node=alpha", json!({})).unwrap()["sessions"].as_array().unwrap().len(), 1);
327    }
328}
hatchery_server/hub.rs

hatchery_server/
hub.rs
