use chrono::Utc;
use serde::{Deserialize, Serialize};
use crate::console::{BootEvent, KernelEventLog, LogLevel};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SseEventType {
Init,
Ingest,
Delete,
Compact,
ChainAppend,
AgentSpawn,
AgentStop,
Heartbeat,
}
impl SseEventType {
pub fn as_str(&self) -> &'static str {
match self {
SseEventType::Init => "init",
SseEventType::Ingest => "ingest",
SseEventType::Delete => "delete",
SseEventType::Compact => "compact",
SseEventType::ChainAppend => "chain_append",
SseEventType::AgentSpawn => "agent_spawn",
SseEventType::AgentStop => "agent_stop",
SseEventType::Heartbeat => "heartbeat",
}
}
}
impl std::fmt::Display for SseEventType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SseMessage {
pub event_type: SseEventType,
pub data: serde_json::Value,
pub timestamp: String,
}
impl SseMessage {
pub fn new(event_type: SseEventType, data: serde_json::Value) -> Self {
Self {
event_type,
data,
timestamp: Utc::now().to_rfc3339(),
}
}
pub fn to_sse_frame(&self) -> String {
let json = serde_json::to_string(&SsePayload {
data: &self.data,
timestamp: &self.timestamp,
})
.unwrap_or_else(|_| "{}".to_string());
format!("event: {}\ndata: {}\n\n", self.event_type.as_str(), json)
}
pub fn heartbeat() -> Self {
Self::new(SseEventType::Heartbeat, serde_json::json!({}))
}
}
#[derive(Serialize)]
struct SsePayload<'a> {
data: &'a serde_json::Value,
timestamp: &'a str,
}
pub const SSE_HEARTBEAT_COMMENT: &str = ":heartbeat\n\n";
pub const SSE_HEARTBEAT_INTERVAL_SECS: u64 = 15;
pub fn classify_boot_event(event: &BootEvent) -> Option<SseEventType> {
let msg = &event.message;
if msg.contains("[agent]") && msg.contains("spawned") {
Some(SseEventType::AgentSpawn)
} else if msg.contains("[agent]") && msg.contains("stopped") {
Some(SseEventType::AgentStop)
} else if msg.contains("ingest") || msg.contains("ingested") {
Some(SseEventType::Ingest)
} else if msg.contains("delete") || msg.contains("deleted") {
Some(SseEventType::Delete)
} else if msg.contains("compact") {
Some(SseEventType::Compact)
} else if msg.contains("chain") && msg.contains("append") {
Some(SseEventType::ChainAppend)
} else if msg.contains("booting") || msg.contains("ready") {
Some(SseEventType::Init)
} else {
None
}
}
pub fn poll_events(event_log: &KernelEventLog, cursor: usize) -> (Vec<SseMessage>, usize) {
let all = event_log.tail(0);
let total = all.len();
if cursor >= total {
return (Vec::new(), total);
}
let new_events = &all[cursor..];
let messages: Vec<SseMessage> = new_events
.iter()
.filter(|e| e.level != LogLevel::Debug)
.map(|e| {
let event_type = classify_boot_event(e).unwrap_or(SseEventType::Init);
SseMessage::new(
event_type,
serde_json::json!({
"phase": e.phase.tag(),
"level": format!("{:?}", e.level).to_lowercase(),
"message": e.message,
}),
)
})
.collect();
(messages, total)
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WitnessRequest {
pub event_type: String,
pub payload: serde_json::Value,
pub signature: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WitnessResponse {
pub accepted: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub chain_hash: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub sequence: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
impl WitnessResponse {
pub fn accepted(hash_hex: String, sequence: u64) -> Self {
Self {
accepted: true,
chain_hash: Some(hash_hex),
sequence: Some(sequence),
error: None,
}
}
pub fn rejected(reason: impl Into<String>) -> Self {
Self {
accepted: false,
chain_hash: None,
sequence: None,
error: Some(reason.into()),
}
}
}
#[cfg(feature = "exochain")]
pub fn verify_witness_signature(
event_type: &str,
payload: &serde_json::Value,
signature_hex: &str,
verifying_key_bytes: &[u8; 32],
) -> Result<(), String> {
use ed25519_dalek::{Signature, Verifier, VerifyingKey};
let sig_bytes = hex_decode(signature_hex)
.map_err(|e| format!("invalid signature hex: {e}"))?;
if sig_bytes.len() != 64 {
return Err(format!(
"signature must be 64 bytes, got {}",
sig_bytes.len()
));
}
let sig = Signature::from_bytes(
sig_bytes
.as_slice()
.try_into()
.map_err(|_| "signature length mismatch")?,
);
let vk = VerifyingKey::from_bytes(verifying_key_bytes)
.map_err(|e| format!("invalid verifying key: {e}"))?;
let payload_bytes = serde_json::to_vec(payload).unwrap_or_default();
let mut message = Vec::with_capacity(event_type.len() + payload_bytes.len());
message.extend_from_slice(event_type.as_bytes());
message.extend_from_slice(&payload_bytes);
vk.verify(&message, &sig)
.map_err(|e| format!("signature verification failed: {e}"))
}
fn hex_decode(hex: &str) -> Result<Vec<u8>, String> {
if hex.len() % 2 != 0 {
return Err("odd-length hex string".into());
}
(0..hex.len())
.step_by(2)
.map(|i| {
u8::from_str_radix(&hex[i..i + 2], 16)
.map_err(|e| format!("invalid hex at offset {i}: {e}"))
})
.collect()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HttpMethod {
Get,
Post,
Delete,
}
impl HttpMethod {
pub fn parse(s: &str) -> Option<Self> {
match s.to_ascii_uppercase().as_str() {
"GET" => Some(HttpMethod::Get),
"POST" => Some(HttpMethod::Post),
"DELETE" => Some(HttpMethod::Delete),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FacadeRoute {
Events,
CustodyWitness,
Rpc {
method: &'static str,
params: Option<RpcParam>,
},
NotFound,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RpcParam {
Pid(u64),
}
pub fn match_facade_route(method: HttpMethod, path: &str) -> FacadeRoute {
let path = path.split('?').next().unwrap_or(path);
let path = path.trim_end_matches('/');
match (method, path) {
(HttpMethod::Get, "/events") => FacadeRoute::Events,
(HttpMethod::Post, "/custody/witness") => FacadeRoute::CustodyWitness,
(HttpMethod::Get, "/api/status") => FacadeRoute::Rpc {
method: "kernel.status",
params: None,
},
(HttpMethod::Get, "/api/processes") => FacadeRoute::Rpc {
method: "kernel.ps",
params: None,
},
(HttpMethod::Get, "/api/services") => FacadeRoute::Rpc {
method: "kernel.services",
params: None,
},
(HttpMethod::Get, "/api/chain/status") => FacadeRoute::Rpc {
method: "chain.status",
params: None,
},
(HttpMethod::Get, "/api/chain/events") => FacadeRoute::Rpc {
method: "kernel.logs",
params: None,
},
(HttpMethod::Get, "/api/vectors/status") => FacadeRoute::Rpc {
method: "ecc.status",
params: None,
},
(HttpMethod::Post, "/api/vectors/search") => FacadeRoute::Rpc {
method: "ecc.search",
params: None,
},
(HttpMethod::Get, "/api/ecc/calibration") => FacadeRoute::Rpc {
method: "ecc.calibrate",
params: None,
},
(HttpMethod::Get, "/api/custody/attest") => FacadeRoute::Rpc {
method: "custody.attest",
params: None,
},
(HttpMethod::Post, "/api/agents/spawn") => FacadeRoute::Rpc {
method: "agent.spawn",
params: None,
},
(HttpMethod::Delete, p) if p.starts_with("/api/agents/") => {
let pid_str = &p["/api/agents/".len()..];
match pid_str.parse::<u64>() {
Ok(pid) => FacadeRoute::Rpc {
method: "agent.stop",
params: Some(RpcParam::Pid(pid)),
},
Err(_) => FacadeRoute::NotFound,
}
}
_ => FacadeRoute::NotFound,
}
}
pub fn build_rpc_params(
route: &FacadeRoute,
body: &[u8],
query: &str,
) -> serde_json::Value {
match route {
FacadeRoute::Rpc {
method: "kernel.logs",
..
} => {
let count = parse_query_param(query, "count")
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(50);
serde_json::json!({ "count": count })
}
FacadeRoute::Rpc {
method: "agent.stop",
params: Some(RpcParam::Pid(pid)),
} => {
serde_json::json!({ "pid": pid, "graceful": true })
}
FacadeRoute::Rpc { .. } => {
serde_json::from_slice(body).unwrap_or(serde_json::json!({}))
}
_ => serde_json::json!({}),
}
}
fn parse_query_param<'a>(query: &'a str, key: &str) -> Option<&'a str> {
query
.split('&')
.find_map(|pair| {
let (k, v) = pair.split_once('=')?;
if k == key { Some(v) } else { None }
})
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FacadeResponse {
pub status: u16,
pub body: serde_json::Value,
}
impl FacadeResponse {
pub fn ok(body: serde_json::Value) -> Self {
Self { status: 200, body }
}
pub fn bad_request(message: impl Into<String>) -> Self {
Self {
status: 400,
body: serde_json::json!({ "error": message.into() }),
}
}
pub fn not_found() -> Self {
Self {
status: 404,
body: serde_json::json!({ "error": "not found" }),
}
}
pub fn internal_error(message: impl Into<String>) -> Self {
Self {
status: 500,
body: serde_json::json!({ "error": message.into() }),
}
}
pub fn unauthorized(message: impl Into<String>) -> Self {
Self {
status: 401,
body: serde_json::json!({ "error": message.into() }),
}
}
}
pub const CONTENT_TYPE_SSE: &str = "text/event-stream";
pub const CONTENT_TYPE_JSON: &str = "application/json";
pub fn sse_response_headers() -> Vec<(&'static str, &'static str)> {
vec![
("Content-Type", CONTENT_TYPE_SSE),
("Cache-Control", "no-cache"),
("Connection", "keep-alive"),
("X-Accel-Buffering", "no"),
]
}
#[cfg(test)]
mod tests {
use super::*;
use crate::console::{BootEvent, BootPhase, KernelEventLog};
#[test]
fn sse_message_format() {
let msg = SseMessage::new(
SseEventType::AgentSpawn,
serde_json::json!({"pid": 42, "agent_id": "coder-1"}),
);
let frame = msg.to_sse_frame();
assert!(frame.starts_with("event: agent_spawn\n"));
assert!(frame.contains("data: "));
assert!(frame.ends_with("\n\n"));
let data_line = frame
.lines()
.find(|l| l.starts_with("data: "))
.unwrap();
let json_str = &data_line["data: ".len()..];
let parsed: serde_json::Value = serde_json::from_str(json_str).unwrap();
assert!(parsed.get("data").is_some());
assert!(parsed.get("timestamp").is_some());
}
#[test]
fn sse_heartbeat_comment() {
assert_eq!(SSE_HEARTBEAT_COMMENT, ":heartbeat\n\n");
}
#[test]
fn sse_heartbeat_message() {
let msg = SseMessage::heartbeat();
assert_eq!(msg.event_type, SseEventType::Heartbeat);
let frame = msg.to_sse_frame();
assert!(frame.starts_with("event: heartbeat\n"));
}
#[test]
fn sse_event_type_display() {
assert_eq!(SseEventType::Init.as_str(), "init");
assert_eq!(SseEventType::ChainAppend.as_str(), "chain_append");
assert_eq!(SseEventType::AgentSpawn.as_str(), "agent_spawn");
assert_eq!(SseEventType::Heartbeat.as_str(), "heartbeat");
}
#[test]
fn classify_agent_spawn_event() {
let event = BootEvent::info(BootPhase::Ready, "[agent] spawned coder-1 (PID 5)");
assert_eq!(classify_boot_event(&event), Some(SseEventType::AgentSpawn));
}
#[test]
fn classify_agent_stop_event() {
let event = BootEvent::info(BootPhase::Ready, "[agent] stopped PID 5");
assert_eq!(classify_boot_event(&event), Some(SseEventType::AgentStop));
}
#[test]
fn classify_ingest_event() {
let event = BootEvent::info(BootPhase::Ready, "content ingested into store");
assert_eq!(classify_boot_event(&event), Some(SseEventType::Ingest));
}
#[test]
fn classify_compact_event() {
let event = BootEvent::info(BootPhase::Ready, "vector store compaction complete");
assert_eq!(classify_boot_event(&event), Some(SseEventType::Compact));
}
#[test]
fn classify_init_event() {
let event = BootEvent::info(BootPhase::Init, "WeftOS booting...");
assert_eq!(classify_boot_event(&event), Some(SseEventType::Init));
}
#[test]
fn classify_unknown_event() {
let event = BootEvent::info(BootPhase::Ready, "some random message");
assert_eq!(classify_boot_event(&event), None);
}
#[test]
fn poll_events_empty_log() {
let log = KernelEventLog::new();
let (messages, cursor) = poll_events(&log, 0);
assert!(messages.is_empty());
assert_eq!(cursor, 0);
}
#[test]
fn poll_events_with_cursor() {
let log = KernelEventLog::new();
log.info("agent", "spawned coder-1 (PID 5)");
log.info("test", "second event");
let (msgs, cursor) = poll_events(&log, 0);
assert_eq!(msgs.len(), 2);
assert_eq!(cursor, 2);
let (msgs, cursor2) = poll_events(&log, cursor);
assert!(msgs.is_empty());
assert_eq!(cursor2, 2);
log.info("agent", "stopped PID 5");
let (msgs, cursor3) = poll_events(&log, cursor2);
assert_eq!(msgs.len(), 1);
assert_eq!(cursor3, 3);
}
#[test]
fn witness_response_accepted() {
let resp = WitnessResponse::accepted("abcd1234".into(), 42);
assert!(resp.accepted);
assert_eq!(resp.chain_hash.as_deref(), Some("abcd1234"));
assert_eq!(resp.sequence, Some(42));
assert!(resp.error.is_none());
}
#[test]
fn witness_response_rejected() {
let resp = WitnessResponse::rejected("bad signature");
assert!(!resp.accepted);
assert!(resp.chain_hash.is_none());
assert_eq!(resp.error.as_deref(), Some("bad signature"));
}
#[test]
fn witness_request_serde() {
let req = WitnessRequest {
event_type: "audit.external".into(),
payload: serde_json::json!({"source": "client-A"}),
signature: "deadbeef".into(),
};
let json = serde_json::to_string(&req).unwrap();
let restored: WitnessRequest = serde_json::from_str(&json).unwrap();
assert_eq!(restored.event_type, "audit.external");
assert_eq!(restored.signature, "deadbeef");
}
#[test]
fn hex_decode_valid() {
let bytes = hex_decode("deadbeef").unwrap();
assert_eq!(bytes, vec![0xde, 0xad, 0xbe, 0xef]);
}
#[test]
fn hex_decode_empty() {
let bytes = hex_decode("").unwrap();
assert!(bytes.is_empty());
}
#[test]
fn hex_decode_odd_length() {
assert!(hex_decode("abc").is_err());
}
#[test]
fn hex_decode_invalid_chars() {
assert!(hex_decode("zzzz").is_err());
}
#[test]
fn route_events() {
assert_eq!(
match_facade_route(HttpMethod::Get, "/events"),
FacadeRoute::Events
);
}
#[test]
fn route_events_with_trailing_slash() {
assert_eq!(
match_facade_route(HttpMethod::Get, "/events/"),
FacadeRoute::Events
);
}
#[test]
fn route_custody_witness() {
assert_eq!(
match_facade_route(HttpMethod::Post, "/custody/witness"),
FacadeRoute::CustodyWitness
);
}
#[test]
fn route_api_status() {
let route = match_facade_route(HttpMethod::Get, "/api/status");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "kernel.status",
params: None
}
);
}
#[test]
fn route_api_processes() {
let route = match_facade_route(HttpMethod::Get, "/api/processes");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "kernel.ps",
params: None
}
);
}
#[test]
fn route_api_chain_status() {
let route = match_facade_route(HttpMethod::Get, "/api/chain/status");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "chain.status",
params: None
}
);
}
#[test]
fn route_api_chain_events() {
let route = match_facade_route(HttpMethod::Get, "/api/chain/events");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "kernel.logs",
params: None
}
);
}
#[test]
fn route_api_chain_events_with_query() {
let route = match_facade_route(HttpMethod::Get, "/api/chain/events?count=20");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "kernel.logs",
params: None
}
);
}
#[test]
fn route_api_vectors_status() {
let route = match_facade_route(HttpMethod::Get, "/api/vectors/status");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "ecc.status",
params: None
}
);
}
#[test]
fn route_api_vectors_search() {
let route = match_facade_route(HttpMethod::Post, "/api/vectors/search");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "ecc.search",
params: None
}
);
}
#[test]
fn route_api_agents_spawn() {
let route = match_facade_route(HttpMethod::Post, "/api/agents/spawn");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "agent.spawn",
params: None
}
);
}
#[test]
fn route_api_agents_delete() {
let route = match_facade_route(HttpMethod::Delete, "/api/agents/42");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "agent.stop",
params: Some(RpcParam::Pid(42))
}
);
}
#[test]
fn route_api_agents_delete_invalid_pid() {
let route = match_facade_route(HttpMethod::Delete, "/api/agents/notanumber");
assert_eq!(route, FacadeRoute::NotFound);
}
#[test]
fn route_not_found() {
assert_eq!(
match_facade_route(HttpMethod::Get, "/nonexistent"),
FacadeRoute::NotFound
);
}
#[test]
fn route_wrong_method() {
assert_eq!(
match_facade_route(HttpMethod::Post, "/api/status"),
FacadeRoute::NotFound
);
}
#[test]
fn route_ecc_calibration() {
let route = match_facade_route(HttpMethod::Get, "/api/ecc/calibration");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "ecc.calibrate",
params: None
}
);
}
#[test]
fn route_custody_attest() {
let route = match_facade_route(HttpMethod::Get, "/api/custody/attest");
assert_eq!(
route,
FacadeRoute::Rpc {
method: "custody.attest",
params: None
}
);
}
#[test]
fn build_params_kernel_logs_with_count() {
let route = FacadeRoute::Rpc {
method: "kernel.logs",
params: None,
};
let params = build_rpc_params(&route, b"", "count=25");
assert_eq!(params["count"], 25);
}
#[test]
fn build_params_kernel_logs_default_count() {
let route = FacadeRoute::Rpc {
method: "kernel.logs",
params: None,
};
let params = build_rpc_params(&route, b"", "");
assert_eq!(params["count"], 50);
}
#[test]
fn build_params_agent_stop() {
let route = FacadeRoute::Rpc {
method: "agent.stop",
params: Some(RpcParam::Pid(99)),
};
let params = build_rpc_params(&route, b"", "");
assert_eq!(params["pid"], 99);
assert_eq!(params["graceful"], true);
}
#[test]
fn build_params_from_body() {
let route = FacadeRoute::Rpc {
method: "agent.spawn",
params: None,
};
let body = br#"{"agent_id":"coder-1"}"#;
let params = build_rpc_params(&route, body, "");
assert_eq!(params["agent_id"], "coder-1");
}
#[test]
fn build_params_invalid_body_fallback() {
let route = FacadeRoute::Rpc {
method: "agent.spawn",
params: None,
};
let params = build_rpc_params(&route, b"not json", "");
assert_eq!(params, serde_json::json!({}));
}
#[test]
fn parse_query_param_found() {
assert_eq!(parse_query_param("count=20&level=info", "count"), Some("20"));
assert_eq!(parse_query_param("count=20&level=info", "level"), Some("info"));
}
#[test]
fn parse_query_param_not_found() {
assert_eq!(parse_query_param("count=20", "level"), None);
}
#[test]
fn parse_query_param_empty() {
assert_eq!(parse_query_param("", "count"), None);
}
#[test]
fn http_method_parse() {
assert_eq!(HttpMethod::parse("GET"), Some(HttpMethod::Get));
assert_eq!(HttpMethod::parse("get"), Some(HttpMethod::Get));
assert_eq!(HttpMethod::parse("POST"), Some(HttpMethod::Post));
assert_eq!(HttpMethod::parse("DELETE"), Some(HttpMethod::Delete));
assert_eq!(HttpMethod::parse("PATCH"), None);
}
#[test]
fn facade_response_ok() {
let resp = FacadeResponse::ok(serde_json::json!({"status": "running"}));
assert_eq!(resp.status, 200);
}
#[test]
fn facade_response_not_found() {
let resp = FacadeResponse::not_found();
assert_eq!(resp.status, 404);
}
#[test]
fn facade_response_bad_request() {
let resp = FacadeResponse::bad_request("missing field");
assert_eq!(resp.status, 400);
assert_eq!(resp.body["error"], "missing field");
}
#[test]
fn facade_response_unauthorized() {
let resp = FacadeResponse::unauthorized("invalid token");
assert_eq!(resp.status, 401);
}
#[test]
fn sse_headers() {
let headers = sse_response_headers();
assert!(headers.iter().any(|(k, v)| *k == "Content-Type" && *v == "text/event-stream"));
assert!(headers.iter().any(|(k, _)| *k == "Cache-Control"));
}
#[test]
fn content_types() {
assert_eq!(CONTENT_TYPE_SSE, "text/event-stream");
assert_eq!(CONTENT_TYPE_JSON, "application/json");
}
}