//! MCP server handler using rmcp's #[tool_router] macro.
//!
//! Each #[tool] method becomes an MCP tool that Hermes can discover and call.
//! The rmcp macro auto-generates JSON Schema from the parameter structs in tools.rs.
use std::collections::HashMap;
use std::sync::Mutex;
use std::time::{Duration, Instant};
use chrono::{SecondsFormat, Utc};
use rmcp::{
handler::server::{router::tool::ToolRouter, wrapper::Parameters},
tool, tool_handler, tool_router, ErrorData, Json, ServerHandler,
};
use serde_json::Value;
use std::path::PathBuf;
use crate::evidence::{digest, validate_witness_capture, WitnessCapture, WitnessError};
use crate::run_manager::{initial_state_for_input, RunBudgets, RunManager};
use crate::spec::{
ensure_size, parse_and_validate, validate_max_graphs, GraphSpec, DEFAULT_MAX_GRAPHS,
MAX_INPUT_BYTES,
};
use crate::store::{
ApprovalError, ApprovalRecord, CheckpointError, CheckpointRecord, GraphDeleteResult,
PersistentStore,
};
use crate::templates;
use crate::tools::*;
fn internal_error(message: impl Into<std::borrow::Cow<'static, str>>) -> ErrorData {
ErrorData::internal_error(message, None)
}
fn invalid_params(message: impl Into<std::borrow::Cow<'static, str>>) -> ErrorData {
ErrorData::invalid_params(message, None)
}
fn structured_output(value: Value) -> Json<StructuredOutput> {
Json(StructuredOutput {
ok: true,
status: None,
data: Some(value),
error: None,
error_code: None,
graph_id: None,
graph_version: None,
run_id: None,
})
}
fn error_output(message: impl Into<String>, code: impl Into<String>) -> Json<StructuredOutput> {
Json(StructuredOutput {
ok: false,
status: None,
data: None,
error: Some(message.into()),
error_code: Some(code.into()),
graph_id: None,
graph_version: None,
run_id: None,
})
}
fn structured_from_value(value: Value) -> Result<Json<StructuredOutput>, ErrorData> {
serde_json::from_value::<StructuredOutput>(value)
.map(Json)
.map_err(|e| internal_error(format!("cached idempotency decode: {e}")))
}
fn canonical_request_value(value: &Value) -> Value {
match value {
Value::String(raw) => serde_json::from_str(raw).unwrap_or_else(|_| value.clone()),
_ => value.clone(),
}
}
fn check_idempotency(
store: Option<&PersistentStore>,
key: Option<&str>,
request_digest: &str,
) -> Result<Option<Json<StructuredOutput>>, ErrorData> {
let Some((store, key)) = store.zip(key) else {
return Ok(None);
};
let Some((stored_digest, cached)) = store.check_idempotency(key).map_err(internal_error)?
else {
return Ok(None);
};
if stored_digest.as_deref() == Some(request_digest) {
return structured_from_value(cached).map(Some);
}
Ok(Some(error_output(
"idempotency key is already bound to different request material",
"IDEMPOTENCY_CONFLICT",
)))
}
fn persist_idempotency(
store: &PersistentStore,
key: &str,
request_digest: &str,
output: &Json<StructuredOutput>,
) -> Result<Option<Json<StructuredOutput>>, ErrorData> {
let result_json =
serde_json::to_string(&output.0).map_err(|e| internal_error(e.to_string()))?;
if store
.save_idempotency(key, request_digest, &result_json)
.map_err(internal_error)?
{
return Ok(None);
}
// Another request won the insert. Return its exact cached result so a
// concurrent same-key caller cannot observe a result that was not stored.
check_idempotency(Some(store), Some(key), request_digest)
}
fn output_with_meta(
data: Value,
graph_id: Option<&str>,
graph_version: Option<&str>,
run_id: Option<&str>,
) -> Json<StructuredOutput> {
Json(StructuredOutput {
ok: true,
status: None,
data: Some(data),
error: None,
error_code: None,
graph_id: graph_id.map(String::from),
graph_version: graph_version.map(String::from),
run_id: run_id.map(String::from),
})
}
fn checkpoint_error_output(error: CheckpointError) -> Json<StructuredOutput> {
error_output(error.message(), error.code())
}
fn approval_error_output(error: ApprovalError) -> Json<StructuredOutput> {
error_output(error.message(), error.code())
}
fn approval_value(record: &ApprovalRecord) -> Value {
serde_json::json!({
"approval_id": record.approval_id,
"checkpoint_id": record.checkpoint_id,
"run_id": record.run_id,
"graph_id": record.graph_id,
"graph_version": record.graph_version,
"checkpoint_digest": record.checkpoint_digest,
"audience": record.audience,
"prompt_digest": record.prompt_digest,
"allowed_decisions": record.allowed_decisions,
"approval_digest": record.approval_digest,
"status": record.status,
"decision": record.decision,
"decided_by": record.decided_by,
"decided_at": record.decided_at,
"expires_at": record.expires_at,
"created_at": record.created_at,
})
}
fn checkpoint_value(record: &CheckpointRecord) -> Value {
serde_json::json!({
"checkpoint_id": record.checkpoint_id,
"run_id": record.run_id,
"graph_id": record.graph_id,
"graph_version": record.graph_version,
"next_node_cursor": record.next_node_cursor,
"state": record.state,
"state_digest": record.state_digest,
"budgets": record.budgets,
"budget_counters": record.budget_counters,
"dependency_summary": record.dependency_summary,
"dependency_digest": record.dependency_digest,
"terminal_cursor": record.terminal_cursor,
"event_cursor": record.event_cursor,
"checkpoint_digest": record.checkpoint_digest,
"created_at": record.created_at,
"consumed_at": record.consumed_at,
"status": if record.consumed_at.is_some() { "consumed" } else { "available" },
"resume_capability": "deterministic_local_resume",
})
}
#[derive(Clone)]
struct RegisteredGraph {
spec: GraphSpec,
normalized: Value,
version: String,
warnings: Vec<String>,
}
pub struct AgentGraphServer {
tool_router: ToolRouter<Self>,
base_url: String,
default_model: String,
/// Provider API key for http(s) llm-pipeline calls. Attached as a Bearer
/// header only for the http(s) path; the codex-app-server:// path carries no
/// auth. Never serialized into status, receipts, or logs.
api_key: Option<String>,
graphs: Mutex<HashMap<String, RegisteredGraph>>,
runs: Mutex<RunManager>,
store: Option<PersistentStore>,
max_graphs: usize,
}
impl AgentGraphServer {
fn graph_requires_witness_store(spec: &GraphSpec) -> bool {
spec.nodes.iter().any(|node| node.evidence_required)
}
fn witness_error_output(error: WitnessError) -> Json<StructuredOutput> {
error_output(error.message, error.code)
}
pub fn new(
base_url: String,
default_model: String,
data_dir: Option<PathBuf>,
integrity_key_path: Option<PathBuf>,
) -> Result<Self, String> {
Self::new_with_max_graphs(
base_url,
default_model,
data_dir,
integrity_key_path,
DEFAULT_MAX_GRAPHS,
)
}
pub fn new_with_max_graphs(
base_url: String,
default_model: String,
data_dir: Option<PathBuf>,
integrity_key_path: Option<PathBuf>,
max_graphs: usize,
) -> Result<Self, String> {
Self::new_with_max_graphs_and_key(
base_url,
default_model,
data_dir,
integrity_key_path,
max_graphs,
None,
)
}
pub fn new_with_max_graphs_and_key(
base_url: String,
default_model: String,
data_dir: Option<PathBuf>,
integrity_key_path: Option<PathBuf>,
max_graphs: usize,
api_key: Option<String>,
) -> Result<Self, String> {
let max_graphs = validate_max_graphs(max_graphs)?;
let store = match data_dir {
Some(ref dir) => Some(PersistentStore::open_with_integrity_key(
dir,
integrity_key_path.as_deref(),
)?),
None => None,
};
if let Some(ref store) = store {
store.recover_incomplete_executions()?;
}
let runs = RunManager::default().with_api_key(api_key.clone());
let server = Self {
base_url,
default_model,
api_key,
graphs: Mutex::new(HashMap::new()),
runs: Mutex::new(runs),
store,
max_graphs,
tool_router: Self::tool_router(),
};
// Restore persisted graphs on startup
if let Some(ref store) = server.store {
if let Ok(graphs) = store.list_graphs() {
for (name, hash, _created) in graphs {
if let Ok(Some((spec_json, _))) = store.load_graph(&name) {
if let Ok(spec) = serde_json::from_str::<GraphSpec>(&spec_json) {
let normalized = serde_json::to_value(&spec).unwrap_or_default();
server.graphs.lock().unwrap().insert(
name,
RegisteredGraph {
spec,
normalized,
version: hash,
warnings: Vec::new(),
},
);
}
}
}
}
}
Ok(server)
}
fn safe_provider_label(&self) -> String {
let url = &self.base_url;
let without_fragment = url.split(['?', '#']).next().unwrap_or(url);
if let Some((scheme, rest)) = without_fragment.split_once("://") {
let authority_and_path = rest.rsplit_once('@').map(|(_, safe)| safe).unwrap_or(rest);
format!("{scheme}://{authority_and_path}")
} else {
"server-configured".into()
}
}
fn persist_terminal(
store: Option<PersistentStore>,
record: crate::run_manager::RunRecord,
) -> Result<(), String> {
let Some(store) = store else {
return Ok(());
};
let final_state = serde_json::to_string(&record.final_state)
.map_err(|e| format!("serialize terminal state error: {e}"))?;
// Persist one bounded terminal projection atomically. This is not replayable
// execution history and does not make the run resumable.
let events = record
.events
.iter()
.map(|entry| {
let seq = entry.get("cursor").and_then(Value::as_u64).unwrap_or(0);
let event = entry.get("event").cloned().unwrap_or_else(|| {
serde_json::json!({"receipt": "terminal event persisted with reduced fidelity"})
});
let event_type = event
.as_object()
.and_then(|object| object.keys().next().cloned())
.unwrap_or_else(|| "run_event".into());
Ok((seq, event_type, event.to_string()))
})
.collect::<Result<Vec<_>, String>>()?;
let mut durable_receipt = record.receipt.clone();
if let Some(object) = durable_receipt.as_object_mut() {
object.insert(
"persistence_status".into(),
Value::String("durable_terminal".into()),
);
}
let receipt = serde_json::to_string(&durable_receipt)
.map_err(|e| format!("serialize terminal receipt error: {e}"))?;
let durable_bundle = crate::evidence::bundle(
&record.run_id,
&record.graph_version,
&record.input,
&record.state,
&durable_receipt,
);
let bundle = serde_json::to_string(&durable_bundle)
.map_err(|e| format!("serialize terminal bundle error: {e}"))?;
store.persist_terminal_projection(
&record.run_id,
&record.status,
&final_state,
record.steps.len(),
&events,
&receipt,
&bundle,
)?;
Ok(())
}
fn persist_terminal_and_mark(
runs: crate::run_manager::RunManager,
store: Option<PersistentStore>,
record: crate::run_manager::RunRecord,
) {
if store.is_none() {
runs.mark_persistence(&record.run_id, "volatile_no_store", None);
return;
}
match Self::persist_terminal(store, record.clone()) {
Ok(()) => runs.mark_persistence(&record.run_id, "durable_terminal", None),
Err(error) => {
tracing::error!(%error, "terminal run persistence failed; run remains volatile");
runs.mark_persistence(&record.run_id, "volatile_persistence_failed", Some(error));
}
}
}
fn stored_run(&self, run_id: &str) -> Result<Option<Value>, ErrorData> {
let Some(store) = &self.store else {
return Ok(None);
};
let Some(mut record) = store.load_execution(run_id).map_err(internal_error)? else {
return Ok(None);
};
if let Some(receipt) = store
.load_terminal_receipt(run_id)
.map_err(internal_error)?
.and_then(|value| value.get("receipt").cloned())
{
if let Some(object) = record.as_object_mut() {
for key in ["budgets", "budget_counters", "budget_exhausted"] {
if let Some(value) = receipt.get(key) {
object.insert(key.into(), value.clone());
}
}
object.insert("receipt".into(), receipt);
}
}
Ok(Some(record))
}
fn resolve_graph(
&self,
graph_id: &str,
requested_version: Option<&str>,
) -> Result<RegisteredGraph, ErrorData> {
let current = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?
.get(graph_id)
.cloned()
.ok_or_else(|| invalid_params(format!("graph '{graph_id}' not found")))?;
let Some(requested_version) = requested_version else {
return Ok(current);
};
if requested_version == current.version {
return Ok(current);
}
let store = self.store.as_ref().ok_or_else(|| {
invalid_params("historical graph versions require SQLite persistence")
})?;
let serialized = store
.load_graph_version(graph_id, requested_version)
.map_err(internal_error)?
.ok_or_else(|| invalid_params("requested graph version was not found"))?;
let normalized: Value = serde_json::from_str(&serialized)
.map_err(|e| internal_error(format!("stored graph version JSON error: {e}")))?;
let spec = parse_and_validate(&normalized)
.map_err(|e| internal_error(format!("stored graph version validation error: {e}")))?;
let canonical = serde_json::to_value(&spec).map_err(|e| internal_error(e.to_string()))?;
let actual_version = digest(&canonical);
if actual_version != requested_version {
return Err(internal_error(
"stored graph version digest does not match its normalized specification",
));
}
Ok(RegisteredGraph {
warnings: spec.warnings(),
spec,
normalized: canonical,
version: actual_version,
})
}
fn mermaid(spec: &GraphSpec) -> String {
let mut s = String::from("graph TD\n");
for edge in &spec.edges {
s.push_str(&format!(" {} --> {}\n", edge.from, edge.to));
}
s
}
fn delete_registered_graph(&self, graph_id: &str) -> Result<Json<StructuredOutput>, ErrorData> {
let exists = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?
.contains_key(graph_id);
if !exists {
return Ok(error_output(
format!("graph '{graph_id}' not found"),
"GRAPH_NOT_FOUND",
));
}
if let Some(store) = &self.store {
match store.delete_graph(graph_id).map_err(internal_error)? {
GraphDeleteResult::Deleted => {}
GraphDeleteResult::Referenced => {
return Ok(error_output(
format!("graph '{graph_id}' is referenced by a durable execution"),
"GRAPH_REFERENCED",
));
}
GraphDeleteResult::ReferencedBySubgraph => {
return Ok(error_output(
format!(
"graph '{graph_id}' is referenced by another graph's subgraph node"
),
"GRAPH_SUBGRAPH_REFERENCED",
));
}
GraphDeleteResult::RetentionApprovalRequired => {
return Ok(error_output(
format!("graph '{graph_id}' requires delete_candidate then delete_approved retention states"),
"RETENTION_APPROVAL_REQUIRED",
));
}
GraphDeleteResult::NotFound => {
return Ok(error_output(
format!(
"graph '{graph_id}' is present in memory but missing from durable storage"
),
"GRAPH_PERSISTENCE_MISMATCH",
));
}
}
}
self.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?
.remove(graph_id);
Ok(output_with_meta(
serde_json::json!({"status": "deleted"}),
Some(graph_id),
None,
None,
))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::run_manager::RunManager;
use crate::store::PersistentStore;
fn configure_test_integrity_key() {
let path = std::env::temp_dir().join("agent-graph-mcp-unit-integrity.key");
std::fs::write(&path, [0x5au8; 32]).expect("test integrity key");
std::env::set_var("AGENT_GRAPH_INTEGRITY_KEY_PATH", path);
}
#[test]
fn terminal_projection_failure_rolls_back_sqlite_and_marks_run_volatile() {
configure_test_integrity_key();
let temp = tempfile::tempdir().expect("temp graph database");
let store = PersistentStore::open(temp.path()).expect("store");
let spec: GraphSpec = serde_json::from_value(serde_json::json!({
"name":"fault-injection",
"entry":"x",
"nodes":[{"id":"x","type":"passthrough"}],
"edges":[{"from":"x","to":"END"}]
}))
.expect("graph spec");
let spec_json = serde_json::to_string(&spec).expect("spec JSON");
store
.save_graph("fault-injection", &spec_json, "version", false)
.expect("graph");
let runs = RunManager::default();
let run_id = runs
.allocate("fault-injection", "version", serde_json::json!({"x":1}))
.expect("run");
store
.save_execution(
&run_id,
"fault-injection",
"version",
"running",
"{\"x\":1}",
)
.expect("execution");
runs.execute(
&run_id,
spec,
"http://localhost".into(),
"test-model".into(),
)
.expect("execution completes");
store.fail_terminal_projection_after_events();
AgentGraphServer::persist_terminal_and_mark(
runs.clone(),
Some(store.clone()),
runs.get(&run_id).expect("terminal record"),
);
let public = runs.get(&run_id).expect("volatile record").public();
assert_eq!(public["persistence_status"], "volatile_persistence_failed");
assert_eq!(public["storage_class"], "volatile");
let reopened = PersistentStore::open(temp.path()).expect("fresh store");
assert_eq!(
reopened.load_execution(&run_id).unwrap().unwrap()["status"],
"running"
);
assert!(reopened.load_events(&run_id, 0, 100).unwrap().is_none());
assert!(reopened.load_terminal_receipt(&run_id).unwrap().is_none());
}
#[test]
fn capacity_is_reserved_before_direct_or_approved_checkpoint_consumption() {
configure_test_integrity_key();
let temp = tempfile::tempdir().expect("checkpoint database");
let server = AgentGraphServer::new(
"http://localhost".into(),
"test-model".into(),
Some(temp.path().to_owned()),
None,
)
.expect("server");
server
.graph_create(Parameters(GraphCreateParams {
spec: Some(serde_json::json!({
"name":"capacity-resume", "entry":"first",
"nodes":[
{"id":"first","type":"passthrough"},
{"id":"second","type":"state_transform","config":{"operations":[{"op":"set","path":"done","value":true}]}}
],
"edges":[{"from":"first","to":"second"},{"from":"second","to":"END"}]
})),
action: None,
graph_id: None,
idempotency_key: None,
template: None,
overwrite: None,
}))
.expect("create graph");
let checkpoint = |server: &AgentGraphServer| {
server
.graph_run_start(Parameters(RunStartParams {
graph_id: "capacity-resume".into(),
input: None,
graph_version: None,
thread_id: None,
idempotency_key: None,
budgets: None,
checkpoint: Some(true),
}))
.expect("checkpoint start")
.0
.data
.unwrap()["checkpoint_id"]
.as_str()
.unwrap()
.to_owned()
};
let direct_checkpoint = checkpoint(&server);
{
let runs = server.runs.lock().expect("runs");
for index in 0..8 {
let run_id = runs
.allocate("capacity", "v1", serde_json::json!({"index":index}))
.expect("slot record");
runs.admit_async(&run_id).expect("slot admission");
}
}
let direct = server
.graph_run_resume(Parameters(RunResumeParams {
checkpoint_id: Some(direct_checkpoint.clone()),
run_id: None,
}))
.expect("resume response");
assert_eq!(direct.0.error_code.as_deref(), Some("RUN_CAPACITY"));
let store = server.store.as_ref().expect("store");
assert!(store
.load_resume_checkpoint(Some(&direct_checkpoint), None)
.expect("checkpoint")
.expect("record")
.consumed_at
.is_none());
let approval_checkpoint = checkpoint(&server);
let approval = server
.graph_approval_request(Parameters(ApprovalRequestParams {
checkpoint_id: approval_checkpoint.clone(),
audience: "operator".into(),
prompt: "approve after capacity is available".into(),
allowed_decisions: vec!["approve".into()],
expiration: (Utc::now() + chrono::Duration::hours(1)).to_rfc3339(),
}))
.expect("approval request");
let approval_id = approval.0.data.unwrap()["approval_id"]
.as_str()
.unwrap()
.to_owned();
let decided = server
.graph_approval_decide(Parameters(ApprovalDecideParams {
approval_id: approval_id.clone(),
decision: "approve".into(),
claimed_actor_label: "operator".into(),
}))
.expect("approval response");
assert_eq!(
decided.0.error_code.as_deref(),
Some("AUTHENTICATED_OPERATOR_REQUIRED")
);
assert_eq!(
store
.get_checkpoint_approval(&approval_id)
.expect("approval")
.expect("approval row")
.status,
"pending"
);
assert!(store
.load_resume_checkpoint(Some(&approval_checkpoint), None)
.expect("checkpoint")
.expect("record")
.consumed_at
.is_none());
}
}
#[tool_router]
impl AgentGraphServer {
// ── graph_create ──────────────────────────────────────────────────
#[tool(
description = "Create, validate, or delete a graph-orchestrated workflow from a JSON spec. Supports template instantiation and idempotency keys."
)]
fn graph_create(
&self,
Parameters(GraphCreateParams {
spec,
action,
graph_id,
idempotency_key,
template,
overwrite,
}): Parameters<GraphCreateParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let action = action.as_deref().unwrap_or("create");
let request_digest = digest(&serde_json::json!({
"operation": "graph_create",
"action": action,
"spec": spec.as_ref().map(canonical_request_value).unwrap_or(Value::Null),
"template": template.as_ref().map(canonical_request_value).unwrap_or(Value::Null),
"graph_id": graph_id,
"overwrite": overwrite.unwrap_or(false),
}));
if action != "delete" {
if let Some(cached) = check_idempotency(
self.store.as_ref(),
idempotency_key.as_deref(),
&request_digest,
)? {
return Ok(cached);
}
}
// Lifecycle deletion is never model-authorized. The authenticated
// operator IPC service is the sole mutation boundary and must perform
// its own request, peer, nonce, and state-digest validation before it
// reaches `delete_registered_graph`.
if action == "delete" {
return Ok(error_output(
"graph lifecycle deletion requires the authenticated operator service",
"AUTHENTICATED_OPERATOR_REQUIRED",
));
}
if action != "create" && action != "validate" {
return Ok(error_output(
format!("unsupported graph_create action '{action}'"),
"INVALID_ACTION",
));
}
// ── create / validate ──
let raw = if let Some(ref tpl) = template {
let tpl_val = if let Value::String(s) = tpl {
serde_json::from_str(s).unwrap_or_else(|_| tpl.clone())
} else {
tpl.clone()
};
let tpl_id = tpl_val
.get("id")
.and_then(Value::as_str)
.ok_or_else(|| invalid_params("template.id required"))?;
let tpl_name = tpl_val
.get("name")
.and_then(Value::as_str)
.or_else(|| graph_id.as_deref())
.unwrap_or(tpl_id);
templates::instantiate(tpl_id, tpl_name)
.map_err(|e| internal_error(format!("template error: {e}")))?
} else {
let spec = spec
.clone()
.ok_or_else(|| invalid_params("missing spec for create/validate"))?;
if let Value::String(s) = spec {
serde_json::from_str(&s)
.map_err(|e| invalid_params(format!("spec string parse error: {e}")))?
} else {
spec
}
};
let original_version = raw
.get("spec_version")
.and_then(Value::as_str)
.unwrap_or("1")
.to_owned();
let warnings_preview = serde_json::from_value::<GraphSpec>(raw.clone())
.ok()
.map(|s| s.warnings())
.unwrap_or_default();
let spec_parsed =
parse_and_validate(&raw).map_err(|e| invalid_params(format!("invalid spec: {e}")))?;
if let Some(node) = spec_parsed
.nodes
.iter()
.find(|node| crate::spec::GraphSpec::executable_node_type(&node.node_type).is_err())
{
return Ok(error_output(
format!("node '{}' declares an unsupported executable type", node.id),
"UNSUPPORTED_NODE_TYPE",
));
}
let normalized =
serde_json::to_value(&spec_parsed).map_err(|e| internal_error(e.to_string()))?;
let version = digest(&normalized);
let warnings = if original_version == "1" {
warnings_preview
} else {
spec_parsed.warnings()
};
if action == "validate" {
let output = output_with_meta(
serde_json::json!({
"graph_id": spec_parsed.name,
"graph_version": version,
"digest": version,
"normalized_spec_version": "2",
"warnings": warnings,
"storage_class": "volatile",
"status": "valid"
}),
Some(&spec_parsed.name),
Some(&version),
None,
);
if let Some(ref store) = self.store {
if let Some(idem) = idempotency_key {
if let Some(cached) =
persist_idempotency(store, &idem, &request_digest, &output)?
{
return Ok(cached);
}
}
}
return Ok(output);
}
if Self::graph_requires_witness_store(&spec_parsed) && self.store.is_none() {
return Ok(error_output(
"evidence-required graphs require SQLite witness persistence",
"WITNESS_STORE_REQUIRED",
));
}
// ── register ──
let mut graphs = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let name = spec_parsed.name.clone();
let overwrite = overwrite.unwrap_or(false);
if !overwrite && !graphs.contains_key(&name) && graphs.len() >= self.max_graphs {
return Ok(error_output(
format!("graph capacity ({}) exhausted — {} graphs registered", self.max_graphs, graphs.len()),
"CAPACITY_EXHAUSTED",
));
}
let id = name.clone();
if let Some(ref store) = self.store {
let spec_str = serde_json::to_string(&normalized).unwrap_or_default();
if let Err(error) = store.save_graph(&id, &spec_str, &version, overwrite) {
return Ok(error_output(error, "GRAPH_VERSION_CONFLICT"));
}
}
graphs.insert(
id.clone(),
RegisteredGraph {
spec: spec_parsed,
normalized: normalized.clone(),
version: version.clone(),
warnings: warnings.clone(),
},
);
drop(graphs);
let output = output_with_meta(
serde_json::json!({
"graph_id": id,
"graph_version": version,
"digest": version,
"normalized_spec_version": "2",
"warnings": warnings,
"storage_class": "volatile",
"status": "created"
}),
Some(&id),
Some(&version),
None,
);
if let Some(ref store) = self.store {
if let Some(idem) = idempotency_key {
if let Some(cached) = persist_idempotency(store, &idem, &request_digest, &output)? {
return Ok(cached);
}
}
}
Ok(output)
}
// ── graph_execute ─────────────────────────────────────────────────
#[tool(
description = "Execute a registered graph. Sync mode blocks until completion; async mode returns immediately with a run_id."
)]
fn graph_execute(
&self,
Parameters(GraphExecuteParams {
graph_id,
input,
graph_version,
thread_id,
mode,
idempotency_key,
}): Parameters<GraphExecuteParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let input = input.unwrap_or(Value::Null);
ensure_size(&input, MAX_INPUT_BYTES, "execution input").map_err(|e| invalid_params(e))?;
let sync = mode.as_deref().is_none_or(|m| m == "sync");
if sync {
tracing::warn!(
"graph_execute sync mode is deprecated — use graph_run_start + graph_run_get"
);
}
if let Some(store) = self.store.as_ref() {
if !store
.graph_execution_allowed(&graph_id)
.map_err(internal_error)?
{
return Ok(error_output(
format!("graph '{graph_id}' is archived or pending deletion"),
"GRAPH_RETIRED",
));
}
}
let graph = self.resolve_graph(&graph_id, graph_version.as_deref())?;
if Self::graph_requires_witness_store(&graph.spec) && self.store.is_none() {
return Ok(error_output(
"evidence-required graphs require SQLite witness persistence",
"WITNESS_STORE_REQUIRED",
));
}
let request_digest = digest(&serde_json::json!({
"operation": "graph_execute",
"graph_id": graph_id,
"graph_spec": graph.normalized,
"graph_version": graph.version,
"input": input,
"mode": mode.clone().unwrap_or_else(|| "sync".into()),
"thread_id": thread_id,
}));
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
if let Some(idem) = idempotency_key.as_deref() {
if let Some(cached) =
check_idempotency(self.store.as_ref(), Some(idem), &request_digest)?
{
return Ok(cached);
}
}
let run_id = runs
.allocate(&graph_id, &graph.version, input.clone())
.map_err(|e| internal_error(e))?;
if let Err(e) = runs.admit_async(&run_id) {
runs.remove(&run_id);
return Ok(error_output(e, "RUN_CAPACITY"));
}
if let Some(ref store) = self.store {
let _ = store.save_execution(
&run_id,
&graph_id,
&graph.version,
"running",
&input.to_string(),
);
}
let is_async = mode.as_deref() == Some("async");
if is_async {
let terminal_store = self.store.clone();
let completion_runs = runs.clone();
runs.start_with_completion_with_store(
run_id.clone(),
graph.spec,
self.base_url.clone(),
self.default_model.clone(),
self.store.clone(),
move |record| {
Self::persist_terminal_and_mark(completion_runs, terminal_store, record)
},
);
let output = output_with_meta(
serde_json::json!({
"run_id": run_id,
"status": "accepted",
"thread_id": thread_id,
"storage_class": "volatile",
"cancellation": "provider_future_best_effort_drop; underlying_request_may_continue"
}),
Some(&graph_id),
Some(&graph.version),
Some(&run_id),
);
if let Some(ref store) = self.store {
if let Some(idem) = idempotency_key {
if let Some(cached) =
persist_idempotency(store, &idem, &request_digest, &output)?
{
return Ok(cached);
}
}
}
return Ok(output);
}
let terminal_store = self.store.clone();
let completion_runs = runs.clone();
runs.start_with_completion_with_store(
run_id.clone(),
graph.spec,
self.base_url.clone(),
self.default_model.clone(),
self.store.clone(),
move |record| Self::persist_terminal_and_mark(completion_runs, terminal_store, record),
);
let deadline = Instant::now() + Duration::from_millis(300_000);
let output = loop {
let r = runs
.get(&run_id)
.ok_or_else(|| internal_error(format!("run '{run_id}' not found")))?;
if matches!(r.status.as_str(), "completed" | "failed" | "cancelled") {
break output_with_meta(
r.public(),
Some(&graph_id),
Some(&graph.version),
Some(&run_id),
);
}
if Instant::now() >= deadline {
let cancellation = runs.cancel(&run_id).unwrap_or_else(
|_| serde_json::json!({"run_id": run_id, "status": "cancellation_requested"}),
);
break output_with_meta(
serde_json::json!({
"run_id": run_id,
"status": r.status,
"timed_out": true,
"completion_unknown": true,
"cancellation": "requested",
"cancellation_result": cancellation,
}),
Some(&graph_id),
Some(&graph.version),
Some(&run_id),
);
}
std::thread::sleep(Duration::from_millis(100));
};
if let Some(ref store) = self.store {
let status = output
.0
.data
.as_ref()
.and_then(|data| data.get("status").and_then(Value::as_str))
.unwrap_or("failed");
let final_state = output
.0
.data
.as_ref()
.and_then(|data| data.get("final_state").cloned())
.map(|v| serde_json::to_string(&v).unwrap_or_default());
let _ = store.save_execution(
&run_id,
&graph_id,
&graph.version,
status,
&input.to_string(),
);
let _ =
store.update_execution_status(&run_id, status, final_state.as_deref(), None, None);
}
if let Some(ref store) = self.store {
if let Some(idem) = idempotency_key {
if let Some(cached) = persist_idempotency(store, &idem, &request_digest, &output)? {
return Ok(cached);
}
}
}
Ok(output)
}
// ── Local source witness capture ─────────────────────────────────
#[tool(
description = "Persist caller-supplied UTF-8 source content as a local witness receipt. The locator is metadata only; this tool never fetches or verifies it."
)]
fn graph_source_witness_capture(
&self,
Parameters(WitnessCaptureParams {
locator,
content,
media_type,
authority_class,
retrieved_at,
}): Parameters<WitnessCaptureParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let capture = WitnessCapture {
locator,
content,
media_type,
authority_class,
retrieved_at: retrieved_at
.unwrap_or_else(|| Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true)),
};
if let Err(error) = validate_witness_capture(capture.clone()) {
return Ok(Self::witness_error_output(error));
}
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for source witness capture",
"WITNESS_STORE_REQUIRED",
));
};
match store.capture_witness(capture) {
Ok(record) => Ok(structured_output(serde_json::json!({
"witness_id": record.witness_id,
"digest": record.digest,
"locator_digest": digest(&Value::String(record.locator)),
"media_type": record.media_type,
"authority_class": record.authority_class,
"retrieved_at": record.retrieved_at,
"content_bytes": record.content.len(),
"storage_class": "sqlite_source_witness"
}))),
Err(error) => Ok(Self::witness_error_output(error)),
}
}
#[tool(
description = "Read one exact local source witness ID, verifying its HMAC-SHA256 authentication tag before returning metadata and captured content."
)]
fn graph_source_witness_get(
&self,
Parameters(WitnessGetParams { witness_id }): Parameters<WitnessGetParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for source witness reads",
"WITNESS_STORE_REQUIRED",
));
};
match store.get_witness(&witness_id) {
Ok(Some(record)) => {
let locator_digest = digest(&Value::String(record.locator.clone()));
Ok(structured_output(serde_json::json!({
"witness_id": record.witness_id,
"digest": record.digest,
"locator": record.locator,
"locator_digest": locator_digest,
"content": record.content,
"media_type": record.media_type,
"authority_class": record.authority_class,
"retrieved_at": record.retrieved_at,
"storage_class": "sqlite_source_witness"
})))
}
Ok(None) => Ok(error_output(
"source witness was not found",
"WITNESS_NOT_FOUND",
)),
Err(error) => Ok(Self::witness_error_output(error)),
}
}
// ── graph_status ──────────────────────────────────────────────────
#[tool(
description = "Query server state, graph details, run status, events, receipts, or templates."
)]
fn graph_status(
&self,
Parameters(GraphStatusParams {
resource,
graph_id,
run_id,
cursor,
limit,
}): Parameters<GraphStatusParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let resource = resource.as_deref();
// Server-level summary (no resource or resource="server")
if resource.is_none() || resource == Some("server") {
let graphs = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let graph_names: Vec<&String> = graphs.keys().collect();
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let run_ids = runs.list();
let durable_integrity = self
.store
.as_ref()
.is_some_and(PersistentStore::has_integrity_key);
return Ok(structured_output(serde_json::json!({
"graphs": graph_names,
"graph_count": graphs.len(),
"execution_count": run_ids.len(),
"retained_execution_count": run_ids.len(),
"total_execution_count": run_ids.len(),
"base_url": self.safe_provider_label(),
"default_model": self.default_model,
"storage_class": if self.store.is_none() {
"process_local"
} else if durable_integrity {
"persisted_integrity_verified"
} else {
"persisted_unverified"
},
"capabilities": {
"runtime": "agent_graph",
"async_start": true,
"api_key_configured": self.api_key.is_some(),
"cancellation": "provider_future_best_effort_drop; underlying_request_may_continue",
"durable_resume": if durable_integrity {
Value::String("deterministic_local_resume_only".into())
} else {
Value::Bool(false)
},
"terminal_persistence": if durable_integrity { "sqlite_projection_only" } else { "disabled_without_integrity_key" },
"checkpointing": if durable_integrity { "deterministic_local_pre_execution" } else { "unavailable" },
"events": if self.store.is_some() { "terminal_persisted_projection_with_sqlite_fallback" } else { "volatile_in_memory_only" },
"event_replay": "not_replayable_execution",
"restart_recovery": "interrupted_non_resumable",
"budgets": {
"max_wall_clock_ms": "enforced",
"max_nodes": "enforced_at_engine_superstep_boundary",
"max_llm_calls": "enforced_before_provider_invocation"
},
"state_write_conflicts": "rejected_without_explicit_reducer",
"evidence": "witness_bound_local_capture_only; locators_not_fetched; source_authority_not_verified",
"evidence_authority": "caller_supplied_unverified_or_local_primary_capture",
"hitl": if durable_integrity { "checkpoint_bound_durable_approval_only" } else { "unavailable" },
"replay": "integrity_only"
},
"limits": {"graphs": self.max_graphs},
"capacity_state": if graphs.len() <= self.max_graphs {
"within_limit"
} else {
"over_limit_legacy"
}
})));
}
match resource.unwrap() {
"templates" => Ok(structured_output(templates::list())),
"graph" => {
let id = graph_id
.as_deref()
.ok_or_else(|| invalid_params("missing graph_id"))?;
let graphs = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let g = graphs
.get(id)
.ok_or_else(|| invalid_params(format!("graph '{id}' not found")))?;
Ok(output_with_meta(
serde_json::json!({
"graph_id": id,
"graph_version": g.version,
"normalized_spec": g.normalized,
"mermaid": Self::mermaid(&g.spec),
"warnings": g.warnings,
"storage_class": "volatile"
}),
Some(id),
Some(&g.version),
None,
))
}
"run" => {
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
if run_id.is_none() {
// List all runs
return Ok(structured_output(serde_json::json!({
"runs": runs.list()
})));
}
let id = run_id.as_deref().unwrap();
let r = runs
.get(id)
.ok_or_else(|| invalid_params(format!("run '{id}' not found")))?;
Ok(structured_output(r.public()))
}
"events" => {
let id = run_id
.as_deref()
.ok_or_else(|| invalid_params("missing run_id for events"))?;
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let cursor_val = cursor.unwrap_or(0);
let limit_val = limit.unwrap_or(100) as usize;
let result = runs
.events(self.store.as_ref(), id, cursor_val, limit_val)
.map_err(|e| invalid_params(e))?;
Ok(output_with_meta(result, None, None, Some(id)))
}
"receipt" => {
let id = run_id
.as_deref()
.ok_or_else(|| invalid_params("missing run_id for receipt"))?;
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let r = runs
.get(id)
.ok_or_else(|| invalid_params(format!("run '{id}' not found")))?;
Ok(output_with_meta(r.receipt.clone(), None, None, Some(id)))
}
"bundle" => {
let id = run_id
.as_deref()
.ok_or_else(|| invalid_params("missing run_id for bundle"))?;
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let r = runs
.get(id)
.ok_or_else(|| invalid_params(format!("run '{id}' not found")))?;
Ok(output_with_meta(r.bundle.clone(), None, None, Some(id)))
}
_ => Ok(error_output(
format!("unknown status resource '{}'", resource.unwrap_or("")),
"INVALID_RESOURCE",
)),
}
}
// ── graph_list (NEW) ──────────────────────────────────────────────
#[tool(
description = "List all registered graphs with metadata (name, node count, edge count, version)."
)]
fn graph_list(
&self,
Parameters(GraphListParams { query, limit }): Parameters<GraphListParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let graphs = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let mut entries: Vec<Value> = graphs
.iter()
.filter(|(name, _)| {
let visible = self
.store
.as_ref()
.map(|store| store.graph_is_tombstoned(name).map(|v| !v).unwrap_or(false))
.unwrap_or(true);
query
.as_ref()
.map(|q| name.contains(q.as_str()))
.unwrap_or(true)
&& visible
})
.take(limit.unwrap_or(50) as usize)
.map(|(name, g)| {
let version_history = self
.store
.as_ref()
.and_then(|store| store.list_graph_versions(name).ok())
.unwrap_or_else(|| vec![g.version.clone()]);
serde_json::json!({
"name": name,
"version": g.version,
"current_version": g.version,
"version_history": version_history,
"historical_specs": self.store.is_some(),
"node_count": g.spec.nodes.len(),
"edge_count": g.spec.edges.len(),
"entry": g.spec.entry,
"warnings": g.warnings,
})
})
.collect();
entries.sort_by(|a, b| {
a.get("name")
.and_then(Value::as_str)
.cmp(&b.get("name").and_then(Value::as_str))
});
Ok(structured_output(serde_json::json!({
"graphs": entries,
"count": entries.len(),
})))
}
#[tool(
description = "Read a durable graph retention inventory with execution, version, and inbound-subgraph reference counts. This tool never changes graph state."
)]
fn graph_retention_review(
&self,
Parameters(GraphRetentionReviewParams {
graph_id,
state,
limit,
}): Parameters<GraphRetentionReviewParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for graph retention review",
"RETENTION_STORE_REQUIRED",
));
};
let reports = store
.graph_retention_review(
graph_id.as_deref(),
state.as_deref(),
limit.unwrap_or(100).clamp(1, 256) as usize,
)
.map_err(internal_error)?;
let graphs: Vec<Value> = reports
.into_iter()
.map(|report| {
serde_json::json!({
"graph_id": report.graph_id,
"state": report.state,
"reason": report.reason,
"actor": report.actor,
"review_after": report.review_after,
"created_at": report.created_at,
"updated_at": report.updated_at,
"version_count": report.version_count,
"execution_count": report.execution_count,
"last_execution_at": report.last_execution_at,
"state_digest": report.state_digest,
"tombstoned": report.tombstoned,
"inbound_subgraph_refs": report.inbound_subgraph_refs,
"deletion_eligible": report.deletion_eligible,
})
})
.collect();
Ok(structured_output(serde_json::json!({
"graphs": graphs,
"count": graphs.len(),
"storage_class": "sqlite_graph_retention",
})))
}
#[tool(
description = "Set an explicit durable graph lifecycle state. delete_approved requires a prior delete_candidate state and no execution or inbound-subgraph references."
)]
fn graph_retention_set(
&self,
Parameters(GraphRetentionSetParams {
graph_id,
state,
reason,
actor,
review_after,
}): Parameters<GraphRetentionSetParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let _ = (graph_id, state, reason, actor, review_after);
Ok(error_output(
"graph lifecycle updates require the authenticated operator service",
"AUTHENTICATED_OPERATOR_REQUIRED",
))
}
// ── graph_delete (NEW) ────────────────────────────────────────────
#[allow(dead_code)]
fn graph_delete(
&self,
Parameters(GraphDeleteParams { graph_id }): Parameters<GraphDeleteParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
self.delete_registered_graph(&graph_id)
}
// ── graph_inspect (NEW) ───────────────────────────────────────────
#[tool(
description = "Get a graph's full topology: nodes, edges, Mermaid diagram, and topology hash."
)]
fn graph_inspect(
&self,
Parameters(GraphInspectParams { graph_id }): Parameters<GraphInspectParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let graphs = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let g = graphs
.get(&graph_id)
.ok_or_else(|| invalid_params(format!("graph '{graph_id}' not found")))?;
let nodes: Vec<Value> = g
.spec
.nodes
.iter()
.map(|n| {
serde_json::json!({
"id": n.id,
"type": n.node_type,
"config": n.config,
})
})
.collect();
let edges: Vec<Value> = g
.spec
.edges
.iter()
.map(|e| {
serde_json::json!({
"from": e.from,
"to": e.to,
})
})
.collect();
Ok(output_with_meta(
serde_json::json!({
"name": graph_id,
"version": g.version,
"current_version": g.version,
"version_history": self.store.as_ref().and_then(|store| store.list_graph_versions(&graph_id).ok()).unwrap_or_else(|| vec![g.version.clone()]),
"historical_specs": self.store.is_some(),
"entry": g.spec.entry,
"max_iterations": g.spec.max_iterations,
"max_parallelism": g.spec.max_parallelism,
"nodes": nodes,
"node_count": nodes.len(),
"edges": edges,
"edge_count": edges.len(),
"mermaid": Self::mermaid(&g.spec),
"topology_hash": g.version,
"reducers": g.spec.reducers,
"warnings": g.warnings,
}),
Some(&graph_id),
Some(&g.version),
None,
))
}
// ── Approval lifecycle ────────────────────────────────────────────
fn validate_resume_checkpoint(
&self,
store: &PersistentStore,
checkpoint: &CheckpointRecord,
) -> Result<
(
crate::store::ExecutionContract,
RegisteredGraph,
Option<RunBudgets>,
),
(String, String),
> {
let contract = store
.load_execution_contract(&checkpoint.run_id)
.map_err(|error| (error, "CHECKPOINT_PERSISTENCE_FAILURE".into()))?
.ok_or_else(|| {
(
"checkpoint execution contract was not found".into(),
"CHECKPOINT_INTEGRITY_FAILURE".into(),
)
})?;
if contract.graph_id != checkpoint.graph_id
|| contract.graph_version != checkpoint.graph_version
|| checkpoint.terminal_cursor != 0
|| checkpoint.event_cursor != 0
{
return Err((
"checkpoint integrity validation failed".into(),
"CHECKPOINT_INTEGRITY_FAILURE".into(),
));
}
let graph = self
.resolve_graph(&checkpoint.graph_id, Some(&checkpoint.graph_version))
.map_err(|_| {
(
"checkpoint graph version is unavailable".into(),
"CHECKPOINT_INTEGRITY_FAILURE".into(),
)
})?;
let eligibility = graph.spec.resume_eligibility().map_err(|_| {
(
"checkpoint graph is no longer in the deterministic local resume subset".into(),
"RESUME_INELIGIBLE".into(),
)
})?;
if graph.version != checkpoint.graph_version
|| checkpoint.next_node_cursor != eligibility.next_node_cursor
|| checkpoint.dependency_summary != eligibility.dependency_summary
|| checkpoint.dependency_digest != digest(&eligibility.dependency_summary)
|| checkpoint.state != initial_state_for_input(&contract.input)
|| checkpoint.budgets != contract.budgets
|| checkpoint.budget_counters
!= serde_json::json!({"nodes":0,"llm_calls":0,"wall_clock_ms":0})
{
return Err((
"checkpoint integrity validation failed".into(),
"CHECKPOINT_INTEGRITY_FAILURE".into(),
));
}
let budgets = RunBudgets::parse(Some(&checkpoint.budgets)).map_err(|_| {
(
"checkpoint budgets failed validation".into(),
"CHECKPOINT_INTEGRITY_FAILURE".into(),
)
})?;
Ok((contract, graph, budgets))
}
fn launch_resumed(
&self,
checkpoint: CheckpointRecord,
contract: crate::store::ExecutionContract,
graph: RegisteredGraph,
budgets: Option<RunBudgets>,
approval: Option<Value>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
if runs.get(&checkpoint.run_id).is_some() {
let _ = runs.remove(&checkpoint.run_id);
}
let run_id = match runs.allocate_resumed(
&checkpoint.run_id,
&checkpoint.graph_id,
&checkpoint.graph_version,
contract.input,
checkpoint.state.clone(),
budgets,
&checkpoint.checkpoint_id,
&checkpoint.checkpoint_digest,
approval.clone(),
) {
Ok(run_id) => run_id,
Err(error) => {
runs.release_async_slot();
return Ok(error_output(error, "RUN_CAPACITY"));
}
};
if let Err(error) = runs.admit_reserved_async(&run_id) {
runs.remove(&run_id);
runs.release_async_slot();
return Ok(error_output(error, "RUN_CAPACITY"));
}
self.store
.as_ref()
.expect("resumed launch requires SQLite")
.update_execution_status(&run_id, "running", None, None, None)
.map_err(internal_error)?;
let terminal_store = self.store.clone();
let completion_runs = runs.clone();
runs.start_resumed_with_completion(
run_id.clone(),
graph.spec,
self.base_url.clone(),
self.default_model.clone(),
self.store.clone(),
move |record| Self::persist_terminal_and_mark(completion_runs, terminal_store, record),
);
Ok(output_with_meta(
serde_json::json!({
"run_id": run_id,
"status": "running",
"checkpoint": checkpoint_value(&checkpoint),
"resume_capability": "deterministic_local_resume",
"approval": approval,
}),
Some(&checkpoint.graph_id),
Some(&checkpoint.graph_version),
Some(&run_id),
))
}
#[tool(
description = "Create a durable approval request bound to one unconsumed deterministic-local checkpoint."
)]
fn graph_approval_request(
&self,
Parameters(ApprovalRequestParams {
checkpoint_id,
audience,
prompt,
allowed_decisions,
expiration,
}): Parameters<ApprovalRequestParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for durable approvals",
"APPROVAL_STORE_REQUIRED",
));
};
if audience.trim().is_empty() || audience.len() > 256 {
return Ok(error_output(
"audience must be non-empty and at most 256 bytes",
"INVALID_PARAMS",
));
}
if allowed_decisions.is_empty()
|| allowed_decisions
.iter()
.any(|decision| !matches!(decision.as_str(), "approve" | "reject"))
{
return Ok(error_output(
"allowed_decisions must be a non-empty subset of approve and reject",
"INVALID_PARAMS",
));
}
if chrono::DateTime::parse_from_rfc3339(&expiration).is_err() {
return Ok(error_output("expiration must be RFC3339", "INVALID_PARAMS"));
}
if prompt.len() > 16 * 1024 {
return Ok(error_output(
"prompt exceeds the bounded approval prompt size",
"INVALID_PARAMS",
));
}
let checkpoint = match store.load_resume_checkpoint(Some(&checkpoint_id), None) {
Ok(Some(checkpoint)) => checkpoint,
Ok(None) => return Ok(checkpoint_error_output(CheckpointError::NotFound)),
Err(error) => return Ok(checkpoint_error_output(error)),
};
if checkpoint.consumed_at.is_some() {
return Ok(checkpoint_error_output(CheckpointError::Consumed));
}
if let Err((message, code)) = self.validate_resume_checkpoint(store, &checkpoint) {
return Ok(error_output(message, code));
}
let prompt_digest = digest(&Value::String(prompt));
let approval = match store.create_checkpoint_approval(
&checkpoint.checkpoint_id,
&checkpoint.graph_id,
&checkpoint.graph_version,
&checkpoint.next_node_cursor,
&checkpoint.state,
&checkpoint.budgets,
&checkpoint.budget_counters,
&checkpoint.dependency_summary,
&audience,
&prompt_digest,
&allowed_decisions,
&expiration,
) {
Ok(approval) => approval,
Err(error) => return Ok(approval_error_output(error)),
};
Ok(output_with_meta(
approval_value(&approval),
Some(&approval.graph_id),
Some(&approval.graph_version),
Some(&approval.run_id),
))
}
#[tool(
description = "Read durable checkpoint-bound approval metadata from SQLite without raw prompt or checkpoint state."
)]
fn graph_approval_list(
&self,
Parameters(ApprovalListParams {
run_id,
status,
limit,
}): Parameters<ApprovalListParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for durable approvals",
"APPROVAL_STORE_REQUIRED",
));
};
let approvals = store
.list_checkpoint_approvals(
run_id.as_deref(),
status.as_deref(),
limit.unwrap_or(50) as usize,
)
.map_err(|error| internal_error(error.message()))?;
Ok(structured_output(serde_json::json!({
"approvals": approvals.iter().map(approval_value).collect::<Vec<_>>(),
"count": approvals.len(),
"storage_class": "sqlite_durable_approval_metadata",
})))
}
#[tool(
description = "Read one durable checkpoint-bound approval's metadata from SQLite without raw prompt or checkpoint state."
)]
fn graph_approval_get(
&self,
Parameters(ApprovalGetParams { approval_id }): Parameters<ApprovalGetParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for durable approvals",
"APPROVAL_STORE_REQUIRED",
));
};
match store
.get_checkpoint_approval(&approval_id)
.map_err(|error| internal_error(error.message()))?
{
Some(approval) => Ok(output_with_meta(
approval_value(&approval),
Some(&approval.graph_id),
Some(&approval.graph_version),
Some(&approval.run_id),
)),
None => Ok(approval_error_output(ApprovalError::NotFound)),
}
}
#[allow(dead_code)]
fn graph_approval_decide(
&self,
Parameters(ApprovalDecideParams {
approval_id: _,
decision: _,
claimed_actor_label: _,
}): Parameters<ApprovalDecideParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
return Ok(error_output(
"approval decisions require authenticated operator transport",
"AUTHENTICATED_OPERATOR_REQUIRED",
));
}
// ── Async run lifecycle ───────────────────────────────────────────
#[tool(
description = "Start an async graph run. Returns run_id immediately; use graph_run_wait to block on completion. Optional budgets accept only positive integer max_wall_clock_ms, max_nodes, or max_llm_calls fields; max_llm_calls is enforced before each provider invocation."
)]
fn graph_run_start(
&self,
Parameters(RunStartParams {
graph_id,
input,
graph_version,
thread_id,
idempotency_key,
budgets,
checkpoint,
}): Parameters<RunStartParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let requested_budgets = match RunBudgets::parse(budgets.as_ref()) {
Ok(budgets) => budgets,
Err(error) => return Ok(error_output(error, "INVALID_BUDGETS")),
};
let input = input.unwrap_or(Value::Null);
let checkpoint_requested = checkpoint.unwrap_or(false);
ensure_size(&input, MAX_INPUT_BYTES, "execution input").map_err(|e| invalid_params(e))?;
let RegisteredGraph {
spec,
normalized,
version,
..
} = self.resolve_graph(&graph_id, graph_version.as_deref())?;
if Self::graph_requires_witness_store(&spec) && self.store.is_none() {
return Ok(error_output(
"evidence-required graphs require SQLite witness persistence",
"WITNESS_STORE_REQUIRED",
));
}
let eligibility = if checkpoint_requested {
match spec.resume_eligibility() {
Ok(eligibility) => Some(eligibility),
Err(reason) => return Ok(error_output(reason, "RESUME_INELIGIBLE")),
}
} else {
None
};
let request_digest = digest(&serde_json::json!({
"operation": "graph_run_start",
"graph_id": graph_id,
"graph_spec": normalized,
"graph_version": version,
"input": input,
"thread_id": thread_id,
"budgets": requested_budgets
.as_ref()
.map(RunBudgets::requested_value)
.unwrap_or(Value::Null),
"checkpoint": checkpoint_requested,
}));
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
if let Some(idem) = idempotency_key.as_deref() {
if let Some(cached) =
check_idempotency(self.store.as_ref(), Some(idem), &request_digest)?
{
return Ok(cached);
}
}
if checkpoint_requested {
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for deterministic checkpoints",
"CHECKPOINT_STORE_REQUIRED",
));
};
let eligibility = eligibility.expect("checkpoint eligibility");
let state = initial_state_for_input(&input);
let budgets_value = requested_budgets
.as_ref()
.map(RunBudgets::requested_value)
.unwrap_or(Value::Null);
let counters = serde_json::json!({"nodes":0,"llm_calls":0,"wall_clock_ms":0});
let run_id = runs
.allocate_with_budgets(
&graph_id,
&version,
input.clone(),
requested_budgets.clone(),
)
.map_err(|e| internal_error(e))?;
if let Err(error) = store.save_execution_with_budgets(
&run_id,
&graph_id,
&version,
"checkpointed",
&input.to_string(),
Some(&budgets_value.to_string()),
) {
runs.remove(&run_id);
return Ok(error_output(error, "CHECKPOINT_PERSISTENCE_FAILURE"));
}
let checkpoint_record = match store.create_resume_checkpoint(
&run_id,
&graph_id,
&version,
&eligibility.next_node_cursor,
&state,
&budgets_value,
&counters,
&eligibility.dependency_summary,
0,
0,
) {
Ok(record) => record,
Err(error) => {
let _ = store.update_execution_status(&run_id, "failed", None, None, None);
runs.remove(&run_id);
return Ok(checkpoint_error_output(error));
}
};
runs.mark_checkpointed(
&run_id,
&checkpoint_record.checkpoint_id,
&checkpoint_record.checkpoint_digest,
)
.map_err(internal_error)?;
let output = output_with_meta(
serde_json::json!({
"run_id": run_id,
"status": "checkpointed",
"thread_id": thread_id,
"checkpoint_id": checkpoint_record.checkpoint_id,
"checkpoint_digest": checkpoint_record.checkpoint_digest,
"checkpoint": checkpoint_value(&checkpoint_record),
"resume_capability": "deterministic_local_resume",
}),
Some(&graph_id),
Some(&version),
Some(&run_id),
);
if let Some(idem) = idempotency_key {
if let Some(cached) = persist_idempotency(store, &idem, &request_digest, &output)? {
return Ok(cached);
}
}
return Ok(output);
}
let run_id = runs
.allocate_with_budgets(&graph_id, &version, input.clone(), requested_budgets)
.map_err(|e| internal_error(e))?;
if let Err(e) = runs.admit_async(&run_id) {
runs.remove(&run_id);
return Ok(error_output(e, "RUN_CAPACITY"));
}
if let Some(ref store) = self.store {
let _ =
store.save_execution(&run_id, &graph_id, &version, "running", &input.to_string());
}
let terminal_store = self.store.clone();
let completion_runs = runs.clone();
runs.start_with_completion_with_store(
run_id.clone(),
spec,
self.base_url.clone(),
self.default_model.clone(),
self.store.clone(),
move |record| Self::persist_terminal_and_mark(completion_runs, terminal_store, record),
);
let output = output_with_meta(
serde_json::json!({
"run_id": run_id,
"status": "running",
"thread_id": thread_id,
}),
Some(&graph_id),
Some(&version),
Some(&run_id),
);
if let Some(ref store) = self.store {
if let Some(idem) = idempotency_key {
if let Some(cached) = persist_idempotency(store, &idem, &request_digest, &output)? {
return Ok(cached);
}
}
}
Ok(output)
}
#[tool(
description = "Read one durable deterministic-local checkpoint, including its integrity-bound state and resume metadata."
)]
fn graph_run_checkpoint(
&self,
Parameters(RunCheckpointParams {
run_id,
checkpoint_id,
}): Parameters<RunCheckpointParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for checkpoint reads",
"CHECKPOINT_STORE_REQUIRED",
));
};
if run_id.is_none() && checkpoint_id.is_none() {
return Ok(error_output(
"run_id or checkpoint_id is required for checkpoint reads",
"INVALID_PARAMS",
));
}
match store.load_resume_checkpoint(checkpoint_id.as_deref(), run_id.as_deref()) {
Ok(Some(record))
if run_id
.as_deref()
.is_none_or(|run_id| record.run_id == run_id) =>
{
Ok(output_with_meta(
checkpoint_value(&record),
Some(&record.graph_id),
Some(&record.graph_version),
Some(&record.run_id),
))
}
Ok(Some(_)) => Ok(checkpoint_error_output(CheckpointError::Integrity)),
Ok(None) => Ok(checkpoint_error_output(CheckpointError::NotFound)),
Err(error) => Ok(checkpoint_error_output(error)),
}
}
#[tool(
description = "Consume one deterministic-local checkpoint atomically and resume its pinned run exactly once."
)]
fn graph_run_resume(
&self,
Parameters(RunResumeParams {
checkpoint_id,
run_id,
}): Parameters<RunResumeParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let Some(store) = self.store.as_ref() else {
return Ok(error_output(
"SQLite persistence is required for deterministic resume",
"CHECKPOINT_STORE_REQUIRED",
));
};
if checkpoint_id.is_none() && run_id.is_none() {
return Ok(error_output(
"checkpoint_id or run_id is required for resume",
"INVALID_PARAMS",
));
}
let checkpoint =
match store.load_resume_checkpoint(checkpoint_id.as_deref(), run_id.as_deref()) {
Ok(Some(record)) => record,
Ok(None) => return Ok(checkpoint_error_output(CheckpointError::NotFound)),
Err(error) => return Ok(checkpoint_error_output(error)),
};
if store
.checkpoint_approval_status(&checkpoint.checkpoint_id)
.map_err(|error| internal_error(error.message()))?
.as_deref()
== Some("pending")
{
return Ok(error_output(
"checkpoint resume is pending its durable approval decision",
"APPROVAL_PENDING",
));
}
if checkpoint.consumed_at.is_some() {
return Ok(checkpoint_error_output(CheckpointError::Consumed));
}
if run_id
.as_deref()
.is_some_and(|run_id| run_id != checkpoint.run_id)
{
return Ok(checkpoint_error_output(CheckpointError::Integrity));
}
let Some(contract) = store
.load_execution_contract(&checkpoint.run_id)
.map_err(internal_error)?
else {
return Ok(checkpoint_error_output(CheckpointError::Integrity));
};
if contract.graph_id != checkpoint.graph_id
|| contract.graph_version != checkpoint.graph_version
|| checkpoint.terminal_cursor != 0
|| checkpoint.event_cursor != 0
{
return Ok(checkpoint_error_output(CheckpointError::Integrity));
}
let graph = match self.resolve_graph(&checkpoint.graph_id, Some(&checkpoint.graph_version))
{
Ok(graph) => graph,
Err(_) => return Ok(checkpoint_error_output(CheckpointError::Integrity)),
};
if graph.version != checkpoint.graph_version {
return Ok(checkpoint_error_output(CheckpointError::Integrity));
}
let eligibility = match graph.spec.resume_eligibility() {
Ok(eligibility) => eligibility,
Err(_) => {
return Ok(error_output(
"checkpoint graph is no longer in the deterministic local resume subset",
"RESUME_INELIGIBLE",
));
}
};
if checkpoint.next_node_cursor != eligibility.next_node_cursor
|| checkpoint.dependency_summary != eligibility.dependency_summary
|| checkpoint.dependency_digest != digest(&eligibility.dependency_summary)
|| checkpoint.state != initial_state_for_input(&contract.input)
|| checkpoint.budgets != contract.budgets
|| checkpoint.budget_counters
!= serde_json::json!({"nodes":0,"llm_calls":0,"wall_clock_ms":0})
{
return Ok(checkpoint_error_output(CheckpointError::Integrity));
}
let budgets = match RunBudgets::parse(Some(&checkpoint.budgets)) {
Ok(budgets) => budgets,
Err(_) => return Ok(checkpoint_error_output(CheckpointError::Integrity)),
};
let reserved_runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
if let Err(error) = reserved_runs.reserve_async_slot() {
return Ok(error_output(error, "RUN_CAPACITY"));
}
drop(reserved_runs);
let consumed = match store.consume_resume_checkpoint(&checkpoint.checkpoint_id) {
Ok(record) => record,
Err(error) => {
if let Ok(runs) = self.runs.lock() {
runs.release_async_slot();
}
return Ok(checkpoint_error_output(error));
}
};
self.launch_resumed(consumed, contract, graph, budgets, None)
}
#[tool(description = "Wait for an async run to complete, with optional timeout.")]
fn graph_run_wait(
&self,
Parameters(RunWaitParams { run_id, timeout_ms }): Parameters<RunWaitParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let timeout = Duration::from_millis(timeout_ms.unwrap_or(300_000));
let deadline = Instant::now() + timeout;
loop {
let r = {
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
runs.get(&run_id)
.ok_or_else(|| invalid_params(format!("run '{run_id}' not found")))?
};
if matches!(r.status.as_str(), "completed" | "failed" | "cancelled") {
let persist = Self::persist_terminal(self.store.clone(), r.clone());
if let Ok(runs) = self.runs.lock() {
if self.store.is_none() {
runs.mark_persistence(&run_id, "volatile_no_store", None);
} else {
match persist {
Ok(()) => runs.mark_persistence(&run_id, "durable_terminal", None),
Err(error) => runs.mark_persistence(
&run_id,
"volatile_persistence_failed",
Some(error),
),
}
}
}
let public = self
.runs
.lock()
.ok()
.and_then(|runs| runs.get(&run_id).map(|record| record.public()))
.unwrap_or_else(|| r.public());
return Ok(output_with_meta(public, None, None, Some(&run_id)));
}
if Instant::now() >= deadline {
return Ok(output_with_meta(
serde_json::json!({
"run_id": run_id,
"status": r.status,
"timed_out": true,
}),
None,
None,
Some(&run_id),
));
}
std::thread::sleep(Duration::from_millis(100));
}
}
#[tool(description = "Cancel a running execution.")]
fn graph_run_cancel(
&self,
Parameters(RunCancelParams { run_id, reason: _ }): Parameters<RunCancelParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
match runs.cancel(&run_id) {
Ok(_) => {}
Err(error) if error == "RUN_NOT_CANCELLABLE" => {
return Ok(error_output(
"terminal or checkpointed runs cannot be cancelled",
"RUN_NOT_CANCELLABLE",
));
}
Err(error) if error == "run not found" => {
drop(runs);
if self
.store
.as_ref()
.and_then(|store| store.load_execution(&run_id).ok().flatten())
.is_some_and(|stored| {
matches!(
stored.get("status").and_then(Value::as_str),
Some("completed" | "failed" | "cancelled" | "checkpointed")
)
})
{
return Ok(error_output(
"terminal or checkpointed runs cannot be cancelled",
"RUN_NOT_CANCELLABLE",
));
}
return Err(invalid_params(error));
}
Err(error) => return Err(invalid_params(error)),
}
Ok(output_with_meta(
serde_json::json!({
"run_id": run_id,
"status": "cancellation_requested",
"cancellation_effect": "best_effort_drop_provider_future",
"provider_request_may_still_be_in_flight": true,
"effective_at": "provider_completion_or_cancellation_observation"
}),
None,
None,
Some(&run_id),
))
}
#[tool(description = "Get current run status, budget usage, and pending approvals.")]
fn graph_run_get(
&self,
Parameters(RunGetParams { run_id }): Parameters<RunGetParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
if let Some(r) = runs.get(&run_id) {
return Ok(output_with_meta(r.public(), None, None, Some(&run_id)));
}
drop(runs);
if let Some(record) = self.stored_run(&run_id)? {
return Ok(output_with_meta(record, None, None, Some(&run_id)));
}
Err(invalid_params(format!("run '{run_id}' not found")))
}
#[tool(
description = "Read the in-memory state projection from a live run; use graph_run_checkpoint for a durable checkpoint state."
)]
fn graph_run_state(
&self,
Parameters(RunStateParams {
run_id,
checkpoint_id: _,
json_pointer,
}): Parameters<RunStateParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let r = runs
.get(&run_id)
.ok_or_else(|| invalid_params(format!("run '{run_id}' not found")))?;
let state = if let Some(pointer) = json_pointer.as_deref() {
if pointer.is_empty() {
r.state.clone()
} else {
r.state.pointer(pointer).cloned().unwrap_or(Value::Null)
}
} else {
r.state.clone()
};
Ok(output_with_meta(
serde_json::json!({
"state": state,
"run_id": run_id,
"status": r.status,
}),
None,
None,
Some(&run_id),
))
}
#[tool(
description = "Read bounded events. With SQLite, terminal emitted events remain available as a persisted projection after restart; this is not replayable execution or resume support."
)]
fn graph_run_events(
&self,
Parameters(RunEventsParams {
run_id,
cursor,
limit,
}): Parameters<RunEventsParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let runs = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let result = runs
.events(
self.store.as_ref(),
&run_id,
cursor.unwrap_or(0),
limit.unwrap_or(100) as usize,
)
.map_err(|e| invalid_params(e))?;
Ok(output_with_meta(result, None, None, Some(&run_id)))
}
#[tool(description = "Fetch the canonical execution receipt for a run.")]
fn graph_run_receipt(
&self,
Parameters(RunReceiptParams { run_id }): Parameters<RunReceiptParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
if let Some(r) = self
.runs
.lock()
.map_err(|e| internal_error(e.to_string()))?
.get(&run_id)
{
// Canonical wrapper: identical shape to the durable read-back, so
// consumers always read data.receipt. The HMAC receipt_digest is a
// persistence artifact, so it is null until the terminal projection
// is durably stored; storage_class marks the live-resident path.
return Ok(output_with_meta(
serde_json::json!({
"receipt": r.receipt.clone(),
"receipt_digest": Value::Null,
"storage_class": "volatile_live",
"replay_capability": r
.receipt
.get("replay_capability")
.and_then(Value::as_str)
.unwrap_or("integrity_only"),
}),
None,
None,
Some(&run_id),
));
}
if let Some(store) = &self.store {
match store.load_terminal_receipt(&run_id) {
Ok(Some(receipt)) => {
return Ok(output_with_meta(receipt, None, None, Some(&run_id)));
}
Ok(None) => {}
Err(error) if error == "RECEIPT_INTEGRITY_FAILURE" => {
return Ok(error_output(
"terminal receipt integrity validation failed",
"RECEIPT_INTEGRITY_FAILURE",
));
}
Err(error) if error == "INTEGRITY_KEY_REQUIRED" => {
return Ok(error_output(
"an external integrity key is required for terminal receipt reads",
"INTEGRITY_KEY_REQUIRED",
));
}
Err(error) => return Err(internal_error(error)),
}
}
Ok(error_output(
format!("run '{run_id}' not found"),
"RUN_NOT_FOUND",
))
}
// ── Policy + render ───────────────────────────────────────────────
#[tool(description = "Preflight a graph against policy before execution.")]
fn graph_policy_check(
&self,
Parameters(PolicyCheckParams { graph_id, input: _ }): Parameters<PolicyCheckParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let graphs = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let g = graphs
.get(&graph_id)
.ok_or_else(|| invalid_params(format!("graph '{graph_id}' not found")))?;
let node_count = g.spec.nodes.len();
let edge_count = g.spec.edges.len();
let issues: Vec<String> = Vec::new();
Ok(structured_output(serde_json::json!({
"graph_id": graph_id,
"passed": issues.is_empty(),
"issues": issues,
"stats": {
"node_count": node_count,
"edge_count": edge_count,
"max_iterations": g.spec.max_iterations,
"max_parallelism": g.spec.max_parallelism,
},
"capabilities": {
"models": [self.default_model.clone()],
"tools": [],
}
})))
}
#[tool(description = "Render a graph as Mermaid diagram or JSON topology.")]
fn graph_render(
&self,
Parameters(RenderParams { graph_id, format }): Parameters<RenderParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
let graphs = self
.graphs
.lock()
.map_err(|e| internal_error(e.to_string()))?;
let g = graphs
.get(&graph_id)
.ok_or_else(|| invalid_params(format!("graph '{graph_id}' not found")))?;
let fmt = format.as_deref().unwrap_or("mermaid");
match fmt {
"json" => Ok(output_with_meta(
serde_json::json!({
"name": graph_id,
"nodes": g.spec.nodes.iter().map(|n| serde_json::json!({
"id": n.id, "type": n.node_type
})).collect::<Vec<_>>(),
"edges": g.spec.edges.iter().map(|e| serde_json::json!({
"from": e.from, "to": e.to
})).collect::<Vec<_>>(),
}),
Some(&graph_id),
Some(&g.version),
None,
)),
_ => Ok(output_with_meta(
serde_json::json!({
"mermaid": Self::mermaid(&g.spec),
"name": graph_id,
}),
Some(&graph_id),
Some(&g.version),
None,
)),
}
}
// ── Templates ─────────────────────────────────────────────────────
#[tool(description = "List available built-in graph templates.")]
fn graph_template_list(
&self,
Parameters(TemplateListParams { query: _ }): Parameters<TemplateListParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
Ok(structured_output(templates::list()))
}
#[tool(
description = "Instantiate a template into a graph spec that can be passed to graph_create."
)]
fn graph_template_instantiate(
&self,
Parameters(TemplateInstantiateParams { template_id, name }): Parameters<
TemplateInstantiateParams,
>,
) -> Result<Json<StructuredOutput>, ErrorData> {
match templates::instantiate(&template_id, &name) {
Ok(spec) => Ok(structured_output(serde_json::json!({
"template_id": template_id,
"name": name,
"spec": spec,
}))),
Err(e) => Ok(error_output(e, "GRAPH_INVALID")),
}
}
#[tool(description = "Read-only list of template promotion candidates.")]
fn graph_template_candidates(
&self,
Parameters(TemplateCandidatesParams { state: _ }): Parameters<TemplateCandidatesParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
Ok(structured_output(serde_json::json!({ "candidates": [] })))
}
#[tool(description = "Read-only list of recorded outcomes for a template.")]
fn graph_template_outcomes(
&self,
Parameters(TemplateOutcomesParams { template_id }): Parameters<TemplateOutcomesParams>,
) -> Result<Json<StructuredOutput>, ErrorData> {
Ok(structured_output(serde_json::json!({
"template_id": template_id,
"outcomes": [],
})))
}
}
#[tool_handler(
router = self.tool_router,
name = "agent-graph-mcp",
version = "0.2.0",
instructions = "Graph orchestration for bounded multi-step LLM workflows with parallel fan-out, conditional routing, state transforms, joins, cooperative cancellation, and optional enforced max_wall_clock_ms/max_nodes/max_llm_calls run budgets (max_llm_calls reserves each provider attempt before invocation; failed or timed-out attempts still count). Parallel unordered state writes require an explicit reducer. Cancellation can drop the local provider future on request, best effort; an underlying provider request may continue. Optional SQLite stores terminal projections plus explicit pre-execution checkpoints. Durable checkpoints, approvals, terminal receipts, and source witnesses require an external key file named by AGENT_GRAPH_INTEGRITY_KEY_PATH; without it their operations fail closed with INTEGRITY_KEY_REQUIRED. Deterministic local resume is limited to linear passthrough/state_transform chains and is never generic replay; uncheckpointed or ineligible runs remain interrupted_non_resumable after restart. SQLite-backed approvals can decide only an immutable deterministic-local checkpoint and resume that checkpoint; HumanApproval nodes and arbitrary external actions remain unsupported. Source witnesses are caller-supplied local captures: locators are never fetched, HMAC-authenticated witness integrity and bounded evidence spans are checked against SQLite, and source authority is not independently verified. Receipts provide integrity_only except a successfully resumed deterministic-local path, which reports deterministic_local_resume. Define graphs with graph_create, execute with graph_execute or graph_run_start, checkpoint with checkpoint:true, inspect with graph_run_get/wait/cancel/state/events/receipt/checkpoint, request or decide checkpoint approvals with graph_approval_request/decide, and resume with graph_run_resume."
)]
impl ServerHandler for AgentGraphServer {}