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(&ws, &ch);
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) if ev.id > last_id => {
407                        last_id = ev.id;
408                        redact_event_payload(&mut ev, redaction);
409                        let data = serde_json::to_string(&ev).unwrap_or_else(|_| "{}".to_string());
410                        let evt = SseEvent::default()
411                            .id(ev.id.to_string())
412                            .event(ev.kind)
413                            .data(data);
414                        return Some((
415                            Ok(evt),
416                            (pending, rx, ws, ch, last_id, redaction, bus, metrics),
417                        ));
418                    }
419                    Ok(_) => {}
420                    Err(broadcast::error::RecvError::Closed) => return None,
421                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
422                        let missed = bus.read(&ws, &ch, last_id, skipped as usize);
423                        metrics.record_events_replayed(missed.len() as u64);
424                        for ev in missed {
425                            last_id = last_id.max(ev.id);
426                            pending.push_back(ev);
427                        }
428                    }
429                }
430            }
431        },
432    );
433
434    let metrics_ref = rt.metrics.clone();
435    let guarded = SseDisconnectGuard {
436        inner: Box::pin(stream),
437        metrics: metrics_ref,
438    };
439
440    Sse::new(guarded).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
441}
442
443async fn v1_metrics(State(_state): State<AppState>) -> impl IntoResponse {
444    let rt = crate::core::context_os::runtime();
445    let snap = rt.metrics.snapshot();
446    (
447        StatusCode::OK,
448        Json(serde_json::to_value(snap).unwrap_or_default()),
449    )
450}
451
452pub async fn serve(cfg: HttpServerConfig) -> Result<()> {
453    cfg.validate()?;
454
455    let addr: SocketAddr = format!("{}:{}", cfg.host, cfg.port)
456        .parse()
457        .context("invalid host/port")?;
458
459    let project_root = cfg.project_root.to_string_lossy().to_string();
460    // IMPORTANT: Create a fresh server per MCP session in *shared* mode.
461    // This avoids per-client state clobbering while still sharing the Context OS store.
462    let service_project_root = project_root.clone();
463    let service_factory = move || -> Result<LeanCtxServer, std::io::Error> {
464        Ok(LeanCtxServer::new_shared_with_context(
465            &service_project_root,
466            "default",
467            "default",
468        ))
469    };
470    let mcp_http = StreamableHttpService::new(
471        service_factory,
472        Arc::new(
473            rmcp::transport::streamable_http_server::session::local::LocalSessionManager::default(),
474        ),
475        cfg.mcp_http_config(),
476    );
477
478    let state = AppState {
479        token: cfg.auth_token.clone().filter(|t| !t.is_empty()),
480        concurrency: Arc::new(tokio::sync::Semaphore::new(cfg.max_concurrency.max(1))),
481        rate: Arc::new(RateLimiter::new(cfg.max_rps, cfg.rate_burst)),
482        project_root: project_root.clone(),
483        timeout: Duration::from_millis(cfg.request_timeout_ms.max(1)),
484    };
485
486    let app = Router::new()
487        .route("/health", get(health))
488        .route("/v1/manifest", get(v1_manifest))
489        .route("/v1/tools", get(v1_tools))
490        .route("/v1/tools/call", axum::routing::post(v1_tool_call))
491        .route("/v1/events", get(v1_events))
492        .route(
493            "/v1/context/summary",
494            get(context_views::v1_context_summary),
495        )
496        .route("/v1/events/search", get(context_views::v1_events_search))
497        .route("/v1/events/lineage", get(context_views::v1_event_lineage))
498        .route("/v1/metrics", get(v1_metrics))
499        .fallback_service(mcp_http)
500        .layer(axum::extract::DefaultBodyLimit::max(cfg.max_body_bytes))
501        .layer(middleware::from_fn_with_state(
502            state.clone(),
503            rate_limit_middleware,
504        ))
505        .layer(middleware::from_fn_with_state(
506            state.clone(),
507            concurrency_middleware,
508        ))
509        .layer(middleware::from_fn_with_state(
510            state.clone(),
511            auth_middleware,
512        ))
513        .with_state(state);
514
515    let listener = tokio::net::TcpListener::bind(addr)
516        .await
517        .with_context(|| format!("bind {addr}"))?;
518
519    tracing::info!(
520        "lean-ctx Streamable HTTP server listening on http://{addr} (project_root={})",
521        cfg.project_root.display()
522    );
523
524    axum::serve(listener, app)
525        .with_graceful_shutdown(async move {
526            let _ = tokio::signal::ctrl_c().await;
527        })
528        .await
529        .context("http server")?;
530    Ok(())
531}
532
533#[cfg(unix)]
534pub async fn serve_uds(cfg: HttpServerConfig, socket_path: PathBuf) -> Result<()> {
535    cfg.validate()?;
536
537    if socket_path.exists() {
538        std::fs::remove_file(&socket_path)
539            .with_context(|| format!("remove stale socket {}", socket_path.display()))?;
540    }
541
542    let project_root = cfg.project_root.to_string_lossy().to_string();
543    let service_project_root = project_root.clone();
544    let service_factory = move || -> Result<LeanCtxServer, std::io::Error> {
545        Ok(LeanCtxServer::new_shared_with_context(
546            &service_project_root,
547            "default",
548            "default",
549        ))
550    };
551    let mcp_http = StreamableHttpService::new(
552        service_factory,
553        Arc::new(
554            rmcp::transport::streamable_http_server::session::local::LocalSessionManager::default(),
555        ),
556        cfg.mcp_http_config(),
557    );
558
559    let state = AppState {
560        token: cfg.auth_token.clone().filter(|t| !t.is_empty()),
561        concurrency: Arc::new(tokio::sync::Semaphore::new(cfg.max_concurrency.max(1))),
562        rate: Arc::new(RateLimiter::new(cfg.max_rps, cfg.rate_burst)),
563        project_root: project_root.clone(),
564        timeout: Duration::from_millis(cfg.request_timeout_ms.max(1)),
565    };
566
567    let app = Router::new()
568        .route("/health", get(health))
569        .route("/v1/manifest", get(v1_manifest))
570        .route("/v1/tools", get(v1_tools))
571        .route("/v1/tools/call", axum::routing::post(v1_tool_call))
572        .route("/v1/events", get(v1_events))
573        .route(
574            "/v1/context/summary",
575            get(context_views::v1_context_summary),
576        )
577        .route("/v1/events/search", get(context_views::v1_events_search))
578        .route("/v1/events/lineage", get(context_views::v1_event_lineage))
579        .route("/v1/metrics", get(v1_metrics))
580        .fallback_service(mcp_http)
581        .layer(axum::extract::DefaultBodyLimit::max(cfg.max_body_bytes))
582        .layer(middleware::from_fn_with_state(
583            state.clone(),
584            rate_limit_middleware,
585        ))
586        .layer(middleware::from_fn_with_state(
587            state.clone(),
588            concurrency_middleware,
589        ))
590        .layer(middleware::from_fn_with_state(
591            state.clone(),
592            auth_middleware,
593        ))
594        .with_state(state);
595
596    let listener = tokio::net::UnixListener::bind(&socket_path)
597        .with_context(|| format!("bind UDS {}", socket_path.display()))?;
598
599    tracing::info!(
600        "lean-ctx daemon listening on {} (project_root={})",
601        socket_path.display(),
602        cfg.project_root.display()
603    );
604
605    axum::serve(listener, app.into_make_service())
606        .with_graceful_shutdown(async move {
607            let _ = tokio::signal::ctrl_c().await;
608        })
609        .await
610        .context("uds server")?;
611    Ok(())
612}
613
614#[cfg(test)]
615mod tests {
616    use super::*;
617    use axum::body::Body;
618    use axum::http::Request;
619    use futures::StreamExt;
620    use rmcp::transport::{StreamableHttpServerConfig, StreamableHttpService};
621    use serde_json::json;
622    use tower::ServiceExt;
623
624    async fn read_first_sse_message(body: Body) -> String {
625        let mut stream = body.into_data_stream();
626        let mut buf: Vec<u8> = Vec::new();
627        for _ in 0..32 {
628            let next = tokio::time::timeout(Duration::from_secs(2), stream.next()).await;
629            let Ok(Some(Ok(bytes))) = next else {
630                break;
631            };
632            buf.extend_from_slice(&bytes);
633            if buf.windows(2).any(|w| w == b"\n\n") {
634                break;
635            }
636        }
637        String::from_utf8_lossy(&buf).to_string()
638    }
639
640    #[tokio::test]
641    async fn auth_token_blocks_requests_without_bearer_header() {
642        let dir = tempfile::tempdir().expect("tempdir");
643        let root_str = dir.path().to_string_lossy().to_string();
644        let service_project_root = root_str.clone();
645        let service_factory = move || -> Result<LeanCtxServer, std::io::Error> {
646            Ok(LeanCtxServer::new_shared_with_context(
647                &service_project_root,
648                "default",
649                "default",
650            ))
651        };
652        let cfg = StreamableHttpServerConfig::default()
653            .with_stateful_mode(false)
654            .with_json_response(true);
655
656        let mcp_http = StreamableHttpService::new(
657            service_factory,
658            Arc::new(
659                rmcp::transport::streamable_http_server::session::local::LocalSessionManager::default(),
660            ),
661            cfg,
662        );
663
664        let state = AppState {
665            token: Some("secret".to_string()),
666            concurrency: Arc::new(tokio::sync::Semaphore::new(4)),
667            rate: Arc::new(RateLimiter::new(50, 100)),
668            project_root: root_str.clone(),
669            timeout: Duration::from_millis(30_000),
670        };
671
672        let app = Router::new()
673            .fallback_service(mcp_http)
674            .layer(middleware::from_fn_with_state(
675                state.clone(),
676                auth_middleware,
677            ))
678            .with_state(state);
679
680        let body = json!({
681            "jsonrpc": "2.0",
682            "id": 1,
683            "method": "tools/list",
684            "params": {}
685        })
686        .to_string();
687
688        let req = Request::builder()
689            .method("POST")
690            .uri("/")
691            .header("Host", "localhost")
692            .header("Accept", "application/json, text/event-stream")
693            .header("Content-Type", "application/json")
694            .body(Body::from(body))
695            .expect("request");
696
697        let resp = app.clone().oneshot(req).await.expect("resp");
698        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
699    }
700
701    #[tokio::test]
702    async fn mcp_service_factory_isolates_per_client_state() {
703        let dir = tempfile::tempdir().expect("tempdir");
704        let root_str = dir.path().to_string_lossy().to_string();
705
706        // Mirrors the serve() setup: service_factory must create a fresh server per MCP session.
707        let service_project_root = root_str.clone();
708        let service_factory = move || -> Result<LeanCtxServer, std::convert::Infallible> {
709            Ok(LeanCtxServer::new_shared_with_context(
710                &service_project_root,
711                "default",
712                "default",
713            ))
714        };
715
716        let s1 = service_factory().expect("server 1");
717        let s2 = service_factory().expect("server 2");
718
719        // If the two servers accidentally share the same Arc-backed fields, these writes would
720        // clobber each other. This test stays independent of rmcp's InitializeRequestParams API.
721        *s1.client_name.write().await = "client-a".to_string();
722        *s2.client_name.write().await = "client-b".to_string();
723
724        let a = s1.client_name.read().await.clone();
725        let b = s2.client_name.read().await.clone();
726        assert_eq!(a, "client-a");
727        assert_eq!(b, "client-b");
728    }
729
730    #[tokio::test]
731    async fn rate_limit_returns_429_when_exhausted() {
732        let state = AppState {
733            token: None,
734            concurrency: Arc::new(tokio::sync::Semaphore::new(16)),
735            rate: Arc::new(RateLimiter::new(1, 1)),
736            project_root: ".".to_string(),
737            timeout: Duration::from_millis(30_000),
738        };
739
740        let app = Router::new()
741            .route("/limited", get(|| async { (StatusCode::OK, "ok\n") }))
742            .layer(middleware::from_fn_with_state(
743                state.clone(),
744                rate_limit_middleware,
745            ))
746            .with_state(state);
747
748        let req1 = Request::builder()
749            .method("GET")
750            .uri("/limited")
751            .header("Host", "localhost")
752            .body(Body::empty())
753            .expect("req1");
754        let resp1 = app.clone().oneshot(req1).await.expect("resp1");
755        assert_eq!(resp1.status(), StatusCode::OK);
756
757        let req2 = Request::builder()
758            .method("GET")
759            .uri("/limited")
760            .header("Host", "localhost")
761            .body(Body::empty())
762            .expect("req2");
763        let resp2 = app.clone().oneshot(req2).await.expect("resp2");
764        assert_eq!(resp2.status(), StatusCode::TOO_MANY_REQUESTS);
765    }
766
767    #[tokio::test]
768    async fn events_endpoint_replays_tool_call_event() {
769        let dir = tempfile::tempdir().expect("tempdir");
770        std::fs::create_dir_all(dir.path().join(".git")).expect("git marker");
771        std::fs::write(dir.path().join("a.txt"), "ok").expect("file");
772        let root_str = dir.path().to_string_lossy().to_string();
773
774        let state = AppState {
775            token: None,
776            concurrency: Arc::new(tokio::sync::Semaphore::new(16)),
777            rate: Arc::new(RateLimiter::new(50, 100)),
778            project_root: root_str.clone(),
779            timeout: Duration::from_millis(30_000),
780        };
781
782        let app = Router::new()
783            .route("/v1/tools/call", axum::routing::post(v1_tool_call))
784            .route("/v1/events", get(v1_events))
785            .with_state(state);
786
787        let body = json!({
788            "name": "ctx_session",
789            "arguments": { "action": "status" },
790            "workspaceId": "ws1",
791            "channelId": "ch1"
792        })
793        .to_string();
794        let req = Request::builder()
795            .method("POST")
796            .uri("/v1/tools/call")
797            .header("Host", "localhost")
798            .header("Content-Type", "application/json")
799            .body(Body::from(body))
800            .expect("req");
801        let resp = app.clone().oneshot(req).await.expect("call");
802        assert_eq!(resp.status(), StatusCode::OK);
803
804        // Allow async event persistence to complete (Windows CI disk IO is slower).
805        tokio::time::sleep(Duration::from_millis(250)).await;
806
807        // Subscribe with replay semantics; read the first SSE message.
808        let req = Request::builder()
809            .method("GET")
810            .uri("/v1/events?workspaceId=ws1&channelId=ch1&since=0&limit=1")
811            .header("Host", "localhost")
812            .header("Accept", "text/event-stream")
813            .body(Body::empty())
814            .expect("req");
815        let resp = app.clone().oneshot(req).await.expect("events");
816        assert_eq!(resp.status(), StatusCode::OK);
817
818        let msg = read_first_sse_message(resp.into_body()).await;
819        assert!(msg.contains("event: tool_call_recorded"), "msg={msg:?}");
820        assert!(msg.contains("\"workspaceId\":\"ws1\""), "msg={msg:?}");
821        assert!(msg.contains("\"channelId\":\"ch1\""), "msg={msg:?}");
822    }
823}
lean_ctx/http_server/mod.rs

lean_ctx/http_server/
mod.rs
