refactor(web_ui): Bus::effective_context_window, sigint_claude + per-connection hello, relocate fetch_reminder_stats, drop redundant guard
This commit is contained in:
parent
77f31b44bb
commit
9cebc128e2
7 changed files with 63 additions and 49 deletions
|
|
@ -901,6 +901,16 @@ impl Bus {
|
||||||
*self.api_context_window.lock().unwrap()
|
*self.api_context_window.lock().unwrap()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The effective context window for `model`: the API-reported window if a
|
||||||
|
/// turn has completed ([`api_context_window`]), else the per-model default
|
||||||
|
/// ([`context_window_tokens`]). Single accessor so the state + dashboard
|
||||||
|
/// endpoints agree by construction.
|
||||||
|
#[must_use]
|
||||||
|
pub fn effective_context_window(&self, model: &str) -> u64 {
|
||||||
|
self.api_context_window()
|
||||||
|
.unwrap_or_else(|| context_window_tokens(model))
|
||||||
|
}
|
||||||
|
|
||||||
/// Walk a stream-json value for `tool_use` blocks and bump the
|
/// Walk a stream-json value for `tool_use` blocks and bump the
|
||||||
/// per-turn counter for each one we find. Called by the stdout
|
/// per-turn counter for each one we find. Called by the stdout
|
||||||
/// pump on every parsed line. Cheap when the line isn't an
|
/// pump on every parsed line. Cheap when the line isn't an
|
||||||
|
|
|
||||||
|
|
@ -44,10 +44,7 @@ pub(super) async fn post_send(
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response {
|
pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response {
|
||||||
let out = tokio::process::Command::new("pkill")
|
let out = super::sigint_claude().await;
|
||||||
.args(["-INT", "claude"])
|
|
||||||
.output()
|
|
||||||
.await;
|
|
||||||
let note = match out {
|
let note = match out {
|
||||||
Ok(o) if o.status.success() => "operator: /cancel — sent SIGINT to claude".to_owned(),
|
Ok(o) if o.status.success() => "operator: /cancel — sent SIGINT to claude".to_owned(),
|
||||||
Ok(o) if o.status.code() == Some(1) => {
|
Ok(o) if o.status.code() == Some(1) => {
|
||||||
|
|
|
||||||
|
|
@ -84,10 +84,7 @@ pub(super) async fn post_login_cancel(State(state): State<AppState>) -> Response
|
||||||
/// preservation invariants.
|
/// preservation invariants.
|
||||||
pub(super) async fn post_logout(State(state): State<AppState>) -> Response {
|
pub(super) async fn post_logout(State(state): State<AppState>) -> Response {
|
||||||
// Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`).
|
// Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`).
|
||||||
let _ = tokio::process::Command::new("pkill")
|
let _ = super::sigint_claude().await;
|
||||||
.args(["-INT", "claude"])
|
|
||||||
.output()
|
|
||||||
.await;
|
|
||||||
// Step 2: delete OAuth credential files only — login::clear_session owns
|
// Step 2: delete OAuth credential files only — login::clear_session owns
|
||||||
// the file set and preserves session-history files alongside them.
|
// the file set and preserves session-history files alongside them.
|
||||||
let dir = crate::paths::claude_dir();
|
let dir = crate::paths::claude_dir();
|
||||||
|
|
|
||||||
|
|
@ -259,6 +259,17 @@ fn read_gui_vnc_port() -> Option<u16> {
|
||||||
std::env::var("HIVE_GUI_VNC_PORT").ok()?.parse().ok()
|
std::env::var("HIVE_GUI_VNC_PORT").ok()?.parse().ok()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// SIGINT any running `claude` process in this container (best-effort). Shared
|
||||||
|
/// by `/api/cancel` and `/api/logout`. Returns the `pkill` `Output` so callers
|
||||||
|
/// can inspect the exit status (0 = signalled, 1 = no process matched) or
|
||||||
|
/// ignore it.
|
||||||
|
async fn sigint_claude() -> std::io::Result<std::process::Output> {
|
||||||
|
tokio::process::Command::new("pkill")
|
||||||
|
.args(["-INT", "claude"])
|
||||||
|
.output()
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
fn error_response(status: StatusCode, message: &str) -> Response {
|
fn error_response(status: StatusCode, message: &str) -> Response {
|
||||||
// Plain text — JS app surfaces in `alert()`, HTML wrapping would just
|
// Plain text — JS app surfaces in `alert()`, HTML wrapping would just
|
||||||
// be noise. Status is per-caller: 400 for bad input, 409 for a
|
// be noise. Status is per-caller: 400 for bad input, 409 for a
|
||||||
|
|
|
||||||
|
|
@ -36,10 +36,7 @@ pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<State
|
||||||
let inbox = recent_inbox(&state.socket).await;
|
let inbox = recent_inbox(&state.socket).await;
|
||||||
let (turn_state, turn_state_since) = state.bus.state_snapshot();
|
let (turn_state, turn_state_since) = state.bus.state_snapshot();
|
||||||
let model = state.bus.model();
|
let model = state.bus.model();
|
||||||
let context_window_tokens = state
|
let context_window_tokens = state.bus.effective_context_window(&model);
|
||||||
.bus
|
|
||||||
.api_context_window()
|
|
||||||
.unwrap_or_else(|| crate::events::context_window_tokens(&model));
|
|
||||||
let ctx_usage = state.bus.last_ctx_usage();
|
let ctx_usage = state.bus.last_ctx_usage();
|
||||||
let cost_usage = state.bus.last_cost_usage();
|
let cost_usage = state.bus.last_cost_usage();
|
||||||
let effort = state.bus.effort();
|
let effort = state.bus.effort();
|
||||||
|
|
@ -87,10 +84,7 @@ pub(super) async fn api_dashboard_state(
|
||||||
let (status_text, status_set_at) = read_own_status();
|
let (status_text, status_set_at) = read_own_status();
|
||||||
let rate_limited = state.bus.is_rate_limited();
|
let rate_limited = state.bus.is_rate_limited();
|
||||||
let model = state.bus.model();
|
let model = state.bus.model();
|
||||||
let context_window_tokens = state
|
let context_window_tokens = state.bus.effective_context_window(&model);
|
||||||
.bus
|
|
||||||
.api_context_window()
|
|
||||||
.unwrap_or_else(|| crate::events::context_window_tokens(&model));
|
|
||||||
// Full context-window size = input + cache-read + cache-creation. Using
|
// Full context-window size = input + cache-read + cache-creation. Using
|
||||||
// raw `input_tokens` here reported only the *uncached* sliver, which is
|
// raw `input_tokens` here reported only the *uncached* sliver, which is
|
||||||
// ~0 once prompt caching kicks in — so every card showed `ctx·0k`. Match
|
// ~0 once prompt caching kicks in — so every card showed `ctx·0k`. Match
|
||||||
|
|
@ -378,38 +372,16 @@ async fn recent_inbox(socket: &std::path::Path) -> Vec<hive_sh4re::InboxRow> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Fetch reminder activity stats from the broker via the per-agent /
|
|
||||||
/// manager socket. Returns None on any transport / decode failure — the
|
|
||||||
/// stats are decorative, not authoritative.
|
|
||||||
pub(super) async fn fetch_reminder_stats(
|
|
||||||
socket: &std::path::Path,
|
|
||||||
window_secs: u64,
|
|
||||||
) -> Option<hive_sh4re::ReminderStats> {
|
|
||||||
match super::broker_request(
|
|
||||||
socket,
|
|
||||||
&hive_sh4re::Request::ReminderRollup {
|
|
||||||
since_secs: window_secs,
|
|
||||||
agent: None,
|
|
||||||
},
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
Ok(hive_sh4re::Response::ReminderRollup(stats)) => Some(stats),
|
|
||||||
_ => None,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Read `HIVE_AVAILABLE_MODELS` (comma-separated short names injected by
|
/// Read `HIVE_AVAILABLE_MODELS` (comma-separated short names injected by
|
||||||
/// `services.hyperhive.availableModels`) and return the parsed list.
|
/// `services.hyperhive.availableModels`) and return the parsed list.
|
||||||
/// Falls back to `["haiku", "sonnet", "opus"]` when the env var is absent
|
/// Falls back to `["haiku", "sonnet", "opus"]` when the env var is absent
|
||||||
/// or resolves to an empty list after trimming.
|
/// or resolves to an empty list after trimming.
|
||||||
fn available_models() -> Vec<String> {
|
fn available_models() -> Vec<String> {
|
||||||
const DEFAULT: &[&str] = &["haiku", "sonnet", "opus"];
|
const DEFAULT: &[&str] = &["haiku", "sonnet", "opus"];
|
||||||
let raw = match std::env::var("HIVE_AVAILABLE_MODELS") {
|
// Absent / empty / all-whitespace env all funnel to the single
|
||||||
Ok(v) if !v.trim().is_empty() => v,
|
// emptiness check below — no separate up-front guard needed.
|
||||||
_ => return DEFAULT.iter().map(ToString::to_string).collect(),
|
let models: Vec<String> = std::env::var("HIVE_AVAILABLE_MODELS")
|
||||||
};
|
.unwrap_or_default()
|
||||||
let models: Vec<String> = raw
|
|
||||||
.split(',')
|
.split(',')
|
||||||
.map(|s| s.trim().to_string())
|
.map(|s| s.trim().to_string())
|
||||||
.filter(|s| !s.is_empty())
|
.filter(|s| !s.is_empty())
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,6 @@ use axum::http::StatusCode;
|
||||||
use axum::response::{IntoResponse, Response};
|
use axum::response::{IntoResponse, Response};
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
|
|
||||||
use super::state::fetch_reminder_stats;
|
|
||||||
use super::{AppState, error_response};
|
use super::{AppState, error_response};
|
||||||
|
|
||||||
#[derive(Deserialize)]
|
#[derive(Deserialize)]
|
||||||
|
|
@ -27,6 +26,27 @@ pub(super) async fn api_stats(
|
||||||
axum::Json(snapshot)
|
axum::Json(snapshot)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Fetch reminder activity stats from the broker via the per-agent / manager
|
||||||
|
/// socket. Returns None on any transport / decode failure — the stats are
|
||||||
|
/// decorative, not authoritative.
|
||||||
|
async fn fetch_reminder_stats(
|
||||||
|
socket: &std::path::Path,
|
||||||
|
window_secs: u64,
|
||||||
|
) -> Option<hive_sh4re::ReminderStats> {
|
||||||
|
match super::broker_request(
|
||||||
|
socket,
|
||||||
|
&hive_sh4re::Request::ReminderRollup {
|
||||||
|
since_secs: window_secs,
|
||||||
|
agent: None,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
Ok(hive_sh4re::Response::ReminderRollup(stats)) => Some(stats),
|
||||||
|
_ => None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Proxy this agent's loose-ends list via the per-agent socket. The
|
/// Proxy this agent's loose-ends list via the per-agent socket. The
|
||||||
/// web UI surfaces the result as a collapsible section in the page
|
/// web UI surfaces the result as a collapsible section in the page
|
||||||
/// so the operator can see at a glance what's pending against the
|
/// so the operator can see at a glance what's pending against the
|
||||||
|
|
|
||||||
|
|
@ -56,15 +56,22 @@ pub(super) async fn events_stream(
|
||||||
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
|
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
|
||||||
tracing::info!("sse: client subscribed");
|
tracing::info!("sse: client subscribed");
|
||||||
let rx = state.bus.subscribe();
|
let rx = state.bus.subscribe();
|
||||||
// Drop a "hello" note into the bus so every new subscriber sees at
|
// Prime THIS connection with a one-off "hello" so it can clear the
|
||||||
// least one event immediately and can clear the connecting placeholder.
|
// connecting placeholder immediately. Injected into this subscriber's own
|
||||||
state.bus.emit(crate::events::LiveEvent::Note {
|
// stream rather than emitted to the bus — a bus emit would spam every
|
||||||
|
// already-connected client with a spurious note each time anyone opens
|
||||||
|
// the stream.
|
||||||
|
let hello = Event::default().data(
|
||||||
|
serde_json::to_string(&crate::events::LiveEvent::Note {
|
||||||
text: "live stream attached".into(),
|
text: "live stream attached".into(),
|
||||||
});
|
})
|
||||||
let stream = BroadcastStream::new(rx).filter_map(|res| {
|
.unwrap_or_default(),
|
||||||
|
);
|
||||||
|
let live = BroadcastStream::new(rx).filter_map(|res| {
|
||||||
let ev = res.ok()?;
|
let ev = res.ok()?;
|
||||||
let json = serde_json::to_string(&ev).ok()?;
|
let json = serde_json::to_string(&ev).ok()?;
|
||||||
Some(Ok(Event::default().data(json)))
|
Some(Ok(Event::default().data(json)))
|
||||||
});
|
});
|
||||||
|
let stream = tokio_stream::once(Ok(hello)).chain(live);
|
||||||
Sse::new(stream).keep_alive(KeepAlive::default())
|
Sse::new(stream).keep_alive(KeepAlive::default())
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue