1use std::io::{BufRead, Write};
20use std::path::PathBuf;
21
22use serde_json::{Value, json};
23
24use crate::render::json_string;
25use crate::{EXIT_FAILURE, EXIT_OK, Rendered, doctor, explain, flows, init, status};
26
27const LATEST_PROTOCOL: &str = "2025-06-18";
30
31const SUPPORTED_PROTOCOLS: [&str; 3] = ["2024-11-05", "2025-03-26", "2025-06-18"];
35
36const INSTRUCTIONS: &str = "Keel adds production-grade resilience (retry, backoff, timeout, breaker, \
38rate limit, cache) and durable flows to this project with zero code changes; policy lives in keel.toml. \
39Every tool's text result is byte-identical to the matching CLI --json output: \
40get_status = `keel status --json`, get_doctor_report = `keel doctor --json`, \
41propose_policy = `keel init --diff --json` (an applyable keel.toml patch — never writes), \
42list_flows = `keel flows --json`, get_trace = `keel trace <flow> --json`, \
43explain_error = `keel explain <code> --json`. Outputs are deterministic \
44(sorted keys, no timestamps), so two calls can be diffed to see real change.";
45
46const PARSE_ERROR: i64 = -32700;
48const INVALID_REQUEST: i64 = -32600;
49const METHOD_NOT_FOUND: i64 = -32601;
50const INVALID_PARAMS: i64 = -32602;
51
52struct RpcError {
55 code: i64,
56 message: String,
57}
58
59fn rpc_error(code: i64, message: impl Into<String>) -> RpcError {
61 RpcError {
62 code,
63 message: message.into(),
64 }
65}
66
67#[derive(Debug)]
71pub struct Server {
72 project: PathBuf,
73 now_ms: fn() -> i64,
74}
75
76impl Server {
77 #[must_use]
79 pub fn new(project: PathBuf, now_ms: fn() -> i64) -> Self {
80 Self { project, now_ms }
81 }
82
83 pub fn serve<R: BufRead, W: Write>(&self, input: R, mut output: W) -> i32 {
87 for line in input.lines() {
88 let Ok(line) = line else {
89 return EXIT_FAILURE;
90 };
91 if line.trim().is_empty() {
92 continue;
93 }
94 let Some(response) = self.handle_line(&line) else {
95 continue;
96 };
97 let Ok(text) = serde_json::to_string(&response) else {
98 return EXIT_FAILURE;
99 };
100 if writeln!(output, "{text}")
101 .and_then(|()| output.flush())
102 .is_err()
103 {
104 return EXIT_FAILURE;
105 }
106 }
107 EXIT_OK
108 }
109
110 fn handle_line(&self, line: &str) -> Option<Value> {
113 let Ok(message) = serde_json::from_str::<Value>(line) else {
114 return Some(error_response(
115 &Value::Null,
116 PARSE_ERROR,
117 "Parse error: the line is not valid JSON. Send one JSON-RPC 2.0 message per line.",
118 ));
119 };
120 self.handle_message(&message)
121 }
122
123 fn handle_message(&self, message: &Value) -> Option<Value> {
125 let Some(frame) = message.as_object() else {
126 return Some(error_response(
127 &Value::Null,
128 INVALID_REQUEST,
129 "Invalid request: expected a JSON-RPC 2.0 object (batches are not supported).",
130 ));
131 };
132 let id = frame.get("id").cloned();
133 let Some(method) = frame.get("method").and_then(Value::as_str) else {
134 return id.map(|id| {
137 error_response(
138 &id,
139 INVALID_REQUEST,
140 "Invalid request: missing `method`. This server accepts initialize, ping, tools/list, and tools/call.",
141 )
142 });
143 };
144 let params = frame.get("params").cloned().unwrap_or(Value::Null);
145 let id = id?;
147 let outcome = match method {
148 "initialize" => Ok(initialize_result(¶ms)),
149 "ping" => Ok(json!({})),
150 "tools/call" => self.tools_call(¶ms),
151 "tools/list" => Ok(json!({ "tools": tool_catalog() })),
152 other => Err(rpc_error(
153 METHOD_NOT_FOUND,
154 format!(
155 "Method not found: {other:?}. This server supports initialize, ping, tools/list, and tools/call."
156 ),
157 )),
158 };
159 Some(match outcome {
160 Ok(result) => json!({ "id": id, "jsonrpc": "2.0", "result": result }),
161 Err(e) => error_response(&id, e.code, &e.message),
162 })
163 }
164
165 fn tools_call(&self, params: &Value) -> Result<Value, RpcError> {
170 let Some(name) = params.get("name").and_then(Value::as_str) else {
171 return Err(rpc_error(
172 INVALID_PARAMS,
173 "tools/call requires a string `name` parameter naming the tool.",
174 ));
175 };
176 let args = params
177 .get("arguments")
178 .cloned()
179 .unwrap_or_else(|| json!({}));
180 let rendered = self.call_tool(name, &args)?;
181 Ok(json!({
182 "content": [ { "text": json_string(&rendered.json), "type": "text" } ],
183 "isError": rendered.exit != EXIT_OK,
184 }))
185 }
186
187 fn call_tool(&self, name: &str, args: &Value) -> Result<Rendered, RpcError> {
191 match name {
192 "explain_error" => Ok(explain::run(require_str(args, "code", name)?)),
193 "get_doctor_report" => Ok(doctor::run(&self.project)),
194 "get_status" => Ok(status::run(&self.project, (self.now_ms)())),
195 "get_trace" => Ok(flows::trace(
196 &self.project,
197 require_str(args, "flow", name)?,
198 )),
199 "list_flows" => Ok(flows::flows(
200 &self.project,
201 optional_bool(args, "dead", name)?,
202 (self.now_ms)(),
203 )),
204 "propose_policy" => Ok(init::run(
205 &self.project,
206 init::InitOptions {
207 diff: true,
208 stamp: false,
209 agents: false,
210 },
211 )),
212 other => Err(rpc_error(
213 INVALID_PARAMS,
214 format!(
215 "Unknown tool: {other:?}. Available tools: explain_error, get_doctor_report, get_status, get_trace, list_flows, propose_policy."
216 ),
217 )),
218 }
219 }
220}
221
222fn initialize_result(params: &Value) -> Value {
225 let requested = params
226 .get("protocolVersion")
227 .and_then(Value::as_str)
228 .unwrap_or(LATEST_PROTOCOL);
229 let version = if SUPPORTED_PROTOCOLS.contains(&requested) {
230 requested
231 } else {
232 LATEST_PROTOCOL
233 };
234 json!({
235 "capabilities": { "tools": {} },
236 "instructions": INSTRUCTIONS,
237 "protocolVersion": version,
238 "serverInfo": { "name": "keel", "version": env!("CARGO_PKG_VERSION") },
239 })
240}
241
242pub const TOOL_NAMES: [&str; 6] = [
246 "explain_error",
247 "get_doctor_report",
248 "get_status",
249 "get_trace",
250 "list_flows",
251 "propose_policy",
252];
253
254fn tool_catalog() -> Value {
257 json!([
258 {
259 "description": "Explain a KEEL-E0NN error code: what happened, why, and what to do next. Byte-identical to `keel explain <code> --json`.",
260 "inputSchema": {
261 "properties": {
262 "code": { "description": "The error code, e.g. \"KEEL-E014\".", "type": "string" }
263 },
264 "required": ["code"],
265 "type": "object"
266 },
267 "name": "explain_error"
268 },
269 {
270 "description": "The honesty report: what is wrapped, what is visible but unwrapped and why, adapter pins, policy validity, and the journal backend — findings carry applyable fixes where possible. Byte-identical to `keel doctor --json`.",
271 "inputSchema": { "properties": {}, "type": "object" },
272 "name": "get_doctor_report"
273 },
274 {
275 "description": "One screen of what Keel is doing for this project: coverage, calls, retries saved, breaker opens, cache hit rate, and durable-flow counts. Byte-identical to `keel status --json`.",
276 "inputSchema": { "properties": {}, "type": "object" },
277 "name": "get_status"
278 },
279 {
280 "description": "Trace one durable (Tier 2) flow step by step: outcomes, attempts, timings. Byte-identical to `keel trace <flow> --json`.",
281 "inputSchema": {
282 "properties": {
283 "flow": { "description": "A flow_id, or a substring of an id/entrypoint that names exactly one flow.", "type": "string" }
284 },
285 "required": ["flow"],
286 "type": "object"
287 },
288 "name": "get_trace"
289 },
290 {
291 "description": "List durable (Tier 2) flows: id, entrypoint, status, steps done/total. Byte-identical to `keel flows --json`.",
292 "inputSchema": {
293 "properties": {
294 "dead": { "description": "List only dead flows (those that exhausted their resume cap).", "type": "boolean" }
295 },
296 "type": "object"
297 },
298 "name": "list_flows"
299 },
300 {
301 "description": "Propose policy changes as a keel.toml diff from static + observed evidence (never writes): an applyable unified patch plus structured changes. Byte-identical to `keel init --diff --json`.",
302 "inputSchema": { "properties": {}, "type": "object" },
303 "name": "propose_policy"
304 }
305 ])
306}
307
308fn error_response(id: &Value, code: i64, message: &str) -> Value {
310 json!({
311 "error": { "code": code, "message": message },
312 "id": id,
313 "jsonrpc": "2.0",
314 })
315}
316
317fn require_str<'a>(args: &'a Value, key: &str, tool: &str) -> Result<&'a str, RpcError> {
319 args.get(key).and_then(Value::as_str).ok_or_else(|| {
320 rpc_error(
321 INVALID_PARAMS,
322 format!("{tool} requires a string `{key}` argument."),
323 )
324 })
325}
326
327fn optional_bool(args: &Value, key: &str, tool: &str) -> Result<bool, RpcError> {
330 match args.get(key) {
331 None | Some(Value::Null) => Ok(false),
332 Some(Value::Bool(b)) => Ok(*b),
333 Some(_) => Err(rpc_error(
334 INVALID_PARAMS,
335 format!("{tool}'s `{key}` argument must be a boolean."),
336 )),
337 }
338}
339
340#[cfg(test)]
341mod tests {
342 use super::*;
343
344 const T0: i64 = 1_783_728_000_000;
345
346 fn t0() -> i64 {
347 T0
348 }
349
350 fn server_in(dir: &std::path::Path) -> Server {
351 Server::new(dir.to_path_buf(), t0)
352 }
353
354 fn empty_project() -> tempfile::TempDir {
355 tempfile::TempDir::new().unwrap()
356 }
357
358 fn respond(server: &Server, line: &str) -> Value {
360 server.handle_line(line).expect("a response is due")
361 }
362
363 #[test]
364 fn initialize_echoes_a_supported_version_and_names_the_server() {
365 let dir = empty_project();
366 let r = respond(
367 &server_in(dir.path()),
368 r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"t","version":"0"}}}"#,
369 );
370 assert_eq!(r["id"], 1);
371 assert_eq!(r["result"]["protocolVersion"], "2025-03-26");
372 assert_eq!(r["result"]["serverInfo"]["name"], "keel");
373 assert_eq!(
374 r["result"]["serverInfo"]["version"],
375 env!("CARGO_PKG_VERSION")
376 );
377 assert!(r["result"]["capabilities"]["tools"].is_object());
378 assert!(
379 r["result"]["instructions"]
380 .as_str()
381 .unwrap()
382 .contains("byte-identical")
383 );
384 }
385
386 #[test]
387 fn initialize_with_an_unknown_version_offers_the_latest_supported() {
388 let dir = empty_project();
389 let r = respond(
390 &server_in(dir.path()),
391 r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2099-01-01"}}"#,
392 );
393 assert_eq!(r["result"]["protocolVersion"], LATEST_PROTOCOL);
394 }
395
396 #[test]
397 fn notifications_get_no_response() {
398 let dir = empty_project();
399 let s = server_in(dir.path());
400 assert!(
401 s.handle_line(r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#)
402 .is_none()
403 );
404 assert!(
406 s.handle_line(r#"{"jsonrpc":"2.0","method":"notifications/whatever"}"#)
407 .is_none()
408 );
409 }
410
411 #[test]
412 fn parse_error_answers_with_null_id() {
413 let dir = empty_project();
414 let r = respond(&server_in(dir.path()), "{not json");
415 assert_eq!(r["error"]["code"], PARSE_ERROR);
416 assert!(r["id"].is_null());
417 }
418
419 #[test]
420 fn non_object_frames_are_invalid_requests() {
421 let dir = empty_project();
422 let r = respond(&server_in(dir.path()), "[1,2,3]");
423 assert_eq!(r["error"]["code"], INVALID_REQUEST);
424 }
425
426 #[test]
427 fn unknown_method_is_method_not_found() {
428 let dir = empty_project();
429 let r = respond(
430 &server_in(dir.path()),
431 r#"{"jsonrpc":"2.0","id":7,"method":"resources/list"}"#,
432 );
433 assert_eq!(r["error"]["code"], METHOD_NOT_FOUND);
434 assert_eq!(r["id"], 7);
435 }
436
437 #[test]
438 fn ping_answers_an_empty_object() {
439 let dir = empty_project();
440 let r = respond(
441 &server_in(dir.path()),
442 r#"{"jsonrpc":"2.0","id":2,"method":"ping"}"#,
443 );
444 assert_eq!(r["result"], json!({}));
445 }
446
447 #[test]
448 fn tools_list_is_the_six_spec_tools_alphabetically() {
449 let dir = empty_project();
450 let r = respond(
451 &server_in(dir.path()),
452 r#"{"jsonrpc":"2.0","id":3,"method":"tools/list"}"#,
453 );
454 let names: Vec<&str> = r["result"]["tools"]
455 .as_array()
456 .unwrap()
457 .iter()
458 .map(|t| t["name"].as_str().unwrap())
459 .collect();
460 assert_eq!(
461 names,
462 [
463 "explain_error",
464 "get_doctor_report",
465 "get_status",
466 "get_trace",
467 "list_flows",
468 "propose_policy",
469 ]
470 );
471 for tool in r["result"]["tools"].as_array().unwrap() {
473 assert_eq!(tool["inputSchema"]["type"], "object");
474 assert!(tool["description"].as_str().unwrap().contains("--json"));
475 }
476 }
477
478 #[test]
479 fn unknown_tool_is_invalid_params() {
480 let dir = empty_project();
481 let r = respond(
482 &server_in(dir.path()),
483 r#"{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"get_everything"}}"#,
484 );
485 assert_eq!(r["error"]["code"], INVALID_PARAMS);
486 assert!(
487 r["error"]["message"]
488 .as_str()
489 .unwrap()
490 .contains("Available tools")
491 );
492 }
493
494 #[test]
495 fn missing_required_argument_is_invalid_params() {
496 let dir = empty_project();
497 let s = server_in(dir.path());
498 let r = respond(
499 &s,
500 r#"{"jsonrpc":"2.0","id":5,"method":"tools/call","params":{"name":"get_trace"}}"#,
501 );
502 assert_eq!(r["error"]["code"], INVALID_PARAMS);
503 assert!(r["error"]["message"].as_str().unwrap().contains("`flow`"));
504 let r = respond(
505 &s,
506 r#"{"jsonrpc":"2.0","id":6,"method":"tools/call","params":{"name":"explain_error","arguments":{}}}"#,
507 );
508 assert_eq!(r["error"]["code"], INVALID_PARAMS);
509 assert!(r["error"]["message"].as_str().unwrap().contains("`code`"));
510 }
511
512 #[test]
513 fn mistyped_dead_argument_is_invalid_params_not_coerced() {
514 let dir = empty_project();
515 let r = respond(
516 &server_in(dir.path()),
517 r#"{"jsonrpc":"2.0","id":8,"method":"tools/call","params":{"name":"list_flows","arguments":{"dead":"yes"}}}"#,
518 );
519 assert_eq!(r["error"]["code"], INVALID_PARAMS);
520 assert!(r["error"]["message"].as_str().unwrap().contains("boolean"));
521 }
522
523 #[test]
524 fn explain_error_text_is_byte_identical_to_the_json_twin() {
525 let dir = empty_project();
526 let r = respond(
527 &server_in(dir.path()),
528 r#"{"jsonrpc":"2.0","id":9,"method":"tools/call","params":{"name":"explain_error","arguments":{"code":"KEEL-E014"}}}"#,
529 );
530 assert_eq!(r["result"]["isError"], false);
531 assert_eq!(
532 r["result"]["content"][0]["text"].as_str().unwrap(),
533 json_string(&explain::run("KEEL-E014").json)
534 );
535 assert_eq!(r["result"]["content"][0]["type"], "text");
536 }
537
538 #[test]
539 fn a_failing_tool_is_a_result_with_is_error_not_a_protocol_error() {
540 let dir = empty_project();
541 let r = respond(
542 &server_in(dir.path()),
543 r#"{"jsonrpc":"2.0","id":10,"method":"tools/call","params":{"name":"explain_error","arguments":{"code":"KEEL-E999"}}}"#,
544 );
545 assert!(r.get("error").is_none(), "tool failures are not RPC errors");
546 assert_eq!(r["result"]["isError"], true);
547 assert_eq!(
548 r["result"]["content"][0]["text"].as_str().unwrap(),
549 json_string(&explain::run("KEEL-E999").json)
550 );
551 }
552
553 #[test]
554 fn list_flows_defaults_dead_to_false_and_accepts_true() {
555 let dir = empty_project();
556 let s = server_in(dir.path());
557 let r = respond(
558 &s,
559 r#"{"jsonrpc":"2.0","id":11,"method":"tools/call","params":{"name":"list_flows"}}"#,
560 );
561 let text = r["result"]["content"][0]["text"].as_str().unwrap();
562 assert_eq!(text, json_string(&flows::flows(dir.path(), false, T0).json));
563 let r = respond(
564 &s,
565 r#"{"jsonrpc":"2.0","id":12,"method":"tools/call","params":{"name":"list_flows","arguments":{"dead":true}}}"#,
566 );
567 let text = r["result"]["content"][0]["text"].as_str().unwrap();
568 assert_eq!(text, json_string(&flows::flows(dir.path(), true, T0).json));
569 }
570
571 #[test]
572 fn serve_skips_blank_lines_and_exits_ok_on_eof() {
573 let dir = empty_project();
574 let script = "\n\
575 {\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"ping\"}\n\
576 \n\
577 {\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n";
578 let mut out = Vec::new();
579 let code = server_in(dir.path()).serve(std::io::Cursor::new(script), &mut out);
580 assert_eq!(code, EXIT_OK);
581 let lines: Vec<&str> = std::str::from_utf8(&out).unwrap().lines().collect();
582 assert_eq!(lines.len(), 1, "one request → one response line");
583 let v: Value = serde_json::from_str(lines[0]).unwrap();
584 assert_eq!(v["id"], 1);
585 }
586}