polyc-a2a 2026.8.3

//! The durable task store behind `GetTask`/`CancelTask`/`ListTasks`.
//!
//! An A2A peer is handed a task id and comes back for it later — possibly much
//! later, and possibly to a different replica of this edge. That record has to
//! outlive both the turn and the process, so it lives in the state plane and
//! this edge reaches it through the control plane's `AgentTaskService`
//! ([`TaskDialerStore`]), exactly as it reaches a turn through `AgentService`
//! and an approval decision through `ApprovalService`.
//!
//! [`TaskStore`] is the seam. It is written in the terms the JSON-RPC surface
//! already speaks — [`Task`], [`Message`], [`Artifact`] — and the dialer
//! translates them into the opaque frames the record carries.
//!
//! # The record says that a message happened, never what it said
//!
//! This family holds lifecycle facts: identifiers, roles, timestamps, and
//! references. A history frame carries a message's id, its author, and the
//! task and context it belongs to — and no content parts at all. An artifact
//! frame carries its id and its name on the same terms. The conversation
//! journal is the sole authority for what was actually said, and a second copy
//! here would make two families authoritative for the same bytes. So a later
//! `GetTask` answers an honest skeleton of what happened rather than a
//! truncated or fabricated copy of it.
//!
//! Metadata is the one free-form field the record keeps, and it keeps it for
//! the same reason: it carries the reference a paused task needs to resolve
//! its approval gate, not transcript.
//!
//! # Content is opaque to everything behind this seam
//!
//! The control plane and the state plane store and return the status detail,
//! artifacts, history, and metadata as bytes. Neither grows a copy of the A2A
//! wire union, so this module owns both directions of that translation: JSON
//! in, JSON out, one frame per message or artifact.
//!
//! # What a store failure means
//!
//! Every operation returns a [`TaskStoreError`] and never a silent default. A
//! store that cannot be reached is [`TaskStoreError::Unavailable`], and the
//! JSON-RPC surface turns it into an error response — a peer learns the task
//! could not be read, instead of being told it does not exist.

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

use chrono::{DateTime, SecondsFormat};
use polyc_rpc_client::{
    AgentTaskRecord, AgentTaskState, AgentTaskTransition, DialError, ErrorCode, TaskDialer,
};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::types::{Artifact, Message, Role, Task, TaskState, TaskStatus};

/// The longest identifier the durable record accepts, mirroring the state
/// plane's own identifier bound.
///
/// A peer chooses its own `messageId`, `artifactId`, and `taskId`, so this is
/// the bound that keeps a peer-supplied identifier from growing a frame the
/// record would refuse.
pub const MAX_ID_BYTES: usize = 256;

/// The largest metadata a task record accepts, mirroring the state plane's own
/// bound.
///
/// Metadata carries the reference a paused task resolves its approval gate
/// through, so an oversized one is refused rather than shortened — dropping
/// the reference would lose the way to answer the gate.
pub const MAX_METADATA_BYTES: usize = 8 * 1024;

/// The ceiling on the `pageSize` a peer may ask `ListTasks` for. A larger
/// request is clamped to this rather than refused.
///
/// Clamping to it is the first of two bounds and rarely the one that binds:
/// the durable read asks for at most a hundred tasks (`DURABLE_PAGE_SIZE`) and
/// answers with fewer when the page reaches the state plane's byte budget. So
/// a peer asking for a thousand gets at most a hundred, and often fewer. Every
/// page that stops short carries the token to resume from, so the tasks past
/// it are a page away rather than lost.
pub const MAX_PAGE_SIZE: usize = 1000;

/// The most tasks one durable read asks for.
///
/// The state plane's own page ceiling (`polyc_state::tasks::MAX_PAGE_SIZE`),
/// which is what one multi-key read answers. It is a count, and a count cannot
/// bound a reply on its own — a record's size is a peer's choice — so the
/// durable read also fills the page against `polyc_state::tasks::MAX_PAGE_BYTES`
/// and returns a cursor when it stops there. A context of long-lived tasks
/// therefore lists as a run of short pages rather than as one reply too large
/// to decode.
const DURABLE_PAGE_SIZE: usize = 100;

/// A boxed future one [`TaskStore`] operation resolves to. Object-safe, so the
/// live dialer and the test double are interchangeable behind one `Arc<dyn>` —
/// the same shape [`crate::task::TurnRunner`] uses for the same reason.
type StoreFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, TaskStoreError>> + Send + 'a>>;

/// Why one task operation could not be applied.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum TaskStoreError {
    /// No task exists under the named id.
    #[error("no task exists under this id")]
    NotFound,
    /// A task already exists under the id a create asked to mint.
    #[error("a task already exists under this id")]
    AlreadyExists,
    /// The task already finished, so it can neither move nor be canceled.
    #[error("this task already finished")]
    Terminal,
    /// A turn is already running under this task, so a second one cannot claim
    /// it. Not [`Self::Terminal`]: the task has not finished.
    #[error("this task is already running")]
    Running,
    /// The store refused the record itself — a bound, or a malformed request.
    #[error("{0}")]
    Refused(String),
    /// The store could not be reached at all.
    #[error("{0}")]
    Unavailable(String),
}

/// One page of a context's tasks.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct TaskPage {
    /// The page's tasks, in id order.
    pub tasks: Vec<Task>,
    /// The opaque token a peer passes back as `pageToken` for the next page;
    /// `None` when this page reached the end of the context's tasks.
    pub next_page_token: Option<String>,
}

/// The successor one [`TaskStore::transition`] call records.
///
/// Status, artifacts, and metadata REPLACE what the stored task carries;
/// history is extended by [`Self::appended_history`] and never rewritten —
/// which is what makes a transition safe to compose from a record this process
/// never read.
#[derive(Debug, Clone)]
pub struct TaskUpdate<'a> {
    /// The status the task carries after this transition.
    pub status: &'a TaskStatus,
    /// The artifacts it carries after this transition.
    pub artifacts: Option<&'a Vec<Artifact>>,
    /// The metadata it carries after this transition.
    pub metadata: Option<&'a HashMap<String, Value>>,
    /// The messages this transition appends to the history, in order.
    pub appended_history: &'a [Message],
}

/// Exact State-issued ownership echoed on every claimed task update.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TaskOwnership {
    /// Stable dispatch identity returned by durable ingress.
    pub dispatch_id: String,
    /// Stable identity of this edge process.
    pub worker_instance: String,
    /// Stable identity of this delivery attempt.
    pub attempt_id: String,
    /// State-wide fencing token issued to this attempt.
    pub fence: u64,
    /// Server-derived digest binding the history attached by the claim.
    pub claim_digest: Vec<u8>,
}

/// The durable task lifecycle this edge drives.
///
/// Object-safe (the server holds an `Arc<dyn TaskStore>`) so the live dialer
/// and the in-memory double are interchangeable, mirroring
/// [`crate::task::TurnRunner`] and [`crate::task::ApprovalResponder`].
pub trait TaskStore: Send + Sync {
    /// Mint `task` in `submitted` and index it under its context.
    ///
    /// The task's own `history` is its opening frames; its status, artifacts,
    /// and metadata are not recorded — a new task carries none by
    /// construction, and a first transition is what gives it any.
    fn create<'a>(&'a self, task: &'a Task) -> StoreFuture<'a, ()>;

    /// Claim the task named by `task_id` for one source dispatch, moving it
    /// from `submitted` or an interrupted state to `working` and appending
    /// `appended_history`.
    ///
    /// The claim is what separates a task a turn is running from a task that
    /// was recorded and never dispatched — nothing else is written between the
    /// two — so exactly one caller may win it. A task that is neither
    /// `submitted` nor interrupted refuses the claim with
    /// [`TaskStoreError::Running`]. Exact retries recover the same durable
    /// capability; independent attempts never share one.
    ///
    /// `appended_history` carries the message that drove this dispatch when the
    /// record does not already hold it, so a reader can tell which message the
    /// turn ran.
    ///
    /// `dispatch_id` is the D1 source-event identity, never the task id. One
    /// task may span several continuation messages and receives a newer
    /// per-task fence for each distinct dispatch.
    fn claim<'a>(
        &'a self,
        task_id: &'a str,
        appended_history: &'a [Message],
        dispatch_id: &'a str,
    ) -> StoreFuture<'a, TaskOwnership>;

    /// Move the task named by `task_id` to `update`'s successor.
    fn transition<'a>(
        &'a self,
        task_id: &'a str,
        update: TaskUpdate<'a>,
        ownership: Option<&'a TaskOwnership>,
    ) -> StoreFuture<'a, ()>;

    /// Renew one exact ownership capability under a monotonic ordinal.
    fn renew<'a>(
        &'a self,
        task_id: &'a str,
        ownership: &'a TaskOwnership,
        ordinal: u64,
    ) -> StoreFuture<'a, ()>;

    /// Cancel the task named by `task_id`, returning it as it now stands.
    fn cancel<'a>(&'a self, task_id: &'a str) -> StoreFuture<'a, Task>;

    /// Read one task by id. `Ok(None)` is a plain absence.
    fn get<'a>(&'a self, task_id: &'a str) -> StoreFuture<'a, Option<Task>>;

    /// Read one page of `context_id`'s tasks, resuming after `page_token`.
    fn list<'a>(
        &'a self,
        context_id: &'a str,
        page_size: usize,
        page_token: Option<&'a str>,
    ) -> StoreFuture<'a, TaskPage>;
}

/// The lifecycle of one message, as one opaque frame.
///
/// Every field here answers "which message, in which task, by whom". The
/// content parts are deliberately absent: the conversation journal records
/// what was said.
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct MessageFrame {
    message_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    context_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    task_id: Option<String>,
    role: Role,
}

impl From<&Message> for MessageFrame {
    fn from(message: &Message) -> Self {
        Self {
            message_id: message.message_id.clone(),
            context_id: message.context_id.clone(),
            task_id: message.task_id.clone(),
            role: message.role,
        }
    }
}

impl From<MessageFrame> for Message {
    fn from(frame: MessageFrame) -> Self {
        Self {
            message_id: frame.message_id,
            context_id: frame.context_id,
            task_id: frame.task_id,
            role: frame.role,
            parts: Vec::new(),
            metadata: None,
        }
    }
}

/// The lifecycle of one artifact, as one opaque frame — which artifact, and
/// what it is called.
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ArtifactFrame {
    artifact_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    name: Option<String>,
}

impl From<&Artifact> for ArtifactFrame {
    fn from(artifact: &Artifact) -> Self {
        Self {
            artifact_id: artifact.artifact_id.clone(),
            name: artifact.name.clone(),
        }
    }
}

impl From<ArtifactFrame> for Artifact {
    fn from(frame: ArtifactFrame) -> Self {
        Self {
            artifact_id: frame.artifact_id,
            name: frame.name,
            description: None,
            parts: Vec::new(),
            metadata: None,
        }
    }
}

/// The message and timestamp half of a [`TaskStatus`], as one opaque frame.
///
/// The lifecycle state itself is NOT in here: the durable record types it, and
/// carrying it twice would let the two copies disagree about what a task is
/// doing.
#[derive(Debug, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct StatusDetailFrame {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    message: Option<MessageFrame>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    timestamp: Option<String>,
}

/// Refuse a peer-chosen string longer than the record accepts, naming the
/// bound and what was sent so a peer can fix it.
///
/// Every string a frame carries comes from a peer, so every one of them is
/// bounded here rather than left to be refused somewhere further in. An
/// identifier that only the state plane would have rejected is a refusal a
/// peer reads two layers from the field it names.
fn bounded_id(what: &str, value: &str) -> Result<(), TaskStoreError> {
    if value.len() > MAX_ID_BYTES {
        return Err(TaskStoreError::Refused(format!(
            "a task record holds at most {MAX_ID_BYTES} bytes of {what}, and this one is {} bytes",
            value.len()
        )));
    }
    Ok(())
}

/// Refuse any of a frame's optional peer-chosen strings that is too long.
fn bounded_optional(what: &str, value: Option<&String>) -> Result<(), TaskStoreError> {
    value.map_or(Ok(()), |value| bounded_id(what, value))
}

/// Encode one frame, turning an unencodable value into a refusal rather than
/// an empty frame the record would reject for a reason that names nothing.
fn encode<T: Serialize>(what: &str, value: &T) -> Result<Vec<u8>, TaskStoreError> {
    serde_json::to_vec(value).map_err(|err| {
        TaskStoreError::Refused(format!("this task's {what} could not be saved: {err}"))
    })
}

/// Encode one message as a history frame.
fn message_frame(message: &Message) -> Result<Vec<u8>, TaskStoreError> {
    bounded_id("message id", &message.message_id)?;
    bounded_optional("context id", message.context_id.as_ref())?;
    bounded_optional("task id", message.task_id.as_ref())?;
    encode("message", &MessageFrame::from(message))
}

/// Encode one artifact frame.
fn artifact_frame(artifact: &Artifact) -> Result<Vec<u8>, TaskStoreError> {
    bounded_id("artifact id", &artifact.artifact_id)?;
    // Free text rather than an identifier, and bounded on the same terms: a
    // frame that carries an unbounded name is a frame a peer sizes.
    bounded_optional("artifact name", artifact.name.as_ref())?;
    encode("artifact", &ArtifactFrame::from(artifact))
}

/// Encode a status's message and timestamp. Empty when it carries neither.
fn status_detail_frame(status: &TaskStatus) -> Result<Vec<u8>, TaskStoreError> {
    let Some(detail) = status_detail(status)? else {
        return Ok(Vec::new());
    };
    encode("status", &detail)
}

/// The status detail a record keeps, or `None` when the status carries none.
fn status_detail(status: &TaskStatus) -> Result<Option<StatusDetailFrame>, TaskStoreError> {
    if status.message.is_none() && status.timestamp.is_none() {
        return Ok(None);
    }
    if let Some(message) = status.message.as_ref() {
        bounded_id("message id", &message.message_id)?;
        bounded_optional("context id", message.context_id.as_ref())?;
        bounded_optional("task id", message.task_id.as_ref())?;
    }
    Ok(Some(StatusDetailFrame {
        message: status.message.as_ref().map(MessageFrame::from),
        timestamp: status.timestamp.clone(),
    }))
}

/// Encode a task's metadata. Empty when it carries none.
fn metadata_frame(metadata: Option<&HashMap<String, Value>>) -> Result<Vec<u8>, TaskStoreError> {
    let Some(metadata) = metadata.filter(|metadata| !metadata.is_empty()) else {
        return Ok(Vec::new());
    };
    let frame = encode("metadata", metadata)?;
    if frame.len() > MAX_METADATA_BYTES {
        return Err(TaskStoreError::Refused(format!(
            "a task record holds metadata of at most {MAX_METADATA_BYTES} bytes, and this one is \
             {} bytes",
            frame.len()
        )));
    }
    Ok(frame)
}

/// Map the durable lifecycle state onto the A2A one.
const fn from_record_state(state: AgentTaskState) -> TaskState {
    match state {
        AgentTaskState::Submitted => TaskState::Submitted,
        AgentTaskState::Working => TaskState::Working,
        AgentTaskState::InputRequired => TaskState::InputRequired,
        AgentTaskState::AuthRequired => TaskState::AuthRequired,
        AgentTaskState::Completed => TaskState::Completed,
        AgentTaskState::Failed => TaskState::Failed,
        AgentTaskState::Canceled => TaskState::Canceled,
        AgentTaskState::Rejected => TaskState::Rejected,
    }
}

/// Map the A2A lifecycle state onto the durable one.
///
/// `Unspecified` has no durable counterpart — a task always moves to a state
/// it names — so it is refused rather than guessed at.
const fn to_record_state(state: TaskState) -> Option<AgentTaskState> {
    match state {
        TaskState::Submitted => Some(AgentTaskState::Submitted),
        TaskState::Working => Some(AgentTaskState::Working),
        TaskState::InputRequired => Some(AgentTaskState::InputRequired),
        TaskState::AuthRequired => Some(AgentTaskState::AuthRequired),
        TaskState::Completed => Some(AgentTaskState::Completed),
        TaskState::Failed => Some(AgentTaskState::Failed),
        TaskState::Canceled => Some(AgentTaskState::Canceled),
        TaskState::Rejected => Some(AgentTaskState::Rejected),
        TaskState::Unspecified => None,
    }
}

/// Render a record's last-transition time as the RFC3339 stamp an A2A status
/// carries. `None` for a record that names no time at all, which is the one
/// honest answer when there is nothing to report.
fn transition_timestamp(updated_at_ms: u64) -> Option<String> {
    if updated_at_ms == 0 {
        return None;
    }
    let millis = i64::try_from(updated_at_ms).ok()?;
    DateTime::from_timestamp_millis(millis)
        .map(|at| at.to_rfc3339_opts(SecondsFormat::Millis, true))
}

/// Rebuild the A2A task one durable record stands for.
///
/// Names every field of both sides explicitly. A frame that no longer decodes
/// is dropped rather than failing the whole read: the record's identity and
/// lifecycle state are what a peer polls for, and those are typed.
///
/// # The status always carries a time
///
/// The status frame is where a transition puts a time it *states*, and a claim
/// states none — it writes an empty frame, because the only thing it changes is
/// the lifecycle state. Reading the timestamp out of that frame alone therefore
/// answered `GetTask` on a claimed task with a bare `working` and no time, at
/// minute one and at day thirty alike. That is precisely the task a peer is
/// told to poll for and then judge: a claimed task whose run died stays
/// `working` forever, and the refusal copy tells the peer to follow it with
/// `GetTask` and start over under a new id once it clearly will not finish.
/// With no time on either answer, the peer decides between abandoning live work
/// and double-running dead work by guessing, while an operator reads the same
/// record's `updated_at_ms` and knows.
///
/// So the record's own `updated_at_ms` — the time the most recent accepted
/// transition landed, which every transition including a claim stamps — fills
/// in whenever the frame states nothing. A stated time still wins: it is what
/// the transition asserted about itself, and the record's is when the plane
/// accepted it.
fn task_from_record(record: &AgentTaskRecord) -> Task {
    let detail: StatusDetailFrame = if record.status_detail.is_empty() {
        StatusDetailFrame::default()
    } else {
        serde_json::from_slice(&record.status_detail).unwrap_or_default()
    };
    let artifacts: Vec<Artifact> = record
        .artifacts
        .iter()
        .filter_map(|frame| serde_json::from_slice::<ArtifactFrame>(frame).ok())
        .map(Artifact::from)
        .collect();
    let history: Vec<Message> = record
        .history
        .iter()
        .filter_map(|frame| serde_json::from_slice::<MessageFrame>(frame).ok())
        .map(Message::from)
        .collect();
    let metadata: Option<HashMap<String, Value>> = if record.metadata.is_empty() {
        None
    } else {
        serde_json::from_slice(&record.metadata).ok()
    };
    Task {
        id: record.task_id.clone(),
        context_id: record.context_id.clone(),
        status: TaskStatus {
            state: from_record_state(record.state),
            message: detail.message.map(Message::from),
            timestamp: detail
                .timestamp
                .or_else(|| transition_timestamp(record.updated_at_ms)),
        },
        artifacts: (!artifacts.is_empty()).then_some(artifacts),
        history: (!history.is_empty()).then_some(history),
        metadata,
    }
}

/// Map a dial failure onto the outcome the JSON-RPC surface renders.
///
/// The refusals the control plane states in its codes stay refusals; anything
/// else is an outage, so a peer is never told a task does not exist when the
/// truth is that nothing could be reached.
fn from_dial_error(error: &DialError) -> TaskStoreError {
    match error.code() {
        Some(ErrorCode::NotFound) => TaskStoreError::NotFound,
        Some(ErrorCode::AlreadyExists) => TaskStoreError::AlreadyExists,
        Some(ErrorCode::FailedPrecondition) => TaskStoreError::Terminal,
        // Losing a concurrency check, which for this family means exactly one
        // thing: a turn already claimed the task. Mapped explicitly so it
        // never falls through to the outage arm and reads as something a
        // retry could fix.
        Some(ErrorCode::Aborted) => TaskStoreError::Running,
        Some(ErrorCode::InvalidArgument) => TaskStoreError::Refused(error.to_string()),
        _ => TaskStoreError::Unavailable(error.to_string()),
    }
}

/// A [`TaskStore`] backed by the live control plane's `AgentTaskService`.
///
/// A thin dialer, like [`crate::task::AgentDialerRunner`] and
/// [`crate::task::ApprovalDialerResponder`] beside it: this edge holds no
/// state-plane client of its own and never could — it reaches durable state
/// only through the control plane.
#[derive(Clone)]
pub struct TaskDialerStore {
    dialer: TaskDialer,
    worker_instance: String,
}

impl TaskDialerStore {
    /// Build a store dialing the control plane at `addr` (`http://host:port`)
    /// — the SAME endpoint the turn and approval dialers use.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` is not a valid URI.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        Ok(Self {
            dialer: TaskDialer::new(addr)?,
            worker_instance: uuid::Uuid::now_v7().to_string(),
        })
    }

    /// Build a store dialing the control plane at `addr`, authenticated with
    /// `bearer` — every call carries the edge's bearer header. No envelope is
    /// signed: `AgentTaskService` calls carry no `AgentStart`.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` is not a valid URI, or
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value.
    pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            dialer: TaskDialer::with_bearer(addr, bearer)?,
            worker_instance: uuid::Uuid::now_v7().to_string(),
        })
    }
}

impl TaskStore for TaskDialerStore {
    fn create<'a>(&'a self, task: &'a Task) -> StoreFuture<'a, ()> {
        Box::pin(async move {
            let history = task
                .history
                .as_deref()
                .unwrap_or_default()
                .iter()
                .map(message_frame)
                .collect::<Result<Vec<_>, _>>()?;
            self.dialer
                .create_task(&task.id, &task.context_id, history)
                .await
                .map(|_| ())
                .map_err(|err| from_dial_error(&err))
        })
    }

    fn claim<'a>(
        &'a self,
        task_id: &'a str,
        appended_history: &'a [Message],
        dispatch_id: &'a str,
    ) -> StoreFuture<'a, TaskOwnership> {
        Box::pin(async move {
            let attempt_id = uuid::Uuid::now_v7().to_string();
            let (_, ownership) = self
                .dialer
                .claim_task(
                    task_id,
                    dispatch_id,
                    &self.worker_instance,
                    &attempt_id,
                    appended_history
                        .iter()
                        .map(message_frame)
                        .collect::<Result<Vec<_>, _>>()?,
                )
                .await
                .map_err(|err| from_dial_error(&err))?;
            Ok(TaskOwnership {
                dispatch_id: ownership.dispatch_id,
                worker_instance: ownership.worker_instance,
                attempt_id: ownership.attempt_id,
                fence: ownership.fence,
                claim_digest: ownership.claim_digest,
            })
        })
    }

    fn transition<'a>(
        &'a self,
        task_id: &'a str,
        update: TaskUpdate<'a>,
        ownership: Option<&'a TaskOwnership>,
    ) -> StoreFuture<'a, ()> {
        Box::pin(async move {
            let ownership = ownership.ok_or_else(|| {
                TaskStoreError::Refused(
                    "a claimed task update carries no State ownership".to_owned(),
                )
            })?;
            let state = to_record_state(update.status.state).ok_or_else(|| {
                TaskStoreError::Refused(
                    "this task cannot move to a state it does not name".to_owned(),
                )
            })?;
            let successor = AgentTaskTransition {
                status_detail: status_detail_frame(update.status)?,
                artifacts: update
                    .artifacts
                    .map(|artifacts| {
                        artifacts
                            .iter()
                            .map(artifact_frame)
                            .collect::<Result<Vec<_>, _>>()
                    })
                    .transpose()?
                    .unwrap_or_default(),
                metadata: metadata_frame(update.metadata)?,
                appended_history: update
                    .appended_history
                    .iter()
                    .map(message_frame)
                    .collect::<Result<Vec<_>, _>>()?,
            };
            self.dialer
                .transition_task(
                    task_id,
                    state,
                    successor,
                    &polyc_rpc_client::AgentTaskOwnership {
                        dispatch_id: ownership.dispatch_id.clone(),
                        worker_instance: ownership.worker_instance.clone(),
                        attempt_id: ownership.attempt_id.clone(),
                        fence: ownership.fence,
                        claim_digest: ownership.claim_digest.clone(),
                    },
                )
                .await
                .map(|_| ())
                .map_err(|err| from_dial_error(&err))
        })
    }

    fn renew<'a>(
        &'a self,
        task_id: &'a str,
        ownership: &'a TaskOwnership,
        ordinal: u64,
    ) -> StoreFuture<'a, ()> {
        Box::pin(async move {
            self.dialer
                .renew_task_claim(
                    task_id,
                    &polyc_rpc_client::AgentTaskOwnership {
                        dispatch_id: ownership.dispatch_id.clone(),
                        worker_instance: ownership.worker_instance.clone(),
                        attempt_id: ownership.attempt_id.clone(),
                        fence: ownership.fence,
                        claim_digest: ownership.claim_digest.clone(),
                    },
                    ordinal,
                )
                .await
                .map_err(|err| from_dial_error(&err))
        })
    }

    fn cancel<'a>(&'a self, task_id: &'a str) -> StoreFuture<'a, Task> {
        Box::pin(async move {
            self.dialer
                .cancel_task(task_id)
                .await
                .map(|record| task_from_record(&record))
                .map_err(|err| from_dial_error(&err))
        })
    }

    fn get<'a>(&'a self, task_id: &'a str) -> StoreFuture<'a, Option<Task>> {
        Box::pin(async move {
            self.dialer
                .get_task(task_id)
                .await
                .map(|record| record.as_ref().map(task_from_record))
                .map_err(|err| from_dial_error(&err))
        })
    }

    fn list<'a>(
        &'a self,
        context_id: &'a str,
        page_size: usize,
        page_token: Option<&'a str>,
    ) -> StoreFuture<'a, TaskPage> {
        Box::pin(async move {
            let page_size = u32::try_from(page_size.clamp(1, DURABLE_PAGE_SIZE)).unwrap_or(1);
            self.dialer
                .list_tasks(context_id, page_size, page_token)
                .await
                .map(|page| TaskPage {
                    tasks: page.tasks.iter().map(task_from_record).collect(),
                    next_page_token: page.next_after,
                })
                .map_err(|err| from_dial_error(&err))
        })
    }
}

/// The failure text every [`UnconfiguredTaskStore`] operation returns. Names
/// the unset variable so the cause is actionable from the error alone,
/// matching [`crate::task::UnconfiguredRunner`].
const UNCONFIGURED_STORE_MESSAGE: &str = "control-plane address is unset \
     (POLYCHROME_AGENT_ADDR); this edge serves its Agent Card but cannot record or read tasks \
     until it is set";

/// A [`TaskStore`] for an edge brought up without a control-plane address.
///
/// Mirrors [`crate::task::UnconfiguredRunner`]: the Agent Card still serves,
/// and every task operation fails closed naming the unset address rather than
/// dialing an endpoint that was never configured.
#[derive(Clone, Copy, Debug, Default)]
pub struct UnconfiguredTaskStore;

impl UnconfiguredTaskStore {
    fn refuse<T>() -> Result<T, TaskStoreError> {
        Err(TaskStoreError::Unavailable(
            UNCONFIGURED_STORE_MESSAGE.to_owned(),
        ))
    }
}

impl TaskStore for UnconfiguredTaskStore {
    fn create<'a>(&'a self, _task: &'a Task) -> StoreFuture<'a, ()> {
        Box::pin(async { Self::refuse() })
    }

    fn claim<'a>(
        &'a self,
        _task_id: &'a str,
        _appended_history: &'a [Message],
        _dispatch_id: &'a str,
    ) -> StoreFuture<'a, TaskOwnership> {
        Box::pin(async { Self::refuse() })
    }

    fn transition<'a>(
        &'a self,
        _task_id: &'a str,
        _update: TaskUpdate<'a>,
        _ownership: Option<&'a TaskOwnership>,
    ) -> StoreFuture<'a, ()> {
        Box::pin(async { Self::refuse() })
    }

    fn renew<'a>(
        &'a self,
        _task_id: &'a str,
        _ownership: &'a TaskOwnership,
        _ordinal: u64,
    ) -> StoreFuture<'a, ()> {
        Box::pin(async { Self::refuse() })
    }

    fn cancel<'a>(&'a self, _task_id: &'a str) -> StoreFuture<'a, Task> {
        Box::pin(async { Self::refuse() })
    }

    fn get<'a>(&'a self, _task_id: &'a str) -> StoreFuture<'a, Option<Task>> {
        Box::pin(async { Self::refuse() })
    }

    fn list<'a>(
        &'a self,
        _context_id: &'a str,
        _page_size: usize,
        _page_token: Option<&'a str>,
    ) -> StoreFuture<'a, TaskPage> {
        Box::pin(async { Self::refuse() })
    }
}

/// Test doubles for [`TaskStore`].
///
/// Behind the `test-util` feature, which only this crate's own tests turn on.
/// The production path has exactly one store — [`TaskDialerStore`] — and the
/// process-local map this crate used to ship as that store is gone: it forgot
/// every task on restart, and a mint and a poll routed to different replicas
/// never matched.
#[cfg(any(test, feature = "test-util"))]
pub mod test_double {
    use std::collections::HashMap;
    use std::sync::{
        RwLock,
        atomic::{AtomicU64, Ordering},
    };

    use super::{
        Artifact, ArtifactFrame, Message, MessageFrame, StoreFuture, Task, TaskOwnership, TaskPage,
        TaskStatus, TaskStore, TaskStoreError, TaskUpdate, artifact_frame, message_frame,
        metadata_frame,
    };

    /// A thread-safe in-memory task store.
    ///
    /// Enforces the SAME refusals the durable one does — a duplicate create,
    /// an unknown task, a task that already finished, a claim on a task that
    /// is neither `submitted` nor interrupted, and an identifier past the
    /// record's bound are all refused — and keeps the SAME lifecycle-only skeleton, so a case that
    /// passes here is a case about the surface rather than about which store
    /// is behind it.
    ///
    /// A test double, never a deployment's store: it forgets every task on
    /// restart, and two replicas of it never agree about a task id.
    pub struct InMemoryTaskStore {
        tasks: RwLock<HashMap<String, Task>>,
        next_fence: AtomicU64,
        /// When set, every operation fails with this outage instead of
        /// touching the map — how the fail-closed behavior of a store that
        /// cannot be reached is proven.
        outage: Option<String>,
    }

    impl Default for InMemoryTaskStore {
        fn default() -> Self {
            Self {
                tasks: RwLock::default(),
                next_fence: AtomicU64::new(1),
                outage: None,
            }
        }
    }

    /// The message a durable record keeps of `message`: its lifecycle, and no
    /// content. Runs the real encoder, so the double cannot drift from it.
    fn recorded_message(message: &Message) -> Result<Message, TaskStoreError> {
        let frame = message_frame(message)?;
        let decoded: MessageFrame = serde_json::from_slice(&frame)
            .map_err(|err| TaskStoreError::Refused(err.to_string()))?;
        Ok(Message::from(decoded))
    }

    /// The artifact a durable record keeps of `artifact`, on the same terms.
    fn recorded_artifact(artifact: &Artifact) -> Result<Artifact, TaskStoreError> {
        let frame = artifact_frame(artifact)?;
        let decoded: ArtifactFrame = serde_json::from_slice(&frame)
            .map_err(|err| TaskStoreError::Refused(err.to_string()))?;
        Ok(Artifact::from(decoded))
    }

    /// The status a durable record keeps: the typed lifecycle state, plus the
    /// message's lifecycle and the timestamp.
    fn recorded_status(status: &TaskStatus) -> Result<TaskStatus, TaskStoreError> {
        Ok(TaskStatus {
            state: status.state,
            message: status.message.as_ref().map(recorded_message).transpose()?,
            timestamp: status.timestamp.clone(),
        })
    }

    impl InMemoryTaskStore {
        /// An empty store.
        #[must_use]
        pub fn new() -> Self {
            Self::default()
        }

        /// A store that answers every operation with an outage — how the
        /// fail-closed behavior of an unreachable store is proven.
        #[must_use]
        pub fn unreachable(reason: &str) -> Self {
            Self {
                tasks: RwLock::default(),
                next_fence: AtomicU64::new(1),
                outage: Some(reason.to_owned()),
            }
        }

        fn outage<T>(&self) -> Option<Result<T, TaskStoreError>> {
            self.outage
                .as_ref()
                .map(|reason| Err(TaskStoreError::Unavailable(reason.clone())))
        }

        fn read(&self) -> std::sync::RwLockReadGuard<'_, HashMap<String, Task>> {
            self.tasks
                .read()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
        }

        fn write(&self) -> std::sync::RwLockWriteGuard<'_, HashMap<String, Task>> {
            self.tasks
                .write()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
        }
    }

    impl TaskStore for InMemoryTaskStore {
        fn create<'a>(&'a self, task: &'a Task) -> StoreFuture<'a, ()> {
            Box::pin(async move {
                if let Some(outage) = self.outage() {
                    return outage;
                }
                let history = task
                    .history
                    .as_deref()
                    .unwrap_or_default()
                    .iter()
                    .map(recorded_message)
                    .collect::<Result<Vec<_>, _>>()?;
                let mut tasks = self.write();
                {
                    if tasks.contains_key(&task.id) {
                        return Err(TaskStoreError::AlreadyExists);
                    }
                    let created = Task {
                        id: task.id.clone(),
                        context_id: task.context_id.clone(),
                        status: TaskStatus {
                            state: crate::types::TaskState::Submitted,
                            message: None,
                            timestamp: None,
                        },
                        artifacts: None,
                        history: Some(history),
                        metadata: None,
                    };
                    tasks.insert(created.id.clone(), created);
                }
                drop(tasks);
                Ok(())
            })
        }

        fn claim<'a>(
            &'a self,
            task_id: &'a str,
            appended_history: &'a [Message],
            dispatch_id: &'a str,
        ) -> StoreFuture<'a, TaskOwnership> {
            Box::pin(async move {
                if let Some(outage) = self.outage() {
                    return outage;
                }
                let appended = appended_history
                    .iter()
                    .map(recorded_message)
                    .collect::<Result<Vec<_>, _>>()?;
                let mut tasks = self.write();
                {
                    let task = tasks.get_mut(task_id).ok_or(TaskStoreError::NotFound)?;
                    // The SAME lifecycle rule the durable authority applies:
                    // a new source dispatch may claim a submitted task or
                    // continue one that is waiting on input/authentication.
                    if task.status.state != crate::types::TaskState::Submitted
                        && !matches!(
                            task.status.state,
                            crate::types::TaskState::InputRequired
                                | crate::types::TaskState::AuthRequired
                        )
                    {
                        return Err(TaskStoreError::Running);
                    }
                    task.status = TaskStatus {
                        state: crate::types::TaskState::Working,
                        message: None,
                        timestamp: None,
                    };
                    let mut history = task.history.take().unwrap_or_default();
                    history.extend(appended);
                    task.history = Some(history);
                }
                drop(tasks);
                Ok(TaskOwnership {
                    dispatch_id: dispatch_id.to_owned(),
                    worker_instance: "in-memory-worker".to_owned(),
                    attempt_id: uuid::Uuid::now_v7().to_string(),
                    fence: self.next_fence.fetch_add(1, Ordering::Relaxed),
                    claim_digest: vec![0; 32],
                })
            })
        }

        fn transition<'a>(
            &'a self,
            task_id: &'a str,
            update: TaskUpdate<'a>,
            _ownership: Option<&'a TaskOwnership>,
        ) -> StoreFuture<'a, ()> {
            Box::pin(async move {
                if let Some(outage) = self.outage() {
                    return outage;
                }
                let status = recorded_status(update.status)?;
                let artifacts = update
                    .artifacts
                    .map(|artifacts| {
                        artifacts
                            .iter()
                            .map(recorded_artifact)
                            .collect::<Result<Vec<_>, _>>()
                    })
                    .transpose()?;
                let metadata = metadata_frame(update.metadata)?;
                let appended = update
                    .appended_history
                    .iter()
                    .map(recorded_message)
                    .collect::<Result<Vec<_>, _>>()?;
                let mut tasks = self.write();
                {
                    let task = tasks.get_mut(task_id).ok_or(TaskStoreError::NotFound)?;
                    if task.status.state.is_terminal() {
                        return Err(TaskStoreError::Terminal);
                    }
                    task.status = status;
                    task.artifacts = artifacts;
                    task.metadata = (!metadata.is_empty())
                        .then(|| update.metadata.cloned())
                        .flatten();
                    let mut history = task.history.take().unwrap_or_default();
                    history.extend(appended);
                    task.history = Some(history);
                }
                drop(tasks);
                Ok(())
            })
        }

        fn renew<'a>(
            &'a self,
            _task_id: &'a str,
            _ownership: &'a TaskOwnership,
            _ordinal: u64,
        ) -> StoreFuture<'a, ()> {
            Box::pin(async { Ok(()) })
        }

        fn cancel<'a>(&'a self, task_id: &'a str) -> StoreFuture<'a, Task> {
            Box::pin(async move {
                if let Some(outage) = self.outage() {
                    return outage;
                }
                let mut tasks = self.write();
                let canceled = {
                    let task = tasks.get_mut(task_id).ok_or(TaskStoreError::NotFound)?;
                    if task.status.state.is_terminal() {
                        return Err(TaskStoreError::Terminal);
                    }
                    task.status = TaskStatus {
                        state: crate::types::TaskState::Canceled,
                        message: None,
                        timestamp: None,
                    };
                    task.clone()
                };
                drop(tasks);
                Ok(canceled)
            })
        }

        fn get<'a>(&'a self, task_id: &'a str) -> StoreFuture<'a, Option<Task>> {
            Box::pin(async move {
                if let Some(outage) = self.outage() {
                    return outage;
                }
                Ok(self.read().get(task_id).cloned())
            })
        }

        fn list<'a>(
            &'a self,
            context_id: &'a str,
            page_size: usize,
            page_token: Option<&'a str>,
        ) -> StoreFuture<'a, TaskPage> {
            Box::pin(async move {
                if let Some(outage) = self.outage() {
                    return outage;
                }
                let mut all: Vec<Task> = self
                    .read()
                    .values()
                    .filter(|task| task.context_id == context_id)
                    .filter(|task| page_token.is_none_or(|after| task.id.as_str() > after))
                    .cloned()
                    .collect();
                all.sort_by(|a, b| a.id.cmp(&b.id));
                let has_more = all.len() > page_size;
                all.truncate(page_size);
                let next_page_token = has_more.then(|| all.last().map(|t| t.id.clone())).flatten();
                Ok(TaskPage {
                    tasks: all,
                    next_page_token,
                })
            })
        }
    }
}

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

    use super::*;
    use crate::types::Part;
    use test_double::InMemoryTaskStore;

    fn message(text: &str) -> Message {
        Message {
            message_id: "m1".to_owned(),
            context_id: Some("ctx-1".to_owned()),
            task_id: Some("t1".to_owned()),
            role: Role::Agent,
            parts: vec![Part::text(text)],
            metadata: None,
        }
    }

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

    /// A history frame keeps everything that says WHICH message this was, and
    /// none of what it said — the conversation journal is the authority for
    /// that, and a second copy here would make two families authoritative for
    /// the same bytes.
    #[test]
    fn a_message_frame_keeps_its_lifecycle_and_drops_its_content() {
        let frame = message_frame(&message("hello")).expect("encodes");
        let decoded =
            Message::from(serde_json::from_slice::<MessageFrame>(&frame).expect("decodes"));
        assert_eq!(decoded.message_id, "m1");
        assert_eq!(decoded.context_id.as_deref(), Some("ctx-1"));
        assert_eq!(decoded.task_id.as_deref(), Some("t1"));
        assert_eq!(decoded.role, Role::Agent);
        assert!(decoded.parts.is_empty(), "no content is recorded");
        assert!(
            !String::from_utf8_lossy(&frame).contains("hello"),
            "the text must not reach the record: {}",
            String::from_utf8_lossy(&frame)
        );
    }

    /// An artifact frame is the same bargain: which artifact, and what it is
    /// called.
    #[test]
    fn an_artifact_frame_keeps_its_identity_and_drops_its_content() {
        let artifact = Artifact {
            artifact_id: "a1".to_owned(),
            name: Some("answer".to_owned()),
            description: Some("the long description".to_owned()),
            parts: vec![Part::text("the answer text")],
            metadata: None,
        };
        let frame = artifact_frame(&artifact).expect("encodes");
        let decoded =
            Artifact::from(serde_json::from_slice::<ArtifactFrame>(&frame).expect("decodes"));
        assert_eq!(decoded.artifact_id, "a1");
        assert_eq!(decoded.name.as_deref(), Some("answer"));
        assert!(decoded.parts.is_empty());
        let raw = String::from_utf8_lossy(&frame).into_owned();
        assert!(
            !raw.contains("the answer text"),
            "content is not stored: {raw}"
        );
        assert!(
            !raw.contains("the long description"),
            "content is not stored: {raw}"
        );
    }

    /// A megabyte of text costs the record nothing, because the record never
    /// holds it. This is the whole boundary: no bound to tune, no truncation
    /// note to read back, no second authority for the transcript.
    #[tokio::test]
    async fn a_large_payload_is_not_stored_in_the_family() {
        let huge = "x".repeat(1024 * 1024);
        let frame = message_frame(&message(&huge)).expect("encodes");
        assert!(
            frame.len() < 256,
            "a lifecycle frame is a handful of identifiers: {} bytes",
            frame.len()
        );

        let store = InMemoryTaskStore::new();
        let mut opening = task("t1", "c1");
        opening.history = Some(vec![message(&huge)]);
        store.create(&opening).await.expect("creates");

        let stored = store.get("t1").await.expect("reads").expect("exists");
        let history = stored.history.expect("the frame is kept");
        assert_eq!(
            history.len(),
            1,
            "the message is recorded as having happened"
        );
        assert_eq!(history[0].message_id, "m1");
        assert!(
            history[0].parts.is_empty(),
            "and its content is not: {:?}",
            history[0].parts
        );
    }

    /// A peer chooses its own `messageId`. An unbounded one is refused before
    /// anything is written, rather than composing a record the state plane
    /// would reject for a reason that names nothing the peer can act on.
    #[tokio::test]
    async fn an_oversized_message_id_is_refused_rather_than_recorded() {
        let mut long = message("hello");
        long.message_id = "m".repeat(MAX_ID_BYTES + 1);

        match message_frame(&long) {
            Err(TaskStoreError::Refused(reason)) => {
                assert!(
                    reason.contains("message id") && reason.contains(&MAX_ID_BYTES.to_string()),
                    "the refusal must name the bound: {reason}"
                );
            }
            other => panic!("an oversized message id must be refused, got {other:?}"),
        }

        let store = InMemoryTaskStore::new();
        let mut opening = task("t1", "c1");
        opening.history = Some(vec![long]);
        assert!(
            matches!(
                store.create(&opening).await,
                Err(TaskStoreError::Refused(_))
            ),
            "and the create is refused with it"
        );
        assert!(
            store.get("t1").await.expect("reads").is_none(),
            "nothing is recorded under a refused create"
        );
    }

    /// Metadata carries the reference a paused task resolves its gate through,
    /// so an oversized one is refused rather than shortened.
    #[test]
    fn oversized_metadata_is_refused() {
        let mut metadata = HashMap::new();
        metadata.insert(
            "pendingApprovalToolName".to_owned(),
            Value::from("x".repeat(MAX_METADATA_BYTES + 1)),
        );
        assert!(matches!(
            metadata_frame(Some(&metadata)),
            Err(TaskStoreError::Refused(_))
        ));
    }

    /// A status with neither a message nor a timestamp records no detail at
    /// all — an empty frame, not an encoded empty object.
    #[test]
    fn an_empty_status_records_no_detail() {
        assert!(
            status_detail_frame(&TaskStatus::default())
                .expect("encodes")
                .is_empty()
        );
    }

    /// The durable record's typed state is what a rebuilt task reports, and
    /// its frames fill in which message and which artifact — never their text.
    #[test]
    fn a_record_rebuilds_the_task_it_stands_for() {
        let status = TaskStatus {
            state: TaskState::Completed,
            message: Some(message("42")),
            timestamp: None,
        };
        let record = AgentTaskRecord {
            task_id: "t1".to_owned(),
            context_id: "ctx-1".to_owned(),
            state: AgentTaskState::Completed,
            status_detail: status_detail_frame(&status).expect("encodes"),
            artifacts: vec![
                artifact_frame(&Artifact {
                    artifact_id: "a1".to_owned(),
                    name: None,
                    description: None,
                    parts: vec![Part::text("42")],
                    metadata: None,
                })
                .expect("encodes"),
            ],
            history: vec![message_frame(&message("hello")).expect("encodes")],
            metadata: Vec::new(),
            created_at_ms: 1,
            updated_at_ms: 2,
        };
        let rebuilt = task_from_record(&record);
        assert_eq!(rebuilt.id, "t1");
        assert_eq!(rebuilt.context_id, "ctx-1");
        assert_eq!(rebuilt.status.state, TaskState::Completed);
        let status_message = rebuilt.status.message.expect("the status names a message");
        assert_eq!(status_message.message_id, "m1");
        assert!(status_message.parts.is_empty(), "no content is recorded");
        assert_eq!(rebuilt.artifacts.as_ref().map(Vec::len), Some(1));
        assert_eq!(rebuilt.history.as_ref().map(Vec::len), Some(1));
        assert!(rebuilt.metadata.is_none());
    }

    /// A claim states no time of its own, and the task it leaves behind is the
    /// one a peer is told to poll and then judge. So the record's own
    /// last-transition time answers instead of nothing at all.
    #[test]
    fn a_claim_that_states_no_time_still_reports_when_it_landed() {
        let record = AgentTaskRecord {
            task_id: "t1".to_owned(),
            context_id: "ctx-1".to_owned(),
            state: AgentTaskState::Working,
            // Exactly what `TaskDialerStore::claim` writes.
            status_detail: Vec::new(),
            artifacts: Vec::new(),
            history: Vec::new(),
            metadata: Vec::new(),
            created_at_ms: 1_767_225_600_000,
            updated_at_ms: 1_767_225_600_000,
        };
        let rebuilt = task_from_record(&record);
        assert_eq!(rebuilt.status.state, TaskState::Working);
        assert_eq!(
            rebuilt.status.timestamp.as_deref(),
            Some("2026-01-01T00:00:00.000Z"),
            "a stranded task must say when it last moved"
        );
    }

    /// A stated time is what the transition asserted about itself, so it wins
    /// over the time the plane accepted it.
    #[test]
    fn a_stated_time_wins_over_the_records_own() {
        let status = TaskStatus {
            state: TaskState::Completed,
            message: None,
            timestamp: Some("2026-02-02T02:02:02Z".to_owned()),
        };
        let record = AgentTaskRecord {
            task_id: "t1".to_owned(),
            context_id: "ctx-1".to_owned(),
            state: AgentTaskState::Completed,
            status_detail: status_detail_frame(&status).expect("encodes"),
            artifacts: Vec::new(),
            history: Vec::new(),
            metadata: Vec::new(),
            created_at_ms: 1,
            updated_at_ms: 1_767_225_600_000,
        };
        assert_eq!(
            task_from_record(&record).status.timestamp.as_deref(),
            Some("2026-02-02T02:02:02Z")
        );
    }

    /// A record naming no time is answered with none, rather than with the
    /// epoch dressed up as a transition that happened in 1970.
    #[test]
    fn a_record_naming_no_time_reports_none() {
        assert_eq!(transition_timestamp(0), None);
    }

    /// A frame that no longer decodes is dropped, not fatal: the identity and
    /// lifecycle state a peer polls for are typed on the record itself.
    #[test]
    fn an_undecodable_frame_is_dropped_rather_than_failing_the_read() {
        let record = AgentTaskRecord {
            task_id: "t1".to_owned(),
            context_id: "ctx-1".to_owned(),
            state: AgentTaskState::Failed,
            status_detail: b"not json".to_vec(),
            artifacts: Vec::new(),
            history: vec![b"not json".to_vec()],
            metadata: b"not json".to_vec(),
            created_at_ms: 1,
            updated_at_ms: 1,
        };
        let rebuilt = task_from_record(&record);
        assert_eq!(rebuilt.status.state, TaskState::Failed);
        assert!(rebuilt.status.message.is_none());
        assert!(rebuilt.history.is_none());
        assert!(rebuilt.metadata.is_none());
    }

    /// Every dialer refusal keeps its own meaning, and anything that is not a
    /// stated refusal is an outage — never a "task not found" a peer would
    /// read as final.
    #[test]
    fn dial_failures_keep_refusals_apart_from_outages() {
        let of = |code: ErrorCode| from_dial_error(&DialError::Connect(connect_error(code)));
        assert_eq!(of(ErrorCode::NotFound), TaskStoreError::NotFound);
        assert_eq!(of(ErrorCode::AlreadyExists), TaskStoreError::AlreadyExists);
        assert_eq!(of(ErrorCode::FailedPrecondition), TaskStoreError::Terminal);
        assert!(matches!(
            of(ErrorCode::InvalidArgument),
            TaskStoreError::Refused(_)
        ));
        assert!(matches!(
            of(ErrorCode::Unavailable),
            TaskStoreError::Unavailable(_)
        ));
        assert!(matches!(
            of(ErrorCode::Internal),
            TaskStoreError::Unavailable(_)
        ));
    }

    fn connect_error(code: ErrorCode) -> polyc_rpc_client::ConnectError {
        polyc_rpc_client::ConnectError::new(code, "boom")
    }

    #[tokio::test]
    async fn unconfigured_store_fails_naming_the_unset_address() {
        let err = UnconfiguredTaskStore
            .get("t1")
            .await
            .expect_err("must fail closed");
        match err {
            TaskStoreError::Unavailable(message) => assert!(
                message.contains("POLYCHROME_AGENT_ADDR"),
                "message must name the unset variable: {message}"
            ),
            other => panic!("expected an outage, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn the_double_refuses_a_duplicate_create_and_a_terminal_move() {
        let store = InMemoryTaskStore::new();
        store.create(&task("t1", "c1")).await.expect("creates");
        assert_eq!(
            store.create(&task("t1", "c1")).await,
            Err(TaskStoreError::AlreadyExists)
        );

        let done = TaskStatus {
            state: TaskState::Completed,
            message: None,
            timestamp: None,
        };
        store
            .transition(
                "t1",
                TaskUpdate {
                    status: &done,
                    artifacts: None,
                    metadata: None,
                    appended_history: &[],
                },
                None,
            )
            .await
            .expect("first move");
        assert_eq!(
            store
                .transition(
                    "t1",
                    TaskUpdate {
                        status: &done,
                        artifacts: None,
                        metadata: None,
                        appended_history: &[],
                    },
                    None,
                )
                .await,
            Err(TaskStoreError::Terminal),
            "terminal is final for a transition, not only for a cancel"
        );
        assert_eq!(store.cancel("t1").await, Err(TaskStoreError::Terminal));
        assert_eq!(store.cancel("nope").await, Err(TaskStoreError::NotFound));
    }

    #[tokio::test]
    async fn the_double_pages_one_context_by_keyset() {
        let store = InMemoryTaskStore::new();
        for id in ["t1", "t2", "t3"] {
            store.create(&task(id, "c1")).await.expect("creates");
        }
        store.create(&task("t9", "c2")).await.expect("creates");

        let first = store.list("c1", 2, None).await.expect("lists");
        assert_eq!(
            first
                .tasks
                .iter()
                .map(|t| t.id.as_str())
                .collect::<Vec<_>>(),
            ["t1", "t2"]
        );
        assert_eq!(first.next_page_token.as_deref(), Some("t2"));

        let second = store
            .list("c1", 2, first.next_page_token.as_deref())
            .await
            .expect("lists");
        assert_eq!(
            second
                .tasks
                .iter()
                .map(|t| t.id.as_str())
                .collect::<Vec<_>>(),
            ["t3"]
        );
        assert_eq!(second.next_page_token, None);
    }
}