lean_ctx/http_server/
mod.rs

1use std::net::SocketAddr;
2use std::path::PathBuf;
3use std::sync::Arc;
4
5use anyhow::{anyhow, Context, Result};
6use axum::{
7    extract::Json,
8    extract::Query,
9    extract::State,
10    http::{header, Request, StatusCode},
11    middleware::{self, Next},
12    response::sse::{Event as SseEvent, KeepAlive, Sse},
13    response::{IntoResponse, Response},
14    routing::get,
15    Router,
16};
17use futures::Stream;
18use rmcp::transport::{StreamableHttpServerConfig, StreamableHttpService};
19use serde::Deserialize;
20use serde_json::Value;
21use tokio::sync::broadcast;
22use tokio::time::{Duration, Instant};
23
24use crate::core::context_os::ContextOsMetrics;
25use crate::engine::ContextEngine;
26use crate::tools::LeanCtxServer;
27
28pub mod context_views;
29
30#[cfg(feature = "team-server")]
31pub mod team;
32
33/// Wrapper stream that calls `record_sse_disconnect` on drop.
34use std::pin::Pin;
35
36pub(crate) struct SseDisconnectGuard<I> {
37    pub(crate) inner: Pin<Box<dyn Stream<Item = I> + Send>>,
38    pub(crate) metrics: Arc<ContextOsMetrics>,
39}
40
41impl<I> Stream for SseDisconnectGuard<I> {
42    type Item = I;
43
44    fn poll_next(
45        mut self: Pin<&mut Self>,
46        cx: &mut std::task::Context<'_>,
47    ) -> std::task::Poll<Option<Self::Item>> {
48        self.inner.as_mut().poll_next(cx)
49    }
50}
51
52impl<I> Drop for SseDisconnectGuard<I> {
53    fn drop(&mut self) {
54        self.metrics.record_sse_disconnect();
55    }
56}
57
58#[derive(Clone, Debug)]
59pub struct HttpServerConfig {
60    pub host: String,
61    pub port: u16,
62    pub project_root: PathBuf,
63    pub auth_token: Option<String>,
64    pub stateful_mode: bool,
65    pub json_response: bool,
66    pub disable_host_check: bool,
67    pub allowed_hosts: Vec<String>,
68    pub max_body_bytes: usize,
69    pub max_concurrency: usize,
70    pub max_rps: u32,
71    pub rate_burst: u32,
72    pub request_timeout_ms: u64,
73}
74
75impl Default for HttpServerConfig {
76    fn default() -> Self {
77        let project_root = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
78        Self {
79            host: "127.0.0.1".to_string(),
80            port: 8080,
81            project_root,
82            auth_token: None,
83            stateful_mode: false,
84            json_response: true,
85            disable_host_check: false,
86            allowed_hosts: Vec::new(),
87            max_body_bytes: 2 * 1024 * 1024,
88            max_concurrency: 32,
89            max_rps: 50,
90            rate_burst: 100,
91            request_timeout_ms: 30_000,
92        }
93    }
94}
95
96impl HttpServerConfig {
97    pub fn validate(&self) -> Result<()> {
98        let host = self.host.trim().to_lowercase();
99        let is_loopback = host == "127.0.0.1" || host == "localhost" || host == "::1";
100        if !is_loopback && self.auth_token.as_deref().unwrap_or("").is_empty() {
101            return Err(anyhow!(
102                "Refusing to bind to host='{host}' without auth. Provide --auth-token (or bind to 127.0.0.1)."
103            ));
104        }
105        Ok(())
106    }
107
108    fn mcp_http_config(&self) -> StreamableHttpServerConfig {
109        let mut cfg = StreamableHttpServerConfig::default()
110            .with_stateful_mode(self.stateful_mode)
111            .with_json_response(self.json_response);
112
113        if self.disable_host_check {
114            cfg = cfg.disable_allowed_hosts();
115            return cfg;
116        }
117
118        if !self.allowed_hosts.is_empty() {
119            cfg = cfg.with_allowed_hosts(self.allowed_hosts.clone());
120            return cfg;
121        }
122
123        // Keep rmcp's secure loopback defaults; also allow the configured host (if it's loopback).
124        let host = self.host.trim();
125        if host == "127.0.0.1" || host == "localhost" || host == "::1" {
126            cfg.allowed_hosts.push(host.to_string());
127        }
128
129        cfg
130    }
131}
132
133#[derive(Clone)]
134struct AppState {
135    token: Option<String>,
136    concurrency: Arc<tokio::sync::Semaphore>,
137    rate: Arc<RateLimiter>,
138    project_root: String,
139    timeout: Duration,
140}
141
142#[derive(Debug)]
143struct RateLimiter {
144    max_rps: f64,
145    burst: f64,
146    state: tokio::sync::Mutex<RateState>,
147}
148
149#[derive(Debug, Clone, Copy)]
150struct RateState {
151    tokens: f64,
152    last: Instant,
153}
154
155impl RateLimiter {
156    fn new(max_rps: u32, burst: u32) -> Self {
157        let now = Instant::now();
158        Self {
159            max_rps: (max_rps.max(1)) as f64,
160            burst: (burst.max(1)) as f64,
161            state: tokio::sync::Mutex::new(RateState {
162                tokens: (burst.max(1)) as f64,
163                last: now,
164            }),
165        }
166    }
167
168    async fn allow(&self) -> bool {
169        let mut s = self.state.lock().await;
170        let now = Instant::now();
171        let elapsed = now.saturating_duration_since(s.last);
172        let refill = elapsed.as_secs_f64() * self.max_rps;
173        s.tokens = (s.tokens + refill).min(self.burst);
174        s.last = now;
175        if s.tokens >= 1.0 {
176            s.tokens -= 1.0;
177            true
178        } else {
179            false
180        }
181    }
182}
183
184async fn auth_middleware(
185    State(state): State<AppState>,
186    req: Request<axum::body::Body>,
187    next: Next,
188) -> Response {
189    if state.token.is_none() {
190        return next.run(req).await;
191    }
192
193    if req.uri().path() == "/health" {
194        return next.run(req).await;
195    }
196
197    let expected = state.token.as_deref().unwrap_or("");
198    let Some(h) = req.headers().get(header::AUTHORIZATION) else {
199        return StatusCode::UNAUTHORIZED.into_response();
200    };
201    let Ok(s) = h.to_str() else {
202        return StatusCode::UNAUTHORIZED.into_response();
203    };
204    let Some(token) = s
205        .strip_prefix("Bearer ")
206        .or_else(|| s.strip_prefix("bearer "))
207    else {
208        return StatusCode::UNAUTHORIZED.into_response();
209    };
210    if !constant_time_eq(token.as_bytes(), expected.as_bytes()) {
211        return StatusCode::UNAUTHORIZED.into_response();
212    }
213
214    next.run(req).await
215}
216
217fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
218    if a.len() != b.len() {
219        return false;
220    }
221    a.iter()
222        .zip(b.iter())
223        .fold(0u8, |acc, (x, y)| acc | (x ^ y))
224        == 0
225}
226
227async fn rate_limit_middleware(
228    State(state): State<AppState>,
229    req: Request<axum::body::Body>,
230    next: Next,
231) -> Response {
232    if req.uri().path() == "/health" {
233        return next.run(req).await;
234    }
235    if !state.rate.allow().await {
236        return StatusCode::TOO_MANY_REQUESTS.into_response();
237    }
238    next.run(req).await
239}
240
241async fn concurrency_middleware(
242    State(state): State<AppState>,
243    req: Request<axum::body::Body>,
244    next: Next,
245) -> Response {
246    if req.uri().path() == "/health" {
247        return next.run(req).await;
248    }
249    let Ok(permit) = state.concurrency.clone().try_acquire_owned() else {
250        return StatusCode::TOO_MANY_REQUESTS.into_response();
251    };
252    let resp = next.run(req).await;
253    drop(permit);
254    resp
255}
256
257async fn health() -> impl IntoResponse {
258    (StatusCode::OK, "ok\n")
259}
260
261#[derive(Debug, Deserialize)]
262#[serde(rename_all = "camelCase")]
263struct ToolCallBody {
264    name: String,
265    #[serde(default)]
266    arguments: Option<Value>,
267    #[serde(default)]
268    workspace_id: Option<String>,
269    #[serde(default)]
270    channel_id: Option<String>,
271}
272
273#[derive(Debug, Deserialize)]
274#[serde(rename_all = "camelCase")]
275struct EventsQuery {
276    #[serde(default)]
277    workspace_id: Option<String>,
278    #[serde(default)]
279    channel_id: Option<String>,
280    #[serde(default)]
281    since: Option<i64>,
282    #[serde(default)]
283    limit: Option<usize>,
284}
285
286async fn v1_manifest(State(state): State<AppState>) -> impl IntoResponse {
287    let _ = state;
288    let v = crate::core::mcp_manifest::manifest_value();
289    (StatusCode::OK, Json(v))
290}
291
292#[derive(Debug, Deserialize)]
293#[serde(rename_all = "camelCase")]
294struct ToolsQuery {
295    #[serde(default)]
296    offset: Option<usize>,
297    #[serde(default)]
298    limit: Option<usize>,
299}
300
301async fn v1_tools(State(state): State<AppState>, Query(q): Query<ToolsQuery>) -> impl IntoResponse {
302    let _ = state;
303    let v = crate::core::mcp_manifest::manifest_value();
304    let tools = v
305        .get("tools")
306        .and_then(|t| t.get("granular"))
307        .cloned()
308        .unwrap_or(Value::Array(vec![]));
309
310    let all = tools.as_array().cloned().unwrap_or_default();
311    let total = all.len();
312    let offset = q.offset.unwrap_or(0).min(total);
313    let limit = q.limit.unwrap_or(200).min(500);
314    let page = all.into_iter().skip(offset).take(limit).collect::<Vec<_>>();
315
316    (
317        StatusCode::OK,
318        Json(serde_json::json!({
319            "tools": page,
320            "total": total,
321            "offset": offset,
322            "limit": limit,
323        })),
324    )
325}
326
327async fn v1_tool_call(
328    State(state): State<AppState>,
329    Json(body): Json<ToolCallBody>,
330) -> impl IntoResponse {
331    let ws = body.workspace_id.as_deref().unwrap_or("default");
332    let ch = body.channel_id.as_deref().unwrap_or("default");
333    let server = LeanCtxServer::new_shared_with_context(&state.project_root, ws, ch);
334    let engine = ContextEngine::from_server(server);
335    match tokio::time::timeout(
336        state.timeout,
337        engine.call_tool_value(&body.name, body.arguments),
338    )
339    .await
340    {
341        Ok(Ok(v)) => (StatusCode::OK, Json(serde_json::json!({ "result": v }))).into_response(),
342        Ok(Err(e)) => (
343            StatusCode::BAD_REQUEST,
344            Json(serde_json::json!({ "error": e.to_string() })),
345        )
346            .into_response(),
347        Err(_) => (
348            StatusCode::GATEWAY_TIMEOUT,
349            Json(serde_json::json!({ "error": "request_timeout" })),
350        )
351            .into_response(),
352    }
353}
354
355async fn v1_events(
356    State(_state): State<AppState>,
357    Query(q): Query<EventsQuery>,
358) -> Sse<impl Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
359    use crate::core::context_os::{redact_event_payload, ContextEventV1, RedactionLevel};
360
361    let ws = q.workspace_id.unwrap_or_else(|| "default".to_string());
362    let ch = q.channel_id.unwrap_or_else(|| "default".to_string());
363    let since = q.since.unwrap_or(0);
364    let limit = q.limit.unwrap_or(200).min(1000);
365    let redaction = RedactionLevel::RefsOnly;
366
367    let rt = crate::core::context_os::runtime();
368    let replay = rt.bus.read(&ws, &ch, since, limit);
369    let rx = rt.bus.subscribe();
370    rt.metrics.record_sse_connect();
371    rt.metrics.record_events_replayed(replay.len() as u64);
372    rt.metrics.record_workspace_active(&ws);
373
374    let bus = rt.bus.clone();
375    let metrics = rt.metrics.clone();
376    let pending: std::collections::VecDeque<ContextEventV1> = replay.into();
377
378    let stream = futures::stream::unfold(
379        (
380            pending,
381            rx,
382            ws.clone(),
383            ch.clone(),
384            since,
385            redaction,
386            bus,
387            metrics,
388        ),
389        |(mut pending, mut rx, ws, ch, mut last_id, redaction, bus, metrics)| async move {
390            if let Some(mut ev) = pending.pop_front() {
391                last_id = ev.id;
392                redact_event_payload(&mut ev, redaction);
393                let data = serde_json::to_string(&ev).unwrap_or_else(|_| "{}".to_string());
394                let evt = SseEvent::default()
395                    .id(ev.id.to_string())
396                    .event(ev.kind)
397                    .data(data);
398                return Some((
399                    Ok(evt),
400                    (pending, rx, ws, ch, last_id, redaction, bus, metrics),
401                ));
402            }
403
404            loop {
405                match rx.recv().await {
406                    Ok(mut ev) => {
407                        if ev.workspace_id == ws && ev.channel_id == ch && ev.id > last_id {
408                            last_id = ev.id;
409                            redact_event_payload(&mut ev, redaction);
410                            let data =
411                                serde_json::to_string(&ev).unwrap_or_else(|_| "{}".to_string());
412                            let evt = SseEvent::default()
413                                .id(ev.id.to_string())
414                                .event(ev.kind)
415                                .data(data);
416                            return Some((
417                                Ok(evt),
418                                (pending, rx, ws, ch, last_id, redaction, bus, metrics),
419                            ));
420                        }
421                    }
422                    Err(broadcast::error::RecvError::Closed) => return None,
423                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
424                        let missed = bus.read(&ws, &ch, last_id, skipped as usize);
425                        metrics.record_events_replayed(missed.len() as u64);
426                        for ev in missed {
427                            last_id = last_id.max(ev.id);
428                            pending.push_back(ev);
429                        }
430                    }
431                }
432            }
433        },
434    );
435
436    let metrics_ref = rt.metrics.clone();
437    let guarded = SseDisconnectGuard {
438        inner: Box::pin(stream),
439        metrics: metrics_ref,
440    };
441
442    Sse::new(guarded).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
443}
444
445async fn v1_metrics(State(_state): State<AppState>) -> impl IntoResponse {
446    let rt = crate::core::context_os::runtime();
447    let snap = rt.metrics.snapshot();
448    (
449        StatusCode::OK,
450        Json(serde_json::to_value(snap).unwrap_or_default()),
451    )
452}
453
454pub async fn serve(cfg: HttpServerConfig) -> Result<()> {
455    cfg.validate()?;
456
457    let addr: SocketAddr = format!("{}:{}", cfg.host, cfg.port)
458        .parse()
459        .context("invalid host/port")?;
460
461    let project_root = cfg.project_root.to_string_lossy().to_string();
462    // IMPORTANT: Create a fresh server per MCP session in *shared* mode.
463    // This avoids per-client state clobbering while still sharing the Context OS store.
464    let service_project_root = project_root.clone();
465    let service_factory = move || -> Result<LeanCtxServer, std::io::Error> {
466        Ok(LeanCtxServer::new_shared_with_context(
467            &service_project_root,
468            "default",
469            "default",
470        ))
471    };
472    let mcp_http = StreamableHttpService::new(
473        service_factory,
474        Arc::new(
475            rmcp::transport::streamable_http_server::session::local::LocalSessionManager::default(),
476        ),
477        cfg.mcp_http_config(),
478    );
479
480    let state = AppState {
481        token: cfg.auth_token.clone().filter(|t| !t.is_empty()),
482        concurrency: Arc::new(tokio::sync::Semaphore::new(cfg.max_concurrency.max(1))),
483        rate: Arc::new(RateLimiter::new(cfg.max_rps, cfg.rate_burst)),
484        project_root: project_root.clone(),
485        timeout: Duration::from_millis(cfg.request_timeout_ms.max(1)),
486    };
487
488    let app = Router::new()
489        .route("/health", get(health))
490        .route("/v1/manifest", get(v1_manifest))
491        .route("/v1/tools", get(v1_tools))
492        .route("/v1/tools/call", axum::routing::post(v1_tool_call))
493        .route("/v1/events", get(v1_events))
494        .route(
495            "/v1/context/summary",
496            get(context_views::v1_context_summary),
497        )
498        .route("/v1/events/search", get(context_views::v1_events_search))
499        .route("/v1/events/lineage", get(context_views::v1_event_lineage))
500        .route("/v1/metrics", get(v1_metrics))
501        .fallback_service(mcp_http)
502        .layer(axum::extract::DefaultBodyLimit::max(cfg.max_body_bytes))
503        .layer(middleware::from_fn_with_state(
504            state.clone(),
505            rate_limit_middleware,
506        ))
507        .layer(middleware::from_fn_with_state(
508            state.clone(),
509            concurrency_middleware,
510        ))
511        .layer(middleware::from_fn_with_state(
512            state.clone(),
513            auth_middleware,
514        ))
515        .with_state(state);
516
517    let listener = tokio::net::TcpListener::bind(addr)
518        .await
519        .with_context(|| format!("bind {addr}"))?;
520
521    tracing::info!(
522        "lean-ctx Streamable HTTP server listening on http://{addr} (project_root={})",
523        cfg.project_root.display()
524    );
525
526    axum::serve(listener, app)
527        .with_graceful_shutdown(async move {
528            let _ = tokio::signal::ctrl_c().await;
529        })
530        .await
531        .context("http server")?;
532    Ok(())
533}
534
535#[cfg(unix)]
536pub async fn serve_uds(cfg: HttpServerConfig, socket_path: PathBuf) -> Result<()> {
537    cfg.validate()?;
538
539    if socket_path.exists() {
540        std::fs::remove_file(&socket_path)
541            .with_context(|| format!("remove stale socket {}", socket_path.display()))?;
542    }
543
544    let project_root = cfg.project_root.to_string_lossy().to_string();
545    let service_project_root = project_root.clone();
546    let service_factory = move || -> Result<LeanCtxServer, std::io::Error> {
547        Ok(LeanCtxServer::new_shared_with_context(
548            &service_project_root,
549            "default",
550            "default",
551        ))
552    };
553    let mcp_http = StreamableHttpService::new(
554        service_factory,
555        Arc::new(
556            rmcp::transport::streamable_http_server::session::local::LocalSessionManager::default(),
557        ),
558        cfg.mcp_http_config(),
559    );
560
561    let state = AppState {
562        token: cfg.auth_token.clone().filter(|t| !t.is_empty()),
563        concurrency: Arc::new(tokio::sync::Semaphore::new(cfg.max_concurrency.max(1))),
564        rate: Arc::new(RateLimiter::new(cfg.max_rps, cfg.rate_burst)),
565        project_root: project_root.clone(),
566        timeout: Duration::from_millis(cfg.request_timeout_ms.max(1)),
567    };
568
569    let app = Router::new()
570        .route("/health", get(health))
571        .route("/v1/manifest", get(v1_manifest))
572        .route("/v1/tools", get(v1_tools))
573        .route("/v1/tools/call", axum::routing::post(v1_tool_call))
574        .route("/v1/events", get(v1_events))
575        .route(
576            "/v1/context/summary",
577            get(context_views::v1_context_summary),
578        )
579        .route("/v1/events/search", get(context_views::v1_events_search))
580        .route("/v1/events/lineage", get(context_views::v1_event_lineage))
581        .route("/v1/metrics", get(v1_metrics))
582        .fallback_service(mcp_http)
583        .layer(axum::extract::DefaultBodyLimit::max(cfg.max_body_bytes))
584        .layer(middleware::from_fn_with_state(
585            state.clone(),
586            rate_limit_middleware,
587        ))
588        .layer(middleware::from_fn_with_state(
589            state.clone(),
590            concurrency_middleware,
591        ))
592        .layer(middleware::from_fn_with_state(
593            state.clone(),
594            auth_middleware,
595        ))
596        .with_state(state);
597
598    let listener = tokio::net::UnixListener::bind(&socket_path)
599        .with_context(|| format!("bind UDS {}", socket_path.display()))?;
600
601    tracing::info!(
602        "lean-ctx daemon listening on {} (project_root={})",
603        socket_path.display(),
604        cfg.project_root.display()
605    );
606
607    axum::serve(listener, app.into_make_service())
608        .with_graceful_shutdown(async move {
609            let _ = tokio::signal::ctrl_c().await;
610        })
611        .await
612        .context("uds server")?;
613    Ok(())
614}
615
616#[cfg(test)]
617mod tests {
618    use super::*;
619    use axum::body::Body;
620    use axum::http::Request;
621    use futures::StreamExt;
622    use rmcp::transport::{StreamableHttpServerConfig, StreamableHttpService};
623    use serde_json::json;
624    use tower::ServiceExt;
625
626    async fn read_first_sse_message(body: Body) -> String {
627        let mut stream = body.into_data_stream();
628        let mut buf: Vec<u8> = Vec::new();
629        for _ in 0..32 {
630            let next = tokio::time::timeout(Duration::from_secs(2), stream.next()).await;
631            let Ok(Some(Ok(bytes))) = next else {
632                break;
633            };
634            buf.extend_from_slice(&bytes);
635            if buf.windows(2).any(|w| w == b"\n\n") {
636                break;
637            }
638        }
639        String::from_utf8_lossy(&buf).to_string()
640    }
641
642    #[tokio::test]
643    async fn auth_token_blocks_requests_without_bearer_header() {
644        let dir = tempfile::tempdir().expect("tempdir");
645        let root_str = dir.path().to_string_lossy().to_string();
646        let service_project_root = root_str.clone();
647        let service_factory = move || -> Result<LeanCtxServer, std::io::Error> {
648            Ok(LeanCtxServer::new_shared_with_context(
649                &service_project_root,
650                "default",
651                "default",
652            ))
653        };
654        let cfg = StreamableHttpServerConfig::default()
655            .with_stateful_mode(false)
656            .with_json_response(true);
657
658        let mcp_http = StreamableHttpService::new(
659            service_factory,
660            Arc::new(
661                rmcp::transport::streamable_http_server::session::local::LocalSessionManager::default(),
662            ),
663            cfg,
664        );
665
666        let state = AppState {
667            token: Some("secret".to_string()),
668            concurrency: Arc::new(tokio::sync::Semaphore::new(4)),
669            rate: Arc::new(RateLimiter::new(50, 100)),
670            project_root: root_str.clone(),
671            timeout: Duration::from_millis(30_000),
672        };
673
674        let app = Router::new()
675            .fallback_service(mcp_http)
676            .layer(middleware::from_fn_with_state(
677                state.clone(),
678                auth_middleware,
679            ))
680            .with_state(state);
681
682        let body = json!({
683            "jsonrpc": "2.0",
684            "id": 1,
685            "method": "tools/list",
686            "params": {}
687        })
688        .to_string();
689
690        let req = Request::builder()
691            .method("POST")
692            .uri("/")
693            .header("Host", "localhost")
694            .header("Accept", "application/json, text/event-stream")
695            .header("Content-Type", "application/json")
696            .body(Body::from(body))
697            .expect("request");
698
699        let resp = app.clone().oneshot(req).await.expect("resp");
700        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
701    }
702
703    #[tokio::test]
704    async fn mcp_service_factory_isolates_per_client_state() {
705        let dir = tempfile::tempdir().expect("tempdir");
706        let root_str = dir.path().to_string_lossy().to_string();
707
708        // Mirrors the serve() setup: service_factory must create a fresh server per MCP session.
709        let service_project_root = root_str.clone();
710        let service_factory = move || -> Result<LeanCtxServer, std::convert::Infallible> {
711            Ok(LeanCtxServer::new_shared_with_context(
712                &service_project_root,
713                "default",
714                "default",
715            ))
716        };
717
718        let s1 = service_factory().expect("server 1");
719        let s2 = service_factory().expect("server 2");
720
721        // If the two servers accidentally share the same Arc-backed fields, these writes would
722        // clobber each other. This test stays independent of rmcp's InitializeRequestParams API.
723        *s1.client_name.write().await = "client-a".to_string();
724        *s2.client_name.write().await = "client-b".to_string();
725
726        let a = s1.client_name.read().await.clone();
727        let b = s2.client_name.read().await.clone();
728        assert_eq!(a, "client-a");
729        assert_eq!(b, "client-b");
730    }
731
732    #[tokio::test]
733    async fn rate_limit_returns_429_when_exhausted() {
734        let state = AppState {
735            token: None,
736            concurrency: Arc::new(tokio::sync::Semaphore::new(16)),
737            rate: Arc::new(RateLimiter::new(1, 1)),
738            project_root: ".".to_string(),
739            timeout: Duration::from_millis(30_000),
740        };
741
742        let app = Router::new()
743            .route("/limited", get(|| async { (StatusCode::OK, "ok\n") }))
744            .layer(middleware::from_fn_with_state(
745                state.clone(),
746                rate_limit_middleware,
747            ))
748            .with_state(state);
749
750        let req1 = Request::builder()
751            .method("GET")
752            .uri("/limited")
753            .header("Host", "localhost")
754            .body(Body::empty())
755            .expect("req1");
756        let resp1 = app.clone().oneshot(req1).await.expect("resp1");
757        assert_eq!(resp1.status(), StatusCode::OK);
758
759        let req2 = Request::builder()
760            .method("GET")
761            .uri("/limited")
762            .header("Host", "localhost")
763            .body(Body::empty())
764            .expect("req2");
765        let resp2 = app.clone().oneshot(req2).await.expect("resp2");
766        assert_eq!(resp2.status(), StatusCode::TOO_MANY_REQUESTS);
767    }
768
769    #[tokio::test]
770    async fn events_endpoint_replays_tool_call_event() {
771        let dir = tempfile::tempdir().expect("tempdir");
772        std::fs::create_dir_all(dir.path().join(".git")).expect("git marker");
773        std::fs::write(dir.path().join("a.txt"), "ok").expect("file");
774        let root_str = dir.path().to_string_lossy().to_string();
775
776        let state = AppState {
777            token: None,
778            concurrency: Arc::new(tokio::sync::Semaphore::new(16)),
779            rate: Arc::new(RateLimiter::new(50, 100)),
780            project_root: root_str.clone(),
781            timeout: Duration::from_millis(30_000),
782        };
783
784        let app = Router::new()
785            .route("/v1/tools/call", axum::routing::post(v1_tool_call))
786            .route("/v1/events", get(v1_events))
787            .with_state(state);
788
789        let body = json!({
790            "name": "ctx_session",
791            "arguments": { "action": "status" },
792            "workspaceId": "ws1",
793            "channelId": "ch1"
794        })
795        .to_string();
796        let req = Request::builder()
797            .method("POST")
798            .uri("/v1/tools/call")
799            .header("Host", "localhost")
800            .header("Content-Type", "application/json")
801            .body(Body::from(body))
802            .expect("req");
803        let resp = app.clone().oneshot(req).await.expect("call");
804        assert_eq!(resp.status(), StatusCode::OK);
805
806        // Subscribe with replay semantics; read the first SSE message.
807        let req = Request::builder()
808            .method("GET")
809            .uri("/v1/events?workspaceId=ws1&channelId=ch1&since=0&limit=1")
810            .header("Host", "localhost")
811            .header("Accept", "text/event-stream")
812            .body(Body::empty())
813            .expect("req");
814        let resp = app.clone().oneshot(req).await.expect("events");
815        assert_eq!(resp.status(), StatusCode::OK);
816
817        let msg = read_first_sse_message(resp.into_body()).await;
818        assert!(msg.contains("event: tool_call_recorded"), "msg={msg:?}");
819        assert!(msg.contains("\"workspaceId\":\"ws1\""), "msg={msg:?}");
820        assert!(msg.contains("\"channelId\":\"ch1\""), "msg={msg:?}");
821    }
822}
lean_ctx/http_server/mod.rs

lean_ctx/http_server/
mod.rs
