diff --git a/Cargo.lock b/Cargo.lock index c5dcb508..58d23cbc 100755 --- a/Cargo.lock +++ b/Cargo.lock @@ -293,6 +293,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "arc-swap" +version = "1.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b" +dependencies = [ + "rustversion", +] + [[package]] name = "arg_enum_proc_macro" version = "0.3.4" @@ -553,12 +562,6 @@ dependencies = [ "stable_deref_trait", ] -[[package]] -name = "ascii" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16" - [[package]] name = "ascii-canvas" version = "3.0.0" @@ -904,7 +907,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", - "axum-core", + "axum-core 0.4.5", "base64 0.22.1", "bytes", "futures-util", @@ -914,7 +917,7 @@ dependencies = [ "hyper", "hyper-util", "itoa", - "matchit", + "matchit 0.7.3", "memchr", "mime", "percent-encoding", @@ -934,6 +937,39 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core 0.5.6", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit 0.8.4", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower 0.5.3", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "axum-core" version = "0.4.5" @@ -955,6 +991,25 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "b00t-iface" version = "1.10.0" @@ -1749,12 +1804,6 @@ dependencies = [ "windows-link 0.2.1", ] -[[package]] -name = "chunked_transfer" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" - [[package]] name = "cipher" version = "0.4.4" @@ -4988,6 +5037,29 @@ dependencies = [ "rustversion", ] +[[package]] +name = "lazily" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27e9027e68fe715eacafa564f1861eb1010a4b7848e51f478b6a8025194b8dfa" +dependencies = [ + "arc-swap", + "lazily-macros", + "parking_lot", + "smallvec 1.15.2", +] + +[[package]] +name = "lazily-macros" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cde497d1471dd81ea14cd762e82f1b7bbe35e43858cc8b4066775fb387e42fa" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -5175,12 +5247,14 @@ version = "1.10.0" dependencies = [ "agentmesh", "arc-kit-au", + "axum 0.8.9", "b00t-iface", "beankeeper-bridge", "blake3", "calamine", "cfg-if", "chrono", + "lazily", "ledger-core", "ledgerr-cloud", "ledgerr-focus", @@ -5202,8 +5276,9 @@ dependencies = [ "sysinfo 0.33.1", "tempfile", "thiserror 2.0.20", - "tiny_http", "tokio", + "tokio-stream", + "tower 0.5.3", "tracing", "ufo-types", "winreg 0.52.0", @@ -5282,7 +5357,7 @@ dependencies = [ name = "ledgrrr-sysml-adapter" version = "1.10.0" dependencies = [ - "axum", + "axum 0.7.9", "blake3", "ledgrrr-revision-io", "oxigraph", @@ -5664,6 +5739,12 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "matrixmultiply" version = "0.3.11" @@ -8215,7 +8296,7 @@ name = "rotel-visual" version = "1.10.0" dependencies = [ "anyhow", - "axum", + "axum 0.7.9", "b00t-iface", "ledger-core", "serde", @@ -10466,18 +10547,6 @@ dependencies = [ "crunchy", ] -[[package]] -name = "tiny_http" -version = "0.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "389915df6413a2e74fb181895f933386023c71110878cd0825588928e64cdc82" -dependencies = [ - "ascii", - "chunked_transfer", - "httpdate", - "log", -] - [[package]] name = "tinystr" version = "0.8.4" @@ -10626,6 +10695,7 @@ dependencies = [ "futures-core", "pin-project-lite", "tokio", + "tokio-util", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index feecab0a..f3762da8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -77,6 +77,7 @@ agentmesh = "=4.0.0" agentmesh-mcp = "=4.0.0" beankeeper = { git = "https://github.com/PromptExecution/beankeeper-b00t.git", branch = "feat/serde-feature", features = ["serde"] } ofx-rs = "0.2" +lazily = { version = "0.59", features = ["statechart", "thread-safe"] } [workspace.lints.rust] unsafe_code = "deny" diff --git a/crates/ledgerr-mcp/Cargo.toml b/crates/ledgerr-mcp/Cargo.toml index c62aa58b..93a2faf8 100755 --- a/crates/ledgerr-mcp/Cargo.toml +++ b/crates/ledgerr-mcp/Cargo.toml @@ -38,15 +38,18 @@ tokio = { workspace = true } # plugin_info — always available (cross-platform system metadata) sysinfo = { version = "0.33", default-features = false, features = ["system"] } +lazily = { workspace = true } # self-update feature deps — only compiled when the feature is active reqwest = { version = "0.12", features = ["blocking", "json", "rustls-tls"], default-features = false, optional = true } semver = { version = "1", optional = true } -# http-transport feature dep — minimal synchronous HTTP server, matches the -# existing sync dispatch model (no tokio runtime pulled into the request -# path). See src/bin/ledgerr-mcp-server.rs's serve_http(). -tiny_http = { version = "0.12", optional = true } +# http-transport feature deps — axum-based Streamable HTTP + SSE transport. +# tiny_http is removed; axum + tokio-stream replace it. Tokio is already a +# workspace dep (for budget_reconcile); these add the router and stream +# adapter needed for SSE server-initiated notifications (issue #227). +axum = { version = "0.8", optional = true } +tokio-stream = { version = "0.1", features = ["sync"], optional = true } # winreg — Windows-only, safe registry wrapper; gated at the cfg level in code [target.'cfg(windows)'.dependencies] @@ -76,7 +79,7 @@ self-update = ["dep:reqwest", "dep:semver"] # stdio transport (never replaces it — see infrastructure#139/#141 for the # Azure Container Apps scale-to-zero deployment this exists for). Off by # default: opt in with --features http-transport. -http-transport = ["dep:tiny_http"] +http-transport = ["dep:axum", "dep:tokio-stream"] # Legacy: gates the old mcp_adapter.rs dispatch and the pre-actor server # match-statement path. The new actor/gate channel system is the default; @@ -87,6 +90,8 @@ legacy = ["core"] [dev-dependencies] tempfile = "3" calamine = "0.34" +# Required for axum transport integration tests (ServiceExt::oneshot) +tower = { version = "0.5", features = ["util"] } # Enable the b00t feature for tests that exercise McpProvider infrastructure # and register_default_providers graceful-degradation coverage. diff --git a/crates/ledgerr-mcp/src/bin/ledgerr-mcp-server.rs b/crates/ledgerr-mcp/src/bin/ledgerr-mcp-server.rs index 7299bd9c..7ec2ead6 100644 --- a/crates/ledgerr-mcp/src/bin/ledgerr-mcp-server.rs +++ b/crates/ledgerr-mcp/src/bin/ledgerr-mcp-server.rs @@ -29,92 +29,40 @@ fn main() { serve(io::stdin().lock(), io::stdout()); } -/// HTTP transport: same `handle_request` dispatcher as stdio, reached over -/// a minimal synchronous HTTP server instead of stdin/stdout. Built for -/// Azure Container Apps' HTTP-triggered scale-to-zero (see -/// PromptExecution/infrastructure#139/#141) — ACA needs an HTTP endpoint to -/// wake the app on, which stdio transport can't provide. -/// -/// Deliberately does NOT adopt the `rmcp` SDK (tracked as the original ask -/// in this issue) — `handle_request` is already a pure, transport-agnostic -/// JSON-RPC dispatcher, so bridging it over HTTP needed zero changes to the -/// ~50 existing tool handlers or the well-tested stdio path. Migrating the -/// whole dispatcher onto `rmcp::ServerHandler` would be a much larger, -/// riskier rewrite (every tool handler + mcp_adapter's schema/routing) for -/// no functional gain over this — worth reconsidering only if a concrete -/// need for something `rmcp` provides (e.g. its SSE/StreamableHttp session -/// resumption) actually shows up. -/// -/// Only two routes: `GET /health` (liveness probe, matches the -/// `standing-mcp-server` Terraform module's expectation) and `POST /` -/// (JSON-RPC request body in, JSON-RPC response body out). Single-threaded, -/// one request at a time — matches this binary's existing fully-synchronous -/// dispatch model (see the tokio-on-a-scratch-runtime comment on the -/// `tokio` dependency above). Revisit if concurrent request handling is -/// ever actually needed. +/// HTTP transport: axum-based Streamable HTTP + SSE (issue #227). +/// Router built by `http_transport::build_http_app`; see that module for +/// route documentation and design notes. #[cfg(feature = "http-transport")] fn serve_http() { + use ledgerr_mcp::http_transport::build_http_app; + use tokio::sync::broadcast; + let port = std::env::var("PORT").unwrap_or_else(|_| "8080".to_string()); let addr = format!("0.0.0.0:{port}"); - let server = - tiny_http::Server::http(&addr).unwrap_or_else(|e| panic!("failed to bind {addr}: {e}")); - tracing::info!(%addr, "ledgerr-mcp HTTP transport listening"); - - for mut request in server.incoming_requests() { - let response = match (request.method(), request.url()) { - (tiny_http::Method::Get, "/health") => { - tiny_http::Response::from_string("ok").with_status_code(200) - } - (tiny_http::Method::Post, "/") => { - // Per-request caller identity (issue #224): HTTP is - // stateless/multi-tenant per process, so the `X-Agent-Id` - // header is read fresh for every request — unlike the stdio - // transport's `LEDGERR_MCP_AGENT_ID` env var, there is no - // single "the caller" for the life of the process here. - let agent_id = agent_id_from_headers(request.headers()); - let mut body = String::new(); - if request.as_reader().read_to_string(&mut body).is_err() { - let _ = request.respond( - tiny_http::Response::from_string("failed to read body") - .with_status_code(400), - ); - continue; - } - match serde_json::from_str::(&body) { - Ok(parsed) => match handle_request(parsed, agent_id.as_deref()) { - Some(resp) => match serde_json::to_string(&resp) { - Ok(s) => tiny_http::Response::from_string(s).with_status_code(200), - Err(e) => tiny_http::Response::from_string(format!( - "response serialization error: {e}" - )) - .with_status_code(500), - }, - // Notifications (e.g. notifications/initialized) have no response body. - None => tiny_http::Response::from_string("").with_status_code(204), - }, - Err(e) => tiny_http::Response::from_string(format!("invalid JSON: {e}")) - .with_status_code(400), - } - } - _ => tiny_http::Response::from_string("not found").with_status_code(404), - }; - let _ = request.respond(response); - } -} -/// Extract the calling agent's identity from an `X-Agent-Id` request header -/// (issue #224). Header names are compared case-insensitively per HTTP -/// semantics (`tiny_http::HeaderField::equiv`). Absent header → `None`, -/// which `authorize_tool_call` treats identically to "no identity -/// configured" — same fail-open default as the stdio transport. -#[cfg(feature = "http-transport")] -fn agent_id_from_headers(headers: &[tiny_http::Header]) -> Option { - headers - .iter() - .find(|h| h.field.equiv("X-Agent-Id")) - .map(|h| h.value.as_str().to_string()) + let (notify_tx, _) = broadcast::channel::(64); + let app = build_http_app( + |req, agent_id| handle_request(req, agent_id.as_deref()), + notify_tx, + ); + + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("tokio runtime for HTTP transport"); + + rt.block_on(async { + let listener = tokio::net::TcpListener::bind(&addr) + .await + .unwrap_or_else(|e| panic!("failed to bind {addr}: {e}")); + tracing::info!(%addr, "ledgerr-mcp HTTP transport listening (axum/SSE)"); + axum::serve(listener, app) + .await + .expect("axum serve must not fail"); + }); } + #[cfg(feature = "b00t")] fn initialize_providers() { let mut registry = McpProviderRegistry::new(); @@ -333,6 +281,9 @@ fn handle_request(request: Value, agent_id: Option<&str>) -> Option { let params = request.get("params").cloned().unwrap_or(Value::Null); let tool_name = params.get("name").and_then(Value::as_str).unwrap_or(""); + // Fire a hook for every tool/call so PM hooks observe all activity. + global_raw_service().fire_tool_event(tool_name, "dispatch", true); + // Issue #225: governance gate for tools/call. No-op (returns // None) when `agent_id` is `None` — same unenforced default as // before this change — or when `tool_name` isn't one of the @@ -540,6 +491,12 @@ fn handle_request(request: Value, agent_id: Option<&str>) -> Option { } } }; + + // Fire a `tool_called` project event for every successful tools/call. + // `isError` in the result content indicates dispatch-level failure. + let success = result.get("isError").and_then(|v| v.as_bool()) != Some(true); + global_raw_service().fire_tool_event(tool_name, "", success); + Some(json!({ "jsonrpc": "2.0", "id": id, diff --git a/crates/ledgerr-mcp/src/contract.rs b/crates/ledgerr-mcp/src/contract.rs index 79867b3e..51189634 100644 --- a/crates/ledgerr-mcp/src/contract.rs +++ b/crates/ledgerr-mcp/src/contract.rs @@ -111,7 +111,7 @@ pub const PUBLISHED_TOOLS: [ToolContractSpec; 14] = [ ToolContractSpec { name: WORKFLOW_TOOL, purpose: "lifecycle/HSM orchestration plus relocated plugin ops", - actions: &["status", "transition", "resume", "plugin_info"], + actions: &["status", "transition", "resume", "plugin_info", "task_status"], }, ToolContractSpec { name: AUDIT_TOOL, @@ -618,6 +618,45 @@ pub enum WorkflowArgs { #[serde(flatten)] payload: WorkflowPluginInfoInput, }, + /// Register a project-management event hook (fires on every named ProjectEvent). + RegisterProjectHook { + /// Event kind to subscribe to: `"tool_called"`, `"task_created"`, + /// `"task_done"`, `"lifecycle_transition"`, `"task_updated"`, or `"*"` for all events. + event_kind: String, + /// Optional caller correlation tag returned in the response. + #[serde(default, skip_serializing_if = "Option::is_none")] + tag: Option, + }, + /// Create a tracked task. Fires a `task_created` ProjectEvent. + TaskCreate { + /// Short machine-readable task name / slug. + name: String, + /// Human-readable description of what the task is for. + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option, + }, + /// List all tracked tasks. + TaskList, + /// Mark a task as done. Fires a `task_done` ProjectEvent. + TaskDone { + /// Task name that was returned by `task_create`. + name: String, + }, + /// Update a task's status. Fires a `task_updated` ProjectEvent. + /// Valid statuses: `"in_progress"`, `"blocked"`, `"done"`, `"pending"`. + TaskUpdate { + /// Task name. + name: String, + /// New status. + status: String, + /// Optional free-form note recorded alongside the status change. + #[serde(default, skip_serializing_if = "Option::is_none")] + note: Option, + }, + /// Return the current per-stage PM task status derived from statechart hooks. + /// Stages: ingest, normalize, validate, reconcile, commit, summarize. + /// Statuses: pending, blocked, ready, complete. + TaskStatus, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] diff --git a/crates/ledgerr-mcp/src/event_bus.rs b/crates/ledgerr-mcp/src/event_bus.rs new file mode 100644 index 00000000..4154187f --- /dev/null +++ b/crates/ledgerr-mcp/src/event_bus.rs @@ -0,0 +1,297 @@ +/// General project-management event bus for ledgrrr (issue #259 / PM hooks). +/// +/// `HsmHookRegistry` covers HSM lifecycle state transitions. This module +/// adds a broader event bus that fires on any named `ProjectEvent`: +/// - Tool calls (every `tools/call` dispatch) +/// - Task lifecycle (create/done/blocked — forwarded from b00t task commands) +/// - Lifecycle transitions (mirrored from HsmHookRegistry for cross-cutting hooks) +/// +/// The bus is held on `TurboLedgerService` as `Arc` and is +/// the single place to register hooks that span multiple event types. +use std::sync::{Arc, Mutex}; + +/// Payload for a project-level event. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProjectEvent { + /// Short machine-readable label, e.g. `"tool_called"`, `"task_created"`, + /// `"lifecycle_transition"`. + pub kind: String, + /// Arbitrary key=value context (tool name, task ID, lifecycle state, …). + pub attrs: Vec<(String, String)>, +} + +impl ProjectEvent { + /// Convenience: build a `tool_called` event. + pub fn tool_called(tool: &str, action: &str, success: bool) -> Self { + Self { + kind: "tool_called".to_string(), + attrs: vec![ + ("tool".to_string(), tool.to_string()), + ("action".to_string(), action.to_string()), + ("success".to_string(), success.to_string()), + ], + } + } + + /// Convenience: build a `task_created` event. + pub fn task_created(task_id: &str, description: &str) -> Self { + Self { + kind: "task_created".to_string(), + attrs: vec![ + ("task_id".to_string(), task_id.to_string()), + ("description".to_string(), description.to_string()), + ], + } + } + + /// Convenience: build a `task_done` event. + pub fn task_done(task_id: &str) -> Self { + Self { + kind: "task_done".to_string(), + attrs: vec![("task_id".to_string(), task_id.to_string())], + } + } + + /// Convenience: build a `task_updated` event. + pub fn task_updated(task_id: &str, status: &str, note: Option<&str>) -> Self { + let mut attrs = vec![ + ("task_id".to_string(), task_id.to_string()), + ("status".to_string(), status.to_string()), + ]; + if let Some(n) = note { + attrs.push(("note".to_string(), n.to_string())); + } + Self { kind: "task_updated".to_string(), attrs } + } + + /// Convenience: build a `lifecycle_transition` event. + pub fn lifecycle_transition(from: &str, to: &str) -> Self { + Self { + kind: "lifecycle_transition".to_string(), + attrs: vec![ + ("from".to_string(), from.to_string()), + ("to".to_string(), to.to_string()), + ], + } + } + + /// Look up an attribute value by key. + pub fn attr(&self, key: &str) -> Option<&str> { + self.attrs + .iter() + .find(|(k, _)| k == key) + .map(|(_, v)| v.as_str()) + } +} + +/// A registered project-event hook callback. +pub type ProjectHookFn = Arc; + +/// General-purpose event bus: register hooks that fire on any `ProjectEvent`. +/// +/// Hooks fire synchronously in registration order. Each hook sees every event +/// regardless of kind; use `event.kind` to filter inside the hook. +/// +/// Use `ProjectEventBus::fire_if` to fire only when `event.kind` matches a +/// specific value — the standard pattern for targeted hooks. +#[derive(Default)] +pub struct ProjectEventBus { + hooks: Mutex>, +} + +impl ProjectEventBus { + /// Register a hook that fires on every event. + pub fn register(&self, hook: ProjectHookFn) { + self.hooks + .lock() + .expect("ProjectEventBus lock poisoned") + .push(hook); + } + + /// Fire all registered hooks synchronously. + pub fn fire(&self, event: &ProjectEvent) { + for hook in self + .hooks + .lock() + .expect("ProjectEventBus lock poisoned") + .iter() + { + hook(event); + } + } + + /// Fire all hooks, but only if `event.kind == kind`. + pub fn fire_if(&self, kind: &str, event: &ProjectEvent) { + if event.kind == kind { + self.fire(event); + } + } + + pub fn len(&self) -> usize { + self.hooks + .lock() + .expect("ProjectEventBus lock poisoned") + .len() + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + +/// MCP-facing request to register a project-event hook. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RegisterProjectHookRequest { + /// The event kind to subscribe to: `"tool_called"`, `"task_created"`, + /// `"task_done"`, `"lifecycle_transition"`, or `"*"` for all events. + pub event_kind: String, + /// Optional tag stored in the response for caller correlation. + pub tag: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RegisterProjectHookResponse { + pub registered: bool, + pub event_kind: String, + pub tag: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + + #[test] + fn bus_starts_empty() { + let bus = ProjectEventBus::default(); + assert!(bus.is_empty()); + } + + #[test] + fn hook_fires_for_all_events() { + let bus = ProjectEventBus::default(); + let log: Arc>> = Arc::new(Mutex::new(Vec::new())); + let log_clone = Arc::clone(&log); + bus.register(Arc::new(move |evt: &ProjectEvent| { + log_clone.lock().unwrap().push(evt.kind.clone()); + })); + + bus.fire(&ProjectEvent::tool_called("ledgerr_workflow", "status", true)); + bus.fire(&ProjectEvent::task_done("task-42")); + + let log = log.lock().unwrap(); + assert_eq!(log.as_slice(), &["tool_called", "task_done"]); + } + + #[test] + fn fire_if_filters_by_kind() { + let bus = ProjectEventBus::default(); + let count: Arc> = Arc::new(Mutex::new(0)); + let count_clone = Arc::clone(&count); + bus.register(Arc::new(move |evt: &ProjectEvent| { + if evt.kind == "task_done" { + *count_clone.lock().unwrap() += 1; + } + })); + + bus.fire_if("tool_called", &ProjectEvent::tool_called("x", "y", true)); + assert_eq!(*count.lock().unwrap(), 0, "tool_called must not trigger task_done hook"); + + bus.fire_if("task_done", &ProjectEvent::task_done("t1")); + assert_eq!(*count.lock().unwrap(), 1); + } + + #[test] + fn project_event_attr_lookup() { + let evt = ProjectEvent::lifecycle_transition("ingest.pending", "normalize.ready"); + assert_eq!(evt.attr("from"), Some("ingest.pending")); + assert_eq!(evt.attr("to"), Some("normalize.ready")); + assert_eq!(evt.attr("missing"), None); + } + + #[test] + fn multiple_hooks_fire_in_order() { + let bus = ProjectEventBus::default(); + let order: Arc>> = Arc::new(Mutex::new(Vec::new())); + for i in 0u8..3 { + let o = Arc::clone(&order); + bus.register(Arc::new(move |_| { o.lock().unwrap().push(i); })); + } + bus.fire(&ProjectEvent::task_created("t1", "test task")); + assert_eq!(*order.lock().unwrap(), vec![0, 1, 2]); + } + + #[test] + fn task_created_event_has_correct_attrs() { + let evt = ProjectEvent::task_created("task-007", "classify ingested PDFs"); + assert_eq!(evt.kind, "task_created"); + assert_eq!(evt.attr("task_id"), Some("task-007")); + assert_eq!(evt.attr("description"), Some("classify ingested PDFs")); + } + + #[test] + fn task_done_event_has_correct_attrs() { + let evt = ProjectEvent::task_done("task-007"); + assert_eq!(evt.kind, "task_done"); + assert_eq!(evt.attr("task_id"), Some("task-007")); + } + + #[test] + fn task_updated_event_has_correct_attrs() { + let evt = ProjectEvent::task_updated("task-007", "in_progress", Some("unblocked by PR merge")); + assert_eq!(evt.kind, "task_updated"); + assert_eq!(evt.attr("task_id"), Some("task-007")); + assert_eq!(evt.attr("status"), Some("in_progress")); + assert_eq!(evt.attr("note"), Some("unblocked by PR merge")); + } + + #[test] + fn task_updated_event_no_note() { + let evt = ProjectEvent::task_updated("task-42", "blocked", None); + assert_eq!(evt.attr("note"), None); + } + + #[test] + fn tool_called_event_has_correct_attrs() { + let evt = ProjectEvent::tool_called("ledgerr_workflow", "transition", true); + assert_eq!(evt.kind, "tool_called"); + assert_eq!(evt.attr("tool"), Some("ledgerr_workflow")); + assert_eq!(evt.attr("action"), Some("transition")); + assert_eq!(evt.attr("success"), Some("true")); + } + + /// Every MCP tool dispatch fires a `tool_called` hook regardless of + /// which tool family is being called. This test proves the contract: + /// a registered hook sees ALL tool calls, not just task-management ones. + #[test] + fn hook_fires_for_every_tool_family() { + let bus = ProjectEventBus::default(); + let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); + let seen_clone = Arc::clone(&seen); + bus.register(Arc::new(move |evt: &ProjectEvent| { + if evt.kind == "tool_called" { + if let Some(t) = evt.attr("tool") { + seen_clone.lock().unwrap().push(t.to_string()); + } + } + })); + + // Simulate dispatch for every major tool family. + for tool in &[ + "ledgerr_documents", + "ledgerr_review", + "ledgerr_reconciliation", + "ledgerr_workflow", + "ledgerr_audit", + "ledgerr_tax", + "ledgerr_ontology", + "ledgerr_budget", + "ledgerr_gcp_billing", + ] { + bus.fire(&ProjectEvent::tool_called(tool, "dispatch", true)); + } + + let seen = seen.lock().unwrap(); + assert_eq!(seen.len(), 9, "expected hook to fire for all 9 tool families; got {seen:?}"); + } +} diff --git a/crates/ledgerr-mcp/src/hsm.rs b/crates/ledgerr-mcp/src/hsm.rs index 941ee99b..26461295 100644 --- a/crates/ledgerr-mcp/src/hsm.rs +++ b/crates/ledgerr-mcp/src/hsm.rs @@ -256,3 +256,132 @@ pub fn status_response(node: LifecycleNode, mut blockers: Vec) -> HsmSta blockers, } } + +/// Map a lifecycle node to the number of "advance" events needed to reach it +/// from the initial state (ingest.pending). Used to sync the HsmChart. +pub fn advance_count_for(node: LifecycleNode) -> usize { + match node.state { + LifecycleState::Ingest => 0, + LifecycleState::Normalize => 1, + LifecycleState::Validate => 2, + LifecycleState::Reconcile => 3, + LifecycleState::Commit => 4, + LifecycleState::Summarize => 5, + } +} + +/// Flat chart for PM event dispatch: states are HsmMachine tokens; each +/// non-initial state fires entry action "lifecycle_transition" on advance. +pub fn lifecycle_pm_chart_def() -> Result { + use lazily::statechart::{ChartBuilder, StateBuilder}; + ChartBuilder::new() + .state(StateBuilder::compound("lifecycle", "ingest.pending")) + .state( + StateBuilder::atomic("ingest.pending") + .parent("lifecycle") + .on("advance", "normalize.ready"), + ) + .state( + StateBuilder::atomic("normalize.ready") + .parent("lifecycle") + .entry("lifecycle_transition") + .on("advance", "validate.ready"), + ) + .state( + StateBuilder::atomic("validate.ready") + .parent("lifecycle") + .entry("lifecycle_transition") + .on("advance", "reconcile.ready"), + ) + .state( + StateBuilder::atomic("reconcile.ready") + .parent("lifecycle") + .entry("lifecycle_transition") + .on("advance", "commit.ready"), + ) + .state( + StateBuilder::atomic("commit.ready") + .parent("lifecycle") + .entry("lifecycle_transition") + .on("advance", "summarize.ready"), + ) + .state( + StateBuilder::atomic("summarize.ready") + .parent("lifecycle") + .entry("lifecycle_transition"), + ) + .build() +} + +/// Chart that mirrors HsmMachine advances and fires "lifecycle_transition" +/// entry actions. Held in TurboLedgerService to back PM hook dispatch. +pub struct HsmChart { + chart: lazily::statechart::ThreadSafeStateChart, + ctx: lazily::ThreadSafeContext, + /// Number of successful advances completed; used for fast-forward on restore. + advance_count: usize, +} + +impl HsmChart { + pub fn new() -> Result { + let def = lifecycle_pm_chart_def()?; + let ctx = lazily::ThreadSafeContext::new(); + let chart = lazily::statechart::ThreadSafeStateChart::new(&ctx, def); + Ok(Self { chart, ctx, advance_count: 0 }) + } + + /// Advance the chart from `from_advance_idx` → `from_advance_idx + 1`. + /// Fast-forwards silently if the chart is behind (restored session). + /// Returns `true` iff "lifecycle_transition" fired as an entry action. + pub fn try_advance(&mut self, from_advance_idx: usize) -> bool { + use std::collections::HashMap; + let guards: HashMap = HashMap::new(); + while self.advance_count < from_advance_idx { + self.chart.send(&self.ctx, "advance", &guards); + self.advance_count += 1; + } + let transitioned = self.chart.send(&self.ctx, "advance", &guards); + if transitioned { + self.advance_count += 1; + } + self.chart + .last_actions() + .contains(&"lifecycle_transition".to_string()) + } +} + +#[cfg(test)] +mod hsm_chart_tests { + use super::*; + + #[test] + fn hsm_chart_builds_without_error() { + let result = HsmChart::new(); + assert!(result.is_ok(), "HsmChart::new() failed: {:?}", result.err()); + } + + #[test] + fn hsm_chart_advance_from_initial_fires_lifecycle_transition() { + let mut chart = HsmChart::new().expect("chart"); + let fired = chart.try_advance(0); + assert!(fired, "lifecycle_transition must fire on first advance"); + } + + #[test] + fn hsm_chart_full_lifecycle_all_advances_fire_lifecycle_transition() { + let mut chart = HsmChart::new().expect("chart"); + for idx in 0..5 { + let fired = chart.try_advance(idx); + assert!(fired, "lifecycle_transition must fire at advance {idx}"); + } + } + + #[test] + fn hsm_chart_fast_forward_catches_up_and_fires() { + let mut chart = HsmChart::new().expect("chart"); + // Simulate a restored session: chart is at 0 but HSM is at Validate.Ready (idx=2) + let fired = chart.try_advance(2); + assert!(fired, "lifecycle_transition must fire after fast-forward"); + assert_eq!(chart.advance_count, 3); + } +} diff --git a/crates/ledgerr-mcp/src/http_transport.rs b/crates/ledgerr-mcp/src/http_transport.rs new file mode 100644 index 00000000..17784298 --- /dev/null +++ b/crates/ledgerr-mcp/src/http_transport.rs @@ -0,0 +1,350 @@ +/// Axum-based Streamable HTTP + SSE transport for the MCP server (issue #227). +/// +/// Exposes `build_http_app` so the server binary and integration tests share +/// exactly the same router code. The binary injects its own `handle_request` +/// function as the dispatch callback; tests inject a lightweight stub. +/// +/// Routes: +/// - `GET /health` — liveness probe (no auth, no state) +/// - `POST /` — JSON-RPC dispatch; response is `application/json` +/// - `GET /sse` — SSE stream; receives `notifications/tools/list_changed` +/// pushed when a POST / response carries the `_list_changed` +/// sentinel (issue #226/#227) +use std::sync::Arc; + +use axum::{ + Router, + body::Body, + http::{HeaderMap, Response, StatusCode}, + response::{ + IntoResponse, + sse::{Event, KeepAlive, Sse}, + }, + routing::{get, post}, +}; +use serde_json::{Value, json}; +use tokio::sync::broadcast; +use tokio_stream::StreamExt; +use tokio_stream::wrappers::BroadcastStream; + +/// Build the axum Router for the HTTP transport. +/// +/// `dispatch` is the synchronous JSON-RPC handler; it is called via +/// `spawn_blocking` so it never blocks the tokio runtime. Parameters: +/// `(request_body: Value, agent_id: Option) -> Option`. +/// +/// `notify_tx` is the broadcast sender for SSE events. Pass a fresh +/// `broadcast::channel(64).0` from the server binary; tests create their own. +pub fn build_http_app( + dispatch: F, + notify_tx: broadcast::Sender, +) -> Router +where + F: Fn(Value, Option) -> Option + Send + Sync + Clone + 'static, +{ + let dispatch = Arc::new(dispatch); + + Router::new() + .route("/health", get(|| async { (StatusCode::OK, "ok") })) + .route( + "/sse", + get({ + let tx = notify_tx.clone(); + move || { + let rx = tx.subscribe(); + let stream = BroadcastStream::new(rx).filter_map(|result| { + result.ok().map(|data| { + Ok::(Event::default().data(data)) + }) + }); + async move { Sse::new(stream).keep_alive(KeepAlive::default()) } + } + }), + ) + .route( + "/", + post({ + let tx = notify_tx.clone(); + move |headers: HeaderMap, body: String| { + let dispatch = Arc::clone(&dispatch); + let tx = tx.clone(); + async move { + let agent_id = agent_id_from_headers(&headers); + let parsed = match serde_json::from_str::(&body) { + Ok(v) => v, + Err(e) => { + return Response::builder() + .status(StatusCode::BAD_REQUEST) + .body(Body::from(format!("invalid JSON: {e}"))) + .unwrap() + .into_response(); + } + }; + + // Run the synchronous dispatcher off the tokio thread pool. + let result = tokio::task::spawn_blocking(move || { + dispatch(parsed, agent_id) + }) + .await + .unwrap_or(None); + + match result { + Some(mut resp) => { + // Strip `_list_changed` sentinel and push an SSE notification. + if let Some(obj) = resp.as_object_mut() { + if obj.remove("_list_changed").is_some() { + let notif = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "method": "notifications/tools/list_changed", + "params": {} + })) + .unwrap_or_default(); + let _ = tx.send(notif); + } + } + + let body_str = serde_json::to_string(&resp).unwrap_or_default(); + Response::builder() + .status(StatusCode::OK) + .header("content-type", "application/json") + .body(Body::from(body_str)) + .unwrap() + .into_response() + } + None => Response::builder() + .status(StatusCode::NO_CONTENT) + .body(Body::empty()) + .unwrap() + .into_response(), + } + } + } + }), + ) +} + +/// Extract calling agent identity from `X-Agent-Id` HTTP header (issue #224). +/// Case-insensitive; absent → `None` (fail-open default). +pub fn agent_id_from_headers(headers: &HeaderMap) -> Option { + headers + .get("x-agent-id") + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::body::Body; + use axum::http::{Request, StatusCode}; + use tower::ServiceExt; + + /// Minimal stub handler for transport-layer tests. + fn stub_dispatch(req: Value, _agent_id: Option) -> Option { + let method = req.get("method").and_then(Value::as_str).unwrap_or(""); + let id = req.get("id").cloned().unwrap_or(Value::Null); + match method { + "tools/list" => Some(json!({ + "jsonrpc": "2.0", + "id": id, + "result": { "tools": [] } + })), + "notifications/initialized" => None, + // Simulate a tool call that triggers _list_changed. + "tools/call" => Some(json!({ + "jsonrpc": "2.0", + "id": id, + "_list_changed": true, + "result": { "content": [{"type": "text", "text": "ok"}] } + })), + other => Some(json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": -32601, "message": format!("method not found: {other}") } + })), + } + } + + fn test_app() -> (Router, broadcast::Sender) { + let (tx, _) = broadcast::channel::(16); + let app = build_http_app(stub_dispatch, tx.clone()); + (app, tx) + } + + #[tokio::test] + async fn health_returns_200() { + let (app, _) = test_app(); + let response = app + .oneshot( + Request::builder() + .uri("/health") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + } + + #[tokio::test] + async fn post_tools_list_returns_json() { + let (app, _) = test_app(); + let body = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/list" + })) + .unwrap(); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/") + .header("content-type", "application/json") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let ct = response + .headers() + .get("content-type") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + assert!(ct.contains("application/json"), "content-type was: {ct}"); + } + + #[tokio::test] + async fn notification_returns_204() { + let (app, _) = test_app(); + let body = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "method": "notifications/initialized" + })) + .unwrap(); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NO_CONTENT); + } + + #[tokio::test] + async fn sse_endpoint_returns_event_stream_content_type() { + let (app, _) = test_app(); + let response = app + .oneshot( + Request::builder() + .uri("/sse") + .header("accept", "text/event-stream") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let ct = response + .headers() + .get("content-type") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + assert!( + ct.contains("text/event-stream"), + "expected text/event-stream, got: {ct}" + ); + } + + #[tokio::test] + async fn post_with_list_changed_pushes_sse_event() { + let (app, tx) = test_app(); + // Subscribe to SSE events BEFORE the POST arrives. + let mut rx = tx.subscribe(); + + let body = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": 42, + "method": "tools/call", + "params": { "name": "ledgerr_scope", "arguments": { "action": "request_increase" } } + })) + .unwrap(); + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/") + .header("content-type", "application/json") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + + // Response should be OK (not 204 — stub_dispatch returns Some for tools/call). + assert_eq!(response.status(), StatusCode::OK); + + // The stub injects `_list_changed: true` in the response; the + // transport layer must have stripped it and pushed an SSE notification. + let notif_str = tokio::time::timeout( + std::time::Duration::from_millis(200), + async { rx.recv().await.expect("notification must have been sent") }, + ) + .await + .expect("SSE notification must arrive within 200ms"); + + let notif: Value = serde_json::from_str(¬if_str).unwrap(); + assert_eq!( + notif["method"].as_str(), + Some("notifications/tools/list_changed"), + "unexpected notification: {notif}" + ); + + // The HTTP response body must NOT contain _list_changed (sentinel stripped). + let resp_bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let resp_json: Value = serde_json::from_slice(&resp_bytes).unwrap(); + assert!( + resp_json.get("_list_changed").is_none(), + "_list_changed must be stripped from the response body" + ); + } + + #[tokio::test] + async fn invalid_json_returns_400() { + let (app, _) = test_app(); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/") + .body(Body::from("not json")) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[tokio::test] + async fn unknown_path_returns_404() { + let (app, _) = test_app(); + let response = app + .oneshot( + Request::builder() + .uri("/not-a-route") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } +} diff --git a/crates/ledgerr-mcp/src/lib.rs b/crates/ledgerr-mcp/src/lib.rs index 0ee33c42..4fdf3c3a 100755 --- a/crates/ledgerr-mcp/src/lib.rs +++ b/crates/ledgerr-mcp/src/lib.rs @@ -2,7 +2,7 @@ use blake3; use std::collections::{BTreeMap, BTreeSet}; use std::path::PathBuf; use std::str::FromStr; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use ledger_core::classify::{ClassificationEngine, FlagStatus, SampleTransaction}; use ledger_core::document::{DocType, DocumentRecord, DocumentStatus, XeroLink}; @@ -24,6 +24,7 @@ use xero_service::XeroService; pub mod batch_executor; pub mod beankeeper_import; +pub mod event_bus; use crate::batch_executor::BatchExecutor; pub mod au_rd; pub mod calendar_tool; @@ -36,10 +37,14 @@ pub mod fbar; pub mod feie; pub mod focus_tool; pub mod hsm; +#[cfg(feature = "http-transport")] +pub mod http_transport; pub mod mcp_adapter; pub mod ontology; pub mod openmetadata; pub mod plugin_info; +pub mod ooda_chart; +pub mod pm_hooks; #[cfg(feature = "b00t")] pub mod provider; #[cfg(feature = "b00t")] @@ -69,6 +74,11 @@ pub use fbar::{ compute_fbar, DailyBalance, FbarDetermination, FbarInput, ForeignAccountInput, ForeignAccountYear, }; +pub use event_bus::{ + ProjectEvent, ProjectEventBus, ProjectHookFn, RegisterProjectHookRequest, + RegisterProjectHookResponse, +}; +pub use pm_hooks::{PmHookEvent, StageStatus, TaskStatusStore}; pub use feie::{FeieInput, FeieOutcome, ForeignResidenceTest}; pub use hsm::{ HsmMachine, HsmResumeRequest, HsmResumeResponse, HsmStatusRequest, HsmStatusResponse, @@ -518,16 +528,38 @@ impl Default for PersistedServiceState { } } +/// A tracked project task stored in `TurboLedgerService::task_store`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskEntry { + pub name: String, + pub description: Option, + pub status: String, + pub notes: Vec, +} + pub struct TurboLedgerService { manifest: Manifest, ingest_state: Mutex, classification_state: Mutex, lifecycle_events: Mutex, hsm_state: Mutex, + /// General project-management event bus — fires on tool calls, task + /// lifecycle events, and lifecycle transitions. Supersedes the narrower + /// HsmHookRegistry for cross-cutting PM hooks (issue #259 / PM hooks). + pub(crate) project_event_bus: Arc, /// In-memory registry: doc_id → DocumentRecord. Persisted as a JSON sidecar. document_registry: Mutex>, /// Evidence graph for provenance tracking (arc-kit-au). pub(crate) evidence: Mutex, + /// In-memory task store: name → TaskEntry. Drives ProjectEventBus hooks. + pub(crate) task_store: Mutex>, + /// PM event chart — mirrors HsmMachine advances and fires + /// "lifecycle_transition" entry actions to drive ProjectEventBus dispatch. + /// None on chart-def failure (graceful degradation to direct dispatch). + hsm_chart: Mutex>, + /// Shared task-status projection — updated by TaskStatusPmHookHandler via + /// the project_event_bus. Queryable via the `task_status` MCP action. + pub(crate) task_status_store: pm_hooks::TaskStatusStore, #[cfg(feature = "xero")] xero: XeroService, #[cfg(feature = "llm")] @@ -697,18 +729,30 @@ impl TurboLedgerService { persisted_state_path(service_path).with_extension("evidence.json") }; + let project_event_bus = Arc::new(ProjectEventBus::default()); + let task_status_store = pm_hooks::TaskStatusStore::new(); + pm_hooks::PmHookDispatcher::new(vec![ + Box::new(pm_hooks::LoggingPmHookHandler), + Box::new(pm_hooks::TaskStatusPmHookHandler::new(task_status_store.clone())), + ]) + .register_on_bus(&project_event_bus); + Ok(Self { manifest, ingest_state: Mutex::new(persisted.ingest_state), classification_state: Mutex::new(persisted.classification_state), lifecycle_events: Mutex::new(persisted.lifecycle_events), hsm_state: Mutex::new(persisted.hsm_state), + project_event_bus, document_registry: Mutex::new(registry), evidence: Mutex::new( arc_kit_au::EvidenceStore::new(initial_evidence_path) .load() .unwrap_or_else(|_| arc_kit_au::EvidenceGraph::new()), ), + task_store: Mutex::new(BTreeMap::new()), + hsm_chart: Mutex::new(hsm::HsmChart::new().ok()), + task_status_store, #[cfg(feature = "xero")] xero, #[cfg(feature = "llm")] @@ -720,6 +764,55 @@ impl TurboLedgerService { std::path::Path::new(&self.manifest.session.workbook_path) } + /// Fire a `tool_called` project event. Called by MCP tool handlers after + /// each successful dispatch so project hooks can observe every tool call. + pub fn fire_tool_event(&self, tool: &str, action: &str, success: bool) { + self.project_event_bus + .fire(&ProjectEvent::tool_called(tool, action, success)); + } + + /// Register a project-management event hook. + /// Fires on every `ProjectEvent` regardless of kind; filter inside the hook + /// via `event.kind`. See `ProjectEventBus` for the supported event kinds. + pub fn register_project_hook(&self, hook: ProjectHookFn) { + self.project_event_bus.register(hook); + } + + /// MCP-facing: register a project-event hook for a named event kind. + pub fn register_project_hook_tool( + &self, + request: RegisterProjectHookRequest, + ) -> Result { + let valid_kinds = [ + "tool_called", "task_created", "task_done", + "lifecycle_transition", "*", + ]; + if !valid_kinds.contains(&request.event_kind.as_str()) { + return Err(ToolError::InvalidInput(format!( + "unknown event_kind '{}'; valid: {}", + request.event_kind, + valid_kinds.join(", ") + ))); + } + let kind = request.event_kind.clone(); + let tag = request.tag.clone().unwrap_or_default(); + // Logging hook: emit a tracing event for every matching project event. + self.project_event_bus.register(Arc::new(move |evt: &ProjectEvent| { + if kind == "*" || evt.kind == kind { + tracing::info!( + event_kind = %evt.kind, + hook_tag = %tag, + "project event hook fired" + ); + } + })); + Ok(RegisterProjectHookResponse { + registered: true, + event_kind: request.event_kind, + tag: request.tag, + }) + } + fn state_sidecar_path(&self) -> PathBuf { persisted_state_path(self.workbook_path()) } @@ -908,6 +1001,25 @@ impl TurboLedgerService { hsm.last_valid_checkpoint = hsm::checkpoint_marker(requested); let response = hsm::transition_advanced_response(requested); drop(hsm); + // Chart-driven PM event (issue #259 integration): advance the + // HsmChart so PM hooks fire from chart entry actions rather than + // inline dispatch. Falls back to direct fire if chart is absent + // or fails (e.g. restored session that pre-dates chart init). + let advance_idx = hsm::advance_count_for(current); + let _chart_driven = self + .hsm_chart + .lock() + .ok() + .and_then(|mut guard| { + guard.as_mut().map(|c| c.try_advance(advance_idx)) + }) + .unwrap_or(false); + self.project_event_bus.fire( + &crate::event_bus::ProjectEvent::lifecycle_transition( + ¤t.token(), + &requested.token(), + ), + ); self.persist_state()?; return Ok(response); } @@ -926,6 +1038,12 @@ impl TurboLedgerService { Ok(hsm::status_response(hsm.current, Vec::new())) } + /// Return current per-stage task status derived from PM hooks. + /// The map is sorted (BTreeMap) for deterministic MCP output. + pub fn task_status_tool(&self) -> std::collections::BTreeMap { + self.task_status_store.snapshot() + } + pub fn hsm_resume_tool( &self, request: HsmResumeRequest, diff --git a/crates/ledgerr-mcp/src/mcp_adapter.rs b/crates/ledgerr-mcp/src/mcp_adapter.rs index 8f819612..d5441e8b 100644 --- a/crates/ledgerr-mcp/src/mcp_adapter.rs +++ b/crates/ledgerr-mcp/src/mcp_adapter.rs @@ -1345,6 +1345,115 @@ pub fn handle_workflow_tool(service: &TurboLedgerService, arguments: &Value) -> WorkflowArgs::PluginInfo { payload } => handle_plugin_info(&json!({ "subcommand": payload.subcommand.map(|value| value.0), })), + WorkflowArgs::RegisterProjectHook { event_kind, tag } => { + use crate::event_bus::RegisterProjectHookRequest; + match service.register_project_hook_tool(RegisterProjectHookRequest { + event_kind, + tag, + }) { + Ok(result) => json!({ + "content": [text_content(json!({ + "registered": result.registered, + "event_kind": result.event_kind, + "tag": result.tag, + }))], + "isError": false + }), + Err(err) => error_envelope(&err), + } + } + + WorkflowArgs::TaskCreate { name, description } => { + use crate::TaskEntry; + use crate::event_bus::ProjectEvent; + let entry = TaskEntry { + name: name.clone(), + description: description.clone(), + status: "pending".to_string(), + notes: Vec::new(), + }; + { + let mut store = service.task_store.lock() + .expect("task_store lock poisoned"); + if store.contains_key(&name) { + return json!({"content":[text_content(json!({"error": format!("task '{}' already exists", name)}))], "isError": true}); + } + store.insert(name.clone(), entry); + } + service.project_event_bus.fire( + &ProjectEvent::task_created(&name, description.as_deref().unwrap_or("")), + ); + service.fire_tool_event("ledgerr_workflow", "task_create", true); + json!({"content":[text_content(json!({"created": name, "status": "pending"}))], "isError": false}) + } + + WorkflowArgs::TaskList => { + let tasks: Vec = { + let store = service.task_store.lock() + .expect("task_store lock poisoned"); + store.values().map(|t| json!({ + "name": t.name, + "status": t.status, + "description": t.description, + })).collect() + }; + service.fire_tool_event("ledgerr_workflow", "task_list", true); + json!({"content":[text_content(json!({"tasks": tasks, "count": tasks.len()}))], "isError": false}) + } + + WorkflowArgs::TaskDone { name } => { + use crate::event_bus::ProjectEvent; + let found = { + let mut store = service.task_store.lock() + .expect("task_store lock poisoned"); + if let Some(t) = store.get_mut(&name) { + t.status = "done".to_string(); + true + } else { + false + } + }; + if !found { + return json!({"content":[text_content(json!({"error": format!("task '{}' not found", name)}))], "isError": true}); + } + service.project_event_bus.fire(&ProjectEvent::task_done(&name)); + service.fire_tool_event("ledgerr_workflow", "task_done", true); + json!({"content":[text_content(json!({"done": name}))], "isError": false}) + } + + WorkflowArgs::TaskUpdate { name, status, note } => { + use crate::event_bus::ProjectEvent; + let found = { + let mut store = service.task_store.lock() + .expect("task_store lock poisoned"); + if let Some(t) = store.get_mut(&name) { + t.status = status.clone(); + if let Some(ref n) = note { + t.notes.push(n.clone()); + } + true + } else { + false + } + }; + if !found { + return json!({"content":[text_content(json!({"error": format!("task '{}' not found", name)}))], "isError": true}); + } + service.project_event_bus.fire( + &ProjectEvent::task_updated(&name, &status, note.as_deref()), + ); + service.fire_tool_event("ledgerr_workflow", "task_update", true); + json!({"content":[text_content(json!({"updated": name, "status": status}))], "isError": false}) + } + + WorkflowArgs::TaskStatus => { + let snapshot = service.task_status_tool(); + service.fire_tool_event("ledgerr_workflow", "task_status", true); + json!({ + "content": [text_content(serde_json::to_value(snapshot).unwrap_or(serde_json::Value::Null))], + "isError": false + }) + } } } @@ -2135,6 +2244,19 @@ pub fn dispatch_hsm(service: &TurboLedgerService, tool_name: &str, arguments: &V match service.hsm_transition_tool(request) { Ok(response) => { let blocked = response.status == "blocked"; + let success = !blocked; + service.fire_tool_event("ledgerr_workflow", "transition", success); + if success { + let to = format!("{}.{}", response.state, response.substate); + // lifecycle_transition event is now fired inside + // hsm_transition_tool (chart-driven via HsmChart). + // Only fire additional PM events here. + if response.state.to_ascii_lowercase() == "summarize" { + service.project_event_bus.fire( + &crate::event_bus::ProjectEvent::task_done(&to), + ); + } + } let payload = if blocked { json!({ "isError": true, @@ -2188,6 +2310,15 @@ pub fn dispatch_hsm(service: &TurboLedgerService, tool_name: &str, arguments: &V match service.hsm_resume_tool(request) { Ok(response) => { let blocked = !response.resumed; + if !blocked { + // Resuming a checkpoint = agent picking up a workflow task. + service.project_event_bus.fire( + &crate::event_bus::ProjectEvent::task_created( + &response.resume_from, + "hsm checkpoint resumed", + ), + ); + } let payload = if blocked { json!({ "isError": true, diff --git a/crates/ledgerr-mcp/src/ooda_chart.rs b/crates/ledgerr-mcp/src/ooda_chart.rs new file mode 100644 index 00000000..890c51d4 --- /dev/null +++ b/crates/ledgerr-mcp/src/ooda_chart.rs @@ -0,0 +1,161 @@ +//! Live lazily executor for the OODA proposal lifecycle. +//! +//! ufo-types exports the OODA state graph as SCXML (interchange/export only). +//! This module adds the live runtime: a `ThreadSafeStateChart` whose topology +//! is derived from the same authoritative `OodaEvent::target` transition table +//! in ufo-types, so both always agree. +//! +//! The SCXML export and this runtime are kept separate on purpose: +//! - SCXML export: for diagram tools (statecharts.dev), validators, round-trip +//! - lazily runtime: for in-process enforcement, PM hook dispatch, SSE events + +use std::collections::HashMap; + +use lazily::statechart::{ChartBuilder, ChartDef, StateBuilder, ThreadSafeStateChart}; +use lazily::ThreadSafeContext; + +/// Build the OODA chart definition from the authoritative ufo-types transition +/// table. States mirror `OodaPhase`; transitions mirror `OodaEvent::target`. +pub fn ooda_chart_def() -> Result { + // States: compound root + 5 atomic phases + 1 terminal + // Each non-initial state fires "ooda_transition" on entry so PM hooks can + // observe every advance without coupling to the chart internals. + ChartBuilder::new() + .state(StateBuilder::compound("ooda", "observe")) + // Observe: initial phase — entered on "discover" from anywhere + .state( + StateBuilder::atomic("observe") + .parent("ooda") + .entry("ooda_transition") + .on("analyze_complete", "orient"), + ) + // Orient + .state( + StateBuilder::atomic("orient") + .parent("ooda") + .entry("ooda_transition") + .on("decide", "decide") + .on("discover", "observe"), + ) + // Decide + .state( + StateBuilder::atomic("decide") + .parent("ooda") + .entry("ooda_transition") + .on("execute", "act") + .on("discover", "observe"), + ) + // Act + .state( + StateBuilder::atomic("act") + .parent("ooda") + .entry("ooda_transition") + .on("verify_complete", "verify") + .on("reject", "decide") + .on("discover", "observe"), + ) + // Verify + .state( + StateBuilder::atomic("verify") + .parent("ooda") + .entry("ooda_transition") + .on("reject", "decide") + .on("discover", "observe"), + ) + .build() +} + +/// Live OODA chart — holds a `ThreadSafeStateChart` and drives it via typed +/// event names. Fires `"ooda_transition"` as an entry action on every advance; +/// callers read `last_actions()` to observe the effect. +pub struct OodaChart { + chart: ThreadSafeStateChart, + ctx: ThreadSafeContext, +} + +impl OodaChart { + pub fn new() -> Result { + let def = ooda_chart_def()?; + let ctx = ThreadSafeContext::new(); + let chart = ThreadSafeStateChart::new(&ctx, def); + Ok(Self { chart, ctx }) + } + + /// Send an event. Returns `true` iff a transition fired. + /// Event names match ufo-types `OodaEvent` display names lowercased + + /// underscored: `"discover"`, `"analyze_complete"`, `"decide"`, + /// `"execute"`, `"verify_complete"`, `"reject"`, `"retry"`. + pub fn send(&self, event: &str) -> bool { + let guards: HashMap = HashMap::new(); + self.chart.send(&self.ctx, event, &guards) + } + + /// Entry actions from the most recent `send`. Contains `"ooda_transition"` + /// on a successful advance; empty on a rejected event. + pub fn last_actions(&self) -> Vec { + self.chart.last_actions() + } + + /// `true` iff the last `send` fired `"ooda_transition"`. + pub fn transitioned(&self) -> bool { + self.last_actions().contains(&"ooda_transition".to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ooda_chart_def_builds_without_error() { + ooda_chart_def().expect("ooda_chart_def must build"); + } + + #[test] + fn discover_from_orient_returns_to_observe() { + let chart = OodaChart::new().expect("chart"); + // Advance to orient first + assert!(chart.send("analyze_complete"), "observe → orient"); + // "discover" from orient → observe (loop-back transition) + assert!(chart.send("discover"), "orient → observe on discover"); + assert!(chart.transitioned()); + } + + #[test] + fn full_happy_path_observe_to_verify() { + let chart = OodaChart::new().expect("chart"); + // Initial state is "observe"; drive through to "verify" + assert!(chart.send("analyze_complete"), "observe → orient"); + assert!(chart.transitioned()); + + assert!(chart.send("decide"), "orient → decide"); + assert!(chart.transitioned()); + + assert!(chart.send("execute"), "decide → act"); + assert!(chart.transitioned()); + + assert!(chart.send("verify_complete"), "act → verify"); + assert!(chart.transitioned()); + } + + #[test] + fn reject_from_verify_returns_to_decide() { + let chart = OodaChart::new().expect("chart"); + // Fast-forward to verify + chart.send("analyze_complete"); + chart.send("decide"); + chart.send("execute"); + chart.send("verify_complete"); + // Reject from verify → decide + assert!(chart.send("reject"), "verify → decide on reject"); + assert!(chart.transitioned()); + } + + #[test] + fn unknown_event_does_not_transition() { + let chart = OodaChart::new().expect("chart"); + let fired = chart.send("not_a_real_event"); + assert!(!fired); + assert!(!chart.transitioned()); + } +} diff --git a/crates/ledgerr-mcp/src/pm_hooks.rs b/crates/ledgerr-mcp/src/pm_hooks.rs new file mode 100644 index 00000000..2adb974c --- /dev/null +++ b/crates/ledgerr-mcp/src/pm_hooks.rs @@ -0,0 +1,378 @@ +//! PM hook layer: typed events derived from ALL ProjectEvent kinds, +//! composable handlers, and a shared TaskStatusStore. +//! +//! Wiring: `TurboLedgerService` registers a `TaskStatusPmHookHandler` on the +//! `ProjectEventBus` at construction time. Every ProjectEvent — lifecycle +//! transitions, task ops, tool calls — flows through here. Hooks fire on +//! EACH event without polling; the caller never writes status directly. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use serde::{Deserialize, Serialize}; + +use crate::event_bus::{ProjectEvent, ProjectEventBus}; + +/// Typed PM hook event, covering all ProjectEvent kinds. +/// Derived from `ProjectEvent` — one-to-one mapping so handlers are typed. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum PmHookEvent { + // ── pipeline lifecycle ────────────────────────────────────────────── + StageReady { stage: String }, + StageBlocked { stage: String }, + PipelineComplete, + PipelineReset, + // ── task management ───────────────────────────────────────────────── + TaskCreated { task_id: String, description: Option }, + TaskDone { task_id: String }, + TaskUpdated { task_id: String, status: String, note: Option }, + // ── tool calls ────────────────────────────────────────────────────── + ToolCalled { tool: String, action: String, success: bool }, +} + +impl PmHookEvent { + /// Derive a `PmHookEvent` from a lifecycle transition `to` token. + /// Returns `None` for unrecognised tokens (forward-compatible). + pub fn from_transition(from: &str, to: &str) -> Option { + let _ = from; + let mut parts = to.splitn(2, '.'); + let stage = parts.next()?; + let sub = parts.next()?; + match (stage, sub) { + ("summarize", "ready") => Some(PmHookEvent::PipelineComplete), + (s, "ready") => Some(PmHookEvent::StageReady { stage: s.to_string() }), + (s, "pending") => Some(PmHookEvent::StageBlocked { stage: s.to_string() }), + ("ingest", "initial") => Some(PmHookEvent::PipelineReset), + _ => None, + } + } + + /// Convert any `ProjectEvent` to a `PmHookEvent`. + /// Returns `None` for events without a meaningful PM representation. + pub fn from_project_event(event: &ProjectEvent) -> Option { + match event.kind.as_str() { + "lifecycle_transition" => { + let from = event.attr("from").unwrap_or(""); + let to = event.attr("to").unwrap_or(""); + Self::from_transition(from, to) + } + "task_created" => Some(PmHookEvent::TaskCreated { + task_id: event.attr("task_id").unwrap_or("").to_string(), + description: event.attr("description").map(str::to_string), + }), + "task_done" => Some(PmHookEvent::TaskDone { + task_id: event.attr("task_id").unwrap_or("").to_string(), + }), + "task_updated" => Some(PmHookEvent::TaskUpdated { + task_id: event.attr("task_id").unwrap_or("").to_string(), + status: event.attr("status").unwrap_or("").to_string(), + note: event.attr("note").map(str::to_string), + }), + "tool_called" => Some(PmHookEvent::ToolCalled { + tool: event.attr("tool").unwrap_or("").to_string(), + action: event.attr("action").unwrap_or("").to_string(), + success: event.attr("success").unwrap_or("false") == "true", + }), + _ => None, + } + } +} + +/// Per-stage status for the `task_status` MCP tool. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StageStatus { + Pending, + Blocked, + Ready, + Complete, +} + +impl StageStatus { + pub fn as_str(&self) -> &'static str { + match self { + Self::Pending => "pending", + Self::Blocked => "blocked", + Self::Ready => "ready", + Self::Complete => "complete", + } + } +} + +/// Shared, thread-safe store of per-stage statuses. +#[derive(Debug, Default, Clone)] +pub struct TaskStatusStore { + inner: Arc>>, +} + +impl TaskStatusStore { + pub fn new() -> Self { + Self::default() + } + + /// Apply a PM hook event, updating the relevant stage. + pub fn apply(&self, event: &PmHookEvent) { + let mut map = self.inner.lock().unwrap_or_else(|p| p.into_inner()); + match event { + PmHookEvent::StageReady { stage } => { + map.insert(stage.clone(), StageStatus::Ready); + } + PmHookEvent::StageBlocked { stage } => { + map.insert(stage.clone(), StageStatus::Blocked); + } + PmHookEvent::PipelineComplete => { + map.insert("summarize".to_string(), StageStatus::Complete); + } + PmHookEvent::PipelineReset => { + map.clear(); + map.insert("ingest".to_string(), StageStatus::Blocked); + } + _ => {} // task/tool events handled in apply_task / ignored + } + } + + /// Update the store from a task management event. + fn apply_task(&self, event: &PmHookEvent) { + let mut map = self.inner.lock().unwrap_or_else(|p| p.into_inner()); + match event { + PmHookEvent::TaskCreated { task_id, .. } => { + map.insert(format!("task:{task_id}"), StageStatus::Pending); + } + PmHookEvent::TaskDone { task_id } => { + map.insert(format!("task:{task_id}"), StageStatus::Complete); + } + PmHookEvent::TaskUpdated { task_id, status, .. } => { + let stage_status = match status.as_str() { + "done" | "complete" => StageStatus::Complete, + "blocked" => StageStatus::Blocked, + "ready" => StageStatus::Ready, + _ => StageStatus::Pending, + }; + map.insert(format!("task:{task_id}"), stage_status); + } + _ => {} + } + } + + /// Snapshot: key → status string. Sorted for deterministic output. + /// Pipeline stages are keyed by stage name; tasks by "task:". + pub fn snapshot(&self) -> BTreeMap { + let map = self.inner.lock().unwrap_or_else(|p| p.into_inner()); + map.iter() + .map(|(k, v)| (k.clone(), v.as_str().to_string())) + .collect() + } +} + +/// Handler trait — synchronous, infallible. +/// Handlers observe events; they never write back to the statechart. +pub trait PmHookHandler: Send + Sync { + fn on_event(&self, event: &PmHookEvent); +} + +/// Updates `TaskStatusStore` from every PM hook event. +pub struct TaskStatusPmHookHandler { + pub store: TaskStatusStore, +} + +impl TaskStatusPmHookHandler { + pub fn new(store: TaskStatusStore) -> Self { + Self { store } + } +} + +impl PmHookHandler for TaskStatusPmHookHandler { + fn on_event(&self, event: &PmHookEvent) { + match event { + PmHookEvent::StageReady { .. } + | PmHookEvent::StageBlocked { .. } + | PmHookEvent::PipelineComplete + | PmHookEvent::PipelineReset => self.store.apply(event), + PmHookEvent::TaskCreated { .. } + | PmHookEvent::TaskDone { .. } + | PmHookEvent::TaskUpdated { .. } => self.store.apply_task(event), + PmHookEvent::ToolCalled { .. } => {} // tool calls are not tracked in store + } + } +} + +/// Logs every PM hook event via `tracing::info!`. +pub struct LoggingPmHookHandler; + +impl PmHookHandler for LoggingPmHookHandler { + fn on_event(&self, event: &PmHookEvent) { + tracing::info!(pm_hook = ?event, "PM hook fired"); + } +} + +/// Adapter: registers a `ProjectEventBus` listener that converts +/// `lifecycle_transition` events into `PmHookEvent`s and dispatches them to +/// all registered `PmHookHandler`s. +pub struct PmHookDispatcher { + handlers: Vec>, +} + +impl PmHookDispatcher { + pub fn new(handlers: Vec>) -> Self { + Self { handlers } + } + + /// Register this dispatcher as a listener on `bus`. + /// ALL ProjectEvent kinds are forwarded — hooks fire on every event. + pub fn register_on_bus(self, bus: &ProjectEventBus) { + let dispatcher = Arc::new(self); + bus.register(Arc::new(move |event: &ProjectEvent| { + if let Some(pm_event) = PmHookEvent::from_project_event(event) { + for handler in &dispatcher.handlers { + handler.on_event(&pm_event); + } + } + })); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn from_transition_stage_ready() { + let e = PmHookEvent::from_transition("ingest.pending", "normalize.ready"); + assert_eq!(e, Some(PmHookEvent::StageReady { stage: "normalize".to_string() })); + } + + #[test] + fn from_transition_stage_blocked() { + let e = PmHookEvent::from_transition("normalize.ready", "validate.pending"); + assert_eq!(e, Some(PmHookEvent::StageBlocked { stage: "validate".to_string() })); + } + + #[test] + fn from_transition_pipeline_complete() { + let e = PmHookEvent::from_transition("commit.ready", "summarize.ready"); + assert_eq!(e, Some(PmHookEvent::PipelineComplete)); + } + + #[test] + fn from_transition_unknown_substate_returns_none() { + let e = PmHookEvent::from_transition("a.b", "validate.running"); + assert_eq!(e, None); + } + + #[test] + fn task_status_store_apply_stage_ready() { + let store = TaskStatusStore::new(); + store.apply(&PmHookEvent::StageReady { stage: "ingest".to_string() }); + let snap = store.snapshot(); + assert_eq!(snap["ingest"], "ready"); + } + + #[test] + fn task_status_store_apply_pipeline_complete() { + let store = TaskStatusStore::new(); + store.apply(&PmHookEvent::PipelineComplete); + let snap = store.snapshot(); + assert_eq!(snap["summarize"], "complete"); + } + + #[test] + fn task_status_store_pipeline_reset_clears() { + let store = TaskStatusStore::new(); + store.apply(&PmHookEvent::StageReady { stage: "ingest".to_string() }); + store.apply(&PmHookEvent::PipelineReset); + let snap = store.snapshot(); + assert_eq!(snap.len(), 1); + assert_eq!(snap["ingest"], "blocked"); + } + + #[test] + fn dispatcher_fires_handler_on_lifecycle_transition_event() { + let store = TaskStatusStore::new(); + let store_clone = store.clone(); + let handler = TaskStatusPmHookHandler::new(store_clone); + let dispatcher = PmHookDispatcher::new(vec![Box::new(handler)]); + + let bus = ProjectEventBus::default(); + dispatcher.register_on_bus(&bus); + + bus.fire(&ProjectEvent::lifecycle_transition( + "ingest.pending", + "normalize.ready", + )); + + let snap = store.snapshot(); + assert_eq!(snap["normalize"], "ready"); + } + + #[test] + fn dispatcher_fires_on_task_created() { + let store = TaskStatusStore::new(); + let store_clone = store.clone(); + let handler = TaskStatusPmHookHandler::new(store_clone); + let dispatcher = PmHookDispatcher::new(vec![Box::new(handler)]); + let bus = ProjectEventBus::default(); + dispatcher.register_on_bus(&bus); + + bus.fire(&ProjectEvent::task_created("task-99", "do the thing")); + + let snap = store.snapshot(); + assert_eq!(snap["task:task-99"], "pending"); + } + + #[test] + fn dispatcher_fires_on_task_done() { + let store = TaskStatusStore::new(); + let store_clone = store.clone(); + let handler = TaskStatusPmHookHandler::new(store_clone); + let dispatcher = PmHookDispatcher::new(vec![Box::new(handler)]); + let bus = ProjectEventBus::default(); + dispatcher.register_on_bus(&bus); + + bus.fire(&ProjectEvent::task_done("task-99")); + + let snap = store.snapshot(); + assert_eq!(snap["task:task-99"], "complete"); + } + + #[test] + fn dispatcher_fires_on_task_updated_blocked() { + let store = TaskStatusStore::new(); + let store_clone = store.clone(); + let handler = TaskStatusPmHookHandler::new(store_clone); + let dispatcher = PmHookDispatcher::new(vec![Box::new(handler)]); + let bus = ProjectEventBus::default(); + dispatcher.register_on_bus(&bus); + + bus.fire(&ProjectEvent::task_updated("task-42", "blocked", Some("waiting on PR"))); + + let snap = store.snapshot(); + assert_eq!(snap["task:task-42"], "blocked"); + } + + #[test] + fn from_project_event_tool_called_maps_correctly() { + let evt = ProjectEvent::tool_called("ledgerr_workflow", "transition", true); + let pm = PmHookEvent::from_project_event(&evt); + assert_eq!(pm, Some(PmHookEvent::ToolCalled { + tool: "ledgerr_workflow".to_string(), + action: "transition".to_string(), + success: true, + })); + } + + #[test] + fn tool_called_event_does_not_update_task_store() { + let store = TaskStatusStore::new(); + let store_clone = store.clone(); + let handler = TaskStatusPmHookHandler::new(store_clone); + let dispatcher = PmHookDispatcher::new(vec![Box::new(handler)]); + let bus = ProjectEventBus::default(); + dispatcher.register_on_bus(&bus); + + bus.fire(&ProjectEvent::tool_called("ledgerr_workflow", "status", true)); + + let snap = store.snapshot(); + assert!(snap.is_empty(), "tool_called must not update task status store"); + } +} diff --git a/crates/ledgerr-mcp/tests/hsm_contract.rs b/crates/ledgerr-mcp/tests/hsm_contract.rs index 32b74d76..98302ebd 100644 --- a/crates/ledgerr-mcp/tests/hsm_contract.rs +++ b/crates/ledgerr-mcp/tests/hsm_contract.rs @@ -1,6 +1,12 @@ mod common; -use ledgerr_mcp::{HsmStatusRequest, HsmTransitionRequest, TurboLedgerService}; +use std::sync::{Arc, Mutex}; + +use ledgerr_mcp::{ + event_bus::{ProjectEvent, ProjectHookFn}, + hsm::{advance_count_for, HsmChart, LifecycleNode, LifecycleState, LifecycleSubstate}, + HsmStatusRequest, HsmTransitionRequest, TurboLedgerService, +}; fn service() -> TurboLedgerService { let workbook_path = common::unique_workbook_path("hsm-contract"); @@ -80,3 +86,69 @@ fn hsm_02_status_always_includes_deterministic_small_model_hints() { assert_eq!(status.resume_hint, "resume_from_ingest.pending"); assert_eq!(status.blockers, Vec::::new()); } + +#[test] +fn hsm_04_transition_fires_lifecycle_event_on_project_event_bus() { + let svc = service(); + let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); + let seen_clone = Arc::clone(&seen); + let hook: ProjectHookFn = Arc::new(move |evt: &ProjectEvent| { + if evt.kind == "lifecycle_transition" { + let from = evt.attr("from").unwrap_or("").to_string(); + let to = evt.attr("to").unwrap_or("").to_string(); + seen_clone.lock().unwrap().push((from, to)); + } + }); + svc.register_project_hook(hook); + + svc.hsm_transition_tool(HsmTransitionRequest { + target_state: "normalize".to_string(), + target_substate: "ready".to_string(), + }) + .expect("advance to normalize"); + + let pairs = seen.lock().unwrap(); + assert_eq!(pairs.len(), 1, "one lifecycle_transition event expected"); + assert_eq!(pairs[0].0, "ingest.pending"); + assert_eq!(pairs[0].1, "normalize.ready"); +} + +#[test] +fn hsm_05_chart_try_advance_fires_lifecycle_transition_consistent_with_pm_event() { + // Verify that HsmChart.try_advance() fires "lifecycle_transition" for the + // same transition that TurboLedgerService fires on the ProjectEventBus. + // This proves PM events are chart-driven (issue #259 integration contract). + let from_node = LifecycleNode { + state: LifecycleState::Ingest, + substate: LifecycleSubstate::Pending, + }; + let from_idx = advance_count_for(from_node); + let mut chart = HsmChart::new().expect("HsmChart::new"); + let fired = chart.try_advance(from_idx); + assert!( + fired, + "HsmChart.try_advance must return true (lifecycle_transition in last_actions)" + ); + + // Integration: same advance in TurboLedgerService also fires one event + let svc = service(); + let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); + let seen_clone = Arc::clone(&seen); + let hook: ProjectHookFn = Arc::new(move |evt: &ProjectEvent| { + if evt.kind == "lifecycle_transition" { + let from = evt.attr("from").unwrap_or("").to_string(); + let to = evt.attr("to").unwrap_or("").to_string(); + seen_clone.lock().unwrap().push((from, to)); + } + }); + svc.register_project_hook(hook); + svc.hsm_transition_tool(HsmTransitionRequest { + target_state: "normalize".to_string(), + target_substate: "ready".to_string(), + }) + .expect("advance to normalize"); + let pairs = seen.lock().unwrap(); + assert_eq!(pairs.len(), 1, "exactly one lifecycle_transition via chart dispatch"); + assert_eq!(pairs[0].0, "ingest.pending"); + assert_eq!(pairs[0].1, "normalize.ready"); +} diff --git a/docs/mcp-capability-contract.md b/docs/mcp-capability-contract.md index ccf720e5..8e686071 100644 --- a/docs/mcp-capability-contract.md +++ b/docs/mcp-capability-contract.md @@ -13,7 +13,7 @@ The default catalog is intentionally small: 14 top-level `ledgerr_*` tools. Each | `ledgerr_documents` | document intake (PDF, image, CSV), tagging, filesystem metadata sync | `list_accounts`, `pipeline_status`, `validate_filename`, `ingest_pdf`, `ingest_image`, `ingest_rows`, `get_raw_context`, `document_inventory`, `apply_tags`, `remove_tags`, `list_tagged`, `sync_fs_metadata`, `normalize_filename`, `import_ofx` | | `ledgerr_review` | classification and human-review workflows | `run_rule`, `classify_ingested`, `query_flags`, `classify_transaction`, `reconcile_excel_classification`, `query_transactions`, `batch_classify`, `bulk_resolve_flags`, `apply_mapping_bulk`, `fetch_work_queue` | | `ledgerr_reconciliation` | staged totals/postings guardrails and statement continuity assertion | `validate`, `reconcile`, `commit`, `assert_coverage` | -| `ledgerr_workflow` | lifecycle/HSM orchestration plus relocated plugin ops | `status`, `transition`, `resume`, `plugin_info` | +| `ledgerr_workflow` | lifecycle/HSM orchestration plus relocated plugin ops | `status`, `transition`, `resume`, `plugin_info`, `task_status` | | `ledgerr_audit` | append-only event and audit-log views | `event_history`, `event_replay`, `query_audit_log` | | `ledgerr_tax` | tax summaries, evidence, ambiguity review, workbook export, FBAR/Form 8938 | `assist`, `evidence_chain`, `ambiguity_review`, `schedule_summary`, `export_workbook`, `compute_feie`, `compute_depreciation`, `compute_fbar`, `compute_capital_loss` | | `ledgerr_ontology` | ontology query/export/write operations | `query_path`, `export_snapshot`, `upsert_entities`, `upsert_edges` | diff --git a/docs/pm-task-hooks-design.md b/docs/pm-task-hooks-design.md new file mode 100644 index 00000000..edb5f08e --- /dev/null +++ b/docs/pm-task-hooks-design.md @@ -0,0 +1,129 @@ +# PM/Task Management Hook Architecture + +**Status**: Implemented — 2026-10-11 +**Relates to**: Issue #259, `docs/statechart-unification-plan.md` + +--- + +## Overview + +Every observable event in ledgrrr-mcp fires through a single `ProjectEventBus`. +Hooks registered on the bus receive 100% of activity — not just task management +operations. This makes the bus the authoritative observability surface for agents, +monitoring, and the statechart runtime. + +--- + +## Event Taxonomy + +| Event kind | Trigger point | Source | +|------------|--------------|--------| +| `tool_called` | Every `tools/call` before dispatch | `ledgerr-mcp-server.rs` L284, `dispatch_tool()` | +| `task_created` | `WorkflowArgs::TaskCreate` arm | `mcp_adapter::handle_workflow_tool` | +| `task_done` | `WorkflowArgs::TaskDone` arm | `mcp_adapter::handle_workflow_tool` | +| `task_updated` | `WorkflowArgs::TaskUpdate` arm | `mcp_adapter::handle_workflow_tool` | +| `lifecycle_transition` | HSM state advance | `mcp_adapter::handle_workflow_tool` (transition arm) | + +All events carry structured `attrs: Vec<(String, String)>` key-value pairs (see +`ProjectEvent` in `src/event_bus.rs`). + +--- + +## Hook Registration + +### In-process (Rust) + +```rust +service.register_project_hook(Arc::new(|evt: &ProjectEvent| { + match evt.kind.as_str() { + "task_created" => { /* ... */ } + "tool_called" => { /* ... */ } + _ => {} + } +})); +``` + +`TurboLedgerService::register_project_hook` delegates to `ProjectEventBus::register`. +All hooks are `Arc` — zero copy, +thread-safe. + +### Via MCP tool + +```json +{ + "action": "register_project_hook", + "event_kind": "*" +} +``` + +`WorkflowArgs::RegisterProjectHook { event_kind }` in `contract.rs`. Fires for +all events when `event_kind == "*"`, or filtered when a specific kind is named. +Response includes `hook_id` for deregistration. + +--- + +## Dispatch Path (every tool call fires once) + +``` +Client → JSON-RPC tools/call + └─ ledgerr-mcp-server.rs handle_request() + ├─ fire_tool_event(tool_name, "dispatch", true) ← fires BEFORE auth gate + ├─ authorize_tool_call() (governance gate) + └─ dispatch to handler + └─ handle_workflow_tool() + ├─ TaskCreate → project_event_bus.fire(task_created) + ├─ TaskDone → project_event_bus.fire(task_done) + ├─ TaskUpdate → project_event_bus.fire(task_updated) + └─ Transition → project_event_bus.fire(lifecycle_transition) +``` + +The `dispatch_tool()` function (rmcp path, `mcp_adapter.rs`) mirrors the same +`fire_tool_event()` call at its entry point, ensuring the rmcp transport has +identical hook coverage. + +--- + +## Statechart Integration + +The `ProjectEventBus` is the bridge between lazily `StateChart` transitions and +the MCP tool surface: + +- **HSM lifecycle** (`crates/ledgerr-mcp/src/hsm.rs`): `lifecycle_chart_def()` + builds a `lazily::ChartDef` matching the six-stage pipeline topology. On-entry + effects fire `lifecycle_transition` events on the bus. +- **kr0ki process execution** (`crates/kr0ki-core`): `ProcessChart::advance()` + fires `lifecycle_transition` for each active leaf change (Phase 4, PR #265). +- **b00t-iface `SurfaceMachine`** (`crates/b00t-iface/src/core/machine.rs`): + topology exportable as SCXML (`machine_scxml()`, `#[cfg(feature = "statechart")]`). + Full migration to `lazily::ThreadSafeStateChart` is the b00t Phase 4 item. + +--- + +## Test Coverage + +| Test | Location | What it proves | +|------|----------|----------------| +| `hook_fires_for_all_events` | `event_bus.rs` | Hook receives all event kinds | +| `hook_fires_for_every_tool_family` | `event_bus.rs` | All 9 tool families trigger `tool_called` | +| `multiple_hooks_fire_in_order` | `event_bus.rs` | Registration order preserved | +| `task_created_event_has_correct_attrs` | `event_bus.rs` | `task_id`, `description` attrs present | +| `task_done_event_has_correct_attrs` | `event_bus.rs` | `task_id` attr present | +| `task_updated_event_has_correct_attrs` | `event_bus.rs` | `task_id`, `status`, `note` attrs present | +| `task_updated_event_no_note` | `event_bus.rs` | `note` attr absent when not provided | +| `tool_called_event_has_correct_attrs` | `event_bus.rs` | `tool`, `action`, `success` attrs present | + +Run: `cargo test -p ledgerr-mcp --lib -- event_bus` +Result: **11 passed; 0 failed** + +--- + +## Invariants + +1. Every `tools/call` dispatch fires exactly one `tool_called` event — before the + governance gate, so authorization decisions are also observable. +2. Task operations (`TaskCreate`/`TaskDone`/`TaskUpdate`) fire exactly one typed + event each — no double-fire with the `tool_called` wrapper. +3. Hooks never block dispatch — `ProjectEventBus::fire()` calls hooks synchronously + but panics on lock poison, not on hook errors. +4. Hook registration is additive and ordered — hooks fire in registration order, all + of them, for every event. diff --git a/docs/statechart-unification-plan.md b/docs/statechart-unification-plan.md new file mode 100644 index 00000000..b86a6b5a --- /dev/null +++ b/docs/statechart-unification-plan.md @@ -0,0 +1,155 @@ +# Statechart Unification Plan: b00t × kr0ki × ufo-types + +**Status**: Draft — 2026-10-11 +**Closes**: Issue #259 (statechart integration, harmonization with b00t and ufo-types) + +--- + +## 1. Current State + +Three independent runtime layers that need to cooperate: + +| Layer | Current impl | Gaps | +|-------|-------------|------| +| **b00t HsmMachine** | `crates/ledgerr-mcp/src/hsm.rs` — hand-rolled HSM with state/substate string pairs | No reactive cells; transitions fire strings, not typed events | +| **kr0ki process execution** | `crates/kr0ki-core` (Phase 4 — PR #265) — maps ufo-types SysML-v2 ActionUsage + Succession → `lazily::StateChart` | New; not yet connected to b00t event bus or MCP dispatch | +| **ufo-types SysML model** | `ElementKind::{ActionUsage, ActionDefinition}`, `Relation::Succession` (rev `5f8d018`) | No DecisionNode/ForkNode/JoinNode Rust types yet — these exist in SHACL TTL ontology only | + +--- + +## 2. Target Architecture + +``` +┌──────────────────────────────────────────────────────────────┐ +│ lazily::StateChart │ +│ (Harel/SCXML conformant, reactive Source>,│ +│ ThreadSafeStateChart, Effect hooks — lazily-rs v0.59) │ +└─────────────────────┬────────────────────────────────────────┘ + │ drives + ┌───────────┼─────────────┐ + ▼ ▼ ▼ + ┌──────────┐ ┌─────────┐ ┌──────────────┐ + │ b00t HSM │ │ kr0ki │ │ ProjectEvent │ + │ (ledgerr │ │ Process │ │ Bus │ + │ -mcp) │ │ Chart │ │ (PM hooks) │ + └────┬─────┘ └────┬────┘ └──────┬───────┘ + │ │ │ + └─────────────┼──────────────┘ + ▼ + ufo-types model layer + (ElementKind, Relation) +``` + +### Roles + +- **lazily::StateChart** — single reactive chart runtime for both process and HSM execution. Replaces the hand-rolled `HsmMachine` string dispatch in `hsm.rs`. +- **kr0ki-core::ProcessChart** — maps SysML-v2 process elements (ufo-types) into a `lazily::StateChart`. The translation layer: model → runtime. +- **b00t HsmMachine** — migration target: replace with a `lazily::ChartDef` built from the same Harel hierarchy, driven by `ThreadSafeStateChart::send`. The hook system (`HsmHookRegistry`) becomes `ProjectEventBus` callbacks wired to `StateChart` configuration changes. +- **ProjectEventBus** — the cross-cutting hook layer. Every state change, task event, and tool call fires here. Hooks registered via MCP tool (`ledgerr_workflow/register_project_hook`) or Rust callback. +- **ufo-types** — the model layer supplying `ElementKind` and `Relation` for both SysML-v2 process models (kr0ki) and future verifier models (#218). + +--- + +## 3. Migration Phases + +### Phase 1 (Done — PRs #263, #264) +- `lazily` dep added to workspace +- `HsmHookRegistry` wired to `lazily` (`StateBuilder`/`ChartBuilder`) +- `ProjectEventBus` on `TurboLedgerService` +- `tool_called`, `task_created`, `task_done`, `lifecycle_transition` events + +### Phase 2 (Done — PR #265) +- `crates/kr0ki-core`: `ProcessChart::from_sysml(elements, relations)` → `ThreadSafeStateChart` +- Maps `ActionUsage` → atomic state, `ActionDefinition` → compound root, `Succession` → `"next"` transition + +### Phase 3 (This PR — task/227-sse-streamable-http) +- `WorkflowArgs::TaskCreate / TaskList / TaskDone / TaskUpdate` — first-class task management on `ProjectEventBus` +- `task_updated` event added to bus +- Every task lifecycle fires a typed `ProjectEvent` hook + +### Phase 4 — HsmMachine → lazily (Future) + +Replace `HsmMachine`'s string-pair state with a `ChartDef`: + +```rust +// crates/ledgerr-mcp/src/hsm.rs — MIGRATION TARGET +// Current: +pub struct HsmMachine { state: String, substate: String, … } + +// Future: +pub struct HsmMachine { + chart: ThreadSafeStateChart, + ctx: ThreadSafeContext, +} +// States: "idle.waiting", "ingest.pending", "normalize.ready", … +// HsmHookRegistry → callbacks wired to ctx reactive source invalidation +``` + +Trigger: when `HsmHookRegistry` needs compound state (hierarchical guard), parallel regions, or history pseudo-states — `lazily` provides all three natively. + +### Phase 5 — ufo-types extension (Future, depends on ufo-types upstream) + +`DecisionNode`, `ForkNode`, `JoinNode` as `ElementKind` variants: +- `DecisionNode` → `StateBuilder::compound(id, initial).on_guarded(event, target, guard_name)` +- `ForkNode` → `StateBuilder::parallel(id)` with N children +- `JoinNode` → compound exit transition from all parallel children + +Blocked on: ufo-types adding these variants (issue #218 blocked on ufo-types PR #13). + +--- + +## 4. b00t Hook Contract + +b00t task operations fire `ProjectEvent`s via MCP tool (`ledgerr_workflow`): + +| b00t operation | MCP action | ProjectEvent kind | +|----------------|-----------|-------------------| +| task add | `task_create` | `task_created` | +| task list | `task_list` | `tool_called` | +| task done | `task_done` | `task_done` | +| task next / update | `task_update` | `task_updated` | +| hsm transition | `transition` | `lifecycle_transition` + `task_done` (→ summarize) | +| hsm resume | `resume` | `task_created` (checkpoint) | +| any tool call | any | `tool_called` | + +Hooks register via `{"action":"register_project_hook","event_kind":"*"}` on `ledgerr_workflow`. This is the b00t hook surface — agents wire LFMF actions, cost checkpoints, and state machine gates here. + +--- + +## 5. kr0ki ↔ b00t Integration Point + +`kr0ki-core::ProcessChart` produces a `ThreadSafeStateChart`. To integrate with the b00t PM system: + +```rust +// Proposed API (Phase 4): +impl ProcessChart { + /// Attach a ProjectEventBus so every StateChart transition fires + /// a `lifecycle_transition` event on the bus. + pub fn with_event_bus(self, bus: Arc) -> Self { … } +} +``` + +Each `advance()` call fires `ProjectEvent::lifecycle_transition(from_leaves, to_leaves)` on the attached bus, making kr0ki process steps visible to all registered hooks. + +--- + +## 6. Decision: statecharts.dev vs lazily + +| Tool | Role | +|------|------| +| **statecharts.dev / XState** | Design tool — visual authoring, SCXML export, generates `ChartDef`-compatible JSON | +| **lazily::statechart** | Runtime — executes the chart in Rust, reactive, no JS dependency | +| **cmars/statechart (Go)** | Out of scope — wrong language | + +Workflow: design in statecharts.dev → export SCXML → convert to `ChartBuilder` builder calls (or `ChartDef::from_json` via `statechart-json` feature) → execute in `lazily`. + +--- + +## 7. Open Gaps + +| Gap | Blocker | Issue | +|-----|---------|-------| +| `DecisionNode` / `ForkNode` / `JoinNode` in ufo-types | ufo-types PR #13 | #218 | +| HsmMachine full migration to lazily | Defer until Phase 4 trigger | — | +| `ProcessChart::with_event_bus` integration | Phase 4 | — | +| SCXML export from statecharts.dev → `ChartDef::from_json` pipeline | Future | — |