polyc-a2a 2026.8.3

//! The A2A **v1.0** JSON-RPC surface.
//!
//! Methods are `PascalCase` (`SendMessage`, `GetTask`, `CancelTask`, `ListTasks`) —
//! the capability-ungated set every v1.0 server MUST implement — and `params`
//! is the request struct itself (not a positional array). Results follow the
//! `ProtoJSON` shapes in [`crate::types`]: `SendMessage` returns a wrapped
//! `{"task":…}`/`{"message":…}` oneof; `GetTask`/`CancelTask` return a bare
//! `Task`; `ListTasks` returns `{tasks, nextPageToken?}`. Errors carry the A2A
//! codes (`-32001…`).
//!
//! `SendStreamingMessage` and `TaskSubscription` (the card's `streaming`
//! capability, `#371`) are dispatched by `handle_streaming` instead of
//! `handle` — the HTTP layer (`crate::server`'s `json_rpc`) peeks the method
//! name and routes to whichever returns the right transport (a plain JSON
//! body vs. an SSE stream of JSON-RPC responses) before either ever touches
//! the body. Push config (webhook delivery) remains a capability-gated
//! follow-up (the card advertises it off).
//!
//! # One turn per task id
//!
//! A peer supplies its own `taskId` when it wants a send to be safe to repeat,
//! and this surface honors that: the same id twice must find the first send's
//! work rather than start it again. Nothing about the record says so on its
//! own, though — no write happens between a create and the turn's outcome, so
//! a task that a turn is running looks exactly like a task nobody ever
//! dispatched.
//!
//! So a send claims the record before it dials: `submitted` → `working`, under
//! the revision it read, one winner. A send that finds `working` is refused,
//! and a send that finds `submitted` knows it is looking at the one case the
//! claim leaves behind — a record that committed while its reply was lost —
//! and drives it. Without the claim, a peer whose client timed out and retried
//! would get a second turn, a fresh `exec_id`, and every side effect again.
//!
//! Every source dispatch claims, including a continuation of an
//! `input-required` task. State binds that claim to the durable ingress
//! `dispatch_id`, the authenticated edge process, and this delivery attempt,
//! then issues a new per-task fence. A late outcome from the message that
//! originally paused cannot therefore overwrite its continuation.
//!
//! Exact claim-RPC retries carry the same execution identity and recover the
//! same receipt. A different process or delivery attempt cannot piggyback on
//! it; after expiry it competes for a strictly newer State grant.

use std::collections::HashMap;
use std::pin::Pin;

use futures::{Stream, StreamExt as _};
use polyc_rpc_client::{IngressIdentity, IngressIdentityError};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use uuid::Uuid;

use crate::server::AppState;
use crate::store::{MAX_PAGE_SIZE, TaskStoreError, TaskUpdate};
use crate::task::{IngressReceiptError, TurnOutcome, TurnRequest, TurnStreamEvent};
use crate::types::{
    Artifact, Message, Part, Role, SendMessageResponse, StreamResponse, Task,
    TaskArtifactUpdateEvent, TaskState, TaskStatus, TaskStatusUpdateEvent,
};

/// Task metadata keys stashing a paused turn's approval `request_id`/
/// `tool_name` (`#792`) — the only place they survive between the task
/// entering `input-required` and a later continuation resolving it, since
/// `TurnOutcome` itself is not persisted.
const PENDING_APPROVAL_REQUEST_ID_KEY: &str = "pendingApprovalRequestId";
const PENDING_APPROVAL_TURN_ID_KEY: &str = "pendingApprovalTurnId";
const PENDING_APPROVAL_TOOL_NAME_KEY: &str = "pendingApprovalToolName";
const PENDING_APPROVAL_RESOLVE_TOKEN_KEY: &str = "pendingApprovalResolveToken";

/// Namespace prefix for peer conversation ids (`a2a:<uuid>`).
///
/// Also this edge's tenancy claim (#1691). The two are the same string on
/// purpose: an operator granting `allowed_namespaces` reads the prefix off a
/// conversation id and grants exactly that.
pub(crate) const NAMESPACE: &str = "a2a";

/// This edge's tenancy claim, for every turn it dials.
///
/// # Panics
///
/// Panics when [`NAMESPACE`] is not a valid claim. It is a compile-time
/// constant covered by `the_namespace_is_a_valid_claim`, so a panic here means
/// that constant changed to a value no operator can grant.
pub(crate) fn claimed_namespace() -> polyc_rpc_client::ClaimedNamespace {
    polyc_rpc_client::ClaimedNamespace::new(NAMESPACE)
        .expect("this edge's namespace is a valid claim")
}

/// Namespace the peer context coordinate is derived under.
const CONTEXT_ID_NAMESPACE: Uuid = Uuid::from_u128(0xa2a0_0000_0000_5000_8000_0000_0000_0002);

/// The conversation id for a peer's `contextId`.
///
/// A peer chooses its own `contextId`, and the durable record accepts one up to
/// [`MAX_ID_BYTES`](crate::store::MAX_ID_BYTES) — far longer than a journal
/// partition name can be once the storage codec escapes it. So the coordinate
/// is DERIVED rather than carried: the id is always `a2a:<uuid>`, whatever the
/// peer sent.
///
/// Always, never past a threshold. An identity that changes shape at a length
/// boundary is one a peer cannot predict, and one that two code paths can
/// disagree about for the same conversation.
///
/// The coordinate is length-prefix framed before hashing, because a `contextId`
/// may itself hold any byte a peer likes.
fn peer_conversation_id(context_id: &str) -> String {
    polyc_rpc_client::namespaced_id(
        NAMESPACE,
        &polyc_rpc_client::framed_conversation_id(CONTEXT_ID_NAMESPACE, &[context_id]),
    )
}

/// Default `ListTasks` page size. The ceiling a peer may ask for is
/// [`MAX_PAGE_SIZE`]; a request past it is clamped rather than refused, and
/// the durable read below it may answer with fewer still. A page that stops
/// short always carries the token to resume from, so a peer reads the rest by
/// asking again.
const DEFAULT_PAGE_SIZE: usize = 50;

/// The v1.0 JSON-RPC method names (`PascalCase`, matching the gRPC service).
mod methods {
    pub(super) const SEND_MESSAGE: &str = "SendMessage";
    pub(super) const SEND_STREAMING_MESSAGE: &str = "SendStreamingMessage";
    pub(super) const TASK_SUBSCRIPTION: &str = "TaskSubscription";
    pub(super) const GET_TASK: &str = "GetTask";
    pub(super) const CANCEL_TASK: &str = "CancelTask";
    pub(super) const LIST_TASKS: &str = "ListTasks";
}

// JSON-RPC standard + A2A-specific error codes (§5.4).
const PARSE_ERROR: i64 = -32700;
const INVALID_REQUEST: i64 = -32600;
const METHOD_NOT_FOUND: i64 = -32601;
const INVALID_PARAMS: i64 = -32602;
const INTERNAL_ERROR: i64 = -32603;
const TASK_NOT_FOUND: i64 = -32001;
const TASK_NOT_CANCELABLE: i64 = -32002;

/// A JSON-RPC 2.0 request envelope (the fields this edge reads).
#[derive(Debug, Deserialize)]
struct JsonRpcRequest {
    #[serde(default)]
    jsonrpc: String,
    #[serde(default)]
    id: Option<Value>,
    #[serde(default)]
    method: String,
    #[serde(default)]
    params: Value,
}

/// `SendMessage` params (the request struct is the JSON-RPC `params`).
#[derive(Debug, Deserialize)]
struct SendMessageRequest {
    message: Message,
}

/// `GetTask` params.
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct GetTaskRequest {
    id: String,
    #[serde(default)]
    history_length: Option<usize>,
}

/// `CancelTask` params.
#[derive(Debug, Deserialize)]
struct CancelTaskRequest {
    id: String,
}

/// `TaskSubscription` params — resubscribe to `id`'s event stream.
#[derive(Debug, Deserialize)]
struct TaskSubscriptionRequest {
    id: String,
}

/// `ListTasks` params.
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ListTasksRequest {
    #[serde(default)]
    context_id: Option<String>,
    #[serde(default)]
    page_size: Option<usize>,
    #[serde(default)]
    page_token: Option<String>,
}

/// Dispatch one inbound JSON-RPC request body to its v1.0 method, returning the
/// JSON-RPC response value (success or error). Never fails the HTTP layer.
pub(crate) async fn handle_for_peer(state: &AppState, body: &[u8], peer_id: &str) -> Value {
    let request: JsonRpcRequest = match serde_json::from_slice(body) {
        Ok(request) => request,
        Err(_) => return error_response(Value::Null, PARSE_ERROR, "Parse error"),
    };
    let id = request.id.clone().unwrap_or(Value::Null);

    if request.jsonrpc != "2.0" || request.method.is_empty() {
        return error_response(id, INVALID_REQUEST, "Invalid Request");
    }

    match request.method.as_str() {
        methods::SEND_MESSAGE => match parse::<SendMessageRequest>(request.params) {
            Ok(req) => send_message(state, id, req, peer_id).await,
            Err(err) => error_response(id, INVALID_PARAMS, &err),
        },
        methods::GET_TASK => match parse::<GetTaskRequest>(request.params) {
            Ok(req) => {
                let task_id = scoped_id("task", peer_id, &req.id);
                match state.store.get(&task_id).await {
                    Ok(Some(task)) => ok(id, &with_history_length(task, req.history_length)),
                    Ok(None) => {
                        error_response(id, TASK_NOT_FOUND, &format!("task not found: {}", req.id))
                    }
                    // A store that cannot be reached is NOT a missing task: a peer
                    // told "not found" would stop asking about work that is still
                    // recorded. Fail closed with the reason instead.
                    Err(err) => store_error_response(id, &err, &task_subject(&req.id)),
                }
            }
            Err(err) => error_response(id, INVALID_PARAMS, &err),
        },
        methods::CANCEL_TASK => match parse::<CancelTaskRequest>(request.params) {
            Ok(req) => cancel_task(state, id, &scoped_id("task", peer_id, &req.id)).await,
            Err(err) => error_response(id, INVALID_PARAMS, &err),
        },
        methods::LIST_TASKS => match parse::<ListTasksRequest>(request.params) {
            Ok(mut req) => {
                req.context_id = req
                    .context_id
                    .map(|context| scoped_id("context", peer_id, &context));
                list_tasks(state, id, &req).await
            }
            Err(err) => error_response(id, INVALID_PARAMS, &err),
        },
        other => error_response(id, METHOD_NOT_FOUND, &format!("Method not found: {other}")),
    }
}

#[cfg(test)]
async fn handle(state: &AppState, body: &[u8]) -> Value {
    handle_for_peer(state, body, "").await
}

/// Whether `body`'s JSON-RPC `method` names a streaming transport
/// (`SendStreamingMessage`/`TaskSubscription`, `#371`) — checked by the HTTP
/// layer ([`crate::server::json_rpc`]) BEFORE the body reaches [`handle`], so
/// those two methods are routed to the SSE response instead of the unary
/// JSON one. A malformed body, or any other method, is `false`; [`handle`]
/// still runs for it either way and reports the right JSON-RPC error.
#[must_use]
pub(crate) fn is_streaming_method(body: &[u8]) -> bool {
    let Ok(value) = serde_json::from_slice::<Value>(body) else {
        return false;
    };
    matches!(
        value.get("method").and_then(Value::as_str),
        Some(methods::SEND_STREAMING_MESSAGE | methods::TASK_SUBSCRIPTION)
    )
}

/// Whether an SSE request is itself a source-ingress acknowledgement.
#[must_use]
pub(crate) fn requires_durable_marker(body: &[u8]) -> bool {
    let Ok(value) = serde_json::from_slice::<Value>(body) else {
        return false;
    };
    value.get("method").and_then(Value::as_str) == Some(methods::SEND_STREAMING_MESSAGE)
}

/// Dispatch one inbound `SendStreamingMessage`/`TaskSubscription` request to
/// its SSE event stream — the streaming sibling of [`handle`]. Each item is a
/// complete JSON-RPC response value (success or error) ready to ride one SSE
/// `data:` line; [`crate::server::json_rpc`] wraps each as an `Event`.
///
/// Takes `state` by value — the caller's `AppState` is cheaply `Clone` — since
/// the returned stream keeps running after this call (and the HTTP handler
/// that made it) returns, so it cannot borrow anything scoped to the call.
pub(crate) fn handle_streaming_for_peer(
    state: AppState,
    body: &[u8],
    peer_id: String,
) -> Pin<Box<dyn Stream<Item = Value> + Send>> {
    let request: JsonRpcRequest = match serde_json::from_slice(body) {
        Ok(request) => request,
        Err(_) => {
            return Box::pin(futures::stream::once(async {
                error_response(Value::Null, PARSE_ERROR, "Parse error")
            }));
        }
    };
    let id = request.id.clone().unwrap_or(Value::Null);

    if request.jsonrpc != "2.0" || request.method.is_empty() {
        return Box::pin(futures::stream::once(async move {
            error_response(id, INVALID_REQUEST, "Invalid Request")
        }));
    }

    match request.method.as_str() {
        methods::SEND_STREAMING_MESSAGE => match parse::<SendMessageRequest>(request.params) {
            Ok(req) => {
                let stream = send_message_streaming(state, req, peer_id)
                    .map(move |item| item.into_response(id.clone()));
                Box::pin(stream)
            }
            Err(err) => Box::pin(futures::stream::once(async move {
                error_response(id, INVALID_PARAMS, &err)
            })),
        },
        methods::TASK_SUBSCRIPTION => match parse::<TaskSubscriptionRequest>(request.params) {
            Ok(req) => Box::pin(task_subscription_stream(
                state,
                id,
                scoped_id("task", &peer_id, &req.id),
            )),
            Err(err) => Box::pin(futures::stream::once(async move {
                error_response(id, INVALID_PARAMS, &err)
            })),
        },
        other => {
            let message = format!("Method not found: {other}");
            Box::pin(futures::stream::once(async move {
                error_response(id, METHOD_NOT_FOUND, &message)
            }))
        }
    }
}

#[cfg(test)]
fn handle_streaming(state: AppState, body: &[u8]) -> Pin<Box<dyn Stream<Item = Value> + Send>> {
    handle_streaming_for_peer(state, body, String::new())
}

/// `TaskSubscription`: resubscribe to `task_id`'s event stream.
///
/// This edge never holds a task in a `working` snapshot visible across two
/// separate connections — `SendMessage`/`SendStreamingMessage` only persist a
/// task once its turn has already reached a terminal or `input-required`
/// outcome (see [`send_message`]/[`send_message_streaming`]) — so a stored
/// task is always one of those two, and resubscribing to it reduces exactly
/// to: report that one snapshot and close the stream. A task genuinely
/// in-flight lives only inside the one `SendStreamingMessage` call driving
/// it; subscribing to ITS live progress from a second connection would need a
/// broader pub/sub than this store provides, and is out of scope here (see
/// the crate's `#371` PR description).
fn task_subscription_stream(
    state: AppState,
    id: Value,
    task_id: String,
) -> Pin<Box<dyn Stream<Item = Value> + Send>> {
    Box::pin(async_stream::stream! {
        match state.store.get(&task_id).await {
            Ok(Some(task)) => {
                let update = StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
                    task_id: task.id,
                    context_id: task.context_id,
                    status: task.status,
                    is_final: true,
                });
                yield ok(id, &update);
            }
            Ok(None) => {
                yield error_response(id, TASK_NOT_FOUND, &format!("task not found: {task_id}"));
            }
            Err(err) => {
                yield store_error_response(id, &err, &task_subject(&task_id));
            }
        }
    })
}

/// One item of a streaming method's response stream: a protocol event, or the
/// JSON-RPC error that ends it.
///
/// A refusal rides as an error rather than a `failed` status event, for the
/// reason the unary path refuses the same way: a status event naming a real
/// task id is a claim about that task's durable record, and this side does not
/// get to make one it has not read.
///
/// Read that as narrowly as it is written. It says this side never invents a
/// status for a record it never read — not that every status event it sends
/// matches what the record currently says. One case is deliberately the
/// second: when a turn's outcome cannot be recorded at all,
/// [`record_outcome`] rewrites the status to `failed`, and
/// [`run_streaming_turn`] yields that as a terminal `statusUpdate` while the
/// record still reads whatever it read before. The alternative is reporting a
/// `completed` a peer will poll as `submitted` forever, which is the worse
/// split of the two; the reasoning is at [`record_outcome`].
///
/// The event variant carries the whole protocol event and dwarfs the failure
/// one; boxing it would cost an allocation on every frame of a live stream to
/// shrink an enum that is only ever moved through one stream at a time.
#[allow(clippy::large_enum_variant)]
enum StreamItem {
    /// State durably acknowledged this source event.
    DurablyReceived,
    /// One protocol event, wrapped as a JSON-RPC success.
    Event(StreamResponse),
    /// The JSON-RPC error this stream ends with.
    Failure {
        /// The A2A or JSON-RPC error code.
        code: i64,
        /// What happened, and what the peer can do about it.
        message: String,
    },
}

impl StreamItem {
    /// Build the failure item one `(code, message)` pair stands for.
    fn failure((code, message): (i64, String)) -> Self {
        Self::Failure { code, message }
    }

    /// Render this item as the JSON-RPC response value that rides one SSE line.
    fn into_response(self, id: Value) -> Value {
        match self {
            Self::DurablyReceived => json!({ "__polychromeIngressDurable": true }),
            Self::Event(event) => ok(id, &event),
            Self::Failure { code, message } => error_response(id, code, &message),
        }
    }
}

fn final_status_update(task: Task) -> StreamItem {
    StreamItem::Event(StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
        task_id: task.id,
        context_id: task.context_id,
        status: task.status,
        is_final: true,
    }))
}

/// Identifies the internal marker the HTTP layer consumes before emitting SSE
/// headers. The marker never reaches an A2A peer.
pub(crate) fn is_durable_marker(value: &Value) -> bool {
    value
        .get("__polychromeIngressDurable")
        .and_then(Value::as_bool)
        == Some(true)
}

/// The streaming sibling of [`send_message`]: same task lifecycle (fresh send
/// vs. an `input-required` continuation, `#792`), but yields each step as it
/// happens — an opening snapshot of the freshly claimed `working` task, then
/// the turn's own stream folded through [`run_streaming_turn`] — instead of
/// folding straight to one terminal `Task`.
#[allow(clippy::too_many_lines)] // one ordered ingress → task claim → streamed-turn protocol
fn send_message_streaming(
    state: AppState,
    req: SendMessageRequest,
    peer_id: String,
) -> Pin<Box<dyn Stream<Item = StreamItem> + Send>> {
    Box::pin(async_stream::stream! {
        let (mut inbound, source_identity) = match prepare_peer_message(&peer_id, req.message) {
            Ok(prepared) => prepared,
            Err(err) => {
                yield StreamItem::failure((INVALID_PARAMS, err.to_string()));
                return;
            }
        };
        let turn = turn_request(&inbound, source_identity);
        let receipt = match state.runner.receive_ingress(turn.clone()).await {
            Ok(receipt) => receipt,
            Err(err) => {
                yield StreamItem::failure(ingress_error(&err));
                return;
            }
        };
        yield StreamItem::DurablyReceived;
        // Set when the peer's task id names a record that was accepted but
        // never driven; this send continues it instead of minting a second one.
        let mut recorded: Option<Task> = None;

        if let Some(task_id) = non_empty(inbound.task_id.clone()) {
            match state.store.get(&task_id).await {
                Ok(Some(existing)) if existing.status.state == TaskState::InputRequired => {
                    let inner = continue_input_required_streaming(
                        state,
                        inbound,
                        existing,
                        peer_id,
                        turn,
                        receipt.dispatch_id,
                    );
                    futures::pin_mut!(inner);
                    while let Some(item) = inner.next().await {
                        yield item;
                    }
                    return;
                }
                Ok(Some(existing)) if existing.status.state == TaskState::Submitted => {
                    recorded = Some(existing);
                }
                Ok(Some(existing)) => {
                    if let Some(items) = streaming_redelivery(&existing, &inbound) {
                        for item in items {
                            yield item;
                        }
                        return;
                    }
                    yield StreamItem::failure(
                        already_recorded_failure(&task_id, existing.status.state),
                    );
                    return;
                }
                Ok(None) => {}
                // Whether this id names a paused task is unknown, so refuse
                // rather than start a fresh turn that could duplicate or
                // clobber a live one — the unary path's own reasoning.
                Err(err) => {
                    yield StreamItem::failure((
                        INTERNAL_ERROR,
                        format!(
                            "{}. Retry with the same taskId {task_id}",
                            store_failure_text("this task could not be started", &err)
                        ),
                    ));
                    return;
                }
            }
        }

        let (context_id, task_id) = send_ids(recorded.as_ref(), &inbound, &peer_id);
        inbound.context_id = Some(context_id.clone());
        inbound.task_id = Some(task_id.clone());
        inbound.role = Role::User;

        // Already durable, and its opening history with it: this send drives
        // the turn the earlier one never reached.
        let (mut history, driving) = if let Some(existing) = recorded {
            driven_history(existing, inbound)
        } else {
            let submitted = submitted_task(&task_id, &context_id, inbound);
                // The snapshot this transport opens with is a DURABLE fact,
                // not just a stream frame: it is minted before the turn is
                // dialed, so the id the peer is about to be given already
                // resolves.
                if let Err(err) = state.store.create(&submitted).await {
                    yield StreamItem::failure(create_failure(&task_id, &err));
                    return;
                }
                (submitted.history.unwrap_or_default(), Vec::new())
        };

        // Claimed before the opening snapshot goes out, so a refusal ends the
        // stream having claimed nothing and yielded nothing — and so the
        // snapshot describes the record as it now stands rather than as it
        // stood a write ago. The unary path's own reasoning; see the module
        // doc.
        let ownership = match state.store.claim(&task_id, &driving, &receipt.dispatch_id).await {
            Ok(ownership) => ownership,
            Err(err) => {
                yield StreamItem::failure(claim_failure(&state, &task_id, &err).await);
                return;
            }
        };
        history.extend(driving);

        yield StreamItem::Event(StreamResponse::Task(Task {
            id: task_id.clone(),
            context_id: context_id.clone(),
            status: TaskStatus {
                state: TaskState::Working,
                message: None,
                timestamp: None,
            },
            artifacts: None,
            history: Some(history.clone()),
            metadata: None,
        }));

        let inner = run_streaming_turn(state, context_id, task_id, history, turn, ownership);
        futures::pin_mut!(inner);
        while let Some(item) = inner.next().await {
            yield item;
        }
    })
}

/// The streaming sibling of [`continue_input_required`]: submits the peer's
/// approve/deny decision the same way, then either re-drives the turn through
/// [`run_streaming_turn`] (a persisted decision) or reports the task
/// unchanged/re-prompted (the idempotent-no-op and unparseable-reply cases) as
/// a single terminal `statusUpdate`.
#[allow(clippy::too_many_lines)] // one ordered claim → approval decision → durable reply protocol
fn continue_input_required_streaming(
    state: AppState,
    inbound: Message,
    mut existing: Task,
    peer_id: String,
    mut turn: TurnRequest,
    dispatch_id: String,
) -> Pin<Box<dyn Stream<Item = StreamItem> + Send>> {
    Box::pin(async_stream::stream! {
        let ownership = match state
            .store
            .claim(&existing.id, std::slice::from_ref(&inbound), &dispatch_id)
            .await
        {
            Ok(ownership) => ownership,
            Err(err) => {
                yield StreamItem::failure(claim_failure(&state, &existing.id, &err).await);
                return;
            }
        };

        let Some((turn_id, request_id, tool_name, resolve_token)) = pending_approval(&existing) else {
            existing.status = status_with_message(
                TaskState::Failed,
                agent_message(
                    polyc_proto::approval_card_expired_text(),
                    &existing.context_id,
                    &existing.id,
                ),
            );
            existing
                .history
                .get_or_insert_with(Vec::new)
                .push(inbound);
            record_outcome(&state, &mut existing, &[], Some(&ownership)).await;
            yield final_status_update(existing);
            return;
        };

        let Some(approved) = parse_decision(&inbound.text()) else {
            let prompt = format!(
                "reply \"approve\" or \"deny\" to decide `{tool_name}` — anything else leaves it \
                pending"
            );
            existing.status = status_with_message(
                TaskState::InputRequired,
                agent_message(&prompt, &existing.context_id, &existing.id),
            );
            existing
                .history
                .get_or_insert_with(Vec::new)
                .push(inbound);
            record_outcome(&state, &mut existing, &[], Some(&ownership)).await;
            yield final_status_update(existing);
            return;
        };

        let conversation_id = peer_conversation_id(&existing.context_id);
        turn.conversation_id.clone_from(&conversation_id);
        // The redrive carries NO new utterance, the way every other edge's
        // does. The peer's "approve"/"deny" is a decision, not something said
        // to the agent: it is consumed by `respond` just below and delivered
        // as the signed response the resumed turn reads. Sending it as the
        // turn's text made this the one edge whose redrive looked like a
        // person typing, so the control plane decided the room afresh instead
        // of inheriting what the paused turn proved — which withholds, and
        // refuses the very note that was approved. Inert only while this edge
        // asserts "unknown"; it stops being inert the day it asserts what
        // §4.2 says it should.
        turn.text = String::new();
        match state
            .approvals
            .respond(
                &turn_id,
                &request_id,
                approved,
                &approval_reason(&peer_id),
                &conversation_id,
                &resolve_token,
            )
            .await
        {
            Ok(true) => {
                let mut history = existing.history.take().unwrap_or_default();
                history.push(inbound);
                let inner = run_streaming_turn(
                    state,
                    existing.context_id.clone(),
                    existing.id.clone(),
                    history,
                    turn,
                    ownership,
                );
                futures::pin_mut!(inner);
                while let Some(item) = inner.next().await {
                    yield item;
                }
            }
            // The approval ceremony was already settled, but this source
            // message was durably admitted and claimed. Persist its input and
            // settle that exact claim without redriving the turn.
            Ok(false) => {
                existing.status = already_decided_status(&existing, &tool_name);
                existing
                    .history
                    .get_or_insert_with(Vec::new)
                    .push(inbound);
                record_outcome(&state, &mut existing, &[], Some(&ownership)).await;
                yield final_status_update(existing);
            }
            Err(message) => {
                let mut appended = Vec::new();
                let (status, artifacts, metadata) = apply_outcome(
                    TurnOutcome::Failed { message },
                    &existing.context_id,
                    &existing.id,
                    &mut appended,
                );
                let mut history = existing.history.take().unwrap_or_default();
                history.push(inbound);
                history.extend(appended.iter().cloned());
                existing.status = status;
                existing.artifacts = artifacts;
                existing.history = Some(history);
                existing.metadata = metadata;
                record_outcome(&state, &mut existing, &appended, Some(&ownership)).await;
                yield StreamItem::Event(StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
                    task_id: existing.id.clone(),
                    context_id: existing.context_id.clone(),
                    status: existing.status,
                    is_final: true,
                }));
            }
        }
    })
}

/// Drive one turn to completion, streaming its text as `artifactUpdate`
/// chunks and yielding one terminal `statusUpdate` (`final: true`) built by
/// the SAME [`apply_outcome`] the unary path uses — persists the resulting
/// task exactly like [`send_message`]/[`continue_input_required`] do, so a
/// later `GetTask`/`CancelTask`/`ListTasks`/`TaskSubscription` sees a
/// consistent picture regardless of which transport produced it.
fn run_streaming_turn(
    state: AppState,
    context_id: String,
    task_id: String,
    mut history: Vec<Message>,
    turn: TurnRequest,
    ownership: crate::store::TaskOwnership,
) -> Pin<Box<dyn Stream<Item = StreamItem> + Send>> {
    Box::pin(async_stream::stream! {
        let turn_stream = dial_with_admission_streaming(state.clone(), turn);
        futures::pin_mut!(turn_stream);
        let mut outcome = None;
        let mut chunk_seen = false;
        let renewal = tokio::time::sleep(std::time::Duration::from_secs(90));
        tokio::pin!(renewal);
        let mut renewal_ordinal = 0_u64;
        loop {
            let event = tokio::select! {
                event = turn_stream.next() => match event {
                    Some(event) => event,
                    None => break,
                },
                () = &mut renewal => {
                    renewal_ordinal = renewal_ordinal.saturating_add(1);
                    if let Err(error) = state
                        .store
                        .renew(&task_id, &ownership, renewal_ordinal)
                        .await
                    {
                        outcome = Some(TurnOutcome::Failed {
                            message: store_failure_text(
                                "the task lost its State ownership while its turn ran",
                                &error,
                            ),
                        });
                        break;
                    }
                    renewal.as_mut().reset(
                        tokio::time::Instant::now() + std::time::Duration::from_secs(90),
                    );
                    continue;
                }
            };
            match event {
                TurnStreamEvent::DurablyReceived => {
                    yield StreamItem::DurablyReceived;
                }
                TurnStreamEvent::TextDelta(text) => {
                    if text.is_empty() {
                        continue;
                    }
                    yield StreamItem::Event(StreamResponse::ArtifactUpdate(TaskArtifactUpdateEvent {
                        task_id: task_id.clone(),
                        context_id: context_id.clone(),
                        artifact: Artifact {
                            artifact_id: format!("{task_id}-answer"),
                            name: None,
                            description: None,
                            parts: vec![Part::text(text)],
                            metadata: None,
                        },
                        append: chunk_seen.then_some(true),
                        last_chunk: None,
                    }));
                    chunk_seen = true;
                }
                TurnStreamEvent::Outcome(o) => outcome = Some(o),
            }
        }
        // `dial_with_admission_streaming`/`TurnRunner::run_turn_streaming`
        // always yield exactly one `Outcome` before ending; this only fires
        // if that contract were ever violated, so fail closed rather than
        // silently report an empty completion.
        let outcome = outcome.unwrap_or_else(|| TurnOutcome::Failed {
            message: "turn stream ended without a terminal outcome".to_owned(),
        });
        let mut appended = Vec::new();
        let (status, artifacts, metadata) =
            apply_outcome(outcome, &context_id, &task_id, &mut appended);
        history.extend(appended.iter().cloned());
        let mut task = Task {
            id: task_id,
            context_id,
            status,
            artifacts,
            history: Some(history),
            metadata,
        };
        // Recorded BEFORE the terminal frame goes out, so the status a peer
        // reads on the stream is the status a later `GetTask` answers with —
        // a record that could not be saved closes the stream as `failed`.
        record_outcome(&state, &mut task, &appended, Some(&ownership)).await;
        yield StreamItem::Event(StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
            task_id: task.id.clone(),
            context_id: task.context_id.clone(),
            status: task.status,
            is_final: true,
        }));
    })
}

/// The streaming sibling of [`dial_with_admission`]: holds the SAME admission
/// permit for the lifetime of the turn's event stream (rather than one
/// `.await`), so a burst beyond `turn_limit` sheds a `failed` outcome instead
/// of dialing an already-loaded agent, exactly as the unary path does.
fn dial_with_admission_streaming(
    state: AppState,
    turn: TurnRequest,
) -> Pin<Box<dyn Stream<Item = TurnStreamEvent> + Send>> {
    Box::pin(async_stream::stream! {
        let Some(permit) = state.turn_limit.try_admit() else {
            tracing::warn!("overloaded: shedding A2A streaming message/send");
            yield TurnStreamEvent::Outcome(TurnOutcome::Failed {
                message: polyc_proto::admission_shed_text().to_owned(),
            });
            return;
        };
        let inner = state.runner.run_turn_streaming(turn);
        futures::pin_mut!(inner);
        while let Some(event) = inner.next().await {
            yield event;
        }
        drop(permit);
    })
}

/// Run one turn for a `SendMessage` and return the wrapped task result. The
/// produced task is persisted in the store so a later
/// `GetTask`/`CancelTask`/`ListTasks` sees it.
///
/// When the inbound message names an existing task that's paused in
/// `input-required`, this is instead a continuation (`#792`): the text is a
/// human's approve/deny decision, not a fresh prompt — see
/// [`continue_input_required`].
#[allow(clippy::too_many_lines)] // unary mirror of the ordered durable streaming protocol above
async fn send_message(
    state: &AppState,
    id: Value,
    req: SendMessageRequest,
    peer_id: &str,
) -> Value {
    let (mut inbound, source_identity) = match prepare_peer_message(peer_id, req.message) {
        Ok(prepared) => prepared,
        Err(err) => return error_response(id, INVALID_PARAMS, &err.to_string()),
    };
    let turn = turn_request(&inbound, source_identity);
    let receipt = match state.runner.receive_ingress(turn.clone()).await {
        Ok(receipt) => receipt,
        Err(err) => return ingress_error_response(id, &err),
    };
    // Set when the peer's task id names a record that was accepted but never
    // driven; this send continues it instead of minting a second one.
    let mut recorded: Option<Task> = None;

    if let Some(task_id) = non_empty(inbound.task_id.clone()) {
        match state.store.get(&task_id).await {
            Ok(Some(existing)) if existing.status.state == TaskState::InputRequired => {
                return match continue_input_required(
                    state,
                    inbound,
                    existing,
                    peer_id,
                    turn,
                    &receipt.dispatch_id,
                )
                .await
                {
                    Ok(response) => ok(id, &response),
                    Err((code, message)) => error_response(id, code, &message),
                };
            }
            // Accepted and never started — a create that committed while its
            // reply was lost leaves exactly this, and only this: a send that
            // reached the dial has already claimed the record into `working`.
            // Driving it here is the way out; leaving it would strand the task
            // forever, since every retry under the same id would find the
            // record already there.
            Ok(Some(existing)) if existing.status.state == TaskState::Submitted => {
                tracing::warn!(
                    task_id = %task_id,
                    "this task was recorded and never started; driving it under the same id"
                );
                recorded = Some(existing);
            }
            Ok(Some(existing)) => {
                if let Some(response) = unary_redelivery(id.clone(), &existing, &inbound) {
                    return response;
                }
                let (code, message) = already_recorded_failure(&task_id, existing.status.state);
                return error_response(id, code, &message);
            }
            Ok(None) => {}
            // Reading the store failed, so whether this id names a paused task
            // is unknown. Starting a fresh turn under it would either duplicate
            // work or clobber a live task, so refuse instead.
            Err(err) => {
                return error_response(
                    id,
                    INTERNAL_ERROR,
                    &format!(
                        "{}. Retry with the same taskId {task_id}",
                        store_failure_text("this task could not be started", &err)
                    ),
                );
            }
        }
    }

    let (context_id, task_id) = send_ids(recorded.as_ref(), &inbound, peer_id);
    inbound.context_id = Some(context_id.clone());
    inbound.task_id = Some(task_id.clone());
    inbound.role = Role::User;

    // Already durable, and its opening history with it.
    let (mut history, driving) = if let Some(existing) = recorded {
        driven_history(existing, inbound)
    } else {
        let submitted = Task {
            id: task_id.clone(),
            context_id: context_id.clone(),
            status: TaskStatus {
                state: TaskState::Submitted,
                message: None,
                timestamp: None,
            },
            artifacts: None,
            history: Some(vec![inbound]),
            metadata: None,
        };
        // The record is minted BEFORE the turn is dialed, so a peer that
        // polls a task id it was just handed finds it — and so a control
        // plane that cannot record the task refuses it before anything
        // runs, rather than after.
        if let Err(err) = state.store.create(&submitted).await {
            let (code, message) = create_failure(&task_id, &err);
            return error_response(id, code, &message);
        }
        (submitted.history.unwrap_or_default(), Vec::new())
    };

    // Claim the record for this dispatch. Only the winner dials, so a peer's
    // retry under its own task id finds a running task instead of starting a
    // second turn under it — see the module doc.
    let ownership = match state
        .store
        .claim(&task_id, &driving, &receipt.dispatch_id)
        .await
    {
        Ok(ownership) => ownership,
        Err(err) => {
            let (code, message) = claim_failure(state, &task_id, &err).await;
            return error_response(id, code, &message);
        }
    };
    history.extend(driving);

    // Admission control (`#795`): acquire a turn slot before dialing so
    // concurrency is bounded. If none is free, the task fails closed with
    // the shared shed copy rather than dialing an already-loaded agent.
    let outcome = dial_with_admission(state, turn, &task_id, &ownership).await;
    let mut appended = Vec::new();
    let (status, artifacts, metadata) =
        apply_outcome(outcome, &context_id, &task_id, &mut appended);

    history.extend(appended.iter().cloned());
    let mut task = Task {
        id: task_id,
        context_id,
        status,
        artifacts,
        history: Some(history),
        metadata,
    };
    record_outcome(state, &mut task, &appended, Some(&ownership)).await;
    ok(id, &SendMessageResponse::Task(task))
}

/// The history a send that picks up a recorded-but-unstarted task runs with,
/// and the frames its claim appends.
///
/// The record's own history opens it; the message that actually drove the turn
/// is appended, because a reader that sees only the original frame cannot tell
/// which message ran. A retry resending the same message under the same id
/// carries the same `messageId`, and that one is already recorded, so it is
/// not written twice.
fn driven_history(existing: Task, driving: Message) -> (Vec<Message>, Vec<Message>) {
    let history = existing.history.unwrap_or_default();
    let already_recorded = history
        .iter()
        .any(|frame| frame.message_id == driving.message_id);
    let driving = if already_recorded {
        Vec::new()
    } else {
        vec![driving]
    };
    (history, driving)
}

/// The error a `SendMessage` answers with when its task id names a record that
/// is neither resumable nor startable.
///
/// Never a synthetic `Task`: a record wearing a real task id and a state this
/// side made up would say `failed` while `GetTask` on the same id says
/// something else.
fn already_recorded_failure(task_id: &str, state: TaskState) -> (i64, String) {
    // A running task is the one state with a different way out. It is not
    // finished, so there is nothing to read back yet, and it can stay running
    // for good if the run that claimed it ended without recording an outcome —
    // so the copy names both what to do while it runs and what to do when it
    // clearly will not finish.
    let next = if state == TaskState::Working {
        "Follow it with GetTask. If the run that owns it ended, it stays here and will not move \
         again — send this message under a new taskId to start over."
    } else {
        "Read it with GetTask, or send this message without a taskId to start a new task."
    };
    (
        INVALID_PARAMS,
        format!(
            "task {task_id} is already recorded as {}, so this message cannot start it. {next}",
            state_text(state)
        ),
    )
}

/// The error a `SendMessage` answers with when it could not claim the task it
/// was about to dispatch.
///
/// The claim is refused for one reason in practice — another send won it — and
/// the honest answer is what the record now says, so it is read back and
/// reported the way every other lost race on this surface is. A record that
/// still reads `submitted`, or that cannot be read at all, leaves the claim's
/// own failure as the only thing this side knows.
async fn claim_failure(state: &AppState, task_id: &str, err: &TaskStoreError) -> (i64, String) {
    if let Ok(Some(durable)) = state.store.get(task_id).await
        && durable.status.state != TaskState::Submitted
    {
        tracing::warn!(
            task_id = %task_id,
            recorded = ?durable.status.state,
            "this task was claimed by another send; reporting the recorded state"
        );
        return already_recorded_failure(task_id, durable.status.state);
    }
    let message = store_failure_text("this task could not be started", err);
    match err {
        TaskStoreError::Unavailable(_) => (INTERNAL_ERROR, message),
        _ => (INVALID_PARAMS, message),
    }
}

/// The error a refused create answers with, on the same terms.
///
/// The outage arm names the task id and says what a retry does with it: the
/// record either committed before the reply was lost or never committed at
/// all, and sending the same message again under that id resolves either one.
fn create_failure(task_id: &str, err: &TaskStoreError) -> (i64, String) {
    match err {
        TaskStoreError::AlreadyExists => (
            INVALID_PARAMS,
            format!(
                "task {task_id} was started by another request a moment ago. Read it with \
                 GetTask, or send this message without a taskId to start a new task."
            ),
        ),
        TaskStoreError::Refused(_) => (
            INVALID_PARAMS,
            store_failure_text("this task could not be started", err),
        ),
        _ => (
            INTERNAL_ERROR,
            format!(
                "{}. The record for task {task_id} may or may not have been saved; sending this \
                 message again with taskId {task_id} either starts it or continues it.",
                store_failure_text("this task could not be started", err)
            ),
        ),
    }
}

/// Plain wording for one lifecycle state, for the copy a peer reads. The wire
/// enum names ([`TaskState`]) stay on the wire.
const fn state_text(state: TaskState) -> &'static str {
    match state {
        TaskState::Submitted => "accepted and not yet started",
        TaskState::Working => "still running",
        TaskState::InputRequired => "waiting on a decision",
        TaskState::AuthRequired => "waiting on authentication",
        TaskState::Completed => "finished",
        TaskState::Failed => "failed",
        TaskState::Canceled => "canceled",
        TaskState::Rejected => "declined",
        TaskState::Unspecified => "in a state it does not name",
    }
}

/// Record `task`'s terminal (or paused) outcome.
///
/// The durable record is what a later `GetTask` answers from, so what commits
/// here is what the task IS. Two things stop the commit, and they mean
/// opposite things:
///
/// - **The record already finished or a newer source dispatch owns it.** That
///   is a race the design intends: a peer canceled the task while its turn ran,
///   a second `SendMessage` continued the same task, or a continuation raced
///   the connection that started it. The recorded outcome is the durable truth,
///   so it is read back and reported — inventing a `failed` here would tell the
///   peer one thing while `GetTask` told it another.
/// - **The record could not be reached.** The side effects ran either way, and
///   reporting `completed` for a task that polls as `submitted` forever is the
///   worse split, so the task is reported as failed naming the record failure.
///   This is the one place a status a peer is told deliberately outruns the
///   record behind it, on both transports: the streaming path yields it as a
///   terminal `statusUpdate` under the real task id. [`StreamItem`]'s doc
///   states the rule that bounds it.
async fn record_outcome(
    state: &AppState,
    task: &mut Task,
    appended: &[Message],
    ownership: Option<&crate::store::TaskOwnership>,
) {
    let update = TaskUpdate {
        status: &task.status,
        artifacts: task.artifacts.as_ref(),
        metadata: task.metadata.as_ref(),
        appended_history: appended,
    };
    let Err(err) = state.store.transition(&task.id, update, ownership).await else {
        return;
    };

    if matches!(err, TaskStoreError::Terminal | TaskStoreError::Running) {
        if let Ok(Some(durable)) = state.store.get(&task.id).await {
            tracing::warn!(
                task_id = %task.id,
                recorded = ?durable.status.state,
                ran = ?task.status.state,
                "this task finished while its turn was running; reporting the recorded outcome"
            );
            *task = durable;
        } else {
            tracing::warn!(
                task_id = %task.id,
                "this task finished while its turn was running, and the record could not be read \
                 back"
            );
            task.status = status_with_message(
                TaskState::Failed,
                agent_message(
                    "the turn ran, and this task had already finished, but its recorded outcome \
                     could not be read back",
                    &task.context_id,
                    &task.id,
                ),
            );
        }
        return;
    }

    tracing::error!(
        task_id = %task.id,
        error = %err,
        state = ?task.status.state,
        "the task's outcome could not be recorded; reporting it as failed"
    );
    task.status = status_with_message(
        TaskState::Failed,
        agent_message(
            &store_failure_text("the turn ran, but this task did not finish", &err),
            &task.context_id,
            &task.id,
        ),
    );
}

/// Plain-language wording for a store failure, prefixed with what it stopped.
fn store_failure_text(what_failed: &str, err: &TaskStoreError) -> String {
    match err {
        TaskStoreError::Unavailable(reason) => {
            format!("{what_failed}: the task record store could not be reached — {reason}")
        }
        TaskStoreError::Refused(reason) => {
            format!("{what_failed}: the task record store refused the record — {reason}")
        }
        TaskStoreError::AlreadyExists => {
            format!("{what_failed}: a task already exists under this id")
        }
        TaskStoreError::NotFound => format!("{what_failed}: no task exists under this id"),
        TaskStoreError::Terminal => format!("{what_failed}: this task already finished"),
        TaskStoreError::Running => format!("{what_failed}: a turn is already running under it"),
    }
}

/// What a read was asking for, as the error copy names it. `ListTasks` asks
/// for a context's tasks and every other read asks for one task, so the
/// subject is passed in rather than assumed — a peer debugging a list should
/// never be handed its context id labelled as a task id.
fn task_subject(task_id: &str) -> String {
    format!("task {task_id}")
}

/// The subject a `ListTasks` failure names.
fn context_subject(context_id: &str) -> String {
    format!("the tasks in context {context_id}")
}

/// The JSON-RPC error one store failure answers a read with.
///
/// An outage is an internal error, never `task not found`: a peer told a task
/// does not exist stops asking about work that is still recorded.
fn store_error_response(id: Value, err: &TaskStoreError, subject: &str) -> Value {
    match err {
        TaskStoreError::NotFound => {
            error_response(id, TASK_NOT_FOUND, &format!("{subject} was not found"))
        }
        TaskStoreError::Terminal => error_response(
            id,
            TASK_NOT_CANCELABLE,
            &format!("{subject} already finished, so it cannot be canceled"),
        ),
        // No read refuses this way — only a claim does, and `claim_failure`
        // answers that one by reading the record back. Reaching it here would
        // mean a read raced a claim, so it says what the record is doing
        // rather than reporting an outage that did not happen.
        TaskStoreError::Running => error_response(
            id,
            INVALID_PARAMS,
            &format!("{subject} is still running, so this request cannot be answered yet"),
        ),
        TaskStoreError::Refused(reason) => error_response(
            id,
            INVALID_PARAMS,
            &format!("the task record store refused this request: {reason}"),
        ),
        TaskStoreError::AlreadyExists | TaskStoreError::Unavailable(_) => error_response(
            id,
            INTERNAL_ERROR,
            &format!("the task record store is unreachable, so {subject} could not be read: {err}"),
        ),
    }
}

/// Resolve an `input-required` task's continuation (`#792`): `inbound`'s task
/// id names `existing`, a task already paused on an approval gate, so its
/// text is a decision rather than a fresh prompt. Submits the decision via
/// [`AppState::approvals`], then — once persisted — re-drives the SAME
/// conversation under the same source identity so the harness executes the
/// approved call (or synthesizes a denial) and the task reaches a real outcome
/// instead of hanging.
async fn continue_input_required(
    state: &AppState,
    inbound: Message,
    mut existing: Task,
    peer_id: &str,
    mut turn: TurnRequest,
    dispatch_id: &str,
) -> Result<SendMessageResponse, (i64, String)> {
    let ownership = match state
        .store
        .claim(&existing.id, std::slice::from_ref(&inbound), dispatch_id)
        .await
    {
        Ok(ownership) => ownership,
        Err(err) => {
            return Err(claim_failure(state, &existing.id, &err).await);
        }
    };

    let Some((turn_id, request_id, tool_name, resolve_token)) = pending_approval(&existing) else {
        // Defensive: an `input-required` task without a complete stashed
        // occurrence can't be resolved (including a card persisted before
        // turn identity existed). Fail the task rather than
        // silently ignoring the decision.
        existing.status = status_with_message(
            TaskState::Failed,
            agent_message(
                polyc_proto::approval_card_expired_text(),
                &existing.context_id,
                &existing.id,
            ),
        );
        existing.history.get_or_insert_with(Vec::new).push(inbound);
        record_outcome(state, &mut existing, &[], Some(&ownership)).await;
        return Ok(SendMessageResponse::Task(existing));
    };

    let Some(approved) = parse_decision(&inbound.text()) else {
        // Unparseable reply: re-prompt rather than guess. The task stays
        // `input-required`; nothing is submitted, nothing re-drives.
        let prompt = format!(
            "reply \"approve\" or \"deny\" to decide `{tool_name}` — anything else leaves it \
             pending"
        );
        existing.status = status_with_message(
            TaskState::InputRequired,
            agent_message(&prompt, &existing.context_id, &existing.id),
        );
        existing.history.get_or_insert_with(Vec::new).push(inbound);
        record_outcome(state, &mut existing, &[], Some(&ownership)).await;
        return Ok(SendMessageResponse::Task(existing));
    };

    let conversation_id = peer_conversation_id(&existing.context_id);
    turn.conversation_id.clone_from(&conversation_id);
    // No new utterance on a redrive — the same rule the streaming twin
    // states at length. The decision is consumed by `respond` below and
    // delivered as the signed response the resumed turn reads.
    turn.text = String::new();
    let outcome = match state
        .approvals
        .respond(
            &turn_id,
            &request_id,
            approved,
            &approval_reason(peer_id),
            &conversation_id,
            &resolve_token,
        )
        .await
    {
        Ok(true) => {
            // The resume dial is bounded by the same admission gate (`#795`)
            // as a fresh `SendMessage` — it's still a dial into the agent.
            dial_with_admission(state, turn, &existing.id, &ownership).await
        }
        // The approval ceremony was already settled, but this source message
        // was durably admitted and claimed. Persist its input and settle that
        // exact claim without redriving the turn.
        Ok(false) => {
            existing.status = already_decided_status(&existing, &tool_name);
            existing.history.get_or_insert_with(Vec::new).push(inbound);
            record_outcome(state, &mut existing, &[], Some(&ownership)).await;
            return Ok(SendMessageResponse::Task(existing));
        }
        Err(message) => TurnOutcome::Failed { message },
    };

    let mut appended = Vec::new();
    let (status, artifacts, metadata) =
        apply_outcome(outcome, &existing.context_id, &existing.id, &mut appended);
    let mut history = existing.history.take().unwrap_or_default();
    history.push(inbound);
    history.extend(appended.iter().cloned());
    existing.status = status;
    existing.artifacts = artifacts;
    existing.history = Some(history);
    existing.metadata = metadata;
    record_outcome(state, &mut existing, &appended, Some(&ownership)).await;
    Ok(SendMessageResponse::Task(existing))
}

/// The status an already-answered decision is reported with.
///
/// Nothing re-drives here because the decision was recorded elsewhere. The
/// source input and its newer ownership fence are still durably recorded; this
/// status explains why the claimed continuation did not launch another turn.
fn already_decided_status(existing: &Task, tool_name: &str) -> TaskStatus {
    status_with_message(
        existing.status.state,
        agent_message(
            &format!("the decision on `{tool_name}` was already recorded, so nothing new ran"),
            &existing.context_id,
            &existing.id,
        ),
    )
}

/// Dial the agent for `turn`, bounded by the shared admission gate (`#795`).
/// Acquires a turn slot before dialing so concurrency is bounded; if none is
/// free, sheds with the shared shed copy rather than dialing an
/// already-loaded agent. Shared by the fresh-send and continuation-resume
/// dial sites — both are a dial into the agent.
async fn dial_with_admission(
    state: &AppState,
    turn: TurnRequest,
    task_id: &str,
    ownership: &crate::store::TaskOwnership,
) -> TurnOutcome {
    let dial = async {
        let Some(permit) = state.turn_limit.try_admit() else {
            tracing::warn!("overloaded: shedding A2A message/send");
            return TurnOutcome::Failed {
                message: polyc_proto::admission_shed_text().to_owned(),
            };
        };
        let outcome = state.runner.run_turn(turn).await;
        drop(permit);
        outcome
    };
    tokio::pin!(dial);
    let renewal = tokio::time::sleep(std::time::Duration::from_secs(90));
    tokio::pin!(renewal);
    let mut ordinal = 0_u64;
    loop {
        tokio::select! {
            outcome = &mut dial => return outcome,
            () = &mut renewal => {
                ordinal = ordinal.saturating_add(1);
                if let Err(error) = state.store.renew(task_id, ownership, ordinal).await {
                    return TurnOutcome::Failed {
                        message: store_failure_text(
                            "the task lost its State ownership while its turn ran",
                            &error,
                        ),
                    };
                }
                renewal.as_mut().reset(
                    tokio::time::Instant::now() + std::time::Duration::from_secs(90),
                );
            }
        }
    }
}

/// Fold one [`TurnOutcome`] into the pieces a [`Task`] needs: `status`,
/// `artifacts`, and — for [`TurnOutcome::InputRequired`] — the `metadata`
/// stashing the pending occurrence's `turn_id`, `request_id`, `tool_name`, and
/// `resolve_token` so a later [`continue_input_required`] can find them again.
/// Shared by the fresh-send and continuation paths so they can't drift on what
/// an outcome means.
///
/// `appended` collects the messages this outcome ADDS to the task's history,
/// never the history itself: the durable record extends its own history and
/// never rewrites it, so what a transition carries is the delta.
fn apply_outcome(
    outcome: TurnOutcome,
    context_id: &str,
    task_id: &str,
    appended: &mut Vec<Message>,
) -> (
    TaskStatus,
    Option<Vec<Artifact>>,
    Option<HashMap<String, Value>>,
) {
    match outcome {
        TurnOutcome::Completed { text } => {
            let reply = agent_message(&text, context_id, task_id);
            appended.push(reply.clone());
            // v1.0 carries outputs as artifacts; also keep the reply on the
            // terminal status (where most clients read a final answer).
            let artifact = Artifact {
                artifact_id: new_uuid(),
                name: None,
                description: None,
                parts: vec![Part::text(text)],
                metadata: None,
            };
            (
                status_with_message(TaskState::Completed, reply),
                Some(vec![artifact]),
                None,
            )
        }
        // The HITL approval pause: the peer must answer before the task can
        // finish. Stash the occurrence identity, tool name, and resolve token
        // so a follow-up SendMessage naming this task is recognized as the
        // decision for this exact occurrence.
        TurnOutcome::InputRequired {
            turn_id,
            request_id,
            tool_name,
            prompt,
            resolve_token,
        } => {
            let mut metadata = HashMap::new();
            metadata.insert(
                PENDING_APPROVAL_TURN_ID_KEY.to_owned(),
                Value::from(turn_id),
            );
            metadata.insert(
                PENDING_APPROVAL_REQUEST_ID_KEY.to_owned(),
                Value::from(request_id),
            );
            metadata.insert(
                PENDING_APPROVAL_TOOL_NAME_KEY.to_owned(),
                Value::from(tool_name),
            );
            metadata.insert(
                PENDING_APPROVAL_RESOLVE_TOKEN_KEY.to_owned(),
                Value::from(resolve_token),
            );
            (
                status_with_message(
                    TaskState::InputRequired,
                    agent_message(&prompt, context_id, task_id),
                ),
                None,
                Some(metadata),
            )
        }
        TurnOutcome::Failed { message } => (
            status_with_message(
                TaskState::Failed,
                agent_message(&message, context_id, task_id),
            ),
            None,
            None,
        ),
    }
}

/// Read back the `turn_id`/`request_id`/`tool_name`/`resolve_token`
/// [`apply_outcome`] stashed in a task's metadata when it entered
/// `input-required`.
fn pending_approval(task: &Task) -> Option<(String, String, String, String)> {
    let metadata = task.metadata.as_ref()?;
    let turn_id = metadata
        .get(PENDING_APPROVAL_TURN_ID_KEY)?
        .as_str()?
        .to_owned();
    let request_id = metadata
        .get(PENDING_APPROVAL_REQUEST_ID_KEY)?
        .as_str()?
        .to_owned();
    let tool_name = metadata
        .get(PENDING_APPROVAL_TOOL_NAME_KEY)
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let resolve_token = metadata
        .get(PENDING_APPROVAL_RESOLVE_TOKEN_KEY)
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    Some((turn_id, request_id, tool_name, resolve_token))
}

/// Parse a plain-text approval decision from an `input-required`
/// continuation message. Case-insensitive; anything else is `None` (the
/// caller re-prompts rather than guessing what an ambiguous reply meant).
fn parse_decision(text: &str) -> Option<bool> {
    match text.trim().to_ascii_lowercase().as_str() {
        "approve" | "approved" | "yes" | "y" => Some(true),
        "deny" | "denied" | "no" | "n" | "reject" | "rejected" => Some(false),
        _ => None,
    }
}

/// Cancel a stored task: not-found / not-cancelable (terminal) errors, else
/// flip it to `CANCELED` and return the updated task.
///
/// The terminality check and the flip happen inside ONE durable command, on
/// the same row premise the check read: a get-then-write across two calls
/// could clobber a task that finished in between. Refusing a finished task is
/// part of the contract rather than a check this side performs, so a peer can
/// never cancel work that already completed.
async fn cancel_task(state: &AppState, id: Value, task_id: &str) -> Value {
    match state.store.cancel(task_id).await {
        Ok(task) => ok(id, &task),
        Err(err) => store_error_response(id, &err, &task_subject(task_id)),
    }
}

/// Build the `ListTasksResponse` value (`{tasks, nextPageToken?}`).
///
/// One context at a time. Tasks are indexed per context so a page costs one
/// index row plus the rows it returns, whatever the total; there is no
/// cross-context listing to fall back on, so a request that names no context
/// is refused rather than answered from a scan.
async fn list_tasks(state: &AppState, id: Value, req: &ListTasksRequest) -> Value {
    let Some(context_id) = req.context_id.as_deref().filter(|ctx| !ctx.is_empty()) else {
        return error_response(
            id,
            INVALID_PARAMS,
            "list tasks one context at a time: name the contextId whose tasks you want",
        );
    };
    let page_size = req
        .page_size
        .unwrap_or(DEFAULT_PAGE_SIZE)
        .clamp(1, MAX_PAGE_SIZE);
    // The page token IS the last id of the previous page — a keyset cursor
    // over the context's sorted index, opaque to the peer either way.
    match state
        .store
        .list(context_id, page_size, req.page_token.as_deref())
        .await
    {
        Ok(page) => {
            let mut result = json!({ "tasks": page.tasks });
            if let Some(next) = page.next_page_token {
                result["nextPageToken"] = Value::from(next);
            }
            success_response(id, result)
        }
        Err(err) => store_error_response(id, &err, &context_subject(context_id)),
    }
}

/// Truncate a task's history to its last `n` messages, when requested.
fn with_history_length(mut task: Task, history_length: Option<usize>) -> Task {
    if let (Some(n), Some(history)) = (history_length, task.history.as_mut())
        && history.len() > n
    {
        history.drain(0..history.len() - n);
    }
    task
}

/// Build an `agent`-role message carrying `text`, bound to the task/context.
fn agent_message(text: &str, context_id: &str, task_id: &str) -> Message {
    Message {
        message_id: new_uuid(),
        context_id: Some(context_id.to_owned()),
        task_id: Some(task_id.to_owned()),
        role: Role::Agent,
        parts: vec![Part::text(text)],
        metadata: None,
    }
}

/// A status carrying an agent message.
///
/// Named for what it builds rather than for where it is used: three of its
/// callers name a terminal state and two name `input-required`, which is
/// explicitly not terminal, and terminal-is-final is now a rule the durable
/// record enforces rather than a shade of meaning.
const fn status_with_message(state: TaskState, message: Message) -> TaskStatus {
    TaskStatus {
        state,
        message: Some(message),
        timestamp: None,
    }
}

/// Deserialize JSON-RPC `params` into a request struct, returning a
/// human-readable message on failure (for an `INVALID_PARAMS` error).
fn parse<T: for<'de> Deserialize<'de>>(params: Value) -> Result<T, String> {
    serde_json::from_value(params).map_err(|err| format!("Invalid params: {err}"))
}

/// A JSON-RPC success response carrying `value` as `result`, or an internal
/// error if `value` cannot be serialized.
fn ok<T: Serialize>(id: Value, value: &T) -> Value {
    match serde_json::to_value(value) {
        Ok(result) => success_response(id, result),
        Err(_) => error_response(id, INTERNAL_ERROR, "Internal error"),
    }
}

fn non_empty(value: Option<String>) -> Option<String> {
    value.filter(|s| !s.is_empty())
}

/// Pinned namespace for peer-scoped A2A task and context ids.
const PEER_SCOPE_NAMESPACE: Uuid = Uuid::from_u128(0xa2a0_0000_0000_5000_8000_0000_0000_0001);

const fn effective_peer_id(peer_id: &str) -> &str {
    if peer_id.is_empty() {
        "test-peer"
    } else {
        peer_id
    }
}

fn prepare_peer_message(
    peer_id: &str,
    mut message: Message,
) -> Result<(Message, IngressIdentity), IngressIdentityError> {
    let source_identity = a2a_source_identity(peer_id, &message.message_id)?;
    message.task_id = Some(message.task_id.map_or_else(
        || scoped_id("task", peer_id, &message.message_id),
        |task_id| scoped_id("task", peer_id, &task_id),
    ));
    message.context_id = Some(message.context_id.map_or_else(
        || scoped_id("context", peer_id, &message.message_id),
        |context_id| scoped_id("context", peer_id, &context_id),
    ));
    Ok((message, source_identity))
}

fn turn_request(inbound: &Message, source_identity: IngressIdentity) -> TurnRequest {
    let context_id = inbound
        .context_id
        .as_deref()
        .expect("prepare_peer_message always supplies a context id");
    TurnRequest {
        conversation_id: peer_conversation_id(context_id),
        exec_id: new_uuid(),
        source_identity,
        text: inbound.text(),
    }
}

fn ingress_error(err: &IngressReceiptError) -> (i64, String) {
    let code = if err.content_conflict {
        INVALID_PARAMS
    } else {
        INTERNAL_ERROR
    };
    (code, err.message.clone())
}

fn ingress_error_response(id: Value, err: &IngressReceiptError) -> Value {
    let (code, message) = ingress_error(err);
    error_response(id, code, &message)
}

fn approval_reason(peer_id: &str) -> String {
    format!("a2a:{}", effective_peer_id(peer_id))
}

fn send_ids(recorded: Option<&Task>, inbound: &Message, peer_id: &str) -> (String, String) {
    let context_id = recorded.map_or_else(
        || {
            non_empty(inbound.context_id.clone())
                .unwrap_or_else(|| scoped_id("context", peer_id, &inbound.message_id))
        },
        |existing| existing.context_id.clone(),
    );
    let task_id = non_empty(inbound.task_id.clone())
        .unwrap_or_else(|| scoped_id("task", peer_id, &inbound.message_id));
    (context_id, task_id)
}

/// Reports whether this task already recorded the inbound message's id.
///
/// The match is on the id alone. This does not compare content.
///
/// The guard against a reused id carrying new content is `receive_ingress`.
/// It settles that against the durable authority before the store is read.
/// The authority sees only the message TEXT: the edge sends
/// `Message::text`, which drops data, file, and url parts. A redelivery that
/// changes only a non-text part is therefore not refused. Those parts reach
/// neither the control plane nor the agent, and the store drops them too, so
/// nothing downstream can observe the difference.
///
/// This returned `Option<bool>` until `POLY-154`. A `false` arm answered a
/// content conflict. `.then_some(true)` yields `None` on a miss and never
/// `Some(false)`, so that arm was unreachable. The name and its refusal text
/// described a comparison this never performed.
///
/// Rebuilding the comparison here is not possible. The task record keeps no
/// content parts at all — see this crate's `store` module. So `existing`
/// carries empty parts, whatever the peer sent, and an edge-side content
/// check would have nothing to compare.
fn redelivered_message_id(existing: &Task, inbound: &Message) -> bool {
    existing.history.as_ref().is_some_and(|history| {
        history
            .iter()
            .any(|recorded| recorded.message_id == inbound.message_id)
    })
}

fn submitted_task(task_id: &str, context_id: &str, inbound: Message) -> Task {
    Task {
        id: task_id.to_owned(),
        context_id: context_id.to_owned(),
        status: TaskStatus {
            state: TaskState::Submitted,
            message: None,
            timestamp: None,
        },
        artifacts: None,
        history: Some(vec![inbound]),
        metadata: None,
    }
}

/// Replays a recorded task when the inbound message is a redelivery.
///
/// The receipt marker repeats here. The caller already yielded one, so a
/// redelivered send emits two, and a sibling case asserts the receipt arrives
/// exactly once on the refusal path. The transport hides it: `server.rs`
/// filters every marker out of the SSE stream.
///
/// It stays anyway. This change removes an unreachable branch and alters no
/// behavior, and dropping the second marker would alter some. That belongs in
/// a change whose claim can be about the stream.
fn streaming_redelivery(existing: &Task, inbound: &Message) -> Option<Vec<StreamItem>> {
    redelivered_message_id(existing, inbound).then(|| {
        vec![
            StreamItem::DurablyReceived,
            StreamItem::Event(StreamResponse::Task(existing.clone())),
        ]
    })
}

fn unary_redelivery(id: Value, existing: &Task, inbound: &Message) -> Option<Value> {
    redelivered_message_id(existing, inbound)
        .then(|| ok(id, &SendMessageResponse::Task(existing.clone())))
}

fn a2a_source_identity(
    peer_id: &str,
    message_id: &str,
) -> Result<IngressIdentity, IngressIdentityError> {
    if message_id.trim().is_empty() {
        return Err(IngressIdentityError::EmptyReportedId);
    }
    IngressIdentity::reported_components(
        format!("a2a:{}", effective_peer_id(peer_id)),
        &[message_id],
    )
}

fn scoped_id(kind: &str, peer_id: &str, raw: &str) -> String {
    if peer_id.is_empty() {
        return raw.to_owned();
    }
    let peer = effective_peer_id(peer_id);
    let prefix = polyc_rpc_client::framed_conversation_id(PEER_SCOPE_NAMESPACE, &[kind, peer]);
    if raw
        .strip_prefix(&prefix)
        .is_some_and(|suffix| suffix.starts_with(':'))
    {
        return raw.to_owned();
    }
    let id = polyc_rpc_client::framed_conversation_id(PEER_SCOPE_NAMESPACE, &[kind, peer, raw]);
    format!("{prefix}:{id}")
}

fn new_uuid() -> String {
    Uuid::now_v7().to_string()
}

/// A JSON-RPC 2.0 success response carrying `result`. Builds the object by hand
/// so `id`/`result` are moved in (the `json!` macro would only borrow them).
fn success_response(id: Value, result: Value) -> Value {
    let mut obj = serde_json::Map::new();
    obj.insert("jsonrpc".to_owned(), Value::from("2.0"));
    obj.insert("id".to_owned(), id);
    obj.insert("result".to_owned(), result);
    Value::Object(obj)
}

/// A JSON-RPC 2.0 error response.
fn error_response(id: Value, code: i64, message: &str) -> Value {
    let mut obj = serde_json::Map::new();
    obj.insert("jsonrpc".to_owned(), Value::from("2.0"));
    obj.insert("id".to_owned(), id);
    obj.insert(
        "error".to_owned(),
        json!({ "code": code, "message": message }),
    );
    Value::Object(obj)
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    /// #1691: the constant this edge claims is one an operator can grant.
    /// Calls the production helper so the `expect()` in `claimed_namespace`
    /// is what this exercises, not a copy of the literal.
    #[test]
    fn the_namespace_is_a_valid_claim() {
        assert_eq!(super::claimed_namespace().as_str(), super::NAMESPACE);
    }

    use std::future::Future;
    use std::pin::Pin;
    use std::sync::Arc;

    use std::collections::VecDeque;
    use std::sync::Mutex;

    use super::*;

    const TEST_TURN: &str = "00000000-0000-0000-0000-000000000001";
    use crate::store::test_double::InMemoryTaskStore;
    use crate::task::{ApprovalResponder, TurnRunner};

    struct StubRunner(TurnOutcome);
    impl TurnRunner for StubRunner {
        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            let outcome = self.0.clone();
            Box::pin(async move { outcome })
        }
    }

    struct ReceiptCheckingRunner {
        outcome: TurnOutcome,
        seen: Mutex<HashMap<IngressIdentity, String>>,
    }

    impl ReceiptCheckingRunner {
        fn new(outcome: TurnOutcome) -> Self {
            Self {
                outcome,
                seen: Mutex::new(HashMap::new()),
            }
        }
    }

    impl TurnRunner for ReceiptCheckingRunner {
        fn receive_ingress<'a>(
            &'a self,
            req: TurnRequest,
        ) -> Pin<
            Box<
                dyn Future<Output = Result<crate::task::IngressReceipt, IngressReceiptError>>
                    + Send
                    + 'a,
            >,
        > {
            let result = {
                let mut seen = self.seen.lock().unwrap();
                match seen.get(&req.source_identity) {
                    Some(text) if text != &req.text => Err(IngressReceiptError {
                        message: "source event was already received with different content"
                            .to_owned(),
                        retryable: false,
                        content_conflict: true,
                    }),
                    Some(_) => Ok(crate::task::IngressReceipt {
                        dispatch_id: "test-dispatch".to_owned(),
                    }),
                    None => {
                        seen.insert(req.source_identity, req.text);
                        Ok(crate::task::IngressReceipt {
                            dispatch_id: "test-dispatch".to_owned(),
                        })
                    }
                }
            };
            Box::pin(async move { result })
        }

        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            let outcome = self.outcome.clone();
            Box::pin(async move { outcome })
        }
    }

    /// A [`TurnRunner`] that panics if called — used to prove an admission
    /// shed never reaches the dial (`#795`).
    struct PanickingRunner;
    impl TurnRunner for PanickingRunner {
        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            Box::pin(async { panic!("a shed request must never reach the agent dial") })
        }
    }

    /// A [`TurnRunner`] that returns one queued [`TurnOutcome`] per call, in
    /// order — for tests that drive an initial turn plus a resume and need
    /// each call to see a different outcome (`SequenceRunner` — a
    /// continuation "approve" resumes to a fresh `Completed`, not the same
    /// `InputRequired` it started from).
    struct SequenceRunner(Mutex<VecDeque<TurnOutcome>>);
    impl SequenceRunner {
        fn new(outcomes: Vec<TurnOutcome>) -> Self {
            Self(Mutex::new(outcomes.into()))
        }
    }
    impl TurnRunner for SequenceRunner {
        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            let next = self
                .0
                .lock()
                .unwrap()
                .pop_front()
                .unwrap_or(TurnOutcome::Failed {
                    message: "SequenceRunner: no more stubbed outcomes".to_owned(),
                });
            Box::pin(async move { next })
        }
    }

    /// A [`TurnRunner`] that counts its dispatches and holds every turn open
    /// until the test releases it.
    ///
    /// The only way to have a second send arrive while a turn is GENUINELY in
    /// flight: a runner that returns promptly closes the window the claim
    /// exists to cover, and the record's convergence would then hide a second
    /// dispatch that really happened.
    struct GatedRunner {
        dispatches: Arc<std::sync::atomic::AtomicUsize>,
        started: Arc<tokio::sync::Notify>,
        release: Arc<tokio::sync::Notify>,
        outcome: TurnOutcome,
    }

    impl GatedRunner {
        /// The runner, its dispatch counter, its "a turn has begun" signal,
        /// and the handle that lets the held turn finish.
        fn build(
            outcome: TurnOutcome,
        ) -> (
            Arc<Self>,
            Arc<std::sync::atomic::AtomicUsize>,
            Arc<tokio::sync::Notify>,
            Arc<tokio::sync::Notify>,
        ) {
            let dispatches = Arc::new(std::sync::atomic::AtomicUsize::new(0));
            let started = Arc::new(tokio::sync::Notify::new());
            let release = Arc::new(tokio::sync::Notify::new());
            let runner = Arc::new(Self {
                dispatches: Arc::clone(&dispatches),
                started: Arc::clone(&started),
                release: Arc::clone(&release),
                outcome,
            });
            (runner, dispatches, started, release)
        }

        async fn begin(&self) {
            self.dispatches
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            self.started.notify_one();
            self.release.notified().await;
        }
    }

    impl TurnRunner for GatedRunner {
        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            Box::pin(async move {
                self.begin().await;
                self.outcome.clone()
            })
        }

        fn run_turn_streaming<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Stream<Item = TurnStreamEvent> + Send + 'a>> {
            Box::pin(async_stream::stream! {
                self.begin().await;
                yield TurnStreamEvent::Outcome(self.outcome.clone());
            })
        }
    }

    /// An [`ApprovalResponder`] that records every call and always resolves
    /// to a fixed `persisted` outcome.
    struct StubApprovalResponder {
        persisted: bool,
        calls: Mutex<Vec<(String, bool, String)>>,
    }
    impl StubApprovalResponder {
        fn new(persisted: bool) -> Self {
            Self {
                persisted,
                calls: Mutex::new(Vec::new()),
            }
        }
    }
    impl ApprovalResponder for StubApprovalResponder {
        fn respond<'a>(
            &'a self,
            turn_id: &'a str,
            request_id: &'a str,
            approved: bool,
            _reason: &'a str,
            conversation_id: &'a str,
            _resolve_token: &'a str,
        ) -> Pin<Box<dyn Future<Output = Result<bool, String>> + Send + 'a>> {
            self.calls.lock().unwrap().push((
                format!("{turn_id}:{request_id}"),
                approved,
                conversation_id.to_owned(),
            ));
            let persisted = self.persisted;
            Box::pin(async move { Ok(persisted) })
        }
    }

    fn state(outcome: TurnOutcome) -> AppState {
        state_with_limit(outcome, 64)
    }

    fn state_with_limit(outcome: TurnOutcome, max_concurrent_turns: usize) -> AppState {
        AppState {
            card: Arc::new(json!({})),
            runner: Arc::new(StubRunner(outcome)),
            approvals: Arc::new(StubApprovalResponder::new(true)),
            store: Arc::new(InMemoryTaskStore::new()),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(max_concurrent_turns),
            // `rpc::handle` is dispatch-only — the auth gate lives in
            // `server::json_rpc`, so these tests never touch it.
            peers: crate::server::PeerAuthenticator::single("test-peer", "unused-in-rpc-tests")
                .unwrap(),
        }
    }

    /// Build state around a [`SequenceRunner`] and a [`StubApprovalResponder`]
    /// whose recorded calls the test can inspect via the returned `Arc`.
    fn state_with_sequence(
        outcomes: Vec<TurnOutcome>,
        approval_persisted: bool,
    ) -> (AppState, Arc<StubApprovalResponder>) {
        let approvals = Arc::new(StubApprovalResponder::new(approval_persisted));
        let state = AppState {
            card: Arc::new(json!({})),
            runner: Arc::new(SequenceRunner::new(outcomes)),
            approvals: approvals.clone(),
            store: Arc::new(InMemoryTaskStore::new()),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(64),
            // `rpc::handle` is dispatch-only — the auth gate lives in
            // `server::json_rpc`, so these tests never touch it.
            peers: crate::server::PeerAuthenticator::single("test-peer", "unused-in-rpc-tests")
                .unwrap(),
        };
        (state, approvals)
    }

    /// Collect one streaming method's protocol events, failing the test on the
    /// JSON-RPC error a refusal would end the stream with.
    async fn stream_events(
        stream: Pin<Box<dyn Stream<Item = StreamItem> + Send>>,
    ) -> Vec<StreamResponse> {
        stream
            .filter_map(|item| {
                futures::future::ready(match item {
                    StreamItem::Event(event) => Some(event),
                    StreamItem::DurablyReceived => None,
                    StreamItem::Failure { code, message } => {
                        panic!("unexpected stream failure {code}: {message}")
                    }
                })
            })
            .collect()
            .await
    }

    fn send_body(text: &str) -> Vec<u8> {
        serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 1, "method": "SendMessage",
            "params": { "message": {
                "messageId": "m1", "role": "ROLE_USER",
                "parts": [{ "text": text }], "contextId": "ctx-1"
            }}
        }))
        .unwrap()
    }

    #[tokio::test]
    async fn message_id_is_required_before_a_task_is_minted() {
        let state = state(TurnOutcome::Completed {
            text: "unused".to_owned(),
        });
        let body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 1, "method": "SendMessage", "params": {
                "message": { "role": "ROLE_USER", "parts": [{"text": "q"}] }
            }
        }))
        .unwrap();
        let response = handle_for_peer(&state, &body, "weather-peer").await;
        assert_eq!(response["error"]["code"], INVALID_PARAMS);
        assert!(
            response["error"]["message"]
                .as_str()
                .unwrap()
                .contains("event id")
        );
    }

    #[tokio::test]
    async fn same_message_retry_reuses_task_but_changed_content_conflicts() {
        let state = state_with_dyn_runner(Arc::new(ReceiptCheckingRunner::new(
            TurnOutcome::Completed {
                text: "done".to_owned(),
            },
        )));
        let first = handle_for_peer(&state, &send_body("q"), "weather-peer").await;
        let retry = handle_for_peer(&state, &send_body("q"), "weather-peer").await;
        assert_eq!(first["result"]["task"]["id"], retry["result"]["task"]["id"]);

        let conflict = handle_for_peer(&state, &send_body("different"), "weather-peer").await;
        assert_eq!(conflict["error"]["code"], INVALID_PARAMS);
        assert!(
            conflict["error"]["message"]
                .as_str()
                .unwrap()
                .contains("different content")
        );
    }

    #[tokio::test]
    async fn identical_peer_message_ids_have_isolated_tasks_and_contexts() {
        let state = state(TurnOutcome::Completed {
            text: "done".to_owned(),
        });
        let peer_a = handle_for_peer(&state, &send_body("q"), "peer-a").await;
        let peer_b = handle_for_peer(&state, &send_body("q"), "peer-b").await;
        assert_ne!(
            peer_a["result"]["task"]["id"],
            peer_b["result"]["task"]["id"]
        );
        assert_ne!(
            peer_a["result"]["task"]["contextId"],
            peer_b["result"]["task"]["contextId"]
        );

        let task_a = peer_a["result"]["task"]["id"].as_str().unwrap();
        let get = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "GetTask", "params": { "id": task_a }
        }))
        .unwrap();
        let collision = handle_for_peer(&state, &get, "peer-b").await;
        assert_eq!(collision["error"]["code"], TASK_NOT_FOUND);
    }

    /// A `SendMessageRequest` naming `task_id`, as a peer supplying its own id
    /// sends it on the streaming transport.
    fn send_streaming_request(seq: i64, task_id: &str, text: &str) -> SendMessageRequest {
        SendMessageRequest {
            message: Message {
                message_id: format!("m{seq}"),
                context_id: Some("ctx-1".to_owned()),
                task_id: Some(task_id.to_owned()),
                role: Role::User,
                parts: vec![Part::text(text)],
                metadata: None,
            },
        }
    }

    /// A `SendMessageRequest` naming `context_id`, e.g. for driving
    /// [`send_message_streaming`] directly (which — unlike `handle` — takes
    /// the already-deserialized request, not a raw JSON-RPC body).
    fn send_message_request(text: &str, context_id: &str) -> SendMessageRequest {
        SendMessageRequest {
            message: Message {
                message_id: "m1".to_owned(),
                context_id: Some(context_id.to_owned()),
                task_id: None,
                role: Role::User,
                parts: vec![Part::text(text)],
                metadata: None,
            },
        }
    }

    /// Build [`AppState`] around an arbitrary task store — for the cases that
    /// prove what the JSON-RPC surface answers when the store refuses or
    /// cannot be reached at all.
    fn state_with_store(outcome: TurnOutcome, store: Arc<dyn crate::store::TaskStore>) -> AppState {
        AppState {
            card: Arc::new(json!({})),
            runner: Arc::new(StubRunner(outcome)),
            approvals: Arc::new(StubApprovalResponder::new(true)),
            store,
            turn_limit: polyc_runtime::admission::AdmissionGate::new(64),
            peers: crate::server::PeerAuthenticator::single("test-peer", "unused-in-rpc-tests")
                .unwrap(),
        }
    }

    /// A store whose every operation reports an outage.
    fn unreachable_store() -> Arc<dyn crate::store::TaskStore> {
        Arc::new(InMemoryTaskStore::unreachable("state plane is unreachable"))
    }

    /// Build [`AppState`] around an arbitrary `Arc<dyn TurnRunner>` — for tests
    /// that need a runner overriding [`TurnRunner::run_turn_streaming`]
    /// ([`StreamingStubRunner`]) rather than the unary-only [`StubRunner`].
    fn state_with_dyn_runner(runner: Arc<dyn TurnRunner>) -> AppState {
        AppState {
            card: Arc::new(json!({})),
            runner,
            approvals: Arc::new(StubApprovalResponder::new(true)),
            store: Arc::new(InMemoryTaskStore::new()),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(64),
            // `rpc::handle`/`rpc::handle_streaming` are dispatch-only — the
            // auth gate lives in `server::json_rpc`, so these tests never
            // touch it.
            peers: crate::server::PeerAuthenticator::single("test-peer", "unused-in-rpc-tests")
                .unwrap(),
        }
    }

    /// A [`TurnRunner`] that streams a FIXED sequence of [`TurnStreamEvent`]s
    /// rather than folding to one [`TurnOutcome`] — proves
    /// `send_message_streaming`'s own orchestration (chunk ordering, the
    /// terminal status) independent of `AgentDialerRunner`'s own dial/fold
    /// logic (`#371`).
    struct StreamingStubRunner(Vec<TurnStreamEvent>);
    impl TurnRunner for StreamingStubRunner {
        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            // Not exercised by the streaming-path tests below; fold like the
            // trait's default would, for completeness.
            let outcome = self
                .0
                .iter()
                .find_map(|e| match e {
                    TurnStreamEvent::Outcome(o) => Some(o.clone()),
                    TurnStreamEvent::DurablyReceived | TurnStreamEvent::TextDelta(_) => None,
                })
                .unwrap_or(TurnOutcome::Completed {
                    text: String::new(),
                });
            Box::pin(async move { outcome })
        }

        fn run_turn_streaming<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Stream<Item = TurnStreamEvent> + Send + 'a>> {
            Box::pin(futures::stream::iter(self.0.clone()))
        }
    }

    #[tokio::test]
    async fn send_message_returns_wrapped_completed_task() {
        let state = state(TurnOutcome::Completed {
            text: "42".to_owned(),
        });
        let resp = handle(&state, &send_body("q")).await;
        let task = &resp["result"]["task"];
        assert_eq!(task["status"]["state"], "TASK_STATE_COMPLETED");
        assert_eq!(task["contextId"], "ctx-1");
        assert_eq!(task["status"]["message"]["parts"][0]["text"], "42");
        assert_eq!(task["artifacts"][0]["parts"][0]["text"], "42");
        assert_eq!(task["history"][1]["role"], "ROLE_AGENT");
    }

    /// `#795`: with the admission gate already at zero capacity, `SendMessage`
    /// must fail closed with the shared shed copy — and must NOT reach the
    /// agent dial (`PanickingRunner` proves that) — rather than queue
    /// unboundedly onto an already-loaded turn.
    #[tokio::test]
    async fn send_message_sheds_when_over_the_admission_limit() {
        let state = AppState {
            card: Arc::new(json!({})),
            runner: Arc::new(PanickingRunner),
            approvals: Arc::new(StubApprovalResponder::new(true)),
            store: Arc::new(InMemoryTaskStore::new()),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(0),
            peers: crate::server::PeerAuthenticator::single("test-peer", "unused-in-rpc-tests")
                .unwrap(),
        };
        let resp = handle(&state, &send_body("q")).await;
        let task = &resp["result"]["task"];
        assert_eq!(task["status"]["state"], "TASK_STATE_FAILED");
        assert_eq!(
            task["status"]["message"]["parts"][0]["text"],
            polyc_proto::admission_shed_text()
        );
    }

    #[tokio::test]
    async fn get_task_round_trips_then_404s() {
        let state = state(TurnOutcome::Completed {
            text: "hi".to_owned(),
        });
        let send = handle(&state, &send_body("q")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();

        let get_body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "GetTask", "params": { "id": task_id }
        }))
        .unwrap();
        let got = handle(&state, &get_body).await;
        assert_eq!(got["result"]["status"]["state"], "TASK_STATE_COMPLETED");

        let missing = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 3, "method": "GetTask", "params": { "id": "nope" }
        }))
        .unwrap();
        assert_eq!(
            handle(&state, &missing).await["error"]["code"],
            TASK_NOT_FOUND
        );
    }

    #[tokio::test]
    async fn cancel_terminal_task_is_not_cancelable() {
        let state = state(TurnOutcome::Completed {
            text: "hi".to_owned(),
        });
        let send = handle(&state, &send_body("q")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();
        let cancel = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 4, "method": "CancelTask", "params": { "id": task_id }
        }))
        .unwrap();
        assert_eq!(
            handle(&state, &cancel).await["error"]["code"],
            TASK_NOT_CANCELABLE
        );
    }

    #[tokio::test]
    async fn cancel_non_terminal_task_succeeds() {
        // input-required is non-terminal → cancelable; the flip is atomic.
        let state = state(TurnOutcome::InputRequired {
            turn_id: TEST_TURN.to_owned(),
            request_id: "r".to_owned(),
            tool_name: "t".to_owned(),
            prompt: "approve?".to_owned(),
            resolve_token: "resolve-token-1".to_owned(),
        });
        let send = handle(&state, &send_body("q")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();
        let cancel = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 8, "method": "CancelTask", "params": { "id": task_id }
        }))
        .unwrap();
        let resp = handle(&state, &cancel).await;
        assert_eq!(resp["result"]["status"]["state"], "TASK_STATE_CANCELED");
    }

    #[tokio::test]
    async fn missing_method_is_invalid_request() {
        let state = state(TurnOutcome::Completed {
            text: "x".to_owned(),
        });
        let body = serde_json::to_vec(&json!({ "jsonrpc": "2.0", "id": 1 })).unwrap();
        assert_eq!(
            handle(&state, &body).await["error"]["code"],
            INVALID_REQUEST
        );
    }

    #[tokio::test]
    async fn input_required_maps_to_input_required_state() {
        let state = state(TurnOutcome::InputRequired {
            turn_id: TEST_TURN.to_owned(),
            request_id: "r1".to_owned(),
            tool_name: "send_email".to_owned(),
            prompt: "approve send_email?".to_owned(),
            resolve_token: "resolve-token-1".to_owned(),
        });
        let resp = handle(&state, &send_body("q")).await;
        assert_eq!(
            resp["result"]["task"]["status"]["state"],
            "TASK_STATE_INPUT_REQUIRED"
        );
        assert_eq!(
            resp["result"]["task"]["status"]["message"]["parts"][0]["text"],
            "approve send_email?"
        );
    }

    /// `#792`: a follow-up `SendMessage` naming an `input-required` task's id
    /// with an "approve" reply must submit the decision via `ApprovalDialer`
    /// and re-drive the SAME conversation to completion — this is the
    /// acceptance gate's "resolving it resumes the turn to completion".
    #[tokio::test]
    async fn approve_reply_resolves_input_required_task_to_completion() {
        let (state, approvals) = state_with_sequence(
            vec![
                TurnOutcome::InputRequired {
                    turn_id: TEST_TURN.to_owned(),
                    request_id: "r1".to_owned(),
                    tool_name: "send_email".to_owned(),
                    prompt: "approve send_email?".to_owned(),
                    resolve_token: "resolve-token-1".to_owned(),
                },
                TurnOutcome::Completed {
                    text: "sent!".to_owned(),
                },
            ],
            true,
        );
        let send = handle(&state, &send_body("send an email")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();
        assert_eq!(
            send["result"]["task"]["status"]["state"],
            "TASK_STATE_INPUT_REQUIRED"
        );

        let approve_body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "SendMessage",
            "params": { "message": {
                "messageId": "m2", "role": "ROLE_USER",
                "parts": [{ "text": "approve" }],
                "contextId": "ctx-1", "taskId": task_id.clone(),
            }}
        }))
        .unwrap();
        let resp = handle(&state, &approve_body).await;

        assert_eq!(
            resp["result"]["task"]["status"]["state"], "TASK_STATE_COMPLETED",
            "the approve reply must resume the turn to a completed task, not \
             leave it (or a new task) hanging: {resp}"
        );
        assert_eq!(
            resp["result"]["task"]["status"]["message"]["parts"][0]["text"],
            "sent!"
        );
        assert_eq!(resp["result"]["task"]["id"], task_id, "same task, resumed");

        let calls = approvals.calls.lock().unwrap();
        assert_eq!(calls.len(), 1);
        assert_eq!(
            calls[0].0,
            format!("{TEST_TURN}:r1"),
            "the stashed occurrence was submitted"
        );
        assert!(calls[0].1, "approve maps to an approved decision");
        assert_eq!(
            calls[0].2,
            super::peer_conversation_id("ctx-1"),
            "scoped to the conversation this peer context resolves to"
        );
        assert!(
            calls[0].2.starts_with("a2a:") && calls[0].2.len() < 64,
            "and that conversation id is namespaced and bounded: {}",
            calls[0].2
        );
    }

    /// The deny path mirrors approve: submits `approved: false` and still
    /// resumes (the harness synthesizes a denial result rather than hanging).
    #[tokio::test]
    async fn deny_reply_submits_denial_and_resumes() {
        let (state, approvals) = state_with_sequence(
            vec![
                TurnOutcome::InputRequired {
                    turn_id: TEST_TURN.to_owned(),
                    request_id: "r1".to_owned(),
                    tool_name: "send_email".to_owned(),
                    prompt: "approve send_email?".to_owned(),
                    resolve_token: "resolve-token-1".to_owned(),
                },
                TurnOutcome::Completed {
                    text: "ok, not sent.".to_owned(),
                },
            ],
            true,
        );
        let send = handle(&state, &send_body("send an email")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();

        let deny_body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "SendMessage",
            "params": { "message": {
                "messageId": "m2", "role": "ROLE_USER",
                "parts": [{ "text": "deny" }],
                "contextId": "ctx-1", "taskId": task_id.clone(),
            }}
        }))
        .unwrap();
        let resp = handle(&state, &deny_body).await;

        assert_eq!(
            resp["result"]["task"]["status"]["state"],
            "TASK_STATE_COMPLETED"
        );
        assert!(
            !approvals.calls.lock().unwrap()[0].1,
            "deny maps to approved: false"
        );
    }

    /// An unparseable reply to an `input-required` task must re-prompt, not
    /// guess a decision or silently start a fresh turn.
    #[tokio::test]
    async fn unparseable_reply_reprompts_without_submitting() {
        let (state, approvals) = state_with_sequence(
            vec![TurnOutcome::InputRequired {
                turn_id: TEST_TURN.to_owned(),
                request_id: "r1".to_owned(),
                tool_name: "send_email".to_owned(),
                prompt: "approve send_email?".to_owned(),
                resolve_token: "resolve-token-1".to_owned(),
            }],
            true,
        );
        let send = handle(&state, &send_body("send an email")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();

        let ambiguous_body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "SendMessage",
            "params": { "message": {
                "messageId": "m2", "role": "ROLE_USER",
                "parts": [{ "text": "what does this do?" }],
                "contextId": "ctx-1", "taskId": task_id,
            }}
        }))
        .unwrap();
        let resp = handle(&state, &ambiguous_body).await;

        assert_eq!(
            resp["result"]["task"]["status"]["state"], "TASK_STATE_INPUT_REQUIRED",
            "an unparseable reply must not resolve the gate"
        );
        assert!(
            approvals.calls.lock().unwrap().is_empty(),
            "nothing should be submitted for an unparseable reply"
        );
        let read = handle(
            &state,
            &serde_json::to_vec(&json!({
                "jsonrpc": "2.0", "id": 3, "method": "GetTask",
                "params": { "id": task_id }
            }))
            .unwrap(),
        )
        .await;
        assert_eq!(
            read["result"]["status"]["state"], resp["result"]["task"]["status"]["state"],
            "the continuation reply must name the durable task state"
        );
        assert_eq!(
            read["result"]["status"]["message"]["messageId"],
            resp["result"]["task"]["status"]["message"]["messageId"],
            "the redacted durable status must be the same persisted message"
        );
    }

    #[tokio::test]
    async fn list_tasks_filters_by_context() {
        let state = state(TurnOutcome::Completed {
            text: "hi".to_owned(),
        });
        handle(&state, &send_body("q")).await;
        let list = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 5, "method": "ListTasks", "params": { "contextId": "ctx-1" }
        }))
        .unwrap();
        let resp = handle(&state, &list).await;
        assert_eq!(resp["result"]["tasks"].as_array().unwrap().len(), 1);
        assert_eq!(resp["result"]["tasks"][0]["contextId"], "ctx-1");
    }

    #[tokio::test]
    async fn slash_method_is_method_not_found() {
        // The v0.x `message/send` name must be rejected — proof we moved to v1.0.
        let state = state(TurnOutcome::Completed {
            text: "x".to_owned(),
        });
        let body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 6, "method": "message/send", "params": {}
        }))
        .unwrap();
        assert_eq!(
            handle(&state, &body).await["error"]["code"],
            METHOD_NOT_FOUND
        );
    }

    // `SendStreamingMessage`/`TaskSubscription` (`#371`).

    #[test]
    fn is_streaming_method_detects_only_the_two_streaming_methods() {
        let body = |method: &str| serde_json::to_vec(&json!({ "method": method })).unwrap();
        assert!(is_streaming_method(&body("SendStreamingMessage")));
        assert!(is_streaming_method(&body("TaskSubscription")));
        assert!(!is_streaming_method(&body("SendMessage")));
        assert!(!is_streaming_method(&body("GetTask")));
        assert!(!is_streaming_method(b"not json"));
    }

    /// A completed turn: `submitted` snapshot, each `TextDelta` as an ordered
    /// `artifactUpdate` (the first chunk omits `append`, later ones set it),
    /// then one terminal `statusUpdate` with `final: true` — and nothing
    /// after it.
    #[tokio::test]
    async fn completed_turn_streams_submitted_artifacts_then_final_status_in_order() {
        let state = state_with_dyn_runner(Arc::new(StreamingStubRunner(vec![
            TurnStreamEvent::TextDelta("Hello, ".to_owned()),
            TurnStreamEvent::TextDelta("world.".to_owned()),
            TurnStreamEvent::Outcome(TurnOutcome::Completed {
                text: "Hello, world.".to_owned(),
            }),
        ])));
        let events = stream_events(send_message_streaming(
            state,
            send_message_request("hi", "ctx-1"),
            String::new(),
        ))
        .await;

        assert_eq!(
            events.len(),
            4,
            "opening snapshot + 2 chunks + 1 final: {events:?}"
        );
        assert!(
            matches!(&events[0], StreamResponse::Task(t) if t.status.state == TaskState::Working),
            "the first event is the snapshot of the record this send just claimed, and a claimed \
             record is working: {:?}",
            events[0]
        );
        match &events[1] {
            StreamResponse::ArtifactUpdate(update) => {
                assert_eq!(update.artifact.parts[0].as_text(), Some("Hello, "));
                assert!(update.append.is_none(), "the first chunk omits `append`");
            }
            other => panic!("expected the first ArtifactUpdate, got {other:?}"),
        }
        match &events[2] {
            StreamResponse::ArtifactUpdate(update) => {
                assert_eq!(update.artifact.parts[0].as_text(), Some("world."));
                assert_eq!(
                    update.append,
                    Some(true),
                    "a later chunk sets `append: true`"
                );
            }
            other => panic!("expected the second ArtifactUpdate, got {other:?}"),
        }
        match &events[3] {
            StreamResponse::StatusUpdate(update) => {
                assert_eq!(update.status.state, TaskState::Completed);
                assert!(update.is_final, "the terminal event must set `final: true`");
            }
            other => panic!("expected the terminal StatusUpdate, got {other:?}"),
        }
    }

    /// A turn that pauses on an approval gate still closes the stream — with
    /// `final: true` on the `input-required` status, since the peer must send
    /// a fresh `SendStreamingMessage`/`SendMessage` to continue.
    #[tokio::test]
    async fn input_required_turn_ends_the_stream_with_final_true() {
        let state = state_with_dyn_runner(Arc::new(StreamingStubRunner(vec![
            TurnStreamEvent::TextDelta("thinking about it".to_owned()),
            TurnStreamEvent::Outcome(TurnOutcome::InputRequired {
                turn_id: TEST_TURN.to_owned(),
                request_id: "r1".to_owned(),
                tool_name: "send_email".to_owned(),
                prompt: "approve send_email?".to_owned(),
                resolve_token: "resolve-token-1".to_owned(),
            }),
        ])));
        let events = stream_events(send_message_streaming(
            state,
            send_message_request("hi", "ctx-1"),
            String::new(),
        ))
        .await;

        match events.last().expect("at least one event") {
            StreamResponse::StatusUpdate(update) => {
                assert_eq!(update.status.state, TaskState::InputRequired);
                assert!(update.is_final);
                assert_eq!(
                    update
                        .status
                        .message
                        .as_ref()
                        .map(crate::types::Message::text),
                    Some("approve send_email?".to_owned())
                );
            }
            other => panic!("expected the terminal StatusUpdate, got {other:?}"),
        }
    }

    /// `#792` over the streaming transport: an "approve" reply naming an
    /// `input-required` task's id resumes the SAME task to completion,
    /// mirroring `approve_reply_resolves_input_required_task_to_completion`
    /// for the unary path.
    #[tokio::test]
    async fn streaming_approve_reply_resolves_input_required_task_to_completion() {
        let (state, approvals) = state_with_sequence(
            vec![
                TurnOutcome::InputRequired {
                    turn_id: TEST_TURN.to_owned(),
                    request_id: "r1".to_owned(),
                    tool_name: "send_email".to_owned(),
                    prompt: "approve send_email?".to_owned(),
                    resolve_token: "resolve-token-1".to_owned(),
                },
                TurnOutcome::Completed {
                    text: "sent!".to_owned(),
                },
            ],
            true,
        );

        let first = stream_events(send_message_streaming(
            state.clone(),
            send_message_request("send an email", "ctx-1"),
            String::new(),
        ))
        .await;
        let task_id = match &first[0] {
            StreamResponse::Task(task) => task.id.clone(),
            other => panic!("expected the submitted snapshot, got {other:?}"),
        };

        let approve = SendMessageRequest {
            message: Message {
                message_id: "m2".to_owned(),
                context_id: Some("ctx-1".to_owned()),
                task_id: Some(task_id),
                role: Role::User,
                parts: vec![Part::text("approve")],
                metadata: None,
            },
        };
        let resumed = stream_events(send_message_streaming(state, approve, String::new())).await;
        match resumed.last().expect("at least one event") {
            StreamResponse::StatusUpdate(update) => {
                assert_eq!(
                    update.status.state,
                    TaskState::Completed,
                    "the approve reply must resume the SAME task to completion"
                );
                assert!(update.is_final);
            }
            other => panic!("expected the terminal StatusUpdate, got {other:?}"),
        }
        // Resuming an `input-required` continuation never re-emits a
        // `submitted` snapshot — it's a continuation, not a fresh task.
        assert!(
            !resumed
                .iter()
                .any(|event| matches!(event, StreamResponse::Task(_))),
            "a continuation must not re-submit: {resumed:?}"
        );

        let calls = approvals.calls.lock().unwrap();
        assert_eq!(calls.len(), 1);
        assert_eq!(
            calls[0].0,
            format!("{TEST_TURN}:r1"),
            "the stashed occurrence was submitted"
        );
    }

    #[tokio::test]
    async fn streaming_unparseable_continuation_persists_its_reprompt() {
        let (state, approvals) = state_with_sequence(
            vec![TurnOutcome::InputRequired {
                turn_id: TEST_TURN.to_owned(),
                request_id: "r1".to_owned(),
                tool_name: "send_email".to_owned(),
                prompt: "approve send_email?".to_owned(),
                resolve_token: "resolve-token-1".to_owned(),
            }],
            true,
        );
        let first = handle(&state, &send_body("send an email")).await;
        let task_id = first["result"]["task"]["id"]
            .as_str()
            .expect("task id")
            .to_owned();
        let continuation = SendMessageRequest {
            message: Message {
                message_id: "m2".to_owned(),
                context_id: Some("ctx-1".to_owned()),
                task_id: Some(task_id.clone()),
                role: Role::User,
                parts: vec![Part::text("what does this do?")],
                metadata: None,
            },
        };
        let events = stream_events(send_message_streaming(
            state.clone(),
            continuation,
            String::new(),
        ))
        .await;
        let streamed = match events.last().expect("terminal event") {
            StreamResponse::StatusUpdate(update) => &update.status,
            other => panic!("expected terminal status update, got {other:?}"),
        };
        let durable = state
            .store
            .get(&task_id)
            .await
            .expect("store read")
            .expect("task");
        assert_eq!(streamed.state, durable.status.state);
        assert_eq!(
            streamed.message.as_ref().map(|message| &message.message_id),
            durable
                .status
                .message
                .as_ref()
                .map(|message| &message.message_id),
            "the redacted durable status must be the streamed message"
        );
        assert!(approvals.calls.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn task_subscription_reports_a_stored_task_once_then_closes() {
        let state = state(TurnOutcome::Completed {
            text: "hi".to_owned(),
        });
        let send = handle(&state, &send_body("q")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();

        let events: Vec<Value> = task_subscription_stream(state, json!(1), task_id.clone())
            .collect()
            .await;
        assert_eq!(events.len(), 1, "one snapshot, then the stream closes");
        assert_eq!(events[0]["id"], 1);
        assert_eq!(events[0]["result"]["statusUpdate"]["taskId"], task_id);
        assert_eq!(events[0]["result"]["statusUpdate"]["final"], true);
        assert_eq!(
            events[0]["result"]["statusUpdate"]["status"]["state"],
            "TASK_STATE_COMPLETED"
        );
    }

    #[tokio::test]
    async fn task_subscription_of_an_unknown_task_errors() {
        let state = state(TurnOutcome::Completed {
            text: "hi".to_owned(),
        });
        let events: Vec<Value> = task_subscription_stream(state, json!(1), "nope".to_owned())
            .collect()
            .await;
        assert_eq!(events.len(), 1);
        assert_eq!(events[0]["error"]["code"], TASK_NOT_FOUND);
    }

    #[tokio::test]
    async fn handle_streaming_repeats_the_request_id_on_every_event() {
        let state = state_with_dyn_runner(Arc::new(StreamingStubRunner(vec![
            TurnStreamEvent::Outcome(TurnOutcome::Completed {
                text: "42".to_owned(),
            }),
        ])));
        let body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 7, "method": "SendStreamingMessage",
            "params": { "message": {
                "messageId": "m1", "role": "ROLE_USER",
                "parts": [{ "text": "q" }], "contextId": "ctx-1"
            }}
        }))
        .unwrap();
        let events: Vec<Value> = handle_streaming(state, &body)
            .filter(|event| futures::future::ready(!is_durable_marker(event)))
            .collect()
            .await;
        assert!(
            events.len() >= 2,
            "at least a submitted snapshot + final status: {events:?}"
        );
        for event in &events {
            assert_eq!(event["jsonrpc"], "2.0");
            assert_eq!(event["id"], 7);
        }
        assert_eq!(
            events.last().unwrap()["result"]["statusUpdate"]["final"],
            true
        );
    }

    // The durable store's own failures, as this surface answers them.

    /// `#1565` chunk C4.2: a store this edge cannot reach is NOT a missing
    /// task. A peer told `task not found` stops asking about work that is
    /// still recorded, so every read fails closed with an internal error
    /// carrying the reason instead.
    #[tokio::test]
    async fn get_task_on_a_store_outage_errors_rather_than_reporting_it_missing() {
        let state = state_with_store(
            TurnOutcome::Completed {
                text: "hi".to_owned(),
            },
            unreachable_store(),
        );
        let body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 1, "method": "GetTask", "params": { "id": "t1" }
        }))
        .unwrap();
        let resp = handle(&state, &body).await;
        assert_eq!(
            resp["error"]["code"], INTERNAL_ERROR,
            "an unreachable store must never read as a missing task: {resp}"
        );
        assert!(
            resp["error"]["message"]
                .as_str()
                .unwrap()
                .contains("unreachable"),
            "the refusal must say what happened: {resp}"
        );
    }

    /// The same for `CancelTask`: an outage is not "already finished".
    #[tokio::test]
    async fn cancel_task_on_a_store_outage_errors() {
        let state = state_with_store(
            TurnOutcome::Completed {
                text: "hi".to_owned(),
            },
            unreachable_store(),
        );
        let body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "CancelTask", "params": { "id": "t1" }
        }))
        .unwrap();
        let resp = handle(&state, &body).await;
        assert_eq!(resp["error"]["code"], INTERNAL_ERROR);
    }

    /// And for `ListTasks`: an outage is not an empty page.
    #[tokio::test]
    async fn list_tasks_on_a_store_outage_errors_rather_than_reporting_an_empty_page() {
        let state = state_with_store(
            TurnOutcome::Completed {
                text: "hi".to_owned(),
            },
            unreachable_store(),
        );
        let body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 3, "method": "ListTasks", "params": { "contextId": "ctx-1" }
        }))
        .unwrap();
        let resp = handle(&state, &body).await;
        assert_eq!(resp["error"]["code"], INTERNAL_ERROR);
        assert!(
            resp["result"].is_null(),
            "an outage must not answer with a page: {resp}"
        );
    }

    /// `TaskSubscription` reads the same store, so it fails closed the same way.
    #[tokio::test]
    async fn task_subscription_on_a_store_outage_errors() {
        let state = state_with_store(
            TurnOutcome::Completed {
                text: "hi".to_owned(),
            },
            unreachable_store(),
        );
        let events: Vec<Value> = task_subscription_stream(state, json!(1), "t1".to_owned())
            .collect()
            .await;
        assert_eq!(events.len(), 1);
        assert_eq!(events[0]["error"]["code"], INTERNAL_ERROR);
    }

    /// A turn that cannot be recorded never runs: the record is minted before
    /// the dial, so a store that refuses it fails the send before anything
    /// with side effects happens.
    ///
    /// The answer is a JSON-RPC error, never a `Task`. A synthetic task
    /// wearing the real id would report `failed` for a record this side never
    /// wrote and cannot read — and, when the create actually committed and
    /// only its reply was lost, would contradict the record outright.
    #[tokio::test]
    async fn send_message_refuses_before_dialing_when_the_record_cannot_be_minted() {
        let state = AppState {
            card: Arc::new(json!({})),
            runner: Arc::new(PanickingRunner),
            approvals: Arc::new(StubApprovalResponder::new(true)),
            store: unreachable_store(),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(64),
            peers: crate::server::PeerAuthenticator::single("test-peer", "unused-in-rpc-tests")
                .unwrap(),
        };
        let resp = handle(&state, &send_body("q")).await;
        assert!(
            resp["result"].is_null(),
            "a task that never started must not come back as a task: {resp}"
        );
        assert_eq!(resp["error"]["code"], INTERNAL_ERROR);
        let message = resp["error"]["message"].as_str().unwrap();
        assert!(
            message.contains("could not be reached"),
            "the failure must say the record could not be saved: {resp}"
        );
        assert!(
            message.contains("taskId"),
            "an ambiguous outcome must name the way out — retry under the same \
             task id: {resp}"
        );
    }

    /// The ambiguous outcome has a way out. A create whose reply was lost
    /// leaves the record durably `submitted` and nothing driving it; sending
    /// the same message again under that task id drives it, rather than
    /// finding `AlreadyExists` forever.
    #[tokio::test]
    async fn a_recorded_but_unstarted_task_is_driven_by_the_next_send() {
        let state = state(TurnOutcome::Completed {
            text: "finished on the retry".to_owned(),
        });
        // Exactly what a create that committed while its reply was lost
        // leaves behind: a `submitted` record with its opening history.
        let stranded = Task {
            id: "t-stranded".to_owned(),
            context_id: "ctx-1".to_owned(),
            status: TaskStatus {
                state: TaskState::Submitted,
                message: None,
                timestamp: None,
            },
            artifacts: None,
            history: Some(vec![Message {
                message_id: "m1".to_owned(),
                context_id: Some("ctx-1".to_owned()),
                task_id: Some("t-stranded".to_owned()),
                role: Role::User,
                parts: vec![Part::text("q")],
                metadata: None,
            }]),
            metadata: None,
        };
        state.store.create(&stranded).await.expect("records");

        let retry = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "SendMessage",
            "params": { "message": {
                "messageId": "m1", "role": "ROLE_USER",
                "parts": [{ "text": "q" }],
                "contextId": "ctx-1", "taskId": "t-stranded",
            }}
        }))
        .unwrap();
        let resp = handle(&state, &retry).await;
        assert_eq!(
            resp["result"]["task"]["status"]["state"], "TASK_STATE_COMPLETED",
            "a recorded-but-unstarted task must be driven, not refused: {resp}"
        );
        assert_eq!(resp["result"]["task"]["id"], "t-stranded", "same task");

        let get = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 3, "method": "GetTask", "params": { "id": "t-stranded" }
        }))
        .unwrap();
        assert_eq!(
            handle(&state, &get).await["result"]["status"]["state"],
            "TASK_STATE_COMPLETED",
            "and the record agrees with what the peer was told"
        );
    }

    /// A `SendMessage` naming `task_id`, as a peer that supplies its own id
    /// sends it.
    fn send_body_for_task(rpc_id: i64, task_id: &str, text: &str) -> Vec<u8> {
        serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": rpc_id, "method": "SendMessage",
            "params": { "message": {
                "messageId": format!("m{rpc_id}"), "role": "ROLE_USER",
                "parts": [{ "text": text }],
                "contextId": "ctx-1", "taskId": task_id,
            }}
        }))
        .unwrap()
    }

    /// Await a send that must be refused without dialing, failing rather than
    /// hanging if it dials instead.
    ///
    /// A send that wrongly dispatches a second turn blocks on the gate the
    /// first turn holds and never answers. Bounding the wait is what turns
    /// that into a test failure a reader can act on.
    async fn refused_promptly<T>(send: impl Future<Output = T>) -> T {
        tokio::time::timeout(std::time::Duration::from_secs(5), send)
            .await
            .expect("a send landing on a running task is refused, never dispatched")
    }

    /// [`AppState`] over a [`GatedRunner`] and one shared in-memory store, so
    /// two concurrent sends see the same records.
    fn gated_state(runner: Arc<GatedRunner>) -> AppState {
        AppState {
            card: Arc::new(json!({})),
            runner,
            approvals: Arc::new(StubApprovalResponder::new(true)),
            store: Arc::new(InMemoryTaskStore::new()),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(64),
            peers: crate::server::PeerAuthenticator::single("test-peer", "unused-in-rpc-tests")
                .unwrap(),
        }
    }

    /// The retry a peer-chosen task id exists to make safe. A client that
    /// times out at 30 s and resends the same `taskId` while the first turn is
    /// still running must not get a SECOND turn: the side effects would run
    /// twice, and the record would converge on one outcome without ever
    /// showing that two turns ran.
    ///
    /// The assertion is on the DISPATCH COUNT, not on the reported state.
    /// Reporting one task while dispatching two is exactly the defect, and a
    /// state assertion passes right through it.
    #[tokio::test]
    async fn a_retry_arriving_mid_turn_dispatches_exactly_once() {
        let (runner, dispatches, started, release) = GatedRunner::build(TurnOutcome::Completed {
            text: "the first turn finished".to_owned(),
        });
        let state = gated_state(runner);

        let first_state = state.clone();
        let first = tokio::spawn(async move {
            handle(&first_state, &send_body_for_task(1, "job-42", "q")).await
        });
        // The turn is genuinely in flight from here until `release`.
        started.notified().await;
        assert_eq!(
            dispatches.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the first send dispatches its turn"
        );

        // Bounded, because the failure this guards against does not return an
        // answer: a send that dispatches a second turn parks on the same gate
        // the first one is holding. A regression has to read as a failure, not
        // as a run that never ends.
        let retry = refused_promptly(handle(&state, &send_body_for_task(2, "job-42", "q"))).await;
        // Asserted first and on its own terms: the count is the property, and
        // what the retry was told is only how the property is reported.
        assert_eq!(
            dispatches.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the retry must not dispatch a second turn"
        );
        assert!(
            retry["result"].is_null(),
            "a retry landing on a running task must not be answered with a task: {retry}"
        );
        assert_eq!(retry["error"]["code"], INVALID_PARAMS);
        let message = retry["error"]["message"].as_str().unwrap();
        assert!(
            message.contains("still running"),
            "the refusal must say the task is running: {retry}"
        );
        assert!(
            message.contains("new taskId"),
            "and must say what to do when the run that owns it ended: {retry}"
        );

        release.notify_waiters();
        let first = first.await.expect("the first send completes");
        assert_eq!(
            first["result"]["task"]["status"]["state"], "TASK_STATE_COMPLETED",
            "and the send that owns the task still reports its own outcome: {first}"
        );
        assert_eq!(
            dispatches.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "one send, one turn"
        );
    }

    /// A completed task's stream replays it, on the streaming transport too.
    ///
    /// `POLY-154` changed both redelivery paths, and only the unary one had
    /// coverage: forcing `streaming_redelivery` to return `None` left the
    /// whole suite green. This holds the streaming half.
    #[tokio::test]
    async fn a_streaming_redelivery_replays_the_recorded_task() {
        let state = state_with_dyn_runner(Arc::new(ReceiptCheckingRunner::new(
            TurnOutcome::Completed {
                text: "done".to_owned(),
            },
        )));

        let first: Vec<StreamItem> = send_message_streaming(
            state.clone(),
            send_streaming_request(1, "job-77", "q"),
            String::new(),
        )
        .collect()
        .await;
        let recorded = first
            .iter()
            .find_map(|item| match item {
                StreamItem::Event(StreamResponse::Task(task)) => Some(task.id.clone()),
                _ => None,
            })
            .expect("the first send records a task");

        // The SAME message id — `send_streaming_request` derives it from the
        // sequence number, so a different one is a different message and never
        // reaches the redelivery path at all. The authority admits this as a
        // replay, and the recorded task comes back rather than a second turn.
        let retry: Vec<StreamItem> = send_message_streaming(
            state,
            send_streaming_request(1, "job-77", "q"),
            String::new(),
        )
        .collect()
        .await;
        let replayed = retry
            .iter()
            .find_map(|item| match item {
                StreamItem::Event(StreamResponse::Task(task)) => Some(task.id.clone()),
                _ => None,
            })
            .expect("a redelivery must replay the recorded task");
        assert_eq!(
            replayed, recorded,
            "the redelivery must not mint a new task"
        );
        let refusals = retry
            .iter()
            .filter(|item| matches!(item, StreamItem::Failure { .. }))
            .count();
        assert_eq!(refusals, 0, "a matching redelivery is not a refusal");
    }

    /// The same hole on the streaming transport, refused the same way: the
    /// retry's stream carries no events and ends as a JSON-RPC error.
    #[tokio::test]
    async fn a_streaming_retry_arriving_mid_turn_dispatches_exactly_once() {
        let (runner, dispatches, started, release) = GatedRunner::build(TurnOutcome::Completed {
            text: "the first turn finished".to_owned(),
        });
        let state = gated_state(runner);

        let first_state = state.clone();
        let first = tokio::spawn(async move {
            send_message_streaming(
                first_state,
                send_streaming_request(1, "job-42", "q"),
                String::new(),
            )
            .collect::<Vec<_>>()
            .await
        });
        started.notified().await;
        assert_eq!(dispatches.load(std::sync::atomic::Ordering::SeqCst), 1);

        let retry: Vec<StreamItem> = refused_promptly(
            send_message_streaming(
                state.clone(),
                send_streaming_request(2, "job-42", "q"),
                String::new(),
            )
            .collect(),
        )
        .await;
        assert_eq!(
            dispatches.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the retry must not dispatch a second turn"
        );
        assert_eq!(retry.len(), 2, "receipt precedes the refused task lookup");
        assert!(matches!(retry[0], StreamItem::DurablyReceived));
        match &retry[1] {
            StreamItem::Failure { code, message } => {
                assert_eq!(*code, INVALID_PARAMS);
                assert!(
                    message.contains("still running"),
                    "the refusal must say the task is running: {message}"
                );
            }
            StreamItem::Event(event) => {
                panic!("a retry landing on a running task must not emit an event: {event:?}")
            }
            StreamItem::DurablyReceived => panic!("the receipt must be emitted exactly once"),
        }

        release.notify_waiters();
        let first = first.await.expect("the first stream completes");
        assert_eq!(
            dispatches.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "one send, one turn"
        );
        assert!(
            matches!(
                first.iter().find(|item| !matches!(item, StreamItem::DurablyReceived)),
                Some(StreamItem::Event(StreamResponse::Task(task)))
                    if task.status.state == TaskState::Working
            ),
            "the opening snapshot reports the record it just claimed"
        );
    }

    /// The orphan the claim leaves behind is still picked up. A record created
    /// while its reply was lost never reached a claim, so it is the ONE
    /// `submitted` record a later send may drive — and it dispatches once.
    #[tokio::test]
    async fn a_claim_that_never_happened_still_leaves_the_task_drivable() {
        let (runner, dispatches, _started, release) = GatedRunner::build(TurnOutcome::Completed {
            text: "finished on the retry".to_owned(),
        });
        let state = gated_state(runner);
        // Exactly what a create that committed while its reply was lost leaves
        // behind: `submitted`, with no claim on it.
        let stranded = Task {
            id: "t-stranded".to_owned(),
            context_id: "ctx-1".to_owned(),
            status: TaskStatus {
                state: TaskState::Submitted,
                message: None,
                timestamp: None,
            },
            artifacts: None,
            history: Some(vec![Message {
                message_id: "m0".to_owned(),
                context_id: Some("ctx-1".to_owned()),
                task_id: Some("t-stranded".to_owned()),
                role: Role::User,
                parts: vec![Part::text("q")],
                metadata: None,
            }]),
            metadata: None,
        };
        state.store.create(&stranded).await.expect("records");

        release.notify_waiters();
        let driven_state = state.clone();
        let driven = tokio::spawn(async move {
            handle(&driven_state, &send_body_for_task(2, "t-stranded", "q")).await
        });
        // The gate is released ahead of the send as well as after it, so the
        // turn cannot park on a notification that already fired.
        release.notify_waiters();
        tokio::task::yield_now().await;
        release.notify_waiters();
        let driven = driven.await.expect("the send completes");

        assert_eq!(
            driven["result"]["task"]["status"]["state"], "TASK_STATE_COMPLETED",
            "a task that was recorded and never claimed must be driven: {driven}"
        );
        assert_eq!(
            dispatches.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "and driven exactly once"
        );
        let stored = state.store.get("t-stranded").await.expect("reads").unwrap();
        assert!(
            stored
                .history
                .as_deref()
                .unwrap_or_default()
                .iter()
                .any(|frame| frame.message_id == "m2"),
            "the message that drove the turn is in the history: {stored:?}"
        );
    }

    /// Terminal is final. A `SendMessage` naming a task that already finished
    /// is refused instead of starting a fresh turn under the same id — the
    /// process-local store used to overwrite the finished record silently.
    ///
    /// The refusal is a JSON-RPC error naming the recorded state. Answering
    /// with a `Task` carrying the real id and a made-up `failed` would put the
    /// send and a `GetTask` on the same id in direct contradiction.
    #[tokio::test]
    async fn send_message_naming_a_finished_task_refuses_instead_of_clobbering_it() {
        let state = state(TurnOutcome::Completed {
            text: "first".to_owned(),
        });
        let send = handle(&state, &send_body("q")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();
        assert_eq!(
            send["result"]["task"]["status"]["state"],
            "TASK_STATE_COMPLETED"
        );

        let again = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "SendMessage",
            "params": { "message": {
                "messageId": "m2", "role": "ROLE_USER",
                "parts": [{ "text": "again" }],
                "contextId": "ctx-1", "taskId": task_id,
            }}
        }))
        .unwrap();
        let resp = handle(&state, &again).await;
        assert!(
            resp["result"].is_null(),
            "a finished task must not be answered with a made-up record: {resp}"
        );
        assert_eq!(resp["error"]["code"], INVALID_PARAMS);
        assert!(
            resp["error"]["message"]
                .as_str()
                .unwrap()
                .contains("finished"),
            "the refusal must say what the record actually is: {resp}"
        );

        // The finished record is untouched.
        let get = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 3, "method": "GetTask", "params": { "id": task_id }
        }))
        .unwrap();
        let got = handle(&state, &get).await;
        assert_eq!(got["result"]["status"]["state"], "TASK_STATE_COMPLETED");
    }

    /// A peer cancels while the turn is still running. `tasks/cancel` is an
    /// independent method and `submitted` is not terminal, so the cancel
    /// succeeds — and the finishing turn must then report what the record
    /// says, not the `completed` it happened to produce.
    #[tokio::test]
    async fn a_cancel_during_a_running_turn_is_reported_as_canceled() {
        let state = state(TurnOutcome::Completed {
            text: "the turn finished anyway".to_owned(),
        });
        let mut task = Task {
            id: "t-raced".to_owned(),
            context_id: "ctx-1".to_owned(),
            status: TaskStatus {
                state: TaskState::Submitted,
                message: None,
                timestamp: None,
            },
            artifacts: None,
            history: None,
            metadata: None,
        };
        state.store.create(&task).await.expect("records");
        state.store.cancel("t-raced").await.expect("cancels");

        // The turn now finishes and tries to record its own outcome.
        task.status = status_with_message(
            TaskState::Completed,
            agent_message("the turn finished anyway", "ctx-1", "t-raced"),
        );
        record_outcome(&state, &mut task, &[], None).await;

        assert_eq!(
            task.status.state,
            TaskState::Canceled,
            "the durable record is the truth a peer is told, not the turn's own outcome"
        );
        let stored = state.store.get("t-raced").await.expect("reads").unwrap();
        assert_eq!(
            stored.status.state, task.status.state,
            "and the answer matches what GetTask reports"
        );
    }

    /// The same race from the other two directions: a second `SendMessage` on
    /// one `input-required` task, and a continuation whose task finished on
    /// another connection. Both reach `record_outcome` against a record that
    /// already finished, and both must report the record.
    #[tokio::test]
    async fn a_turn_that_lost_the_race_reports_the_recorded_outcome() {
        let state = state(TurnOutcome::Completed {
            text: "second".to_owned(),
        });
        let mut task = Task {
            id: "t-raced".to_owned(),
            context_id: "ctx-1".to_owned(),
            status: TaskStatus {
                state: TaskState::Submitted,
                message: None,
                timestamp: None,
            },
            artifacts: None,
            history: None,
            metadata: None,
        };
        state.store.create(&task).await.expect("records");

        // The first caller finishes the task.
        let mut winner = task.clone();
        winner.status = status_with_message(
            TaskState::Completed,
            agent_message("first", "ctx-1", "t-raced"),
        );
        record_outcome(&state, &mut winner, &[], None).await;
        assert_eq!(winner.status.state, TaskState::Completed);

        // The second caller finishes a moment later.
        task.status = status_with_message(
            TaskState::Failed,
            agent_message("second", "ctx-1", "t-raced"),
        );
        record_outcome(&state, &mut task, &[], None).await;
        assert_eq!(
            task.status.state,
            TaskState::Completed,
            "the loser of the race reports the recorded outcome, never a fabricated failure"
        );
    }

    /// An already-settled approval does not re-drive a turn, but the newly
    /// admitted source message is still claimed and durably reflected.
    #[tokio::test]
    async fn an_already_answered_decision_records_input_without_redriving() {
        let (state, approvals) = state_with_sequence(
            vec![TurnOutcome::InputRequired {
                turn_id: TEST_TURN.to_owned(),
                request_id: "r1".to_owned(),
                tool_name: "send_email".to_owned(),
                prompt: "approve send_email?".to_owned(),
                resolve_token: "resolve-token-1".to_owned(),
            }],
            // The control plane reports the decision was already persisted.
            false,
        );
        let send = handle(&state, &send_body("send an email")).await;
        let task_id = send["result"]["task"]["id"].as_str().unwrap().to_owned();

        let approve = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 2, "method": "SendMessage",
            "params": { "message": {
                "messageId": "m2", "role": "ROLE_USER",
                "parts": [{ "text": "approve" }],
                "contextId": "ctx-1", "taskId": task_id.clone(),
            }}
        }))
        .unwrap();
        let resp = handle(&state, &approve).await;

        assert_eq!(
            resp["result"]["task"]["status"]["state"], "TASK_STATE_INPUT_REQUIRED",
            "an already-answered decision must leave the task as it stands: {resp}"
        );
        assert!(
            resp["result"]["task"]["status"]["message"]["parts"][0]["text"]
                .as_str()
                .unwrap()
                .contains("already recorded"),
            "the reply says the decision was already made — nothing was re-driven: {resp}"
        );
        assert_eq!(
            approvals.calls.lock().unwrap().len(),
            1,
            "the decision is submitted exactly once"
        );
        let durable = state
            .store
            .get(&task_id)
            .await
            .expect("store read")
            .expect("task");
        assert_eq!(durable.status.state, TaskState::InputRequired);
        assert_eq!(
            durable
                .status
                .message
                .as_ref()
                .map(|message| message.message_id.as_str()),
            resp["result"]["task"]["status"]["message"]["messageId"].as_str(),
            "the redacted durable status must be the claimed continuation message"
        );
    }

    /// Tasks are indexed one context at a time, so a `ListTasks` that names no
    /// context is refused rather than answered from a scan across all of them.
    #[tokio::test]
    async fn list_tasks_without_a_context_is_invalid_params() {
        let state = state(TurnOutcome::Completed {
            text: "hi".to_owned(),
        });
        let body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 9, "method": "ListTasks", "params": {}
        }))
        .unwrap();
        let resp = handle(&state, &body).await;
        assert_eq!(resp["error"]["code"], INVALID_PARAMS);
        assert!(
            resp["error"]["message"]
                .as_str()
                .unwrap()
                .contains("contextId"),
            "the refusal must say what to name: {resp}"
        );
    }

    /// The page token is a keyset cursor over the context's tasks, not an
    /// offset: the second page resumes after the last id of the first.
    #[tokio::test]
    async fn list_tasks_pages_by_keyset_over_one_context() {
        let state = state(TurnOutcome::Completed {
            text: "hi".to_owned(),
        });
        for seq in 0..3 {
            let body = serde_json::to_vec(&json!({
                "jsonrpc": "2.0", "id": seq, "method": "SendMessage", "params": {
                    "message": {
                        "role": "ROLE_USER", "messageId": format!("m{seq}"),
                        "contextId": "ctx-1", "parts": [{"text": "q"}]
                    }
                }
            }))
            .unwrap();
            handle(&state, &body).await;
        }
        let page = |token: Option<&str>| {
            let mut params = json!({ "contextId": "ctx-1", "pageSize": 2 });
            if let Some(token) = token {
                params["pageToken"] = Value::from(token.to_owned());
            }
            serde_json::to_vec(&json!({
                "jsonrpc": "2.0", "id": 5, "method": "ListTasks", "params": params
            }))
            .unwrap()
        };
        let first = handle(&state, &page(None)).await;
        assert_eq!(first["result"]["tasks"].as_array().unwrap().len(), 2);
        let token = first["result"]["nextPageToken"]
            .as_str()
            .unwrap()
            .to_owned();
        assert_eq!(
            token,
            first["result"]["tasks"][1]["id"].as_str().unwrap(),
            "the token is the last id of the page it closes"
        );

        let second = handle(&state, &page(Some(&token))).await;
        assert_eq!(second["result"]["tasks"].as_array().unwrap().len(), 1);
        assert!(second["result"]["nextPageToken"].is_null());
    }

    #[tokio::test]
    async fn handle_streaming_reports_method_not_found_for_an_unknown_method() {
        let state = state(TurnOutcome::Completed {
            text: "x".to_owned(),
        });
        let body = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": 1, "method": "Bogus", "params": {}
        }))
        .unwrap();
        let events: Vec<Value> = handle_streaming(state, &body).collect().await;
        assert_eq!(events.len(), 1);
        assert_eq!(events[0]["error"]["code"], METHOD_NOT_FOUND);
    }
}