727 lines
26 KiB
Rust
727 lines
26 KiB
Rust
//! Daemon-side HTTP API for clients (NAP-243).
|
|
//!
|
|
//! These routes live on the daemon's `bind_addr` (default `127.0.0.1:7070`).
|
|
//! They're the contract the browser SPA, the TUI, and the CLI call against.
|
|
//!
|
|
//! ## Trust model
|
|
//!
|
|
//! Whoever can reach `bind_addr` can use these routes. No JWT, no cookie. The
|
|
//! default `127.0.0.1` bind makes this same-trust as tmux (one-user-wide).
|
|
//! Operators who bind a LAN/VPN address opt into one-segment-wide trust — the
|
|
//! perimeter is theirs to choose, Maestro does not police it.
|
|
|
|
use std::sync::Arc;
|
|
|
|
use axum::extract::{Path, Query, State};
|
|
use axum::http::StatusCode;
|
|
use axum::response::{IntoResponse, Response};
|
|
use axum::routing::{any, delete, get, post};
|
|
use axum::{Json, Router};
|
|
use maestro_config::SpawnPreset;
|
|
use maestro_protocol::{
|
|
routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, NameBatch,
|
|
RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session, SetFocusRequest,
|
|
StructuredApprovalDecision, StructuredConfigRequest, StructuredEvent, StructuredPromptRequest,
|
|
PROTOCOL_VERSION,
|
|
};
|
|
use serde::Deserialize;
|
|
use uuid::Uuid;
|
|
|
|
use crate::session::{QueueAcpPrompt, SessionError, SessionManager, SetAcpConfig};
|
|
use crate::sidecar::EventCursor;
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct LocalApiState {
|
|
pub manager: SessionManager,
|
|
pub hostname: Arc<String>,
|
|
/// Home directory of the OS user this daemon runs as, advertised so clients
|
|
/// can prefill a spawn form's working directory. Empty if unresolved.
|
|
pub home_dir: Arc<String>,
|
|
/// Resolved spawn presets (#22/#26) advertised in `GET /api/info`.
|
|
pub presets: Arc<Vec<SpawnPreset>>,
|
|
/// Project discovery (#54) backing `GET /api/projects`.
|
|
pub projects: Arc<crate::projects::ProjectScanner>,
|
|
/// Configured leader key (`[tui].leader`) advertised in `GET /api/info`
|
|
/// (#159), so the web client binds the chord the operator configured
|
|
/// rather than its own default.
|
|
pub leader: Arc<String>,
|
|
}
|
|
|
|
pub fn router(state: LocalApiState) -> Router {
|
|
#[allow(unused_mut)]
|
|
let mut router = Router::new()
|
|
.route(routes::SESSIONS, get(list_sessions).post(create_session))
|
|
.route(routes::patterns::SESSION_ACTIVITY, post(report_activity))
|
|
.route(routes::patterns::SESSION_INPUT, post(send_input))
|
|
.route(routes::patterns::SESSION_FOCUS, post(set_focus))
|
|
.route(
|
|
routes::patterns::SESSION,
|
|
delete(kill_session).patch(rename_session),
|
|
)
|
|
.route(routes::DIRS, get(list_dir))
|
|
.route(routes::RECENT_DIRS, get(recent_dirs))
|
|
.route(routes::PROJECTS, get(projects))
|
|
.route(routes::GITROOT, get(git_root))
|
|
.route(routes::BRANCHES, get(crate::branches::handler))
|
|
.route(routes::NAMES, get(names))
|
|
.route(routes::INFO, get(daemon_info))
|
|
.route(routes::EVENTS, any(events_ws))
|
|
.route(routes::patterns::ATTACH, any(attach_ws))
|
|
.route(routes::patterns::STRUCTURED, any(structured_ws))
|
|
.route(routes::patterns::STRUCTURED_REPLAY, get(structured_replay))
|
|
.route(routes::patterns::STRUCTURED_PROMPT, post(structured_prompt))
|
|
.route(routes::patterns::STRUCTURED_CANCEL, post(structured_cancel))
|
|
.route(routes::patterns::STRUCTURED_CONFIG, post(structured_config))
|
|
.route(
|
|
routes::patterns::STRUCTURED_APPROVAL,
|
|
post(structured_approval),
|
|
);
|
|
#[cfg(feature = "haiku-autoname")]
|
|
{
|
|
router = router.route(routes::patterns::SESSION_PROMPT, post(session_prompt));
|
|
}
|
|
with_web_ui(router.with_state(state))
|
|
}
|
|
|
|
/// Mount the embedded SPA under the same listener as the API, so the browser
|
|
/// talks to the daemon over one connection instead of through a proxy.
|
|
///
|
|
/// The fallback must not answer an unmatched `/api/...` path with the SPA
|
|
/// shell: a client asking for JSON would get HTML and a confusing parse error
|
|
/// instead of a 404.
|
|
#[cfg(feature = "web")]
|
|
fn with_web_ui(router: Router) -> Router {
|
|
use axum::http::Uri;
|
|
|
|
async fn serve_spa(uri: Uri) -> Response {
|
|
let path = uri.path();
|
|
if path == routes::API_PREFIX || path.starts_with(&format!("{}/", routes::API_PREFIX)) {
|
|
return StatusCode::NOT_FOUND.into_response();
|
|
}
|
|
crate::static_bundle::serve_path(path)
|
|
}
|
|
|
|
router.fallback(serve_spa)
|
|
}
|
|
|
|
#[cfg(not(feature = "web"))]
|
|
fn with_web_ui(router: Router) -> Router {
|
|
router
|
|
}
|
|
|
|
/// `POST /api/sessions/:id/prompt` (#131): an agent reports a prompt it was
|
|
/// just given. The first one a provisionally-named session receives starts a
|
|
/// background naming run; every later one, and every prompt to a session
|
|
/// already named, is accepted and dropped. Answers `204 No Content` either
|
|
/// way — the hook posting this has nothing useful to do with a failure, and
|
|
/// naming is never the point of the turn.
|
|
#[cfg(feature = "haiku-autoname")]
|
|
async fn session_prompt(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<maestro_protocol::SessionPromptRequest>,
|
|
) -> Response {
|
|
match s.manager.claim_autoname(id).await {
|
|
Ok(true) => {
|
|
crate::autoname::spawn_naming(s.manager.clone(), id, body.prompt);
|
|
StatusCode::NO_CONTENT.into_response()
|
|
}
|
|
Ok(false) => StatusCode::NO_CONTENT.into_response(),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(
|
|
session_error_status(&info.code),
|
|
info.code.as_str(),
|
|
info.message.as_str(),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn list_sessions(State(s): State<LocalApiState>) -> Json<Vec<Session>> {
|
|
Json(s.manager.snapshot().await)
|
|
}
|
|
|
|
/// Query string for `GET /api/dirs`. `path` is the directory to list (the
|
|
/// parent of what the user is typing); empty/`~`/relative are resolved against
|
|
/// the daemon's home dir.
|
|
#[derive(Debug, Deserialize)]
|
|
struct DirQuery {
|
|
#[serde(default)]
|
|
path: String,
|
|
}
|
|
|
|
/// List one directory's immediate subdirectories, for the spawn form's
|
|
/// working-directory autocomplete (NAP-329).
|
|
async fn list_dir(State(s): State<LocalApiState>, Query(q): Query<DirQuery>) -> Response {
|
|
match crate::dirlist::list_dir(&s.home_dir, &q.path) {
|
|
Ok(listing) => Json(listing).into_response(),
|
|
Err(info) => error_response(dir_error_status(&info.code), &info.code, &info.message),
|
|
}
|
|
}
|
|
|
|
/// HTTP status for a local [`crate::dirlist`] error code.
|
|
fn dir_error_status(code: &str) -> StatusCode {
|
|
match code {
|
|
"not_found" => StatusCode::NOT_FOUND,
|
|
"permission_denied" => StatusCode::FORBIDDEN,
|
|
"not_a_directory" => StatusCode::BAD_REQUEST,
|
|
_ => StatusCode::BAD_GATEWAY,
|
|
}
|
|
}
|
|
|
|
/// HTTP status for a [`crate::session::SessionError`] code. Everything used to
|
|
/// collapse to 502, which reports a client mistake as a daemon failure.
|
|
fn session_error_status(code: &str) -> StatusCode {
|
|
match code {
|
|
"not_found" => StatusCode::NOT_FOUND,
|
|
"invalid_env" => StatusCode::BAD_REQUEST,
|
|
// A guard refusal is the daemon doing its job, not a failure (#79).
|
|
"kill_guarded" => StatusCode::CONFLICT,
|
|
// The session outlived its pane: nothing to type into, and nothing
|
|
// the daemon can do about it.
|
|
"not_running" => StatusCode::CONFLICT,
|
|
"not_terminal" => StatusCode::CONFLICT,
|
|
"shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
|
|
_ => StatusCode::BAD_GATEWAY,
|
|
}
|
|
}
|
|
|
|
/// Recent working dirs for the spawn form's recents picker (#26).
|
|
async fn recent_dirs(State(s): State<LocalApiState>) -> Response {
|
|
Json(crate::spawn_history::to_wire(s.manager.recent_dirs())).into_response()
|
|
}
|
|
|
|
/// Discovered git repositories and worktrees, newest activity first (#54).
|
|
/// `last_activity` is the later of the repo's git HEAD mtime and the newest
|
|
/// spawn-history use of any directory inside it.
|
|
async fn projects(State(s): State<LocalApiState>) -> Response {
|
|
let repos = s.projects.repos().await;
|
|
let history = s.manager.all_recent_dirs();
|
|
Json(crate::projects::to_projects(&repos, &history)).into_response()
|
|
}
|
|
|
|
/// Probe the enclosing git repo of `path`, to infer the worktree `base_repo`
|
|
/// for the spawn form (#26). `path` is resolved against the daemon's home dir
|
|
/// (empty/`~`/relative), like [`list_dir`].
|
|
async fn git_root(State(s): State<LocalApiState>, Query(q): Query<DirQuery>) -> Response {
|
|
let dir = crate::dirlist::resolve(&s.home_dir, &q.path);
|
|
Json(GitRoot {
|
|
root: crate::gitroot::find_root(&dir),
|
|
})
|
|
.into_response()
|
|
}
|
|
|
|
/// Query string for `GET /api/names`. `count` defaults to 1 and is capped at
|
|
/// [`maestro_names::MAX_BATCH`] (#53).
|
|
#[derive(Debug, Deserialize)]
|
|
struct NamesQuery {
|
|
count: Option<usize>,
|
|
}
|
|
|
|
/// Fresh session handles for the spawn form's instant reroll (#53). The web
|
|
/// client fetches one batch per spawn form and rerolls locally from it.
|
|
async fn names(Query(q): Query<NamesQuery>) -> Json<NameBatch> {
|
|
let count = q.count.unwrap_or(1).clamp(1, maestro_names::MAX_BATCH);
|
|
Json(NameBatch {
|
|
names: maestro_names::generate_batch(count),
|
|
})
|
|
}
|
|
|
|
async fn daemon_info(State(s): State<LocalApiState>) -> Json<DaemonInfo> {
|
|
Json(DaemonInfo {
|
|
hostname: (*s.hostname).clone(),
|
|
version: PROTOCOL_VERSION.into(),
|
|
home_dir: (*s.home_dir).clone(),
|
|
presets: (*s.presets).clone(),
|
|
build: crate::release::build_info(),
|
|
instance: crate::release::instance_id().to_string(),
|
|
leader: (*s.leader).clone(),
|
|
})
|
|
}
|
|
|
|
async fn create_session(
|
|
State(s): State<LocalApiState>,
|
|
Json(body): Json<CreateSessionRequest>,
|
|
) -> Response {
|
|
match s.manager.create(body).await {
|
|
Ok(session) => (StatusCode::CREATED, Json(session)).into_response(),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(
|
|
session_error_status(&info.code),
|
|
info.code.as_str(),
|
|
info.message.as_str(),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// `kill` already reports a missing session as `SessionError::NotFound`, which
|
|
/// maps to 404. Checking first would only add a window in which the liveness
|
|
/// watcher reaps the session between the two awaits, turning a 404 into a 502.
|
|
///
|
|
/// The full error info is passed through rather than rebuilt from
|
|
/// code + message: a `kill_guarded` refusal (#79) carries the unsaved-work
|
|
/// breakdown clients render as "3 uncommitted files, 2 unpushed commits".
|
|
///
|
|
/// With `force=true` (#81) the refusal is overridden and the worktree is left
|
|
/// on disk rather than removed. Either way the answer is `204`.
|
|
#[derive(Debug, Deserialize)]
|
|
struct KillQuery {
|
|
#[serde(default)]
|
|
force: bool,
|
|
}
|
|
|
|
async fn kill_session(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Query(q): Query<KillQuery>,
|
|
) -> Response {
|
|
match s.manager.kill(id, q.force).await {
|
|
Ok(()) => StatusCode::NO_CONTENT.into_response(),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
let status = session_error_status(&info.code);
|
|
(status, Json(ApiError { error: info })).into_response()
|
|
}
|
|
}
|
|
}
|
|
|
|
/// `PATCH /api/sessions/:id` (#125): rename a session. The name is trimmed
|
|
/// and capped server-side; a blank one clears `metadata.name` so the session
|
|
/// falls back to its command. Answers with the updated session, which also
|
|
/// goes out as `SessionUpdated` to every events subscriber.
|
|
async fn rename_session(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<RenameSessionRequest>,
|
|
) -> Response {
|
|
match s.manager.rename(id, &body.name).await {
|
|
Ok(session) => (StatusCode::OK, Json(session)).into_response(),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(
|
|
session_error_status(&info.code),
|
|
info.code.as_str(),
|
|
info.message.as_str(),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// `POST /api/sessions/:id/activity` (#105): an agent reports a terminal
|
|
/// session's state. Structured sessions acknowledge and ignore these reports
|
|
/// because their activity comes from ACP. Unknown session is 404; an unknown
|
|
/// state value is rejected by the JSON decoder (422). Answers `204 No Content`.
|
|
async fn report_activity(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<ReportActivityRequest>,
|
|
) -> Response {
|
|
match s.manager.report_activity(id, body.state, body.detail).await {
|
|
Ok(()) => StatusCode::NO_CONTENT.into_response(),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(
|
|
session_error_status(&info.code),
|
|
info.code.as_str(),
|
|
info.message.as_str(),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// `POST /api/sessions/:id/input`: type text into a session's pane, then
|
|
/// press Enter unless the request turns `submit` off. Answers `204 No
|
|
/// Content`. A session that has exited is refused (409) rather than accepted
|
|
/// into nowhere.
|
|
async fn send_input(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<SendInputRequest>,
|
|
) -> Response {
|
|
match s.manager.send_input(id, &body.text, body.submit).await {
|
|
Ok(()) => StatusCode::NO_CONTENT.into_response(),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(
|
|
session_error_status(&info.code),
|
|
info.code.as_str(),
|
|
info.message.as_str(),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// `POST /api/sessions/:id/focus`: a client reports whether its pane is the
|
|
/// one currently showing this session — read on the next activity tick to
|
|
/// decide whether an `AwaitingInput` glow has been looked at long enough to
|
|
/// clear (docs/SIDEBAR.md). Unknown session is 404. Answers `204 No Content`.
|
|
async fn set_focus(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<SetFocusRequest>,
|
|
) -> Response {
|
|
match s.manager.set_focused(id, body.focused).await {
|
|
Ok(()) => StatusCode::NO_CONTENT.into_response(),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(
|
|
session_error_status(&info.code),
|
|
info.code.as_str(),
|
|
info.message.as_str(),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
fn error_response(status: StatusCode, code: &str, message: &str) -> Response {
|
|
(status, Json(ApiError::new(code, message))).into_response()
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct ReplayQuery {
|
|
since: Option<u64>,
|
|
before: Option<u64>,
|
|
limit: Option<usize>,
|
|
}
|
|
|
|
const MAX_STRUCTURED_REPLAY_LIMIT: usize = 500;
|
|
|
|
async fn structured_replay(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Query(query): Query<ReplayQuery>,
|
|
) -> Response {
|
|
if query.since.is_some() && query.before.is_some() {
|
|
return error_response(
|
|
StatusCode::BAD_REQUEST,
|
|
"invalid_cursor",
|
|
"use either since or before, not both",
|
|
);
|
|
}
|
|
let cursor = match query.before {
|
|
Some(before) => EventCursor::Before(Some(before)),
|
|
None => EventCursor::Since(query.since.unwrap_or(0)),
|
|
};
|
|
match s
|
|
.manager
|
|
.read_acp_projection_replay(
|
|
id,
|
|
cursor,
|
|
query
|
|
.limit
|
|
.unwrap_or(100)
|
|
.clamp(1, MAX_STRUCTURED_REPLAY_LIMIT),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Some(events)) => Json(events).into_response(),
|
|
// A terminal session, or a structured session whose adapter is no
|
|
// longer live, cannot provide a coherent replay continuation.
|
|
Ok(None) => error_response(
|
|
StatusCode::CONFLICT,
|
|
"not_structured",
|
|
"session is not a live structured session",
|
|
),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(session_error_status(&info.code), &info.code, &info.message)
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn structured_prompt(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<StructuredPromptRequest>,
|
|
) -> Response {
|
|
match s.manager.queue_acp_prompt(id, body.prompt).await {
|
|
Ok(QueueAcpPrompt::Queued) => StatusCode::NO_CONTENT.into_response(),
|
|
Ok(QueueAcpPrompt::InFlight) => error_response(
|
|
StatusCode::CONFLICT,
|
|
"turn_in_flight",
|
|
"the structured session already has an active turn",
|
|
),
|
|
Ok(QueueAcpPrompt::Unavailable) => error_response(
|
|
StatusCode::CONFLICT,
|
|
"not_structured",
|
|
"session is not a live structured session",
|
|
),
|
|
Ok(QueueAcpPrompt::Failed(message)) => {
|
|
error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
|
|
}
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(session_error_status(&info.code), &info.code, &info.message)
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn structured_cancel(State(s): State<LocalApiState>, Path(id): Path<Uuid>) -> Response {
|
|
match s.manager.cancel_acp_turn(id).await {
|
|
Ok(true) => StatusCode::NO_CONTENT.into_response(),
|
|
Ok(false) => error_response(
|
|
StatusCode::CONFLICT,
|
|
"not_structured",
|
|
"session is not a live structured session",
|
|
),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(session_error_status(&info.code), &info.code, &info.message)
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn structured_config(
|
|
State(s): State<LocalApiState>,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<StructuredConfigRequest>,
|
|
) -> Response {
|
|
match s
|
|
.manager
|
|
.set_acp_config(id, body.config_id, body.value)
|
|
.await
|
|
{
|
|
Ok(SetAcpConfig::Updated) => StatusCode::NO_CONTENT.into_response(),
|
|
Ok(SetAcpConfig::InvalidValue) => error_response(
|
|
StatusCode::BAD_REQUEST,
|
|
"invalid_config_value",
|
|
"configuration values must be strings or booleans",
|
|
),
|
|
Ok(SetAcpConfig::Unavailable) => error_response(
|
|
StatusCode::CONFLICT,
|
|
"not_structured",
|
|
"session is not a live structured session",
|
|
),
|
|
Ok(SetAcpConfig::Failed(message)) => {
|
|
error_response(StatusCode::BAD_GATEWAY, "acp_config_failed", &message)
|
|
}
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(session_error_status(&info.code), &info.code, &info.message)
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn structured_approval(
|
|
State(s): State<LocalApiState>,
|
|
Path((id, nonce)): Path<(Uuid, String)>,
|
|
Json(decision): Json<StructuredApprovalDecision>,
|
|
) -> Response {
|
|
let decision = match decision {
|
|
StructuredApprovalDecision::AllowOnce => crate::session::ApprovalDecision::AllowOnce,
|
|
StructuredApprovalDecision::AllowForSession => {
|
|
crate::session::ApprovalDecision::AllowForSession
|
|
}
|
|
StructuredApprovalDecision::Deny => crate::session::ApprovalDecision::Deny,
|
|
};
|
|
match s.manager.resolve_acp_approval(id, &nonce, decision).await {
|
|
Ok(Some(())) => StatusCode::NO_CONTENT.into_response(),
|
|
Ok(None) => error_response(
|
|
StatusCode::CONFLICT,
|
|
"not_structured",
|
|
"session is not a live structured session",
|
|
),
|
|
Err(e) => {
|
|
let info = e.to_error_info();
|
|
error_response(session_error_status(&info.code), &info.code, &info.message)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Convenience constructor — pulls the bits `LocalApiState` needs from
|
|
/// `DaemonConfig` so callers don't have to assemble it by hand. The project
|
|
/// scanner starts with its cache cold; wire
|
|
/// [`ProjectScanner::start_background_refresh`](crate::projects::ProjectScanner::start_background_refresh)
|
|
/// on the returned handle's scanner to keep it eventually fresh.
|
|
pub fn state_for(
|
|
manager: SessionManager,
|
|
hostname: impl Into<String>,
|
|
home_dir: impl Into<String>,
|
|
presets: Vec<SpawnPreset>,
|
|
projects: Arc<crate::projects::ProjectScanner>,
|
|
leader: impl Into<String>,
|
|
) -> LocalApiState {
|
|
LocalApiState {
|
|
manager,
|
|
hostname: Arc::new(hostname.into()),
|
|
home_dir: Arc::new(home_dir.into()),
|
|
presets: Arc::new(presets),
|
|
projects,
|
|
leader: Arc::new(leader.into()),
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// WS /api/attach/:id — raw PTY bytes for one session
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[allow(clippy::result_large_err)]
|
|
/// Subscribing *before* the upgrade makes the 404 and the subscription one
|
|
/// atomic step. Checking existence and then subscribing inside the callback
|
|
/// left a window in which the liveness watcher could reap the session, so a
|
|
/// client that passed the check still got a socket that immediately closed.
|
|
async fn attach_ws(
|
|
Path(id): Path<Uuid>,
|
|
State(s): State<LocalApiState>,
|
|
ws: axum::extract::WebSocketUpgrade,
|
|
) -> Result<Response, Response> {
|
|
let (replay, rx) = match s.manager.subscribe_pty_with_replay(id).await {
|
|
Ok(subscription) => subscription,
|
|
Err(SessionError::NotFound(_)) => {
|
|
return Err(error_response(
|
|
StatusCode::NOT_FOUND,
|
|
"unknown_session",
|
|
"no such session on this daemon",
|
|
));
|
|
}
|
|
Err(error) => {
|
|
let info = error.to_error_info();
|
|
return Err(error_response(
|
|
session_error_status(&info.code),
|
|
&info.code,
|
|
&info.message,
|
|
));
|
|
}
|
|
};
|
|
let manager = s.manager.clone();
|
|
Ok(ws.on_upgrade(move |socket| async move {
|
|
let _ = crate::attach::serve(socket, manager, id, replay, rx).await;
|
|
}))
|
|
}
|
|
|
|
#[allow(clippy::result_large_err)]
|
|
async fn structured_ws(
|
|
Path(id): Path<Uuid>,
|
|
State(s): State<LocalApiState>,
|
|
ws: axum::extract::WebSocketUpgrade,
|
|
) -> Result<Response, Response> {
|
|
let Some((snapshot, state, rx)) = s
|
|
.manager
|
|
.subscribe_acp_projection(id, EventCursor::Since(0))
|
|
.await
|
|
.map_err(|e| {
|
|
let info = e.to_error_info();
|
|
error_response(session_error_status(&info.code), &info.code, &info.message)
|
|
})?
|
|
else {
|
|
return Err(error_response(
|
|
StatusCode::CONFLICT,
|
|
"not_structured",
|
|
"session is not a live structured session",
|
|
));
|
|
};
|
|
Ok(ws.on_upgrade(move |socket| async move {
|
|
let _ = pump_structured(socket, snapshot, state, rx).await;
|
|
}))
|
|
}
|
|
|
|
async fn pump_structured(
|
|
socket: axum::extract::ws::WebSocket,
|
|
snapshot: maestro_protocol::TranscriptSnapshot,
|
|
state: maestro_protocol::ReducedState,
|
|
mut rx: tokio::sync::broadcast::Receiver<crate::projection::ProjectionUpdate>,
|
|
) -> Result<(), axum::Error> {
|
|
use axum::extract::ws::Message;
|
|
use futures_util::{SinkExt, StreamExt};
|
|
|
|
let (mut sink, mut stream) = socket.split();
|
|
for event in [
|
|
StructuredEvent::TranscriptSnapshot { snapshot },
|
|
StructuredEvent::ReducedState { state },
|
|
] {
|
|
if sink
|
|
.send(Message::Text(
|
|
serde_json::to_string(&event).expect("structured event encode"),
|
|
))
|
|
.await
|
|
.is_err()
|
|
{
|
|
return Ok(());
|
|
}
|
|
}
|
|
loop {
|
|
tokio::select! {
|
|
update = rx.recv() => match update {
|
|
Ok(update) => {
|
|
let event = StructuredEvent::ProjectionUpdate { transcript: update.transcript, state: update.state };
|
|
if sink.send(Message::Text(serde_json::to_string(&event).expect("structured event encode"))).await.is_err() { break; }
|
|
}
|
|
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => break,
|
|
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
|
|
},
|
|
frame = stream.next() => match frame {
|
|
None | Some(Err(_)) | Some(Ok(Message::Close(_))) => break,
|
|
Some(Ok(_)) => {},
|
|
},
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// WS /api/events — live catalog deltas for browser/TUI subscribers (NAP-248)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
async fn events_ws(
|
|
State(s): State<LocalApiState>,
|
|
ws: axum::extract::WebSocketUpgrade,
|
|
) -> Response {
|
|
ws.on_upgrade(move |socket| async move {
|
|
let _ = pump_events(socket, s).await;
|
|
})
|
|
}
|
|
|
|
async fn pump_events(
|
|
socket: axum::extract::ws::WebSocket,
|
|
s: LocalApiState,
|
|
) -> Result<(), axum::Error> {
|
|
use axum::extract::ws::Message;
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use tokio::sync::broadcast::error::RecvError;
|
|
|
|
let (mut sink, mut stream) = socket.split();
|
|
|
|
// Subscribe BEFORE building the snapshot so deltas applied while we're
|
|
// marshaling can't slip between the two and disappear. The UI is
|
|
// idempotent for repeated *Added events, so any overlap is harmless.
|
|
let mut rx = s.manager.events();
|
|
|
|
let snapshot = CatalogEvent::Snapshot {
|
|
sessions: s.manager.snapshot().await,
|
|
};
|
|
let text = serde_json::to_string(&snapshot).expect("snapshot encode");
|
|
if sink.send(Message::Text(text)).await.is_err() {
|
|
return Ok(());
|
|
}
|
|
|
|
loop {
|
|
tokio::select! {
|
|
ev = rx.recv() => {
|
|
let event = match ev {
|
|
Ok(e) => e,
|
|
Err(RecvError::Lagged(n)) => CatalogEvent::Lagged { skipped: n },
|
|
Err(RecvError::Closed) => break,
|
|
};
|
|
let text = serde_json::to_string(&event).expect("event encode");
|
|
if sink.send(Message::Text(text)).await.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
frame = stream.next() => {
|
|
// Server-push only. Anything from the client either closes or
|
|
// is ignored.
|
|
match frame {
|
|
None => break,
|
|
Some(Err(_)) => break,
|
|
Some(Ok(Message::Close(_))) => break,
|
|
Some(Ok(_)) => {}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|