brokkr-agent::broker

Rust

Broker communication module for agent-broker interaction.

All v1 API calls go through brokkr_client::BrokkrClient (T-D1 migration). The only remaining reqwest::Client usage is in wait_for_broker_ready for the /readyz health probe, which lives outside the v1 surface and is therefore not part of the generated SDK.

Public function signatures continue to exchange brokkr-models types (Agent, DeploymentObject, …) so callers (cli/commands.rs) didn’t have to change. Inside each function we JSON-round-trip between the SDK’s generated types and the workspace’s brokkr-models types — the wire form is byte-identical and the conversion is cheap on the frequencies we operate at (seconds-scale).

Functions

fn wait_​for_​broker_​ready

pub

async fn wait_for_broker_ready(config: &Settings)

Waits for the broker service to become ready.

Not migrated to the SDK: /readyz is outside the v1 spec surface.

Source

pub async fn wait_for_broker_ready(config: &Settings) {
    let client = reqwest::Client::new();
    let readyz_url = format!("{}/readyz", config.agent.broker_url);

    for attempt in 1..=config.agent.max_retries {
        match client.get(&readyz_url).send().await {
            Ok(response) => {
                if response.status().is_success() {
                    info!("Successfully connected to broker at {}", readyz_url);
                    return;
                }
                warn!(
                    "Broker at {} returned non-success status: {}",
                    readyz_url,
                    response.status()
                );
            }
            Err(e) => {
                warn!(
                    "Failed to connect to broker at {} (attempt {}/{}): {:?}",
                    readyz_url, attempt, config.agent.max_retries, e
                );
            }
        }
        if attempt < config.agent.max_retries {
            info!(
                "Waiting for broker to be ready at {} (attempt {}/{})",
                readyz_url, attempt, config.agent.max_retries
            );
            sleep(Duration::from_secs(1)).await;
        }
    }
    error!(
        "Failed to connect to broker at {} after {} attempts. Exiting.",
        readyz_url, config.agent.max_retries
    );
    std::process::exit(1);
}

fn verify_​agent_​pak

pub

async fn verify_agent_pak(
    config: &Settings) -> Result<(,
) -> Result<(), Box<dyn std::error::Error>>

Verifies the agent’s Prefixed API Key (PAK) with the broker.

Source

pub async fn verify_agent_pak(config: &Settings) -> Result<(), Box<dyn std::error::Error>> {
    debug!("Verifying agent PAK at {}", config.agent.broker_url);
    let sdk = crate::broker_sdk::build_client(config)
        .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

    match sdk.api().verify_pak().send().await {
        Ok(_) => {
            info!("Successfully verified agent PAK");
            Ok(())
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            if status_u16(&wrapped) == Some(401) {
                error!("Agent PAK verification failed: unauthorized");
                Err("Invalid agent PAK".into())
            } else {
                error!("PAK verification failed: {}", wrapped);
                Err(boxed("PAK verification failed", wrapped))
            }
        }
    }
}

fn fetch_​agent_​details

pub

async fn fetch_agent_details(
    config: &Settings,
    client: &BrokkrClient,
) -> Result<Agent, Box<dyn std::error::Error>>

Fetches the details of the agent from the broker.

Source

pub async fn fetch_agent_details(
    config: &Settings,
    client: &BrokkrClient,
) -> Result<Agent, Box<dyn std::error::Error>> {
    debug!(
        "Fetching agent details for name={} cluster={}",
        config.agent.agent_name, config.agent.cluster_name
    );

    let result = client
        .api()
        .search_agent()
        .name(&config.agent.agent_name)
        .cluster_name(&config.agent.cluster_name)
        .send()
        .await;

    match result {
        Ok(rv) => {
            let agent: Agent = convert(rv.into_inner()).map_err(|e| {
                error!("Failed to convert agent details: {}", e);
                Box::new(e) as Box<dyn std::error::Error>
            })?;
            info!(
                "Successfully fetched details for agent {} in cluster {}",
                agent.name, agent.cluster_name
            );
            Ok(agent)
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            if status_u16(&wrapped) == Some(404) {
                error!(
                    "Agent not found: name={}, cluster={}",
                    config.agent.agent_name, config.agent.cluster_name
                );
                Err("Agent not found".into())
            } else {
                error!("Failed to fetch agent details: {}", wrapped);
                Err(boxed("Failed to fetch agent details", wrapped))
            }
        }
    }
}

fn fetch_​associated_​stack_​ids

pub

async fn fetch_associated_stack_ids(
    client: &BrokkrClient,
    agent: &Agent,
) -> Result<std::collections::HashSet<Uuid>, Box<dyn std::error::Error>>

Fetches the ids of the stacks the broker associates with this agent: its targets, and the stacks its labels and annotations match, under the generators it is registered with. This is the set gap detection compares between polls (BROKKR-T-0342). The target-state list cannot serve: it is incremental, and an object leaves it once this agent reported it deployed, so a stack with nothing new to apply looks like a stack that was removed.

Source

pub async fn fetch_associated_stack_ids(
    client: &BrokkrClient,
    agent: &Agent,
) -> Result<std::collections::HashSet<Uuid>, Box<dyn std::error::Error>> {
    let rv = client
        .api()
        .get_associated_stacks()
        .id(agent.id)
        .send()
        .await
        .map_err(|e| boxed("Failed to fetch the agent's stacks", BrokkrError::from(e)))?;
    let stacks: Vec<brokkr_models::models::stacks::Stack> = convert(rv.into_inner())?;
    Ok(stacks.into_iter().map(|s| s.id).collect())
}

fn fetch_​and_​process_​deployment_​objects

pub

async fn fetch_and_process_deployment_objects(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
) -> Result<Vec<DeploymentObject>, Box<dyn std::error::Error>>

Fetches deployment objects to apply from the broker’s target-state view.

Source

pub async fn fetch_and_process_deployment_objects(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
) -> Result<Vec<DeploymentObject>, Box<dyn std::error::Error>> {
    debug!(
        "Fetching target-state deployment objects for agent {}",
        agent.name
    );

    let start = Instant::now();
    let result = client.api().get_target_state().id(agent.id).send().await;
    let duration = start.elapsed().as_secs_f64();

    match result {
        Ok(rv) => {
            let objects: Vec<DeploymentObject> = convert(rv.into_inner()).map_err(|e| {
                error!("Failed to convert deployment objects: {}", e);
                metrics::poll_requests_total()
                    .with_label_values(&["error"])
                    .inc();
                metrics::poll_duration_seconds()
                    .with_label_values(&[])
                    .observe(duration);
                Box::new(e) as Box<dyn std::error::Error>
            })?;
            info!(
                "Successfully fetched {} deployment objects for agent {}",
                objects.len(),
                agent.name
            );
            metrics::poll_requests_total()
                .with_label_values(&["success"])
                .inc();
            metrics::poll_duration_seconds()
                .with_label_values(&[])
                .observe(duration);
            Ok(objects)
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            error!("Failed to fetch deployment objects: {}", wrapped);
            metrics::poll_requests_total()
                .with_label_values(&["error"])
                .inc();
            metrics::poll_duration_seconds()
                .with_label_values(&[])
                .observe(duration);
            Err(boxed("Failed to fetch deployment objects", wrapped))
        }
    }
}

fn send_​success_​event

pub

async fn send_success_event(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
    deployment_object_id: Uuid,
    message: Option<String>,
    ws_uplink: Option<&WsUplink>,
) -> Result<(), Box<dyn std::error::Error>>

Sends a success event to the broker for the given deployment object.

Source

pub async fn send_success_event(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
    deployment_object_id: Uuid,
    message: Option<String>,
    ws_uplink: Option<&WsUplink>,
) -> Result<(), Box<dyn std::error::Error>> {
    debug!(
        "Sending success event for deployment {} for agent {}",
        deployment_object_id, agent.name
    );

    let event = NewAgentEvent {
        agent_id: agent.id,
        deployment_object_id,
        event_type: "DEPLOY".to_string(),
        status: "SUCCESS".to_string(),
        message,
    };

    // Prefer WS when the channel is up. Broker's reader_task dispatches
    // the same DAL write the REST handler would do (see WS-05 in
    // BROKKR-A-0008).
    if try_ws_send(ws_uplink, || synth_agent_event(&event)) {
        debug!(
            "Sent success event for deployment {} via WS",
            deployment_object_id
        );
        return Ok(());
    }

    let sdk_event: brokkr_client::types::NewAgentEvent = convert(event).map_err(|e| {
        error!("Failed to convert NewAgentEvent: {}", e);
        Box::new(e) as Box<dyn std::error::Error>
    })?;

    match client
        .api()
        .create_event()
        .id(agent.id)
        .body(sdk_event)
        .send()
        .await
    {
        Ok(_) => {
            info!(
                "Successfully reported deployment success for object {}",
                deployment_object_id
            );
            Ok(())
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            error!("Failed to send success event: {}", wrapped);
            Err(boxed("Failed to send success event", wrapped))
        }
    }
}

fn send_​failure_​event

pub

async fn send_failure_event(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
    deployment_object_id: Uuid,
    error_message: String,
    ws_uplink: Option<&WsUplink>,
) -> Result<(), Box<dyn std::error::Error>>

Sends a failure event to the broker for the given deployment object.

Source

pub async fn send_failure_event(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
    deployment_object_id: Uuid,
    error_message: String,
    ws_uplink: Option<&WsUplink>,
) -> Result<(), Box<dyn std::error::Error>> {
    debug!(
        "Sending failure event for deployment {} for agent {}",
        deployment_object_id, agent.name
    );

    let event = NewAgentEvent {
        agent_id: agent.id,
        deployment_object_id,
        event_type: "DEPLOY".to_string(),
        status: "FAILURE".to_string(),
        message: Some(error_message),
    };

    if try_ws_send(ws_uplink, || synth_agent_event(&event)) {
        debug!(
            "Sent failure event for deployment {} via WS",
            deployment_object_id
        );
        return Ok(());
    }

    let sdk_event: brokkr_client::types::NewAgentEvent = convert(event).map_err(|e| {
        error!("Failed to convert NewAgentEvent: {}", e);
        Box::new(e) as Box<dyn std::error::Error>
    })?;

    match client
        .api()
        .create_event()
        .id(agent.id)
        .body(sdk_event)
        .send()
        .await
    {
        Ok(_) => {
            info!(
                "Successfully reported deployment failure for object {}",
                deployment_object_id
            );
            Ok(())
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            error!(
                "Failed to send failure event for deployment {}: {}",
                deployment_object_id, wrapped
            );
            Err(boxed("Failed to send failure event", wrapped))
        }
    }
}

fn send_​heartbeat

pub

async fn send_heartbeat(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
    ws_uplink: Option<&WsUplink>,
    k8s_reachable: Option<bool>,
    k8s_api_latency_ms: Option<i32>,
) -> Result<(), Box<dyn std::error::Error>>

Sends a heartbeat to the broker for the given agent.

k8s_reachable / k8s_api_latency_ms carry the agent’s self-reported Kubernetes API connectivity (BROKKR-T-0227). They are optional: when the agent could not probe its cluster, pass None and the fields are omitted from the report (the broker leaves the columns untouched / NULL).

Source

pub async fn send_heartbeat(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
    ws_uplink: Option<&WsUplink>,
    k8s_reachable: Option<bool>,
    k8s_api_latency_ms: Option<i32>,
) -> Result<(), Box<dyn std::error::Error>> {
    if try_ws_send(ws_uplink, || {
        WsMessage::Heartbeat(Heartbeat {
            agent_id: agent.id,
            sent_at: Utc::now(),
            k8s_reachable,
            k8s_api_latency_ms,
        })
    }) {
        trace!("Heartbeat sent via WS for agent {}", agent.name);
        metrics::heartbeat_sent_total().inc();
        metrics::last_successful_poll_timestamp().set(
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs_f64(),
        );
        return Ok(());
    }

    let report = brokkr_client::types::HeartbeatReport {
        k8s_reachable,
        k8s_api_latency_ms,
    };
    match client
        .api()
        .record_heartbeat()
        .id(agent.id)
        .body(report)
        .send()
        .await
    {
        Ok(_) => {
            trace!("Heartbeat sent successfully for agent {}", agent.name);
            metrics::heartbeat_sent_total().inc();
            metrics::last_successful_poll_timestamp().set(
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap_or_default()
                    .as_secs_f64(),
            );
            Ok(())
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            metrics::heartbeat_sent_total().inc();
            if status_u16(&wrapped) == Some(401) {
                error!("Heartbeat unauthorized for agent {}", agent.name);
                Err("Unauthorized: Invalid agent PAK".into())
            } else {
                error!("Heartbeat failed for agent {}: {}", agent.name, wrapped);
                Err(boxed("Heartbeat failed", wrapped))
            }
        }
    }
}

fn send_​health_​status

pub

async fn send_health_status(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
    health_updates: Vec<DeploymentObjectHealthUpdate>,
    ws_uplink: Option<&WsUplink>,
) -> Result<(), Box<dyn std::error::Error>>

Sends health status updates for deployment objects to the broker.

Source

pub async fn send_health_status(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
    health_updates: Vec<DeploymentObjectHealthUpdate>,
    ws_uplink: Option<&WsUplink>,
) -> Result<(), Box<dyn std::error::Error>> {
    if health_updates.is_empty() {
        return Ok(());
    }

    debug!(
        "Sending health status update for {} deployment objects for agent {}",
        health_updates.len(),
        agent.name
    );

    // Try WS first — one frame per deployment-object update. If the lane
    // can absorb the whole batch we're done; otherwise fall through to
    // the REST batch endpoint with the full set.
    if let Some(uplink) = ws_uplink
        && uplink.is_up()
    {
        let now = Utc::now();
        let all_ok = health_updates.iter().all(|u| {
            let summary_json = u
                .summary
                .as_ref()
                .and_then(|s| serde_json::to_string(s).ok());
            let msg = WsMessage::AgentHealth(DeploymentHealth {
                id: Uuid::nil(), // server assigns
                agent_id: agent.id,
                deployment_object_id: u.id,
                status: u.status.to_string(),
                summary: summary_json,
                checked_at: u.checked_at,
                created_at: now,
                updated_at: now,
            });
            uplink.try_send(msg).is_ok()
        });
        if all_ok {
            debug!(
                "Sent {} health updates via WS for agent {}",
                health_updates.len(),
                agent.name
            );
            return Ok(());
        }
        warn!("WS health-update lane back-pressured; falling back to REST batch");
    }

    let update = HealthStatusUpdate {
        deployment_objects: health_updates,
    };
    let count = update.deployment_objects.len();
    let sdk_update: brokkr_client::types::HealthStatusUpdate = convert(update).map_err(|e| {
        error!("Failed to convert HealthStatusUpdate: {}", e);
        Box::new(e) as Box<dyn std::error::Error>
    })?;

    match client
        .api()
        .update_health_status()
        .id(agent.id)
        .body(sdk_update)
        .send()
        .await
    {
        Ok(_) => {
            debug!(
                "Successfully sent health status for {} deployment objects",
                count
            );
            Ok(())
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            if status_u16(&wrapped) == Some(401) {
                error!("Health status update unauthorized for agent {}", agent.name);
                Err("Unauthorized: Invalid agent PAK".into())
            } else {
                error!(
                    "Health status update failed for agent {}: {}",
                    agent.name, wrapped
                );
                Err(boxed("Health status update failed", wrapped))
            }
        }
    }
}

fn fetch_​pending_​diagnostics

pub

async fn fetch_pending_diagnostics(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
) -> Result<Vec<DiagnosticRequest>, Box<dyn std::error::Error>>

Fetches pending diagnostic requests for the agent.

Source

pub async fn fetch_pending_diagnostics(
    _config: &Settings,
    client: &BrokkrClient,
    agent: &Agent,
) -> Result<Vec<DiagnosticRequest>, Box<dyn std::error::Error>> {
    debug!("Fetching pending diagnostics for agent {}", agent.name);

    match client
        .api()
        .get_pending_diagnostics()
        .id(agent.id)
        .send()
        .await
    {
        Ok(rv) => {
            let requests: Vec<DiagnosticRequest> = convert(rv.into_inner()).map_err(|e| {
                error!("Failed to convert diagnostic requests: {}", e);
                Box::new(e) as Box<dyn std::error::Error>
            })?;
            if !requests.is_empty() {
                debug!(
                    "Found {} pending diagnostic requests for agent {}",
                    requests.len(),
                    agent.name
                );
            }
            Ok(requests)
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            error!("Failed to fetch pending diagnostics: {}", wrapped);
            Err(boxed("Failed to fetch pending diagnostics", wrapped))
        }
    }
}

fn fetch_​deployment_​object

pub

async fn fetch_deployment_object(
    client: &BrokkrClient,
    deployment_object_id: Uuid,
) -> Result<DeploymentObject, Box<dyn std::error::Error>>

Claims a diagnostic request for processing. Fetches a single deployment object by id. Used by the diagnostics loop to derive the namespaces to search from the object’s manifests (BROKKR-T-0190).

Source

pub async fn fetch_deployment_object(
    client: &BrokkrClient,
    deployment_object_id: Uuid,
) -> Result<DeploymentObject, Box<dyn std::error::Error>> {
    debug!("Fetching deployment object {}", deployment_object_id);

    match client
        .api()
        .get_deployment_object()
        .id(deployment_object_id)
        .send()
        .await
    {
        Ok(rv) => {
            let object: DeploymentObject = convert(rv.into_inner()).map_err(|e| {
                error!("Failed to convert deployment object: {}", e);
                Box::new(e) as Box<dyn std::error::Error>
            })?;
            Ok(object)
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            error!(
                "Failed to fetch deployment object {}: {}",
                deployment_object_id, wrapped
            );
            Err(Box::new(wrapped))
        }
    }
}

fn claim_​diagnostic_​request

pub

async fn claim_diagnostic_request(
    _config: &Settings,
    client: &BrokkrClient,
    request_id: Uuid,
) -> Result<DiagnosticRequest, Box<dyn std::error::Error>>

Source

pub async fn claim_diagnostic_request(
    _config: &Settings,
    client: &BrokkrClient,
    request_id: Uuid,
) -> Result<DiagnosticRequest, Box<dyn std::error::Error>> {
    debug!("Claiming diagnostic request {}", request_id);

    match client.api().claim_diagnostic().id(request_id).send().await {
        Ok(rv) => {
            let request: DiagnosticRequest = convert(rv.into_inner()).map_err(|e| {
                error!("Failed to convert claimed diagnostic request: {}", e);
                Box::new(e) as Box<dyn std::error::Error>
            })?;
            info!("Successfully claimed diagnostic request {}", request_id);
            Ok(request)
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            if status_u16(&wrapped) == Some(409) {
                warn!(
                    "Diagnostic request {} already claimed or completed",
                    request_id
                );
                Err(format!("Diagnostic request {request_id} already claimed or completed").into())
            } else {
                error!(
                    "Failed to claim diagnostic request {}: {}",
                    request_id, wrapped
                );
                Err(boxed("Failed to claim diagnostic request", wrapped))
            }
        }
    }
}

fn submit_​diagnostic_​result

pub

async fn submit_diagnostic_result(
    _config: &Settings,
    client: &BrokkrClient,
    request_id: Uuid,
    result: SubmitDiagnosticResult,
) -> Result<(), Box<dyn std::error::Error>>

Submits diagnostic results for a request.

Source

pub async fn submit_diagnostic_result(
    _config: &Settings,
    client: &BrokkrClient,
    request_id: Uuid,
    result: SubmitDiagnosticResult,
) -> Result<(), Box<dyn std::error::Error>> {
    debug!("Submitting diagnostic result for request {}", request_id);

    let sdk_result: brokkr_client::types::SubmitDiagnosticResult =
        convert(result).map_err(|e| {
            error!("Failed to convert SubmitDiagnosticResult: {}", e);
            Box::new(e) as Box<dyn std::error::Error>
        })?;

    match client
        .api()
        .submit_diagnostic_result()
        .id(request_id)
        .body(sdk_result)
        .send()
        .await
    {
        Ok(_) => {
            info!(
                "Successfully submitted diagnostic result for request {}",
                request_id
            );
            Ok(())
        }
        Err(raw) => {
            let wrapped = BrokkrError::from(raw);
            error!(
                "Failed to submit diagnostic result for request {}: {}",
                request_id, wrapped
            );
            Err(boxed("Failed to submit diagnostic result", wrapped))
        }
    }
}