feat(server): operator removal of stale sessions/machines (SPEC-004 Task 5, server)

Admin-gated soft-delete + purge so operators can clear ghost machines/sessions
(the ~15-rows-for-one-host accumulation) from the console.

- migration 009: deleted_at on connect_sessions + connect_machines, with partial
  indexes WHERE deleted_at IS NULL.
- DELETE /api/machines/:agent_id?purge=true and DELETE /api/sessions/:id?purge=true
  soft-delete the row and purge the in-memory session (remove_session); the
  non-purge path keeps the legacy hard-delete / live-only disconnect. POST
  /api/machines/bulk-remove handles multi-select (batch cap 500). All admin-gated
  (AdminUser -> 403; tightens the prior any-user delete) and audited to
  connect_session_events (actor + target + trusted client IP).
- list/get queries filter deleted_at IS NULL so removed units leave the console;
  upsert revives (deleted_at = NULL) a genuinely-reconnecting machine. The
  keyed-reattach identity resolver (get_machine_by_id) is intentionally unfiltered.

Dashboard removal UI is the A3b follow-up. 86 server tests pass; fmt/clippy/test
clean. Implements specs/v2-stable-identity/plan.md Task 5 (server portion).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

server/src/api/mod.rs

@@ -6,6 +6,7 @@ pub mod changelog;
 pub mod downloads;
 pub mod machine_keys;
 pub mod releases;
+pub mod removal;
 pub mod sessions;
 pub mod users;
 
@@ -172,7 +173,7 @@ impl From<db::sessions::DbSession> for SessionRecord {
 #[derive(Debug, Serialize)]
 pub struct EventRecord {
     pub id: i64,
-    pub session_id: String,
+    pub session_id: Option<String>,
     pub event_type: String,
     pub timestamp: String,
     pub viewer_id: Option<String>,
@@ -185,7 +186,7 @@ impl From<db::events::SessionEvent> for EventRecord {
     fn from(e: db::events::SessionEvent) -> Self {
         Self {
             id: e.id,
-            session_id: e.session_id.to_string(),
+            session_id: e.session_id.map(|id| id.to_string()),
             event_type: e.event_type,
             timestamp: e.timestamp.to_rfc3339(),
             viewer_id: e.viewer_id,
@@ -208,12 +209,17 @@ pub struct MachineHistory {
 /// Query parameters for machine deletion
 #[derive(Debug, Deserialize)]
 pub struct DeleteMachineParams {
-    /// If true, send uninstall command to agent (if online)
+    /// If true, send uninstall command to agent (if online). Legacy (non-purge) path.
     #[serde(default)]
     pub uninstall: bool,
-    /// If true, include history in response before deletion
+    /// If true, include history in response before deletion. Legacy (non-purge) path.
     #[serde(default)]
     pub export: bool,
+    /// If true, take the Task-5 SOFT-DELETE path: set `connect_machines.deleted_at`,
+    /// drop the live in-memory session, and audit the removal — instead of the legacy
+    /// hard delete. This is the operator-removal mechanism that purges ghost rows.
+    #[serde(default)]
+    pub purge: bool,
 }
 
 /// Response for machine deletion

server/src/api/removal.rs

@@ -0,0 +1,612 @@
+//! Operator-removal endpoints (admin plane) — SPEC-004 / v2-stable-identity Task 5.
+//!
+//! Lets an administrator PURGE stale machines and sessions: soft-delete the DB row
+//! (`deleted_at = NOW()`, retaining the audit history), drop the live in-memory
+//! session via [`SessionManager::remove_session`], and write an audit row to
+//! `connect_session_events`. This is the mechanism that removes the ghost duplicate
+//! `connect_machines` rows left by the historical duplicate-registration bug.
+//!
+//! Removal semantics vs. the pre-existing live-only controls:
+//!   - `DELETE /api/sessions/:id` (live-only "disconnect") sends a `Disconnect`
+//!     message to the agent. It does NOT touch the DB. Left unchanged.
+//!   - `DELETE /api/machines/:agent_id` WITHOUT `?purge=true` keeps the legacy
+//!     behavior (optional `?uninstall=true` admin command, optional `?export=true`
+//!     history dump, then a HARD delete). Left unchanged.
+//!   - `?purge=true` on either route is the NEW soft-delete path defined here.
+//!
+//! Auth: dashboard JWT + admin role (the [`AdminUser`] extractor). A non-admin gets
+//! 403. Every purge is audited with the acting admin and the target id.
+//!
+//! Errors use the shared [`ApiError`] envelope; raw `sqlx` strings are never sent to
+//! the client (they are logged server-side via `tracing`).
+
+use std::net::IpAddr;
+
+use axum::{
+    extract::{ConnectInfo, Path, Query, State},
+    http::{HeaderMap, StatusCode},
+    Json,
+};
+use serde::{Deserialize, Serialize};
+use std::net::SocketAddr;
+use uuid::Uuid;
+
+use crate::auth::AdminUser;
+use crate::db;
+use crate::db::events::EventTypes;
+use crate::proto;
+use crate::AppState;
+
+use super::machine_keys::ApiError;
+use super::{DeleteMachineParams, DeleteMachineResponse, MachineHistory};
+
+type ApiResult<T> = Result<T, (StatusCode, Json<ApiError>)>;
+
+/// Build the shared error envelope (mirrors `machine_keys`/`sessions`).
+fn err(status: StatusCode, code: &str, detail: &str) -> (StatusCode, Json<ApiError>) {
+    (
+        status,
+        Json(ApiError {
+            detail: detail.to_string(),
+            error_code: code.to_string(),
+            status_code: status.as_u16(),
+        }),
+    )
+}
+
+/// Resolve the live `Database` handle or 503 (DB is optional in this server).
+fn require_db(state: &AppState) -> ApiResult<&db::Database> {
+    state.db.as_ref().ok_or_else(|| {
+        err(
+            StatusCode::SERVICE_UNAVAILABLE,
+            "DATABASE_UNAVAILABLE",
+            "Database not available",
+        )
+    })
+}
+
+/// Real client IP for the audit row (trusted-proxy aware, matching the rest of the
+/// server). Best-effort: a missing/garbled forwarded header just yields the peer.
+fn audit_ip(state: &AppState, addr: SocketAddr, headers: &HeaderMap) -> IpAddr {
+    crate::utils::ip_extract::client_ip(&addr, headers, &state.trusted_proxies)
+}
+
+// ============================================================================
+// Single-machine removal — DELETE /api/machines/:agent_id[?purge=true]
+// ============================================================================
+
+/// DELETE /api/machines/:agent_id
+///
+/// Admin-only. Two modes, selected by the `purge` query flag:
+///   - `?purge=true` → SOFT-DELETE (Task 5): set `connect_machines.deleted_at`,
+///     drop the machine's live in-memory session via `remove_session`, and write a
+///     `machine_removed` audit event. The row + its history are retained; the
+///     machine disappears from `/api/machines` and is not restored on startup.
+///   - otherwise → the legacy HARD delete (optional uninstall command + history
+///     export, then a cascading `DELETE`), preserved verbatim for compatibility.
+pub async fn remove_machine(
+    AdminUser(admin): AdminUser,
+    State(state): State<AppState>,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    headers: HeaderMap,
+    Path(agent_id): Path<String>,
+    Query(params): Query<DeleteMachineParams>,
+) -> ApiResult<Json<DeleteMachineResponse>> {
+    // `agent_id` is a UUID minted by the agent (`generate_agent_id()`); validate the
+    // shape before touching the DB so a malformed path is a clean 400, not a lookup.
+    if Uuid::parse_str(&agent_id).is_err() {
+        return Err(err(
+            StatusCode::BAD_REQUEST,
+            "INVALID_AGENT_ID",
+            "agent_id must be a valid UUID",
+        ));
+    }
+
+    let db = require_db(&state)?;
+
+    // The machine must currently exist and be live (the get filters soft-deleted),
+    // so a re-purge or an unknown id is a clean 404 rather than a silent success.
+    let machine = db::machines::get_machine_by_agent_id(db.pool(), &agent_id)
+        .await
+        .map_err(|e| {
+            tracing::error!("DB error loading machine for removal: {}", e);
+            err(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "INTERNAL_ERROR",
+                "Internal server error",
+            )
+        })?
+        .ok_or_else(|| {
+            err(
+                StatusCode::NOT_FOUND,
+                "MACHINE_NOT_FOUND",
+                "Machine not found",
+            )
+        })?;
+
+    if params.purge {
+        return purge_machine(
+            &state,
+            db,
+            &admin,
+            machine,
+            audit_ip(&state, addr, &headers),
+        )
+        .await;
+    }
+
+    // -------- Legacy hard-delete path (unchanged behavior) --------
+    let history = if params.export {
+        let sessions = db::sessions::get_sessions_for_machine(db.pool(), machine.id)
+            .await
+            .map_err(|e| {
+                tracing::error!("DB error exporting sessions: {}", e);
+                err(
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    "INTERNAL_ERROR",
+                    "Internal server error",
+                )
+            })?;
+        let events = db::events::get_events_for_machine(db.pool(), machine.id)
+            .await
+            .map_err(|e| {
+                tracing::error!("DB error exporting events: {}", e);
+                err(
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    "INTERNAL_ERROR",
+                    "Internal server error",
+                )
+            })?;
+        Some(MachineHistory {
+            machine: super::MachineInfo::from(machine.clone()),
+            sessions: sessions
+                .into_iter()
+                .map(super::SessionRecord::from)
+                .collect(),
+            events: events.into_iter().map(super::EventRecord::from).collect(),
+            exported_at: chrono::Utc::now().to_rfc3339(),
+        })
+    } else {
+        None
+    };
+
+    let mut uninstall_sent = false;
+    if params.uninstall {
+        if let Some(session) = state.sessions.get_session_by_agent(&agent_id).await {
+            if session.is_online {
+                uninstall_sent = state
+                    .sessions
+                    .send_admin_command(
+                        session.id,
+                        proto::AdminCommandType::AdminUninstall,
+                        "Deleted by administrator",
+                    )
+                    .await;
+                if uninstall_sent {
+                    tracing::info!("Sent uninstall command to agent {}", agent_id);
+                }
+            }
+        }
+    }
+
+    state.sessions.remove_agent(&agent_id).await;
+
+    db::machines::delete_machine(db.pool(), &agent_id)
+        .await
+        .map_err(|e| {
+            tracing::error!("DB error hard-deleting machine: {}", e);
+            err(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "INTERNAL_ERROR",
+                "Failed to delete machine",
+            )
+        })?;
+
+    tracing::info!(
+        "Admin {} hard-deleted machine {} (uninstall_sent: {})",
+        admin.username,
+        agent_id,
+        uninstall_sent
+    );
+
+    Ok(Json(DeleteMachineResponse {
+        success: true,
+        message: format!("Machine {} deleted", machine.hostname),
+        uninstall_sent,
+        history,
+    }))
+}
+
+/// Soft-delete + in-memory removal + audit for one machine (the `?purge=true` path).
+async fn purge_machine(
+    state: &AppState,
+    db: &db::Database,
+    admin: &crate::auth::AuthenticatedUser,
+    machine: db::machines::Machine,
+    ip: IpAddr,
+) -> ApiResult<Json<DeleteMachineResponse>> {
+    // 1. Soft-delete the DB row. 0 rows means it was purged between the load and
+    //    here (a race) — treat as already-removed success, idempotent.
+    let affected = db::machines::soft_delete_machine(db.pool(), &machine.agent_id)
+        .await
+        .map_err(|e| {
+            tracing::error!("DB error soft-deleting machine: {}", e);
+            err(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "INTERNAL_ERROR",
+                "Failed to remove machine",
+            )
+        })?;
+
+    // 2. Drop the live in-memory session (agents + machine_uids indexes) so the
+    //    purged machine cannot keep streaming or be reattached. `remove_agent`
+    //    resolves agent_id -> session_id and routes through `remove_session`.
+    let removed_session = state.sessions.remove_agent(&machine.agent_id).await;
+
+    // 3. Audit. Anchor to the machine's last session when known so the event links
+    //    to that machine's history; the detail JSON independently carries the ids.
+    let details = serde_json::json!({
+        "target_kind": "machine",
+        "agent_id": machine.agent_id,
+        "machine_id": machine.id,
+        "hostname": machine.hostname,
+        "machine_uid": machine.machine_uid,
+        "purge": true,
+        "soft_deleted": affected > 0,
+        "in_memory_session_removed": removed_session.map(|s| s.to_string()),
+    });
+    if let Err(e) = db::events::log_admin_removal(
+        db.pool(),
+        machine.last_session_id,
+        EventTypes::MACHINE_REMOVED,
+        &admin.user_id,
+        &admin.username,
+        details,
+        Some(ip),
+    )
+    .await
+    {
+        // Audit is best-effort: the purge already happened. Log loudly; do not fail
+        // the request (mirrors how the relay treats audit-write failures).
+        tracing::error!("Failed to write machine_removed audit event: {}", e);
+    }
+
+    tracing::info!(
+        "Admin {} purged machine {} (agent_id {}, soft_deleted={}, session_removed={})",
+        admin.username,
+        machine.hostname,
+        machine.agent_id,
+        affected > 0,
+        removed_session.is_some()
+    );
+
+    Ok(Json(DeleteMachineResponse {
+        success: true,
+        message: format!("Machine {} removed", machine.hostname),
+        uninstall_sent: false,
+        history: None,
+    }))
+}
+
+// ============================================================================
+// Single-session removal — DELETE /api/sessions/:id[?purge=true]
+// ============================================================================
+
+/// Query flag for the session removal route.
+#[derive(Debug, Deserialize)]
+pub struct PurgeParams {
+    /// When true, soft-delete the session row + remove it in-memory + audit. When
+    /// false/absent, fall back to the live-only disconnect (unchanged behavior).
+    #[serde(default)]
+    pub purge: bool,
+}
+
+/// Result body for a session removal.
+#[derive(Debug, Serialize)]
+pub struct RemoveSessionResponse {
+    pub success: bool,
+    pub message: String,
+    /// Whether the row was soft-deleted in the DB (false if there was no DB row,
+    /// e.g. a live support session that never persisted, or it was already purged).
+    pub soft_deleted: bool,
+}
+
+/// DELETE /api/sessions/:id?purge=true
+///
+/// Admin-only soft-delete of a session: set `connect_sessions.deleted_at`, drop the
+/// live in-memory session via `remove_session`, and write a `session_removed` audit
+/// event. WITHOUT `?purge=true` this delegates to the live-only disconnect so the
+/// pre-existing semantics are preserved.
+pub async fn remove_session(
+    AdminUser(admin): AdminUser,
+    State(state): State<AppState>,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    headers: HeaderMap,
+    Path(id): Path<String>,
+    Query(params): Query<PurgeParams>,
+) -> ApiResult<Json<RemoveSessionResponse>> {
+    let session_id = Uuid::parse_str(&id).map_err(|_| {
+        err(
+            StatusCode::BAD_REQUEST,
+            "INVALID_SESSION_ID",
+            "session id must be a valid UUID",
+        )
+    })?;
+
+    if !params.purge {
+        // Live-only disconnect (unchanged): send a Disconnect to the agent. 404 if
+        // the session is not live in memory.
+        let disconnected = state
+            .sessions
+            .disconnect_session(session_id, "Disconnected by administrator")
+            .await;
+        if disconnected {
+            tracing::info!(
+                "Admin {} disconnected session {} (live-only)",
+                admin.username,
+                session_id
+            );
+            return Ok(Json(RemoveSessionResponse {
+                success: true,
+                message: "Session disconnected".to_string(),
+                soft_deleted: false,
+            }));
+        }
+        return Err(err(
+            StatusCode::NOT_FOUND,
+            "SESSION_NOT_FOUND",
+            "Session not found",
+        ));
+    }
+
+    let db = require_db(&state)?;
+
+    // A session may be live in memory, persisted in the DB, or both. Soft-delete the
+    // DB row (no-op if there is none) and drop any in-memory session. The purge is a
+    // 404 only when NEITHER exists.
+    let in_memory = state.sessions.get_session(session_id).await.is_some();
+    let soft_deleted = db::sessions::soft_delete_session(db.pool(), session_id)
+        .await
+        .map_err(|e| {
+            tracing::error!("DB error soft-deleting session: {}", e);
+            err(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "INTERNAL_ERROR",
+                "Failed to remove session",
+            )
+        })?
+        > 0;
+
+    if !in_memory && !soft_deleted {
+        return Err(err(
+            StatusCode::NOT_FOUND,
+            "SESSION_NOT_FOUND",
+            "Session not found",
+        ));
+    }
+
+    state.sessions.remove_session(session_id).await;
+
+    let details = serde_json::json!({
+        "target_kind": "session",
+        "session_id": session_id,
+        "purge": true,
+        "soft_deleted": soft_deleted,
+        "was_live": in_memory,
+    });
+    if let Err(e) = db::events::log_admin_removal(
+        db.pool(),
+        Some(session_id),
+        EventTypes::SESSION_REMOVED,
+        &admin.user_id,
+        &admin.username,
+        details,
+        Some(audit_ip(&state, addr, &headers)),
+    )
+    .await
+    {
+        tracing::error!("Failed to write session_removed audit event: {}", e);
+    }
+
+    tracing::info!(
+        "Admin {} purged session {} (soft_deleted={}, was_live={})",
+        admin.username,
+        session_id,
+        soft_deleted,
+        in_memory
+    );
+
+    Ok(Json(RemoveSessionResponse {
+        success: true,
+        message: "Session removed".to_string(),
+        soft_deleted,
+    }))
+}
+
+// ============================================================================
+// Bulk machine removal — POST /api/machines/bulk-remove
+// ============================================================================
+
+/// Request body for the bulk machine removal.
+#[derive(Debug, Deserialize)]
+pub struct BulkRemoveRequest {
+    /// `agent_id`s to remove. Each is validated as a UUID; unknown/invalid ids are
+    /// reported per-id rather than failing the whole batch.
+    pub ids: Vec<String>,
+    /// When true, soft-delete + in-memory removal + audit (Task 5). When false, the
+    /// legacy hard delete is applied to each id. Defaults to true: the bulk endpoint
+    /// exists for the purge workflow.
+    #[serde(default = "default_true")]
+    pub purge: bool,
+}
+
+fn default_true() -> bool {
+    true
+}
+
+/// Per-id outcome in a bulk response.
+#[derive(Debug, Serialize)]
+pub struct BulkRemoveItem {
+    pub agent_id: String,
+    /// `removed` | `not_found` | `invalid` | `error`.
+    pub status: String,
+}
+
+/// Summary body for a bulk removal.
+#[derive(Debug, Serialize)]
+pub struct BulkRemoveResponse {
+    pub requested: usize,
+    pub removed: usize,
+    pub results: Vec<BulkRemoveItem>,
+}
+
+/// POST /api/machines/bulk-remove
+///
+/// Admin-only. Removes many machines in one call, auditing ONE `machine_removed`
+/// event per actually-removed id (matching the single-machine granularity). Invalid
+/// or unknown ids are reported in the per-id results and never abort the batch. With
+/// `purge=true` (default) each removal is the Task-5 soft-delete; otherwise the
+/// legacy hard delete.
+pub async fn bulk_remove_machines(
+    AdminUser(admin): AdminUser,
+    State(state): State<AppState>,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    headers: HeaderMap,
+    Json(body): Json<BulkRemoveRequest>,
+) -> ApiResult<Json<BulkRemoveResponse>> {
+    if body.ids.is_empty() {
+        return Err(err(
+            StatusCode::BAD_REQUEST,
+            "EMPTY_ID_LIST",
+            "ids must contain at least one agent_id",
+        ));
+    }
+    // Guard against an unbounded batch.
+    const MAX_BATCH: usize = 500;
+    if body.ids.len() > MAX_BATCH {
+        return Err(err(
+            StatusCode::BAD_REQUEST,
+            "BATCH_TOO_LARGE",
+            "ids exceeds the maximum batch size of 500",
+        ));
+    }
+
+    let db = require_db(&state)?;
+    let ip = audit_ip(&state, addr, &headers);
+
+    let requested = body.ids.len();
+    let mut results = Vec::with_capacity(requested);
+    let mut removed = 0usize;
+
+    for agent_id in body.ids {
+        // Validate shape first.
+        if Uuid::parse_str(&agent_id).is_err() {
+            results.push(BulkRemoveItem {
+                agent_id,
+                status: "invalid".to_string(),
+            });
+            continue;
+        }
+
+        // Must exist + be live.
+        let machine = match db::machines::get_machine_by_agent_id(db.pool(), &agent_id).await {
+            Ok(Some(m)) => m,
+            Ok(None) => {
+                results.push(BulkRemoveItem {
+                    agent_id,
+                    status: "not_found".to_string(),
+                });
+                continue;
+            }
+            Err(e) => {
+                tracing::error!("DB error loading machine {} in bulk: {}", agent_id, e);
+                results.push(BulkRemoveItem {
+                    agent_id,
+                    status: "error".to_string(),
+                });
+                continue;
+            }
+        };
+
+        if body.purge {
+            match db::machines::soft_delete_machine(db.pool(), &agent_id).await {
+                Ok(affected) => {
+                    let removed_session = state.sessions.remove_agent(&agent_id).await;
+                    let details = serde_json::json!({
+                        "target_kind": "machine",
+                        "agent_id": machine.agent_id,
+                        "machine_id": machine.id,
+                        "hostname": machine.hostname,
+                        "machine_uid": machine.machine_uid,
+                        "purge": true,
+                        "bulk": true,
+                        "soft_deleted": affected > 0,
+                        "in_memory_session_removed": removed_session.map(|s| s.to_string()),
+                    });
+                    if let Err(e) = db::events::log_admin_removal(
+                        db.pool(),
+                        machine.last_session_id,
+                        EventTypes::MACHINE_REMOVED,
+                        &admin.user_id,
+                        &admin.username,
+                        details,
+                        Some(ip),
+                    )
+                    .await
+                    {
+                        tracing::error!(
+                            "Failed to write machine_removed audit event (bulk) for {}: {}",
+                            agent_id,
+                            e
+                        );
+                    }
+                    removed += 1;
+                    results.push(BulkRemoveItem {
+                        agent_id,
+                        status: "removed".to_string(),
+                    });
+                }
+                Err(e) => {
+                    tracing::error!("DB error soft-deleting machine {} in bulk: {}", agent_id, e);
+                    results.push(BulkRemoveItem {
+                        agent_id,
+                        status: "error".to_string(),
+                    });
+                }
+            }
+        } else {
+            // Legacy hard-delete per id.
+            state.sessions.remove_agent(&agent_id).await;
+            match db::machines::delete_machine(db.pool(), &agent_id).await {
+                Ok(()) => {
+                    removed += 1;
+                    results.push(BulkRemoveItem {
+                        agent_id,
+                        status: "removed".to_string(),
+                    });
+                }
+                Err(e) => {
+                    tracing::error!("DB error hard-deleting machine {} in bulk: {}", agent_id, e);
+                    results.push(BulkRemoveItem {
+                        agent_id,
+                        status: "error".to_string(),
+                    });
+                }
+            }
+        }
+    }
+
+    tracing::info!(
+        "Admin {} bulk-removed {}/{} machine(s) (purge={})",
+        admin.username,
+        removed,
+        requested,
+        body.purge
+    );
+
+    Ok(Json(BulkRemoveResponse {
+        requested,
+        removed,
+        results,
+    }))
+}