From bd01fb3375d50d7bc280bcafc9ba003b939bf7f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=B6ren=20Nikolaus?= Date: Mon, 7 Sep 2026 10:49:43 +0200 Subject: [PATCH] feat(monitor): attach dashboards to a background proxy --- README.md | 4 + .../docs/reference/command-reference.md | 10 + docs/src/content/docs/reference/http-api.md | 8 + docs/src/content/docs/using/monitor-tui.md | 24 +- src/main.rs | 58 +++- src/monitor.rs | 11 +- src/monitor/remote.rs | 86 +++++ src/monitor/snapshot.rs | 300 ++++++++++++++++++ src/server.rs | 46 ++- src/tui.rs | 171 +++++++--- tests/remote_monitor.rs | 258 +++++++++++++++ 11 files changed, 919 insertions(+), 57 deletions(-) create mode 100644 src/monitor/remote.rs create mode 100644 src/monitor/snapshot.rs create mode 100644 tests/remote_monitor.rs diff --git a/README.md b/README.md index 0a48409e..c9e6f0bb 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,10 @@ Start the proxy in one terminal: claude-code-proxy serve ``` +For a background service, use `claude-code-proxy serve --no-monitor` and attach +from another terminal with `claude-code-proxy monitor`. Closing an attached +dashboard leaves the proxy running. + Start Claude Code in another: ```sh diff --git a/docs/src/content/docs/reference/command-reference.md b/docs/src/content/docs/reference/command-reference.md index 9d5b9d48..eb1c7fd4 100644 --- a/docs/src/content/docs/reference/command-reference.md +++ b/docs/src/content/docs/reference/command-reference.md @@ -30,6 +30,16 @@ Starts the local HTTP proxy and blocks until shutdown. The bind address comes from `CCP_BIND_ADDRESS` or `bindAddress`. Interactive stdout opens the monitor unless `--no-monitor` is present. Non-terminal stdout uses plain mode. +Plain mode continues collecting monitor history and supports separate dashboards. SIGTERM (Unix) and Ctrl-C request graceful service shutdown. + +## `monitor` + +```sh +claude-code-proxy monitor [--url ] +``` + +Attach a read-only dashboard to a running proxy. The default URL is `http://127.0.0.1:`. No provider login is needed in the dashboard process. `q` and `Ctrl-C` detach without stopping the proxy; multiple dashboards are supported. After a connection failure, the dashboard retains its last snapshot and retries. The proxy accepts monitor reads only from loopback peers; use an SSH port forward for another machine. + ## `demo` ```sh diff --git a/docs/src/content/docs/reference/http-api.md b/docs/src/content/docs/reference/http-api.md index e5fbbaf0..5e1257cf 100644 --- a/docs/src/content/docs/reference/http-api.md +++ b/docs/src/content/docs/reference/http-api.md @@ -19,6 +19,14 @@ Liveness check: It does not verify provider credentials or upstream availability. +## `GET /monitor` + +Read-only dashboard snapshot, available when monitor collection is enabled. The CLI enables collection in both plain and interactive `serve` modes. Only actual loopback peers are allowed; forwarded IP headers do not grant access. Other peers receive HTTP 403, and disabled collection returns HTTP 404. Responses use `Cache-Control: no-store`. + +The JSON envelope contains `version: 1` and a `snapshot` with the proxy start time, sessions, active requests and recent requests. It includes the monitor's display metadata, timing, token usage and computed throughput. It does not include credentials or request/response bodies. Process-local monotonic clocks remain in the service; snapshots contain elapsed durations instead. + +The `monitor` CLI polls this endpoint. There is no corresponding mutation or shutdown endpoint. Attaching or disconnecting a viewer does not change proxy lifetime or request accounting. + ## `POST /v1/messages` Accepts an Anthropic Messages request in streaming or non-streaming mode. `POST /v1/messages?beta=true` reaches the same route. diff --git a/docs/src/content/docs/using/monitor-tui.md b/docs/src/content/docs/using/monitor-tui.md index ad70c795..010d7020 100644 --- a/docs/src/content/docs/using/monitor-tui.md +++ b/docs/src/content/docs/using/monitor-tui.md @@ -5,6 +5,20 @@ description: Use the claude-code-proxy monitor to inspect sessions, active and r `claude-code-proxy serve` opens the monitor when stdout is an interactive terminal. The same process runs the HTTP listener. +To run the proxy as a service and attach the dashboard separately: + +```sh +# Run this under your service manager, or leave it in another terminal. +claude-code-proxy serve --no-monitor + +# Attach from any terminal; repeat for additional dashboards. +claude-code-proxy monitor +``` + +Use `claude-code-proxy monitor --url http://127.0.0.1:19999` for a different port. Without `--url`, the port follows the usual proxy configuration. The attached dashboard reads the running service's existing history; it does not start a proxy or need provider credentials. + +In an attached dashboard, `q` and `Ctrl-C` detach immediately and leave the service running. Multiple dashboards can attach independently. If the service becomes unavailable, the dashboard marks its last snapshot as stale and reconnects automatically. Network polling runs outside the terminal event loop. + ![claude-code-proxy monitor showing sessions, active requests, recent requests, and events](/monitor-tui.webp) ## What the monitor shows @@ -27,8 +41,8 @@ description: Use the claude-code-proxy monitor to inspect sessions, active and r | `Esc` | Close details or an overlay | | `?` | Toggle shortcut help | | `b` | Toggle the setup overlay | -| `q` | Request a graceful shutdown | -| `Ctrl-C` | Force shutdown | +| `q` | Detach an attached dashboard; in the built-in dashboard, confirm proxy shutdown | +| `Ctrl-C` | Detach an attached dashboard; in the built-in dashboard, start shutdown (press again to force exit) | The request table changes columns as the terminal width changes. @@ -42,6 +56,8 @@ claude-code-proxy serve --no-monitor Non-terminal stdout also selects plain mode. `CCP_LOG_STDERR=1` mirrors JSONL log events to stderr in plain mode. +Plain mode retains monitor accounting even with no dashboard attached. On Unix, SIGTERM starts graceful proxy shutdown; Ctrl-C does the same. The service manager owns the process lifetime. + ## Demo mode Explore the full interface without binding a port or using provider credentials: @@ -61,3 +77,7 @@ brew services start claude-code-proxy ``` Service output lives in `~/.local/state/claude-code-proxy/service.log` on macOS and Linux. The structured `proxy.log` shares the state directory. Provider login remains an interactive one-time command. + +Run `claude-code-proxy monitor` to inspect that service. The monitor endpoint only accepts loopback connections, even if the inference listener binds a LAN address. For a service on another machine, forward its port with SSH and point `monitor --url` at the local end of the tunnel. + +History remains in the proxy's memory and resets when the proxy restarts. The dashboard polls snapshots every 250 ms and displays server-computed durations and throughput. The attached setup overlay describes its connection; provider setup remains with the service and its built-in dashboard. diff --git a/src/main.rs b/src/main.rs index 68f85548..958267dd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -38,6 +38,11 @@ enum Commands { #[arg(long = "no-monitor", action = ArgAction::SetTrue)] no_monitor: bool, }, + /// Attach a read-only dashboard to a running proxy + Monitor { + #[arg(long)] + url: Option, + }, /// Open the monitor TUI with mock data and no proxy server #[command(hide = true)] Demo, @@ -105,10 +110,10 @@ fn main() -> Result<()> { ServeMode::Plain => { print_server_banner(&bind_address, effective_port, ®istry); runtime - .block_on(server::serve(ServerConfig { + .block_on(run_service(ServerConfig { bind_address, port: effective_port, - monitor: None, + monitor: Some(MonitorHandle::default()), })) .map_err(|err| anyhow::anyhow!(err)) } @@ -157,6 +162,25 @@ fn main() -> Result<()> { let registry = Registry::with_default_alias(); tui::run_mock_monitor(config::port(), ®istry) } + Commands::Monitor { url } => { + let url = url.unwrap_or_else(|| { + format!("http://127.0.0.1:{}", config::port()) + .parse() + .expect("local proxy URL") + }); + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()?; + let client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(std::time::Duration::from_secs(2)) + .build()?; + let monitor = runtime.block_on( + claude_code_proxy::monitor::remote::RemoteMonitor::connect(client, url.clone()), + )?; + tui::run_attached_monitor(|| monitor.snapshot(), url.to_string())?; + Ok(()) + } Commands::Models { full } => { print_models(&Registry::with_default_alias(), full); Ok(()) @@ -168,6 +192,36 @@ fn main() -> Result<()> { } } +async fn run_service(config: ServerConfig) -> Result<()> { + let (shutdown, stopped) = tokio::sync::oneshot::channel(); + let server = server::serve_with_shutdown(config, async { + let _ = stopped.await; + }); + tokio::pin!(server); + tokio::select! { + result = &mut server => result, + signal = service_shutdown_signal() => { + signal?; + let _ = shutdown.send(()); + server.await + } + } +} + +#[cfg(unix)] +async fn service_shutdown_signal() -> std::io::Result<()> { + let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; + tokio::select! { + result = tokio::signal::ctrl_c() => result, + _ = terminate.recv() => Ok(()), + } +} + +#[cfg(not(unix))] +async fn service_shutdown_signal() -> std::io::Result<()> { + tokio::signal::ctrl_c().await +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ServeMode { Monitor, diff --git a/src/monitor.rs b/src/monitor.rs index d8dd4bf6..d6b07d10 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -6,13 +6,16 @@ use std::{ }; mod mock; +pub mod remote; +pub mod snapshot; pub use mock::{MockMonitor, mock_state}; const DEFAULT_RECENT_LIMIT: usize = 200; pub const SESSION_TOKEN_BUCKET_SECS: u64 = 10; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum EndpointKind { Messages, CountTokens, @@ -35,7 +38,8 @@ impl EndpointKind { } } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum RequestStatus { Started, ProviderSelected, @@ -210,7 +214,8 @@ impl CompletedRequest { } } -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "unit", content = "value", rename_all = "snake_case")] pub enum Throughput { TokensPerSecond(f64), BytesPerSecond(f64), diff --git a/src/monitor/remote.rs b/src/monitor/remote.rs new file mode 100644 index 00000000..4fc335cc --- /dev/null +++ b/src/monitor/remote.rs @@ -0,0 +1,86 @@ +use std::time::Duration; + +use anyhow::{Context, Result, bail}; +use reqwest::{Client, Url}; +use tokio::{sync::watch, task::JoinHandle, time::MissedTickBehavior}; + +use super::snapshot::{MonitorResponse, MonitorSnapshot, PROTOCOL_VERSION, SnapshotUpdate}; + +const MAX_SNAPSHOT_BYTES: usize = 16 * 1024 * 1024; +const POLL_INTERVAL: Duration = Duration::from_millis(250); + +/// Owns polling; dropping the dashboard cancels only its outstanding read. +pub struct RemoteMonitor { + updates: watch::Receiver, + poller: JoinHandle<()>, +} + +impl RemoteMonitor { + pub async fn connect(client: Client, base_url: Url) -> Result { + if !matches!(base_url.scheme(), "http" | "https") + || !base_url.username().is_empty() + || base_url.password().is_some() + || base_url.query().is_some() + || base_url.fragment().is_some() + { + bail!("monitor URL must be an HTTP(S) base URL without credentials, query or fragment"); + } + let endpoint = base_url.join("monitor").context("invalid monitor URL")?; + let initial = fetch_snapshot(&client, &endpoint) + .await + .context("cannot attach to proxy monitor")?; + let (sender, updates) = watch::channel(SnapshotUpdate::Live(initial.clone())); + let poller = tokio::spawn(async move { + let mut current = SnapshotUpdate::Live(initial); + let mut interval = tokio::time::interval(POLL_INTERVAL); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + interval.tick().await; + loop { + interval.tick().await; + let result = fetch_snapshot(&client, &endpoint) + .await + .map_err(|error| error.to_string()); + current = current.updated(result); + if sender.send(current.clone()).is_err() { + break; + } + } + }); + Ok(Self { updates, poller }) + } + + pub fn snapshot(&self) -> SnapshotUpdate { + self.updates.borrow().clone() + } +} + +impl Drop for RemoteMonitor { + fn drop(&mut self) { + self.poller.abort(); + } +} + +async fn fetch_snapshot(client: &Client, endpoint: &Url) -> Result { + let mut response = client + .get(endpoint.clone()) + .send() + .await? + .error_for_status()?; + let mut body = Vec::new(); + while let Some(chunk) = response.chunk().await? { + if body.len().saturating_add(chunk.len()) > MAX_SNAPSHOT_BYTES { + bail!("monitor snapshot exceeds the size limit"); + } + body.extend_from_slice(&chunk); + } + let response: MonitorResponse = + serde_json::from_slice(&body).context("invalid monitor snapshot")?; + if response.version != PROTOCOL_VERSION { + bail!( + "unsupported monitor protocol version {}; expected {}", + response.version, + PROTOCOL_VERSION + ); + } + Ok(response.snapshot) +} diff --git a/src/monitor/snapshot.rs b/src/monitor/snapshot.rs new file mode 100644 index 00000000..1a55ddc0 --- /dev/null +++ b/src/monitor/snapshot.rs @@ -0,0 +1,300 @@ +//! Immutable dashboard read models. Process-local clocks and mutable accounting +//! stay in the monitor store; viewers receive elapsed durations and computed rates. +use super::{ + ActiveRequest, CompletedRequest, EndpointKind, MonitorState, RequestStatus, SessionSummary, + Throughput, +}; +use serde::{Deserialize, Serialize}; +use std::{ + path::PathBuf, + time::{Duration, SystemTime}, +}; + +pub const PROTOCOL_VERSION: u32 = 1; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MonitorResponse { + pub version: u32, + pub snapshot: MonitorSnapshot, +} + +impl From for MonitorResponse { + fn from(state: MonitorState) -> Self { + Self { + version: PROTOCOL_VERSION, + snapshot: state.into(), + } + } +} + +#[derive(Debug, Clone)] +pub enum SnapshotUpdate { + Live(MonitorSnapshot), + Disconnected { + snapshot: MonitorSnapshot, + error: String, + }, +} + +impl SnapshotUpdate { + pub fn snapshot(&self) -> &MonitorSnapshot { + match self { + Self::Live(snapshot) | Self::Disconnected { snapshot, .. } => snapshot, + } + } + + pub fn connection_error(&self) -> Option<&str> { + match self { + Self::Live(_) => None, + Self::Disconnected { error, .. } => Some(error), + } + } + + pub(super) fn updated(self, result: Result) -> Self { + match result { + Ok(snapshot) => Self::Live(snapshot), + Err(error) => { + let snapshot = match self { + Self::Live(snapshot) | Self::Disconnected { snapshot, .. } => snapshot, + }; + Self::Disconnected { snapshot, error } + } + } + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MonitorSnapshot { + pub started_at: SystemTime, + pub sessions: Vec, + pub active: Vec, + pub recent: Vec, +} + +impl From for MonitorSnapshot { + fn from(state: MonitorState) -> Self { + Self { + started_at: state.started_at, + sessions: state.sessions.into_iter().map(Into::into).collect(), + active: state.active.into_iter().map(Into::into).collect(), + recent: state.recent.into_iter().map(Into::into).collect(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ActiveSnapshot { + pub request_id: String, + pub session_id: Option, + pub session_seq: Option, + pub project: Option, + pub provider: Option, + pub model: Option, + pub effort: Option, + pub endpoint: EndpointKind, + pub started_at: SystemTime, + pub generation_started_at: Option, + pub generation_finished_at: Option, + pub generation_duration: Option, + pub status: RequestStatus, + pub streamed_bytes: u64, + pub stream_chunks: u64, + pub input_tokens: Option, + pub output_tokens: Option, + pub error: Option, + pub traffic_capture_path: Option, + elapsed: Duration, + throughput: Throughput, +} + +impl ActiveSnapshot { + pub fn elapsed(&self) -> Duration { + self.elapsed + } + pub fn rate(&self) -> Throughput { + self.throughput.clone() + } +} + +impl From for ActiveSnapshot { + fn from(value: ActiveRequest) -> Self { + let throughput = value.rate(); + let elapsed = value.elapsed(); + Self { + request_id: value.request_id, + session_id: value.session_id, + session_seq: value.session_seq, + project: value.project, + provider: value.provider, + model: value.model, + effort: value.effort, + endpoint: value.endpoint, + started_at: value.started_at, + generation_started_at: value.generation_started_at, + generation_finished_at: value.generation_finished_at, + generation_duration: value.generation_duration, + status: value.status, + streamed_bytes: value.streamed_bytes, + stream_chunks: value.stream_chunks, + input_tokens: value.input_tokens, + output_tokens: value.output_tokens, + error: value.error, + traffic_capture_path: value.traffic_capture_path, + elapsed, + throughput, + } + } +} +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct CompletedSnapshot { + pub request_id: String, + pub session_id: Option, + pub session_seq: Option, + pub project: Option, + pub provider: Option, + pub model: Option, + pub effort: Option, + pub endpoint: EndpointKind, + pub started_at: SystemTime, + pub generation_started_at: Option, + pub generation_finished_at: Option, + pub generation_duration: Option, + pub status: RequestStatus, + pub streamed_bytes: u64, + pub stream_chunks: u64, + pub input_tokens: Option, + pub output_tokens: Option, + pub error: Option, + pub traffic_capture_path: Option, + pub finished_at: SystemTime, + pub http_status: Option, + pub latency: Duration, + throughput: Throughput, +} + +impl CompletedSnapshot { + pub fn rate(&self) -> Throughput { + self.throughput.clone() + } +} + +impl From for CompletedSnapshot { + fn from(value: CompletedRequest) -> Self { + let throughput = value.rate(); + Self { + request_id: value.request_id, + session_id: value.session_id, + session_seq: value.session_seq, + project: value.project, + provider: value.provider, + model: value.model, + effort: value.effort, + endpoint: value.endpoint, + started_at: value.started_at, + generation_started_at: value.generation_started_at, + generation_finished_at: value.generation_finished_at, + generation_duration: value.generation_duration, + status: value.status, + streamed_bytes: value.streamed_bytes, + stream_chunks: value.stream_chunks, + input_tokens: value.input_tokens, + output_tokens: value.output_tokens, + error: value.error, + traffic_capture_path: value.traffic_capture_path, + finished_at: value.finished_at, + http_status: value.http_status, + latency: value.latency, + throughput, + } + } +} +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct SessionSnapshot { + pub session_id: Option, + pub project: Option, + pub active_count: usize, + pub request_count: usize, + pub failure_count: usize, + pub provider: Option, + pub model: Option, + pub effort: Option, + pub last_seen: SystemTime, + pub input_tokens: u64, + pub output_tokens: u64, + pub output_token_samples: Vec<(SystemTime, u64)>, + pub generation_duration: Duration, + pub last_status: String, + throughput: Throughput, +} + +impl SessionSnapshot { + pub fn rate(&self) -> Throughput { + self.throughput.clone() + } +} + +impl From for SessionSnapshot { + fn from(value: SessionSummary) -> Self { + let throughput = value.rate(); + Self { + session_id: value.session_id, + project: value.project, + active_count: value.active_count, + request_count: value.request_count, + failure_count: value.failure_count, + provider: value.provider, + model: value.model, + effort: value.effort, + last_seen: value.last_seen, + input_tokens: value.input_tokens, + output_tokens: value.output_tokens, + output_token_samples: value.output_token_samples, + generation_duration: value.generation_duration, + last_status: value.last_status, + throughput, + } + } +} + +impl SessionSnapshot { + pub fn label(&self) -> String { + self.session_id + .clone() + .unwrap_or_else(|| "no-session".to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::monitor::MonitorHandle; + + #[test] + fn snapshot_round_trip_preserves_accounting_without_process_clocks() { + let monitor = MonitorHandle::default(); + monitor.request_started( + "active", + Some("session".into()), + Some(2), + EndpointKind::Messages, + ); + monitor.provider_selected("active", "codex", "gpt-6-astra", Some("high".into())); + monitor.stream_progress("active", 40, 1, Some(8), Some(3)); + monitor.request_started( + "done", + Some("session".into()), + Some(1), + EndpointKind::CountTokens, + ); + monitor.request_completed("done", 200, Some(11), None); + let snapshot = MonitorSnapshot::from(monitor.snapshot()); + let encoded = serde_json::to_string(&snapshot).unwrap(); + assert!(!encoded.contains("instant")); + let decoded: MonitorSnapshot = serde_json::from_str(&encoded).unwrap(); + assert_eq!(decoded, snapshot); + assert_eq!(decoded.active[0].elapsed(), snapshot.active[0].elapsed()); + assert_eq!(decoded.active[0].rate(), snapshot.active[0].rate()); + assert_eq!(decoded.sessions[0].rate(), snapshot.sessions[0].rate()); + assert_eq!(decoded.recent[0].http_status, Some(200)); + } +} diff --git a/src/server.rs b/src/server.rs index 913a71cc..1f14e39a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -34,7 +34,7 @@ use axum::{ body::Body, extract::{DefaultBodyLimit, FromRequest, Multipart, Query, State}, http::{Request, StatusCode}, - response::Response, + response::{IntoResponse, Response}, routing::{get, post}, }; use http_body_util::{BodyExt, StreamBody}; @@ -168,9 +168,12 @@ pub async fn serve_listener( ])), ); let app = app_with_monitor(Arc::new(Registry::with_default_alias()), monitor); - axum::serve(listener, app) - .with_graceful_shutdown(shutdown) - .await?; + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .with_graceful_shutdown(shutdown) + .await?; Ok(()) } @@ -259,6 +262,7 @@ pub fn app_with_features( }); let router = Router::new() .route("/healthz", get(healthz)) + .route("/monitor", get(handler_monitor)) .route("/v1/messages", post(handler_messages)) .route("/v1/messages/count_tokens", post(handler_count_tokens)) .route("/v1/models", get(handler_models)); @@ -301,6 +305,40 @@ struct AppState { transcriptions: Option>, } +async fn handler_monitor( + State(state): State>, + peer: Option>>, +) -> Response { + let Some(axum::Extension(axum::extract::ConnectInfo(peer))) = peer else { + return json_error( + StatusCode::FORBIDDEN, + "permission_error", + "Monitor access requires a local connection", + ); + }; + if !peer.ip().is_loopback() { + return json_error( + StatusCode::FORBIDDEN, + "permission_error", + "Monitor access requires a local connection", + ); + } + match &state.monitor { + Some(monitor) => ( + [(http::header::CACHE_CONTROL, "no-store")], + Json(crate::monitor::snapshot::MonitorResponse::from( + monitor.snapshot(), + )), + ) + .into_response(), + None => json_error( + StatusCode::NOT_FOUND, + "not_found_error", + "Monitor collection is disabled", + ), + } +} + async fn healthz() -> Json { Json(json!({ "ok": true })) } diff --git a/src/tui.rs b/src/tui.rs index 25766c6c..ce5d048f 100644 --- a/src/tui.rs +++ b/src/tui.rs @@ -32,8 +32,10 @@ use tokio::sync::oneshot; use crate::{ monitor::{ - ActiveRequest, CompletedRequest, MockMonitor, MonitorHandle, MonitorState, - SESSION_TOKEN_BUCKET_SECS, SessionSummary, + MockMonitor, MonitorHandle, SESSION_TOKEN_BUCKET_SECS, + snapshot::{ + ActiveSnapshot, CompletedSnapshot, MonitorSnapshot, SessionSnapshot, SnapshotUpdate, + }, }, paths, registry::Registry, @@ -67,41 +69,78 @@ pub struct MonitorUiConfig<'a> { pub enum MonitorExit { ShutdownComplete, ForceQuit, + Detached, } pub fn run_monitor( handle: MonitorHandle, config: MonitorUiConfig<'_>, ) -> Result { - run_monitor_loop(|| handle.snapshot(), config, None) + run_monitor_loop( + || SnapshotUpdate::Live(handle.snapshot().into()), + UiSession::from(config), + ) +} + +pub fn run_attached_monitor( + snapshot: impl FnMut() -> SnapshotUpdate, + listen_url: String, +) -> Result { + let setup_text = format!( + "Attached to {listen_url}\nClosing this dashboard leaves the proxy running.\nConfigure providers and authentication in the proxy service." + ); + run_monitor_loop( + snapshot, + UiSession { + listen_url, + setup_text, + shutdown: None, + shutdown_complete: None, + }, + ) } pub fn run_mock_monitor(port: u16, registry: &Registry) -> Result<(), anyhow::Error> { let mut monitor = MockMonitor::new(); run_monitor_loop( - move || monitor.snapshot(), - MonitorUiConfig { + move || SnapshotUpdate::Live(monitor.snapshot().into()), + UiSession { listen_url: "mock://tui-demo".to_string(), - port, - registry, + setup_text: mock_setup_text(port, registry), shutdown: None, shutdown_complete: None, }, - Some(mock_setup_text(port, registry)), ) .map(|_| ()) } +struct UiSession { + listen_url: String, + setup_text: String, + shutdown: Option>, + shutdown_complete: Option>, +} + +impl From> for UiSession { + fn from(config: MonitorUiConfig<'_>) -> Self { + Self { + setup_text: setup_text(config.port, config.registry), + listen_url: config.listen_url, + shutdown: config.shutdown, + shutdown_complete: config.shutdown_complete, + } + } +} + fn run_monitor_loop( - mut snapshot: impl FnMut() -> MonitorState, - config: MonitorUiConfig<'_>, - setup_text_override: Option, + mut snapshot: impl FnMut() -> SnapshotUpdate, + config: UiSession, ) -> Result { let mut terminal = setup_terminal()?; let _guard = TerminalGuard; let mut app = MonitorApp { listen_url: config.listen_url, - setup_text: setup_text_override.unwrap_or_else(|| setup_text(config.port, config.registry)), + setup_text: config.setup_text, show_setup: false, show_help: false, detail: None, @@ -118,7 +157,7 @@ fn run_monitor_loop( if run_result.is_err() { app.begin_shutdown(); let state = snapshot(); - let _ = terminal.draw(|frame| render(frame, &mut app, &state)); + let _ = terminal.draw(|frame| render(frame, &mut app, state.snapshot())); app.wait_for_shutdown_completion(); } let cursor_result = terminal.show_cursor(); @@ -129,20 +168,43 @@ fn run_monitor_loop( fn run_monitor_events( terminal: &mut Terminal>, - mut snapshot: impl FnMut() -> MonitorState, + mut snapshot: impl FnMut() -> SnapshotUpdate, app: &mut MonitorApp, ) -> Result { loop { - let state = snapshot(); + let update = snapshot(); + let state = update.snapshot(); app.clamp_selection(state.sessions.len(), state.recent.len()); app.tick = app.tick.wrapping_add(1); - terminal.draw(|frame| render(frame, app, &state))?; + terminal.draw(|frame| { + render(frame, app, state); + if let Some(error) = update.connection_error() { + let area = frame.area(); + let banner = Rect::new( + area.x, + area.y + area.height.saturating_sub(1), + area.width, + 1, + ); + frame.render_widget( + Paragraph::new(format!("Reconnecting; showing last snapshot: {error}")) + .style(Style::default().fg(YELLOW).bg(BG)), + banner, + ); + } + })?; if app.shutdown_is_complete() { return Ok(MonitorExit::ShutdownComplete); } if event::poll(Duration::from_millis(250))? { match event::read()? { Event::Key(key) => match key.code { + KeyCode::Char('q') if app.is_attached() => return Ok(MonitorExit::Detached), + KeyCode::Char('c') + if app.is_attached() && key.modifiers.contains(KeyModifiers::CONTROL) => + { + return Ok(MonitorExit::Detached); + } KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => { if app.handle_ctrl_c() { return Ok(MonitorExit::ForceQuit); @@ -245,6 +307,9 @@ struct MonitorApp { } impl MonitorApp { + fn is_attached(&self) -> bool { + self.shutdown.is_none() && self.shutdown_complete.is_none() + } fn handle_ctrl_c(&mut self) -> bool { if self.phase == MonitorPhase::ShuttingDown { true @@ -363,7 +428,7 @@ fn setup_terminal() -> Result>, anyhow::Error> Ok(terminal) } -fn render(frame: &mut ratatui::Frame<'_>, app: &mut MonitorApp, state: &MonitorState) { +fn render(frame: &mut ratatui::Frame<'_>, app: &mut MonitorApp, state: &MonitorSnapshot) { let area = frame.area(); frame.render_widget(Block::default().style(Style::default().bg(BG)), area); @@ -408,7 +473,7 @@ fn render(frame: &mut ratatui::Frame<'_>, app: &mut MonitorApp, state: &MonitorS render_setup_overlay(frame, area, &app.setup_text); } if app.show_help { - render_help_overlay(frame, area); + render_help_overlay(frame, area, app.is_attached()); } match app.phase { MonitorPhase::Running => {} @@ -421,7 +486,7 @@ fn render_header( frame: &mut ratatui::Frame<'_>, area: Rect, app: &MonitorApp, - state: &MonitorState, + state: &MonitorSnapshot, ) { let uptime = state .started_at @@ -649,7 +714,7 @@ fn detail_cell(value: &str) -> Cell<'static> { } } -fn error_indicator(request: &CompletedRequest) -> &'static str { +fn error_indicator(request: &CompletedSnapshot) -> &'static str { if request.status == crate::monitor::RequestStatus::Failed || request.http_status.is_some_and(|status| status >= 400) || request @@ -840,7 +905,7 @@ fn session_columns(tier: LayoutTier, show_full_sparkline: bool) -> Vec, area: Rect, - sessions: &[SessionSummary], + sessions: &[SessionSnapshot], selected: usize, focused: bool, ) { @@ -990,7 +1055,7 @@ fn active_columns(tier: LayoutTier) -> Vec> { fn render_active( frame: &mut ratatui::Frame<'_>, area: Rect, - active: &[ActiveRequest], + active: &[ActiveSnapshot], tick: usize, ) { if active.is_empty() { @@ -1144,7 +1209,7 @@ fn http_code_cell(status: Option) -> Cell<'static> { fn render_recent( frame: &mut ratatui::Frame<'_>, area: Rect, - recent: &[CompletedRequest], + recent: &[CompletedSnapshot], selected: usize, focused: bool, ) { @@ -1258,7 +1323,7 @@ fn event_columns(tier: LayoutTier) -> Vec> { } } -fn render_events(frame: &mut ratatui::Frame<'_>, area: Rect, recent: &[CompletedRequest]) { +fn render_events(frame: &mut ratatui::Frame<'_>, area: Rect, recent: &[CompletedSnapshot]) { let events = recent .iter() .filter(|request| { @@ -1312,7 +1377,7 @@ fn render_events(frame: &mut ratatui::Frame<'_>, area: Rect, recent: &[Completed fn render_session_detail( frame: &mut ratatui::Frame<'_>, area: Rect, - state: &MonitorState, + state: &MonitorSnapshot, selected: usize, ) { let lines = if let Some(session) = state.sessions.get(selected) { @@ -1372,7 +1437,7 @@ fn render_session_detail( fn render_request_detail( frame: &mut ratatui::Frame<'_>, area: Rect, - state: &MonitorState, + state: &MonitorSnapshot, selected: usize, ) { let lines = if let Some(request) = state.recent.get(selected) { @@ -1465,11 +1530,18 @@ fn detail_line<'a>(label: &'static str, value: impl Into, value_color: C ]) } -fn render_footer(frame: &mut ratatui::Frame<'_>, area: Rect, _app: &MonitorApp) { +fn render_footer(frame: &mut ratatui::Frame<'_>, area: Rect, app: &MonitorApp) { let spans = vec![ Span::raw(" "), Span::styled("q", Style::default().fg(TEAL)), - Span::styled(" quit ", Style::default().fg(DIM)), + Span::styled( + if app.is_attached() { + " detach " + } else { + " quit " + }, + Style::default().fg(DIM), + ), Span::styled("?", Style::default().fg(TEAL)), Span::styled(" help ", Style::default().fg(DIM)), Span::styled("b", Style::default().fg(TEAL)), @@ -1560,7 +1632,7 @@ fn render_shutdown_overlay(frame: &mut ratatui::Frame<'_>, area: Rect, tick: usi ); } -fn render_help_overlay(frame: &mut ratatui::Frame<'_>, area: Rect) { +fn render_help_overlay(frame: &mut ratatui::Frame<'_>, area: Rect, attached: bool) { let width = 48.min(area.width.saturating_sub(4)).max(24); let height = 12.min(area.height.saturating_sub(2)).max(8); let popup = Rect { @@ -1579,7 +1651,14 @@ fn render_help_overlay(frame: &mut ratatui::Frame<'_>, area: Rect) { let inner = block.inner(popup); frame.render_widget(block, popup); let lines = [ - ("q / Ctrl-C", "quit proxy"), + ( + "q / Ctrl-C", + if attached { + "detach dashboard" + } else { + "quit proxy" + }, + ), ("?", "toggle help"), ("b", "toggle setup"), ("arrows", "navigate rows and panes"), @@ -1893,7 +1972,7 @@ mod tests { #[test] fn active_table_renders_expected_headers_at_tier_boundaries() { - let state = mock_state(); + let state: MonitorSnapshot = mock_state().into(); let render_at = |width| { let buffer = draw(width, 8, |frame| { render_active(frame, frame.area(), &state.active, 0) @@ -2131,7 +2210,7 @@ mod tests { monitor.provider_selected(&request_id, "codex", "gpt-5.6-sol", None); monitor.request_completed(&request_id, 200, Some(100), Some(tokens)); } - let state = monitor.snapshot(); + let state: MonitorSnapshot = monitor.snapshot().into(); let render_at = |width| { let buffer = draw(width, 8, |frame| { render_sessions(frame, frame.area(), &state.sessions, 0, true) @@ -2201,7 +2280,7 @@ mod tests { let monitor = MonitorHandle::new(10); monitor.request_started("request-1", None, None, EndpointKind::Messages); monitor.upstream_started("request-1"); - let state = monitor.snapshot(); + let state: MonitorSnapshot = monitor.snapshot().into(); let active = draw(88, 6, |frame| { render_active(frame, frame.area(), &state.active, 0) @@ -2216,7 +2295,7 @@ mod tests { let monitor = MonitorHandle::new(10); monitor.request_started("request-1", None, None, EndpointKind::Messages); monitor.compaction_started("request-1"); - let state = monitor.snapshot(); + let state: MonitorSnapshot = monitor.snapshot().into(); let active = draw(88, 6, |frame| { render_active(frame, frame.area(), &state.active, 0) @@ -2243,7 +2322,7 @@ mod tests { "gpt-5.6-sol", Some("high".to_string()), ); - let active_state = monitor.snapshot(); + let active_state: MonitorSnapshot = monitor.snapshot().into(); let sessions = draw(170, 8, |frame| { render_sessions(frame, frame.area(), &active_state.sessions, 0, true) @@ -2264,7 +2343,7 @@ mod tests { assert!(!active_text.contains("No active requests")); monitor.request_completed("request-1", 200, Some(100), Some(25)); - let completed_state = monitor.snapshot(); + let completed_state: MonitorSnapshot = monitor.snapshot().into(); let recent = draw(140, 8, |frame| { render_recent(frame, frame.area(), &completed_state.recent, 0, false) }); @@ -2281,7 +2360,7 @@ mod tests { #[test] fn selected_rows_scroll_into_table_viewports() { - let state = mock_state(); + let state: MonitorSnapshot = mock_state().into(); let sessions = (0..12) .map(|index| { let mut session = state.sessions[0].clone(); @@ -2313,7 +2392,7 @@ mod tests { #[test] fn mock_state_renders_representative_panes_at_wide_width() { - let state = mock_state(); + let state: MonitorSnapshot = mock_state().into(); let mut app = MonitorApp { listen_url: "mock://tui-demo".to_string(), setup_text: String::new(), @@ -2341,7 +2420,7 @@ mod tests { #[test] fn mock_request_detail_exposes_error_and_capture_fields() { - let state = mock_state(); + let state: MonitorSnapshot = mock_state().into(); let failed = state .recent .iter() @@ -2364,7 +2443,7 @@ mod tests { monitor.request_started("request-1", None, None, EndpointKind::Messages); monitor.provider_selected("request-1", "codex", "gpt-5.6-sol", None); monitor.request_failed("request-1", Some(502), "upstream unavailable"); - let state = monitor.snapshot(); + let state: MonitorSnapshot = monitor.snapshot().into(); let recent = draw(110, 8, |frame| { render_recent(frame, frame.area(), &state.recent, 0, true) @@ -2385,7 +2464,7 @@ mod tests { monitor.request_started("request-1", None, None, EndpointKind::Messages); monitor.provider_selected("request-1", "codex", "gpt-5.6-sol", None); monitor.request_failed("request-1", Some(502), "upstream unavailable"); - let state = monitor.snapshot(); + let state: MonitorSnapshot = monitor.snapshot().into(); let recent = draw(180, 8, |frame| { render_recent(frame, frame.area(), &state.recent, 0, false) @@ -2415,7 +2494,7 @@ mod tests { Some("high".to_string()), ); monitor.request_failed("request-1", Some(502), "upstream unavailable"); - let state = monitor.snapshot(); + let state: MonitorSnapshot = monitor.snapshot().into(); let detail = draw(120, 20, |frame| { render_request_detail(frame, frame.area(), &state, 0) @@ -2436,7 +2515,7 @@ mod tests { let monitor = MonitorHandle::new(10); monitor.request_started("request-1", None, None, EndpointKind::Messages); monitor.request_failed("request-1", Some(502), "upstream unavailable"); - let state = monitor.snapshot(); + let state: MonitorSnapshot = monitor.snapshot().into(); let events = draw(100, 8, |frame| { render_events(frame, frame.area(), &state.recent) @@ -2473,7 +2552,7 @@ mod tests { Err(oneshot::error::TryRecvError::Empty) )); - let state = MonitorHandle::default().snapshot(); + let state: MonitorSnapshot = MonitorHandle::default().snapshot().into(); let screen = draw(80, 24, |frame| render(frame, &mut app, &state)); let text = buffer_text(&screen); assert!(text.contains("Shut down proxy?"), "{text}"); @@ -2517,7 +2596,7 @@ mod tests { assert_eq!(app.phase, MonitorPhase::ShuttingDown); assert_eq!(shutdown_rx.try_recv(), Ok(())); - let state = MonitorHandle::default().snapshot(); + let state: MonitorSnapshot = MonitorHandle::default().snapshot().into(); let screen = draw(80, 24, |frame| render(frame, &mut app, &state)); let text = buffer_text(&screen); assert!(text.contains("Shutting down...")); @@ -2569,7 +2648,7 @@ mod tests { shutdown: None, shutdown_complete: Some(mpsc::channel().1), }; - let state = MonitorHandle::default().snapshot(); + let state: MonitorSnapshot = MonitorHandle::default().snapshot().into(); let header = draw(100, 1, |frame| { render_header(frame, frame.area(), &app, &state) diff --git a/tests/remote_monitor.rs b/tests/remote_monitor.rs new file mode 100644 index 00000000..e3e5a46a --- /dev/null +++ b/tests/remote_monitor.rs @@ -0,0 +1,258 @@ +use std::{ + net::SocketAddr, + sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use axum::{ + Json, Router, + body::Body, + extract::ConnectInfo, + http::{Request, StatusCode}, + response::IntoResponse, + routing::get, +}; +use claude_code_proxy::{ + config::AliasProvider, + monitor::{ + EndpointKind, MonitorHandle, + remote::RemoteMonitor, + snapshot::{MonitorResponse, PROTOCOL_VERSION}, + }, + registry::Registry, + server::{AppFeatures, app_with_features}, +}; +use http_body_util::BodyExt; +use tokio::{net::TcpListener, task::JoinHandle}; +use tower::ServiceExt; + +struct Server { + url: reqwest::Url, + task: JoinHandle<()>, +} + +impl Drop for Server { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn serve(app: Router) -> Server { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/", listener.local_addr().unwrap()) + .parse() + .unwrap(); + let task = tokio::spawn(async move { + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); + }); + Server { url, task } +} + +fn client() -> reqwest::Client { + reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(2)) + .build() + .unwrap() +} + +fn app(monitor: Option) -> Router { + app_with_features( + Arc::new(Registry::from_providers(AliasProvider::Codex, [])), + monitor, + AppFeatures { + responses_api: false, + images_api: false, + transcriptions_api: false, + }, + ) +} + +#[tokio::test] +async fn snapshot_access_requires_a_real_loopback_peer() { + let monitor = MonitorHandle::default(); + monitor.request_started( + "r1", + Some("session".into()), + Some(1), + EndpointKind::Messages, + ); + for (peer, expected) in [ + (None, StatusCode::FORBIDDEN), + (Some("192.0.2.1:1111"), StatusCode::FORBIDDEN), + (Some("127.0.0.1:1111"), StatusCode::OK), + (Some("[::1]:1111"), StatusCode::OK), + ] { + let request = Request::get("/monitor").header("x-forwarded-for", "127.0.0.1"); + let request = match peer { + Some(peer) => request.extension(ConnectInfo(peer.parse::().unwrap())), + None => request, + } + .body(Body::empty()) + .unwrap(); + let response = app(Some(monitor.clone())).oneshot(request).await.unwrap(); + assert_eq!(response.status(), expected); + if expected == StatusCode::OK { + assert_eq!(response.headers()["cache-control"], "no-store"); + let bytes = response.into_body().collect().await.unwrap().to_bytes(); + let response: MonitorResponse = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(response.version, PROTOCOL_VERSION); + assert_eq!(response.snapshot.active[0].request_id, "r1"); + } + } +} + +#[tokio::test] +async fn independent_dashboards_observe_existing_history_and_detach_without_stopping_proxy() { + let monitor = MonitorHandle::default(); + monitor.request_started( + "before-attach", + Some("session".into()), + Some(1), + EndpointKind::Messages, + ); + monitor.request_completed("before-attach", 200, Some(13), Some(7)); + let server = serve(app(Some(monitor.clone()))).await; + let first = RemoteMonitor::connect(client(), server.url.clone()) + .await + .unwrap(); + let second = RemoteMonitor::connect(client(), server.url.clone()) + .await + .unwrap(); + assert_eq!( + first.snapshot().snapshot().recent[0].request_id, + "before-attach" + ); + assert_eq!( + second.snapshot().snapshot().recent[0].output_tokens, + Some(7) + ); + drop(first); + monitor.request_started( + "after-detach", + Some("session".into()), + Some(2), + EndpointKind::Messages, + ); + wait_until(|| { + second + .snapshot() + .snapshot() + .active + .iter() + .any(|request| request.request_id == "after-detach") + }) + .await; + drop(second); + assert_eq!( + client() + .get(server.url.join("healthz").unwrap()) + .send() + .await + .unwrap() + .status(), + StatusCode::OK + ); + let reattached = RemoteMonitor::connect(client(), server.url.clone()) + .await + .unwrap(); + assert_eq!( + reattached.snapshot().snapshot().active[0].request_id, + "after-detach" + ); +} + +async fn wait_until(mut predicate: impl FnMut() -> bool) { + tokio::time::timeout(Duration::from_secs(4), async { + while !predicate() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn polling_keeps_last_snapshot_on_failure_and_recovers() { + let failing = Arc::new(AtomicBool::new(false)); + let requests = Arc::new(AtomicUsize::new(0)); + let monitor = MonitorHandle::default(); + monitor.request_started("r1", None, None, EndpointKind::Messages); + let server = serve(Router::new().route( + "/monitor", + get({ + let failing = failing.clone(); + let requests = requests.clone(); + let monitor = monitor.clone(); + move || { + let failed = failing.load(Ordering::SeqCst); + requests.fetch_add(1, Ordering::SeqCst); + let snapshot = MonitorResponse::from(monitor.snapshot()); + async move { + if failed { + StatusCode::SERVICE_UNAVAILABLE.into_response() + } else { + Json(snapshot).into_response() + } + } + } + }), + )) + .await; + let remote = RemoteMonitor::connect(client(), server.url.clone()) + .await + .unwrap(); + failing.store(true, Ordering::SeqCst); + wait_until(|| remote.snapshot().connection_error().is_some()).await; + assert_eq!(remote.snapshot().snapshot().active[0].request_id, "r1"); + monitor.request_completed("r1", 200, Some(10), Some(5)); + failing.store(false, Ordering::SeqCst); + wait_until(|| { + remote.snapshot().connection_error().is_none() + && remote.snapshot().snapshot().recent.len() == 1 + }) + .await; + assert_eq!( + remote.snapshot().snapshot().recent[0].output_tokens, + Some(5) + ); + drop(remote); + tokio::time::sleep(Duration::from_millis(100)).await; + let after_drop = requests.load(Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(350)).await; + assert_eq!(requests.load(Ordering::SeqCst), after_drop); +} + +#[tokio::test] +async fn incompatible_protocol_and_disabled_collection_report_attach_errors() { + let incompatible = serve(Router::new().route( + "/monitor", + get(|| async { + let snapshot = MonitorResponse { + version: PROTOCOL_VERSION + 1, + snapshot: MonitorHandle::default().snapshot().into(), + }; + Json(snapshot) + }), + )) + .await; + let error = RemoteMonitor::connect(client(), incompatible.url.clone()) + .await + .err() + .unwrap(); + assert!(format!("{error:#}").contains("unsupported monitor protocol")); + let disabled = serve(app(None)).await; + let error = RemoteMonitor::connect(client(), disabled.url.clone()) + .await + .err() + .unwrap(); + assert!(format!("{error:#}").contains("404")); +}