use a2a_protocol_types::events::StreamResponse;
use a2a_protocol_types::task::TaskState;
use super::super::report::CheckResult;
use super::{RUN_TIMEOUT, Run, RunOutcome, states};
impl Run {
pub(in crate::conformance) fn did_not_panic(&self, name: &'static str) -> CheckResult {
match &self.outcome {
RunOutcome::Panicked => CheckResult::fail(
name,
"the executor panicked; the handler turns that into a Failed task, \
but the panic message is lost to the caller",
),
RunOutcome::Ok => CheckResult::pass(name, "returned normally"),
RunOutcome::Err(e) => {
CheckResult::pass(name, format!("returned Err({e}) without panicking"))
}
RunOutcome::TimedOut => {
CheckResult::pass(name, "did not return in time, but did not panic")
}
}
}
pub(in crate::conformance) fn returned_in_time(&self) -> CheckResult {
const NAME: &str = "returns_within_the_time_limit";
if matches!(self.outcome, RunOutcome::TimedOut) {
return CheckResult::fail(
NAME,
format!(
"did not return within {RUN_TIMEOUT:?}; the server would hold \
the task open until its own executor timeout, and every \
subscriber with it"
),
);
}
CheckResult::pass(NAME, "returned within the time limit")
}
pub(in crate::conformance) fn cancellation_is_not_an_error(&self) -> CheckResult {
const NAME: &str = "cancellation_is_not_an_error";
match &self.outcome {
RunOutcome::Err(e) => CheckResult::fail(
NAME,
format!(
"returned Err({e}) on a cancelled token; the handler writes \
Failed over the Canceled the cancel path already emitted, so \
the caller is told the work broke rather than that it was \
cancelled"
),
),
_ => CheckResult::pass(NAME, "did not report cancellation as a failure"),
}
}
pub(in crate::conformance) fn stops_when_cancelled_mid_run(&self) -> CheckResult {
const NAME: &str = "stops_when_cancelled_mid_run";
if matches!(self.outcome, RunOutcome::TimedOut) {
return CheckResult::fail(
NAME,
format!(
"still running {RUN_TIMEOUT:?} after the token was cancelled \
mid-work; checking the token once on entry is not enough, \
because cancellation almost always arrives later than that"
),
);
}
if self.events.is_empty() {
return CheckResult::skip(
NAME,
"the executor emitted nothing, so there was no first write to \
cancel at",
);
}
let completed_at = self.events.iter().position(|e| {
matches!(e, StreamResponse::StatusUpdate(u) if u.status.state == TaskState::Completed)
});
if completed_at.is_some_and(|i| i >= 1) {
return CheckResult::fail(
NAME,
"emitted Completed after the token was cancelled during its first \
write; cancellation is cooperative, so an executor that checks \
the token on entry and never again cannot be cancelled once it \
has started — which is when cancellation almost always arrives",
);
}
CheckResult::pass(
NAME,
states(&self.events)
.last()
.map_or_else(|| "stopped".to_owned(), |s| format!("stopped at {s}")),
)
}
pub(in crate::conformance) fn ends_in_terminal_or_interrupt(&self) -> CheckResult {
const NAME: &str = "ends_in_terminal_or_interrupt";
if self.truncated {
return CheckResult::skip(
NAME,
"the event queue lagged, so the sequence this check reads is \
incomplete",
);
}
if matches!(self.outcome, RunOutcome::TimedOut) {
return CheckResult::fail(
NAME,
"never returned, so the task never reached a terminal state",
);
}
if matches!(self.outcome, RunOutcome::Err(_) | RunOutcome::Panicked) {
if self.events.is_empty() {
return CheckResult::fail(
NAME,
"failed without emitting anything, so no protocol behaviour \
could be observed at all. If the executor rejects this \
harness's probe message, drive it with `check_with` and a \
message it accepts.",
);
}
return CheckResult::skip(
NAME,
"the executor failed, so the handler writes the terminal status",
);
}
match states(&self.events).last().copied() {
None => CheckResult::fail(
NAME,
"returned Ok having emitted no status at all; the task stays \
Submitted for ever and every caller blocks or polls until it \
times out",
),
Some(s) if s.is_terminal() || s.is_interrupted() => {
CheckResult::pass(NAME, format!("ended in {s}"))
}
Some(s) => CheckResult::fail(
NAME,
format!(
"returned Ok with the task left in {s}; a caller cannot tell \
that from work still in progress"
),
),
}
}
pub(in crate::conformance) fn transitions_are_legal(&self) -> CheckResult {
const NAME: &str = "transitions_are_legal";
let seen = states(&self.events);
if seen.len() < 2 {
return CheckResult::skip(NAME, "fewer than two statuses emitted");
}
for pair in seen.windows(2) {
let (from, to) = (pair[0], pair[1]);
if !from.can_transition_to(to) {
return CheckResult::fail(
NAME,
format!("{from} -> {to} is not a legal transition; the handler rejects it"),
);
}
}
CheckResult::pass(NAME, format!("{} transitions, all legal", seen.len() - 1))
}
pub(in crate::conformance) fn nothing_after_terminal(&self) -> CheckResult {
const NAME: &str = "nothing_after_terminal";
let mut terminal_at = None;
for (i, event) in self.events.iter().enumerate() {
if let StreamResponse::StatusUpdate(u) = event
&& u.status.state.is_terminal()
{
terminal_at = Some((i, u.status.state));
break;
}
}
let Some((i, state)) = terminal_at else {
return CheckResult::skip(NAME, "no terminal status emitted");
};
let after = self.events.len() - i - 1;
if after > 0 {
return CheckResult::fail(
NAME,
format!(
"{after} event(s) emitted after the terminal {state}; a task \
that has ended cannot produce more, and subscribers have \
already been told it finished"
),
);
}
CheckResult::pass(NAME, format!("{state} was the last event"))
}
pub(in crate::conformance) fn artifacts_have_ids(&self) -> CheckResult {
const NAME: &str = "artifacts_have_ids";
let artifacts: Vec<_> = self
.events
.iter()
.filter_map(|e| match e {
StreamResponse::ArtifactUpdate(u) => Some(&u.artifact),
_ => None,
})
.collect();
if artifacts.is_empty() {
return CheckResult::skip(NAME, "no artifacts emitted");
}
if artifacts.iter().any(|a| a.id.0.trim().is_empty()) {
return CheckResult::fail(
NAME,
"an artifact was emitted with an empty id; appends are matched by \
id, so an empty one makes every later chunk ambiguous",
);
}
CheckResult::pass(
NAME,
format!("{} artifact event(s), all identified", artifacts.len()),
)
}
pub(in crate::conformance) fn parking_is_not_an_error(&self) -> CheckResult {
const NAME: &str = "parking_is_not_an_error";
let parked = states(&self.events).iter().any(|s| s.is_interrupted());
if !parked {
return CheckResult::skip(NAME, "the executor never parked");
}
match &self.outcome {
RunOutcome::Ok => CheckResult::pass(NAME, "parked and returned Ok"),
RunOutcome::Err(e) => CheckResult::fail(
NAME,
format!(
"parked but returned Err({e}); waiting for the caller is a \
normal outcome, and the handler will overwrite the parked \
state with Failed"
),
),
RunOutcome::Panicked => CheckResult::fail(NAME, "panicked after parking"),
RunOutcome::TimedOut => {
CheckResult::fail(NAME, "parked but never returned; the turn never ends")
}
}
}
pub(in crate::conformance) fn honours_cancellation(&self) -> CheckResult {
const NAME: &str = "honours_cancellation";
if matches!(self.outcome, RunOutcome::TimedOut) {
return CheckResult::fail(
NAME,
format!(
"still running {RUN_TIMEOUT:?} after being started with an \
already-cancelled token; it never observed the token at all"
),
);
}
let seen = states(&self.events);
if seen.contains(&TaskState::Completed) {
return CheckResult::fail(
NAME,
"ran to Completed with an already-cancelled token; cancellation is \
cooperative, so an executor that never checks \
ctx.cancellation_token cannot be cancelled at all",
);
}
CheckResult::pass(
NAME,
seen.last().map_or_else(
|| "emitted nothing and stopped".to_owned(),
|s| format!("stopped at {s}"),
),
)
}
}