From 73b8beee7bf64a9b91d2f98ea7f194b16e92b19f Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Wed, 17 Jun 2026 00:25:01 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Implement read-only MCP stdio gateway","authority":"XY-980"} --- README.md | 7 + apps/decodex/src/cli.rs | 99 +- apps/decodex/src/lib.rs | 1 + apps/decodex/src/mcp.rs | 862 ++++++++++++++++++ apps/decodex/src/orchestrator/entrypoints.rs | 69 ++ apps/decodex/src/state/internal.rs | 21 + apps/decodex/src/state/store.rs | 32 + apps/decodex/src/state/tests.rs | 7 + ...p-capability-gateway-and-skill-slimming.md | 7 +- docs/reference/operator-control-plane.md | 9 + docs/spec/runtime.md | 9 + 11 files changed, 1104 insertions(+), 19 deletions(-) create mode 100644 apps/decodex/src/mcp.rs diff --git a/README.md b/README.md index cbcf4b40c..4a2b6e588 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,7 @@ cargo run -p decodex --bin decodex -- intake goal --project decodex --apply cargo run -p decodex --bin decodex -- intake issues --project decodex XY-1 XY-2 --dry-run cargo run -p decodex --bin decodex -- intake issues --project decodex XY-1 XY-2 --apply +cargo run -p decodex --bin decodex -- mcp serve --transport stdio cargo run -p decodex --bin decodex -- radar refresh-upstream-queue cargo run -p decodex --bin decodex -- radar refresh-release-delta cargo run -p decodex --bin decodex -- radar validate @@ -185,6 +186,12 @@ Program, and issue mappings. It never applies or removes service queue labels; r mapped nodes are dispatched directly by the Program scheduler instead of being converted into queued-label work. +`decodex mcp serve --transport stdio` starts the phase-one read-only MCP gateway for +local desktop and CLI clients. The gateway exposes checked-in documentation, checked-in +JSON research reports, runtime Decision Contract readback, local status snapshots, and +lane-control readback as MCP resources only. It does not expose mutating MCP tools. +Stdout is reserved for MCP JSON-RPC messages; diagnostics and logs stay off stdout. + ### Install from Source ```sh diff --git a/apps/decodex/src/cli.rs b/apps/decodex/src/cli.rs index c540ecb19..78aa3db4a 100644 --- a/apps/decodex/src/cli.rs +++ b/apps/decodex/src/cli.rs @@ -20,6 +20,7 @@ use crate::{ archive_hygiene::{self, ArchiveHygieneRequest}, maintenance::{self, MaintenanceMode, MaintenancePruneRequest, MaintenanceScope}, manual::{self, ManualCommitRequest, ManualLandRequest}, + mcp::{self, McpServeRequest, McpTransport}, orchestrator::{ self, DEFAULT_STEER_RESULT_WAIT_TIMEOUT, DiagnoseRequest, EvidenceRequest, IssueDispatchMode, LaneInspectRequest, LaneInterruptRequest, LaneSteerReport, @@ -73,6 +74,7 @@ impl Cli { Command::Land(args) => args.run(), Command::Run(args) => args.run(), Command::Serve(args) => args.run(), + Command::Mcp(args) => args.run(), Command::Project(args) => args.run(), Command::Lane(args) => args.run(), Command::Status(args) => args.run(), @@ -368,6 +370,36 @@ impl ServeCommand { } } +#[derive(Debug, Args)] +struct McpCommand { + #[command(subcommand)] + command: McpSubcommand, +} +impl McpCommand { + fn run(&self) -> Result<()> { + match &self.command { + McpSubcommand::Serve(args) => args.run(), + } + } +} + +#[derive(Debug, Args)] +struct McpServeCommand { + #[command(flatten)] + project_config: ProjectConfigArgs, + /// MCP transport. + #[arg(long, value_enum, default_value_t = McpTransport::Stdio)] + transport: McpTransport, +} +impl McpServeCommand { + fn run(&self) -> Result<()> { + mcp::serve(McpServeRequest { + transport: self.transport, + config_path: self.project_config.as_path(), + }) + } +} + #[derive(Debug, Args)] struct ProjectCommand { #[command(subcommand)] @@ -1634,6 +1666,8 @@ enum Command { Run(RunCommand), /// Run the local multi-project Decodex control plane. Serve(ServeCommand), + /// Serve the read-only Decodex MCP gateway. + Mcp(McpCommand), /// Manage the local Decodex project registry. Project(ProjectCommand), /// Inspect or influence a local lane. @@ -1697,6 +1731,12 @@ enum ProjectSubcommand { Remove(ProjectToggleCommand), } +#[derive(Debug, Subcommand)] +enum McpSubcommand { + /// Serve read-only Decodex resources over MCP. + Serve(McpServeCommand), +} + #[derive(Debug, Subcommand)] enum LaneSubcommand { /// Inspect one local lane by issue identifier or tracker issue id. @@ -1929,22 +1969,26 @@ mod tests { use clap::Parser; - use crate::cli::{ - AccountCommand, AccountSubcommand, AccountUseCommand, AppCommand, AttemptCommand, Cli, - Command, CommitCommand, DiagnoseCommand, EvidenceCommand, IntakeCommand, IntakeGoalCommand, - IntakeIssuesCommand, IntakeSubcommand, LandCommand, LaneCommand, LaneInspectCommand, - LaneInterruptCommand, LaneSteerCommand, LaneSubcommand, LegacyCloseoutRecoveryCommand, - MergedCloseoutRecoveryCommand, ProbeCommand, ProjectCommand, ProjectConfigArgs, - ProjectSubcommand, RadarBackfillReleaseRangeCommand, RadarBundleBuildCommand, - RadarBundleCommand, RadarBundleSubcommand, RadarBundleValidateCommand, RadarCommand, - RadarLedgerCommand, RadarLedgerIngestExistingCommand, RadarLedgerSubcommand, - RadarLedgerSummaryCommand, RadarRefreshReleaseDeltaCommand, - RadarRefreshUpstreamQueueCommand, RadarRenderSignalCommand, RadarSubcommand, - RadarValidateCommand, RecoverCommand, RecoverSubcommand, ResearchCommand, - ResearchCompileCommand, ResearchOutcomeArg, ResearchPromoteCommand, ResearchSubcommand, - ReviewHandoffAdoptCommand, ReviewHandoffDiagnoseCommand, ReviewHandoffRebindCommand, - ReviewHandoffRecoveryCommand, ReviewHandoffRecoverySubcommand, RunCommand, ServeCommand, - StatusCommand, + use crate::{ + cli::{ + AccountCommand, AccountSubcommand, AccountUseCommand, AppCommand, AttemptCommand, Cli, + Command, CommitCommand, DiagnoseCommand, EvidenceCommand, IntakeCommand, + IntakeGoalCommand, IntakeIssuesCommand, IntakeSubcommand, LandCommand, LaneCommand, + LaneInspectCommand, LaneInterruptCommand, LaneSteerCommand, LaneSubcommand, + LegacyCloseoutRecoveryCommand, McpCommand, McpServeCommand, McpSubcommand, + MergedCloseoutRecoveryCommand, ProbeCommand, ProjectCommand, ProjectConfigArgs, + ProjectSubcommand, RadarBackfillReleaseRangeCommand, RadarBundleBuildCommand, + RadarBundleCommand, RadarBundleSubcommand, RadarBundleValidateCommand, RadarCommand, + RadarLedgerCommand, RadarLedgerIngestExistingCommand, RadarLedgerSubcommand, + RadarLedgerSummaryCommand, RadarRefreshReleaseDeltaCommand, + RadarRefreshUpstreamQueueCommand, RadarRenderSignalCommand, RadarSubcommand, + RadarValidateCommand, RecoverCommand, RecoverSubcommand, ResearchCommand, + ResearchCompileCommand, ResearchOutcomeArg, ResearchPromoteCommand, ResearchSubcommand, + ReviewHandoffAdoptCommand, ReviewHandoffDiagnoseCommand, ReviewHandoffRebindCommand, + ReviewHandoffRecoveryCommand, ReviewHandoffRecoverySubcommand, RunCommand, + ServeCommand, StatusCommand, + }, + mcp::McpTransport, }; #[test] @@ -2191,6 +2235,29 @@ mod tests { } } + #[test] + fn parses_mcp_stdio_serve() { + let cli = Cli::parse_from([ + "decodex", + "mcp", + "serve", + "--config", + "./project.toml", + "--transport", + "stdio", + ]); + + assert!(matches!( + cli.command, + Command::Mcp(McpCommand { + command: McpSubcommand::Serve(McpServeCommand { + project_config: ProjectConfigArgs { config: Some(config) }, + transport: McpTransport::Stdio, + }) + }) if config == Path::new("./project.toml") + )); + } + #[test] fn parses_radar_validate_paths() { let cli = Cli::parse_from(["decodex", "radar", "validate", "artifacts/github/bundles"]); diff --git a/apps/decodex/src/lib.rs b/apps/decodex/src/lib.rs index 492fd4293..4dc389cca 100644 --- a/apps/decodex/src/lib.rs +++ b/apps/decodex/src/lib.rs @@ -18,6 +18,7 @@ mod github; mod loop_contract; mod maintenance; mod manual; +mod mcp; mod orchestrator; mod program_intake; mod pull_request; diff --git a/apps/decodex/src/mcp.rs b/apps/decodex/src/mcp.rs new file mode 100644 index 000000000..60274580a --- /dev/null +++ b/apps/decodex/src/mcp.rs @@ -0,0 +1,862 @@ +use std::{ + env, + fmt::Display, + fs, + io::{self, BufRead as _, BufReader, ErrorKind, Read, Write}, + path::{Path, PathBuf}, +}; + +use clap::ValueEnum; +use reqwest::Url; +use serde::{Deserialize, Serialize}; +use serde_json::{self, Value}; + +use crate::{ + config::ServiceConfig, + orchestrator, + prelude::{Result, eyre}, + runtime, + state::StateStore, +}; + +const MCP_PROTOCOL_VERSION: &str = "2025-11-25"; +const SERVER_NAME: &str = "decodex"; +const DOCS_HOST: &str = "docs"; +const RESEARCH_HOST: &str = "research"; +const DECISION_CONTRACTS_HOST: &str = "decision-contracts"; +const PROJECTS_HOST: &str = "projects"; +const RESOURCE_NOT_FOUND_CODE: i64 = -32_002; +const DEFAULT_MCP_STATUS_LIMIT: usize = 10; + +/// MCP transport supported by the native Decodex gateway. +#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] +#[value(rename_all = "kebab-case")] +pub(crate) enum McpTransport { + /// JSON-RPC messages over stdin/stdout. + Stdio, +} + +/// Request to start the native Decodex MCP gateway. +#[derive(Clone, Copy, Debug)] +pub(crate) struct McpServeRequest<'a> { + pub(crate) transport: McpTransport, + pub(crate) config_path: Option<&'a Path>, +} + +struct McpServer { + context: McpContext, +} +impl McpServer { + fn handle_line(&self, line: &str) -> Option { + let parsed = serde_json::from_str::(line); + let value = match parsed { + Ok(value) => value, + Err(_) => return Some(json_rpc_error(Value::Null, -32_700, "Parse error")), + }; + let request = match serde_json::from_value::(value) { + Ok(request) => request, + Err(_) => return Some(json_rpc_error(Value::Null, -32_600, "Invalid Request")), + }; + + self.handle_request(request) + } + + fn handle_request(&self, request: JsonRpcRequest) -> Option { + let id = request.id?; + + if request.jsonrpc.as_deref() != Some("2.0") { + return Some(json_rpc_error(id, -32_600, "Invalid Request")); + } + + let Some(method) = request.method else { + return Some(json_rpc_error(id, -32_600, "Invalid Request")); + }; + let result = match method.as_str() { + "initialize" => Ok(self.initialize()), + "ping" => Ok(serde_json::json!({})), + "resources/list" => self.list_resources(), + "resources/read" => self.read_resource(request.params), + _ => Err(McpError::method_not_found()), + }; + + Some(match result { + Ok(result) => serde_json::json!({ "jsonrpc": "2.0", "id": id, "result": result }), + Err(error) => json_rpc_error(id, error.code, &error.message), + }) + } + + fn initialize(&self) -> Value { + serde_json::json!({ + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": { + "resources": {} + }, + "serverInfo": { + "name": SERVER_NAME, + "version": env!("CARGO_PKG_VERSION") + } + }) + } + + fn list_resources(&self) -> Result { + let mut resources = self.context.docs_resources()?; + + resources.extend(self.context.decision_contract_resources()?); + + if let Some(project_id) = self.context.project_id() { + resources.push(McpResource::json( + format!("decodex://projects/{project_id}/status"), + format!("Project {project_id} status"), + "Read-only local runtime status snapshot.", + )); + resources.push(McpResource::json( + format!("decodex://projects/{project_id}/lane-control"), + format!("Project {project_id} lane-control readback"), + "Read-only lane-control state for current and recent local lanes.", + )); + } + + Ok(serde_json::json!({ "resources": resources })) + } + + fn read_resource(&self, params: Option) -> Result { + let params = params.ok_or_else(McpError::invalid_params)?; + let params = serde_json::from_value::(params) + .map_err(|_| McpError::invalid_params())?; + let content = self.context.read_resource(¶ms.uri)?; + + Ok(serde_json::json!({ + "contents": [ + { + "uri": content.uri, + "mimeType": content.mime_type, + "text": content.text + } + ] + })) + } +} + +struct McpContext { + repo_root: PathBuf, + config_path: Option, + project_id: Option, + state_store: Option, +} +impl McpContext { + fn for_process(config_path: Option<&Path>) -> Result { + let state_store = runtime::open_runtime_store_lazy().ok(); + let config_path = resolve_context_config_path(config_path, state_store.as_ref())?; + let config = config_path.as_ref().map(ServiceConfig::from_path).transpose()?; + let repo_root = config + .as_ref() + .map(|config| config.repo_root().to_path_buf()) + .or_else(|| discover_repo_root_from_current_dir().ok().flatten()) + .ok_or_else(|| { + eyre::eyre!( + "Failed to find the Decodex repository root for MCP docs resources; start from a checkout or pass --config." + ) + })?; + let project_id = config.map(|config| config.service_id().to_owned()); + + Ok(Self { repo_root, config_path, project_id, state_store }) + } + + fn project_id(&self) -> Option<&str> { + self.project_id.as_deref() + } + + fn docs_resources(&self) -> Result, McpError> { + let mut resources = Vec::new(); + + push_file_resource( + &mut resources, + self.repo_root.join("docs/index.md"), + "decodex://docs/index", + "Documentation index", + "Checked-in Decodex documentation router.", + ); + push_file_resource( + &mut resources, + self.repo_root.join("docs/policy.md"), + "decodex://docs/policy", + "Documentation policy", + "Checked-in Decodex documentation policy.", + ); + + for lane in ["spec", "runbook", "reference", "decisions"] { + let docs_dir = self.repo_root.join("docs").join(lane); + + for entry in read_sorted_dir(&docs_dir)? { + let Some(stem) = markdown_stem(&entry) else { + continue; + }; + + resources.push(McpResource::markdown( + format!("decodex://docs/{lane}/{stem}"), + format!("docs/{lane}/{stem}.md"), + "Checked-in Decodex documentation resource.", + )); + } + } + for entry in read_sorted_dir(&self.repo_root.join("docs/research"))? { + let Some(stem) = json_stem(&entry) else { + continue; + }; + + resources.push(McpResource::json( + format!("decodex://research/{stem}"), + format!("docs/research/{stem}.json"), + "Checked-in JSON research report.", + )); + } + + Ok(resources) + } + + fn decision_contract_resources(&self) -> Result, McpError> { + let Some(project_id) = self.project_id.as_deref() else { + return Ok(Vec::new()); + }; + let Some(state_store) = self.state_store.as_ref() else { + return Ok(Vec::new()); + }; + let records = state_store + .list_decision_contracts_for_project(project_id) + .map_err(McpError::internal)?; + + Ok(records + .into_iter() + .map(|record| { + McpResource::json( + format!("decodex://decision-contracts/{}", record.contract_id()), + format!("Decision Contract {}", record.contract_id()), + "Read-only local runtime Decision Contract readback.", + ) + }) + .collect()) + } + + fn read_resource(&self, uri: &str) -> Result { + let resource_uri = ResourceUri::parse(uri)?; + + match resource_uri.host.as_str() { + DOCS_HOST => self.read_docs_resource(&resource_uri), + RESEARCH_HOST => self.read_research_resource(&resource_uri), + DECISION_CONTRACTS_HOST => self.read_decision_contract_resource(&resource_uri), + PROJECTS_HOST => self.read_project_resource(&resource_uri), + _ => Err(McpError::resource_not_found()), + } + } + + fn read_docs_resource(&self, uri: &ResourceUri) -> Result { + let path = match uri.segments.as_slice() { + [segment] if segment == "index" => self.repo_root.join("docs/index.md"), + [segment] if segment == "policy" => self.repo_root.join("docs/policy.md"), + [lane, topic] if docs_lane_allowed(lane) && safe_resource_stem(topic) => + self.repo_root.join("docs").join(lane).join(format!("{topic}.md")), + _ => return Err(McpError::resource_not_found()), + }; + + read_file_resource(&uri.raw, path, "text/markdown") + } + + fn read_research_resource(&self, uri: &ResourceUri) -> Result { + let [artifact] = uri.segments.as_slice() else { + return Err(McpError::resource_not_found()); + }; + + if !safe_research_artifact(artifact) { + return Err(McpError::resource_not_found()); + } + + let file_name = if artifact.ends_with(".json") { + artifact.to_owned() + } else { + format!("{artifact}.json") + }; + + read_file_resource( + &uri.raw, + self.repo_root.join("docs/research").join(file_name), + "application/json", + ) + } + + fn read_decision_contract_resource( + &self, + uri: &ResourceUri, + ) -> Result { + let [contract_id] = uri.segments.as_slice() else { + return Err(McpError::resource_not_found()); + }; + + if !safe_runtime_identifier(contract_id) { + return Err(McpError::resource_not_found()); + } + + let Some(project_id) = self.project_id.as_deref() else { + return Err(McpError::resource_not_found()); + }; + let Some(state_store) = self.state_store.as_ref() else { + return Err(McpError::resource_not_found()); + }; + let Some(record) = + state_store.decision_contract(project_id, contract_id).map_err(McpError::internal)? + else { + return Err(McpError::resource_not_found()); + }; + let value = serde_json::json!({ + "schema": "decodex.mcp.decision_contract_resource/1", + "project_id": record.project_id(), + "source_issue_id": record.source_issue_id(), + "status": record.status(), + "created_at": record.created_at(), + "updated_at": record.updated_at(), + "decision_contract": record.contract() + }); + + ResourceContent::json(&uri.raw, value) + } + + fn read_project_resource(&self, uri: &ResourceUri) -> Result { + let [project_id, resource_kind, rest @ ..] = uri.segments.as_slice() else { + return Err(McpError::resource_not_found()); + }; + + if Some(project_id.as_str()) != self.project_id.as_deref() { + return Err(McpError::resource_not_found()); + } + + let Some(config_path) = self.config_path.as_deref() else { + return Err(McpError::resource_not_found()); + }; + let value = match (resource_kind.as_str(), rest) { + ("status", []) => + orchestrator::build_mcp_status_resource(Some(config_path), DEFAULT_MCP_STATUS_LIMIT), + ("lane-control", []) => orchestrator::build_mcp_lane_control_resource( + Some(config_path), + None, + None, + DEFAULT_MCP_STATUS_LIMIT, + ), + ("lane-control", [issue]) if safe_runtime_identifier(issue) => + orchestrator::build_mcp_lane_control_resource( + Some(config_path), + Some(issue), + None, + DEFAULT_MCP_STATUS_LIMIT, + ), + _ => return Err(McpError::resource_not_found()), + } + .map_err(McpError::internal)?; + + ResourceContent::json(&uri.raw, value) + } +} + +#[derive(Deserialize)] +struct JsonRpcRequest { + jsonrpc: Option, + id: Option, + method: Option, + params: Option, +} + +#[derive(Deserialize)] +struct ReadResourceParams { + uri: String, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +struct McpResource { + uri: String, + name: String, + description: String, + mime_type: String, +} +impl McpResource { + fn markdown( + uri: impl Into, + name: impl Into, + description: impl Into, + ) -> Self { + Self { + uri: uri.into(), + name: name.into(), + description: description.into(), + mime_type: String::from("text/markdown"), + } + } + + fn json( + uri: impl Into, + name: impl Into, + description: impl Into, + ) -> Self { + Self { + uri: uri.into(), + name: name.into(), + description: description.into(), + mime_type: String::from("application/json"), + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct ResourceContent { + uri: String, + mime_type: String, + text: String, +} +impl ResourceContent { + fn json(uri: &str, value: Value) -> Result { + let text = serde_json::to_string_pretty(&value).map_err(McpError::internal)?; + + Ok(Self { uri: uri.to_owned(), mime_type: String::from("application/json"), text }) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct ResourceUri { + raw: String, + host: String, + segments: Vec, +} +impl ResourceUri { + fn parse(uri: &str) -> Result { + let parsed = Url::parse(uri).map_err(|_| McpError::invalid_params())?; + + if parsed.scheme() != "decodex" { + return Err(McpError::invalid_params()); + } + + let host = parsed.host_str().map(str::to_owned).ok_or_else(McpError::invalid_params)?; + let segments = parsed + .path_segments() + .map(|segments| { + segments + .filter(|segment| !segment.is_empty()) + .map(str::to_owned) + .collect::>() + }) + .unwrap_or_default(); + + Ok(Self { raw: uri.to_owned(), host, segments }) + } +} + +#[derive(Debug)] +struct McpError { + code: i64, + message: String, +} +impl McpError { + fn invalid_params() -> Self { + Self { code: -32_602, message: String::from("Invalid params") } + } + + fn method_not_found() -> Self { + Self { code: -32_601, message: String::from("Method not found") } + } + + fn resource_not_found() -> Self { + Self { code: RESOURCE_NOT_FOUND_CODE, message: String::from("Resource not found") } + } + + fn internal(error: impl Display) -> Self { + tracing::warn!(error = %error, "MCP resource read failed."); + + Self { code: -32_603, message: String::from("Internal error") } + } +} + +/// Start the read-only Decodex MCP gateway. +pub(crate) fn serve(request: McpServeRequest<'_>) -> Result<()> { + match request.transport { + McpTransport::Stdio => { + let context = McpContext::for_process(request.config_path)?; + let stdin = io::stdin(); + let stdout = io::stdout(); + + serve_stdio_with_context(stdin.lock(), stdout.lock(), context) + }, + } +} + +fn serve_stdio_with_context(reader: R, mut writer: W, context: McpContext) -> Result<()> +where + R: Read, + W: Write, +{ + let server = McpServer { context }; + let reader = BufReader::new(reader); + + for line in reader.lines() { + let line = line?; + + if line.trim().is_empty() { + continue; + } + + if let Some(response) = server.handle_line(&line) { + write_json_line(&mut writer, &response)?; + } + } + + Ok(()) +} + +fn write_json_line(writer: &mut W, value: &Value) -> Result<()> +where + W: Write, +{ + let line = serde_json::to_string(value)?; + + match writer + .write_all(line.as_bytes()) + .and_then(|()| writer.write_all(b"\n")) + .and_then(|()| writer.flush()) + { + Ok(()) => Ok(()), + Err(error) if error.kind() == ErrorKind::BrokenPipe => Ok(()), + Err(error) => Err(error.into()), + } +} + +fn resolve_context_config_path( + explicit_path: Option<&Path>, + state_store: Option<&StateStore>, +) -> Result> { + if let Some(path) = explicit_path { + return Ok(Some(path.to_path_buf())); + } + + let Some(state_store) = state_store else { + return Ok(None); + }; + + runtime::registered_config_path_for_cwd(state_store, &env::current_dir()?) +} + +fn discover_repo_root_from_current_dir() -> Result> { + let mut candidate = env::current_dir()?; + + loop { + if candidate.join("docs/index.md").is_file() && candidate.join("Cargo.toml").is_file() { + return Ok(Some(candidate)); + } + if !candidate.pop() { + return Ok(None); + } + } +} + +fn json_rpc_error(id: Value, code: i64, message: &str) -> Value { + serde_json::json!({ + "jsonrpc": "2.0", + "id": id, + "error": { + "code": code, + "message": message + } + }) +} + +fn push_file_resource( + resources: &mut Vec, + path: PathBuf, + uri: &str, + name: &str, + description: &str, +) { + if path.is_file() { + resources.push(McpResource::markdown(uri, name, description)); + } +} + +fn read_sorted_dir(path: &Path) -> Result, McpError> { + let entries = match fs::read_dir(path) { + Ok(entries) => entries, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(McpError::internal(error)), + }; + let mut paths = entries + .map(|entry| entry.map(|entry| entry.path()).map_err(McpError::internal)) + .collect::, _>>()?; + + paths.sort(); + + Ok(paths) +} + +fn markdown_stem(path: &Path) -> Option { + if path.extension().and_then(|extension| extension.to_str()) != Some("md") { + return None; + } + + path.file_stem().and_then(|stem| stem.to_str()).map(str::to_owned) +} + +fn json_stem(path: &Path) -> Option { + if path.extension().and_then(|extension| extension.to_str()) != Some("json") { + return None; + } + + path.file_stem().and_then(|stem| stem.to_str()).map(str::to_owned) +} + +fn read_file_resource( + uri: &str, + path: PathBuf, + mime_type: &str, +) -> Result { + let text = fs::read_to_string(path).map_err(|error| match error.kind() { + ErrorKind::NotFound => McpError::resource_not_found(), + _ => McpError::internal(error), + })?; + + Ok(ResourceContent { uri: uri.to_owned(), mime_type: mime_type.to_owned(), text }) +} + +fn docs_lane_allowed(lane: &str) -> bool { + matches!(lane, "spec" | "runbook" | "reference" | "decisions") +} + +fn safe_resource_stem(value: &str) -> bool { + !value.is_empty() + && value.bytes().all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) +} + +fn safe_research_artifact(value: &str) -> bool { + !value.is_empty() + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')) + && !value.contains("..") +} + +fn safe_runtime_identifier(value: &str) -> bool { + !value.is_empty() + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')) + && !value.contains("..") +} + +#[cfg(test)] +mod tests { + use std::{fs, io::Cursor, path::Path}; + + use serde_json::Value; + use tempfile::TempDir; + + use crate::{ + loop_contract::DecisionContract, + mcp::{self, McpContext}, + state::StateStore, + }; + + #[test] + fn initialize_exposes_read_only_resource_capability() { + let repo = test_repo(); + let responses = + run_stdio(repo.path(), r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#); + let response = response_at(&responses, 0); + let result = response.get("result").and_then(Value::as_object).expect("result object"); + let capabilities = + result.get("capabilities").and_then(Value::as_object).expect("capabilities object"); + + assert!(capabilities.contains_key("resources")); + assert!(!capabilities.contains_key("tools")); + } + + #[test] + fn resources_list_includes_docs_decisions_and_research_json() { + let repo = test_repo(); + let responses = run_stdio( + repo.path(), + r#"{"jsonrpc":"2.0","id":1,"method":"resources/list","params":{}}"#, + ); + let resources = + response_at(&responses, 0)["result"]["resources"].as_array().expect("resources array"); + let uris = resources + .iter() + .filter_map(|resource| resource.get("uri").and_then(Value::as_str)) + .collect::>(); + + assert!(uris.contains(&"decodex://docs/index")); + assert!(uris.contains(&"decodex://docs/spec/runtime")); + assert!(uris.contains(&"decodex://docs/decisions/mcp-gateway")); + assert!(uris.contains(&"decodex://research/sample-report")); + } + + #[test] + fn resources_list_includes_runtime_decision_contracts() { + let repo = test_repo(); + let state_store = StateStore::open_in_memory().expect("state store should open"); + + state_store + .upsert_decision_contract("decodex", Some("XY-852"), latent_decision_contract_fixture()) + .expect("decision contract should persist"); + + let responses = run_stdio_with_context( + McpContext { + repo_root: repo.path().to_path_buf(), + config_path: None, + project_id: Some(String::from("decodex")), + state_store: Some(state_store), + }, + r#"{"jsonrpc":"2.0","id":1,"method":"resources/list","params":{}}"#, + ); + let resources = + response_at(&responses, 0)["result"]["resources"].as_array().expect("resources array"); + let uris = resources + .iter() + .filter_map(|resource| resource.get("uri").and_then(Value::as_str)) + .collect::>(); + + assert!(uris.contains(&"decodex://decision-contracts/research-x-loop-contract")); + } + + #[test] + fn resources_read_returns_runtime_decision_contract() { + let repo = test_repo(); + let state_store = StateStore::open_in_memory().expect("state store should open"); + + state_store + .upsert_decision_contract("decodex", Some("XY-852"), latent_decision_contract_fixture()) + .expect("decision contract should persist"); + + let responses = run_stdio_with_context( + McpContext { + repo_root: repo.path().to_path_buf(), + config_path: None, + project_id: Some(String::from("decodex")), + state_store: Some(state_store), + }, + r#"{"jsonrpc":"2.0","id":1,"method":"resources/read","params":{"uri":"decodex://decision-contracts/research-x-loop-contract"}}"#, + ); + let contents = + response_at(&responses, 0)["result"]["contents"].as_array().expect("contents array"); + let text = contents[0]["text"].as_str().expect("text content"); + let content: Value = serde_json::from_str(text).expect("decision contract should be json"); + + assert_eq!(content["project_id"], "decodex"); + assert_eq!(content["decision_contract"]["contract_id"], "research-x-loop-contract"); + } + + #[test] + fn resources_read_returns_checked_in_doc_text() { + let repo = test_repo(); + let responses = run_stdio( + repo.path(), + r#"{"jsonrpc":"2.0","id":1,"method":"resources/read","params":{"uri":"decodex://docs/spec/runtime"}}"#, + ); + let contents = + response_at(&responses, 0)["result"]["contents"].as_array().expect("contents array"); + let text = contents[0]["text"].as_str().expect("text content"); + + assert_eq!(text, "# Runtime\n\nSpec body.\n"); + } + + #[test] + fn resources_read_rejects_invalid_resource_uri() { + let repo = test_repo(); + let responses = run_stdio( + repo.path(), + r#"{"jsonrpc":"2.0","id":1,"method":"resources/read","params":{"uri":"file:///etc/passwd"}}"#, + ); + let error = response_at(&responses, 0).get("error").expect("error response"); + + assert_eq!(error["code"], -32_602); + } + + #[test] + fn stdio_output_contains_only_json_rpc_responses() { + let repo = test_repo(); + let input = [ + r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#, + r#"{"jsonrpc":"2.0","method":"notifications/initialized","params":{}}"#, + r#"{"jsonrpc":"2.0","id":2,"method":"resources/list","params":{}}"#, + ] + .join("\n"); + let output = run_stdio_raw(repo.path(), &input); + let lines = output.lines().collect::>(); + + assert_eq!(lines.len(), 2); + + for line in lines { + let value = serde_json::from_str::(line).expect("stdout line should be JSON"); + + assert_eq!(value["jsonrpc"], "2.0"); + } + } + + fn run_stdio(repo_root: &Path, input: &str) -> Vec { + run_stdio_raw(repo_root, input) + .lines() + .map(|line| serde_json::from_str::(line).expect("response should be JSON")) + .collect() + } + + fn run_stdio_with_context(context: McpContext, input: &str) -> Vec { + run_stdio_raw_with_context(context, input) + .lines() + .map(|line| serde_json::from_str::(line).expect("response should be JSON")) + .collect() + } + + fn run_stdio_raw(repo_root: &Path, input: &str) -> String { + let context = McpContext { + repo_root: repo_root.to_path_buf(), + config_path: None, + project_id: None, + state_store: None, + }; + + run_stdio_raw_with_context(context, input) + } + + fn run_stdio_raw_with_context(context: McpContext, input: &str) -> String { + let mut output = Vec::new(); + + mcp::serve_stdio_with_context(Cursor::new(format!("{input}\n")), &mut output, context) + .expect("stdio server should run"); + + String::from_utf8(output).expect("stdout should be utf-8") + } + + fn response_at(responses: &[Value], index: usize) -> &Value { + responses.get(index).expect("response should exist") + } + + fn test_repo() -> TempDir { + let repo = TempDir::new().expect("temp repo should exist"); + + write_file(repo.path().join("Cargo.toml"), "[workspace]\n"); + write_file(repo.path().join("docs/index.md"), "# Docs\n"); + write_file(repo.path().join("docs/policy.md"), "# Policy\n"); + write_file(repo.path().join("docs/spec/runtime.md"), "# Runtime\n\nSpec body.\n"); + write_file(repo.path().join("docs/decisions/mcp-gateway.md"), "# MCP\n"); + write_file(repo.path().join("docs/research/sample-report.json"), "{}\n"); + + repo + } + + fn write_file(path: std::path::PathBuf, contents: &str) { + let parent = path.parent().expect("test path should have parent"); + + fs::create_dir_all(parent).expect("parent directory should exist"); + fs::write(path, contents).expect("test file should write"); + } + + fn latent_decision_contract_fixture() -> DecisionContract { + serde_json::from_str(include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/fixtures/decision_contract/research_x_latent_contract.json" + ))) + .expect("research X latent contract fixture should deserialize") + } +} diff --git a/apps/decodex/src/orchestrator/entrypoints.rs b/apps/decodex/src/orchestrator/entrypoints.rs index 74514602f..c240b6d63 100644 --- a/apps/decodex/src/orchestrator/entrypoints.rs +++ b/apps/decodex/src/orchestrator/entrypoints.rs @@ -304,6 +304,75 @@ pub(crate) fn print_status( Ok(()) } +pub(crate) fn build_mcp_status_resource( + config_path: Option<&Path>, + limit: usize, +) -> Result { + if limit == 0 { + eyre::bail!("MCP status resource limit must be greater than zero."); + } + + let state_store = runtime::open_runtime_store_lazy()?; + let Some(config_path) = resolve_config_path(config_path, &state_store)? else { + eyre::bail!( + "No Decodex project config found. Start MCP from a registered checkout or pass --config." + ); + }; + let config = ServiceConfig::from_path(&config_path)?; + let mut snapshot = build_operator_status_snapshot_with_account_mode( + &config, + &state_store, + limit, + AccountActivityMode::Snapshot, + )?; + + snapshot.status_source = Some(String::from("local_runtime")); + + serde_json::to_value(snapshot).map_err(Into::into) +} + +pub(crate) fn build_mcp_lane_control_resource( + config_path: Option<&Path>, + issue: Option<&str>, + run_id: Option<&str>, + limit: usize, +) -> Result { + if limit == 0 { + eyre::bail!("MCP lane-control resource limit must be greater than zero."); + } + + let state_store = runtime::open_runtime_store_lazy()?; + let Some(config_path) = resolve_config_path(config_path, &state_store)? else { + eyre::bail!( + "No Decodex project config found. Start MCP from a registered checkout or pass --config." + ); + }; + let config = ServiceConfig::from_path(&config_path)?; + + if let Some(issue) = issue { + let report = lane_control::build_lane_inspect_report(&state_store, &config, issue, run_id)?; + + return serde_json::to_value(report).map_err(Into::into); + } + + let snapshot = build_operator_status_snapshot_with_account_mode( + &config, + &state_store, + limit, + AccountActivityMode::Snapshot, + )?; + + Ok(json!({ + "schema": "decodex.mcp.lane_control_readback/1", + "project_id": snapshot.project_id, + "read_only": true, + "mutating_tools": [], + "current_lanes": snapshot.current_lanes, + "recent_runs": snapshot.recent_runs, + "post_review_lanes": snapshot.post_review_lanes + })) +} + pub(crate) fn run_diagnose(request: DiagnoseRequest<'_>) -> Result<()> { if request.limit == 0 { eyre::bail!("`diagnose --limit` must be greater than zero."); diff --git a/apps/decodex/src/state/internal.rs b/apps/decodex/src/state/internal.rs index 5c397a8de..69b8cd7b1 100644 --- a/apps/decodex/src/state/internal.rs +++ b/apps/decodex/src/state/internal.rs @@ -2167,6 +2167,27 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(records) } + fn list_decision_contracts_for_project( + &self, + project_id: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, contract_id, source_issue_id, status, payload_json, created_at, \ + created_at_unix, updated_at, updated_at_unix \ + FROM decision_contracts \ + WHERE project_id = ?1 \ + ORDER BY created_at_unix ASC, contract_id ASC", + )?; + let rows = statement.query_map(params![project_id], decision_contract_runtime_row_parts)?; + let mut records = Vec::new(); + + for row in rows { + records.push(decision_contract_record_from_row_parts(row?)?); + } + + Ok(records) + } + fn load_execution_programs(&self, state: &mut StateData) -> Result<()> { let mut statement = self.connection.prepare( "SELECT project_id, program_id, source_contract_id, payload_json, created_at, \ diff --git a/apps/decodex/src/state/store.rs b/apps/decodex/src/state/store.rs index b6f1336c8..a66a76254 100644 --- a/apps/decodex/src/state/store.rs +++ b/apps/decodex/src/state/store.rs @@ -1914,6 +1914,38 @@ impl StateStore { Ok(records.into_iter().map(|record| record.as_public()).collect()) } + /// List local Loop/Decision Contracts for one project. + #[allow(dead_code)] + pub(crate) fn list_decision_contracts_for_project( + &self, + project_id: &str, + ) -> Result> { + validate_required_decision_contract_field("project_id", project_id)?; + + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + let records = sqlite + .list_decision_contracts_for_project(project_id)? + .into_iter() + .map(|record| record.as_public()) + .collect(); + + return Ok(records); + } + + let state = self.lock()?; + let mut records = state + .decision_contracts + .values() + .filter(|record| record.project_id == project_id) + .cloned() + .collect::>(); + + records.sort_by(compare_decision_contract_runtime_records); + + Ok(records.into_iter().map(|record| record.as_public()).collect()) + } + /// Promote a latent Loop/Decision Contract into accepted execution authority. #[allow(dead_code)] pub(crate) fn promote_decision_contract( diff --git a/apps/decodex/src/state/tests.rs b/apps/decodex/src/state/tests.rs index 6130cdc24..2fbe7153a 100644 --- a/apps/decodex/src/state/tests.rs +++ b/apps/decodex/src/state/tests.rs @@ -3115,6 +3115,13 @@ fn decision_contracts_persist_reload_and_promote_without_linear_mirror() { assert_eq!(issue_contracts.len(), 1); assert_eq!(issue_contracts[0].contract_id(), "research-x-loop-contract"); + + let project_contracts = reopened + .list_decision_contracts_for_project("decodex") + .expect("project contracts should list"); + + assert_eq!(project_contracts.len(), 1); + assert_eq!(project_contracts[0].contract_id(), "research-x-loop-contract"); } #[test] diff --git a/docs/decisions/mcp-capability-gateway-and-skill-slimming.md b/docs/decisions/mcp-capability-gateway-and-skill-slimming.md index d6ae67366..f6839cdf1 100644 --- a/docs/decisions/mcp-capability-gateway-and-skill-slimming.md +++ b/docs/decisions/mcp-capability-gateway-and-skill-slimming.md @@ -174,6 +174,7 @@ Target shape: ## Open Follow-Up -Implementation is intentionally not included here. The next promoted work should be -split into one read-only MCP gateway issue, one research compile/promote tool issue, -one skill-slimming eval issue, and one docs/resource validation issue. +The first promoted implementation is the read-only stdio gateway exposed as +`decodex mcp serve --transport stdio`. Later promoted work should stay split across +research compile/promote tools, skill-slimming eval, and docs/resource validation +lanes so mutating MCP tools do not bypass Decision Contract or lane-control authority. diff --git a/docs/reference/operator-control-plane.md b/docs/reference/operator-control-plane.md index 8b131a426..e7a26ffe9 100644 --- a/docs/reference/operator-control-plane.md +++ b/docs/reference/operator-control-plane.md @@ -68,6 +68,11 @@ Use `decodex app` to open the installed macOS app from the CLI; use the caller's environment, so `DECODEX_APP_SERVER_URL` remains an explicit App preview override when set. +Local MCP hosts can use `decodex mcp serve --transport stdio` for read-only resource +access. The stdio gateway lists and reads checked-in docs, checked-in JSON research +reports, Decision Contract readback, status snapshots, and lane-control readback. It +does not serve Streamable HTTP and does not expose mutating MCP tools in this phase. + `decodex serve` has two hardcoded scheduler cadences: - The local control-plane loop publishes operator snapshots every 15 seconds. @@ -305,6 +310,10 @@ command surface. After a steer request is handled, current lane protocol activity may show a compact `turn/steer` entry with outcome, failure class, and response turn id. It does not include the operator message. +MCP lane-control resources are readback only and mirror the local status/inspect +projection. Operators must still use `decodex lane inspect`, `decodex lane interrupt`, +`decodex lane steer`, or the local `/api/lane/*` endpoints for supported CLI/API +controls. Snapshot `warnings` remain stable machine-readable tokens. When a warning needs operator action, snapshots may also include `warning_details` entries with the affected `project_id`, `repo_root`, reason, and next action; for example, a stale diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index 6a7fd5761..c9e484866 100644 --- a/docs/spec/runtime.md +++ b/docs/spec/runtime.md @@ -63,6 +63,11 @@ state or this state machine. - `decodex research compile` and `decodex research promote` are runtime-local Decision Contract writes. They update the SQLite `decision_contracts` surface and do not by themselves create Linear issues, queue intent, goals, or executable lanes. +- `decodex mcp serve --transport stdio` is the local read-only MCP gateway. Phase one + exposes MCP resources that read checked-in docs, checked-in JSON research reports, + runtime Decision Contracts, local status snapshots, and lane-control readback. It + must not expose mutating MCP tools, and stdout must contain only valid MCP JSON-RPC + messages. - `decodex intake goal --dry-run` is a tracker-read-only and runtime-read-only operator surface for promoted Decision Contracts. It renders the proposed normal Linear issue split, dependencies, conflict domains, and dispatch plan @@ -878,6 +883,10 @@ After a process restart, recent-run history, run lease ownership, retained post- - `thread_id = null` is expected only before the worker creates the Codex thread for the current run. `event_count = 0`, `last_event_type = null`, and `last_event_at = null` are expected only before the first protocol event for that same run. After the thread exists and protocol activity has started, those empty values indicate missing hydration rather than normal progress. - Operator snapshots may expose an additive `protocol_activity` object derived from app-server structured messages for the current run. The object stays local/operator-only and should summarize turn status, waiting reason, rate-limit status, and a compact recent event list for high-value app-server activity such as `turn/started`, `turn/completed`, plan updates, diff updates, item start/completion, command output deltas, server request responses, account updates, and rate-limit updates. Missing `protocol_activity` means no structured summary was captured yet; consumers must continue to rely on the older `event_count`, `last_event_type`, `last_event_at`, thread fields, and `child_agent_activity` fields when it is absent. Presence in `protocol_activity` is not by itself meaningful progress; non-work account, rate-limit, phase-goal, passive status, warning, model-routing, and token-usage events must remain distinguishable from work-progress events through `last_progress_at` and `progress_diagnostic`. - The operator snapshot transport must stay local/operator-only. `decodex serve` exposes the human-facing operator console from the canonical HTTP `GET /` and `GET /dashboard` routes, serves only the necessary dashboard assets, `GET /livez` liveness probe, and local account-control API over HTTP, and delivers published snapshots, current-lane activity, and dashboard control acknowledgements through the local `GET /dashboard/control` WebSocket upgrade. +- The MCP stdio transport is separate from the operator HTTP/WebSocket transport. It is + local-client integration for resource readback only; it must not replace the app-server + dynamic tool bridge or become a lane-control mutation path before a later authority + design delegates through existing lane-control guards. - `GET /livez` is only a process- and listener-level liveness probe. It must not claim control-plane tick freshness or forward progress by itself. - The dashboard must not depend on a separate HTTP snapshot or readiness endpoint; snapshot freshness belongs to the WebSocket-delivered snapshot payload and the browser connection state. - Reconciliation must mark locally active run attempts as `interrupted` when their