//! Operator action POST handlers (cancel, compact, reset, todos mark-done). use axum::{ Form, extract::State, http::StatusCode, response::{IntoResponse, Response}, }; use serde::{Deserialize, Serialize}; use std::future::Future; use super::{AppState, SOCKET_FETCH_TIMEOUT, error_response}; pub(super) async fn post_cancel_turn(State(state): State) -> Response { super::cancel_turn(&state.bus, &state.interrupted, "operator: /cancel"); (axum::http::StatusCode::OK, "ok").into_response() } /// Operator-initiated `/compact`. Deferred: sets the `compact_pending` flag /// that `turn::drive_turn` consumes at the end of the current/next turn, so it /// works while a turn is in flight (a mid-turn compaction would race the live /// claude process) rather than only when the agent is idle. Returns 200 /// immediately; the compaction stream lands in the live panel when it runs. pub(super) async fn post_compact(State(state): State) -> Response { // No wake prompt: the operator is watching the dashboard, not waiting on // an inbox message — `request_compact`'s wake-prompt arg exists for the // agent's own `compact` MCP tool (`todo_server.rs::compact`). state.bus.request_compact(None); state.bus.emit(crate::events::LiveEvent::Note { text: "operator: /compact queued — runs at the end of the current turn".into(), }); (axum::http::StatusCode::OK, "ok").into_response() } /// Request a session reset. The current session is archived (its backing /// `.jsonl` renamed out of claude's resolution glob) at the next turn /// boundary, so the following turn's `--resume` misses and self-heals into a /// freshly-named session. History is preserved on disk, not deleted. /// /// Deferred (a one-shot flag consumed by `drive_turn`) rather than applied /// here: renaming the session file while a claude turn is mid-write would /// race the live process. Between turns there is no open session file (one /// claude per container, serialized by the serve loop), so the archive is /// safe there. Useful when the session-resume context is poisoned (claude /// went off the rails, hit an unrecoverable refusal, etc.) and a full reset /// is cheaper than asking claude to forget mid-stream. pub(super) async fn post_new_session(State(state): State) -> Response { state.bus.request_session_reset(); state.bus.emit(crate::events::LiveEvent::Note { text: "operator: session reset queued — takes effect at the next turn".into(), }); (axum::http::StatusCode::OK, "ok").into_response() } /// The most ids one mark-done request may carry. const MAX_MARK_DONE_IDS: usize = 200; #[derive(Deserialize)] pub(super) struct MarkTodosDoneForm { /// Comma-separated todo ids. Same "one field, JS joins the checked /// boxes" shape — axum's `Form` extractor doesn't natively decode /// repeated same-name keys. ids: String, } /// `POST /api/todos/mark-done` — dismiss one or more of this agent's own /// todos (loose-ends v2) from the todos flyout. Loops a `MarkTodoDone` call /// per id over the in-agent socket rather than adding a new bulk request to /// `hive-agent-sock`: the todos list is small (single-digit rows most of the /// time), so N same-host socket round-trips isn't a real cost, and it keeps /// the wire protocol's `Request` enum — already used by the `cancel_loose_end` /// MCP tool — unchanged. Unknown/already-acked ids just don't add to the /// `acked` count (same "acking twice is not a new action" semantics as the /// single-id path); a request with no ids or where every id fails to parse /// is rejected as a client error rather than silently acking nothing. An id /// whose store call errors or gets no answer fails the request with a 500 /// naming how many failed. More than [`MAX_MARK_DONE_IDS`] ids is a client /// error, and the calls share one [`SOCKET_FETCH_TIMEOUT`]: ids the store has /// not answered by then count as unanswered, so a stalled store gets its 500 /// after about 3 seconds like the other handlers. pub(super) async fn post_mark_todos_done(Form(form): Form) -> Response { let ids: Vec = form .ids .split(',') .filter_map(|s| s.trim().parse::().ok()) .collect(); if ids.is_empty() { return error_response(StatusCode::BAD_REQUEST, "mark-done: no todo ids selected"); } if ids.len() > MAX_MARK_DONE_IDS { return error_response( StatusCode::BAD_REQUEST, &format!("mark-done: at most {MAX_MARK_DONE_IDS} todo ids per request"), ); } mark_ids_done(ids, SOCKET_FETCH_TIMEOUT, |id| async move { crate::todo_server::dial(&hive_agent_sock::Request::MarkTodoDone { id }).await }) .await } /// The handler body behind `dial`, which answers one id's `MarkTodoDone` /// (`None` when the store doesn't answer). Tests inject it so they never /// reach the live in-agent socket. async fn mark_ids_done(ids: Vec, within: std::time::Duration, dial: F) -> Response where F: Fn(i64) -> Fut, Fut: Future>, { let deadline = tokio::time::Instant::now() + within; let mut answers = Vec::with_capacity(ids.len()); for id in ids { // Past the deadline every remaining id is an unanswered one. answers.push( tokio::time::timeout_at(deadline, dial(id)) .await .unwrap_or(None), ); } match settle_mark_done(answers) { Ok(acked) => axum::Json(MarkTodosDoneBody { acked }).into_response(), Err(message) => error_response(StatusCode::INTERNAL_SERVER_ERROR, &message), } } /// Folds the per-id answers into the acked total, or the operator-facing /// failure. `None` is a dial that got no answer; `Response::Err` and any /// other variant are failures. `Acked { count: 0 }` is a success. fn settle_mark_done(answers: Vec>) -> Result { let total = answers.len(); let mut acked = 0u64; let mut failed = 0usize; let mut first_error = None; for answer in answers { match answer { Some(hive_agent_sock::Response::Acked { count }) => acked += count, Some(hive_agent_sock::Response::Err { message }) => { failed += 1; first_error.get_or_insert(message); } Some(other) => { failed += 1; first_error.get_or_insert(format!("unexpected response: {other:?}")); } None => { failed += 1; first_error.get_or_insert_with(|| "the todo store did not answer".to_owned()); } } } match first_error { None => Ok(acked), Some(error) => Err(format!( "mark-done: {failed} of {total} failed ({acked} marked done): {error}" )), } } #[derive(Serialize)] struct MarkTodosDoneBody { acked: u64, } #[cfg(test)] mod tests { use super::{mark_ids_done, settle_mark_done}; async fn body_text(response: axum::response::Response) -> String { let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) .await .unwrap(); String::from_utf8(bytes.to_vec()).unwrap() } #[tokio::test] async fn mark_done_handler_answers_500_with_the_message_when_the_store_does_not_answer() { let response = mark_ids_done(vec![1, 2], std::time::Duration::from_secs(3), |_| async { None }) .await; assert_eq!( response.status(), axum::http::StatusCode::INTERNAL_SERVER_ERROR ); let body = body_text(response).await; assert!(body.contains("2 of 2 failed"), "{body}"); assert!(body.contains("did not answer"), "{body}"); } #[tokio::test] async fn mark_done_handler_answers_500_with_the_store_error() { let response = mark_ids_done(vec![7], std::time::Duration::from_secs(3), |_| async { Some(hive_agent_sock::Response::Err { message: "database is locked".to_owned(), }) }) .await; assert_eq!( response.status(), axum::http::StatusCode::INTERNAL_SERVER_ERROR ); assert!(body_text(response).await.contains("database is locked")); } #[tokio::test] async fn mark_done_handler_answers_200_with_the_acked_total() { let response = mark_ids_done(vec![1, 2], std::time::Duration::from_secs(3), |_| async { Some(hive_agent_sock::Response::Acked { count: 1 }) }) .await; assert_eq!(response.status(), axum::http::StatusCode::OK); assert_eq!(body_text(response).await, r#"{"acked":2}"#); } #[tokio::test(start_paused = true)] async fn a_stalled_store_gets_one_deadline_for_all_ids_not_one_per_id() { let started = tokio::time::Instant::now(); let response = mark_ids_done( (1..=50).collect(), std::time::Duration::from_secs(3), |_| async { tokio::time::sleep(std::time::Duration::from_hours(1)).await; None }, ) .await; assert_eq!( response.status(), axum::http::StatusCode::INTERNAL_SERVER_ERROR ); assert!(started.elapsed() < std::time::Duration::from_secs(4)); assert!(body_text(response).await.contains("50 of 50 failed")); } #[tokio::test] async fn more_ids_than_the_cap_are_a_client_error() { let ids = (0..=super::MAX_MARK_DONE_IDS) .map(|id| id.to_string()) .collect::>() .join(","); let response = super::post_mark_todos_done(axum::Form(super::MarkTodosDoneForm { ids })).await; assert_eq!(response.status(), axum::http::StatusCode::BAD_REQUEST); assert!(body_text(response).await.contains("at most 200")); } #[test] fn mark_done_with_no_answer_is_an_error() { let err = settle_mark_done(vec![None]).unwrap_err(); assert!(err.contains("1 of 1 failed"), "{err}"); } #[test] fn mark_done_surfaces_the_store_error() { let answers = vec![Some(hive_agent_sock::Response::Err { message: "database is locked".to_owned(), })]; let err = settle_mark_done(answers).unwrap_err(); assert!(err.contains("database is locked"), "{err}"); } #[test] fn mark_done_partial_failure_reports_both_counts() { let answers = vec![Some(hive_agent_sock::Response::Acked { count: 1 }), None]; let err = settle_mark_done(answers).unwrap_err(); assert!(err.contains("1 of 2 failed (1 marked done)"), "{err}"); } #[test] fn mark_done_acking_nothing_new_is_still_success() { let answers = vec![ Some(hive_agent_sock::Response::Acked { count: 0 }), Some(hive_agent_sock::Response::Acked { count: 2 }), ]; assert_eq!(settle_mark_done(answers).unwrap(), 2); } }