a2a-protocol-server 0.14.1

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 Tom F. <tomf@tomtomtech.net> (https://github.com/tomtom215)

//! Tests for the message send path.
//!
//! Split from `mod.rs`, which was 2,395 lines of which 1,674 were these. The
//! production code they cover is ~720 lines, and it holds the hot path plus the
//! destroy/`CleanupGuard` coupling — the part most worth being able to read
//! without scrolling past a test suite three times its length.

use super::decisions::{
    evict_aged_token, second_send_blocked, shape_response_history, token_aged,
    token_still_evictable,
};
use super::*;
use a2a_protocol_types::events::StreamResponse;
use a2a_protocol_types::events::TaskStatusUpdateEvent;
use a2a_protocol_types::message::{Message, MessageId, MessageRole, Part};
use a2a_protocol_types::params::{MessageSendParams, SendMessageConfiguration};
use a2a_protocol_types::responses::SendMessageResponse;
use a2a_protocol_types::task::{ContextId, TaskId, TaskState, TaskStatus};

use crate::agent_executor;
use crate::builder::RequestHandlerBuilder;
use crate::error::ServerError;
use crate::handler::CancellationEntry;
use crate::streaming::EventQueueReader as _;

struct DummyExecutor;
agent_executor!(DummyExecutor, |_ctx, _queue| async { Ok(()) });

fn make_handler() -> RequestHandler {
    RequestHandlerBuilder::new(DummyExecutor)
        .build()
        .expect("default build should succeed")
}

fn make_params(context_id: Option<&str>) -> MessageSendParams {
    MessageSendParams {
        message: Message {
            id: MessageId::new("msg-1"),
            role: MessageRole::User,
            parts: vec![Part::text("hello")],
            context_id: context_id.map(ContextId::new),
            task_id: None,
            reference_task_ids: None,
            extensions: None,
            metadata: None,
        },
        configuration: None,
        metadata: None,
        tenant: None,
    }
}

// ── CleanupGuard ─────────────────────────────────────────────────────────

/// Pins `CleanupGuard::drop`, which is the *only* thing that releases a
/// task's event queue and cancellation token when the executor unwinds.
///
/// Nothing tested it, and the reason is structural rather than an
/// oversight. On the normal path the executor task cleans up explicitly and
/// then defuses the guard (`cleanup_guard.task_id = None`), so `drop`
/// becomes a no-op — every ordinary send, success or handled error, leaves
/// the mutation invisible. The guard earns its place only when the executor
/// panics and the task unwinds before reaching that cleanup, which is
/// exactly what this test arranges. Despite the comment above the executor
/// call, there is no `catch_unwind` here; the guard *is* the panic
/// handling.
///
/// The wait is a bounded poll rather than a sleep because `drop` spawns the
/// cleanup onto the runtime: the work is ordered after the unwind but not
/// synchronous with it, so a fixed sleep would either flake or be
/// needlessly slow.
#[tokio::test]
async fn cleanup_guard_releases_the_token_when_the_executor_panics() {
    struct PanicExec;
    impl crate::executor::AgentExecutor for PanicExec {
        fn execute<'a>(
            &'a self,
            _ctx: &'a crate::request_context::RequestContext,
            _queue: &'a dyn crate::streaming::EventQueueWriter,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>>
                    + Send
                    + 'a,
            >,
        > {
            Box::pin(async { panic!("executor panics before it can clean up") })
        }
    }

    let handler = RequestHandlerBuilder::new(PanicExec)
        .build()
        .expect("build should succeed");

    let _ = handler.on_send_message(make_params(None), true, None).await;

    // The executor task must unwind and its guard must run.
    let mut cleaned = false;
    for _ in 0..200 {
        if handler.cancellation_tokens.read().await.is_empty() {
            cleaned = true;
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    assert!(
        cleaned,
        "a panicking executor must still release its cancellation token; \
         CleanupGuard::drop is the only path that does so"
    );
    let _ = handler.shutdown().await;
}

// ── task history cap ─────────────────────────────────────────────────────

/// Pins the arithmetic that trims an over-long task history.
///
/// The obvious test does not work here. A single send onto a full history
/// gives `len == MAX + 1`, where `len - MAX` and `len / MAX` are both 1 —
/// the mutation is invisible at the boundary it looks like it should be
/// caught at. The two only diverge once the history is at least twice the
/// cap: at 2048, subtraction trims 1024 and leaves exactly the cap, while
/// division trims 2 and leaves 2046.
///
/// An over-long history is reachable in practice — the cap is applied on
/// write, so a task stored by an older build, a different cap, or a direct
/// store write arrives here oversized — which is why the trim is written to
/// bring any length back to the cap rather than to peel off one message.
#[tokio::test]
async fn oversized_stored_history_is_trimmed_back_to_the_cap() {
    let handler = make_handler();
    let task_id = TaskId::new("t-overlong");

    // One short of twice the cap; the incoming message makes it 2048.
    let seeded = MAX_TASK_HISTORY_MESSAGES * 2 - 1;
    let mut task = task_with_history(seeded);
    task.id = task_id.clone();
    handler
        .task_store
        .save(&task)
        .await
        .expect("seed the oversized task");

    let mut params = make_params(None);
    params.message.task_id = Some(task_id.clone());
    let _ = handler.on_send_message(params, false, None).await;

    let stored = handler
        .task_store
        .get(&task_id)
        .await
        .expect("load")
        .expect("task should still exist");
    assert_eq!(
        stored.history.map(|h| h.len()),
        Some(MAX_TASK_HISTORY_MESSAGES),
        "an oversized history must be trimmed back to exactly the cap"
    );
    let _ = handler.shutdown().await;
}

// ── cancellation-token sweep and context-lock pruning ────────────────────

/// Seeds the token map with `n` already-cancelled entries, which the sweep
/// treats as unconditionally evictable, and returns their ids.
async fn seed_cancelled_tokens(handler: &RequestHandler, n: usize) -> Vec<TaskId> {
    let mut ids = Vec::new();
    // Dropped explicitly below rather than at end of scope: holding the
    // write guard across the return is what `significant_drop_tightening`
    // flags, and that lint is deny-by-default here via `-D warnings`.
    let mut tokens = handler.cancellation_tokens.write().await;
    for i in 0..n {
        let id = TaskId::new(format!("stale-{i}"));
        let token = tokio_util::sync::CancellationToken::new();
        token.cancel();
        tokens.insert(
            id.clone(),
            CancellationEntry {
                turn: std::sync::Arc::default(),
                token,
                created_at: Instant::now(),
            },
        );
        ids.push(id);
    }
    drop(tokens);
    ids
}

/// Pins the two decisions that drive cancellation-token eviction: whether
/// the sweep runs at all, and whether phase 2 actually removes anything.
///
/// Both mutants survived the 2026-08-07 sweep despite
/// `stale_cancellation_tokens_cleaned_up` exercising this code, because
/// that test asserted nothing — it ran the sweep and then called
/// `shutdown()`. Driving code is not testing it.
///
/// Cancelled tokens are seeded directly rather than produced by a slow
/// executor: an *aged* token is only evicted once its event queue is gone,
/// so a still-running executor keeps its token alive and the map never
/// shrinks. Cancelled entries are unconditionally evictable, which makes
/// the outcome deterministic instead of a race with a sleep.
#[tokio::test]
async fn sweep_evicts_cancelled_tokens_once_the_map_is_at_capacity() {
    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(
            crate::handler::limits::HandlerLimits::default().with_max_cancellation_tokens(2),
        )
        .build()
        .expect("build should succeed");

    let stale = seed_cancelled_tokens(&handler, 2).await;
    assert_eq!(handler.cancellation_tokens.read().await.len(), 2);

    // len == max, so `len >= max` fires. `<` would skip the sweep, and
    // deleting the `!` on `stale_ids.is_empty()` would skip the removal.
    let _ = handler
        .on_send_message(make_params(None), false, None)
        .await;

    let tokens = handler.cancellation_tokens.read().await;
    for id in &stale {
        assert!(
            !tokens.contains_key(id),
            "cancelled token {id:?} should have been evicted by the sweep"
        );
    }
    drop(tokens);
    let _ = handler.shutdown().await;
}

/// Pins the context-lock pruning *threshold*.
///
/// A first version of this test used a limit of 2 and three sends, and the
/// `>=`-to-`<` mutant survived it: with the limit that low, both the
/// original and the mutant end holding exactly one entry, so the assertion
/// could not tell them apart. The threshold only becomes observable when
/// the map stays *below* it — the original never prunes, while `<` prunes
/// on every send and reclaims each previous context.
#[tokio::test]
async fn context_locks_are_not_pruned_below_the_limit() {
    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(
            crate::handler::limits::HandlerLimits::default().with_max_context_locks(5),
        )
        .build()
        .expect("build should succeed");

    for ctx in ["ctx-a", "ctx-b", "ctx-c"] {
        let _ = handler
            .on_send_message(make_params(Some(ctx)), false, None)
            .await;
    }

    assert_eq!(
        handler.context_locks.read().await.len(),
        3,
        "with a limit of 5, three contexts must all be retained"
    );
    let _ = handler.shutdown().await;
}

/// Pins the staleness predicate itself: pruning must reclaim unused locks
/// and spare one another task still holds.
///
/// `Arc::strong_count(v) > 1` means "someone besides the map owns this".
/// The entries are seeded directly so both populations exist at prune time
/// with certainty — a live lock (a clone held here, count 2) and stale ones
/// (count 1). Driving this through concurrent sends would depend on a
/// scheduler race for whether the live lock is still held when pruning runs.
///
/// This is what separates the three mutations of that predicate:
/// `< 1` is never true and would clear the map including the live lock,
/// `== 1` inverts it and would reclaim exactly the wrong entries, and
/// `>= 1` is always true and would reclaim nothing.
#[tokio::test]
async fn context_lock_pruning_spares_locks_still_in_use() {
    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(
            crate::handler::limits::HandlerLimits::default().with_max_context_locks(2),
        )
        .build()
        .expect("build should succeed");

    // Held for the duration of the test, so the map is not its only owner.
    let live = std::sync::Arc::new(tokio::sync::Mutex::new(()));
    {
        let mut locks = handler.context_locks.write().await;
        locks.insert("live-ctx".to_string(), std::sync::Arc::clone(&live));
        locks.insert(
            "stale-1".to_string(),
            std::sync::Arc::new(tokio::sync::Mutex::new(())),
        );
        locks.insert(
            "stale-2".to_string(),
            std::sync::Arc::new(tokio::sync::Mutex::new(())),
        );
    }

    // 3 entries against a limit of 2, so the next send prunes.
    let _ = handler
        .on_send_message(make_params(Some("ctx-new")), false, None)
        .await;

    let locks = handler.context_locks.read().await;
    assert!(
        locks.contains_key("live-ctx"),
        "a lock another owner still holds must survive pruning"
    );
    assert!(
        !locks.contains_key("stale-1") && !locks.contains_key("stale-2"),
        "locks owned only by the map must be reclaimed"
    );
    drop(locks);
    drop(live);
    let _ = handler.shutdown().await;
}

// ── metadata size limit ──────────────────────────────────────────────────

/// Builds a handler whose metadata budget is `max` bytes.
fn handler_with_metadata_limit(max: usize) -> RequestHandler {
    RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(
            crate::handler::limits::HandlerLimits::default().with_max_metadata_size(max),
        )
        .build()
        .expect("build with custom limits should succeed")
}

/// Metadata serialising to exactly `n` bytes.
///
/// It must be a JSON *object*: `validate_metadata_object` runs before the
/// size check and rejects scalars outright, so a bare string never reaches
/// the code under test here. `{"k":"<pad>"}` serialises to `pad.len() + 8`
/// bytes, and the assertion below keeps that arithmetic honest rather than
/// trusting the comment.
fn metadata_of_exactly(n: usize) -> serde_json::Value {
    let value = serde_json::json!({ "k": "x".repeat(n - 8) });
    assert_eq!(
        serde_json::to_vec(&value).expect("serialise").len(),
        n,
        "fixture must serialise to exactly {n} bytes"
    );
    value
}

/// Pins the `>` in both metadata size checks, which is a limit boundary and
/// therefore worth being exact about: `>=` would reject a payload of
/// precisely the configured maximum.
///
/// Two mutants survived here in the 2026-08-07 sweep, one per check. The
/// existing tests only prove that something far *over* the limit is
/// rejected, which `>=` also does — nothing exercised the boundary itself.
///
/// A small custom limit keeps this exact and cheap; the default is 1 MiB.
#[tokio::test]
async fn metadata_exactly_at_limit_is_accepted_but_one_byte_over_is_not() {
    const MAX: usize = 64;

    // ── message metadata ──
    let mut params = make_params(None);
    params.message.metadata = Some(metadata_of_exactly(MAX));
    assert!(
        !matches!(
            handler_with_metadata_limit(MAX)
                .on_send_message(params, false, None)
                .await,
            Err(ServerError::InvalidParams(_))
        ),
        "message metadata of exactly {MAX} bytes is within the limit"
    );

    let mut params = make_params(None);
    params.message.metadata = Some(metadata_of_exactly(MAX + 1));
    assert!(
        matches!(
            handler_with_metadata_limit(MAX)
                .on_send_message(params, false, None)
                .await,
            Err(ServerError::InvalidParams(_))
        ),
        "message metadata one byte over the limit must be rejected"
    );

    // ── request metadata (the second, separate check) ──
    let mut params = make_params(None);
    params.metadata = Some(metadata_of_exactly(MAX));
    assert!(
        !matches!(
            handler_with_metadata_limit(MAX)
                .on_send_message(params, false, None)
                .await,
            Err(ServerError::InvalidParams(_))
        ),
        "request metadata of exactly {MAX} bytes is within the limit"
    );

    let mut params = make_params(None);
    params.metadata = Some(metadata_of_exactly(MAX + 1));
    assert!(
        matches!(
            handler_with_metadata_limit(MAX)
                .on_send_message(params, false, None)
                .await,
            Err(ServerError::InvalidParams(_))
        ),
        "request metadata one byte over the limit must be rejected"
    );
}

// ── shape_response_history ───────────────────────────────────────────────

/// Builds a task carrying `len` history messages, oldest first, each
/// individually identifiable as `h0`, `h1`, …
fn task_with_history(len: usize) -> Task {
    Task {
        id: TaskId::new("t-hist"),
        context_id: ContextId::new("ctx-hist"),
        status: TaskStatus::new(TaskState::Submitted),
        artifacts: None,
        history: Some(
            (0..len)
                .map(|i| Message {
                    id: MessageId::new(format!("h{i}")),
                    role: MessageRole::User,
                    parts: vec![Part::text("x")],
                    context_id: None,
                    task_id: None,
                    reference_task_ids: None,
                    extensions: None,
                    metadata: None,
                })
                .collect(),
        ),
        metadata: None,
    }
}

/// Shapes a `len`-message history and returns the surviving message ids.
fn shaped_ids(len: usize, history_length: Option<u32>) -> Option<Vec<String>> {
    let mut task = task_with_history(len);
    shape_response_history(&mut task, history_length);
    task.history
        .map(|msgs| msgs.into_iter().map(|m| m.id.0).collect())
}

/// Pins every branch and boundary of `shape_response_history`.
///
/// Six mutants survived here in the 2026-08-07 sweep, because the tests
/// reached this function only through `on_send_message`, which never
/// varies `historyLength` — nothing observed the truncation at all.
///
/// The truncation arithmetic those cases were written against has since
/// moved to [`helpers::truncate_history`](super::super::helpers), where it
/// is unit-tested directly. What this test now guards is the part that
/// stayed behind, and that no mutation operator can reach: that
/// `shape_response_history` still *calls* it, and still maps the two
/// send-response-specific inputs correctly. `None` and `Some(0)` both
/// yield no history here, but they are not the same instruction — only
/// `Some(0)` is a client asking for zero messages, and a refactor that
/// collapsed them would be invisible to every other case below.
#[test]
fn shape_response_history_covers_every_branch() {
    // Default: history is omitted entirely, not echoed back.
    assert_eq!(shaped_ids(3, None), None);

    // Some(0) omits too — distinct from keeping an empty list.
    assert_eq!(shaped_ids(3, Some(0)), None);

    // Truncation keeps the n most recent, oldest dropped.
    assert_eq!(
        shaped_ids(6, Some(2)),
        Some(vec!["h4".to_string(), "h5".to_string()])
    );
    assert_eq!(shaped_ids(4, Some(1)), Some(vec!["h3".to_string()]));

    // Exactly at the boundary, and asking for more than exists: keep all.
    assert_eq!(
        shaped_ids(3, Some(3)),
        Some(vec!["h0".to_string(), "h1".to_string(), "h2".to_string()])
    );
    assert_eq!(
        shaped_ids(2, Some(5)),
        Some(vec!["h0".to_string(), "h1".to_string()])
    );

    // An empty history stays empty rather than becoming None.
    assert_eq!(shaped_ids(0, Some(3)), Some(vec![]));
}

#[tokio::test]
async fn empty_message_parts_returns_invalid_params() {
    let handler = make_handler();
    let mut params = make_params(None);
    params.message.parts = vec![];

    let result = handler.on_send_message(params, false, None).await;

    assert!(
        matches!(result, Err(ServerError::InvalidParams(_))),
        "expected InvalidParams for empty parts"
    );
}

#[tokio::test]
async fn oversized_message_metadata_returns_invalid_params() {
    let handler = make_handler();
    let mut params = make_params(None);
    // An *object* exceeding the default 1 MiB limit. This was a bare JSON
    // string until 2026-08-08, which `validate_metadata_object` rejects as
    // a non-object before the size check ever runs — so the test passed
    // without exercising the limit it names. Both size-check mutants
    // survived the 2026-08-07 sweep for exactly that reason.
    params.message.metadata = Some(serde_json::json!({ "k": "x".repeat(1_100_000) }));

    let result = handler.on_send_message(params, false, None).await;

    assert!(
        matches!(result, Err(ServerError::InvalidParams(_))),
        "expected InvalidParams for oversized message metadata"
    );
}

#[tokio::test]
async fn oversized_request_metadata_returns_invalid_params() {
    let handler = make_handler();
    let mut params = make_params(None);
    // Build a JSON string that exceeds the default 1 MiB limit.
    let big_value = "x".repeat(1_100_000);
    params.metadata = Some(serde_json::json!(big_value));

    let result = handler.on_send_message(params, false, None).await;

    assert!(
        matches!(result, Err(ServerError::InvalidParams(_))),
        "expected InvalidParams for oversized request metadata"
    );
}

#[tokio::test]
async fn non_object_message_metadata_returns_invalid_params() {
    // Cross-binding portability: array/scalar metadata is not representable
    // over gRPC (google.protobuf.Struct) and must be rejected at ingress.
    let handler = make_handler();
    let mut params = make_params(None);
    params.message.metadata = Some(serde_json::json!([1, 2, 3]));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::InvalidParams(ref msg))
            if msg.contains("JSON object") && msg.contains("array")),
        "expected InvalidParams naming the offending kind (array), got: {result:?}"
    );
}

#[tokio::test]
async fn scalar_request_metadata_returns_invalid_params() {
    let handler = make_handler();
    let mut params = make_params(None);
    params.metadata = Some(serde_json::json!("a bare string"));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::InvalidParams(ref msg))
            if msg.contains("JSON object") && msg.contains("string")),
        "expected InvalidParams naming the offending kind (string), got: {result:?}"
    );
}

#[tokio::test]
async fn non_object_part_metadata_returns_invalid_params() {
    let handler = make_handler();
    let mut params = make_params(None);
    params.message.parts[0].metadata = Some(serde_json::json!(42));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::InvalidParams(ref msg))
            if msg.contains("part 0") && msg.contains("number")),
        "expected InvalidParams naming the part index and kind (number), got: {result:?}"
    );
}

#[tokio::test]
async fn object_metadata_is_accepted() {
    // An object metadata value is representable across all bindings.
    let handler = make_handler();
    let mut params = make_params(None);
    params.message.metadata = Some(serde_json::json!({"k": "v"}));
    params.metadata = Some(serde_json::json!({"trace": 1}));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        result.is_ok(),
        "object metadata must be accepted, got: {result:?}"
    );
}

#[tokio::test]
async fn valid_message_returns_ok() {
    let handler = make_handler();
    let params = make_params(None);

    let result = handler.on_send_message(params, false, None).await;

    let send_result = result.expect("expected Ok for valid message");
    assert!(
        matches!(
            send_result,
            SendMessageResult::Response(SendMessageResponse::Task(_))
        ),
        "expected Response(Task) for non-streaming send"
    );
}

#[tokio::test]
async fn return_immediately_returns_task() {
    let handler = make_handler();
    let mut params = make_params(None);
    params.configuration = Some(SendMessageConfiguration {
        accepted_output_modes: vec!["text/plain".into()],
        task_push_notification_config: None,
        history_length: None,
        return_immediately: Some(true),
    });

    let result = handler.on_send_message(params, false, None).await;

    assert!(
        matches!(
            result,
            Ok(SendMessageResult::Response(SendMessageResponse::Task(_)))
        ),
        "expected Response(Task) for return_immediately=true"
    );
}

// An executor that narrates progress to completion via the event queue.
struct CompletingExecutor;
agent_executor!(CompletingExecutor, |ctx, queue| async {
    for state in [TaskState::Working, TaskState::Completed] {
        let ev = StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
            task_id: ctx.task_id.clone(),
            context_id: ContextId::new(ctx.context_id.clone()),
            status: TaskStatus::with_timestamp(state),
            metadata: None,
        });
        let _ = queue.write(ev).await;
    }
    Ok(())
});

// An executor that never finishes, keeping its task in flight (and its
// event queue alive) for the duration of a test.
struct BlockingExecutor;
agent_executor!(BlockingExecutor, |_ctx, _queue| async {
    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
    Ok(())
});

async fn poll_task_state(handler: &RequestHandler, task_id: &TaskId, want: TaskState) -> TaskState {
    for _ in 0..200 {
        if let Ok(Some(t)) = handler.task_store.get(task_id).await
            && t.status.state == want
        {
            return want;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    handler
        .task_store
        .get(task_id)
        .await
        .ok()
        .flatten()
        .map_or(TaskState::Submitted, |t| t.status.state)
}

/// Regression: a `return_immediately` send must still drive the task to
/// completion in the background and persist the final state. Previously it
/// spawned no background processor, so the executor's events went nowhere
/// and the task was stuck in `Submitted` forever.
#[tokio::test]
async fn return_immediately_persists_final_state() {
    let handler = RequestHandlerBuilder::new(CompletingExecutor)
        .build()
        .unwrap();
    let mut params = make_params(Some("ctx-ri"));
    params.configuration = Some(SendMessageConfiguration {
        accepted_output_modes: vec!["text/plain".into()],
        task_push_notification_config: None,
        history_length: None,
        return_immediately: Some(true),
    });

    let SendMessageResult::Response(SendMessageResponse::Task(task)) =
        handler.on_send_message(params, false, None).await.unwrap()
    else {
        panic!("expected an immediate Task response");
    };
    assert_eq!(
        task.status.state,
        TaskState::Submitted,
        "snapshot is Submitted"
    );

    let final_state = poll_task_state(&handler, &task.id, TaskState::Completed).await;
    assert_eq!(
        final_state,
        TaskState::Completed,
        "fire-and-forget task must reach Completed in the store"
    );
}

/// Regression: a second send targeting a task already being processed must
/// be rejected, not spawn a second executor and overwrite the first's
/// cancellation token (leaving the original work uncancelable).
#[tokio::test]
async fn concurrent_send_to_in_flight_task_is_rejected() {
    let handler = RequestHandlerBuilder::new(BlockingExecutor)
        .build()
        .unwrap();

    // First send (fire-and-forget) leaves a live executor + token.
    let mut first = make_params(Some("ctx-dup"));
    first.configuration = Some(SendMessageConfiguration {
        accepted_output_modes: vec!["text/plain".into()],
        task_push_notification_config: None,
        history_length: None,
        return_immediately: Some(true),
    });
    let SendMessageResult::Response(SendMessageResponse::Task(task)) =
        handler.on_send_message(first, false, None).await.unwrap()
    else {
        panic!("expected an immediate Task response");
    };

    // Second send explicitly targets the same in-flight task.
    let mut second = make_params(Some("ctx-dup"));
    second.message.task_id = Some(task.id.clone());
    let result = handler.on_send_message(second, false, None).await;
    assert!(
        matches!(result, Err(ServerError::UnsupportedOperation(_))),
        "expected rejection of a send to an in-flight task, got {result:?}"
    );
}

/// Regression: hitting the concurrent-stream cap must return a clean
/// `Overloaded` error and create NO task (no orphaned `Submitted` row, no
/// leaked queue) — not a misleading internal error after committing the
/// task and token.
#[tokio::test]
async fn stream_cap_exhaustion_returns_overloaded_without_orphan() {
    let handler = RequestHandlerBuilder::new(BlockingExecutor)
        .with_max_concurrent_streams(1)
        .build()
        .unwrap();

    // First send consumes the single slot (its executor blocks, so the
    // queue stays alive).
    let mut first = make_params(Some("ctx-a"));
    first.configuration = Some(SendMessageConfiguration {
        accepted_output_modes: vec!["text/plain".into()],
        task_push_notification_config: None,
        history_length: None,
        return_immediately: Some(true),
    });
    handler.on_send_message(first, false, None).await.unwrap();
    assert_eq!(handler.event_queue_manager.active_count().await, 1);

    // Second send hits the cap.
    let mut second = make_params(Some("ctx-b"));
    second.configuration = Some(SendMessageConfiguration {
        accepted_output_modes: vec!["text/plain".into()],
        task_push_notification_config: None,
        history_length: None,
        return_immediately: Some(true),
    });
    let result = handler.on_send_message(second, false, None).await;
    assert!(
        matches!(result, Err(ServerError::Overloaded(_))),
        "expected Overloaded at capacity, got {result:?}"
    );
    // No queue was created for the rejected send, and no task orphaned.
    assert_eq!(
        handler.event_queue_manager.active_count().await,
        1,
        "capacity rejection must not create a queue"
    );
}

/// An exactly-empty `contextId` is proto3's unset value (the JSON bindings
/// are `ProtoJSON`, and a2a-java's JSON-RPC client prints it that way), so it
/// is accepted and a context is generated. Until 2026-09-09 this returned
/// `InvalidParams`, which failed every JSON-RPC pairing with the Java SDK in
/// the official ITK.
#[tokio::test]
async fn empty_context_id_is_unset_and_a_context_is_generated() {
    let handler = make_handler();
    let params = make_params(Some(""));

    let result = handler
        .on_send_message(params, false, None)
        .await
        .expect("an empty contextId is the unset value, not an invalid id");
    let task = match result {
        SendMessageResult::Response(SendMessageResponse::Task(task)) => task,
        other => panic!("expected a task, got {other:?}"),
    };
    assert!(
        !task.context_id.0.is_empty(),
        "a context must be generated for an unset one"
    );
}

/// Whitespace-only is not a value any `ProtoJSON` printer produces for
/// "unset", so it stays a client error: the normalisation is exactly-empty
/// and nothing wider.
#[tokio::test]
async fn whitespace_only_context_id_returns_invalid_params() {
    let handler = make_handler();
    let params = make_params(Some("   "));

    let result = handler.on_send_message(params, false, None).await;

    assert!(
        matches!(result, Err(ServerError::InvalidParams(_))),
        "expected InvalidParams for whitespace-only context_id, got {result:?}"
    );
}

#[tokio::test]
async fn too_long_context_id_returns_invalid_params() {
    // Covers line 98-99: context_id exceeding max_id_length.
    use crate::handler::limits::HandlerLimits;

    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(HandlerLimits::default().with_max_id_length(10))
        .build()
        .unwrap();
    let long_ctx = "x".repeat(20);
    let params = make_params(Some(&long_ctx));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::InvalidParams(ref msg)) if msg.contains("maximum length")),
        "expected InvalidParams for too-long context_id"
    );
}

#[tokio::test]
async fn too_long_task_id_returns_invalid_params() {
    // Covers lines 108-109: task_id exceeding max_id_length.
    use crate::handler::limits::HandlerLimits;
    use a2a_protocol_types::task::TaskId;

    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(HandlerLimits::default().with_max_id_length(10))
        .build()
        .unwrap();
    let mut params = make_params(None);
    params.message.task_id = Some(TaskId::new("a".repeat(20)));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::InvalidParams(ref msg)) if msg.contains("maximum length")),
        "expected InvalidParams for too-long task_id"
    );
}

/// An exactly-empty `taskId` is likewise unset: a new task is started rather
/// than the request rejected.
#[tokio::test]
async fn empty_task_id_is_unset_and_a_new_task_starts() {
    use a2a_protocol_types::task::TaskId;

    let handler = make_handler();
    let mut params = make_params(None);
    params.message.task_id = Some(TaskId::new(""));

    let result = handler
        .on_send_message(params, false, None)
        .await
        .expect("an empty taskId is the unset value, not an invalid id");
    assert!(
        matches!(result, SendMessageResult::Response(SendMessageResponse::Task(ref task)) if !task.id.0.is_empty()),
        "expected a fresh task, got {result:?}"
    );
}

#[tokio::test]
async fn whitespace_only_task_id_returns_invalid_params() {
    use a2a_protocol_types::task::TaskId;

    let handler = make_handler();
    let mut params = make_params(None);
    params.message.task_id = Some(TaskId::new(" \t"));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::InvalidParams(ref msg)) if msg.contains("empty")),
        "expected InvalidParams for whitespace-only task_id, got {result:?}"
    );
}

/// Saves `id` in `context`, non-terminal unless `state` says otherwise.
async fn seed_task(handler: &RequestHandler, id: &str, context: &str, state: TaskState) {
    use a2a_protocol_types::task::{Task, TaskStatus};

    let task = Task {
        id: TaskId::new(id),
        context_id: ContextId::new(context),
        status: TaskStatus::with_timestamp(state),
        history: None,
        artifacts: None,
        metadata: None,
    };
    handler.task_store.save(&task).await.unwrap();
}

/// A `taskId` naming no task at all is `TaskNotFound`, whether or not the
/// context happens to hold one.
///
/// This test previously expected `InvalidParams`, and that expectation was
/// wrong. §3.4.2: "When a client includes a taskId in a Message, it MUST
/// reference an existing task" — and the handler already returned
/// `TaskNotFound` for this very input when the context was empty, because
/// that branch looks the task up. The only reason the same input produced
/// `InvalidParams` here was that a *different* task existed in the context,
/// which says nothing about whether the named one does. Same input, two
/// answers, decided by something irrelevant to the question.
#[tokio::test]
async fn task_id_naming_no_existing_task_returns_task_not_found() {
    let handler = make_handler();
    seed_task(
        &handler,
        "stored-task-id",
        "ctx-existing",
        TaskState::InputRequired,
    )
    .await;

    let mut params = make_params(Some("ctx-existing"));
    params.message.task_id = Some(TaskId::new("no-such-task"));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::TaskNotFound(ref id)) if id.0 == "no-such-task"),
        "expected TaskNotFound for a taskId that names nothing, got: {result:?}"
    );
}

/// The one mismatch §3.4.3 requires an agent to reject: a `taskId` whose task
/// exists, but under a different `contextId`.
#[tokio::test]
async fn task_id_from_another_context_returns_invalid_params() {
    let handler = make_handler();
    seed_task(&handler, "ours", "ctx-existing", TaskState::InputRequired).await;
    seed_task(&handler, "theirs", "ctx-other", TaskState::InputRequired).await;

    let mut params = make_params(Some("ctx-existing"));
    params.message.task_id = Some(TaskId::new("theirs"));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::InvalidParams(ref msg)) if msg.contains("different context")),
        "expected InvalidParams for a cross-context taskId, got: {result:?}"
    );
}

/// A context holds more than one task, and every live one stays addressable.
///
/// Finding 2 of `docs/swarm-scale-findings.md`: `find_task_by_context` returns
/// a single task, the most recently updated live one, and naming any other
/// used to be rejected outright. So one participant posting without a `taskId`
/// forked the channel and locked everyone else out of the original — which was
/// still live, still in the same context, and by §3.4.1 still part of the group
/// the context denotes. §3.4.3 permits continuing "a specific task"; the only
/// rejection it mandates is a contextId that differs from the referenced
/// task's, which is not this.
#[tokio::test]
async fn a_live_sibling_task_in_the_same_context_is_still_addressable() {
    let handler = make_handler();
    // `older` is the one a participant holds a reference to. `newer` is the
    // fork that displaced it as what `find_task_by_context` returns.
    seed_task(&handler, "older", "ctx-shared", TaskState::InputRequired).await;
    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
    seed_task(&handler, "newer", "ctx-shared", TaskState::InputRequired).await;

    let mut params = make_params(Some("ctx-shared"));
    params.message.task_id = Some(TaskId::new("older"));

    let result = handler
        .on_send_message(params, false, None)
        .await
        .expect("a live task in this context must stay addressable after a fork displaced it");
    match result {
        SendMessageResult::Response(SendMessageResponse::Task(task)) => {
            assert_eq!(
                task.id,
                TaskId::new("older"),
                "the send must land on the task it named, not on the one the \
                 context lookup happens to return"
            );
        }
        other => panic!("expected a task response, got {other:?}"),
    }
}

/// Terminality is judged on the task actually named, not on whichever one the
/// context lookup returned.
#[tokio::test]
async fn a_terminal_sibling_task_is_still_rejected() {
    let handler = make_handler();
    seed_task(&handler, "done", "ctx-shared", TaskState::Completed).await;
    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
    seed_task(&handler, "live", "ctx-shared", TaskState::InputRequired).await;

    let mut params = make_params(Some("ctx-shared"));
    params.message.task_id = Some(TaskId::new("done"));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::UnsupportedOperation(ref msg)) if msg.contains("terminal")),
        "CORE-SEND-002 still applies to the named task, got: {result:?}"
    );
}

#[tokio::test]
async fn send_message_records_user_message_in_history() {
    // Task.history is the conversation record: the incoming user message
    // must be persisted with the task.
    let handler = make_handler();
    let result = handler
        .on_send_message(make_params(None), false, None)
        .await
        .expect("send should succeed");
    let task_id = match result {
        SendMessageResult::Response(SendMessageResponse::Task(t)) => t.id,
        other => panic!("expected task response, got {other:?}"),
    };
    let stored = handler
        .task_store
        .get(&task_id)
        .await
        .expect("get")
        .expect("task stored");
    let history = stored.history.expect("history populated on send");
    assert_eq!(history.len(), 1, "exactly the incoming user message");
    assert_eq!(history[0].role, MessageRole::User);
    assert_eq!(
        history[0].parts[0].text_content(),
        Some("hello"),
        "history records the message content"
    );
}

#[tokio::test]
async fn continuation_appends_history_and_preserves_artifacts() {
    // A continuation must carry the stored task's artifacts and metadata
    // forward and append the new message — not reset the task.
    use a2a_protocol_types::artifact::Artifact;
    let handler = make_handler();
    let prior = Task {
        id: TaskId::new("cont-task"),
        context_id: ContextId::new("ctx-cont"),
        status: TaskStatus::new(TaskState::InputRequired),
        history: Some(vec![Message {
            id: MessageId::new("m-prior"),
            role: MessageRole::User,
            parts: vec![Part::text("first turn")],
            context_id: None,
            task_id: None,
            reference_task_ids: None,
            extensions: None,
            metadata: None,
        }]),
        artifacts: Some(vec![Artifact::new("a1", vec![Part::text("turn-1 output")])]),
        metadata: Some(serde_json::json!({"k": "v"})),
    };
    handler.task_store.save(&prior).await.unwrap();

    let mut params = make_params(Some("ctx-cont"));
    params.message.task_id = Some(TaskId::new("cont-task"));
    handler
        .on_send_message(params, false, None)
        .await
        .expect("continuation should succeed");

    let stored = handler
        .task_store
        .get(&TaskId::new("cont-task"))
        .await
        .expect("get")
        .expect("task stored");
    let history = stored.history.expect("history preserved");
    assert_eq!(history.len(), 2, "prior message + continuation message");
    assert_eq!(history[0].parts[0].text_content(), Some("first turn"));
    assert_eq!(history[1].parts[0].text_content(), Some("hello"));
    assert!(
        stored.artifacts.as_ref().is_some_and(|a| a.len() == 1),
        "continuation must not wipe accumulated artifacts"
    );
    assert_eq!(
        stored.metadata,
        Some(serde_json::json!({"k": "v"})),
        "continuation must not wipe task metadata"
    );
}

/// Turn 1 of the issue-130 sequence: a completed task on `ctx-130` carrying
/// an artifact, a history and metadata that are all its own.
fn issue_130_completed_first_task() -> Task {
    use a2a_protocol_types::artifact::Artifact;
    Task {
        id: TaskId::new("T1"),
        context_id: ContextId::new("ctx-130"),
        status: TaskStatus::new(TaskState::Completed),
        history: Some(vec![Message {
            id: MessageId::new("m-turn-1"),
            role: MessageRole::User,
            parts: vec![Part::text("yo")],
            context_id: None,
            task_id: None,
            reference_task_ids: None,
            extensions: None,
            metadata: None,
        }]),
        artifacts: Some(vec![Artifact::new("A1", vec![Part::text("turn-1 output")])]),
        metadata: Some(serde_json::json!({"belongs-to": "T1"})),
    }
}

/// The other half of `continuation_appends_history_and_preserves_artifacts`:
/// a *new* task on an existing context starts clean, where a continuation of
/// the same task does not.
///
/// Issue #130. A client sent "yo" on a fresh context, the agent produced
/// artifact `A1` and completed the task; the client then sent "yo2" on the
/// same context with no `taskId`, so the server minted a new task id — and
/// returned that new task already holding `A1`, to which the agent's `A2`
/// delta was then merged. The reporter saw every round return the whole
/// context's artifacts, growing by one each turn.
///
/// `build_initial_task` had keyed its carry-forward on "a task exists for
/// this context" rather than "it is this task", so all three of `artifacts`,
/// `history` and `metadata` leaked across the task boundary. Only
/// `artifacts` was visible on the send response: history is omitted from it
/// unless the client asks for a `history_length` (`shape_response_history`),
/// which is why the report named artifacts alone. This asserts the stored
/// task, where all three are observable.
#[tokio::test]
async fn new_task_on_existing_context_does_not_inherit_the_prior_task() {
    let handler = make_handler();
    let t1 = issue_130_completed_first_task();
    handler.task_store.save(&t1).await.unwrap();

    // Turn 2: same context, no task_id — a new round, so a new task id.
    let result = handler
        .on_send_message(make_params(Some("ctx-130")), false, None)
        .await
        .expect("a new round on an existing context should succeed");
    let SendMessageResult::Response(SendMessageResponse::Task(t2)) = result else {
        panic!("expected a Task response, got: {result:?}");
    };
    assert_ne!(
        t2.id, t1.id,
        "turn 2 must be a new task, not a continuation"
    );
    assert_eq!(
        t2.context_id, t1.context_id,
        "the new task stays on the same context"
    );
    assert!(
        t2.artifacts.as_ref().is_none_or(Vec::is_empty),
        "the response task must not carry T1's artifacts, got: {:?}",
        t2.artifacts
    );

    // The stored task is where history and metadata are observable: the send
    // response omits history unless the client requested a history_length.
    let stored = handler
        .task_store
        .get(&t2.id)
        .await
        .expect("get")
        .expect("the new task is stored");
    assert!(
        stored.artifacts.as_ref().is_none_or(Vec::is_empty),
        "the stored task must not carry T1's artifacts, got: {:?}",
        stored.artifacts
    );
    assert_eq!(
        stored.metadata, None,
        "the stored task must not carry T1's metadata"
    );
    let history = stored.history.expect("history holds the incoming message");
    assert_eq!(
        history.len(),
        1,
        "history starts at the incoming message, not T1's turns: {history:?}"
    );
    assert_eq!(history[0].id, MessageId::new("msg-1"));

    // T1 is untouched by any of this.
    let t1_after = handler
        .task_store
        .get(&t1.id)
        .await
        .expect("get")
        .expect("T1 still stored");
    assert!(
        t1_after.artifacts.as_ref().is_some_and(|a| a.len() == 1),
        "T1 keeps its own artifact"
    );
}

#[tokio::test]
async fn history_is_capped_at_max_messages() {
    // The oldest messages are dropped once the cap is reached.
    let handler = make_handler();
    let mut long_history: Vec<Message> = (0..MAX_TASK_HISTORY_MESSAGES)
        .map(|i| Message {
            id: MessageId::new(format!("m-{i}")),
            role: MessageRole::User,
            parts: vec![Part::text(format!("msg {i}"))],
            context_id: None,
            task_id: None,
            reference_task_ids: None,
            extensions: None,
            metadata: None,
        })
        .collect();
    long_history[0].parts = vec![Part::text("OLDEST")];
    let prior = Task {
        id: TaskId::new("cap-task"),
        context_id: ContextId::new("ctx-cap"),
        status: TaskStatus::new(TaskState::InputRequired),
        history: Some(long_history),
        artifacts: None,
        metadata: None,
    };
    handler.task_store.save(&prior).await.unwrap();

    let mut params = make_params(Some("ctx-cap"));
    params.message.task_id = Some(TaskId::new("cap-task"));
    handler
        .on_send_message(params, false, None)
        .await
        .expect("continuation should succeed");

    let stored = handler
        .task_store
        .get(&TaskId::new("cap-task"))
        .await
        .unwrap()
        .unwrap();
    let history = stored.history.unwrap();
    assert_eq!(history.len(), MAX_TASK_HISTORY_MESSAGES, "capped");
    assert_ne!(
        history[0].parts[0].text_content(),
        Some("OLDEST"),
        "the oldest message is dropped first"
    );
    assert_eq!(
        history[MAX_TASK_HISTORY_MESSAGES - 1].parts[0].text_content(),
        Some("hello"),
        "the newest message is retained"
    );
}

#[tokio::test]
async fn send_response_omits_history_by_default_and_honors_history_length() {
    // The store keeps full history, but the send RESPONSE omits it
    // unless SendMessageConfiguration.historyLength asks for it —
    // echoing the just-sent message back doubled response payloads for
    // large sends (caught by the benchmark regression gate).
    use a2a_protocol_types::params::SendMessageConfiguration;
    let handler = make_handler();

    let result = handler
        .on_send_message(make_params(Some("ctx-resp")), false, None)
        .await
        .expect("send should succeed");
    let task = match result {
        SendMessageResult::Response(SendMessageResponse::Task(t)) => t,
        other => panic!("expected task response, got {other:?}"),
    };
    assert!(
        task.history.is_none(),
        "default send response must not echo history"
    );
    let stored = handler
        .task_store
        .get(&task.id)
        .await
        .unwrap()
        .expect("task stored");
    assert_eq!(
        stored.history.as_ref().map(Vec::len),
        Some(1),
        "the store still keeps the full history"
    );

    let mut params = make_params(Some("ctx-resp"));
    params.message.task_id = Some(task.id.clone());
    params.configuration = Some(SendMessageConfiguration {
        history_length: Some(10),
        ..Default::default()
    });
    let result = handler
        .on_send_message(params, false, None)
        .await
        .expect("continuation should succeed");
    let task = match result {
        SendMessageResult::Response(SendMessageResponse::Task(t)) => t,
        other => panic!("expected task response, got {other:?}"),
    };
    assert_eq!(
        task.history.as_ref().map(Vec::len),
        Some(2),
        "historyLength=10 returns the (2) stored messages"
    );
}

#[tokio::test]
async fn send_message_with_request_metadata() {
    // Covers line 186: setting request metadata on context.
    let handler = make_handler();
    let mut params = make_params(None);
    params.metadata = Some(serde_json::json!({"key": "value"}));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        result.is_ok(),
        "send_message with request metadata should succeed"
    );
}

#[tokio::test]
async fn send_message_error_path_records_metrics() {
    // Covers lines 195-199: the Err branch in the outer metrics match.
    use crate::call_context::CallContext;
    use crate::interceptor::ServerInterceptor;
    use std::future::Future;
    use std::pin::Pin;

    struct FailInterceptor;
    impl ServerInterceptor for FailInterceptor {
        fn before<'a>(
            &'a self,
            _ctx: &'a CallContext,
        ) -> Pin<Box<dyn Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>>
        {
            Box::pin(async {
                Err(a2a_protocol_types::error::A2aError::internal(
                    "forced failure",
                ))
            })
        }
        fn after<'a>(
            &'a self,
            _ctx: &'a CallContext,
        ) -> Pin<Box<dyn Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>>
        {
            Box::pin(async { Ok(()) })
        }
    }

    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_interceptor(FailInterceptor)
        .build()
        .unwrap();

    let params = make_params(None);
    let result = handler.on_send_message(params, false, None).await;
    assert!(
        result.is_err(),
        "send_message should fail when interceptor rejects, exercising error metrics path"
    );
}

#[tokio::test]
async fn send_streaming_message_error_path_records_metrics() {
    // Covers the streaming variant of the error metrics path (method_name = "SendStreamingMessage").
    use crate::call_context::CallContext;
    use crate::interceptor::ServerInterceptor;
    use std::future::Future;
    use std::pin::Pin;

    struct FailInterceptor;
    impl ServerInterceptor for FailInterceptor {
        fn before<'a>(
            &'a self,
            _ctx: &'a CallContext,
        ) -> Pin<Box<dyn Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>>
        {
            Box::pin(async {
                Err(a2a_protocol_types::error::A2aError::internal(
                    "forced failure",
                ))
            })
        }
        fn after<'a>(
            &'a self,
            _ctx: &'a CallContext,
        ) -> Pin<Box<dyn Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>>
        {
            Box::pin(async { Ok(()) })
        }
    }

    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_interceptor(FailInterceptor)
        .build()
        .unwrap();

    let params = make_params(None);
    let result = handler.on_send_message(params, true, None).await;
    assert!(
        result.is_err(),
        "streaming send_message should fail when interceptor rejects"
    );
}

#[tokio::test]
async fn streaming_mode_returns_stream_result() {
    // Covers lines 270-280: the streaming=true branch returning SendMessageResult::Stream.
    let handler = make_handler();
    let params = make_params(None);

    let result = handler.on_send_message(params, true, None).await;
    assert!(
        matches!(result, Ok(SendMessageResult::Stream(_))),
        "expected Stream result in streaming mode"
    );
}

#[tokio::test]
async fn send_message_with_stored_task_continuation() {
    // Covers setting stored_task on context when a non-terminal task
    // exists for the given context_id (e.g. input-required continuation).
    use a2a_protocol_types::task::{Task, TaskState, TaskStatus};

    let handler = make_handler();

    // Pre-save a non-terminal task with a known context_id.
    let task = Task {
        id: TaskId::new("existing-task"),
        context_id: ContextId::new("continue-ctx"),
        status: TaskStatus::new(TaskState::InputRequired),
        history: None,
        artifacts: None,
        metadata: None,
    };
    handler.task_store.save(&task).await.unwrap();

    // Send message with the same context_id — should find the stored task.
    let params = make_params(Some("continue-ctx"));
    let result = handler.on_send_message(params, false, None).await;
    assert!(
        result.is_ok(),
        "send_message with existing non-terminal context should succeed"
    );
}

#[tokio::test]
async fn send_message_to_terminal_task_returns_unsupported_operation() {
    // SPEC CORE-SEND-002: Messages explicitly targeting a task in terminal
    // state (via task_id) must be rejected with UnsupportedOperation.
    use a2a_protocol_types::task::{Task, TaskState, TaskStatus};

    let handler = make_handler();

    // Pre-save a completed task.
    let task = Task {
        id: TaskId::new("done-task"),
        context_id: ContextId::new("done-ctx"),
        status: TaskStatus::new(TaskState::Completed),
        history: None,
        artifacts: None,
        metadata: None,
    };
    handler.task_store.save(&task).await.unwrap();

    // Send message with explicit task_id targeting the terminal task.
    let mut params = make_params(Some("done-ctx"));
    params.message.task_id = Some(TaskId::new("done-task"));
    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::UnsupportedOperation(ref msg)) if msg.contains("terminal")),
        "expected UnsupportedOperation for terminal task, got: {result:?}"
    );
}

#[tokio::test]
async fn send_message_to_terminal_context_without_task_id_creates_new_task() {
    // When no task_id is provided but the context has a terminal task,
    // a new task should be created (new conversation round on same context).
    use a2a_protocol_types::task::{Task, TaskState, TaskStatus};

    let handler = make_handler();

    // Pre-save a completed task.
    let task = Task {
        id: TaskId::new("old-task"),
        context_id: ContextId::new("reuse-ctx"),
        status: TaskStatus::new(TaskState::Completed),
        history: None,
        artifacts: None,
        metadata: None,
    };
    handler.task_store.save(&task).await.unwrap();

    // Send message to the same context WITHOUT task_id — should succeed.
    let params = make_params(Some("reuse-ctx"));
    let result = handler.on_send_message(params, false, None).await;
    assert!(
        result.is_ok(),
        "should create new task on terminal context, got: {result:?}"
    );
}

#[tokio::test]
async fn send_message_with_headers() {
    // Covers line 76: build_call_context receives headers.
    let handler = make_handler();
    let params = make_params(None);
    let mut headers = HashMap::new();
    headers.insert("authorization".to_string(), "Bearer test-token".to_string());

    let result = handler.on_send_message(params, false, Some(&headers)).await;
    let send_result = result.expect("send_message with headers should succeed");
    assert!(
        matches!(
            send_result,
            SendMessageResult::Response(SendMessageResponse::Task(_))
        ),
        "expected Response(Task) for send with headers"
    );
}

#[tokio::test]
async fn duplicate_task_id_without_context_match_returns_error() {
    // Task exists under a different context — should return InvalidParams.
    use a2a_protocol_types::task::{Task, TaskId as TId, TaskState, TaskStatus};

    let handler = make_handler();

    // Pre-save a task with task_id "dup-task" but context "other-ctx".
    let task = Task {
        id: TId::new("dup-task"),
        context_id: ContextId::new("other-ctx"),
        status: TaskStatus::new(TaskState::Completed),
        history: None,
        artifacts: None,
        metadata: None,
    };
    handler.task_store.save(&task).await.unwrap();

    // Send a message with a new context_id but the same task_id.
    let mut params = make_params(Some("brand-new-ctx"));
    params.message.task_id = Some(TId::new("dup-task"));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::InvalidParams(ref msg)) if msg.contains("different context")),
        "expected InvalidParams for task_id in different context, got: {result:?}"
    );
}

#[tokio::test]
async fn unknown_task_id_returns_task_not_found() {
    // SPEC §3.4.2: Client-provided task_id must reference existing task.
    use a2a_protocol_types::task::TaskId as TId;

    let handler = make_handler();

    // Send message with a task_id that doesn't exist anywhere.
    let mut params = make_params(Some("fresh-ctx"));
    params.message.task_id = Some(TId::new("nonexistent-task"));

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::TaskNotFound(_))),
        "expected TaskNotFound for unknown task_id, got: {result:?}"
    );
}

#[tokio::test]
async fn send_message_with_tenant() {
    // Covers line 46: tenant scoping with non-default tenant.
    let handler = make_handler();
    let mut params = make_params(None);
    params.tenant = Some("test-tenant".to_string());

    let result = handler.on_send_message(params, false, None).await;
    let send_result = result.expect("send_message with tenant should succeed");
    assert!(
        matches!(
            send_result,
            SendMessageResult::Response(SendMessageResponse::Task(_))
        ),
        "expected Response(Task) for send with tenant"
    );
}

#[tokio::test]
async fn executor_timeout_returns_failed_task() {
    // Covers lines 228-236: the executor timeout path.
    use a2a_protocol_types::error::A2aResult;
    use std::time::Duration;

    struct SlowExecutor;
    impl crate::executor::AgentExecutor for SlowExecutor {
        fn execute<'a>(
            &'a self,
            _ctx: &'a crate::request_context::RequestContext,
            _queue: &'a dyn crate::streaming::EventQueueWriter,
        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = A2aResult<()>> + Send + 'a>>
        {
            Box::pin(async {
                tokio::time::sleep(Duration::from_secs(60)).await;
                Ok(())
            })
        }
    }

    let handler = RequestHandlerBuilder::new(SlowExecutor)
        .with_executor_timeout(Duration::from_millis(50))
        .build()
        .unwrap();

    let params = make_params(None);
    // The executor times out; collect_events sees the Failed status update,
    // which names the timeout so the blocking caller learns why.
    let task = match handler.on_send_message(params, false, None).await {
        Ok(SendMessageResult::Response(SendMessageResponse::Task(t))) => t,
        other => panic!("executor timeout should still return a task, got {other:?}"),
    };
    assert_eq!(task.status.state, TaskState::Failed);
    let text = status_text(&task).expect("the Failed status carries a message");
    assert!(
        text.contains("executor timed out after 0s"),
        "the status message names the timeout, got {text:?}"
    );
}

/// The text of a task's `status.message`, if it carries one.
fn status_text(task: &Task) -> Option<&str> {
    task.status.message.as_ref().and_then(Message::text)
}

#[tokio::test]
async fn executor_failure_writes_failed_event() {
    // Covers lines 243-258: executor error path writes a failed status event.
    use a2a_protocol_types::error::{A2aError, A2aResult};

    struct FailExecutor;
    impl crate::executor::AgentExecutor for FailExecutor {
        fn execute<'a>(
            &'a self,
            _ctx: &'a crate::request_context::RequestContext,
            _queue: &'a dyn crate::streaming::EventQueueWriter,
        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = A2aResult<()>> + Send + 'a>>
        {
            Box::pin(async { Err(A2aError::internal("executor exploded")) })
        }
    }

    let handler = RequestHandlerBuilder::new(FailExecutor).build().unwrap();
    let params = make_params(None);

    // The blocking caller sees the executor's error on the Failed task
    // itself: the failure event's `status.message` is what the sync
    // collector copies onto the task. Until 2026-09-10 the text rode only on
    // the event's `metadata.error`, which a blocking caller never receives
    // (`examples/resilient-agent`, Act 2).
    let task = match handler.on_send_message(params, false, None).await {
        Ok(SendMessageResult::Response(SendMessageResponse::Task(t))) => t,
        other => panic!("executor failure should produce a task result, got {other:?}"),
    };
    assert_eq!(task.status.state, TaskState::Failed);
    let message = task
        .status
        .message
        .as_ref()
        .expect("the Failed status carries the error as a message");
    assert_eq!(message.role, MessageRole::Agent);
    assert_eq!(message.task_id.as_ref(), Some(&task.id));
    assert_eq!(message.context_id.as_ref(), Some(&task.context_id));
    let text = message.text().expect("one text part");
    assert!(
        text.contains("executor exploded"),
        "the status message carries the executor's error, got {text:?}"
    );

    // And it is persisted, so `GetTask` after the fact says the same.
    let fetched = handler
        .on_get_task(
            a2a_protocol_types::params::TaskQueryParams {
                tenant: None,
                id: task.id.0.clone(),
                history_length: None,
            },
            None,
        )
        .await
        .expect("the task is fetchable after it failed");
    assert_eq!(fetched.status.state, TaskState::Failed);
    assert_eq!(
        status_text(&fetched),
        Some(text),
        "GetTask returns the same status message the send did"
    );
}

#[tokio::test]
async fn cancellation_token_sweep_runs_when_map_is_full() {
    // Covers lines 194-199: the cancellation token sweep when the map
    // exceeds max_cancellation_tokens.
    use crate::handler::limits::HandlerLimits;

    // Use a slow executor so tokens accumulate before being cleaned up.
    struct SlowExec;
    impl crate::executor::AgentExecutor for SlowExec {
        fn execute<'a>(
            &'a self,
            _ctx: &'a crate::request_context::RequestContext,
            _queue: &'a dyn crate::streaming::EventQueueWriter,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>>
                    + Send
                    + 'a,
            >,
        > {
            Box::pin(async {
                // Hold the token for a bit so tokens accumulate.
                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
                Ok(())
            })
        }
    }

    let handler = RequestHandlerBuilder::new(SlowExec)
        .with_handler_limits(HandlerLimits::default().with_max_cancellation_tokens(2))
        .build()
        .unwrap();

    // Send multiple streaming messages so tokens accumulate (streaming returns
    // immediately without waiting for executor to finish).
    for _ in 0..3 {
        let params = make_params(None);
        let _ = handler.on_send_message(params, true, None).await;
    }
    // If we get here without panic, the sweep logic ran successfully.
    // Clean up the slow executors.
    let _ = handler.shutdown().await;
}

#[tokio::test]
async fn stale_cancellation_tokens_cleaned_up() {
    // Covers lines 224-228: stale cancellation tokens are removed during sweep.
    use crate::handler::limits::HandlerLimits;
    use std::time::Duration;

    // Use a slow executor so tokens accumulate and become stale.
    struct SlowExec2;
    impl crate::executor::AgentExecutor for SlowExec2 {
        fn execute<'a>(
            &'a self,
            _ctx: &'a crate::request_context::RequestContext,
            _queue: &'a dyn crate::streaming::EventQueueWriter,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>>
                    + Send
                    + 'a,
            >,
        > {
            Box::pin(async {
                tokio::time::sleep(Duration::from_secs(10)).await;
                Ok(())
            })
        }
    }

    let handler = RequestHandlerBuilder::new(SlowExec2)
        .with_handler_limits(
            HandlerLimits::default()
                .with_max_cancellation_tokens(2)
                // Very short max_token_age so tokens become stale quickly.
                .with_max_token_age(Duration::from_millis(1)),
        )
        .build()
        .unwrap();

    // Send two streaming messages to fill up the token map.
    for _ in 0..2 {
        let params = make_params(None);
        let _ = handler.on_send_message(params, true, None).await;
    }

    // Wait for tokens to become stale.
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Send a third message; this should trigger the cleanup sweep
    // because the map is at capacity (>= max_cancellation_tokens)
    // and the existing tokens are stale (age > max_token_age).
    let params = make_params(None);
    let _ = handler.on_send_message(params, true, None).await;

    // The stale tokens should have been cleaned up.
    let _ = handler.shutdown().await;
}

#[tokio::test]
async fn streaming_executor_failure_writes_error_event() {
    // Covers lines 243-258 in streaming mode: executor error path.
    use a2a_protocol_types::error::{A2aError, A2aResult};

    struct FailExecutor;
    impl crate::executor::AgentExecutor for FailExecutor {
        fn execute<'a>(
            &'a self,
            _ctx: &'a crate::request_context::RequestContext,
            _queue: &'a dyn crate::streaming::EventQueueWriter,
        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = A2aResult<()>> + Send + 'a>>
        {
            Box::pin(async { Err(A2aError::internal("streaming fail")) })
        }
    }

    let handler = RequestHandlerBuilder::new(FailExecutor).build().unwrap();
    let params = make_params(None);

    let Ok(SendMessageResult::Stream(mut reader)) =
        handler.on_send_message(params, true, None).await
    else {
        panic!("streaming executor failure should still return stream");
    };
    // The streamed Failed event carries the error twice: as `metadata.error`,
    // where streaming callers have always read it, and as `status.message`,
    // which is what reaches a blocking caller and the store.
    let mut failed = None;
    while let Some(event) = reader.read().await {
        if let Ok(queued) = event
            && let StreamResponse::StatusUpdate(update) = queued.event
            && update.status.state == TaskState::Failed
        {
            failed = Some(update);
        }
    }
    let update = failed.expect("the stream ends with a Failed status update");
    let metadata_error = update
        .metadata
        .as_ref()
        .and_then(|m| m.get("error"))
        .and_then(serde_json::Value::as_str)
        .expect("metadata.error is still set for streaming callers");
    assert!(
        metadata_error.contains("streaming fail"),
        "{metadata_error:?}"
    );
    let message_text = update
        .status
        .message
        .as_ref()
        .and_then(Message::text)
        .expect("status.message carries the same text");
    assert_eq!(message_text, metadata_error);
}

#[tokio::test]
async fn input_required_continuation_reuses_task_id() {
    // When a client sends a task_id matching an existing non-terminal task
    // for the same context_id, the handler should reuse the task_id rather
    // than generating a new one (A2A spec §3.4.3).
    use a2a_protocol_types::task::{Task, TaskId, TaskState, TaskStatus};

    let handler = make_handler();

    // Pre-save a task in InputRequired state (non-terminal).
    let existing_task_id = TaskId::new("input-required-task");
    let task = Task {
        id: existing_task_id.clone(),
        context_id: ContextId::new("ctx-input"),
        status: TaskStatus::new(TaskState::InputRequired),
        history: None,
        artifacts: None,
        metadata: None,
    };
    handler.task_store.save(&task).await.unwrap();

    // Send a continuation message with the same context_id and task_id.
    let mut params = make_params(Some("ctx-input"));
    params.message.task_id = Some(existing_task_id.clone());

    let result = handler.on_send_message(params, false, None).await;
    let send_result = result.expect("continuation should succeed");
    match send_result {
        SendMessageResult::Response(SendMessageResponse::Task(t)) => {
            assert_eq!(
                t.id, existing_task_id,
                "task_id should be reused for input-required continuation"
            );
        }
        _ => panic!("expected Response(Task)"),
    }
}

// ── phases with an observable difference only a direct test can see ─────

/// The in-flight check refuses a resend on the strength of a live token
/// alone. The existing in-flight test cannot tell this check from the queue
/// lease behind it — a running executor has both a token and a queue, and
/// the lease refuses with the same error — so this plants a live token with
/// no queue, which only the token check can see.
#[tokio::test]
async fn a_live_token_alone_refuses_a_resend() {
    let handler = make_handler();
    let task_id = TaskId::new("in-flight-by-token");
    handler
        .task_store
        .save(&Task {
            id: task_id.clone(),
            context_id: ContextId::new("ctx-token-only"),
            status: TaskStatus::new(TaskState::InputRequired),
            history: None,
            artifacts: None,
            metadata: None,
        })
        .await
        .unwrap();
    handler.cancellation_tokens.write().await.insert(
        task_id.clone(),
        CancellationEntry {
            turn: std::sync::Arc::default(),
            token: tokio_util::sync::CancellationToken::new(),
            created_at: Instant::now(),
        },
    );

    let mut resend = make_params(Some("ctx-token-only"));
    resend.message.task_id = Some(task_id.clone());
    let result = handler.on_send_message(resend, false, None).await;
    assert!(
        matches!(result, Err(ServerError::UnsupportedOperation(_))),
        "a live token means an executor is in flight, got {result:?}"
    );
    assert!(
        !handler.event_queue_manager.has_queue(&task_id).await,
        "refused before a queue was leased"
    );
}

/// The other side of the test above: a token already cancelled (its
/// executor was told to stop, and cleanup has not yet removed the entry) is
/// not an executor in flight, so a continuation is admitted.
#[tokio::test]
async fn a_cancelled_token_alone_does_not_refuse_a_resend() {
    let handler = make_handler();
    let task_id = TaskId::new("cancelled-token-only");
    handler
        .task_store
        .save(&Task {
            id: task_id.clone(),
            context_id: ContextId::new("ctx-cancelled-token"),
            status: TaskStatus::new(TaskState::InputRequired),
            history: None,
            artifacts: None,
            metadata: None,
        })
        .await
        .unwrap();
    let token = tokio_util::sync::CancellationToken::new();
    token.cancel();
    handler.cancellation_tokens.write().await.insert(
        task_id.clone(),
        CancellationEntry {
            turn: std::sync::Arc::default(),
            token,
            created_at: Instant::now(),
        },
    );

    let mut resend = make_params(Some("ctx-cancelled-token"));
    resend.message.task_id = Some(task_id.clone());
    let result = handler.on_send_message(resend, false, None).await;
    assert!(
        result.is_ok(),
        "a cancelled token is not an executor in flight, got {result:?}"
    );
}

/// SPEC §3.4.3: a message carrying only a `taskId` continues that task in
/// the task's own context. A constant context in its place would find no
/// stored task there and refuse the continuation as a cross-context send.
#[tokio::test]
async fn a_task_id_only_continuation_infers_the_context() {
    let handler = make_handler();
    let task_id = TaskId::new("infer-my-context");
    handler
        .task_store
        .save(&Task {
            id: task_id.clone(),
            context_id: ContextId::new("ctx-inferred"),
            status: TaskStatus::new(TaskState::InputRequired),
            history: None,
            artifacts: None,
            metadata: None,
        })
        .await
        .unwrap();

    let mut params = make_params(None);
    params.message.task_id = Some(task_id.clone());
    let task = match handler.on_send_message(params, false, None).await {
        Ok(SendMessageResult::Response(SendMessageResponse::Task(t))) => t,
        other => panic!("expected the continuation to be accepted, got {other:?}"),
    };
    assert_eq!(task.id, task_id, "the named task is continued");
    assert_eq!(
        task.context_id.0, "ctx-inferred",
        "the context comes from the stored task, not from the message"
    );
}

/// An inline push config that cannot be registered — here because no push
/// sender is configured — fails the send and releases the queue and token
/// admitted for it, exactly as a store failure does. A leaked queue would
/// hold a concurrent-stream slot for a task that never runs.
#[tokio::test]
async fn a_refused_inline_push_config_releases_the_queue_and_token() {
    let handler = make_handler();
    let mut params = make_params(Some("ctx-push"));
    params.configuration = Some(SendMessageConfiguration {
        accepted_output_modes: vec!["text/plain".into()],
        task_push_notification_config: Some(a2a_protocol_types::push::TaskPushNotificationConfig {
            tenant: None,
            id: None,
            task_id: None,
            url: "https://example.com/hook".into(),
            token: None,
            authentication: None,
        }),
        history_length: None,
        return_immediately: None,
    });

    let result = handler.on_send_message(params, false, None).await;
    assert!(
        matches!(result, Err(ServerError::PushNotSupported)),
        "no push sender is configured, got {result:?}"
    );
    assert_eq!(
        handler.event_queue_manager.active_count().await,
        0,
        "the leased queue is released"
    );
    assert!(
        handler.cancellation_tokens.read().await.is_empty(),
        "the registered token is released"
    );
}

/// Seeds one live token that is already older than `max_token_age`, so the
/// sweep sees it as aged rather than cancelled.
///
/// "Older" is two seconds against the one-second `max_token_age` that
/// [`handler_with_token_cap`] configures, not an hour: `Instant::checked_sub`
/// returns `None` when the monotonic clock's epoch is younger than the
/// amount subtracted, and a freshly booted Windows CI runner is younger than
/// an hour. Measured 2026-09-10 on `Test (1.88, windows-latest)`, where the
/// hour-ago form panicked in both sweep tests.
async fn seed_aged_token(handler: &RequestHandler, id: &str) -> TaskId {
    let id = TaskId::new(id);
    handler.cancellation_tokens.write().await.insert(
        id.clone(),
        CancellationEntry {
            turn: std::sync::Arc::default(),
            token: tokio_util::sync::CancellationToken::new(),
            created_at: Instant::now()
                .checked_sub(std::time::Duration::from_secs(2))
                .expect("two seconds ago is representable on any booted host"),
        },
    );
    id
}

fn handler_with_token_cap(max: usize) -> RequestHandler {
    RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(
            crate::handler::limits::HandlerLimits::default()
                .with_max_cancellation_tokens(max)
                .with_max_token_age(std::time::Duration::from_secs(1)),
        )
        .build()
        .expect("build should succeed")
}

/// An aged token is evicted only once its event queue is gone — the executor
/// finished and the token lingered. `stale_cancellation_tokens_cleaned_up`
/// drives this path but asserts nothing, and the cancelled-token sweep test
/// never produces an aged candidate, so the confirmation step that turns
/// "aged" into "evictable" was pinned by neither.
#[tokio::test]
async fn the_sweep_evicts_an_aged_token_whose_queue_is_gone() {
    let handler = handler_with_token_cap(1);
    let aged = seed_aged_token(&handler, "aged-no-queue").await;

    let _ = handler
        .on_send_message(make_params(None), false, None)
        .await;

    assert!(
        !handler.cancellation_tokens.read().await.contains_key(&aged),
        "an aged token with no queue behind it is swept"
    );
}

/// The counterpart: an aged token whose queue is still registered belongs to
/// a live, long-running executor, and evicting it would make that task
/// uncancelable.
#[tokio::test]
async fn the_sweep_keeps_an_aged_token_whose_queue_is_live() {
    let handler = handler_with_token_cap(1);
    let aged = seed_aged_token(&handler, "aged-live-queue").await;
    let crate::streaming::QueueLease::Created { writer, .. } =
        handler.event_queue_manager.lease(&aged, false, None).await
    else {
        panic!("a fresh id leases a queue");
    };

    let _ = handler
        .on_send_message(make_params(None), false, None)
        .await;

    assert!(
        handler.cancellation_tokens.read().await.contains_key(&aged),
        "an aged token whose queue is live is kept"
    );
    drop(writer);
    handler.event_queue_manager.destroy(&aged).await;
}

// ── Send-path decision helpers ────────────────────────────────────────

#[test]
fn second_send_blocked_iff_token_live() {
    let live = CancellationEntry {
        turn: std::sync::Arc::default(),
        token: tokio_util::sync::CancellationToken::new(),
        created_at: Instant::now(),
    };
    assert!(
        second_send_blocked(&live),
        "a live token means an executor is in flight → block the second send"
    );

    let token = tokio_util::sync::CancellationToken::new();
    token.cancel();
    let cancelled = CancellationEntry {
        turn: std::sync::Arc::default(),
        token,
        created_at: Instant::now(),
    };
    assert!(
        !second_send_blocked(&cancelled),
        "a cancelled token no longer blocks a resend"
    );
}

#[test]
fn token_aged_at_or_past_max_age() {
    let max = std::time::Duration::from_secs(3600);
    assert!(
        !token_aged(std::time::Duration::from_secs(3599), max),
        "younger than max is not aged"
    );
    // Boundary: exactly max_age counts as aged (>=), which distinguishes
    // the correct operator from both `<` and `>`.
    assert!(
        token_aged(std::time::Duration::from_secs(3600), max),
        "exactly max_age is aged"
    );
    assert!(token_aged(std::time::Duration::from_secs(3601), max));
}

#[test]
fn evict_aged_token_only_when_queue_gone() {
    assert!(
        evict_aged_token(false),
        "no live queue → the executor finished → evict the lingering token"
    );
    assert!(
        !evict_aged_token(true),
        "a live queue means the task is still running → keep its token"
    );
}

/// Regression: the Phase-2 sweep removal must re-validate the entry under
/// the write lock. A fresh, live token inserted by a concurrent resend
/// between candidate collection and removal is neither cancelled nor
/// aged — deleting it would leave that executor uncancelable.
#[test]
fn token_still_evictable_spares_fresh_live_token() {
    let max_age = std::time::Duration::from_secs(3600);
    let now = Instant::now();

    // A fresh, live token (the concurrent-resend replacement): spared.
    let fresh = CancellationEntry {
        turn: std::sync::Arc::default(),
        token: tokio_util::sync::CancellationToken::new(),
        created_at: now,
    };
    assert!(
        !token_still_evictable(&fresh, now, max_age),
        "a fresh live token must never be swept"
    );

    // A cancelled token: still evictable.
    let cancelled = CancellationEntry {
        turn: std::sync::Arc::default(),
        token: tokio_util::sync::CancellationToken::new(),
        created_at: now,
    };
    cancelled.token.cancel();
    assert!(token_still_evictable(&cancelled, now, max_age));

    // An aged live token: still evictable (its queue-liveness gate ran
    // during candidate collection). Model "aged" by advancing the
    // comparison instant forward by `max_age` rather than subtracting from
    // `now` — `Instant::checked_sub` returns `None` on platforms whose
    // monotonic-clock epoch is younger than `max_age` (e.g. a freshly
    // booted Windows CI runner), which would spuriously fail the test.
    let aged = CancellationEntry {
        turn: std::sync::Arc::default(),
        token: tokio_util::sync::CancellationToken::new(),
        created_at: now,
    };
    let later = now
        .checked_add(max_age)
        .expect("now + max_age is representable");
    assert!(token_still_evictable(&aged, later, max_age));
}

// ── message.id takes the same rule as the other ids ──────────────────────

/// `message.id` was validated nowhere, while reaching further than either id
/// that was.
///
/// It is stored verbatim in `idempotency_keys.message_id`, it keys the
/// history de-duplication in `helpers.rs`, and it is interpolated into the
/// error a caller sees when a key is already held by a different message. The
/// key beside it in that same row is capped and charset-restricted precisely
/// because a value reaching store keys and log lines is where log injection
/// starts; the message id reached both with no defence at all, bounded only
/// by the dispatcher's request-body limit. An empty id was accepted and
/// stored, and every empty id collides with every other one.
///
/// `validate_id` had its own unit tests throughout. What was missing was the
/// call, so these drive the whole send path rather than the helper: they fail
/// if the call site is removed again.
#[tokio::test]
async fn an_empty_message_id_is_refused() {
    let handler = make_handler();
    let mut params = make_params(None);
    params.message.id = MessageId::new("");

    let err = handler
        .on_send_message(params, false, None)
        .await
        .expect_err("an empty message id must be refused, not stored");

    match err {
        ServerError::InvalidParams(msg) => assert!(
            msg.contains("message.id"),
            "the error must name the field the caller has to fix, got: {msg}"
        ),
        other => panic!("expected InvalidParams, got {other:?}"),
    }
}

#[tokio::test]
async fn an_over_long_message_id_is_refused() {
    let handler = make_handler();
    let over_limit = handler.limits.max_id_length + 1;
    let mut params = make_params(None);
    params.message.id = MessageId::new("m".repeat(over_limit));

    let err = handler
        .on_send_message(params, false, None)
        .await
        .expect_err("a message id past max_id_length must be refused");

    match err {
        ServerError::InvalidParams(msg) => assert!(
            msg.contains("message.id"),
            "the error must name the field the caller has to fix, got: {msg}"
        ),
        other => panic!("expected InvalidParams, got {other:?}"),
    }
}

/// Counter-test: an id *at* the limit is accepted.
///
/// Without it, an off-by-one that rejected every id — or a check that
/// rejected on `>=` — would satisfy both tests above.
#[tokio::test]
async fn a_message_id_at_the_limit_is_accepted() {
    let handler = make_handler();
    let at_limit = handler.limits.max_id_length;
    let mut params = make_params(None);
    params.message.id = MessageId::new("m".repeat(at_limit));

    handler
        .on_send_message(params, false, None)
        .await
        .expect("an id exactly at max_id_length is valid and must be accepted");
}

/// A tightened `max_id_length` must not refuse a conformant client.
///
/// `message.id` was bounded by `max_id_length` for one release. That number
/// is documented as covering task and context ids, which are commonly short,
/// and a deployment tightening it to 32 is doing exactly what the field
/// invites. But a message id is conventionally a UUID — 36 characters, which
/// is what `Message::id`'s own rustdoc tells a caller to send and what every
/// example in this workspace does send — so the tightened deployment refused
/// every message it received.
///
/// Caught by `examples/incident-response`'s handler-limits check, which sets
/// `max_id_length` to 32 and failed on a UUID.
#[tokio::test]
async fn a_uuid_message_id_survives_a_max_id_length_below_it() {
    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(crate::handler::HandlerLimits::default().with_max_id_length(32))
        .build()
        .expect("handler must build");

    let mut params = make_params(None);
    params.message.id = MessageId::new(uuid::Uuid::new_v4().to_string());
    assert_eq!(
        params.message.id.0.len(),
        36,
        "precondition: a hyphenated v4 UUID is 36 characters"
    );

    handler
        .on_send_message(params, false, None)
        .await
        .expect("a UUID message id must be accepted however tight max_id_length is");
}

/// Counter-test: the floor is a floor, not an exemption.
///
/// Without it, dropping the `message.id` check altogether would satisfy the
/// test above while restoring the defect that check exists for — an id
/// reaching `idempotency_keys.message_id` and a caller-visible error with no
/// bound at all.
#[tokio::test]
async fn an_id_past_the_floor_is_still_refused_when_the_limit_is_tight() {
    let handler = RequestHandlerBuilder::new(DummyExecutor)
        .with_handler_limits(crate::handler::HandlerLimits::default().with_max_id_length(32))
        .build()
        .expect("handler must build");

    let mut params = make_params(None);
    params.message.id = MessageId::new("m".repeat(37));

    let err = handler
        .on_send_message(params, false, None)
        .await
        .expect_err("37 characters is past the 36-character floor and must be refused");
    assert!(
        matches!(err, ServerError::InvalidParams(ref msg) if msg.contains("message.id")),
        "the error must still name the field, got: {err:?}"
    );
}

/// And a limit above the floor is honoured as given, so the clamp does not
/// quietly loosen a deployment that asked for more room.
#[tokio::test]
async fn a_limit_above_the_floor_is_used_as_given() {
    let limits = crate::handler::HandlerLimits::default().with_max_id_length(100);
    assert_eq!(limits.effective_max_message_id_length(), 100);

    let tight = crate::handler::HandlerLimits::default().with_max_id_length(8);
    assert_eq!(
        tight.effective_max_message_id_length(),
        crate::handler::MIN_MESSAGE_ID_LENGTH,
        "below the floor, the floor wins"
    );

    // The boundary itself. Both sides of the clamp agree here, which is
    // exactly why the `>` this was once spelled with had an equivalent
    // mutant; the arithmetic that replaced it still has to land on 36.
    let exactly = crate::handler::HandlerLimits::default()
        .with_max_id_length(crate::handler::MIN_MESSAGE_ID_LENGTH);
    assert_eq!(
        exactly.effective_max_message_id_length(),
        crate::handler::MIN_MESSAGE_ID_LENGTH,
        "a limit set to the floor is the floor"
    );
}

// ── multi-turn history accumulation ──────────────────────────────────────
//
// Nothing covered this. The suite has tests that a single send records its
// user message, and tests that the background processor appends an agent
// message, but none that a *continuation* keeps what earlier turns wrote.
// That gap is why `build_initial_task` could stop carrying the stored history
// forward and 2037 tests stayed green while a second turn truncated the
// conversation to the two messages it had itself produced.

/// An executor that replies once, so each turn writes both a user and an
/// agent message into the record.
struct ReplyingExecutor;
agent_executor!(ReplyingExecutor, |_ctx, queue| async {
    use a2a_protocol_types::events::StreamResponse;
    let _ = queue
        .write(StreamResponse::Message(Message {
            id: MessageId::new("agent-reply"),
            role: MessageRole::Agent,
            parts: vec![Part::text("ack")],
            context_id: None,
            task_id: None,
            reference_task_ids: None,
            extensions: None,
            metadata: None,
        }))
        .await;
    // A message alone makes the handler answer with the message rather than
    // the task (the "direct message" path). A non-terminal status keeps the
    // channel continuable, which is what a multi-turn conversation needs.
    let _ = queue
        .write(a2a_protocol_types::events::StreamResponse::StatusUpdate(
            a2a_protocol_types::events::TaskStatusUpdateEvent {
                task_id: _ctx.task_id.clone(),
                context_id: ContextId::new(_ctx.context_id.clone()),
                status: a2a_protocol_types::task::TaskStatus::with_timestamp(
                    a2a_protocol_types::task::TaskState::InputRequired,
                ),
                metadata: None,
            },
        ))
        .await;
    Ok(())
});

/// Three turns on one channel keep every message all three wrote.
#[tokio::test]
async fn a_continuation_keeps_what_earlier_turns_wrote() {
    let handler = RequestHandlerBuilder::new(ReplyingExecutor)
        .build()
        .expect("build handler");

    let mut task_id: Option<TaskId> = None;
    for turn in 0..3 {
        let mut params = make_params(Some("ctx-multi"));
        params.message.id = MessageId::new(format!("user-{turn}"));
        params.message.task_id = task_id.clone();

        let result = handler
            .on_send_message(params, false, None)
            .await
            .unwrap_or_else(|e| panic!("turn {turn} failed: {e:?}"));
        let task = match result {
            SendMessageResult::Response(SendMessageResponse::Task(t)) => t,
            other => panic!("turn {turn}: expected a task, got {other:?}"),
        };
        task_id = Some(task.id);

        // The executor runs in the background; give it a turn to reply and
        // for the reply to be persisted.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }

    let id = task_id.expect("a task id");
    let stored = handler
        .task_store
        .get(&id)
        .await
        .expect("get")
        .expect("the task must still be stored");
    let history = stored.history.unwrap_or_default();
    let user_messages: Vec<_> = history
        .iter()
        .filter(|m| m.role == MessageRole::User)
        .map(|m| m.id.0.as_str())
        .collect();

    assert_eq!(
        user_messages,
        vec!["user-0", "user-1", "user-2"],
        "every turn's user message must survive into the record; got a \
         history of {} message(s) overall, which means a later turn replaced \
         the conversation instead of extending it",
        history.len()
    );
}

/// A send that asks for history gets the conversation, not just its own turn.
///
/// `build_initial_task` stopped carrying the stored history forward, so the
/// task the send path holds now contains only the message this turn added.
/// The response is shaped from that task, which is why this needs its own
/// test: `historyLength` is the one caller that reads history off the send
/// response rather than off `GetTask`.
#[tokio::test]
async fn a_send_asking_for_history_gets_the_whole_conversation() {
    use a2a_protocol_types::params::SendMessageConfiguration;

    let handler = RequestHandlerBuilder::new(ReplyingExecutor)
        .build()
        .expect("build handler");

    let mut first = make_params(Some("ctx-histlen"));
    first.message.id = MessageId::new("user-0");
    let result = handler
        .on_send_message(first, false, None)
        .await
        .expect("first send");
    let task_id = match result {
        SendMessageResult::Response(SendMessageResponse::Task(t)) => t.id,
        other => panic!("expected a task, got {other:?}"),
    };
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    let mut second = make_params(Some("ctx-histlen"));
    second.message.id = MessageId::new("user-1");
    second.message.task_id = Some(task_id.clone());
    second.configuration = Some(SendMessageConfiguration {
        accepted_output_modes: vec![],
        task_push_notification_config: None,
        history_length: Some(10),
        return_immediately: Some(true),
    });

    let result = handler
        .on_send_message(second, false, None)
        .await
        .expect("second send");
    let task = match result {
        SendMessageResult::Response(SendMessageResponse::Task(t)) => t,
        other => panic!("expected a task, got {other:?}"),
    };

    let ids: Vec<_> = task
        .history
        .unwrap_or_default()
        .iter()
        .map(|m| m.id.0.clone())
        .collect();
    assert!(
        ids.contains(&"user-0".to_string()),
        "historyLength asked for 10 messages and the response carried {ids:?}; \
         the earlier turns are missing, so the response was shaped from the \
         send path's own one-message task instead of the stored conversation"
    );
}

// ── N21: a continuation racing the end of a parked turn ─────────────────
//
// `a_continuation_keeps_what_earlier_turns_wrote` above sleeps 50 ms between
// turns, and that sleep is what kept it green: without it, a client that
// answers `input-required` at once can arrive while the executor that parked
// the task is still returning, and was refused as "already being processed".
// `swarm_scale` found it under load (N21). These pin the race directly, by
// having the executor linger after it parks instead of relying on a busy host.

/// Parks the task at `input-required` on its first turn and lingers for
/// `linger` before returning, as an executor that tidies up after asking
/// does; completes the task on the turn that continues it.
struct LingeringExecutor {
    linger: std::time::Duration,
}

impl crate::executor::AgentExecutor for LingeringExecutor {
    fn execute<'a>(
        &'a self,
        ctx: &'a crate::request_context::RequestContext,
        queue: &'a dyn crate::streaming::EventQueueWriter,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>,
    > {
        Box::pin(async move {
            let continuing = ctx
                .stored_task
                .as_ref()
                .is_some_and(|t| t.status.state == TaskState::InputRequired);
            let state = if continuing {
                TaskState::Completed
            } else {
                TaskState::InputRequired
            };
            queue
                .write(StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
                    task_id: ctx.task_id.clone(),
                    context_id: ContextId::new(ctx.context_id.clone()),
                    status: TaskStatus::with_timestamp(state),
                    metadata: None,
                }))
                .await?;
            if !continuing {
                tokio::select! {
                    () = tokio::time::sleep(self.linger) => {}
                    () = ctx.cancellation_token.cancelled() => {}
                }
            }
            Ok(())
        })
    }
}

fn lingering_handler(linger: std::time::Duration, bound: std::time::Duration) -> RequestHandler {
    RequestHandlerBuilder::new(LingeringExecutor { linger })
        .with_handler_limits(
            crate::handler::HandlerLimits::default().with_executor_drain_timeout(bound),
        )
        .build()
        .expect("build handler")
}

/// Sends the first turn blocking, returning the parked task's id.
async fn park(handler: &RequestHandler, context: &str) -> TaskId {
    let SendMessageResult::Response(SendMessageResponse::Task(task)) = handler
        .on_send_message(make_params(Some(context)), false, None)
        .await
        .expect("first turn")
    else {
        panic!("expected a task");
    };
    assert_eq!(task.status.state, TaskState::InputRequired);
    task.id
}

fn continuation(context: &str, task_id: &TaskId) -> MessageSendParams {
    let mut params = make_params(Some(context));
    params.message.id = MessageId::new("msg-2");
    params.message.task_id = Some(task_id.clone());
    params
}

/// The blocking reply says `input-required`; answering it at once must be
/// admitted, not refused because the executor has not yet returned.
#[tokio::test]
async fn an_immediate_answer_to_input_required_is_admitted() {
    let linger = std::time::Duration::from_millis(300);
    let handler = lingering_handler(linger, std::time::Duration::from_secs(10));
    let id = park(&handler, "ctx-n21").await;

    let started = std::time::Instant::now();
    let result = handler
        .on_send_message(continuation("ctx-n21", &id), false, None)
        .await;
    let Ok(SendMessageResult::Response(SendMessageResponse::Task(task))) = result else {
        panic!("the continuation of an input-required task was refused: {result:?}");
    };
    assert_eq!(task.id, id);
    assert_eq!(task.status.state, TaskState::Completed);
    // It waited for the first executor rather than running beside it.
    assert!(
        started.elapsed() < linger + std::time::Duration::from_secs(5),
        "admission waited far past the executor's return: {:?}",
        started.elapsed()
    );
}

/// The same answer from a streaming client, which sees `input-required`
/// on the stream while the executor is still inside `execute`.
#[tokio::test]
async fn a_streaming_answer_to_input_required_is_admitted() {
    let handler = lingering_handler(
        std::time::Duration::from_millis(300),
        std::time::Duration::from_secs(10),
    );
    let SendMessageResult::Stream(mut reader) = handler
        .on_send_message(make_params(Some("ctx-n21s")), true, None)
        .await
        .expect("first turn")
    else {
        panic!("expected a stream");
    };
    let id = loop {
        let event = reader
            .read()
            .await
            .expect("the stream ended before input-required")
            .expect("event");
        if let StreamResponse::StatusUpdate(u) = event.event
            && u.status.state == TaskState::InputRequired
        {
            break u.task_id;
        }
    };

    let result = handler
        .on_send_message(continuation("ctx-n21s", &id), false, None)
        .await;
    let Ok(SendMessageResult::Response(SendMessageResponse::Task(task))) = result else {
        panic!("a continuation sent on seeing input-required was refused: {result:?}");
    };
    assert_eq!(task.status.state, TaskState::Completed);
}

/// The wait is bounded: an executor that parks and then never returns
/// still refuses the continuation, after the bound and not before, and
/// no second executor is started beside it.
#[tokio::test]
async fn a_parked_executor_that_never_returns_is_refused_after_the_bound() {
    let bound = std::time::Duration::from_millis(200);
    let handler = lingering_handler(std::time::Duration::from_secs(3600), bound);
    let id = park(&handler, "ctx-n21b").await;

    let started = std::time::Instant::now();
    let result = handler
        .on_send_message(continuation("ctx-n21b", &id), false, None)
        .await;
    let waited = started.elapsed();
    assert!(
        matches!(result, Err(ServerError::UnsupportedOperation(_))),
        "a continuation beside a live executor must be refused, got {result:?}"
    );
    assert!(
        waited >= bound,
        "refused after {waited:?}, before the {bound:?} bound"
    );
    let _ = handler.shutdown().await;
}

/// A send into a task whose executor has not parked it is refused at once,
/// as before N21: only a parked turn is waited for.
#[tokio::test]
async fn a_send_into_a_working_task_is_refused_without_waiting() {
    let bound = std::time::Duration::from_secs(30);
    let handler = RequestHandlerBuilder::new(BlockingExecutor)
        .with_handler_limits(
            crate::handler::HandlerLimits::default().with_executor_drain_timeout(bound),
        )
        .build()
        .unwrap();
    let mut first = make_params(Some("ctx-n21w"));
    first.configuration = Some(SendMessageConfiguration {
        accepted_output_modes: vec!["text/plain".into()],
        task_push_notification_config: None,
        history_length: None,
        return_immediately: Some(true),
    });
    let SendMessageResult::Response(SendMessageResponse::Task(task)) =
        handler.on_send_message(first, false, None).await.unwrap()
    else {
        panic!("expected an immediate Task response");
    };

    let started = std::time::Instant::now();
    let result = handler
        .on_send_message(continuation("ctx-n21w", &task.id), false, None)
        .await;
    assert!(matches!(result, Err(ServerError::UnsupportedOperation(_))));
    assert!(
        started.elapsed() < std::time::Duration::from_secs(5),
        "a send into a working task waited {:?} before its refusal",
        started.elapsed()
    );
    let _ = handler.shutdown().await;
}

// ── A send dropped mid-commit ────────────────────────────────────────────
//
// hyper drops a request's future when its client goes away, so every await
// in the send path is a place the future can simply stop. Between leasing
// the task's queue and spawning its executor the path holds three things
// only an `Err` used to release: the queue lease, the cancellation token and
// the idempotency key. A drop there released none of them.

/// A store whose `save_appending_history` — the write that persists a new
/// turn — can be held open, so a test can drop the send while it waits.
struct GatedStore {
    inner: crate::store::InMemoryTaskStore,
    armed: std::sync::atomic::AtomicBool,
    entered: tokio::sync::Notify,
}

impl crate::store::TaskStore for GatedStore {
    fn save<'a>(
        &'a self,
        task: &'a a2a_protocol_types::task::Task,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>,
    > {
        self.inner.save(task)
    }
    fn get<'a>(
        &'a self,
        id: &'a TaskId,
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<
                    Output = a2a_protocol_types::error::A2aResult<
                        Option<a2a_protocol_types::task::Task>,
                    >,
                > + Send
                + 'a,
        >,
    > {
        self.inner.get(id)
    }
    fn list<'a>(
        &'a self,
        params: &'a a2a_protocol_types::params::ListTasksParams,
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<
                    Output = a2a_protocol_types::error::A2aResult<
                        a2a_protocol_types::responses::TaskListResponse,
                    >,
                > + Send
                + 'a,
        >,
    > {
        self.inner.list(params)
    }
    fn insert_if_absent<'a>(
        &'a self,
        task: &'a a2a_protocol_types::task::Task,
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<bool>>
                + Send
                + 'a,
        >,
    > {
        self.inner.insert_if_absent(task)
    }
    fn delete<'a>(
        &'a self,
        id: &'a TaskId,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>,
    > {
        self.inner.delete(id)
    }
    fn supports_idempotency(&self) -> bool {
        self.inner.supports_idempotency()
    }
    fn claim_idempotency_key<'a>(
        &'a self,
        key: &'a str,
        message_id: &'a MessageId,
        task_id: &'a TaskId,
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<
                    Output = a2a_protocol_types::error::A2aResult<crate::store::IdempotencyClaim>,
                > + Send
                + 'a,
        >,
    > {
        self.inner.claim_idempotency_key(key, message_id, task_id)
    }
    fn release_idempotency_key<'a>(
        &'a self,
        key: &'a str,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>,
    > {
        self.inner.release_idempotency_key(key)
    }
    fn save_appending_history<'a>(
        &'a self,
        task: &'a a2a_protocol_types::task::Task,
        messages: &'a [Message],
        max_history: usize,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>,
    > {
        Box::pin(async move {
            if self.armed.load(std::sync::atomic::Ordering::SeqCst) {
                self.entered.notify_one();
                std::future::pending::<()>().await;
            }
            self.inner
                .save_appending_history(task, messages, max_history)
                .await
        })
    }
}

/// A continuation whose request is dropped while its turn is being persisted
/// leaves nothing behind: no token, no queue, and the next continuation of
/// the same task is admitted.
#[tokio::test]
async fn a_send_dropped_mid_commit_releases_what_it_took() {
    let (store, handler) = gated_handler(std::time::Duration::ZERO);
    let id = TaskId::new("t-dropped");
    crate::store::TaskStore::save(
        &store.inner,
        &a2a_protocol_types::task::Task {
            id: id.clone(),
            context_id: ContextId::new("ctx-dropped"),
            status: TaskStatus::new(TaskState::InputRequired),
            history: None,
            artifacts: None,
            metadata: None,
        },
    )
    .await
    .unwrap();

    store.armed.store(true, std::sync::atomic::Ordering::SeqCst);
    let entered = store.entered.notified();
    let send = tokio::spawn({
        let handler = std::sync::Arc::clone(&handler);
        let params = continuation("ctx-dropped", &id);
        async move { handler.on_send_message(params, false, None).await }
    });
    tokio::time::timeout(std::time::Duration::from_secs(5), entered)
        .await
        .expect("the send never reached the store write");
    // The client goes away: hyper drops the request future.
    send.abort();
    let _ = send.await;
    store
        .armed
        .store(false, std::sync::atomic::Ordering::SeqCst);

    // Whatever releases a dropped send's resources may be spawned; give it
    // a bounded moment rather than a fixed sleep.
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    while (handler.cancellation_tokens.read().await.contains_key(&id)
        || handler.event_queue_manager.active_count().await > 0)
        && tokio::time::Instant::now() < deadline
    {
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    assert!(
        !handler.cancellation_tokens.read().await.contains_key(&id),
        "a dropped send left its cancellation token registered"
    );
    assert_eq!(
        handler.event_queue_manager.active_count().await,
        0,
        "a dropped send left its event queue leased"
    );
    let retry = handler
        .on_send_message(continuation("ctx-dropped", &id), false, None)
        .await;
    assert!(
        retry.is_ok(),
        "the task is wedged: a continuation after a dropped send was refused with {retry:?}"
    );
}

fn gated_handler(
    linger: std::time::Duration,
) -> (std::sync::Arc<GatedStore>, std::sync::Arc<RequestHandler>) {
    let store = std::sync::Arc::new(GatedStore {
        inner: crate::store::InMemoryTaskStore::new(),
        armed: std::sync::atomic::AtomicBool::new(false),
        entered: tokio::sync::Notify::new(),
    });
    let handler = std::sync::Arc::new(
        RequestHandlerBuilder::new(LingeringExecutor { linger })
            .with_task_store_arc(store.clone())
            .with_handler_limits(
                crate::handler::HandlerLimits::default()
                    .with_executor_drain_timeout(std::time::Duration::from_secs(30)),
            )
            .build()
            .expect("build handler"),
    );
    (store, handler)
}

/// A continuation dropped while it waits for the previous turn's executor
/// (the N21 wait) must leave that executor's token alone: it is still the
/// only way to cancel the work in flight.
#[tokio::test]
async fn a_send_dropped_while_waiting_leaves_the_running_turn_cancelable() {
    let (_store, handler) = gated_handler(std::time::Duration::from_secs(3600));
    let id = park(&handler, "ctx-wait-drop").await;
    let running = handler
        .cancellation_tokens
        .read()
        .await
        .get(&id)
        .map(|e| std::sync::Arc::clone(&e.turn))
        .expect("the parked turn's executor is still running");

    let send = tokio::spawn({
        let handler = std::sync::Arc::clone(&handler);
        let params = continuation("ctx-wait-drop", &id);
        async move { handler.on_send_message(params, false, None).await }
    });
    // Long enough to be inside the admission wait, far short of its bound.
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    assert!(!send.is_finished(), "the continuation should be waiting");
    send.abort();
    let _ = send.await;
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let still = handler
        .cancellation_tokens
        .read()
        .await
        .get(&id)
        .map(|e| std::sync::Arc::clone(&e.turn));
    assert!(
        still.is_some_and(|t| std::sync::Arc::ptr_eq(&t, &running)),
        "dropping a waiting continuation released the running turn's token"
    );
    let _ = handler.shutdown().await;
}

/// A fresh send carrying an idempotency key, dropped before its task was
/// stored, releases the key: the client's retry of the same message runs
/// instead of waiting on a task that will never exist.
#[tokio::test]
async fn a_send_dropped_mid_commit_releases_its_idempotency_key() {
    let (store, handler) = gated_handler(std::time::Duration::ZERO);
    let mut params = make_params(Some("ctx-key-drop"));
    a2a_protocol_types::idempotency::set_key(
        &mut params.message,
        "retry-key-0001-0123456789abcdef",
    )
    .expect("valid key");

    store.armed.store(true, std::sync::atomic::Ordering::SeqCst);
    let entered = store.entered.notified();
    let send = tokio::spawn({
        let handler = std::sync::Arc::clone(&handler);
        let params = params.clone();
        async move { handler.on_send_message(params, false, None).await }
    });
    tokio::time::timeout(std::time::Duration::from_secs(5), entered)
        .await
        .expect("the send never reached the store write");
    send.abort();
    let _ = send.await;
    store
        .armed
        .store(false, std::sync::atomic::Ordering::SeqCst);
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let retry = tokio::time::timeout(
        std::time::Duration::from_secs(10),
        handler.on_send_message(params, false, None),
    )
    .await
    .expect("the retry hung on the dropped send's key");
    assert!(
        matches!(
            retry,
            Ok(SendMessageResult::Response(SendMessageResponse::Task(_)))
        ),
        "the retry of a dropped keyed send did not run: {retry:?}"
    );
}

/// Works for `delay`, then completes the task.
struct SlowCompletingExecutor {
    delay: std::time::Duration,
}

impl crate::executor::AgentExecutor for SlowCompletingExecutor {
    fn execute<'a>(
        &'a self,
        ctx: &'a crate::request_context::RequestContext,
        queue: &'a dyn crate::streaming::EventQueueWriter,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>,
    > {
        Box::pin(async move {
            let status = |state| {
                StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
                    task_id: ctx.task_id.clone(),
                    context_id: ContextId::new(ctx.context_id.clone()),
                    status: TaskStatus::with_timestamp(state),
                    metadata: None,
                })
            };
            queue.write(status(TaskState::Working)).await?;
            tokio::time::sleep(self.delay).await;
            queue.write(status(TaskState::Completed)).await?;
            Ok(())
        })
    }
}

/// A blocking send whose client goes away while the agent works — a client
/// timeout on a slow model call, typically — still leaves the task's
/// outcome in the store. The request future was the only thing persisting a
/// blocking send's events, so the executor finished into nothing and the
/// task stayed `working` for good.
#[tokio::test]
async fn a_blocking_send_dropped_mid_work_still_records_the_outcome() {
    let handler = std::sync::Arc::new(
        RequestHandlerBuilder::new(SlowCompletingExecutor {
            delay: std::time::Duration::from_millis(200),
        })
        .build()
        .expect("build handler"),
    );
    let send = tokio::spawn({
        let handler = std::sync::Arc::clone(&handler);
        async move {
            handler
                .on_send_message(make_params(Some("ctx-blocking-drop")), false, None)
                .await
        }
    });
    // Once the store shows the task `working` — the executor has started and
    // the collection is persisting — drop the request.
    let started = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        let listed = handler
            .task_store
            .list(&a2a_protocol_types::params::ListTasksParams::default())
            .await
            .expect("list");
        if listed.tasks.first().map(|t| t.status.state) == Some(TaskState::Working) {
            break;
        }
        assert!(
            tokio::time::Instant::now() < started,
            "the task never started working"
        );
        tokio::task::yield_now().await;
    }
    send.abort();
    let _ = send.await;

    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    let state = loop {
        let listed = handler
            .task_store
            .list(&a2a_protocol_types::params::ListTasksParams::default())
            .await
            .expect("list");
        let state = listed.tasks.first().map(|t| t.status.state);
        if state == Some(TaskState::Completed) || tokio::time::Instant::now() >= deadline {
            break state;
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    };
    assert_eq!(
        state,
        Some(TaskState::Completed),
        "the agent completed the task, but the store never heard"
    );
}

/// An interceptor whose `after` hook refuses every call.
struct FailingAfter;

impl crate::interceptor::ServerInterceptor for FailingAfter {
    fn before<'a>(
        &'a self,
        _ctx: &'a crate::call_context::CallContext,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>,
    > {
        Box::pin(async { Ok(()) })
    }
    fn after<'a>(
        &'a self,
        _ctx: &'a crate::call_context::CallContext,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = a2a_protocol_types::error::A2aResult<()>> + Send + 'a>,
    > {
        Box::pin(async {
            Err(a2a_protocol_types::error::A2aError::internal(
                "after hook failed",
            ))
        })
    }
}

/// An `after` hook that fails does not orphan the task it ran after. The
/// executor is already running when `after` runs; the hook's error used to
/// drop the send before anything persisting its events was attached, so a
/// task the agent completed stayed `submitted` in the store.
#[tokio::test]
async fn a_failing_after_hook_does_not_orphan_the_running_task() {
    for streaming in [false, true] {
        let handler = RequestHandlerBuilder::new(SlowCompletingExecutor {
            delay: std::time::Duration::from_millis(20),
        })
        .with_interceptor(FailingAfter)
        .build()
        .expect("build handler");
        let context = format!("ctx-after-{streaming}");
        let result = handler
            .on_send_message(make_params(Some(&context)), streaming, None)
            .await;
        assert!(result.is_err(), "the after hook's error is still reported");

        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
        let state = loop {
            let listed = handler
                .task_store
                .list(&a2a_protocol_types::params::ListTasksParams::default())
                .await
                .expect("list");
            let state = listed.tasks.first().map(|t| t.status.state);
            if state == Some(TaskState::Completed) || tokio::time::Instant::now() >= deadline {
                break state;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        };
        assert_eq!(
            state,
            Some(TaskState::Completed),
            "streaming={streaming}: the agent completed the task, but the store never heard"
        );
    }
}