agentmux_srv\server/

reactive.rs

// Copyright 2025-2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

use axum::{
    extract::{Query, State},
    http::StatusCode,
    response::{IntoResponse, Json, Response},
};
use serde_json::json;

use crate::backend::reactive::InjectionRequest;
use crate::backend::reactive::registry as agent_registry;
use crate::backend::subagent_watcher;
use crate::backend::base;

use super::AppState;

pub(super) async fn handle_reactive_inject(
    State(state): State<AppState>,
    Json(req): Json<InjectionRequest>,
) -> Json<serde_json::Value> {
    tracing::info!(
        target_agent = %req.target_agent,
        source_agent = ?req.source_agent,
        msg_len = req.message.len(),
        "reactive inject request received"
    );

    // 1. Try local ReactiveHandler first (fast path — same instance).
    let resp = state.reactive_handler.inject_message(req.clone());
    if resp.success {
        return Json(serde_json::to_value(&resp).unwrap_or_default());
    }

    // 2. On "agent not found", check cross-instance file registry and forward.
    let is_not_found = resp
        .error
        .as_deref()
        .map(|e| e.starts_with("agent not found"))
        .unwrap_or(false);

    if is_not_found {
        // Tier 2: same-host, different sidecar (file registry → HTTP loopback)
        let data_dir = base::get_wave_data_dir();
        if let Some(entry) = agent_registry::lookup(&data_dir, &req.target_agent) {
            // Guard against self-forwarding loops.
            if entry.local_url != state.local_web_url {
                let forward_url = format!("{}/agentmux/reactive/inject", entry.local_url);
                tracing::debug!(
                    target = %req.target_agent,
                    url = %forward_url,
                    "cross-instance inject forward"
                );
                let mut fwd = state.http_client.post(&forward_url).json(&req);
                if !entry.auth_key.is_empty() {
                    fwd = fwd.header("X-AuthKey", &entry.auth_key);
                }
                match fwd.send().await {
                    Ok(r) if r.status().is_success() => {
                        if let Ok(body) = r.json::<serde_json::Value>().await {
                            return Json(body);
                        }
                    }
                    Ok(r) => {
                        tracing::warn!(
                            target = %req.target_agent,
                            status = %r.status(),
                            url = %forward_url,
                            "cross-instance forward: non-success status"
                        );
                    }
                    Err(e) => {
                        tracing::warn!(
                            target = %req.target_agent,
                            error = %e,
                            url = %forward_url,
                            "cross-instance forward failed — removing stale registry entry"
                        );
                        agent_registry::remove(&data_dir, &req.target_agent);
                    }
                }
            }
        }

        // Tier 3: LAN peer (mDNS lookup → HTTP). Runs when tier 2 had no registry
        // entry or its forward failed. Queries each discovered LAN peer for the
        // agent; result is cached for 60s to avoid per-inject mDNS fan-out.
        if let Some((peer_url, peer_auth_key)) = state
            .lan_discovery
            .find_agent(&req.target_agent, &state.http_client)
            .await
        {
            let forward_url = format!("{}/agentmux/reactive/inject", peer_url);
            tracing::debug!(
                target = %req.target_agent,
                url = %forward_url,
                "LAN peer inject forward"
            );
            let mut fwd = state.http_client.post(&forward_url).json(&req);
            if !peer_auth_key.is_empty() {
                fwd = fwd.header("X-AuthKey", &peer_auth_key);
            }
            match fwd.send().await {
                Ok(r) if r.status().is_success() => {
                    if let Ok(body) = r.json::<serde_json::Value>().await {
                        // /reactive/inject always returns HTTP 200; check body.success
                        // to detect "agent not found on that peer" (e.g. after migration).
                        if body.get("success").and_then(|v| v.as_bool()) == Some(false) {
                            tracing::warn!(
                                target = %req.target_agent,
                                url = %forward_url,
                                "LAN peer inject: success=false — evicting stale cache entry"
                            );
                            state.lan_discovery.evict_agent(&req.target_agent);
                        } else {
                            return Json(body);
                        }
                    }
                }
                Ok(r) => {
                    tracing::warn!(
                        target = %req.target_agent,
                        status = %r.status(),
                        url = %forward_url,
                        "LAN peer forward: non-success HTTP status"
                    );
                }
                Err(e) => {
                    tracing::warn!(
                        target = %req.target_agent,
                        error = %e,
                        url = %forward_url,
                        "LAN peer forward failed — evicting cache entry"
                    );
                    state.lan_discovery.evict_agent(&req.target_agent);
                }
            }
        }
    }

    // 4. Return original error (muxbus-client will fall back to cloud relay).
    Json(serde_json::to_value(&resp).unwrap_or_default())
}

pub(super) async fn handle_reactive_agents(
    State(state): State<AppState>,
) -> Json<serde_json::Value> {
    let agents = state.reactive_handler.list_agents();
    Json(serde_json::to_value(&agents).unwrap_or(json!([])))
}

#[derive(serde::Deserialize)]
pub(super) struct AgentQuery {
    id: Option<String>,
}

pub(super) async fn handle_reactive_agent(
    State(state): State<AppState>,
    Query(params): Query<AgentQuery>,
) -> Response {
    let id = match &params.id {
        Some(id) if !id.is_empty() => id.as_str(),
        _ => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "missing id param"})),
            )
                .into_response()
        }
    };
    match state.reactive_handler.get_agent(id) {
        Some(agent) => Json(serde_json::to_value(&agent).unwrap_or_default()).into_response(),
        None => (
            StatusCode::NOT_FOUND,
            Json(json!({"error": "agent not found"})),
        )
            .into_response(),
    }
}

#[derive(serde::Deserialize)]
pub(super) struct AuditQuery {
    #[serde(default = "default_audit_limit")]
    limit: usize,
}
fn default_audit_limit() -> usize {
    100
}

pub(super) async fn handle_reactive_audit(
    State(state): State<AppState>,
    Query(params): Query<AuditQuery>,
) -> Json<serde_json::Value> {
    let log = state.reactive_handler.get_audit_log(params.limit);
    Json(serde_json::to_value(&log).unwrap_or(json!([])))
}

#[derive(serde::Deserialize)]
pub(super) struct RegisterRequest {
    agent_id: String,
    block_id: String,
    tab_id: Option<String>,
}

pub(super) async fn handle_reactive_register(
    State(state): State<AppState>,
    Json(req): Json<RegisterRequest>,
) -> Response {
    tracing::info!(
        agent_id = %req.agent_id,
        block_id = %req.block_id,
        "reactive register request"
    );
    match state
        .reactive_handler
        .register_agent(&req.agent_id, &req.block_id, req.tab_id.as_deref())
    {
        Ok(()) => {
            // Also write to cross-instance file registry so other AgentMux
            // instances can forward inject requests to this one.
            let data_dir = base::get_wave_data_dir();
            agent_registry::write(&data_dir, &req.agent_id, &state.local_web_url, &req.block_id);

            // Auto-watch this agent's Claude Code config dir for subagent JSONL files.
            // Pass block_id so subagent events are stamped with the owning pane,
            // letting the frontend route ⚡ panels to that pane only.
            if let Some(config_dir) = subagent_watcher::derive_claude_config_dir(&req.agent_id) {
                state.subagent_watcher.watch_agent(&req.agent_id, &req.block_id, config_dir);
            }

            // Notify cloud subscriber so it can subscribe for cloud-push delivery
            if let Some(sub) = crate::muxbus::cloud_subscriber::get_global_subscriber() {
                sub.add_agent(&req.agent_id);
            }

            // Notify the Swarm view so it calls AgentTrackedBlocksCommand and
            // shows this pane. We use a dedicated event name so useProcessCount
            // (which subscribes to agent:process-added / agent:process-exited)
            // doesn't treat this as a phantom OS process and show a spurious ⚙ N
            // badge or trigger the kill-tree modal on pane close.
            state.broker.publish(crate::backend::wps::WaveEvent {
                event: "agent:reactive-registered".to_string(),
                scopes: vec![format!("block:{}", req.block_id)],
                sender: String::new(),
                persist: 0,
                data: Some(json!({ "block_id": req.block_id })),
            });

            Json(json!({"success": true})).into_response()
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": e})),
        )
            .into_response(),
    }
}

#[derive(serde::Deserialize)]
pub(super) struct UnregisterRequest {
    agent_id: String,
}

pub(super) async fn handle_reactive_unregister(
    State(state): State<AppState>,
    Json(req): Json<UnregisterRequest>,
) -> Json<serde_json::Value> {
    // Capture block_id before unregistering so we can emit the Swarm refresh event.
    let block_id = state.reactive_handler.get_agent(&req.agent_id)
        .map(|r| r.block_id.clone());

    state.reactive_handler.unregister_agent(&req.agent_id);
    // Also remove from cross-instance file registry.
    let data_dir = base::get_wave_data_dir();
    agent_registry::remove(&data_dir, &req.agent_id);
    // Drop the subagent filesystem watcher (handle + channel + task) — the
    // symmetric teardown for the watch_agent() call in the register handler.
    state.subagent_watcher.unwatch_agent(&req.agent_id);
    // Notify cloud subscriber so it stops subscribing for this agent
    if let Some(sub) = crate::muxbus::cloud_subscriber::get_global_subscriber() {
        sub.remove_agent(&req.agent_id);
    }

    // Symmetric refresh: tell the Swarm view this pane is gone.
    if let Some(bid) = block_id {
        state.broker.publish(crate::backend::wps::WaveEvent {
            event: "agent:reactive-unregistered".to_string(),
            scopes: vec![format!("block:{}", bid)],
            sender: String::new(),
            persist: 0,
            data: Some(json!({ "block_id": bid })),
        });
    }

    Json(json!({"success": true}))
}

pub(super) async fn handle_reactive_poller_stats(
    State(state): State<AppState>,
) -> Json<serde_json::Value> {
    let stats = state.poller.stats();
    Json(serde_json::to_value(&stats).unwrap_or(json!({})))
}

#[derive(serde::Deserialize)]
pub(super) struct PollerConfigRequest {
    url: Option<String>,
    token: Option<String>,
}

pub(super) async fn handle_reactive_poller_config(
    State(state): State<AppState>,
    Json(req): Json<PollerConfigRequest>,
) -> Json<serde_json::Value> {
    state.poller.reconfigure(req.url, req.token);
    Json(json!({"success": true}))
}

pub(super) async fn handle_reactive_poller_status(
    State(state): State<AppState>,
) -> Json<serde_json::Value> {
    let status = state.poller.status();
    Json(serde_json::to_value(&status).unwrap_or(json!({})))
}