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::{Registry, Scheduler};
20use grex_mcp::{GrexMcpServer, ServerState};
21use tokio_util::sync::CancellationToken;
22
23// `_cancel` is intentionally unused here. The verb-level CancellationToken from
24// `cli::run` is plumbed for future global-shutdown wiring (Stage 8+). Per-request
25// cancellation is handled by rmcp's internal local_ct_pool (see grex-mcp Stage 7
26// commit and lib.rs comment block above serve()).
27pub fn run(args: ServeArgs, _global: &GlobalFlags, _cancel: &CancellationToken) -> Result<()> {
28    let workspace = match args.workspace {
29        Some(p) => p,
30        None => std::env::current_dir().context("resolve cwd for --workspace default")?,
31    };
32    let manifest_path = match args.manifest {
33        Some(p) => p,
34        None => workspace.join("grex.jsonl"),
35    };
36    let parallel = resolve_parallel(args.parallel);
37
38    let scheduler = Scheduler::new(parallel);
39    let registry = Registry::default();
40    let state = ServerState::new(scheduler, registry, manifest_path, workspace);
41    let server = GrexMcpServer::new(state);
42
43    // Bridge sync `cli::run` → async rmcp loop. A fresh single-thread
44    // runtime is sufficient: the server has no other in-process work,
45    // and rmcp drives its own request fan-out via tokio::spawn.
46    let rt = tokio::runtime::Builder::new_current_thread()
47        .enable_all()
48        .build()
49        .context("build tokio runtime for grex serve")?;
50
51    rt.block_on(async move {
52        let transport = rmcp::transport::stdio();
53        // Pre-`run` info line so stderr has at least one tracing event
54        // even with the default `grex=info,rmcp=warn` filter that hides
55        // rmcp's "Service initialized as server" log. Useful for ops
56        // visibility (PID, parallel cap) and asserted by 8.T3.
57        tracing::info!(
58            target: "grex",
59            parallel,
60            "grex serve: MCP stdio transport ready",
61        );
62        server.run(transport).await.context("grex-mcp server exited with error")
63    })
64}
65
66/// Resolve the `--parallel` flag to a concrete worker count, falling
67/// back to `std::thread::available_parallelism()` when unset and to `1`
68/// as the conservative floor when even that fails (uncommon — reserved
69/// VMs / sandboxed CI). Matches the harness contract in
70/// `.omne/cfg/concurrency.md`.
71fn resolve_parallel(opt: Option<u32>) -> usize {
72    match opt {
73        Some(n) => n as usize,
74        None => std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1),
75    }
76}
grex_cli/cli/verbs/serve.rs

grex_cli/cli/verbs/
serve.rs
