polyc-a2a 2026.9.0

//! polychrome-a2a — A2A edge binary.
//!
//! Layer config from flags + env, mint the deployment principal, build and sign
//! the Agent Card, build a control-plane-backed turn runner, then serve the A2A
//! router with a health side-server and signal-driven graceful shutdown.

#![forbid(unsafe_code)]
#![warn(missing_docs)]

use std::sync::Arc;

use anyhow::Context as _;
use clap::Parser;
use polyc_a2a::{
    AgentDialerRunner, AppState, ApprovalDialerResponder, ApprovalResponder, TaskDialerStore,
    TaskStore, TurnRunner, UnconfiguredApprovalResponder, UnconfiguredRunner,
    UnconfiguredTaskStore,
    card::{CardConfig, signed_card},
    config::{self, Cli},
    router,
};
use polyc_crypto::Signer;
use polyc_rpc_client::{Sensitive, edge_credentials_from_env_or_fail};
use polyc_runtime::{health::Health, observability, signals, supervise};
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;

#[tokio::main]
async fn main() -> sysexits::ExitCode {
    let cli = match Cli::try_parse() {
        Ok(cli) => cli,
        Err(err) => return cli_error_exit(&err),
    };

    let _ = rustls::crypto::ring::default_provider().install_default();
    let _otel_guard = observability::init("polychrome-a2a");

    match run(cli).await {
        Ok(()) => sysexits::ExitCode::Ok,
        Err(err) => {
            tracing::error!(error = ?err, "fatal error");
            sysexits::ExitCode::Software
        }
    }
}

/// Print a clap parse error and map it to the right exit code.
fn cli_error_exit(err: &clap::Error) -> sysexits::ExitCode {
    use clap::error::ErrorKind;
    let _ = err.print();
    match err.kind() {
        ErrorKind::DisplayHelp
        | ErrorKind::DisplayVersion
        | ErrorKind::DisplayHelpOnMissingArgumentOrSubcommand => sysexits::ExitCode::Ok,
        _ => sysexits::ExitCode::Usage,
    }
}

/// The three control-plane dialers this edge holds: one turn runner, one
/// approval responder, and one durable task store. All three name the same
/// address, so they are resolved together and never one without the others.
type ControlPlaneDialers = (
    Arc<dyn TurnRunner>,
    Arc<dyn ApprovalResponder>,
    Arc<dyn TaskStore>,
);

/// Build the [`TurnRunner`], [`ApprovalResponder`], and [`TaskStore`] dialing
/// the control plane at `cfg.agent_addr`.
///
/// Prefers the authenticated path — [`AgentDialerRunner::with_credentials`] /
/// [`ApprovalDialerResponder::with_bearer`] / [`TaskDialerStore::with_bearer`]
/// — when `edge_id`,
/// `edge_bearer_key`, and `edge_signing_key_hex` are all set; falls back to
/// the unauthenticated [`AgentDialerRunner::new`] / [`ApprovalDialerResponder::new`]
/// with a warning only when `cfg.agent_addr` is loopback (see
/// [`polyc_rpc_client::edge_credentials_from_env_or_fail`]) — otherwise fails
/// startup rather than dialing a remote control plane unauthenticated.
///
/// # Errors
/// Returns an error if `cfg.agent_addr` isn't a valid URI, the edge signing
/// key isn't valid hex/ed25519, the bearer can't be encoded as an HTTP
/// header value, or credentials are unconfigured against a non-loopback
/// `cfg.agent_addr`.
fn build_dialers(cfg: &config::Config) -> anyhow::Result<ControlPlaneDialers> {
    if cfg.agent_addr.is_empty() {
        tracing::warn!(
            "POLYCHROME_AGENT_ADDR is unset — the Agent Card still serves, but message/send \
                 tasks, approval decisions, and task lookups will fail closed until it is set. \
                 Set it in any environment that should run A2A turns."
        );
        return Ok((
            Arc::new(UnconfiguredRunner),
            Arc::new(UnconfiguredApprovalResponder),
            Arc::new(UnconfiguredTaskStore),
        ));
    }

    let creds = edge_credentials_from_env_or_fail(
        &cfg.agent_addr,
        // Blank-vs-unset is normalized inside the helper, so a `""` Secret
        // placeholder reads as unconfigured here exactly as it does on
        // every other edge.
        cfg.edge_id.as_deref(),
        cfg.edge_bearer_key
            .as_ref()
            .map(Sensitive::expose)
            .map(String::as_str),
        cfg.edge_signing_key_hex
            .as_ref()
            .map(Sensitive::expose)
            .map(String::as_str),
    )
    .context("build edge credentials")?;
    match creds {
        Some(creds) => {
            let bearer = creds.bearer().to_owned();
            Ok((
                Arc::new(
                    AgentDialerRunner::with_credentials(&cfg.agent_addr, creds)
                        .context("build authenticated AgentService dialer")?,
                ),
                Arc::new(
                    ApprovalDialerResponder::with_bearer(&cfg.agent_addr, &bearer)
                        .context("build authenticated ApprovalService dialer")?,
                ),
                Arc::new(
                    TaskDialerStore::with_bearer(&cfg.agent_addr, &bearer)
                        .context("build authenticated AgentTaskService dialer")?,
                ),
            ))
        }
        None => Ok((
            Arc::new(AgentDialerRunner::new(&cfg.agent_addr).context("build AgentService dialer")?),
            Arc::new(
                ApprovalDialerResponder::new(&cfg.agent_addr)
                    .context("build ApprovalService dialer")?,
            ),
            Arc::new(
                TaskDialerStore::new(&cfg.agent_addr).context("build AgentTaskService dialer")?,
            ),
        )),
    }
}

async fn run(cli: Cli) -> anyhow::Result<()> {
    let cfg = config::load(&cli).context("load configuration")?;

    let shutdown = CancellationToken::new();
    signals::spawn_handler(shutdown.clone());

    // The Agent Card endpoint serves regardless of control-plane wiring, so an
    // unset `agent_addr` must not abort startup: fall back to a runner that
    // fails `message/send` closed (matching the warning) rather than dialing an
    // address that was never configured. A *non-empty* address that won't parse
    // is an operator error worth failing fast on, so that path still aborts.
    let (runner, approvals, store) = build_dialers(&cfg)?;

    let signer = Signer::from_seed(*cfg.signing_seed.expose());
    let card = signed_card(
        &CardConfig {
            name: cfg.agent_name,
            description: cfg.agent_description,
            url: cfg.public_url,
            version: env!("CARGO_PKG_VERSION").to_owned(),
        },
        &signer,
    );

    let peers = polyc_a2a::PeerAuthenticator::parse(cfg.peer_credentials.expose())
        .context("parse POLYCHROME_A2A_PEER_CREDENTIALS")?;
    if peers.is_empty() {
        tracing::warn!(
            "POLYCHROME_A2A_PEER_CREDENTIALS is unset — the Agent Card still serves, but every \
             JSON-RPC request to POST / will fail closed with 503 until it is set. Set it in \
             any environment that should accept A2A JSON-RPC calls."
        );
    }

    let state = AppState {
        card: Arc::new(card),
        runner,
        approvals,
        store,
        turn_limit: polyc_runtime::admission::AdmissionGate::new(cfg.max_concurrent_turns.max(1)),
        peers,
    };
    let app = router(state);

    let health = Health::new();

    // Bind every listener before flipping readiness: a pod that cannot bind its
    // sockets must fail startup, not report ready while dead.
    let side_listener = TcpListener::bind(cfg.side_addr)
        .await
        .with_context(|| format!("bind side-server on {}", cfg.side_addr))?;
    let listener = TcpListener::bind(cfg.bind)
        .await
        .with_context(|| format!("bind {}", cfg.bind))?;
    tracing::info!(addr = %cfg.bind, "polychrome-a2a listening");

    let mut servers = tokio::task::JoinSet::new();
    servers.spawn({
        let health = health.clone();
        let shutdown = shutdown.clone();
        async move {
            polyc_runtime::health::serve_on(side_listener, health, shutdown)
                .await
                .context("side-server")
        }
    });
    servers.spawn({
        let shutdown = shutdown.clone();
        async move {
            axum::serve(listener, app)
                .with_graceful_shutdown(async move { shutdown.cancelled().await })
                .await
                .context("a2a server")
        }
    });
    health.set_ready(true);

    supervise::until_shutdown(servers, &health, &shutdown).await?;
    tracing::info!("shutdown complete");
    Ok(())
}