grex_cli/cli/verbs/
serve.rs

1//! `grex serve` — launch the MCP stdio server (feat-m7-1 stage 8).
2//!
3//! Wires the rmcp framework (built in `grex-mcp`) to the actual stdio
4//! transport on top of a Tokio runtime. The runtime is constructed
5//! per-invocation: `cli::run` is sync, but rmcp's stdio loop is async,
6//! so we bridge with a current-thread runtime started on demand.
7//!
8//! ## Tracing discipline
9//!
10//! `grex-mcp::GrexMcpServer::run` reinstalls a stderr-pinned subscriber
11//! using `EnvFilter::try_from_default_env()` with `info` fallback. To
12//! silence rmcp's "Service initialized as server" INFO line in
13//! production, `main.rs` defaults `RUST_LOG` to `grex=info,rmcp=warn`
14//! when no env var is present (Stage 5 wiring note #7). Tests can
15//! override with `RUST_LOG=…`.
16
17use crate::cli::args::{GlobalFlags, ServeArgs};
18use anyhow::{Context, Result};
19use grex_core::manifest::{ensure_event_log_migrated, find_workspace_root};
20use grex_core::{Registry, Scheduler};
21use grex_mcp::{GrexMcpServer, ServerState};
22use tokio_util::sync::CancellationToken;
23
24// `_cancel` is intentionally unused here. The verb-level CancellationToken from
25// `cli::run` is plumbed for future global-shutdown wiring (Stage 8+). Per-request
26// cancellation is handled by rmcp's internal local_ct_pool (see grex-mcp Stage 7
27// commit and lib.rs comment block above serve()).
28pub fn run(args: ServeArgs, _global: &GlobalFlags, _cancel: &CancellationToken) -> Result<()> {
29    let workspace = match args.workspace {
30        Some(p) => p,
31        None => {
32            let cwd = std::env::current_dir().context("resolve cwd for --workspace default")?;
33            // Walk up from cwd to find a workspace marker — fixes the
34            // v1.x cwd-relative bug for `grex serve` invoked from a
35            // subdir of the workspace.
36            find_workspace_root(&cwd)
37        }
38    };
39    let manifest_path = match args.manifest {
40        Some(p) => p,
41        None => ensure_event_log_migrated(&workspace).context("migrate v1.x event log")?,
42    };
43    let parallel = resolve_parallel(args.parallel);
44
45    let scheduler = Scheduler::new(parallel);
46    let registry = Registry::default();
47    let state = ServerState::new(scheduler, registry, manifest_path, workspace);
48    let server = GrexMcpServer::new(state);
49
50    // Bridge sync `cli::run` → async rmcp loop. A fresh single-thread
51    // runtime is sufficient: the server has no other in-process work,
52    // and rmcp drives its own request fan-out via tokio::spawn.
53    let rt = tokio::runtime::Builder::new_current_thread()
54        .enable_all()
55        .build()
56        .context("build tokio runtime for grex serve")?;
57
58    rt.block_on(async move {
59        let transport = rmcp::transport::stdio();
60        // Pre-`run` info line so stderr has at least one tracing event
61        // even with the default `grex=info,rmcp=warn` filter that hides
62        // rmcp's "Service initialized as server" log. Useful for ops
63        // visibility (PID, parallel cap) and asserted by 8.T3.
64        tracing::info!(
65            target: "grex",
66            parallel,
67            "grex serve: MCP stdio transport ready",
68        );
69        server.run(transport).await.context("grex-mcp server exited with error")
70    })
71}
72
73/// Resolve the `--parallel` flag to a concrete worker count, falling
74/// back to `std::thread::available_parallelism()` when unset and to `1`
75/// as the conservative floor when even that fails (uncommon — reserved
76/// VMs / sandboxed CI). Matches the harness contract in
77/// `.omne/cfg/concurrency.md`.
78fn resolve_parallel(opt: Option<u32>) -> usize {
79    match opt {
80        Some(n) => n as usize,
81        None => std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1),
82    }
83}
grex_cli/cli/verbs/serve.rs

grex_cli/cli/verbs/
serve.rs
