agentplane/api/

a2a.rs

//! Serving A2A: this plane, as an agent other agents can call.
//!
//! The client side ([`crate::peers::a2a`]) lets this plane *call* peers. This is
//! the other half — a peer calling us — and it is a different problem, because
//! everything arriving here came from somebody else.
//!
//! # Why this is not a route on the operator API
//!
//! Every route on [`crate::api::Api`] authenticates and then authorizes. An
//! Agent Card is public by design: it is what a caller reads *before* it has
//! credentials, and a card behind authentication cannot be discovered. Bolting
//! an unauthenticated path onto a surface whose invariant is "every route
//! authenticates" would delete that invariant for the one route nobody would
//! think to check.
//!
//! So this is its own router with its own rule: **the card is public, every
//! method call is authenticated and authorized**.
//!
//! # What arrives here is untrusted
//!
//! A message from a peer is data written by a party this plane does not control,
//! so it is admitted as `Tainted` with the sending
//! peer's identity as its provenance source — never as trusted input. A skill
//! that wants to act on it has to say so at a gate, and a protected sink field
//! can name the one counterparty it will accept an amount from.
//!
//! This is the same reason the operator API takes an event's `source` from the
//! authenticated caller rather than the body: a party describing itself is not
//! evidence about itself.
//!
//! # The capability is named, never inferred
//!
//! A2A messages do not carry a "call this skill" field — the protocol assumes an
//! agent works out what is being asked. This plane will not: choosing which
//! capability to run on the strength of an untrusted message is a dispatch
//! decision made by inference, and the thing doing the inferring would be a
//! model reading attacker-controlled text.
//!
//! The skill is taken from `message.metadata.skill`, matched against the card's
//! advertised skill ids. When the agent advertises exactly one there is nothing
//! to infer and it is used; when it advertises several and none was named, the
//! call is refused rather than guessed.
//!
//! # Optional capabilities say what is wired
//!
//! Streaming and non-terminal task subscription are durable journal views.
//! Push is advertised only after `with_push` supplies durable registration
//! storage and a retrying transport worker; without that wiring every push
//! method uses `PushNotificationNotSupportedError` and the card advertises
//! false.
//!
//! # A task belongs to the peer that admitted it
//!
//! Every read, write, stream, push configuration and `contextId` join is scoped
//! to the admitting peer; another peer's task answers `TASK_NOT_FOUND`, never a
//! refusal that confirms it exists.

use std::sync::Arc;

use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

use crate::core::{PolicyDecision, PolicyRequest, RunId, Seq, SourceId, Tainted};
use crate::journal::RecordKind;
use crate::manifest::Manifest;
use crate::peers::{AgentCard, ExtendedAgentCard, WELL_KNOWN_PATH};
use crate::push::PushNamespace;
use crate::runtime::Runtime;

use super::{Authenticator, Caller};

/// JSON-RPC method names, exactly as A2A 1.0 spells them.
///
/// 1.0 renamed these: `message/send` was the 0.3 spelling, and a server
/// answering the old names would silently accept clients that have lost half the
/// protocol. Constants rather than inline literals so the dispatch table and the
/// refusal table cannot disagree.
pub mod method {
    pub const SEND_MESSAGE: &str = "SendMessage";
    pub const GET_TASK: &str = "GetTask";
    pub const CANCEL_TASK: &str = "CancelTask";
    pub const GET_EXTENDED_CARD: &str = "GetExtendedAgentCard";

    /// Optional operations implemented by this server.
    pub const SEND_STREAMING: &str = "SendStreamingMessage";
    pub const SUBSCRIBE: &str = "SubscribeToTask";
    pub const LIST_TASKS: &str = "ListTasks";
    pub const CREATE_PUSH: &str = "CreateTaskPushNotificationConfig";
    pub const GET_PUSH: &str = "GetTaskPushNotificationConfig";
    pub const LIST_PUSH: &str = "ListTaskPushNotificationConfigs";
    pub const DELETE_PUSH: &str = "DeleteTaskPushNotificationConfig";
}

/// A2A-specific JSON-RPC error codes, from the spec's mapping table.
pub mod code {
    pub const PARSE_ERROR: i32 = -32700;
    pub const INVALID_REQUEST: i32 = -32600;
    pub const METHOD_NOT_FOUND: i32 = -32601;
    pub const INVALID_PARAMS: i32 = -32602;
    pub const INTERNAL_ERROR: i32 = -32603;

    pub const TASK_NOT_FOUND: i32 = -32001;
    pub const TASK_NOT_CANCELABLE: i32 = -32002;
    pub const PUSH_NOT_SUPPORTED: i32 = -32003;
    pub const UNSUPPORTED_OPERATION: i32 = -32004;
    pub const CONTENT_TYPE_NOT_SUPPORTED: i32 = -32005;
    pub const EXTENDED_CARD_NOT_CONFIGURED: i32 = -32007;
    pub const VERSION_NOT_SUPPORTED: i32 = -32009;

    /// Admission back-pressure: a quota ceiling refused the request before any
    /// work was admitted.
    ///
    /// Not in A2A 1.0's error table — the spec defines only permanent
    /// missing-capability codes and assigns back-pressure no code at all; its
    /// own guidance stops at "return appropriate error responses when rate
    /// limits are exceeded". Reaching into the table anyway would be worse
    /// than inventing a code: `-32004 UnsupportedOperationError` for a full
    /// quota teaches a compliant caller that the *operation does not exist
    /// here* — the correct response to which is to abandon, never to retry —
    /// when the truthful answer is "not right now". So the code is drawn from
    /// JSON-RPC's implementation-defined server-error range.
    ///
    /// The **number is not the identity**. A2A 1.0 reserves `-32001..-32099`
    /// for its own errors, the same band JSON-RPC gives implementations, so a
    /// future spec revision could assign this numeral a meaning of its own.
    /// What makes the refusal identifiable is the `google.rpc.ErrorInfo`
    /// beside it — [`ERROR_DOMAIN`](crate::peers::ERROR_DOMAIN) +
    /// [`QUOTA_EXHAUSTED_REASON`](crate::peers::QUOTA_EXHAUSTED_REASON), a
    /// domain this project controls — and that pair is what this crate's own
    /// client keys its refuse-and-come-back classification on. A bare
    /// `-32029` from a foreign server proves nothing and is classified as an
    /// unknown fault.
    ///
    /// What this code does NOT carry is the quota arithmetic: declines on this
    /// surface are deliberately uniform, and counters or limits in the message
    /// would hand an unauthenticated prober the tenant's ceilings. The message
    /// beside this code is a fixed sentence with no numbers in it.
    pub const QUOTA_EXHAUSTED: i32 = -32029;

    /// This agent is halted by its operator.
    ///
    /// Its own code, beside [`QUOTA_EXHAUSTED`] and
    /// under the same identification rule — the `(domain, reason)` pair, never
    /// the numeral — because the two ask opposite things of a caller. A ceiling
    /// says *come back*; a halt says *somebody is dealing with an incident*,
    /// and a peer told to back off and retry keeps knocking on the one door
    /// that means stop. The message is fixed and carries none of the
    /// operator's reason: the counterparty gets the outcome, not the plane's
    /// internals.
    pub const HALTED: i32 = -32030;

    /// This instance is shutting down and did not admit the request.
    ///
    /// The third admission refusal, under the same identification rule — the
    /// `(domain, reason)` pair, never the numeral. It is its own code because
    /// it is the only one of the three a caller clears by *moving*: a ceiling
    /// and a halt are facts about the agent, and this is a fact about one
    /// process serving it. A peer told to back off waits out a window it never
    /// needed; a peer told to abandon gives up on an agent that is fine.
    pub const DRAINING: i32 = -32031;
}

/// Actions this surface asks the policy engine about.
pub mod action {
    pub const MESSAGE_SEND: &str = "a2a:message.send";
    pub const TASK_READ: &str = "a2a:task.read";
    pub const TASK_CONTINUE: &str = "a2a:task.continue";
    pub const TASK_CANCEL: &str = "a2a:task.cancel";
    pub const CARD_EXTENDED: &str = "a2a:card.extended";
    /// Registering, reading or removing a webhook for a task.
    pub const TASK_PUSH: &str = "a2a:task.push";
    /// Supplying an event of a kind, asked on the kind: a continuation is
    /// delivered as the event its task awaits, so a peer may continue a task
    /// only with input a rule lets it supply.
    pub const EVENT_DELIVER: &str = "a2a:event.deliver";

    /// Every action this surface can ask about, so a deployment can enumerate
    /// what it must write rules for.
    pub const ALL: &[&str] = &[
        MESSAGE_SEND,
        TASK_READ,
        TASK_CONTINUE,
        TASK_CANCEL,
        CARD_EXTENDED,
        TASK_PUSH,
        EVENT_DELIVER,
    ];
}

/// The policy context every A2A action is asked under: roles, the peer's own
/// name, and tenant — no chain, no label, no depth. A task action adds the
/// task's `owner` ([`A2aServer::context`]).
fn peer_context(caller: &Caller) -> Value {
    json!({
        "roles": caller.roles,
        "peer": caller.actor,
        "tenant": caller.tenant.as_str(),
    })
}

/// Whether `engine` can evaluate every request this surface puts to it.
///
/// Each [`action::ALL`] verb is probed in exactly the shapes it is asked in:
/// with the task's `owner` where it acts on a task, and without it where it
/// does not — sending, the extended card, the `tasks` listing, and an inline
/// push registration for a task that does not exist yet. Probing a task action
/// without its owner would refuse a rule set that reads `context.owner`, which
/// every such request carries; probing the listing with one would miss a rule
/// that breaks on the request that carries none. [`A2aServer::hosting`]
/// refuses a runtime whose engine reports anything: a rule this surface
/// cannot evaluate declines every peer, and the decline says nothing to the
/// peer about why.
#[must_use]
pub fn policy_problems(engine: &dyn crate::core::PolicyEngine) -> Vec<String> {
    let caller = Caller::new("preflight", vec!["preflight".to_owned()]);
    let bare = peer_context(&caller);
    let owned = A2aServer::context(&caller, Some(&caller.actor));
    let shapes: &[(&str, &Value)] = &[
        (action::MESSAGE_SEND, &bare),
        (action::TASK_READ, &owned),
        (action::TASK_READ, &bare),
        (action::TASK_CONTINUE, &owned),
        (action::TASK_CANCEL, &owned),
        (action::CARD_EXTENDED, &bare),
        (action::TASK_PUSH, &owned),
        (action::TASK_PUSH, &bare),
        (action::EVENT_DELIVER, &owned),
    ];
    let requests: Vec<PolicyRequest<'_>> = shapes
        .iter()
        .map(|(action, context)| PolicyRequest {
            principal: &caller.actor,
            principal_kind: crate::core::PrincipalKind::Subject,
            action,
            resource: "preflight.resource",
            context,
        })
        .collect();
    engine.preflight(&requests)
}

/// The `A2A-Version` service parameter.
const VERSION_HEADER: &str = "a2a-version";

/// The one sentence a full quota answers with, beside
/// [`code::QUOTA_EXHAUSTED`].
///
/// A single constant rather than `why.to_string()` at each site, because the
/// quota error's own rendering names the counter and its ceiling — numbers an
/// external caller has no business learning from a decline. Deliberately
/// digit-free; the test on this path asserts exactly that.
const QUOTA_EXHAUSTED_MESSAGE: &str =
    "this agent cannot take the request on right now; retry later";

/// The one sentence a halt answers with, beside [`code::HALTED`]. No reason,
/// no scope: an incident's description is for the operator's own worklist.
const HALTED_MESSAGE: &str =
    "this agent is halted by its operator; do not retry until it is lifted";

/// The one sentence a draining instance answers with, beside [`code::DRAINING`].
///
/// No hostname, no instance id and no grace period: which process is going away
/// is this deployment's topology, and a caller's correct behaviour does not
/// depend on knowing it.
const DRAINING_MESSAGE: &str =
    "this instance is shutting down and did not take the request on; retry";

/// Which of the two admission refusals a quota error is, on the wire.
///
/// One function for the three admission sites, because a halt that reached one
/// of them as a ceiling would be the sibling-divergence shape: a peer told
/// *retry* by `message/send` and *stop* by `message/stream`.
fn quota_refusal(e: &crate::quota::QuotaError) -> RpcError {
    match e {
        crate::quota::QuotaError::Halted { .. } => RpcError::new(code::HALTED, HALTED_MESSAGE),
        _ => RpcError::new(code::QUOTA_EXHAUSTED, QUOTA_EXHAUSTED_MESSAGE),
    }
}

/// A message its agent cannot read a data subject from, on the wire: invalid
/// params at every admission site. The error names the binding and why it
/// did not resolve, never a subject.
fn subject_unbound(e: &crate::core::RuntimeError) -> RpcError {
    RpcError::new(code::INVALID_PARAMS, e.to_string())
}

/// A task's lifecycle state, as A2A names them.
///
/// `ProtoJSON` spelling — `TASK_STATE_WORKING`, not `working`. A client matching
/// on the enum gets nothing from a friendlier spelling.
///
/// # Why this is the whole vocabulary, not the part this plane produces
///
/// It is **both** an output and an input: `ListTasks` takes a `status` filter,
/// so a caller may name any state A2A defines and must get a valid, empty
/// answer rather than `INVALID_PARAMS`. A subset would turn a legitimate query
/// into a protocol error.
///
/// Two of these this plane never *produces*: `TASK_STATE_UNSPECIFIED`, the
/// protocol's zero value, and `TASK_STATE_SUBMITTED`, which means accepted but
/// not yet started — there is no such moment here, because a run starts inside
/// the call that admits it.
///
/// `TASK_STATE_AUTH_REQUIRED` is produced for a run whose authority was
/// withdrawn under it: interrupted, not terminal, so lifting the withdrawal
/// continues the same task.
///
/// `TASK_STATE_REJECTED` **is** produced, and not as a refusal: a sweep, a
/// break-glass crossing, a lifted control and an observed session take it,
/// because none is a task any peer submitted. A caller polling one of those is
/// asking about this plane's record of itself, and gets the same answer as for
/// a run id that does not exist at all.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum TaskState {
    #[serde(rename = "TASK_STATE_UNSPECIFIED")]
    Unspecified,
    #[serde(rename = "TASK_STATE_SUBMITTED")]
    Submitted,
    #[serde(rename = "TASK_STATE_WORKING")]
    Working,
    #[serde(rename = "TASK_STATE_COMPLETED")]
    Completed,
    #[serde(rename = "TASK_STATE_FAILED")]
    Failed,
    #[serde(rename = "TASK_STATE_CANCELED")]
    Canceled,
    #[serde(rename = "TASK_STATE_INPUT_REQUIRED")]
    InputRequired,
    #[serde(rename = "TASK_STATE_REJECTED")]
    Rejected,
    #[serde(rename = "TASK_STATE_AUTH_REQUIRED")]
    AuthRequired,
}

/// The A2A state a live run's status surfaces as.
///
/// These are worth stating because the obvious mapping is wrong.
///
/// A **suspended** run is `INPUT_REQUIRED`, not `WORKING`: it is stopped and
/// will not move until something external happens, and a caller polling a
/// `WORKING` task waits forever for a task that is not running.
///
/// A **quarantined** run is `FAILED`, not `REJECTED`. `REJECTED` means the agent
/// declined the work; quarantine means it accepted the work, started, and can no
/// longer be trusted to describe what it did. Reporting that as a refusal tells
/// the caller nothing happened, when something did.
///
/// A **withheld** run is `AUTH_REQUIRED`: its authority was withdrawn under
/// it, its work stands, and it continues once the authorization is restored.
/// Interrupted rather than terminal, so a stream stays open and a push
/// registration stays live. Restoring the authorization is the deployment's
/// act, not the peer's — the status message carries the outcome so nobody
/// reads it as a request for credentials.
///
/// A **failed or exhausted** run is `FAILED` although the runtime can resume
/// it: A2A has no interrupted state for "stopped until this deployment's
/// operator acts", so the answer is the run's state at read time. The stream
/// closes and the push registration is released as for any terminal task; an
/// operator's resume is a fresh act, and `GetTask` then reports it working.
///
/// This and [`sealed_state`] are the **same question asked on two paths**: this
/// one answers the caller who receives the immediate `SendMessage` response, and
/// `sealed_state` answers everyone who reads the task back — `GetTask`,
/// `SubscribeToTask`, and every streamed status update. They must agree, and
/// `a_live_status_and_its_sealed_outcome_agree` holds them to it over every
/// variant.
///
/// **This is the exhaustive one, and that is the point.** The two were unified
/// once by making this function delegate to `sealed_state`, which reads as the
/// tidier direction and is the wrong one: `sealed_state` matches *strings*
/// behind a `_ => Failed`, so delegating to it deleted the only compile-time
/// check on the mapping while leaving a comment claiming the compiler still
/// enforced it. Adding a `RunStatus` variant for something that is **not** a
/// failure — an authorization wait, a rejection — would then have compiled
/// cleanly and reported `Failed` on both paths. That is not agreement; it is
/// the same wrong answer twice, which is strictly harder to notice than two
/// different ones.
///
/// So the enum match is the definition and the string match is checked against
/// it. A new variant fails to compile *here*, which is where the decision
/// belongs.
fn state_of(status: &crate::runtime::RunStatus) -> TaskState {
    use crate::runtime::RunStatus;
    match status {
        RunStatus::Succeeded => TaskState::Completed,
        RunStatus::Suspended(_) => TaskState::InputRequired,
        RunStatus::Cancelled { .. } => TaskState::Canceled,
        RunStatus::Failed(_)
        | RunStatus::Exhausted(_)
        | RunStatus::Quarantined(_)
        // Not `Canceled`: A2A's cancellation means a client asked and the work
        // stopped, which a peer reads as "nothing of mine is standing". An
        // abandoned run is the opposite claim — something may well be standing
        // and nobody could establish what — so it takes the state a peer
        // investigates rather than the one they dismiss.
        | RunStatus::Abandoned { .. }
        | RunStatus::Replanning(_) => TaskState::Failed,
        RunStatus::Withheld { .. } => TaskState::AuthRequired,
        // Neither is a task, and `REJECTED` is the state that says so: no peer
        // submitted a sweep or a break-glass crossing, and a run id that
        // resolves to one is a caller asking about this plane's record of
        // itself. `FAILED` would claim the agent tried and could not; `CANCELED`
        // would claim somebody's work was stopped. Both invent a task where
        // there was none, and a peer reads either as being about *their*
        // request. Nothing narrows further here on purpose — which of this
        // plane's internal runs exists is not a peer's question, so the answer
        // is the same one for a run that does not exist at all.
        // An observed session joins them for the same reason and one more: it
        // is not even this plane's work. No peer submitted it, nothing here
        // executed it, and the only honest thing to tell a caller asking about
        // somebody else's agent is the answer they get for a run that is none
        // of their business.
        RunStatus::Swept
        | RunStatus::BrokeGlass { .. }
        | RunStatus::HaltLifted { .. }
        | RunStatus::HoldReleased { .. }
        | RunStatus::Observed => TaskState::Rejected,
    }
}

/// A2A's `TaskStatus`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TaskStatus {
    pub state: TaskState,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub message: Option<A2aMessage>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timestamp: Option<String>,
}

/// A2A's `Task` — what this plane calls a run.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct A2aTask {
    pub id: String,
    /// The case, when the run belongs to one.
    ///
    /// A2A's `contextId` is "the thing this conversation is part of", which is
    /// exactly a case: several runs, one matter, shared state. Mapping it to
    /// anything else would give a caller a correlation handle that does not
    /// correlate.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub context_id: Option<String>,
    pub status: TaskStatus,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub artifacts: Option<Vec<A2aArtifact>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub history: Option<Vec<A2aMessage>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<Value>,
}

/// One output produced by an A2A task.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct A2aArtifact {
    pub artifact_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    pub parts: Vec<Part>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<Value>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub extensions: Vec<String>,
}

/// One piece of a message.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Part {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub text: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub data: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub raw: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub filename: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub media_type: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<Value>,
}

impl Part {
    /// A text part.
    #[must_use]
    pub fn text(text: impl Into<String>) -> Self {
        Self {
            text: Some(text.into()),
            data: None,
            raw: None,
            url: None,
            filename: None,
            media_type: Some("text/plain".to_owned()),
            metadata: None,
        }
    }

    /// A structured-data part.
    #[must_use]
    pub fn data(data: Value) -> Self {
        Self {
            text: None,
            data: Some(data),
            raw: None,
            url: None,
            filename: None,
            media_type: Some("application/json".to_owned()),
            metadata: None,
        }
    }

    /// A file part carrying its bytes inline, base64-encoded per `ProtoJSON`.
    #[must_use]
    pub fn file_raw(
        raw_base64: impl Into<String>,
        media_type: impl Into<String>,
        filename: impl Into<String>,
    ) -> Self {
        Self {
            text: None,
            data: None,
            raw: Some(raw_base64.into()),
            url: None,
            filename: Some(filename.into()),
            media_type: Some(media_type.into()),
            metadata: None,
        }
    }

    /// A file part referring to bytes by URL.
    #[must_use]
    pub fn file_url(
        url: impl Into<String>,
        media_type: impl Into<String>,
        filename: impl Into<String>,
    ) -> Self {
        Self {
            text: None,
            data: None,
            raw: None,
            url: Some(url.into()),
            filename: Some(filename.into()),
            media_type: Some(media_type.into()),
            metadata: None,
        }
    }
}

/// How a skill shapes its A2A answer, when the default projection is not it.
///
/// The default stands for most skills: a string output becomes a text part and
/// anything else a data part, in one artifact. What that projection cannot say
/// is *file content* — inline bytes or a URL with a filename — or that the
/// answer is a quick, stateless `Message` rather than a task's artifact. Both
/// are ordinary A2A response shapes a peer may expect.
///
/// A skill opts in by returning this value (via [`A2aReply::into_value`]) as
/// its outcome. The runtime journals it as it journals any output — the shape
/// is a *projection instruction* read at the protocol boundary, not a second
/// channel around the journal.
///
/// ```no_run
/// # use agentplane::api::a2a::{A2aReply, Part};
/// # use agentplane::core::{Outcome, Tainted};
/// // A task whose artifact is a file reference:
/// let reply = A2aReply::artifact(vec![Part::file_url(
///     "https://example.com/report.pdf",
///     "application/pdf",
///     "report.pdf",
/// )]);
/// # let _ = Outcome::done(Tainted::trusted(reply.into_value()));
/// ```
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct A2aReply {
    #[serde(skip_serializing_if = "Option::is_none")]
    message: Option<Vec<Part>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    artifacts: Option<Vec<Vec<Part>>>,
}

/// The output key an [`A2aReply`] travels under.
///
/// A `$`-prefixed marker, like `$a2a_message` on the inbound side: ordinary
/// skill output is domain data this runtime never interprets, so the one shape
/// it *does* interpret must be unmistakably deliberate rather than a field
/// name a domain object happens to share.
const REPLY_KEY: &str = "$a2a_reply";

impl A2aReply {
    /// Answer the blocking send with a direct `Message` instead of a task.
    ///
    /// Outside a blocking send — a spawned task, a stream, a later `GetTask` —
    /// there is no message to deliver, and the parts become the task's
    /// artifact instead: the content survives, only the envelope differs.
    #[must_use]
    pub const fn message(parts: Vec<Part>) -> Self {
        Self {
            message: Some(parts),
            artifacts: None,
        }
    }

    /// Answer with one artifact holding these parts.
    #[must_use]
    pub fn artifact(parts: Vec<Part>) -> Self {
        Self {
            message: None,
            artifacts: Some(vec![parts]),
        }
    }

    /// Answer with several artifacts.
    #[must_use]
    pub const fn artifacts(artifacts: Vec<Vec<Part>>) -> Self {
        Self {
            message: None,
            artifacts: Some(artifacts),
        }
    }

    /// The outcome value a skill returns.
    #[must_use]
    pub fn into_value(self) -> Value {
        json!({ REPLY_KEY: self })
    }

    /// Read a reply back out of a run's output, **if the run may declare one**.
    ///
    /// A projection instruction is authority: it decides whether the answer is
    /// a task artifact or a direct `Message`, and what parts it carries —
    /// including a file part naming a URL. Skill output routinely *contains*
    /// untrusted data: a summariser quotes its input, a declarative agent's
    /// answer is a model's words, and an echoing skill returns a peer's own
    /// bytes. So the marker is honoured only from a **trusted** output.
    ///
    /// Otherwise a peer could put the marker in its message, have an ordinary
    /// echoing skill return it, and choose the envelope its own reply arrived
    /// in — a file URL of the attacker's naming, presented as the agent's
    /// answer. Model output is a proposal, never authority; so is a peer's
    /// message.
    ///
    /// An untrusted answer still reaches the caller — as the artifact content
    /// it is, rather than as an instruction about the envelope.
    pub(super) fn of_output(output: &crate::core::Tainted<Value>) -> Option<Self> {
        if output.label().is_untrusted() {
            return None;
        }
        serde_json::from_value(output.peek().get(REPLY_KEY)?.clone()).ok()
    }

    /// The message parts, when this reply is a direct `Message`.
    pub(super) fn message_parts(&self) -> Option<Vec<Part>> {
        self.message.clone()
    }

    /// Every part, for contexts that can only carry artifacts.
    pub(super) fn artifact_parts(&self) -> Vec<Vec<Part>> {
        match (&self.artifacts, &self.message) {
            (Some(artifacts), _) => artifacts.clone(),
            (None, Some(message)) => vec![message.clone()],
            (None, None) => Vec::new(),
        }
    }
}

/// A2A's `Message`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct A2aMessage {
    pub message_id: String,
    pub role: String,
    pub parts: Vec<Part>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub context_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub task_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub metadata: Option<Value>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub extensions: Vec<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub reference_task_ids: Vec<String>,
}

impl A2aMessage {
    fn validate_parts(&self, caller: &Caller) -> Result<(), RpcError> {
        if self.role != "ROLE_USER" {
            return Err(RpcError::new(
                code::INVALID_PARAMS,
                "an inbound Message role must be ROLE_USER",
            ));
        }
        // The id is half of the admission key — an empty one would fold every
        // caller's unnamed messages into a single "retry" of the first.
        if self.message_id.is_empty() {
            return Err(RpcError::new(
                code::INVALID_PARAMS,
                "messageId must be non-empty: it is what lets a retry of this \
                 message be recognised as the same message",
            ));
        }
        // The other half of the same key, bounded by the runtime's admission-key
        // ceiling. The key also carries the caller's name, so the bound is on
        // the key this caller's message makes, not on the id alone — and it is
        // the caller's message that is refused, with a code that says so,
        // rather than a fault the caller would retry unchanged.
        let key = crate::core::origin_key(&admission_source(&caller.actor), &self.message_id);
        if key.len() > crate::runtime::MAX_ADMISSION_KEY_BYTES {
            return Err(RpcError::new(
                code::INVALID_PARAMS,
                format!(
                    "messageId is too long: with the caller's identity it makes an \
                     admission key over {} bytes",
                    crate::runtime::MAX_ADMISSION_KEY_BYTES
                ),
            ));
        }
        if self.parts.is_empty() {
            return Err(RpcError::new(
                code::CONTENT_TYPE_NOT_SUPPORTED,
                "the message has no parts this agent can read; it accepts text and data parts",
            ));
        }
        for (index, part) in self.parts.iter().enumerate() {
            // Only the file modalities are refused. A `mediaType` on a text or
            // data part is a *label* the spec allows on every part type — a
            // peer sending `text/markdown` text is conformant, and the card's
            // input modes are a tailoring hint, not a per-part contract the
            // server may bounce messages on.
            if part.raw.is_some() || part.url.is_some() {
                return Err(RpcError::new(
                    code::CONTENT_TYPE_NOT_SUPPORTED,
                    format!(
                        "message.parts[{index}] is file content; this agent card advertises only text/plain and application/json"
                    ),
                ));
            }
            let variants = usize::from(part.text.is_some()) + usize::from(part.data.is_some());
            if variants != 1 {
                return Err(RpcError::new(
                    code::INVALID_PARAMS,
                    format!("message.parts[{index}] must contain exactly one of text or data"),
                ));
            }
        }
        Ok(())
    }

    /// What the runtime receives as input.
    ///
    /// A stable shape, always the same three keys, rather than a clever unwrapping
    /// that hands a skill a bare string sometimes and an object other times. A
    /// skill parsing its own input should not have to branch on how many parts
    /// the caller happened to send. `$a2a_message` preserves the exact protocol
    /// object for task-history reconstruction; `text` and `data` remain the
    /// ergonomic projections skills normally consume.
    fn to_input(&self) -> Value {
        let text: Vec<&str> = self
            .parts
            .iter()
            .filter_map(|p| p.text.as_deref())
            .collect();
        let data: Vec<Value> = self.parts.iter().filter_map(|p| p.data.clone()).collect();
        json!({
            "text": text.join("\n"),
            "data": data,
            "$a2a_message": self,
        })
    }

    /// The skill this message asks for, if it named one.
    fn requested_skill(&self) -> Option<&str> {
        self.metadata.as_ref()?.get("skill")?.as_str()
    }
}

/// A JSON-RPC 2.0 request.
#[derive(Debug, Clone, Deserialize)]
struct RpcRequest {
    #[serde(default)]
    jsonrpc: String,
    #[serde(default)]
    id: Value,
    method: String,
    #[serde(default)]
    params: Value,
}

/// The parts of `SendMessageConfiguration` this server acts on.
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SendConfiguration {
    /// Return as soon as the task exists, rather than when it finishes.
    ///
    /// Blocking is the spec's default and the default here: unset means wait.
    #[serde(default)]
    return_immediately: bool,
    #[serde(default)]
    accepted_output_modes: Vec<String>,
    #[serde(default)]
    history_length: Option<usize>,
    #[serde(default)]
    task_push_notification_config: Option<PushRequest>,
}

impl SendConfiguration {
    fn validate(&self) -> Result<(), RpcError> {
        if !self.accepted_output_modes.is_empty()
            && !self
                .accepted_output_modes
                .iter()
                .any(|mode| matches!(mode.as_str(), "text/plain" | "application/json"))
        {
            return Err(RpcError::new(
                code::CONTENT_TYPE_NOT_SUPPORTED,
                "acceptedOutputModes contains no mode this agent can produce",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PushAuthenticationRequest {
    scheme: String,
    #[serde(default)]
    credentials: Option<String>,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PushRequest {
    #[serde(default)]
    id: Option<String>,
    #[serde(default)]
    task_id: Option<String>,
    url: String,
    #[serde(default)]
    token: Option<String>,
    #[serde(default)]
    authentication: Option<PushAuthenticationRequest>,
}

/// The longest push configuration id a caller may choose, in bytes.
const MAX_PUSH_ID_LEN: usize = 128;

/// The most push configurations one task holds.
///
/// Every one is a delivery per record, so without a ceiling a peer multiplies
/// the plane's outbound traffic and store rows by however many it registers.
/// Checked before the write rather than inside it, so two registrations racing
/// for the last slot can both land; the bound is on what a peer can build up,
/// not an exact count.
const MAX_PUSH_CONFIGS_PER_TASK: usize = 10;

impl PushRequest {
    /// Refuse an id in the namespace an operator destination owns, or one
    /// longer than [`MAX_PUSH_ID_LEN`].
    ///
    /// A caller and the deployment share one push store, and the two are told
    /// apart by an id prefix. A caller allowed to write into that namespace could
    /// point one of the deployment's own destinations at an address it chose —
    /// and operator destinations are deliberately exempt from the host
    /// allowlist, HTTPS and the public-address check, because there is supposed
    /// to be no caller involved. This is the check that keeps that supposition
    /// true.
    fn validate(&self) -> Result<(), RpcError> {
        if self
            .id
            .as_deref()
            .is_some_and(|id| id.len() > MAX_PUSH_ID_LEN)
        {
            return Err(RpcError::new(
                code::INVALID_PARAMS,
                format!("a pushNotificationConfig id is at most {MAX_PUSH_ID_LEN} bytes"),
            ));
        }
        if self.id.as_deref().is_some_and(crate::push::is_operator_id) {
            return Err(RpcError::new(
                code::INVALID_PARAMS,
                format!(
                    "a pushNotificationConfig id may not begin with '{}': that namespace \
                     belongs to destinations this deployment configured for itself",
                    crate::push::OPERATOR_PREFIX
                ),
            ));
        }
        Ok(())
    }

    fn config(&self, task: RunId) -> crate::push::PushConfig {
        crate::push::PushConfig {
            id: self.id.clone().unwrap_or_else(|| format!("push-{task}")),
            task,
            url: self.url.clone(),
            token: self.token.clone().map(crate::core::Secret::new),
            authentication: self.authentication.as_ref().map(|authentication| {
                crate::push::PushAuthentication {
                    scheme: authentication.scheme.clone(),
                    credentials: crate::core::Secret::new(
                        authentication.credentials.clone().unwrap_or_default(),
                    ),
                }
            }),
        }
    }
}

/// What every method's params may carry.
///
/// One struct for every method, which is the same shape the binary's arguments
/// had before they moved onto per-verb structs: a field belonging to one method
/// was **silently accepted** by another and did nothing. On the wire that reads
/// worse than at a command line, because the caller is a stranger who cannot
/// see the source. `ListTasks` was the case that mattered — a request naming
/// `contxtId`, or the `context_id` the protocol's own conformance kit sends,
/// parsed cleanly, dropped the filter, and answered with **every** task the
/// caller may see, shaped exactly like the scoped list that was asked for.
///
/// Two mechanisms close it, and they are different questions.
/// [`deny_unknown_fields`] refuses a name this surface does not know at all
/// (`contxtId`, `task_id`), which the A2A specification licenses
/// outright: A2A §5.5 says JSON field names **MUST** be camelCase, so `context_id` is not an
/// alternative spelling but a violation. [`FIELDS_BY_METHOD`] refuses a name
/// this surface knows and *this method* does not (`pageSize` on `CancelTask`).
///
/// Neither subsumes the other.
///
/// [`deny_unknown_fields`]: https://serde.rs/container-attrs.html
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct CommonParams {
    /// A2A's opaque routing identifier.
    #[serde(default)]
    tenant: Option<String>,
    #[serde(default)]
    message: Option<A2aMessage>,
    #[serde(default)]
    id: Option<String>,
    #[serde(default)]
    configuration: Option<SendConfiguration>,
    /// `SendMessageRequest.metadata`, accepted and not interpreted.
    ///
    /// Modelled rather than ignored because unknown fields are refused, and
    /// the two are different answers: the specification defines this field, so
    /// a conforming client may send it and must not meet `-32602`. It is
    /// deliberately not read — opaque caller data has no governed meaning here,
    /// and inventing one would be a control nothing enforces. What the runtime
    /// does record about an inbound message is its authenticated sender, which
    /// is provenance rather than a field the sender chose.
    ///
    /// `dead_code` is allowed here for the one case where it is the point: the
    /// field exists so that deserialization *accepts* the name, and reading it
    /// is what would be wrong.
    #[serde(default)]
    #[allow(dead_code)]
    metadata: Option<Value>,
    #[serde(default, rename = "taskId")]
    push_task: Option<String>,
    #[serde(default)]
    url: Option<String>,
    #[serde(default)]
    token: Option<String>,
    #[serde(default)]
    authentication: Option<PushAuthenticationRequest>,
    #[serde(default, rename = "contextId")]
    context_id: Option<String>,
    #[serde(default)]
    status: Option<TaskState>,
    #[serde(default, rename = "pageSize")]
    page_size: Option<usize>,
    #[serde(default, rename = "pageToken")]
    page_token: Option<String>,
    #[serde(default, rename = "historyLength")]
    history_length: Option<usize>,
    #[serde(default, rename = "statusTimestampAfter")]
    status_timestamp_after: Option<String>,
    #[serde(default, rename = "includeArtifacts")]
    include_artifacts: bool,
}

/// A JSON-RPC error, carrying the HTTP status it should be served with.
#[derive(Debug, Clone)]
pub struct RpcError {
    code: i32,
    message: String,
    /// The id of the request being answered — `null` when it could not be read.
    id: Value,
    /// Served as HTTP 401 with a bearer challenge rather than 200.
    unauthenticated: bool,
}

impl RpcError {
    fn new(code: i32, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
            id: Value::Null,
            unauthenticated: false,
        }
    }

    /// A caller whose credential was refused: HTTP 401 with
    /// `WWW-Authenticate`, as A2A requires, so the status says *authenticate*
    /// rather than the body saying *malformed*.
    fn unauthenticated(message: impl Into<String>) -> Self {
        Self {
            unauthenticated: true,
            ..Self::new(code::INVALID_REQUEST, message)
        }
    }

    fn with_id(mut self, id: Value) -> Self {
        self.id = id;
        self
    }

    /// The machine-readable reason A2A 1.0's error-handling rules require
    /// beside an A2A-specific code, as a `google.rpc.ErrorInfo` in
    /// `error.data` — `(domain, reason)`.
    ///
    /// Derived from the code rather than declared per call site, because the
    /// two are one fact: the spec's table maps each code to exactly one reason
    /// token, and a site that could set them independently is a site that can
    /// disagree with itself. Standard JSON-RPC codes carry no reason — the
    /// spec assigns them none — so they return `None` and the error omits
    /// `data` rather than inventing a token.
    ///
    /// The spec's codes carry the spec's domain. [`code::QUOTA_EXHAUSTED`]
    /// carries this plane's own — [`ERROR_DOMAIN`](crate::peers::ERROR_DOMAIN)
    /// — and that domain is load-bearing rather than decorative: the numeric
    /// code sits in space A2A 1.0 reserves for its own future errors, so the
    /// code alone is not proof of what it means. The `(domain, reason)` pair
    /// is, and it is what this crate's client keys its back-off
    /// classification on.
    ///
    /// This adds no information a prober does not already have: the reason is
    /// a restatement of the code in the same response. The uniform-refusal
    /// rule governs what a *model* is told; this is the protocol channel to an
    /// authenticated peer.
    const fn reason(&self) -> Option<(&'static str, &'static str)> {
        const A2A: &str = "a2a-protocol.org";
        const SELF_DOMAIN: &str = crate::peers::ERROR_DOMAIN;
        const QUOTA_EXHAUSTED_REASON: &str = crate::peers::QUOTA_EXHAUSTED_REASON;
        const HALTED_REASON: &str = crate::peers::HALTED_REASON;
        const DRAINING_REASON: &str = crate::peers::DRAINING_REASON;
        match self.code {
            code::TASK_NOT_FOUND => Some((A2A, "TASK_NOT_FOUND")),
            code::TASK_NOT_CANCELABLE => Some((A2A, "TASK_NOT_CANCELABLE")),
            code::PUSH_NOT_SUPPORTED => Some((A2A, "PUSH_NOTIFICATION_NOT_SUPPORTED")),
            code::UNSUPPORTED_OPERATION => Some((A2A, "UNSUPPORTED_OPERATION")),
            code::CONTENT_TYPE_NOT_SUPPORTED => Some((A2A, "CONTENT_TYPE_NOT_SUPPORTED")),
            code::EXTENDED_CARD_NOT_CONFIGURED => Some((A2A, "EXTENDED_AGENT_CARD_NOT_CONFIGURED")),
            code::VERSION_NOT_SUPPORTED => Some((A2A, "VERSION_NOT_SUPPORTED")),
            code::QUOTA_EXHAUSTED => Some((SELF_DOMAIN, QUOTA_EXHAUSTED_REASON)),
            code::HALTED => Some((SELF_DOMAIN, HALTED_REASON)),
            code::DRAINING => Some((SELF_DOMAIN, DRAINING_REASON)),
            _ => None,
        }
    }

    /// The JSON-RPC error member, with the required `ErrorInfo` when one
    /// applies.
    fn body(&self) -> Value {
        match self.reason() {
            Some((domain, reason)) => json!({
                "code": self.code,
                "message": self.message,
                "data": [{
                    "@type": "type.googleapis.com/google.rpc.ErrorInfo",
                    "domain": domain,
                    "reason": reason,
                }],
            }),
            None => json!({ "code": self.code, "message": self.message }),
        }
    }
}

impl IntoResponse for RpcError {
    fn into_response(self) -> Response {
        // HTTP 200 with a JSON-RPC error body, except for a refused credential.
        // JSON-RPC carries its own error channel, and a transport-level status
        // for an application-level refusal is how a client ends up retrying a
        // permanent decline: an A2A client reads `error.code`, and many treat
        // a 5xx as retryable without ever parsing the body. Authentication is
        // the transport's question, and A2A answers it with 401.
        let body = Json(json!({
            "jsonrpc": "2.0",
            "id": self.id,
            "error": self.body(),
        }));
        if self.unauthenticated {
            return (
                StatusCode::UNAUTHORIZED,
                [(axum::http::header::WWW_AUTHENTICATE, "Bearer")],
                body,
            )
                .into_response();
        }
        (StatusCode::OK, body).into_response()
    }
}

/// This plane, served as an A2A agent.
#[derive(Clone)]
pub struct A2aServer {
    runtime: Arc<Runtime>,
    auth: Arc<dyn Authenticator>,
    policy: Arc<dyn crate::core::PolicyEngine>,
    card: AgentCard,
    extended: ExtendedAgentCard,
    /// Each agent's own card, by agent name.
    per_agent: std::collections::BTreeMap<String, crate::peers::AgentCard>,
    /// The card's advertised skill ids — what a caller may ask for.
    skills: Vec<String>,
    push: Option<PushRuntime>,
    /// How many of the caller's own tasks one `ListTasks` may examine.
    ///
    /// See [`Self::filter_scan_budget`] for why this is a ceiling and not a
    /// page size.
    filter_scan_budget: usize,
    /// How many strict replays one `ListTasks` with `includeArtifacts` may
    /// perform. See [`Self::artifact_replay_budget`].
    artifact_replay_budget: usize,
    /// Artifact projections of **sealed** runs, so a poll loop does not buy a
    /// strict replay per read. See [`ArtifactCache`].
    artifact_cache: Arc<std::sync::Mutex<ArtifactCache>>,
    /// Open streams, per caller. See [`Self::streams_per_caller`].
    streams: StreamSlots,
}

/// Default for [`A2aServer::streams_per_caller`].
const STREAMS_PER_CALLER: usize = 8;

/// How many streams each caller holds open on this server.
#[derive(Debug, Clone)]
struct StreamSlots {
    limit: usize,
    open: Arc<std::sync::Mutex<std::collections::HashMap<String, usize>>>,
}

impl StreamSlots {
    fn new(limit: usize) -> Self {
        Self {
            limit,
            open: Arc::default(),
        }
    }

    /// One more stream for `actor`, or `None` at its ceiling.
    fn claim(&self, actor: &str) -> Option<StreamSlot> {
        let mut open = crate::core::poison::recover(&self.open);
        let held = open.entry(actor.to_owned()).or_default();
        if *held >= self.limit {
            return None;
        }
        *held += 1;
        Some(StreamSlot {
            open: Arc::clone(&self.open),
            actor: actor.to_owned(),
        })
    }
}

/// One open stream, given back when the stream is dropped — whether it ended
/// or its client went away.
#[derive(Debug)]
pub(super) struct StreamSlot {
    open: Arc<std::sync::Mutex<std::collections::HashMap<String, usize>>>,
    actor: String,
}

impl Drop for StreamSlot {
    fn drop(&mut self) {
        let mut open = crate::core::poison::recover(&self.open);
        if let Some(held) = open.get_mut(&self.actor) {
            *held = held.saturating_sub(1);
            if *held == 0 {
                open.remove(&self.actor);
            }
        }
    }
}

/// A bounded in-process cache of sealed runs' artifact projections.
///
/// A completed task's artifacts exist only as a projection of its journal, so
/// every read recovers them by **strict replay** — the most expensive read
/// this surface performs. The cache is sound because it holds only sealed
/// runs: a seal is the last record a run ever gets, so the projection is a
/// pure function of an immutable history and can never go stale. Bounded and
/// evicting oldest-first, because an unbounded cache is a leak with a
/// performance story; insertion order is enough where every entry is equally
/// immutable.
#[derive(Debug, Default)]
pub(super) struct ArtifactCache {
    map: std::collections::HashMap<RunId, Option<Vec<A2aArtifact>>>,
    order: std::collections::VecDeque<RunId>,
}

/// How many sealed runs' artifact projections stay cached per server.
const ARTIFACT_CACHE_CAPACITY: usize = 256;

/// What one bounded artifact read produced.
enum ArtifactRead {
    /// The projection — from the cache or from one counted replay.
    Artifacts(Option<Vec<A2aArtifact>>),
    /// The request's replay budget is spent; the caller must say so.
    OverBudget,
}

impl ArtifactCache {
    /// A hit, already wrapped as the read it answers.
    fn get(&self, run: RunId) -> Option<ArtifactRead> {
        self.map
            .get(&run)
            .map(|artifacts| ArtifactRead::Artifacts(artifacts.clone()))
    }

    fn insert(&mut self, run: RunId, artifacts: Option<Vec<A2aArtifact>>) {
        if self.map.insert(run, artifacts).is_none() {
            self.order.push_back(run);
        }
        while self.map.len() > ARTIFACT_CACHE_CAPACITY {
            let Some(evicted) = self.order.pop_front() else {
                break;
            };
            self.map.remove(&evicted);
        }
    }
}

#[derive(Debug, Clone)]
struct PushRuntime {
    store: Arc<dyn crate::push::PushStore>,
    transport: Arc<dyn crate::push::PushTransport>,
}

/// Durable A2A webhook delivery, driven by an operator scheduler.
///
/// A name, not a type of its own: [`A2aServer::push_worker`] hands back the
/// [`DeliveryWorker`](crate::push::DeliveryWorker) itself, bound to the A2A
/// projection. The struct this alias replaced was a pure delegation shell — its
/// own docs said so — that re-stated the loop's retry ceiling and its default
/// in a second place, which is two copies of one contract with nothing holding
/// them together. The cursor discipline lives in `push` because it has nothing
/// to do with A2A; what is A2A about this worker is only the projection it was
/// constructed with.
pub type A2aPushWorker = crate::push::DeliveryWorker;

/// Outcome of one bounded push sweep.
///
/// Re-exported from [`crate::push`], which owns the delivery loop.
pub use crate::push::PushSweepReport;

/// `StreamResponse` payloads, for **caller-registered** webhooks only.
///
/// It claims exactly the registrations an operator destination does not — see
/// [`crate::push::Outbox`] on why the two share one store and must not serve
/// each other's rows.
#[derive(Clone)]
struct A2aProjection {
    runtime: Arc<Runtime>,
}

impl std::fmt::Debug for A2aProjection {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("A2aProjection").finish_non_exhaustive()
    }
}

#[async_trait::async_trait]
impl crate::push::Projection for A2aProjection {
    async fn messages(
        &self,
        record: &crate::journal::Record,
    ) -> Result<Vec<crate::push::PushMessage>, crate::core::StoreError> {
        let case = record.body.case.map(|case| case.to_string());
        let payloads =
            super::a2a_stream::payloads_for_record(&self.runtime, record, case.as_deref())
                .await
                // A projection failure is this plane's own bug and always transient
                // to the worker; the shape it travels in is the store's error type
                // because that is what the seam speaks.
                .map_err(|error| crate::core::StoreError::Backend(error.to_string()))?;
        // One record can produce a status event and an artifact event, so the
        // record's sequence alone is not an identity. The id a receiver
        // deduplicates on is the task, the sequence and the position within
        // the record — stable across every retry of the same message, distinct
        // between the two.
        Ok(payloads
            .into_iter()
            .enumerate()
            .map(|(index, payload)| {
                crate::push::PushMessage::json(
                    format!("{}/{}/{index}", record.body.run, record.body.seq),
                    payload,
                )
                // A2A's media type, so a receiver routes push and streaming
                // through the same parser.
                .typed("application/a2a+json")
            })
            .collect())
    }

    fn terminal(&self, record: &crate::journal::Record) -> bool {
        ends_task(record.kind())
    }

    fn namespace(&self) -> PushNamespace {
        PushNamespace::Caller
    }
}

/// Whether a record ends the task a peer is watching.
///
/// A conclusion in a terminal A2A state, and nothing else: a withheld run has
/// concluded in the journal and is interrupted on this wire, so releasing its
/// push registration there would leave the peer to learn how it ends from
/// nobody.
fn ends_task(kind: &RecordKind) -> bool {
    matches!(kind, RecordKind::RunConcluded { outcome, .. } if outcome_ends_task(outcome))
}

fn outcome_ends_task(outcome: &str) -> bool {
    super::a2a_stream::closes(sealed_state(outcome))
}

impl std::fmt::Debug for A2aServer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("A2aServer")
            .field("agent", &self.card.name)
            .field("skills", &self.skills)
            .finish_non_exhaustive()
    }
}

/// Why a server could not be built.
#[derive(Debug, thiserror::Error)]
pub enum ServerSetupError {
    #[error(
        "the runtime has no policy engine, so every A2A method would be \
         unauthorized. A surface reachable by other agents cannot be the one \
         place that skips the gate"
    )]
    NoPolicy,
    /// The policy set cannot evaluate a request this surface asks. A2A asks
    /// under `{roles, peer, tenant}` alone, so an unscoped rule reading
    /// `context.delegation_depth` or `context.label` unguarded declines every
    /// peer.
    #[error(
        "the policy set cannot evaluate the A2A surface's requests: {problems} — \
         they carry only `roles`, `peer` and `tenant`; scope the rule to the \
         actions it is about, or guard the read with `context has …`"
    )]
    PolicyUnevaluable { problems: String },
    #[error(
        "the runtime has no case layer, so this server cannot mint the \
         contextId A2A 1.0 requires on every task — a generated contextId \
         must be continuable, and continuation here is a case. Build the \
         runtime with `.cases(store)`"
    )]
    NoCases,
    #[error("the agent card could not be derived: {0}")]
    Card(#[from] crate::manifest::ManifestError),
    #[error("push changes the signed Agent Card; configure it before calling signing_cards_with")]
    CardAlreadySigned,
    #[error(
        "an A2A server serves at least one agent, and no manifest was given — the well-known card path must answer with a card describing something"
    )]
    NoAgents,
    #[error(
        "agents '{first}' and '{second}' both advertise skill '{skill}', so a request naming it would be a routing decision the caller did not make. A2A dispatch is named, never inferred: give the skill distinct capability names, or serve the two agents from separate planes"
    )]
    AmbiguousSkill {
        skill: String,
        first: String,
        second: String,
    },
}

impl A2aServer {
    /// Serve `manifest`'s agent at `url`.
    ///
    /// `url` is where callers reach this plane, and it goes on the card — the
    /// same deployment-wiring split as everywhere else: an agent's declaration
    /// must not change when its address does.
    ///
    /// # Errors
    ///
    /// [`ServerSetupError::NoPolicy`] when the runtime has no policy engine, and
    /// [`ServerSetupError::Card`] when the manifest's digest cannot be computed.
    pub fn new(
        runtime: Arc<Runtime>,
        auth: Arc<dyn Authenticator>,
        security: &crate::peers::CardSecurity,
        manifest: &Manifest,
        url: impl Into<String>,
    ) -> Result<Self, ServerSetupError> {
        Self::hosting(runtime, auth, security, &[manifest], url)
    }

    /// Serve several declared agents from one plane.
    ///
    /// A2A's well-known card path is singular per host, so the alternative to
    /// this is a server per agent. The first manifest is the one the **well-known card
    /// describes** — a room's orchestrator, in the shape the CLI already uses —
    /// and every agent additionally gets its own full card at
    /// [`agent_card_path`](crate::peers::agent_card_path), listed in the
    /// [`EXT_AGENT_DIRECTORY`](crate::peers::EXT_AGENT_DIRECTORY) extension so
    /// a caller can find them.
    ///
    /// Skill dispatch spans every agent, because they are all on the runtime
    /// already — what was missing was only discovery. Two agents advertising
    /// one skill id is **refused**: dispatch names a skill, and a name that
    /// resolves to two agents is a routing decision the caller did not make.
    ///
    /// # Errors
    ///
    /// [`ServerSetupError::NoPolicy`], [`ServerSetupError::NoCases`],
    /// [`ServerSetupError::NoAgents`] for an empty slice, or
    /// [`ServerSetupError::AmbiguousSkill`].
    pub fn hosting(
        runtime: Arc<Runtime>,
        auth: Arc<dyn Authenticator>,
        security: &crate::peers::CardSecurity,
        manifests: &[&Manifest],
        url: impl Into<String>,
    ) -> Result<Self, ServerSetupError> {
        let policy = runtime.policy().ok_or(ServerSetupError::NoPolicy)?.clone();
        let mut problems = policy_problems(policy.as_ref());
        // A peer with no chain of its own acts under none on this plane, so
        // the runtime's requests for it take a shape the build did not probe.
        problems.extend(runtime.served_policy_problems());
        if !problems.is_empty() {
            return Err(ServerSetupError::PolicyUnevaluable {
                problems: problems.join("; "),
            });
        }
        if runtime.cases().is_none() {
            return Err(ServerSetupError::NoCases);
        }
        let [primary, rest @ ..] = manifests else {
            return Err(ServerSetupError::NoAgents);
        };
        let url = url.into();

        // Every agent's own card, derived exactly as a lone agent's would be —
        // same digest, same skills, same ceilings. A card that differed because
        // its agent shared a plane would make the plane part of the identity a
        // consumer pins, which the room work already refused for manifests.
        let mut directory = Vec::new();
        let mut per_agent = std::collections::BTreeMap::new();
        let mut owner_of_skill: std::collections::BTreeMap<String, String> =
            std::collections::BTreeMap::new();
        for m in manifests {
            let name = m.metadata.name.clone();
            let mut agent_card = AgentCard::derive(m, url.clone())?;
            security.apply(&mut agent_card);
            for skill in &agent_card.skills {
                if let Some(other) = owner_of_skill.insert(skill.id.clone(), name.clone())
                    && other != name
                {
                    return Err(ServerSetupError::AmbiguousSkill {
                        skill: skill.id.clone(),
                        first: other,
                        second: name,
                    });
                }
            }
            directory.push(serde_json::json!({
                "name": name,
                "version": m.metadata.version,
                "cardPath": crate::peers::agent_card_path(&name),
                "manifestDigest": m.digest()?.to_hex(),
                "skills": agent_card.skills.iter().map(|s| s.id.clone()).collect::<Vec<_>>(),
            }));
            per_agent.insert(name, agent_card);
        }

        let mut card = AgentCard::derive(primary, url.clone())?;
        let mut extended = ExtendedAgentCard::derive(primary, url)?;
        security.apply(&mut card);
        security.apply(&mut extended.public);

        // Only when there is more than one, so a single-agent plane's card is
        // byte-for-byte what it was: an extension nobody needs is a claim a
        // verifier has to understand for nothing.
        if !rest.is_empty() {
            let ext = crate::peers::AgentExtension {
                uri: crate::peers::EXT_AGENT_DIRECTORY.to_owned(),
                description: Some(
                    "Every agent this plane serves, and where each agent's own card is.".to_owned(),
                ),
                required: false,
                params: Some(serde_json::json!({ "agents": directory })),
            };
            card.capabilities.extensions.push(ext.clone());
            extended.public.capabilities.extensions.push(ext);
        }

        // The union, because every agent is already on the runtime and a skill
        // that dispatches but is not accepted here would be a refusal the plane
        // could have answered.
        let skills = owner_of_skill.keys().cloned().collect();
        let tenant = runtime.tenant().to_string();
        let mut server = Self {
            runtime,
            auth,
            policy,
            card,
            extended,
            per_agent,
            skills,
            push: None,
            filter_scan_budget: FILTER_SCAN_BUDGET,
            artifact_replay_budget: ARTIFACT_REPLAY_BUDGET,
            artifact_cache: Arc::new(std::sync::Mutex::new(ArtifactCache::default())),
            streams: StreamSlots::new(STREAMS_PER_CALLER),
        };
        // A card names the tenant only when there is one to route on. A2A's
        // rule is that a client echoes this value back in every request, so
        // advertising `default` would make every caller send a routing
        // identifier that routes nowhere.
        if tenant != crate::core::TenantId::DEFAULT {
            for card in server.published_cards() {
                for iface in &mut card.supported_interfaces {
                    iface.tenant = Some(tenant.clone());
                }
            }
        }
        Ok(server)
    }

    /// Every card this server publishes: the well-known card, the extended
    /// card's public half, and each agent's own.
    ///
    /// The deployment facts — tenant, push capability, signature — go on all of
    /// them through this one iterator. An agent's card is fetched and pinned
    /// exactly as the well-known one is, so one that lacked the tenant would
    /// route a client nowhere, one that lacked the push flag would hide a
    /// capability the server honours, and one left unsigned would be the copy
    /// an attacker substitutes.
    fn published_cards(&mut self) -> impl Iterator<Item = &mut AgentCard> {
        [&mut self.card, &mut self.extended.public]
            .into_iter()
            .chain(self.per_agent.values_mut())
    }

    /// Bound how many streams one caller may hold open at once.
    ///
    /// A `SendStreamingMessage` or `SubscribeToTask` is a connection held for
    /// as long as the run lives and a journal read per poll interval — so an
    /// unbounded count is a cost any authenticated peer chooses for this
    /// plane. Past the ceiling a new stream is refused as back-pressure
    /// ([`code::QUOTA_EXHAUSTED`]), before anything is admitted; one ending,
    /// or its client going away, frees the slot.
    #[must_use]
    pub fn streams_per_caller(mut self, limit: usize) -> Self {
        self.streams = StreamSlots::new(limit.max(1));
        self
    }

    /// Bound what one `ListTasks` may cost.
    ///
    /// The spec's `totalSize` is the exact total, and a `status` or `contextId`
    /// filter is decided only by reading each candidate's journal — so the
    /// cost of a listing is one the *caller* chooses.
    ///
    /// The budget counts the caller's own tasks. The index a listing reads is
    /// narrowed to them, so another caller's volume — or the embedder's — does
    /// not refuse this one's listing.
    ///
    /// Over budget, the request is refused with the narrowing lever named —
    /// `statusTimestampAfter` is answered from the index, so tightening it
    /// shrinks the candidate set without reading anything. A refusal is honest
    /// where a truncated total would be a lie shaped like an answer: the spec
    /// requires the exact count, and a bound that quietly stopped counting
    /// would report a smaller tenant, not a bounded scan.
    #[must_use]
    pub fn filter_scan_budget(mut self, budget: usize) -> Self {
        self.filter_scan_budget = budget.max(1);
        self
    }

    /// Bound what one `ListTasks` with `includeArtifacts` may cost.
    ///
    /// A completed task's artifacts are recovered by strict replay, and a page
    /// holds up to a hundred tasks — so without a ceiling, one authenticated
    /// request buys a hundred full replays, per request, forever. Replays
    /// served from the sealed-run cache cost nothing and do not count; past
    /// the budget a task's artifacts are **omitted and marked** (see
    /// [`ARTIFACTS_OMITTED_KEY`]) rather than the request refused, because
    /// unlike `totalSize` an artifact list is per-task enrichment: omitting
    /// one is honest as long as it does not look complete, and `GetTask` on
    /// the marked id recovers exactly that task's artifacts.
    #[must_use]
    pub fn artifact_replay_budget(mut self, budget: usize) -> Self {
        self.artifact_replay_budget = budget.max(1);
        self
    }

    /// The single-task read, where one replay is the request's stated cost.
    ///
    /// A separate entry point rather than `artifacts_bounded(.., None)`,
    /// because the caller that passes no budget cannot be handed
    /// [`ArtifactRead::OverBudget`] — and saying that with `unreachable!` in a
    /// request handler makes a caller's argument the only thing between a
    /// JSON-RPC method and a panic. Here the type says it instead.
    async fn artifacts_unmetered(
        &self,
        run: RunId,
        state: TaskState,
    ) -> Result<Option<Vec<A2aArtifact>>, crate::core::RuntimeError> {
        match self.artifacts_bounded(run, state, None).await? {
            ArtifactRead::Artifacts(artifacts) => Ok(artifacts),
            // Unreachable by construction — no budget, nothing to exceed — and
            // answered rather than asserted: an empty artifact list is what a
            // task with none looks like, which is the honest reading of "this
            // read produced none".
            ArtifactRead::OverBudget => Ok(None),
        }
    }

    /// A task's artifacts, replaying at most once per sealed run.
    ///
    /// Consults [`ArtifactCache`] first and fills it for completed runs; the
    /// uncached fall-through is [`task_artifacts`]. `budget` is decremented on
    /// every actual replay; `None` means unmetered, which reaches here only
    /// through [`artifacts_unmetered`](Self::artifacts_unmetered).
    ///
    /// A budget hit comes back as [`ArtifactRead::OverBudget`], never as an
    /// absent list, so the caller must decide how to mark it — silence here
    /// would be a bounded result shaped exactly like a complete one.
    async fn artifacts_bounded(
        &self,
        run: RunId,
        state: TaskState,
        budget: Option<&mut usize>,
    ) -> Result<ArtifactRead, crate::core::RuntimeError> {
        if state != TaskState::Completed {
            return Ok(ArtifactRead::Artifacts(None));
        }
        if let Some(cached) = crate::core::poison::recover(&self.artifact_cache).get(run) {
            return Ok(cached);
        }
        if let Some(budget) = &budget
            && **budget == 0
        {
            return Ok(ArtifactRead::OverBudget);
        }
        let artifacts = task_artifacts(&self.runtime, run, state).await?;
        if let Some(budget) = budget {
            *budget -= 1;
        }
        crate::core::poison::recover(&self.artifact_cache).insert(run, artifacts.clone());
        Ok(ArtifactRead::Artifacts(artifacts))
    }

    /// Publish a **signed** card.
    ///
    /// The card is served unauthenticated from a host a caller may not control.
    /// TLS says the bytes came from that host; it says nothing about whether the
    /// host is the party whose capabilities the card describes. A signature says
    /// that, and it keeps saying it after the card has been copied into a
    /// registry, a cache, or somebody's repository.
    ///
    /// Signed here rather than at derivation because the signature covers the
    /// **published** card, interface URL and tenant included — those are
    /// deployment facts, and a signature taken before they were set would cover
    /// a document nobody serves.
    ///
    /// # Errors
    ///
    /// If the card cannot be canonicalized.
    pub fn signing_cards_with(
        mut self,
        signer: &dyn crate::peers::CardSigner,
    ) -> Result<Self, crate::peers::CardSignatureError> {
        for card in self.published_cards() {
            card.sign(signer)?;
        }
        Ok(self)
    }

    /// Enable A2A push configuration and expose a durable delivery worker.
    ///
    /// The worker uses the task journal itself as its outbox. A registration's
    /// cursor advances only after a receiver returns 2xx, so a crash between
    /// POST and acknowledgement persistence repeats an event and never loses
    /// one. Call [`A2aServer::push_worker`] before consuming the server into its
    /// router and schedule
    /// [`DeliveryWorker::run_once`](crate::push::DeliveryWorker::run_once) from
    /// every instance.
    pub fn with_push(
        mut self,
        store: Arc<dyn crate::push::PushStore>,
        transport: Arc<dyn crate::push::PushTransport>,
    ) -> Result<Self, ServerSetupError> {
        if self
            .published_cards()
            .any(|card| !card.signatures.is_empty())
        {
            return Err(ServerSetupError::CardAlreadySigned);
        }
        for card in self.published_cards() {
            card.capabilities.push_notifications = true;
        }
        self.push = Some(PushRuntime { store, transport });
        Ok(self)
    }

    /// A cloneable worker handle, when push is configured.
    ///
    /// The worker is [`crate::push::DeliveryWorker`] itself, bound to the A2A
    /// projection — rows belonging to an operator
    /// [`Outbox`](crate::push::Outbox) are left alone, because the projection
    /// declares the caller namespace and the store's due query filters on it.
    /// Retry ceiling, backoff and abandonment reporting are the loop's own;
    /// see its docs rather than a restatement here.
    #[must_use]
    pub fn push_worker(&self) -> Option<A2aPushWorker> {
        self.push.clone().map(|push| {
            crate::push::DeliveryWorker::new(
                Arc::clone(self.runtime.journal()),
                push.store,
                push.transport,
                Arc::new(A2aProjection {
                    runtime: Arc::clone(&self.runtime),
                }),
            )
        })
    }

    /// The router.
    ///
    /// The card path is unauthenticated by design and everything else is not.
    ///
    /// The RPC endpoint answers with and without a trailing slash, and that is
    /// an interoperability fact rather than a courtesy: mainstream HTTP clients
    /// that take a base URL — httpx among them, which is what the official A2A
    /// conformance kit and the reference Python SDK are built on — resolve a
    /// request for `"/"` against the card's interface URL per RFC 3986, so the
    /// wire carries `POST {interface}/`. A router that 404s the slash form
    /// passes every test written here and fails the first real peer. Exactness
    /// elsewhere (method names, version headers) guards protocol *semantics*;
    /// a trailing slash is a client-side join artifact with none.
    pub fn router(self) -> Router {
        Router::new()
            .route(WELL_KNOWN_PATH, get(agent_card))
            // Unauthenticated like the well-known card, and for the same
            // reason: a card a caller must already be trusted to read is a card
            // that cannot be discovered.
            .route("/agents/{agent}/agent-card.json", get(one_agent_card))
            .route("/a2a", post(rpc))
            .route("/a2a/", post(rpc))
            .with_state(self)
    }

    /// Find the task, and refuse it unless this caller admitted it — then
    /// authorize, with the owner in the rule's context.
    ///
    /// A task id is a bearer of nothing. Policy decides what a caller may do
    /// with *its* tasks; whose task an id names is decided here, before policy
    /// is asked, and a task another peer admitted — or one nobody admitted over
    /// this surface — answers exactly as a task that does not exist. Anything
    /// else tells a peer holding a guessed or leaked id that it is real.
    ///
    /// Hands back the task's journal, read once, because every caller of this
    /// was about to read it anyway.
    async fn gate_task(
        &self,
        caller: &Caller,
        action: &str,
        params: &CommonParams,
    ) -> Result<(RunId, Vec<crate::journal::Record>), RpcError> {
        let run = task_id(params)?;
        let records = self.authorized_task(caller, action, run).await?;
        Ok((run, records))
    }

    /// [`gate_task`](Self::gate_task) for a task id already parsed.
    async fn authorized_task(
        &self,
        caller: &Caller,
        action: &str,
        run: RunId,
    ) -> Result<Vec<crate::journal::Record>, RpcError> {
        let records = self
            .runtime
            .journal()
            .read(run, 1)
            .await
            .map_err(|e| internal("reading a task's journal", &e))?;
        let owner = task_owner(&records);
        if owner != Some(caller.actor.as_str()) {
            return Err(task_not_found(run));
        }
        self.authorize(caller, action, &run.to_string(), owner)?;
        Ok(records)
    }

    /// Who is calling, from the credential — and only if it is this tenant.
    async fn authenticate(&self, headers: &HeaderMap) -> Result<Caller, RpcError> {
        let caller = self
            .auth
            .authenticate(headers)
            .await
            .map_err(|e| RpcError::unauthenticated(e.to_string()))?;
        // The caller's tenant is the second half of the check `check_tenant`
        // starts. That one compares what the *request* asked for against what
        // the card advertises; this compares it against what the *credential*
        // says. A peer that authenticates into one tenant must not be served
        // from another's runs by naming it in a field, and a peer holding a
        // valid credential for a different tenant is exactly who would try.
        // Refused as a credential, in the same sentence as any other this
        // endpoint does not accept: naming the tenant would tell a prober the
        // token is valid somewhere else.
        if caller.tenant != *self.runtime.tenant() {
            return Err(RpcError::unauthenticated(
                crate::api::AuthError::Rejected.to_string(),
            ));
        }
        Ok(caller)
    }

    /// What a rule on this surface can key on.
    ///
    /// `owner` is present on a task action: the peer that admitted the task,
    /// which by the time a rule is asked is always the caller — so a rule set
    /// can say "a peer reads its own tasks" in its own words rather than trust
    /// this surface to have said it.
    fn context(caller: &Caller, owner: Option<&str>) -> Value {
        let mut context = peer_context(caller);
        if let Some(owner) = owner {
            context["owner"] = json!(owner);
        }
        context
    }

    /// The policy's answer for this caller, action and resource — the one
    /// request both [`authorize`](Self::authorize) and
    /// [`permits`](Self::permits) ask.
    fn decision(
        &self,
        caller: &Caller,
        action: &str,
        resource: &str,
        owner: Option<&str>,
    ) -> PolicyDecision {
        let context = Self::context(caller, owner);
        self.policy.authorize(&PolicyRequest {
            principal: &caller.actor,
            principal_kind: crate::core::PrincipalKind::Subject,
            action,
            resource,
            context: &context,
        })
    }

    fn authorize(
        &self,
        caller: &Caller,
        action: &str,
        resource: &str,
        owner: Option<&str>,
    ) -> Result<(), RpcError> {
        match self.decision(caller, action, resource, owner) {
            PolicyDecision::Permit => Ok(()),
            PolicyDecision::Deny { reason } => {
                // The determining policy and its reason stay operator-side. A
                // Cedar denial names the action, the resource, and the policy
                // ids that fired; returned to an external caller that is a
                // probe-able map of the authorization vocabulary. A decline
                // carries no reason on the wire — the runtime's own denial
                // already names the action and resource the gate keyed on. The
                // caller learns only that it was declined; the reason reaches
                // whoever runs the plane.
                tracing::warn!(
                    target: "agentplane::a2a",
                    action,
                    resource,
                    reason,
                    "A2A request denied at admission"
                );
                Err(RpcError::new(
                    code::INVALID_REQUEST,
                    "this request was not permitted",
                ))
            }
            // The rules could not be evaluated. The caller is declined
            // exactly as for a denial — telling a peer that the plane's
            // policy set is broken is a fact about our operations, and a
            // probe would learn which shapes break it — but the operator's
            // side says defect rather than refusal, because the fix is the
            // policy set and nobody should be reading rules looking for the
            // one that fired.
            PolicyDecision::Malformed { reason } => {
                tracing::error!(
                    target: "agentplane::a2a",
                    action,
                    resource,
                    policy_error = true,
                    reason,
                    "A2A request refused because the policy set could not be evaluated"
                );
                Err(RpcError::new(
                    code::INVALID_REQUEST,
                    "this request was not permitted",
                ))
            }
        }
    }

    fn permits(&self, caller: &Caller, action: &str, resource: &str, owner: Option<&str>) -> bool {
        matches!(
            self.decision(caller, action, resource, owner),
            PolicyDecision::Permit
        )
    }

    /// Refuse a version this server does not speak.
    ///
    /// An **absent** header is a refusal, not a default. The spec says an empty
    /// value means 0.3, so a missing header is a 0.3 client — and answering it
    /// with 1.0 semantics is how a caller silently loses half the protocol.
    /// Matching is on `Major.Minor`, as the spec requires.
    fn check_version(headers: &HeaderMap) -> Result<(), RpcError> {
        let claimed = headers
            .get(VERSION_HEADER)
            .and_then(|v| v.to_str().ok())
            .unwrap_or("");
        let claimed_version = crate::peers::protocol_major_minor(claimed);
        if claimed_version == crate::peers::protocol_major_minor(crate::peers::PROTOCOL_VERSION)
            && claimed_version.is_some()
        {
            return Ok(());
        }
        let seen = if claimed.is_empty() {
            "0.3 (no A2A-Version header, which the spec reads as 0.3)".to_owned()
        } else {
            claimed.to_owned()
        };
        Err(RpcError::new(
            code::VERSION_NOT_SUPPORTED,
            format!(
                "this agent speaks A2A {}, and the request asked for {seen}",
                crate::peers::PROTOCOL_VERSION
            ),
        ))
    }

    /// Refuse a request routed to a different tenant.
    ///
    /// A2A's rule is that the client echoes the `tenant` from the interface it
    /// selected on the card, omitting it when the card omits it. So a value that
    /// does not match ours is a request meant for somebody else — plausibly
    /// another plane behind the same address — and answering it would serve one
    /// tenant's caller from another tenant's runs.
    fn check_tenant(&self, params: &CommonParams) -> Result<(), RpcError> {
        let ours = self.runtime.tenant().as_str();
        let advertised = if ours == crate::core::TenantId::DEFAULT {
            ""
        } else {
            ours
        };
        let sent = params.tenant.as_deref().unwrap_or("");
        if sent == advertised {
            return Ok(());
        }
        Err(RpcError::new(
            code::INVALID_PARAMS,
            format!(
                "this endpoint serves the tenant advertised on its card, and \
                 the request named '{sent}'. A2A clients echo the `tenant` from \
                 the interface they selected; a different value is a request \
                 for a different agent"
            ),
        ))
    }
}

/// The public Agent Card.
///
/// Unauthenticated on purpose — see the module docs. It is derived from the
/// manifest, so it cannot describe a capability the plane would refuse to
/// dispatch, which is what makes publishing it safe.
async fn agent_card(State(server): State<A2aServer>) -> Json<AgentCard> {
    Json(server.card)
}

/// One agent's own card.
///
/// A 404 for an unknown name, deliberately naming nothing: the directory
/// extension on the well-known card is how a caller learns which names exist,
/// and answering an unknown one with the list would make this path a way to
/// enumerate a plane without reading the card that governs it.
async fn one_agent_card(
    State(server): State<A2aServer>,
    axum::extract::Path(agent): axum::extract::Path<String>,
) -> Response {
    server.per_agent.get(&agent).map_or_else(
        || (StatusCode::NOT_FOUND, "no such agent on this plane").into_response(),
        |card| Json(card.clone()).into_response(),
    )
}

async fn rpc(
    State(server): State<A2aServer>,
    headers: HeaderMap,
    body: Result<Json<RpcRequest>, axum::extract::rejection::JsonRejection>,
) -> Response {
    let req = match body {
        Ok(Json(req)) => req,
        // Each refusal is its own spec row, and they must not share a code: a
        // body that is not `application/json` is `ContentTypeNotSupported`
        // (-32005) in A2A's own mapping table; a body that is not JSON at all
        // is JSON-RPC's ParseError; valid JSON that is not a request object —
        // a batch array, a missing `method` — is InvalidRequest. Collapsing
        // them tells a caller with a wrong header or a wrong shape that its
        // serializer is broken. A body over the size limit is not a JSON-RPC
        // question and keeps its HTTP 413.
        Err(rejection) => {
            use axum::extract::rejection::JsonRejection;
            let error = match &rejection {
                JsonRejection::MissingJsonContentType(_) => RpcError::new(
                    code::CONTENT_TYPE_NOT_SUPPORTED,
                    "the request body must be application/json",
                ),
                JsonRejection::JsonDataError(_) => RpcError::new(
                    code::INVALID_REQUEST,
                    "the request body is JSON but not a JSON-RPC request object",
                ),
                JsonRejection::BytesRejection(_) => return rejection.into_response(),
                _ => RpcError::new(code::PARSE_ERROR, "the request body is not valid JSON"),
            };
            return error.into_response();
        }
    };
    // JSON-RPC's id is a string, a number or null. Echoing an object or an
    // array back would hand the caller an id its own client cannot match.
    if !matches!(req.id, Value::Null | Value::String(_) | Value::Number(_)) {
        return RpcError::new(
            code::INVALID_REQUEST,
            "`id` must be a string, a number or null",
        )
        .into_response();
    }
    let id = req.id.clone();

    if req.jsonrpc != "2.0" {
        return RpcError::new(
            code::INVALID_REQUEST,
            format!("`jsonrpc` must be \"2.0\", not {:?}", req.jsonrpc),
        )
        .with_id(id)
        .into_response();
    }
    if let Err(e) = A2aServer::check_version(&headers) {
        return e.with_id(id).into_response();
    }
    // Before anything about the request is read. Parameters, the tenant field
    // and the method name all answer differently depending on what was sent,
    // and an unauthenticated caller is owed the challenge and nothing it could
    // map the surface with.
    let caller = match server.authenticate(&headers).await {
        Ok(caller) => caller,
        Err(e) => return e.with_id(id).into_response(),
    };

    // Streaming methods are dispatched first because they answer with a
    // different *kind* of response: an SSE body, not a JSON-RPC envelope. Folding
    // them into `dispatch` would mean a function whose return type is "a value
    // or an entire HTTP response", which is how one of the two paths quietly
    // stops setting its content type.
    if matches!(
        req.method.as_str(),
        method::SEND_STREAMING | method::SUBSCRIBE
    ) {
        return match stream_method(server, caller, req).await {
            Ok(sse) => sse.into_response(),
            Err(e) => e.with_id(id).into_response(),
        };
    }

    // **Boxed, because this future is the whole runtime.** `dispatch` fans out
    // to every A2A method and one of them admits and executes a run, so the
    // state machine inlined here carries the executor's own locals — sixteen
    // kilobytes of them. Held on the stack it is sixteen kilobytes *per
    // concurrent request*, paid by every caller including the ones asking for a
    // task's status. One heap allocation per request buys that back, and a
    // request that reaches this line is already doing far more work than an
    // allocation.
    match Box::pin(dispatch(&server, &caller, &req)).await {
        Ok(result) => Json(json!({ "jsonrpc": "2.0", "id": id, "result": result })).into_response(),
        Err(e) => e.with_id(id).into_response(),
    }
}

/// `SendStreamingMessage` and `SubscribeToTask`.
///
/// Both are the same thing once the run exists: a view of the journal from a
/// point onward. The only difference is whether this call is what created it.
#[allow(clippy::too_many_lines)]
async fn stream_method(
    server: A2aServer,
    caller: Caller,
    req: RpcRequest,
) -> Result<
    axum::response::sse::Sse<
        impl futures_util::stream::Stream<
            Item = Result<axum::response::sse::Event, std::convert::Infallible>,
        >,
    >,
    RpcError,
> {
    let params = parse_params(&req.method, &req.params)?;
    server.check_tenant(&params)?;
    if req.method == method::SEND_STREAMING
        && let Some(configuration) = &params.configuration
    {
        configuration.validate()?;
    }

    // A slot before anything else is done for this caller: a stream is a
    // connection held open and a journal polled for as long as the run lives,
    // so how many one peer may hold is a bound this surface states. Taken
    // before admission, so a caller at its ceiling starts no work it then
    // cannot watch; released when the stream ends or the client goes away.
    let slot = server
        .streams
        .claim(&caller.actor)
        .ok_or_else(|| RpcError::new(code::QUOTA_EXHAUSTED, QUOTA_EXHAUSTED_MESSAGE))?;

    let (run, records) = if req.method == method::SUBSCRIBE {
        let id = task_id(&params)?;
        let records = server
            .authorized_task(&caller, action::TASK_READ, id)
            .await?;
        (id, records)
    } else {
        let Some(message) = params.message.clone() else {
            return Err(RpcError::new(
                code::INVALID_PARAMS,
                "`message` is required by SendStreamingMessage",
            ));
        };
        message.validate_parts(&caller)?;
        refuse_push_on_continuation(&message, params.configuration.as_ref())?;
        let run = if message.task_id.is_some() {
            continue_task(&server, &caller, &message).await?
        } else {
            let skill = resolve_skill(&server, &message)?;
            server.authorize(&caller, action::MESSAGE_SEND, &skill, None)?;
            check_context(&server, &message, &caller).await?;
            if let Some(push) = params
                .configuration
                .as_ref()
                .and_then(|configuration| configuration.task_push_notification_config.as_ref())
            {
                validate_inline_push(&server, &caller, &skill, push)?;
            }
            // Admitted before the stream opens. A stream that begins and *then*
            // reports a refusal has already told the client the work started —
            // and an SSE body cannot carry a JSON-RPC error the client is looking
            // for at that point.
            let source = super::peer_source(&caller.actor);
            let input = Tainted::from_source(message.to_input(), SourceId::new(&source));
            let terms = run_terms(&message, &caller)?;
            match spawn_a2a(&server, &skill, input, terms).await {
                Ok(run) => {
                    if let Some(push) = params.configuration.as_ref().and_then(|configuration| {
                        configuration.task_push_notification_config.as_ref()
                    }) {
                        register_push(&server, push, run, 1).await?;
                    }
                    run
                }
                Err(
                    crate::core::RuntimeError::PolicyDenied(_)
                    | crate::core::RuntimeError::Delegation(_),
                ) => {
                    return Err(RpcError::new(
                        code::UNSUPPORTED_OPERATION,
                        "this agent declined the request",
                    ));
                }
                Err(crate::core::RuntimeError::QuotaExceeded(e)) => {
                    return Err(quota_refusal(&e));
                }
                Err(crate::core::RuntimeError::Draining) => {
                    return Err(RpcError::new(code::DRAINING, DRAINING_MESSAGE));
                }
                Err(e @ crate::core::RuntimeError::SubjectUnbound { .. }) => {
                    return Err(subject_unbound(&e));
                }
                Err(e) => return Err(internal("admitting a streamed message", &e)),
            }
        };
        let records = server
            .runtime
            .journal()
            .read(run, 1)
            .await
            .map_err(|e| internal("reading a task's journal", &e))?;
        (run, records)
    };

    // Decided from the records before anything is replayed: a subscription
    // to a finished task is refused, and refusing it after a strict replay to
    // build artifacts nobody will receive is a cost the caller chose for us.
    let Some((state, detail)) = state_from_history(&records) else {
        return Err(task_not_found(run));
    };
    // `closes` and not a second `matches!` with the same four states: this
    // decides whether a subscription is *refused*, and `closes` decides whether
    // a stream *ends*. Two spellings of one rule disagree the day somebody adds
    // a terminal state to one of them, and the result is either a subscription
    // accepted that shuts immediately or one refused that would have streamed.
    if req.method == method::SUBSCRIBE && super::a2a_stream::closes(state) {
        return Err(RpcError::new(
            code::UNSUPPORTED_OPERATION,
            "SubscribeToTask requires a non-terminal task",
        ));
    }
    let case = records
        .iter()
        .find_map(|r| r.body.case.map(|c| c.to_string()));
    let from = records.last().map_or(1, |last| last.body.seq + 1);
    let mut task = task_of(run, state, &detail, case.clone());
    task.artifacts = server
        .artifacts_unmetered(run, state)
        .await
        .map_err(|e| internal("projecting a task's artifacts", &e))?;

    Ok(super::a2a_stream::tail(
        Arc::clone(&server.runtime),
        Arc::clone(&server.artifact_cache),
        slot,
        run,
        case,
        req.id,
        task,
        from,
    ))
}

async fn dispatch(
    server: &A2aServer,
    caller: &Caller,
    req: &RpcRequest,
) -> Result<Value, RpcError> {
    let params = parse_params(&req.method, &req.params)?;
    server.check_tenant(&params)?;

    match req.method.as_str() {
        // Boxed: a send admits and may run a whole run inline, and that future
        // is larger than the others this match holds.
        method::SEND_MESSAGE => Box::pin(send_message(server, caller, params)).await,
        method::GET_TASK => get_task(server, caller, params).await,
        method::CANCEL_TASK => cancel_task(server, caller, params).await,
        method::GET_EXTENDED_CARD => get_extended_card(server, caller),

        // Streaming is handled before dispatch; reaching here means the router
        // changed and this arm did not.
        method::SEND_STREAMING | method::SUBSCRIBE => Err(RpcError::new(
            code::INTERNAL_ERROR,
            "a streaming method reached the non-streaming dispatcher",
        )),
        method::LIST_TASKS => list_tasks(server, caller, &params).await,
        method::CREATE_PUSH => push_create(server, caller, &params).await,
        method::GET_PUSH => push_get(server, caller, &params).await,
        method::LIST_PUSH => push_list(server, caller, &params).await,
        method::DELETE_PUSH => push_delete(server, caller, &params).await,
        other => Err(RpcError::new(
            code::METHOD_NOT_FOUND,
            format!("no such A2A method: {other}"),
        )),
    }
}

/// Which parameter names each method actually reads.
///
/// `tenant` is on every row rather than special-cased: A2A's routing identifier
/// is orthogonal to the method, and leaving it out of one row would refuse a
/// correctly-routed request for that method alone — the kind of hole a table
/// exists to make visible.
const FIELDS_BY_METHOD: &[(&str, &[&str])] = &[
    (
        method::SEND_MESSAGE,
        &["tenant", "message", "configuration", "metadata"],
    ),
    (
        method::SEND_STREAMING,
        &["tenant", "message", "configuration", "metadata"],
    ),
    (method::GET_TASK, &["tenant", "id", "historyLength"]),
    (method::CANCEL_TASK, &["tenant", "id"]),
    (method::SUBSCRIBE, &["tenant", "id"]),
    (method::GET_EXTENDED_CARD, &["tenant"]),
    (
        method::LIST_TASKS,
        &[
            "tenant",
            "contextId",
            "status",
            "pageSize",
            "pageToken",
            "historyLength",
            "statusTimestampAfter",
            "includeArtifacts",
        ],
    ),
    (
        method::CREATE_PUSH,
        &["tenant", "taskId", "id", "url", "token", "authentication"],
    ),
    (method::GET_PUSH, &["tenant", "taskId", "id"]),
    (
        method::LIST_PUSH,
        &["tenant", "taskId", "pageSize", "pageToken"],
    ),
    (method::DELETE_PUSH, &["tenant", "taskId", "id"]),
];

fn parse_params(method: &str, value: &Value) -> Result<CommonParams, RpcError> {
    if value.is_null() {
        return Ok(CommonParams::default());
    }
    let Some(object) = value.as_object() else {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            "A2A method parameters must be a JSON object",
        ));
    };

    // Before deserializing, because the union struct would accept the field and
    // the method would then ignore it. An unknown method falls through to the
    // dispatcher's own `METHOD_NOT_FOUND`, which is a better answer than a
    // parameter complaint about a method that does not exist.
    if let Some((_, allowed)) = FIELDS_BY_METHOD.iter().find(|(m, _)| *m == method)
        && let Some(stray) = object.keys().find(|k| !allowed.contains(&k.as_str()))
    {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            format!(
                "'{stray}' is not a parameter of {method}; it takes {}",
                allowed.join(", ")
            ),
        ));
    }

    serde_json::from_value(value.clone()).map_err(|error| {
        RpcError::new(
            code::INVALID_PARAMS,
            format!("request parameters do not match the A2A method schema: {error}"),
        )
    })
}

/// Answer a message aimed at a task that has already sealed.
///
/// Decided from the task's own journal: the event store's subscription may
/// already be retired, and routing a retransmit through `deliver_to` would
/// answer a legitimate retry with a terminal-state error. The journal records
/// every delivered continuation message verbatim, so the retry of the message
/// that completed the task is recognisable there — and a *new* message aimed
/// at a finished task stays refused.
fn continue_sealed_task(
    run: RunId,
    records: &[crate::journal::Record],
    message: &A2aMessage,
) -> Result<RunId, RpcError> {
    let delivered_before = records.iter().any(|record| {
        matches!(
            record.kind(),
            RecordKind::EffectDone { output, .. }
                if output
                    .get("$a2a_message")
                    .and_then(|m| m.get("messageId"))
                    .and_then(Value::as_str)
                    == Some(message.message_id.as_str())
        )
    });
    if delivered_before {
        return Ok(run);
    }
    Err(RpcError::new(
        code::UNSUPPORTED_OPERATION,
        "this task has completed and is not waiting for input",
    ))
}

/// Refuse a push configuration on a message that continues a task.
///
/// A continuation answers with the task it continued, and registering a
/// webhook on that task is `CreateTaskPushNotificationConfig`'s job, under its
/// own gate. Accepting the field here and doing nothing with it would tell the
/// caller a webhook was registered that never will be.
fn refuse_push_on_continuation(
    message: &A2aMessage,
    configuration: Option<&SendConfiguration>,
) -> Result<(), RpcError> {
    if message.task_id.is_some()
        && configuration.is_some_and(|c| c.task_push_notification_config.is_some())
    {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            "configuration.taskPushNotificationConfig is not accepted on a message that \
             continues a task; register it with CreateTaskPushNotificationConfig",
        ));
    }
    Ok(())
}

/// Continue one interrupted A2A task with a client message.
///
/// The run remains append-only: the message becomes the output of the exact
/// `event.await` effect on which the task stopped, then ordinary resume replay
/// carries execution forward. `EventStore::deliver_to` is task-addressed and
/// atomic, so another run sharing the same business correlation key cannot
/// consume this message.
async fn continue_task(
    server: &A2aServer,
    caller: &Caller,
    message: &A2aMessage,
) -> Result<RunId, RpcError> {
    let raw = message
        .task_id
        .as_deref()
        .ok_or_else(|| RpcError::new(code::INVALID_PARAMS, "`taskId` is required"))?;
    let run = RunId::parse(raw)
        .map_err(|_| RpcError::new(code::TASK_NOT_FOUND, format!("no such task: {raw}")))?;
    let records = server
        .authorized_task(caller, action::TASK_CONTINUE, run)
        .await?;
    let Some(last) = records.last() else {
        return Err(task_not_found(run));
    };
    if matches!(
        last.kind(),
        RecordKind::RunSuspended {
            reason: crate::core::SuspendReason::AwaitingTime { .. }
        }
    ) {
        return Err(RpcError::new(
            code::UNSUPPORTED_OPERATION,
            "this task is sleeping until a timer fires and cannot accept input",
        ));
    }
    if !matches!(
        last.kind(),
        RecordKind::RunSuspended { .. } | RecordKind::RunConcluded { .. }
    ) {
        return Err(RpcError::new(
            code::UNSUPPORTED_OPERATION,
            "this task is not waiting for input",
        ));
    }
    if let RecordKind::RunConcluded { outcome, .. } = last.kind() {
        if sealed_state(outcome) == TaskState::AuthRequired {
            return Err(RpcError::new(
                code::UNSUPPORTED_OPERATION,
                "this task is paused because the authority it acts under was withdrawn; \
                 it continues when this deployment restores it, not on a message",
            ));
        }
        return continue_sealed_task(run, &records, message);
    }
    // Search history rather than only the last record so a transport retry of
    // the message that completed a task can still be recognized as the same
    // `(source, messageId)` and return the current task instead of a spurious
    // terminal-task error. A different message id finds no live subscription
    // and remains refused.
    let (kind, correlation) = records
        .iter()
        .rev()
        .find_map(|record| match record.kind() {
            RecordKind::RunSuspended {
                reason:
                    crate::core::SuspendReason::AwaitingEvent {
                        kind, correlation, ..
                    },
            } => Some((kind.clone(), correlation.clone())),
            _ => None,
        })
        .ok_or_else(|| {
            RpcError::new(
                code::UNSUPPORTED_OPERATION,
                "this task has no input wait to continue",
            )
        })?;
    let context = records
        .iter()
        .find_map(|record| record.body.case.map(|case| case.to_string()));
    if let Some(sent) = message.context_id.as_deref()
        && context.as_deref() != Some(sent)
    {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            "message.contextId does not match the referenced task",
        ));
    }
    // The task gate says this peer may continue its own task; this one says it
    // may supply the kind the task awaits.
    server.authorize(caller, action::EVENT_DELIVER, &kind, Some(&caller.actor))?;

    let event = crate::core::InboundEvent {
        // `peer:{actor}`, the one spelling a counterparty's provenance has on
        // every transport — see [`super::peer_source`]. This source becomes the
        // delivered value's provenance, so a transport-qualified variant here
        // would give one counterparty two names.
        source: super::peer_source(&caller.actor),
        id: message.message_id.clone(),
        kind,
        correlation,
        payload: message.to_input(),
        // A counterparty's message, so nobody on this plane minted it. The
        // sender is provenance, not an operator, and the one thing this field
        // must not become is somewhere a peer can claim to be a person here.
        by: None,
    };
    match server.runtime.deliver_to(run, &event).await {
        // `Buffered` is a success too: when the owner's lease is stalled the
        // input is held durably and the sweep resumes the run — an internal
        // error here would tell the caller a continuation failed that
        // actually succeeded-pending.
        Ok(
            crate::core::Delivery::Resumed { .. }
            | crate::core::Delivery::Duplicate
            | crate::core::Delivery::Buffered,
        ) => Ok(run),
        Err(crate::core::RuntimeError::PlanContract(_)) => Err(RpcError::new(
            code::UNSUPPORTED_OPERATION,
            "this task is no longer waiting for input",
        )),
        // A run waiting on a human task waits for the worklist's answer, and a
        // message addressed to it by task id is not one.
        Err(crate::core::RuntimeError::ReservedEventKind { .. }) => Err(RpcError::new(
            code::UNSUPPORTED_OPERATION,
            "this task is waiting for a decision on this plane's worklist, not for input",
        )),
        Err(error) => Err(internal("delivering a continuation", &error)),
    }
}

#[allow(clippy::too_many_lines)]
async fn send_message(
    server: &A2aServer,
    caller: &Caller,
    params: CommonParams,
) -> Result<Value, RpcError> {
    if let Some(configuration) = &params.configuration {
        configuration.validate()?;
    }
    let inline_push = params
        .configuration
        .as_ref()
        .and_then(|configuration| configuration.task_push_notification_config.clone());
    let Some(message) = params.message else {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            "`message` is required by SendMessage",
        ));
    };
    message.validate_parts(caller)?;
    refuse_push_on_continuation(&message, params.configuration.as_ref())?;
    if message.task_id.is_some() {
        let history_length = params
            .configuration
            .as_ref()
            .and_then(|configuration| configuration.history_length);
        let run = continue_task(server, caller, &message).await?;
        return get_task(
            server,
            caller,
            CommonParams {
                id: Some(run.to_string()),
                history_length,
                ..CommonParams::default()
            },
        )
        .await
        .map(|task| json!({ "task": task }));
    }
    let skill = resolve_skill(server, &message)?;
    server.authorize(caller, action::MESSAGE_SEND, &skill, None)?;
    check_context(server, &message, caller).await?;
    if let Some(push) = &inline_push {
        validate_inline_push(server, caller, &skill, push)?;
    }

    // Untrusted, and provenanced to the peer that sent it. A protected sink
    // field can then name the one counterparty it will take an amount from, and
    // a skill that wants to act on this has to pass a gate to do it.
    let source = super::peer_source(&caller.actor);
    let input = Tainted::from_source(message.to_input(), SourceId::new(&source));
    let terms = run_terms(&message, caller)?;

    // Non-blocking: the spec requires returning as soon as the task exists,
    // with an in-progress state, leaving the caller to poll `GetTask`. Admission
    // still happens synchronously, so a refusal is still an immediate answer —
    // returning a task id for a run the gate rejected would hand the caller a
    // handle to nothing and turn a decline into a task that never appears.
    if params
        .configuration
        .as_ref()
        .is_some_and(|c| c.return_immediately)
    {
        return match spawn_a2a(server, &skill, input, terms).await {
            Ok(run) => {
                if let Some(push) = &inline_push {
                    register_push(server, push, run, 1).await?;
                }
                let case = task_context(server, run).await?;
                Ok(json!({
                    "task": task_of(run, TaskState::Working, "accepted", case)
                }))
            }
            Err(
                crate::core::RuntimeError::PolicyDenied(_)
                | crate::core::RuntimeError::Delegation(_),
            ) => Ok(json!({ "message": declined(&skill) })),
            Err(crate::core::RuntimeError::QuotaExceeded(e)) => Err(quota_refusal(&e)),
            Err(crate::core::RuntimeError::Draining) => {
                Err(RpcError::new(code::DRAINING, DRAINING_MESSAGE))
            }
            Err(e @ crate::core::RuntimeError::SubjectUnbound { .. }) => Err(subject_unbound(&e)),
            Err(e) => Err(internal("admitting a message", &e)),
        };
    }

    let admission = match run_a2a(server, &skill, input, terms).await {
        Ok(admission) => admission,
        // A policy denial is the agent *declining*, not the agent breaking.
        // Reported as `-32603 Internal error` it reads as "this server is
        // faulty, retry later", and the caller retries a decision that will
        // never change.
        //
        // Answered as a `Message` rather than a rejected `Task` because no task
        // exists: nothing was admitted, so there is no id to poll and no
        // history to fetch. A2A's response is a oneof for exactly this.
        Err(
            crate::core::RuntimeError::PolicyDenied(_) | crate::core::RuntimeError::Delegation(_),
        ) => {
            return Ok(json!({ "message": declined(&skill) }));
        }
        // Back-pressure, not a fault. `-32603` reads as "the far side is
        // broken, retry later" and a caller may well retry the same second;
        // this says *the agent cannot take this on right now*, which is what a
        // ceiling means and what a caller should back off from. The code and
        // message are fixed — see `code::QUOTA_EXHAUSTED` for why neither is
        // the spec's `-32004` and why the quota arithmetic stays out of it.
        Err(crate::core::RuntimeError::QuotaExceeded(e)) => {
            return Err(quota_refusal(&e));
        }
        // Not a fault and not back-pressure from the agent: this process is
        // going away and admitted nothing. A caller retrying now reaches a
        // different instance and is served.
        Err(crate::core::RuntimeError::Draining) => {
            return Err(RpcError::new(code::DRAINING, DRAINING_MESSAGE));
        }
        // The message, not the server: it lacks what the agent's declaration
        // reads its data subject from, and the same message is refused again.
        Err(e @ crate::core::RuntimeError::SubjectUnbound { .. }) => {
            return Err(subject_unbound(&e));
        }
        Err(e) => return Err(internal("admitting a message", &e)),
    };
    let outcome = match admission {
        crate::runtime::Admission::Fresh(outcome)
        | crate::runtime::Admission::Replayed(outcome) => outcome,
        // Another instance is executing this message right now. The honest
        // answer to a retry is *accepted, already in progress* — a Working
        // task naming the run — never a retry-provoking failure.
        crate::runtime::Admission::InFlight(run) => {
            let case = task_context(server, run).await?;
            return Ok(json!({
                "task": task_of(run, TaskState::Working, "accepted", case)
            }));
        }
    };

    let case = task_context(server, outcome.run_id).await?;
    // A declared `Message` reply answers the blocking send directly — A2A's
    // response is a oneof for exactly this, and it is the only path with a
    // caller still waiting to hand a message to. Returned before push
    // registration, because a response with no task has no task to push about;
    // the run and its journal exist either way.
    if matches!(outcome.status, crate::runtime::RunStatus::Succeeded)
        && let Some(reply) = outcome.output.as_ref().and_then(A2aReply::of_output)
        && let Some(parts) = reply.message_parts()
    {
        return Ok(json!({
            "message": A2aMessage {
                message_id: format!("reply-{}", outcome.run_id),
                role: "ROLE_AGENT".to_owned(),
                parts,
                context_id: case,
                task_id: None,
                metadata: None,
                extensions: Vec::new(),
                reference_task_ids: Vec::new(),
            }
        }));
    }
    if let Some(push) = &inline_push {
        register_push(server, push, outcome.run_id, 1).await?;
    }
    let mut task = task_of_outcome(&outcome);
    task.context_id = case;
    // Unset means the full (capped) history, per the protocol's default —
    // only an explicit `0` skips the read.
    let history_length = params
        .configuration
        .as_ref()
        .and_then(|configuration| configuration.history_length);
    if history_length != Some(0) {
        let records = server
            .runtime
            .journal()
            .read(outcome.run_id, 1)
            .await
            .map_err(|_| {
                RpcError::new(code::INTERNAL_ERROR, "the task journal could not be read")
            })?;
        task.history = task_history(
            outcome.run_id,
            &records,
            history_length,
            task.context_id.as_deref(),
        );
    }
    Ok(json!({ "task": task }))
}

async fn run_a2a(
    server: &A2aServer,
    skill: &str,
    input: Tainted<Value>,
    (key, terms): (String, crate::runtime::RunTerms),
) -> Result<crate::runtime::Admission, crate::core::RuntimeError> {
    server
        .runtime
        .start(skill, input)
        .terms(terms)
        .once(key)
        .run()
        .await
}

async fn spawn_a2a(
    server: &A2aServer,
    skill: &str,
    input: Tainted<Value>,
    (key, terms): (String, crate::runtime::RunTerms),
) -> Result<RunId, crate::core::RuntimeError> {
    // Producer-scoped and deduplicating, for the reasons `run_terms` states.
    // A duplicate answers with the run the key already admitted, which is
    // exactly what a retrying caller needs back.
    Ok(server
        .runtime
        .start(skill, input)
        .terms(terms)
        .once(key)
        .spawn()
        .await?
        .run)
}

/// The key and terms every A2A admission runs under: keyed by the producer-scoped
/// message id, inside the case the caller named or a fresh one, and **acting
/// as the caller** — under the chain its credential carried, or under none.
///
/// The key is also what makes the run this caller's: its source half is the
/// authenticated sender, so every later read or write of the task can ask
/// whose it is without a second record — see [`task_owner`].
///
/// The key carries its producer, in
/// [`origin_key`](crate::core::origin_key)'s spelling: a bare `messageId` would
/// let two counterparties swallow each other's messages as apparent retries —
/// or *join* each other's cases, since correlation matches any open case in the
/// tenant. And it is an **admission** key, not only a correlation key: this
/// crate's own client keeps `messageId` stable across retries precisely so a
/// peer can deduplicate, and a server that correlates without deduplicating
/// starts a second run inside the right case. A different key space from an
/// event's, and the same construction, because the forgery it refuses is the
/// same one.
///
/// The chain is the caller's, never the plane's: a plane that admitted every
/// peer's run under its own chain would answer "on whose behalf" with the same
/// name for all of them, and act for a caller whose credential permits less. A
/// caller that presented no chain acts under **none**: the plane's chain
/// bounds what it may be admitted for, and lends it none of its authority —
/// the fallback would make the operator's authority ambient.
fn run_terms(
    message: &A2aMessage,
    caller: &Caller,
) -> Result<(String, crate::runtime::RunTerms), RpcError> {
    let keyed = crate::core::origin_key(&admission_source(&caller.actor), &message.message_id);
    // The authenticated peer, never a body field: the four-eyes exclusion
    // reads it, and a caller who could name it could name somebody else.
    let terms =
        crate::runtime::RunTerms::served(caller.acting_as.clone()).admitted_by(&caller.actor);
    let terms = match message.context_id.as_deref() {
        Some(context) => {
            // `check_context` has already refused this, in the same words; the
            // parse is repeated because the terms need the typed id.
            let case = crate::core::CaseId::parse(context).map_err(|_| {
                RpcError::new(code::TASK_NOT_FOUND, format!("no such context: {context}"))
            })?;
            terms.in_case(case)
        }
        // Always correlated: `A2aServer::new` refuses a runtime without a case
        // layer, so every task gets a real, continuable context — the
        // contextId returned is one a client can send back, not a string that
        // satisfies a schema and continues nothing.
        None => terms.correlated(
            "a2a.context",
            &[crate::core::CorrelationKey::new(
                CONTEXT_KEY_NAMESPACE,
                &keyed,
            )],
        ),
    };
    Ok((keyed, terms))
}

/// The `source` half of every A2A admission key: this surface, then the
/// authenticated sender.
///
/// Its own namespace rather than the bare [`peer_source`](super::peer_source)
/// an inbound event's key carries, because ownership is read back from it: an
/// embedder that keys a run with an event's `dedup_key` — the key this crate
/// recommends — must not thereby hand that run to the event's sender as an A2A
/// task.
fn admission_source(actor: &str) -> String {
    format!("{ADMISSION_NAMESPACE}{}", super::peer_source(actor))
}

const ADMISSION_NAMESPACE: &str = "a2a/";

/// The peer that admitted a task over this surface, read from the run's
/// admission key. `None` for a run nobody admitted here.
fn task_owner(records: &[crate::journal::Record]) -> Option<&str> {
    records
        .first()?
        .admission_source()?
        .strip_prefix(ADMISSION_NAMESPACE)?
        .strip_prefix("peer:")
}

/// The one answer for a task this caller may not address, whether it exists
/// or not.
fn task_not_found(run: RunId) -> RpcError {
    RpcError::new(code::TASK_NOT_FOUND, format!("no such task: {run}"))
}

/// A fault on this plane's side, told to the peer in one fixed sentence —
/// see [`withheld_fault`](crate::core::withheld_fault).
fn internal(doing: &str, error: &dyn std::fmt::Display) -> RpcError {
    RpcError::new(
        code::INTERNAL_ERROR,
        crate::core::withheld_fault("a2a", doing, error),
    )
}

/// Refuse a `contextId` this caller did not open.
///
/// A context is a case, and joining one puts this caller's run beside another
/// peer's work and its case state in reach. So the case must be one a message
/// from this caller opened — its correlation key carries the admitting
/// sender — and anything else answers as a context that does not exist.
async fn check_context(
    server: &A2aServer,
    message: &A2aMessage,
    caller: &Caller,
) -> Result<(), RpcError> {
    let Some(context) = message.context_id.as_deref() else {
        return Ok(());
    };
    let not_found = || RpcError::new(code::TASK_NOT_FOUND, format!("no such context: {context}"));
    let case = crate::core::CaseId::parse(context).map_err(|_| not_found())?;
    let Some(cases) = server.runtime.cases() else {
        return Err(not_found());
    };
    let Some(case) = cases
        .case(case)
        .await
        .map_err(|e| internal("reading a context's case", &e))?
    else {
        return Err(not_found());
    };
    let ours = admission_source(&caller.actor);
    let opened_by_caller = case.correlation.iter().any(|key| {
        key.namespace == CONTEXT_KEY_NAMESPACE
            && crate::core::origin_source(&key.value) == Some(ours.as_str())
    });
    if !opened_by_caller {
        return Err(not_found());
    }
    // Refused here rather than by admission: the runtime's refusal of a closed
    // case is a contract error shared with every other misuse, and reading
    // this one out of it would answer them all as "no such context".
    if case.status.is_closed() {
        return Err(RpcError::new(
            code::TASK_NOT_FOUND,
            format!("context {context} is closed and accepts no new task"),
        ));
    }
    Ok(())
}

/// The correlation namespace a fresh A2A context is opened under.
const CONTEXT_KEY_NAMESPACE: &str = "a2a-message";

async fn task_context(server: &A2aServer, run: RunId) -> Result<Option<String>, RpcError> {
    server
        .runtime
        .journal()
        .read(run, 1)
        .await
        .map_err(|_| RpcError::new(code::INTERNAL_ERROR, "the task journal could not be read"))
        .map(|records| {
            records
                .iter()
                .find_map(|record| record.body.case.map(|case| case.to_string()))
        })
}

pub(super) fn task_of_outcome(outcome: &crate::runtime::RunOutcome) -> A2aTask {
    // A declared reply carries its own parts; otherwise the default projection
    // stands — a string is a text part, anything else a data part.
    let artifacts_parts: Vec<Vec<Part>> =
        match outcome.output.as_ref().and_then(A2aReply::of_output) {
            Some(reply) => reply.artifact_parts(),
            None => vec![vec![match outcome
                .output
                .as_ref()
                .map_or(Value::Null, |o| o.peek().clone())
            {
                Value::String(text) => Part::text(text),
                data => Part::data(data),
            }]],
        };
    A2aTask {
        id: outcome.run_id.to_string(),
        context_id: None,
        status: TaskStatus {
            state: state_of(&outcome.status),
            message: None,
            timestamp: None,
        },
        artifacts: matches!(outcome.status, crate::runtime::RunStatus::Succeeded).then(|| {
            artifacts_parts
                .into_iter()
                .enumerate()
                .map(|(i, parts)| A2aArtifact {
                    artifact_id: format!("{}-result-{i}", outcome.run_id),
                    name: None,
                    description: None,
                    parts,
                    metadata: None,
                    extensions: Vec::new(),
                })
                .collect()
        }),
        history: None,
        metadata: None,
    }
}

/// Which capability this message asks for.
///
/// Named or unambiguous — never inferred. See the module docs: picking a
/// capability from the content of an untrusted message is a dispatch decision
/// made by reading attacker-controlled text.
fn resolve_skill(server: &A2aServer, message: &A2aMessage) -> Result<String, RpcError> {
    if let Some(asked) = message.requested_skill() {
        if server.skills.iter().any(|s| s == asked) {
            return Ok(asked.to_owned());
        }
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            format!(
                "this agent has no skill '{asked}'. Its card advertises: {}",
                server.skills.join(", ")
            ),
        ));
    }
    match server.skills.as_slice() {
        [only] => Ok(only.clone()),
        [] => Err(RpcError::new(
            code::UNSUPPORTED_OPERATION,
            "this agent advertises no skills, so there is nothing to send a \
             message to",
        )),
        many => Err(RpcError::new(
            code::INVALID_PARAMS,
            format!(
                "this agent advertises {} skills, so `message.metadata.skill` \
                 must name one of: {}. It is not inferred from the message — \
                 choosing what to run by reading the text would let the sender \
                 pick the capability",
                many.len(),
                many.join(", ")
            ),
        )),
    }
}

async fn get_task(
    server: &A2aServer,
    caller: &Caller,
    params: CommonParams,
) -> Result<Value, RpcError> {
    let (id, records) = server.gate_task(caller, action::TASK_READ, &params).await?;
    load_task(server, id, &records, params.history_length).await
}

async fn load_task(
    server: &A2aServer,
    id: RunId,
    records: &[crate::journal::Record],
    history_length: Option<usize>,
) -> Result<Value, RpcError> {
    // No records is no such task: a run this plane never admitted and a task id
    // a caller invented are the same fact from here.
    let (state, detail) = state_from_history(records).ok_or_else(|| task_not_found(id))?;
    let case = records
        .iter()
        .find_map(|r| r.body.case.map(|c| c.to_string()));

    let mut task = task_of(id, state, &detail, case.clone());
    task.history = task_history(id, records, history_length, case.as_deref());
    // Unmetered: one task, one replay, and the sealed-run cache spares the
    // poll loop that reads the same finished task every few seconds.
    task.artifacts = server
        .artifacts_unmetered(id, state)
        .await
        .map_err(|error| internal("projecting a task's artifacts", &error))?;
    serde_json::to_value(task).map_err(|error| internal("encoding a task", &error))
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct TaskCursor {
    updated_at: u64,
    run: String,
    context_id: Option<String>,
    status: Option<TaskState>,
    status_timestamp_after: Option<String>,
    /// How many matching tasks the earlier pages delivered.
    ///
    /// A continuation scans only from the cursor on, so without this its
    /// `totalSize` counts what is left rather than the listing's total, and the
    /// number shrinks page by page. Carried rather than recounted because a
    /// recount re-reads every task before the cursor — the cost the scan budget
    /// exists to bound. The token is the caller's, so a value it edits misstates
    /// only its own total.
    before: u64,
}

/// How many index rows `list_tasks` pulls per store round trip.
///
/// Bounded so no path holds the tenant's whole index in memory — which is what
/// the unbounded read this replaced did on *every* call, before going on to
/// read the complete journal of every run it returned.
const TASK_SCAN: usize = 256;

/// Default for [`A2aServer::filter_scan_budget`]: the most candidates one
/// `ListTasks` may examine before being refused as too broad.
///
/// The number is a cost ceiling, not a result limit — results are bounded by
/// `pageSize` already. At the default, the worst request a peer can make costs
/// on the order of a thousand journal reads, once, and is told how to narrow;
/// without it the cost was every run the tenant ever wrote, per request,
/// forever.
const FILTER_SCAN_BUDGET: usize = 1024;

/// Default for [`A2aServer::artifact_replay_budget`]: the most strict replays
/// one `ListTasks` with `includeArtifacts` may perform.
///
/// A cost ceiling in the same spirit as [`FILTER_SCAN_BUDGET`], for the other
/// caller-priced read: each completed task's artifacts are a full strict
/// replay, a page holds up to a hundred tasks, and the sealed-run cache means
/// only first sight of a run pays it. Sixteen replays is a bounded worst case
/// per request; a page that wanted more marks the remainder omitted, and a
/// second request finds the first sixteen cached.
const ARTIFACT_REPLAY_BUDGET: usize = 16;

/// The task-metadata key marking artifacts withheld by the replay budget.
///
/// A bounded result must not be shaped like a complete one — a silent
/// truncation reads as an answer. A
/// task past the budget is returned without artifacts — indistinguishable from
/// several honest states — so the omission says its own name, in `Task.metadata`
/// because that is the field A2A gives a server for exactly this kind of
/// annotation; a new top-level response member would be a schema no client
/// expects. `GetTask` on the marked id recovers the artifacts, and warms the
/// cache doing it.
pub const ARTIFACTS_OMITTED_KEY: &str = "io.agentplane.a2a/artifactsOmitted";

#[allow(clippy::too_many_lines)]
async fn list_tasks(
    server: &A2aServer,
    caller: &Caller,
    params: &CommonParams,
) -> Result<Value, RpcError> {
    server.authorize(caller, action::TASK_READ, "tasks", None)?;
    let page_size = params.page_size.unwrap_or(50);
    if !(1..=100).contains(&page_size) {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            "pageSize must be between 1 and 100",
        ));
    }
    let after = params
        .status_timestamp_after
        .as_deref()
        .map(|value| {
            time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339)
                .map_err(|_| {
                    RpcError::new(
                        code::INVALID_PARAMS,
                        "statusTimestampAfter must be an RFC 3339 timestamp",
                    )
                })
        })
        .transpose()?;
    let cursor = params
        .page_token
        .as_deref()
        .map(decode_task_cursor)
        .transpose()?;
    if let Some(cursor) = &cursor
        && (cursor.context_id != params.context_id
            || cursor.status != params.status
            || cursor.status_timestamp_after != params.status_timestamp_after)
    {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            "pageToken was issued for different ListTasks filters",
        ));
    }

    // Whether the caller asked anything that can only be answered by reading a
    // run's whole journal. `statusTimestampAfter` is not one of those — it
    // compares against the activity index's own timestamp.
    //
    // The candidates are the caller's own runs and nobody else's: the index
    // is narrowed to the admission source this surface keys the caller's runs
    // with, so another peer's runs and the embedder's cost this listing
    // nothing. An unfiltered listing reads the whole journal only of the
    // tasks on the page; a content-filtered one reads the whole journal of
    // every task it must examine.
    let content_filtered = params.context_id.is_some() || params.status.is_some();
    let source = admission_source(&caller.actor);

    let mut cursor_pos = cursor
        .as_ref()
        .and_then(|c| RunId::parse(&c.run).ok().map(|run| (c.updated_at, run)));
    let mut page: Vec<(RunId, u64, A2aTask)> = Vec::new();
    let mut matched: u64 = 0;
    let mut has_more = false;
    let mut examined: usize = 0;

    'scan: loop {
        let batch = server
            .runtime
            .journal()
            .recent_runs_from(&source, cursor_pos, TASK_SCAN)
            .await
            .map_err(|e| internal("reading the task index", &e))?;
        if batch.is_empty() {
            break;
        }
        cursor_pos = batch.last().map(|(run, updated)| (*updated, *run));

        for (run, updated) in batch {
            if after.is_some_and(|cutoff| {
                i64::try_from(updated)
                    .ok()
                    .and_then(|seconds| time::OffsetDateTime::from_unix_timestamp(seconds).ok())
                    .is_none_or(|value| value < cutoff)
            }) {
                // The index is newest first, so every row after this one is
                // older still: the cutoff ends the scan, not just this row.
                break 'scan;
            }

            // Whose task it is was settled by the index: every run in this
            // range was admitted under the caller's source.
            let owner = Some(caller.actor.as_str());
            // The ceiling on what one listing may cost, refused rather than
            // truncated — see `A2aServer::filter_scan_budget`.
            examined += 1;
            if examined > server.filter_scan_budget {
                return Err(RpcError::new(
                    code::INVALID_PARAMS,
                    format!(
                        "this listing would require examining more than {} tasks to answer \
                         exactly — narrow it with statusTimestampAfter and page from there",
                        server.filter_scan_budget
                    ),
                ));
            }
            // Load-bearing for more than cost: a run the caller may not read
            // must not reach the total either, or `totalSize` discloses the
            // existence of tasks it cannot see.
            if !server.permits(caller, action::TASK_READ, &run.to_string(), owner) {
                continue;
            }

            let wanted = page.len() < page_size;
            if !content_filtered && !wanted {
                matched += 1;
                has_more = true;
                continue;
            }
            let records = server
                .runtime
                .journal()
                .read(run, 1)
                .await
                .map_err(|e| internal("reading a task's journal", &e))?;
            let task = task_from_records(run, &records, updated, params.history_length);
            if params
                .context_id
                .as_ref()
                .is_some_and(|context| task.context_id.as_ref() != Some(context))
                || params
                    .status
                    .as_ref()
                    .is_some_and(|status| &task.status.state != status)
            {
                continue;
            }
            matched += 1;
            if wanted {
                page.push((run, updated, task));
            } else {
                has_more = true;
            }
        }
    }

    // Exact, and counted over the same rows the caller was allowed to see. The
    // easy mistake is reporting the page's length, which tells every caller the
    // total is whatever fits on a screen; the dangerous one is counting the
    // store's index directly, which is cheaper and reveals the tasks policy
    // just hid. Saturated at the wire type's ceiling: the proto field is an
    // `int32`, and a count past it must not overflow a conformant client's
    // deserializer.
    let before = cursor.as_ref().map_or(0, |cursor| cursor.before);
    let total_size = before
        .saturating_add(matched)
        .min(u64::try_from(i32::MAX).unwrap_or(u64::MAX));

    let visible = &page[..];
    let next_page_token = if has_more {
        let (run, updated, _) = visible.last().expect("a page with more has a last item");
        encode_task_cursor(&TaskCursor {
            updated_at: *updated,
            run: run.to_string(),
            context_id: params.context_id.clone(),
            status: params.status,
            status_timestamp_after: params.status_timestamp_after.clone(),
            before: before.saturating_add(u64::try_from(visible.len()).unwrap_or(u64::MAX)),
        })?
    } else {
        String::new()
    };
    let mut tasks = Vec::with_capacity(visible.len());
    // The replay budget for this whole request. Cache hits are free; each
    // cache miss is a strict replay and spends one. Past it a task's artifacts
    // are omitted **and marked** — see [`ARTIFACTS_OMITTED_KEY`] for why the
    // omission must say its own name.
    let mut replays_left = server.artifact_replay_budget;
    for (run, _, task) in visible {
        let mut task = task.clone();
        if params.include_artifacts {
            match server
                .artifacts_bounded(*run, task.status.state, Some(&mut replays_left))
                .await
                .map_err(|error| internal("projecting a task's artifacts", &error))?
            {
                ArtifactRead::Artifacts(artifacts) => task.artifacts = artifacts,
                ArtifactRead::OverBudget => {
                    task.metadata = Some(json!({ ARTIFACTS_OMITTED_KEY: true }));
                }
            }
        }
        tasks.push(task);
    }
    Ok(json!({
        "tasks": tasks,
        "nextPageToken": next_page_token,
        "pageSize": page_size,
        "totalSize": total_size,
    }))
}

fn encode_task_cursor(cursor: &TaskCursor) -> Result<String, RpcError> {
    crate::core::canon::to_bytes(cursor)
        .map(crate::core::b64::encode_url)
        .map_err(|_| RpcError::new(code::INTERNAL_ERROR, "the task cursor could not be encoded"))
}

fn decode_task_cursor(token: &str) -> Result<TaskCursor, RpcError> {
    crate::core::b64::decode_url(token)
        .and_then(|bytes| serde_json::from_slice(&bytes).ok())
        .ok_or_else(|| RpcError::new(code::INVALID_PARAMS, "pageToken is not a valid task cursor"))
}

fn task_from_records(
    run: RunId,
    records: &[crate::journal::Record],
    updated: u64,
    history_length: Option<usize>,
) -> A2aTask {
    // A run with no records is one whose first append has not landed. It is
    // reported as working rather than refused, because this builds a row in a
    // listing where a single unreadable run must not fail the page.
    let (state, detail) =
        state_from_history(records).unwrap_or_else(|| (TaskState::Working, "running".to_owned()));
    let case = records
        .iter()
        .find_map(|record| record.body.case.map(|case| case.to_string()));
    let timestamp = i64::try_from(updated)
        .ok()
        .and_then(|seconds| time::OffsetDateTime::from_unix_timestamp(seconds).ok())
        .and_then(|value| {
            value
                .format(&time::format_description::well_known::Rfc3339)
                .ok()
        });
    let history = task_history(run, records, history_length, case.as_deref());
    let mut task = task_of(run, state, &detail, case);
    task.status.timestamp = timestamp;
    task.history = history;
    task
}

/// The most messages an unbounded history request returns.
///
/// The protocol reads an **unset** `historyLength` as "the full history", so
/// the bound is the server's, stated here rather than left to whatever the
/// journal happens to hold — the same shape as the artifact replay budget.
const HISTORY_CAP: usize = 128;

fn task_history(
    run: RunId,
    records: &[crate::journal::Record],
    history_length: Option<usize>,
    case: Option<&str>,
) -> Option<Vec<A2aMessage>> {
    // Unset means full (capped); `0` is the explicit request for none —
    // the protocol's own default, which a conformant client expecting its
    // conversation back depends on.
    let limit = history_length.unwrap_or(HISTORY_CAP);
    {
        if limit == 0 {
            return None;
        }
        let mut history = Vec::new();
        for record in records {
            let input = match record.kind() {
                RecordKind::RunAdmitted { input, .. } => Some(input),
                RecordKind::EffectDone { output, .. } if output.get("$a2a_message").is_some() => {
                    Some(output)
                }
                _ => None,
            };
            let Some(input) = input else { continue };
            if let Some(message) = input.get("$a2a_message")
                && let Ok(mut message) = serde_json::from_value::<A2aMessage>(message.clone())
            {
                message.task_id = Some(run.to_string());
                if message.context_id.is_none() {
                    message.context_id = case.map(ToOwned::to_owned);
                }
                history.push(message);
                continue;
            }
            // Non-A2A admission retained for old/directly-created tasks.
            if history.is_empty() {
                let text = input
                    .get("text")
                    .and_then(Value::as_str)
                    .map(ToOwned::to_owned);
                let media_type = if text.is_some() {
                    "text/plain"
                } else {
                    "application/json"
                };
                history.push(A2aMessage {
                    message_id: format!("{run}-input"),
                    role: "ROLE_USER".to_owned(),
                    parts: vec![Part {
                        data: text.is_none().then(|| input.clone()),
                        text,
                        raw: None,
                        url: None,
                        filename: None,
                        media_type: Some(media_type.to_owned()),
                        metadata: None,
                    }],
                    context_id: case.map(ToOwned::to_owned),
                    task_id: Some(run.to_string()),
                    metadata: None,
                    extensions: Vec::new(),
                    reference_task_ids: Vec::new(),
                });
            }
        }
        let keep_from = history.len().saturating_sub(limit);
        (!history.is_empty()).then(|| history.split_off(keep_from))
    }
}

/// The uncached artifact projection: one strict replay of a completed run.
///
/// Server request paths go through [`A2aServer::artifacts_bounded`], which
/// caches sealed runs and meters `ListTasks`, and a stream through
/// [`cached_artifacts`]; this stays the raw read for the push projection.
pub(super) async fn task_artifacts(
    runtime: &Runtime,
    run: RunId,
    state: TaskState,
) -> Result<Option<Vec<A2aArtifact>>, crate::core::RuntimeError> {
    if state != TaskState::Completed {
        return Ok(None);
    }
    runtime
        .replay(run, crate::runtime::Mode::Strict)
        .await
        .map(|outcome| task_of_outcome(&outcome).artifacts)
}

/// A task's artifacts through the sealed-run cache, unmetered.
///
/// What a stream reads when the run it watches concludes: every subscriber to
/// one task would otherwise buy its own strict replay of the same immutable
/// history.
pub(super) async fn cached_artifacts(
    runtime: &Runtime,
    cache: &std::sync::Mutex<ArtifactCache>,
    run: RunId,
    state: TaskState,
) -> Result<Option<Vec<A2aArtifact>>, crate::core::RuntimeError> {
    if state != TaskState::Completed {
        return Ok(None);
    }
    if let Some(ArtifactRead::Artifacts(artifacts)) = crate::core::poison::recover(cache).get(run) {
        return Ok(artifacts);
    }
    let artifacts = task_artifacts(runtime, run, state).await?;
    crate::core::poison::recover(cache).insert(run, artifacts.clone());
    Ok(artifacts)
}

/// A sealed run's outcome word, as an A2A state.
///
/// The outcome is the same string [`RunStatus::as_str`](crate::runtime::RunStatus::as_str)
/// produces, which is what the executor seals with — so this is [`state_of`]
/// reached through a string, and the two are held to agreement by
/// `a_live_status_and_its_sealed_outcome_agree`.
///
/// The `_` arm is a *wire* decision rather than a modelling shortcut: an A2A
/// client must be told some state, and a word this build cannot interpret is not
/// something it may describe as completed or waiting. It is deliberately **not**
/// how the runtime itself treats an unrecognised outcome — `resume_is_closed`
/// quarantines on one, because refusing to guess is available there and is not
/// available here. The agreement test is what keeps this arm from quietly
/// swallowing a variant somebody added and forgot to map.
pub(super) fn sealed_state(outcome: &str) -> TaskState {
    match outcome {
        "succeeded" => TaskState::Completed,
        "cancelled" => TaskState::Canceled,
        "suspended" => TaskState::InputRequired,
        // `RunStatus::Withheld`'s outcome; `a_live_status_and_its_sealed_outcome_agree`
        // pins the spelling against the enum.
        "withheld" => TaskState::AuthRequired,
        // Not tasks at all: nobody submitted this plane's record of a sweep, of
        // an operator crossing a tenant boundary or lifting a control, or of a session it merely
        // watched somebody else's agent run. `state_of` decides that and this
        // agrees, which is the direction the test beside it enforces.
        "swept"
        | "broke-glass"
        | crate::runtime::HALT_LIFTED_OUTCOME
        | crate::runtime::HOLD_RELEASED_OUTCOME
        | crate::runtime::OBSERVED_OUTCOME => TaskState::Rejected,
        _ => TaskState::Failed,
    }
}

/// The A2A state of a run, read from the last record of its history.
///
/// Every surface that reports a task's state answers here: `tasks/get`,
/// `tasks/list`, and the event stream a client subscribes to. They are three
/// views of one fact, and three copies of this match would agree until somebody
/// added a record kind or reworded a suspension — after which a client polling
/// and the same client streaming would be told different things about the same
/// run, which is worse than either answer being wrong.
///
/// **The last record, not any record.** A run that waited, was resumed and
/// carried on has a suspension in its history and is not suspended now.
///
/// An empty history has no state to report and is not this function's to
/// invent: the caller decides whether that is a task that does not exist or one
/// whose first record has not landed yet, and those get different answers.
pub(super) fn state_from_history(
    records: &[crate::journal::Record],
) -> Option<(TaskState, String)> {
    let last = records.last()?;
    Some(match last.kind() {
        RecordKind::RunSuspended { reason } => (TaskState::InputRequired, reason.to_string()),
        RecordKind::RunConcluded { outcome, .. } => (sealed_state(outcome), outcome.clone()),
        _ => (TaskState::Working, "running".to_owned()),
    })
}

async fn cancel_task(
    server: &A2aServer,
    caller: &Caller,
    params: CommonParams,
) -> Result<Value, RpcError> {
    let (id, records) = server
        .gate_task(caller, action::TASK_CANCEL, &params)
        .await?;
    let Some(last) = records.last() else {
        return Err(task_not_found(id));
    };
    // Which runs a stop can reach is the runtime's decision, not this door's:
    // a failed, exhausted or withheld run has concluded and is still
    // resumable, and stopping it is how its work gets unwound. A closed run
    // (succeeded, cancelled, abandoned) and a quarantined one — whose unwind a
    // quarantine forbids — are refused with A2A's code for exactly this.
    let fresh = server
        .runtime
        .request_cancel(
            id,
            &crate::core::Operator::authenticated(caller.actor.clone())
                .map_err(|e| internal("naming the authenticated caller", &e))?,
            "cancelled over A2A",
        )
        .await
        .map_err(|e| match e {
            crate::core::RuntimeError::AlreadyConcluded { outcome, .. } => RpcError::new(
                code::TASK_NOT_CANCELABLE,
                format!("this task already finished as '{outcome}'"),
            ),
            crate::core::RuntimeError::CannotUnwind { .. } => RpcError::new(
                code::TASK_NOT_CANCELABLE,
                "this task is quarantined; its work cannot be unwound until the \
                 deployment resolves it",
            ),
            other => internal("requesting a cancellation", &other),
        })?;
    // Somebody stopped it already and it has finished stopping: telling this
    // caller a cancellation is under way would describe a stop that is over.
    // (The runtime answers a closed run `false` rather than an error only when
    // the run was cancelled.)
    if let RecordKind::RunConcluded { outcome, .. } = last.kind()
        && !fresh
        && outcome == "cancelled"
    {
        return Err(RpcError::new(
            code::TASK_NOT_CANCELABLE,
            format!("this task already finished as '{outcome}'"),
        ));
    }

    // The same resolution every other path uses, from the records already in
    // hand: A2A 1.0 puts `contextId` on every task, and the one place that
    // hard-coded it absent handed a canceling caller a task it could not
    // continue or correlate.
    let case = records
        .iter()
        .find_map(|record| record.body.case.map(|case| case.to_string()));

    // Still `WORKING`, deliberately. The request is durable, and the run stops
    // at its next step boundary — reporting `CANCELED` here would claim it had
    // already stopped and unwound, which is exactly what has not happened yet.
    serde_json::to_value(task_of(
        id,
        TaskState::Working,
        "cancellation requested",
        case,
    ))
    .map_err(|error| internal("encoding a task", &error))
}

fn get_extended_card(server: &A2aServer, caller: &Caller) -> Result<Value, RpcError> {
    server.authorize(caller, action::CARD_EXTENDED, &server.card.name, None)?;
    serde_json::to_value(&server.extended).map_err(|e| internal("encoding the extended card", &e))
}

fn task_id(params: &CommonParams) -> Result<RunId, RpcError> {
    let Some(raw) = params.id.as_deref() else {
        return Err(RpcError::new(code::INVALID_PARAMS, "`id` is required"));
    };
    RunId::parse(raw).map_err(|_| {
        // Not found rather than invalid params: whether a string is a run id
        // this plane issued is not something a caller should learn from the
        // shape of the refusal.
        RpcError::new(code::TASK_NOT_FOUND, format!("no such task: {raw}"))
    })
}

/// The agent declining, with nothing about *why*.
///
/// The reason is deliberately absent. A denial reason describes the rule that
/// fired, and a rule describes the classification it protects — so a caller who
/// can send messages and read refusals can map the policy by probing it. The
/// operator sees the reason in the journal, where it belongs; the peer sees that
/// it was declined, which is all it can act on anyway.
fn declined(skill: &str) -> A2aMessage {
    A2aMessage {
        message_id: format!("declined-{skill}"),
        role: "ROLE_AGENT".to_owned(),
        parts: vec![Part {
            text: Some("this agent declined the request".to_owned()),
            data: None,
            raw: None,
            url: None,
            filename: None,
            media_type: Some("text/plain".to_owned()),
            metadata: None,
        }],
        context_id: None,
        task_id: None,
        metadata: None,
        extensions: Vec::new(),
        reference_task_ids: Vec::new(),
    }
}

pub(super) fn task_of(run: RunId, state: TaskState, detail: &str, case: Option<String>) -> A2aTask {
    A2aTask {
        id: run.to_string(),
        context_id: case,
        status: TaskStatus {
            state,
            message: Some(A2aMessage {
                message_id: format!("{run}-status"),
                role: "ROLE_AGENT".to_owned(),
                parts: vec![Part {
                    text: Some(detail.to_owned()),
                    data: None,
                    raw: None,
                    url: None,
                    filename: None,
                    media_type: Some("text/plain".to_owned()),
                    metadata: None,
                }],
                context_id: None,
                task_id: Some(run.to_string()),
                metadata: None,
                extensions: Vec::new(),
                reference_task_ids: Vec::new(),
            }),
            timestamp: None,
        },
        artifacts: None,
        history: None,
        metadata: None,
    }
}

fn push_runtime(server: &A2aServer) -> Result<&PushRuntime, RpcError> {
    server.push.as_ref().ok_or_else(push_not_supported_error)
}

fn push_not_supported_error() -> RpcError {
    RpcError::new(
        code::PUSH_NOT_SUPPORTED,
        "this agent does not implement push notifications; its card advertises pushNotifications as false",
    )
}

/// The task a push method addresses, refused unless this caller admitted it.
///
/// Owned before anything is read or written: a webhook registration is a
/// standing instruction to send a task's history somewhere, so one peer
/// registering on — or reading, listing, deleting the registrations of —
/// another's task would be a read of that task with a delivery address
/// attached.
async fn push_task<'a>(
    server: &'a A2aServer,
    caller: &Caller,
    raw: Option<&str>,
) -> Result<(RunId, &'a PushRuntime), RpcError> {
    let push = push_runtime(server)?;
    let raw = raw.ok_or_else(|| RpcError::new(code::INVALID_PARAMS, "`taskId` is required"))?;
    let task = RunId::parse(raw)
        .map_err(|_| RpcError::new(code::TASK_NOT_FOUND, format!("no such task: {raw}")))?;
    server
        .authorized_task(caller, action::TASK_PUSH, task)
        .await?;
    Ok((task, push))
}

fn push_request(params: &CommonParams) -> Result<PushRequest, RpcError> {
    let url = params
        .url
        .clone()
        .ok_or_else(|| RpcError::new(code::INVALID_PARAMS, "`url` is required"))?;
    Ok(PushRequest {
        id: params.id.clone(),
        task_id: params.push_task.clone(),
        url,
        token: params.token.clone(),
        authentication: params.authentication.clone(),
    })
}

fn validate_push_request(server: &A2aServer, request: &PushRequest) -> Result<(), RpcError> {
    let push = push_runtime(server)?;
    request.validate()?;
    let config = request.config(RunId::generate());
    if let Some(authentication) = &config.authentication {
        authentication
            .validate()
            .map_err(|error| RpcError::new(code::INVALID_PARAMS, error.to_string()))?;
    }
    push.transport
        .validate(&config)
        .map_err(|error| RpcError::new(code::INVALID_PARAMS, error.to_string()))
}

fn validate_inline_push(
    server: &A2aServer,
    caller: &Caller,
    skill: &str,
    request: &PushRequest,
) -> Result<(), RpcError> {
    if request
        .task_id
        .as_deref()
        .is_some_and(|task| !task.is_empty())
    {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            "taskPushNotificationConfig.taskId must be empty in SendMessage",
        ));
    }
    server.authorize(caller, action::TASK_PUSH, &format!("new:{skill}"), None)?;
    validate_push_request(server, request)
}

async fn register_push(
    server: &A2aServer,
    request: &PushRequest,
    task: RunId,
    next_seq: Seq,
) -> Result<crate::push::PushConfig, RpcError> {
    let push = push_runtime(server)?;
    if request
        .task_id
        .as_deref()
        .is_some_and(|configured| !configured.is_empty() && configured != task.to_string())
    {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            "push configuration taskId does not match its task",
        ));
    }
    request.validate()?;
    let config = request.config(task);
    if let Some(authentication) = &config.authentication {
        authentication
            .validate()
            .map_err(|error| RpcError::new(code::INVALID_PARAMS, error.to_string()))?;
    }
    push.transport
        .validate(&config)
        .map_err(|error| RpcError::new(code::INVALID_PARAMS, error.to_string()))?;
    // The deployment's own destinations are not the caller's to spend.
    let held: Vec<String> = push
        .store
        .list(task)
        .await
        .map_err(|error| internal("listing push configurations", &error))?
        .into_iter()
        .map(|c| c.id)
        .filter(|id| !crate::push::is_operator_id(id))
        .collect();
    if held.len() >= MAX_PUSH_CONFIGS_PER_TASK && !held.contains(&config.id) {
        return Err(RpcError::new(
            code::INVALID_PARAMS,
            format!(
                "a task holds at most {MAX_PUSH_CONFIGS_PER_TASK} push configurations; \
                 delete one first"
            ),
        ));
    }
    push.store
        .put(&config, next_seq)
        .await
        .map_err(|error| internal("storing a push configuration", &error))?;
    Ok(config)
}

async fn push_create(
    server: &A2aServer,
    caller: &Caller,
    params: &CommonParams,
) -> Result<Value, RpcError> {
    let (task, _) = push_task(server, caller, params.push_task.as_deref()).await?;
    let request = push_request(params)?;
    let head = server
        .runtime
        .journal()
        .head(task)
        .await
        .map_err(|error| internal("reading a task's head", &error))?;
    let tail = server
        .runtime
        .journal()
        .read(task, head.seq)
        .await
        .map_err(|error| internal("reading a task's journal", &error))?;
    let next_seq = if tail
        .last()
        .is_some_and(|record| matches!(record.kind(), RecordKind::RunConcluded { .. }))
    {
        head.seq
    } else {
        head.seq.saturating_add(1)
    };
    Ok(register_push(server, &request, task, next_seq)
        .await?
        .redacted())
}

async fn push_get(
    server: &A2aServer,
    caller: &Caller,
    params: &CommonParams,
) -> Result<Value, RpcError> {
    let (task, push) = push_task(server, caller, params.push_task.as_deref()).await?;
    let (id, reachable) = caller_push_id(params)?;
    let missing = || {
        RpcError::new(
            code::TASK_NOT_FOUND,
            format!("no push configuration '{id}' for task {task}"),
        )
    };
    if !reachable {
        return Err(missing());
    }
    push.store
        .get(task, id)
        .await
        .map_err(|error| internal("reading a push configuration", &error))?
        .map(|config| config.redacted())
        .ok_or_else(missing)
}

/// The `id` a push method names, and whether it lies in the caller's half of
/// the store — the only half a push method may reach.
///
/// A caller's webhooks and the deployment's own destinations share one
/// [`PushStore`](crate::push::PushStore), told apart by
/// [`OPERATOR_PREFIX`](crate::push::OPERATOR_PREFIX), and the outbox registers
/// an operator destination against every run. The A2A methods answer for the
/// caller's namespace and nothing else: an operator id is answered exactly as
/// a configuration that does not exist, so a peer can neither read the
/// deployment's destination URL nor delete it (and its delivery cursor) from
/// under the operator.
fn caller_push_id(params: &CommonParams) -> Result<(&str, bool), RpcError> {
    let id = params
        .id
        .as_deref()
        .ok_or_else(|| RpcError::new(code::INVALID_PARAMS, "`id` is required"))?;
    Ok((id, PushNamespace::Caller.owns_id(id)))
}

async fn push_list(
    server: &A2aServer,
    caller: &Caller,
    params: &CommonParams,
) -> Result<Value, RpcError> {
    let (task, push) = push_task(server, caller, params.push_task.as_deref()).await?;
    let configs = push
        .store
        .list(task)
        .await
        .map_err(|error| internal("listing push configurations", &error))?;
    Ok(json!({
        "configs": configs
            .iter()
            .filter(|config| PushNamespace::Caller.owns_id(&config.id))
            .map(crate::push::PushConfig::redacted)
            .collect::<Vec<_>>(),
        "nextPageToken": "",
    }))
}

async fn push_delete(
    server: &A2aServer,
    caller: &Caller,
    params: &CommonParams,
) -> Result<Value, RpcError> {
    let (task, push) = push_task(server, caller, params.push_task.as_deref()).await?;
    let (id, reachable) = caller_push_id(params)?;
    if !reachable {
        // The same answer a delete of a missing configuration gets.
        return Ok(json!({}));
    }
    push.store
        .delete(task, id)
        .await
        .map_err(|error| internal("deleting a push configuration", &error))?;
    Ok(json!({}))
}

#[cfg(test)]
mod state_agreement_tests {
    use super::{TaskState, sealed_state, state_of};

    /// The same task must not have two states depending on which path a client took.
    ///
    /// `state_of` answers the caller holding the immediate `SendMessage`
    /// response; `sealed_state` answers `GetTask`, `SubscribeToTask` and every
    /// streamed status update. They read the same run.
    ///
    /// The check runs over the crate's one `RunStatus` list, so adding a variant
    /// fails to compile in `state_of` (the match is exhaustive) and, once
    /// somebody maps it there, fails *here* until `sealed_state` is taught the
    /// same answer. That is the ordering the defect needs: the enum decides, the
    /// string agrees. Unifying them the other way — `state_of` delegating to
    /// `sealed_state` — makes both paths return the `_ => Failed` fallback for a
    /// new variant, with nothing to notice, and that is what this test exists to
    /// stop being reintroduced as a tidy-up.
    #[test]
    fn a_live_status_and_its_sealed_outcome_agree() {
        let statuses = crate::runtime::every_status();
        assert_eq!(
            statuses.len(),
            14,
            "a RunStatus variant was added or removed — decide which A2A state it \
             surfaces as, in `state_of` and in `sealed_state` both"
        );
        for status in &statuses {
            assert_eq!(
                state_of(status),
                sealed_state(status.as_str()),
                "'{}' surfaces as one state live and another once sealed",
                status.as_str()
            );
        }
    }

    /// And the mapping is not one answer for everything.
    ///
    /// Without this, `sealed_state` and `state_of` could both be changed to
    /// return `Failed` unconditionally and the agreement test above would pass
    /// perfectly — every A2A client would see every task as failed.
    #[test]
    fn the_mapping_distinguishes_more_than_failure() {
        let seen: std::collections::BTreeSet<_> = crate::runtime::every_status()
            .iter()
            .map(|status| format!("{:?}", state_of(status)))
            .collect();
        assert!(
            seen.len() >= 4,
            "the run-status mapping collapsed to {seen:?}; a client cannot tell \
             completed from cancelled from waiting"
        );
        assert_eq!(
            state_of(&crate::runtime::RunStatus::Succeeded),
            TaskState::Completed
        );
    }

    /// A withheld run is interrupted, not finished.
    ///
    /// Its work stands and it continues once its authority is restored. A
    /// terminal answer would close the peer's stream and release its push
    /// registration, so the peer would never hear how the task ended.
    #[test]
    fn a_withheld_task_is_interrupted_not_terminal() {
        let withheld = crate::runtime::every_status()
            .into_iter()
            .find(|status| matches!(status, crate::runtime::RunStatus::Withheld { .. }))
            .expect("every_status lists Withheld");
        assert_eq!(state_of(&withheld), TaskState::AuthRequired);
        assert!(
            !super::outcome_ends_task(withheld.as_str()),
            "a withheld conclusion released the peer's push registration"
        );
        assert!(super::outcome_ends_task("succeeded"));
        assert!(super::outcome_ends_task("failed"));
    }
}