bcode_tui 0.0.1-alpha.0

//! Generic incremental artifact transport for plugin-owned TUI visuals.

use std::collections::BTreeMap;
use std::time::{Duration, Instant};

use bcode_client::{BcodeClient, ClientError};
use bcode_plugin_sdk::tui::PluginTuiArtifactChunk;
use bcode_session_models::{SessionEventKind, SessionId};

const ACTIVE_ARTIFACT_FETCH_BYTES: u32 = 256 * 1024;
const ACTIVE_ARTIFACT_RETRY_BASE: Duration = Duration::from_millis(100);
const ACTIVE_ARTIFACT_RETRY_MAX: Duration = Duration::from_secs(2);
// Presentation hydration retries are bounded per observed target, not workflow execution.
const ACTIVE_ARTIFACT_MAX_FAILURES: u32 = 8;

type ActiveArtifactKey = (SessionId, String, String, String);

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ArtifactRevision {
    Live(u64),
    Persisted(u64),
}

impl ArtifactRevision {
    // Persistence is an authority transition, not a numeric revision increment.
    const fn supersedes(self, previous: Self) -> bool {
        match (self, previous) {
            (Self::Live(next), Self::Live(old)) | (Self::Persisted(next), Self::Persisted(old)) => {
                next > old
            }
            (Self::Persisted(_), Self::Live(_)) => true,
            (Self::Live(_), Self::Persisted(_)) => false,
        }
    }
}

#[derive(Debug, Clone)]
struct ActiveArtifactTarget {
    producer_plugin_id: String,
    schema: String,
    schema_version: u32,
    content_type: Option<String>,
    committed_bytes: u64,
    revision: ArtifactRevision,
    finalized: bool,
}

#[derive(Debug, Default)]
struct ActiveArtifactFetchState {
    next_offset: u64,
    target: Option<ActiveArtifactTarget>,
    fetching: bool,
    retry_at: Option<Instant>,
    consecutive_failures: u32,
    terminal_error: Option<String>,
    delivered_successfully: bool,
}

#[derive(Debug)]
pub struct ActiveArtifactFetchCompletion {
    session_id: SessionId,
    key: ActiveArtifactKey,
    requested_offset: u64,
    requested_end: u64,
    target_revision: ArtifactRevision,
    result: Result<bcode_client::SessionArtifactRange, ClientError>,
}

fn active_artifact_request_range(next_offset: u64, committed_bytes: u64) -> Option<(u32, u64)> {
    let remaining = committed_bytes.checked_sub(next_offset)?;
    if remaining == 0 {
        return None;
    }
    let length = u32::try_from(remaining)
        .unwrap_or(u32::MAX)
        .min(ACTIVE_ARTIFACT_FETCH_BYTES);
    Some((length, next_offset.saturating_add(u64::from(length))))
}

fn active_artifact_completion_is_current(
    completion_session_id: SessionId,
    current_session_id: Option<SessionId>,
    requested_offset: u64,
    next_offset: u64,
) -> bool {
    Some(completion_session_id) == current_session_id && requested_offset == next_offset
}

fn validate_active_artifact_range(
    range: &bcode_client::SessionArtifactRange,
    next_offset: u64,
    requested_end: u64,
    target: &ActiveArtifactTarget,
    requested_revision: ArtifactRevision,
) -> Result<u64, &'static str> {
    let expected_end = range.next_offset();
    let response_revision = range.finalized_event_seq.map_or(
        ArtifactRevision::Live(range.reference_revision),
        ArtifactRevision::Persisted,
    );
    if range.offset != next_offset
        || expected_end > requested_end
        || range.total_bytes < expected_end
        || (requested_revision != target.revision
            && !target.revision.supersedes(requested_revision))
        || (response_revision != requested_revision
            && !response_revision.supersedes(requested_revision))
        || range
            .finalized_event_seq
            .is_some_and(|sequence| !range.finalized || sequence != range.reference_revision)
    {
        return Err("artifact range response did not match the requested committed prefix");
    }
    if range.bytes.is_empty() && next_offset < requested_end {
        return Err("artifact range response ended before the requested committed boundary");
    }
    Ok(expected_end)
}

fn finalized_artifact_error_is_terminal(error: &ClientError) -> bool {
    matches!(
        error,
        ClientError::Server { code, .. }
            if matches!(code.as_str(), "artifact_not_found" | "artifact_unavailable")
    )
}

const ACTIVE_ARTIFACT_COMPLETION_CAPACITY: usize = 64;

#[derive(Debug, Default)]
pub struct ArtifactStreamStats {
    pub observed_targets: u64,
    pub coalesced_targets: u64,
    pub fetches_started: u64,
    pub completions: u64,
    pub stale_completions: u64,
    pub delivered_chunks: u64,
    pub delivered_bytes: u64,
    pub retries: u64,
    pub terminal_failures: u64,
    pub backlog: u64,
}

pub struct ArtifactStreamCoordinator {
    artifact_fetches: BTreeMap<ActiveArtifactKey, ActiveArtifactFetchState>,
    artifact_fetch_sender: tokio::sync::mpsc::Sender<ActiveArtifactFetchCompletion>,
    artifact_fetch_receiver: tokio::sync::mpsc::Receiver<ActiveArtifactFetchCompletion>,
    passive_client: BcodeClient,
    stats: ArtifactStreamStats,
}

impl ArtifactStreamCoordinator {
    pub(crate) fn new(passive_client: BcodeClient) -> Self {
        let (artifact_fetch_sender, artifact_fetch_receiver) =
            tokio::sync::mpsc::channel(ACTIVE_ARTIFACT_COMPLETION_CAPACITY);
        Self {
            artifact_fetches: BTreeMap::new(),
            artifact_fetch_sender,
            artifact_fetch_receiver,
            passive_client,
            stats: ArtifactStreamStats::default(),
        }
    }

    /// Propose bounded viewport input for nonterminal artifact-backed invocations.
    pub(crate) fn viewport_inputs(
        &self,
        session_id: SessionId,
        presentation: &crate::plugin_tui::PluginTuiPresentation,
        viewport: bmux_tui::geometry::Size,
        terminal: impl Fn(&str) -> bool,
    ) -> Vec<bcode_tool::ToolInvocationInput> {
        let mut seen = std::collections::BTreeSet::new();
        self.artifact_fetches
            .iter()
            .filter_map(|(key, state)| {
                let target = state.target.as_ref()?;
                if key.0 != session_id
                    || target.finalized
                    || terminal(&key.1)
                    || !seen.insert(&key.1)
                {
                    return None;
                }
                presentation.viewport_input(
                    &key.1,
                    &target.producer_plugin_id,
                    &target.schema,
                    target.schema_version,
                    viewport,
                )
            })
            .take(64)
            .collect()
    }

    pub(crate) fn drain_stats(&mut self) -> ArtifactStreamStats {
        self.stats.backlog = u64::try_from(
            self.artifact_fetches
                .values()
                .filter(|state| {
                    state.fetching
                        || state.next_offset
                            < state
                                .target
                                .as_ref()
                                .map_or(0, |target| target.committed_bytes)
                })
                .count(),
        )
        .unwrap_or(u64::MAX);
        std::mem::take(&mut self.stats)
    }

    pub(crate) fn reset_session(&mut self, session_id: SessionId) {
        // The retained adapters still own decoded prefixes. Restarting at zero would
        // deliver conflicting duplicate ranges to those same decoders.
        for (key, state) in &mut self.artifact_fetches {
            if key.0 == session_id {
                state.terminal_error = None;
                state.consecutive_failures = 0;
                state.retry_at = Some(Instant::now());
            }
        }
    }

    pub(crate) fn failed_invocations(&self) -> Vec<String> {
        self.artifact_fetches
            .iter()
            .filter(|(_, state)| state.terminal_error.is_some())
            .map(|(key, _)| key.1.clone())
            .take(256)
            .collect()
    }

    pub(crate) fn recovered_invocations(&self) -> Vec<String> {
        let mut recovered = BTreeMap::new();
        for (key, state) in &self.artifact_fetches {
            let healthy = state.delivered_successfully
                && state.terminal_error.is_none()
                && state.retry_at.is_none()
                && state
                    .target
                    .as_ref()
                    .is_some_and(|target| state.next_offset >= target.committed_bytes);
            recovered
                .entry(&key.1)
                .and_modify(|value| *value &= healthy)
                .or_insert(healthy);
        }
        recovered
            .into_iter()
            .filter(|(_, healthy)| *healthy)
            .map(|(id, _)| id.clone())
            .take(256)
            .collect()
    }

    pub(crate) fn retain_session(&mut self, session_id: Option<SessionId>) {
        self.artifact_fetches
            .retain(|key, _| Some(key.0) == session_id);
    }

    pub(crate) fn observe_finalized_artifact(
        &mut self,
        session_id: SessionId,
        sequence: u64,
        event: &SessionEventKind,
        accepts_reference: impl Fn(&str, &str, u32, &str, Option<&str>) -> bool,
    ) {
        let SessionEventKind::ToolInvocationResultRecorded { record } = event else {
            return;
        };
        let Some(bcode_session_models::ToolInvocationResult::Artifact { artifact }) =
            &record.result
        else {
            return;
        };
        let tool_call_id = &record.invocation_id;
        for reference in &artifact.refs {
            if !accepts_reference(
                &artifact.producer_plugin_id,
                &artifact.schema,
                artifact.schema_version,
                &reference.key,
                reference.content_type.as_deref(),
            ) {
                continue;
            }
            let Some(committed_bytes) = reference.byte_len else {
                continue;
            };
            let availability = reference
                .metadata
                .as_ref()
                .and_then(|metadata| metadata.get("availability"))
                .and_then(serde_json::Value::as_str);
            if matches!(
                availability,
                Some("missing" | "incomplete" | "corrupt" | "evicted" | "unavailable")
            ) {
                continue;
            }
            let key = (
                session_id,
                tool_call_id.clone(),
                artifact.artifact_id.clone(),
                reference.key.clone(),
            );
            let state = self.artifact_fetches.entry(key.clone()).or_default();
            if state.target.as_ref().is_some_and(|target| {
                !ArtifactRevision::Persisted(sequence).supersedes(target.revision)
            }) {
                continue;
            }
            // Live producer revisions and canonical event sequences are different domains.
            // Finalization replaces the live target even when its sequence is numerically lower.
            // Duplicate finalization must not reopen an exhausted retry loop.
            state.terminal_error = None;
            state.consecutive_failures = 0;
            state.retry_at = None;
            state.target = Some(ActiveArtifactTarget {
                producer_plugin_id: artifact.producer_plugin_id.clone(),
                schema: artifact.schema.clone(),
                schema_version: artifact.schema_version,
                content_type: reference.content_type.clone(),
                committed_bytes,
                revision: ArtifactRevision::Persisted(sequence),
                finalized: true,
            });
            self.schedule_active_artifact_fetch(session_id, &key);
        }
    }

    pub(crate) fn observe_session_live_artifact(
        &mut self,
        session_id: SessionId,
        event: &bcode_session_models::SessionLiveEventKind,
        accepts_reference: impl Fn(&str, &str, u32, &str, Option<&str>) -> bool,
    ) {
        match event {
            bcode_session_models::SessionLiveEventKind::ToolContributionPlaced { envelope } => {
                self.observe_contribution(session_id, &envelope.contribution, accepts_reference);
            }
            bcode_session_models::SessionLiveEventKind::ToolPresentationUpdated { update } => {
                self.observe_presentation_update(session_id, update, accepts_reference);
            }
            _ => {}
        }
    }

    pub(crate) fn observe_contribution(
        &mut self,
        session_id: SessionId,
        event: &bcode_session_models::ToolContributionEvent,
        accepts_reference: impl Fn(&str, &str, u32, &str, Option<&str>) -> bool,
    ) {
        self.observe_artifact(
            session_id,
            &event.invocation_id,
            &event.producer_id,
            &event.schema,
            event.schema_version,
            event.artifact.as_ref(),
            accepts_reference,
        );
    }

    fn observe_presentation_update(
        &mut self,
        session_id: SessionId,
        update: &bcode_tool::ToolPresentationUpdate,
        accepts_reference: impl Fn(&str, &str, u32, &str, Option<&str>) -> bool,
    ) {
        self.observe_artifact(
            session_id,
            &update.invocation_id,
            &update.producer_id,
            &update.schema,
            update.schema_version,
            update.artifact.as_ref(),
            accepts_reference,
        );
    }

    #[allow(clippy::too_many_arguments)]
    fn observe_artifact(
        &mut self,
        session_id: SessionId,
        invocation_id: &str,
        producer_plugin_id: &str,
        schema: &str,
        schema_version: u32,
        artifact: Option<&bcode_tool::ToolContributionArtifact>,
        accepts_reference: impl Fn(&str, &str, u32, &str, Option<&str>) -> bool,
    ) {
        let Some(artifact) = artifact else {
            return;
        };
        if !accepts_reference(
            producer_plugin_id,
            schema,
            schema_version,
            &artifact.reference_key,
            artifact.content_type.as_deref(),
        ) {
            return;
        }
        let key = (
            session_id,
            invocation_id.to_owned(),
            artifact.artifact_id.clone(),
            artifact.reference_key.clone(),
        );
        self.observe_artifact_target(
            session_id,
            &key,
            ActiveArtifactTarget {
                producer_plugin_id: producer_plugin_id.to_owned(),
                schema: schema.to_owned(),
                schema_version,
                content_type: artifact.content_type.clone(),
                committed_bytes: artifact.committed_bytes,
                revision: ArtifactRevision::Live(artifact.revision),
                finalized: artifact.finalized,
            },
            artifact.availability.as_deref() == Some("incomplete"),
        );
    }

    fn observe_artifact_target(
        &mut self,
        session_id: SessionId,
        key: &ActiveArtifactKey,
        target: ActiveArtifactTarget,
        incomplete: bool,
    ) {
        self.stats.observed_targets = self.stats.observed_targets.saturating_add(1);
        let state = self.artifact_fetches.entry(key.clone()).or_default();
        if state
            .target
            .as_ref()
            .is_some_and(|current| matches!(current.revision, ArtifactRevision::Persisted(_)))
        {
            return;
        }
        if incomplete {
            state.fetching = false;
            state.retry_at = None;
            state.terminal_error = Some(
                "active artifact is incomplete because its producer stopped before finalization"
                    .to_owned(),
            );
            self.stats.terminal_failures = self.stats.terminal_failures.saturating_add(1);
            return;
        }
        if state.target.as_ref().is_some_and(|current| {
            current.finalized || !target.revision.supersedes(current.revision)
        }) {
            self.stats.coalesced_targets = self.stats.coalesced_targets.saturating_add(1);
            return;
        }
        if state.fetching && state.target.is_some() {
            self.stats.coalesced_targets = self.stats.coalesced_targets.saturating_add(1);
        }
        if state.consecutive_failures >= ACTIVE_ARTIFACT_MAX_FAILURES {
            // A new source revision may repair the failed range. Re-observing the same
            // revision above does not restart a stopped retry loop.
            state.consecutive_failures = 0;
            state.terminal_error = None;
            state.retry_at = None;
        }
        state.target = Some(target);
        self.schedule_active_artifact_fetch(session_id, key);
    }

    fn schedule_active_artifact_fetch(&mut self, session_id: SessionId, key: &ActiveArtifactKey) {
        let Some(state) = self.artifact_fetches.get_mut(key) else {
            return;
        };
        let Some(target) = state.target.as_ref() else {
            return;
        };
        if state.fetching
            || state.next_offset >= target.committed_bytes
            || state.terminal_error.is_some()
            || state
                .retry_at
                .is_some_and(|retry_at| retry_at > Instant::now())
        {
            return;
        }
        let requested_offset = state.next_offset;
        let Some((length, requested_end)) =
            active_artifact_request_range(requested_offset, target.committed_bytes)
        else {
            return;
        };
        let target_revision = target.revision;
        state.fetching = true;
        self.stats.fetches_started = self.stats.fetches_started.saturating_add(1);
        state.retry_at = None;
        let client = self.passive_client.clone();
        let sender = self.artifact_fetch_sender.clone();
        let task_key = key.clone();
        tokio::spawn(async move {
            let result = client
                .session_artifact_range(
                    session_id,
                    task_key.2.clone(),
                    task_key.3.clone(),
                    requested_offset,
                    length,
                )
                .await;
            let _result = sender
                .send(ActiveArtifactFetchCompletion {
                    session_id,
                    key: task_key,
                    requested_offset,
                    requested_end,
                    target_revision,
                    result,
                })
                .await;
        });
    }

    #[allow(clippy::too_many_lines)] // Keeps response validation, delivery, retry, and scheduling as one state transition.
    pub(crate) fn handle_completion(
        &mut self,
        current_session_id: Option<SessionId>,
        completion: ActiveArtifactFetchCompletion,
        deliver: impl FnOnce(&PluginTuiArtifactChunk) -> Result<bool, String>,
    ) -> bool {
        self.stats.completions = self.stats.completions.saturating_add(1);
        let key = completion.key.clone();
        let chunk = {
            let Some(state) = self.artifact_fetches.get_mut(&key) else {
                return false;
            };
            if !active_artifact_completion_is_current(
                completion.session_id,
                current_session_id,
                completion.requested_offset,
                state.next_offset,
            ) {
                self.stats.stale_completions = self.stats.stale_completions.saturating_add(1);
                return false;
            }
            let Some(target) = state.target.clone() else {
                return false;
            };
            if completion.target_revision != target.revision {
                state.fetching = false;
                self.stats.stale_completions = self.stats.stale_completions.saturating_add(1);
                self.schedule_active_artifact_fetch(completion.session_id, &key);
                return false;
            }
            state.fetching = false;
            let range = match completion.result {
                Ok(range) => range,
                Err(error) => {
                    let error_message = error.to_string();
                    if target.finalized && finalized_artifact_error_is_terminal(&error) {
                        tracing::warn!(
                            error = %error_message,
                            "finalized artifact is unavailable; stopping hydration retries"
                        );
                        state.retry_at = None;
                        state.terminal_error = Some(error_message);
                        self.stats.terminal_failures =
                            self.stats.terminal_failures.saturating_add(1);
                    } else {
                        Self::defer_active_artifact_fetch(state, &error_message);
                        if state.terminal_error.is_some() {
                            self.stats.terminal_failures =
                                self.stats.terminal_failures.saturating_add(1);
                        } else {
                            self.stats.retries = self.stats.retries.saturating_add(1);
                        }
                    }
                    return false;
                }
            };
            let expected_end = match validate_active_artifact_range(
                &range,
                state.next_offset,
                completion.requested_end,
                &target,
                completion.target_revision,
            ) {
                Ok(expected_end) => expected_end,
                Err(error) => {
                    Self::defer_active_artifact_fetch(state, error);
                    if state.terminal_error.is_some() {
                        self.stats.terminal_failures =
                            self.stats.terminal_failures.saturating_add(1);
                    } else {
                        self.stats.retries = self.stats.retries.saturating_add(1);
                    }
                    return false;
                }
            };
            if let Some(sequence) = range.finalized_event_seq {
                let mut persisted = target.clone();
                persisted.revision = ArtifactRevision::Persisted(sequence);
                persisted.committed_bytes = range.total_bytes;
                persisted.finalized = true;
                state.target = Some(persisted);
            }
            if range.bytes.is_empty() {
                state.consecutive_failures = 0;
                state.retry_at = None;
                None
            } else {
                Some((
                    PluginTuiArtifactChunk {
                        tool_call_id: key.1.clone(),
                        artifact_id: key.2.clone(),
                        reference_key: key.3.clone(),
                        producer_plugin_id: target.producer_plugin_id,
                        schema: target.schema,
                        schema_version: target.schema_version,
                        content_type: target.content_type,
                        offset: range.offset,
                        total_bytes: range.total_bytes,
                        revision: range.reference_revision,
                        finalized: range.finalized || target.finalized,
                        bytes: range.bytes,
                    },
                    expected_end,
                ))
            }
        };

        let mut redraw = false;
        if let Some((chunk, expected_end)) = chunk {
            let delivery = deliver(&chunk);
            let state = self
                .artifact_fetches
                .get_mut(&key)
                .expect("artifact fetch state remains registered during delivery");
            match delivery {
                Ok(true) => {
                    state.next_offset = expected_end;
                    state.delivered_successfully = true;
                    state.consecutive_failures = 0;
                    state.retry_at = None;
                    self.stats.delivered_chunks = self.stats.delivered_chunks.saturating_add(1);
                    self.stats.delivered_bytes = self
                        .stats
                        .delivered_bytes
                        .saturating_add(u64::try_from(chunk.bytes.len()).unwrap_or(u64::MAX));
                    redraw = true;
                }
                Ok(false) => {
                    let error = "artifact schema has no owning visual adapter".to_owned();
                    tracing::warn!(%error, "active artifact delivery stopped");
                    state.terminal_error = Some(error);
                    self.stats.terminal_failures = self.stats.terminal_failures.saturating_add(1);
                }
                Err(error) => {
                    tracing::warn!(%error, "active artifact delivery failed");
                    state.terminal_error = Some(error);
                    self.stats.terminal_failures = self.stats.terminal_failures.saturating_add(1);
                }
            }
        }
        self.schedule_active_artifact_fetch(completion.session_id, &key);
        redraw
    }

    pub(crate) fn start_due_fetches(&mut self, now: Instant) {
        let due = self
            .artifact_fetches
            .iter()
            .filter(|(_, state)| {
                !state.fetching
                    && state.terminal_error.is_none()
                    && state.retry_at.is_some_and(|retry_at| retry_at <= now)
            })
            .map(|(key, _)| key.clone())
            .collect::<Vec<_>>();
        for key in due {
            self.schedule_active_artifact_fetch(key.0, &key);
        }
    }

    pub(crate) fn next_retry_at(&self) -> Option<Instant> {
        self.artifact_fetches
            .values()
            .filter_map(|state| state.retry_at)
            .min()
    }

    #[cfg(test)]
    pub(crate) async fn next_completion(&mut self) -> Option<ActiveArtifactFetchCompletion> {
        self.artifact_fetch_receiver.recv().await
    }

    #[cfg(test)]
    pub(crate) fn completion_capacity(&self) -> usize {
        self.artifact_fetch_receiver.max_capacity()
    }

    pub(crate) fn take_completion_receiver(
        &mut self,
    ) -> tokio::sync::mpsc::Receiver<ActiveArtifactFetchCompletion> {
        let (_replacement_sender, replacement_receiver) =
            tokio::sync::mpsc::channel(ACTIVE_ARTIFACT_COMPLETION_CAPACITY);
        std::mem::replace(&mut self.artifact_fetch_receiver, replacement_receiver)
    }

    fn defer_active_artifact_fetch(state: &mut ActiveArtifactFetchState, error: &str) {
        tracing::debug!(
            error,
            consecutive_failures = state.consecutive_failures.saturating_add(1),
            "deferring active artifact fetch"
        );
        state.consecutive_failures = state.consecutive_failures.saturating_add(1);
        if state.consecutive_failures >= ACTIVE_ARTIFACT_MAX_FAILURES {
            state.retry_at = None;
            state.terminal_error = Some(
                "artifact hydration paused after repeated failures; reconnect to retry".to_owned(),
            );
            return;
        }
        let exponent = state.consecutive_failures.saturating_sub(1).min(4);
        let multiplier = 1_u32 << exponent;
        let delay = ACTIVE_ARTIFACT_RETRY_BASE
            .saturating_mul(multiplier)
            .min(ACTIVE_ARTIFACT_RETRY_MAX);
        state.retry_at = Some(Instant::now() + delay);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn artifact_completion_channel_is_bounded() {
        let coordinator = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        assert_eq!(
            coordinator.completion_capacity(),
            ACTIVE_ARTIFACT_COMPLETION_CAPACITY
        );
    }

    fn target(committed_bytes: u64, revision: u64, finalized: bool) -> ActiveArtifactTarget {
        ActiveArtifactTarget {
            producer_plugin_id: "test.producer".to_owned(),
            schema: "test.artifact".to_owned(),
            schema_version: 1,
            content_type: Some("application/octet-stream".to_owned()),
            committed_bytes,
            revision: ArtifactRevision::Live(revision),
            finalized,
        }
    }

    fn range(
        offset: u64,
        total_bytes: u64,
        revision: u64,
        bytes: &[u8],
    ) -> bcode_client::SessionArtifactRange {
        bcode_client::SessionArtifactRange {
            artifact_id: "artifact".to_owned(),
            reference_key: "reference".to_owned(),
            content_type: Some("application/octet-stream".to_owned()),
            offset,
            total_bytes,
            reference_bytes: Some(total_bytes),
            reference_revision: revision,
            finalized: false,
            finalized_event_seq: None,
            availability: Some("active".to_owned()),
            complete: Some(false),
            checksum_sha256: None,
            bytes: bytes.to_vec(),
        }
    }

    #[tokio::test]
    async fn presentation_update_hydration_requires_adapter_owned_reference() {
        let session_id = SessionId::new();
        let event = bcode_session_models::SessionLiveEventKind::ToolPresentationUpdated {
            update: bcode_tool::ToolPresentationUpdate {
                invocation_id: "call".to_owned(),
                producer_id: "test.producer".to_owned(),
                generation: 0,
                revision: 2,
                identity: bcode_tool::ToolPresentationIdentity::Primary,
                retention: bcode_tool::ToolPresentationRetention::RetainLatest,
                schema: "test.artifact".to_owned(),
                schema_version: 1,
                artifact: Some(bcode_tool::ToolContributionArtifact {
                    artifact_id: "artifact".to_owned(),
                    reference_key: "reference".to_owned(),
                    content_type: Some("application/octet-stream".to_owned()),
                    storage_uri: "untrusted://must-not-be-read".to_owned(),
                    committed_bytes: 9,
                    revision: 2,
                    finalized: false,
                    availability: None,
                }),
                payload: serde_json::Value::Null,
            },
        };

        let mut rejected = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        rejected.observe_session_live_artifact(session_id, &event, |_, _, _, _, _| false);
        assert!(rejected.artifact_fetches.is_empty());

        let mut accepted = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        accepted.observe_session_live_artifact(
            session_id,
            &event,
            |producer, schema, version, key, ty| {
                producer == "test.producer"
                    && schema == "test.artifact"
                    && version == 1
                    && key == "reference"
                    && ty == Some("application/octet-stream")
            },
        );
        let key = (
            session_id,
            "call".to_owned(),
            "artifact".to_owned(),
            "reference".to_owned(),
        );
        let state = accepted
            .artifact_fetches
            .get(&key)
            .expect("accepted presentation artifact fetch");
        assert!(state.fetching);
        let target = state.target.as_ref().expect("presentation artifact target");
        assert_eq!(target.committed_bytes, 9);
        assert_eq!(target.revision, ArtifactRevision::Live(2));
        assert_eq!(target.schema, "test.artifact");
    }

    #[tokio::test]
    async fn contribution_hydration_requires_adapter_owned_reference() {
        let session_id = SessionId::new();
        let event = bcode_session_models::ToolContributionEvent {
            invocation_id: "call".to_owned(),
            contribution_id: "recording".to_owned(),
            sequence: 2,
            producer_id: "test.producer".to_owned(),
            schema: "test.artifact".to_owned(),
            schema_version: 1,
            operation: bcode_session_models::ToolContributionOperation::Upsert,
            persistence: bcode_session_models::ToolContributionPersistence::Transient,
            artifact: Some(bcode_session_models::ToolContributionArtifact {
                artifact_id: "artifact".to_owned(),
                reference_key: "reference".to_owned(),
                content_type: Some("application/octet-stream".to_owned()),
                storage_uri: "untrusted://must-not-be-read".to_owned(),
                committed_bytes: 9,
                revision: 2,
                finalized: false,
                availability: None,
            }),
            payload: serde_json::Value::Null,
        };

        let mut rejected = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        rejected.observe_contribution(session_id, &event, |_, _, _, _, _| false);
        assert!(rejected.artifact_fetches.is_empty());
        assert_eq!(rejected.stats.observed_targets, 0);

        let mut accepted = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        accepted.observe_contribution(session_id, &event, |producer, schema, version, key, ty| {
            producer == "test.producer"
                && schema == "test.artifact"
                && version == 1
                && key == "reference"
                && ty == Some("application/octet-stream")
        });
        assert_eq!(accepted.artifact_fetches.len(), 1);
        assert_eq!(accepted.stats.observed_targets, 1);
    }

    #[test]
    fn incomplete_contribution_stops_without_scheduling_a_range_fetch() {
        let client = BcodeClient::default_endpoint();
        let mut coordinator = ArtifactStreamCoordinator::new(client);
        let session_id = SessionId::new();
        let event = bcode_session_models::ToolContributionEvent {
            invocation_id: "call".to_owned(),
            contribution_id: "recording".to_owned(),
            sequence: 2,
            producer_id: "test.producer".to_owned(),
            schema: "test.artifact".to_owned(),
            schema_version: 1,
            operation: bcode_session_models::ToolContributionOperation::Upsert,
            persistence: bcode_session_models::ToolContributionPersistence::Transient,
            artifact: Some(bcode_session_models::ToolContributionArtifact {
                artifact_id: "artifact".to_owned(),
                reference_key: "reference".to_owned(),
                content_type: Some("application/octet-stream".to_owned()),
                storage_uri: String::new(),
                committed_bytes: 9,
                revision: 2,
                finalized: false,
                availability: Some("incomplete".to_owned()),
            }),
            payload: serde_json::Value::Null,
        };

        coordinator.observe_contribution(session_id, &event, |_, _, _, _, _| true);

        let key = (
            session_id,
            "call".to_owned(),
            "artifact".to_owned(),
            "reference".to_owned(),
        );
        let state = coordinator
            .artifact_fetches
            .get(&key)
            .expect("incomplete artifact state");
        assert!(!state.fetching);
        assert!(state.retry_at.is_none());
        assert!(state.terminal_error.as_deref().is_some_and(|error| {
            error.contains("incomplete") && error.contains("producer stopped")
        }));
    }

    #[tokio::test]
    async fn finalized_hydration_selects_only_adapter_owned_references() {
        let client = BcodeClient::default_endpoint();
        let mut coordinator = ArtifactStreamCoordinator::new(client);
        let session_id = SessionId::new();
        let event = SessionEventKind::ToolInvocationResultRecorded {
            record: bcode_session_models::ToolInvocationResultRecord {
                invocation_id: "call".to_owned(),
                model_output: String::new(),
                is_error: false,
                presentation: None,
                result: Some(bcode_session_models::ToolInvocationResult::Artifact {
                    artifact: Box::new(bcode_session_models::ToolArtifact {
                        artifact_id: "artifact".to_owned(),
                        producer_plugin_id: "test.producer".to_owned(),
                        schema: "test.visual".to_owned(),
                        schema_version: 1,
                        tool_call_id: Some("call".to_owned()),
                        title: None,
                        metadata: serde_json::Value::Null,
                        refs: vec![
                            bcode_session_models::ToolArtifactRef {
                                key: "ignored".to_owned(),
                                content_type: Some("text/plain".to_owned()),
                                storage_uri: None,
                                byte_len: Some(5),
                                metadata: None,
                            },
                            bcode_session_models::ToolArtifactRef {
                                key: "accepted".to_owned(),
                                content_type: Some("application/test".to_owned()),
                                storage_uri: None,
                                byte_len: Some(10),
                                metadata: None,
                            },
                        ],
                    }),
                }),
                content: Vec::new(),
            },
        };
        let key = (
            session_id,
            "call".to_owned(),
            "artifact".to_owned(),
            "accepted".to_owned(),
        );
        coordinator.artifact_fetches.insert(
            key.clone(),
            ActiveArtifactFetchState {
                next_offset: 10,
                target: Some(target(10, 6145, true)),
                terminal_error: Some("exhausted live retries".to_owned()),
                ..ActiveArtifactFetchState::default()
            },
        );
        coordinator.observe_finalized_artifact(
            session_id,
            7,
            &event,
            |_producer, _schema, _version, key, _content_type| key == "accepted",
        );
        assert_eq!(
            coordinator.artifact_fetches[&key]
                .target
                .as_ref()
                .unwrap()
                .revision,
            ArtifactRevision::Persisted(7)
        );
        assert!(coordinator.artifact_fetches[&key].terminal_error.is_none());
        coordinator.observe_artifact_target(session_id, &key, target(20, 9000, false), true);
        assert_eq!(coordinator.artifact_fetches[&key].next_offset, 10);
        assert!(coordinator.artifact_fetches[&key].terminal_error.is_none());
        assert_eq!(coordinator.artifact_fetches.len(), 1);
        assert!(
            coordinator
                .artifact_fetches
                .keys()
                .all(|key| key.3 == "accepted")
        );
    }

    #[test]
    fn artifact_fetch_ranges_are_chunk_bounded_and_advance_from_committed_offset() {
        let committed_bytes = u64::from(ACTIVE_ARTIFACT_FETCH_BYTES) * 3;
        assert_eq!(
            active_artifact_request_range(17, committed_bytes),
            Some((
                ACTIVE_ARTIFACT_FETCH_BYTES,
                17 + u64::from(ACTIVE_ARTIFACT_FETCH_BYTES)
            ))
        );
        assert_eq!(
            active_artifact_request_range(committed_bytes - 9, committed_bytes),
            Some((9, committed_bytes))
        );
        assert_eq!(
            active_artifact_request_range(committed_bytes, committed_bytes),
            None
        );
        assert_eq!(
            active_artifact_request_range(committed_bytes + 1, committed_bytes),
            None
        );
    }

    #[test]
    fn resetting_a_resident_session_preserves_decoder_progress() {
        let session_id = SessionId::new();
        let other_session_id = SessionId::new();
        let mut coordinator = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        for (owner, name) in [(session_id, "old"), (other_session_id, "retained")] {
            coordinator.artifact_fetches.insert(
                (
                    owner,
                    name.to_owned(),
                    "artifact".to_owned(),
                    "reference".to_owned(),
                ),
                ActiveArtifactFetchState::default(),
            );
        }

        coordinator.reset_session(session_id);
        assert_eq!(coordinator.artifact_fetches.len(), 2);
        assert!(
            coordinator
                .artifact_fetches
                .iter()
                .any(|(key, state)| key.0 == session_id && state.retry_at.is_some())
        );
    }

    #[test]
    fn range_validation_accepts_contiguous_growth_and_rejects_duplicates() {
        let active = target(10, 2, false);
        assert_eq!(
            validate_active_artifact_range(
                &range(0, 10, 2, b"abc"),
                0,
                10,
                &active,
                ArtifactRevision::Live(2)
            ),
            Ok(3)
        );
        let session_id = SessionId::new();
        assert!(active_artifact_completion_is_current(
            session_id,
            Some(session_id),
            0,
            0
        ));
        assert!(!active_artifact_completion_is_current(
            session_id,
            Some(session_id),
            0,
            3
        ));
    }

    #[test]
    fn repeated_invalid_ranges_pause_without_losing_the_committed_prefix() {
        let mut coordinator = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        let session_id = SessionId::new();
        let key = (
            session_id,
            "tool".to_owned(),
            "artifact".to_owned(),
            "reference".to_owned(),
        );
        coordinator.artifact_fetches.insert(
            key.clone(),
            ActiveArtifactFetchState {
                next_offset: 4,
                target: Some(target(8, 3, false)),
                ..ActiveArtifactFetchState::default()
            },
        );
        for _ in 0..ACTIVE_ARTIFACT_MAX_FAILURES {
            let completion = ActiveArtifactFetchCompletion {
                session_id,
                key: key.clone(),
                requested_offset: 4,
                requested_end: 8,
                target_revision: ArtifactRevision::Live(3),
                result: Ok(range(0, 8, 2, b"stale")),
            };
            assert!(
                !coordinator.handle_completion(Some(session_id), completion, |_| {
                    panic!("invalid bytes must not be delivered")
                })
            );
        }
        assert!(coordinator.next_retry_at().is_none());
        assert_eq!(coordinator.failed_invocations(), vec!["tool"]);
        assert_eq!(coordinator.artifact_fetches[&key].next_offset, 4);
        coordinator.observe_artifact_target(session_id, &key, target(8, 3, false), false);
        coordinator.start_due_fetches(Instant::now() + Duration::from_mins(1));
        assert!(!coordinator.artifact_fetches[&key].fetching);
        assert_eq!(coordinator.stats.terminal_failures, 1);
        coordinator.reset_session(session_id);
        assert!(coordinator.failed_invocations().is_empty());
        assert_eq!(coordinator.artifact_fetches[&key].consecutive_failures, 0);
        assert_eq!(coordinator.artifact_fetches[&key].next_offset, 4);
        assert!(coordinator.next_retry_at().is_some());
    }

    #[test]
    fn stale_or_mismatched_ranges_preserve_the_committed_prefix_and_retry() {
        let mut coordinator = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        let session_id = SessionId::new();
        let key = (
            session_id,
            "tool".to_owned(),
            "artifact".to_owned(),
            "reference".to_owned(),
        );
        coordinator.artifact_fetches.insert(
            key.clone(),
            ActiveArtifactFetchState {
                next_offset: 4,
                target: Some(target(8, 3, false)),
                fetching: true,
                ..ActiveArtifactFetchState::default()
            },
        );
        let completion = ActiveArtifactFetchCompletion {
            session_id,
            key: key.clone(),
            requested_offset: 4,
            requested_end: 8,
            target_revision: ArtifactRevision::Live(3),
            result: Ok(range(0, 8, 2, b"stale")),
        };

        let mut delivered = false;
        assert!(
            !coordinator.handle_completion(Some(session_id), completion, |_| {
                delivered = true;
                Ok(true)
            })
        );
        assert!(!delivered);
        let state = coordinator
            .artifact_fetches
            .get(&key)
            .expect("artifact state");
        assert_eq!(state.next_offset, 4);
        assert!(state.retry_at.is_some());
        assert!(state.terminal_error.is_none());
    }

    #[tokio::test]
    async fn in_flight_live_fetch_rejects_a_newer_committed_revision() {
        let mut coordinator = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
        let session_id = SessionId::new();
        let key = (
            session_id,
            "tool".to_owned(),
            "artifact".to_owned(),
            "reference".to_owned(),
        );
        coordinator.artifact_fetches.insert(
            key.clone(),
            ActiveArtifactFetchState {
                target: Some(target(20, 3, false)),
                fetching: true,
                ..ActiveArtifactFetchState::default()
            },
        );
        let completion = ActiveArtifactFetchCompletion {
            session_id,
            key: key.clone(),
            requested_offset: 0,
            requested_end: 10,
            target_revision: ArtifactRevision::Live(2),
            result: Ok(range(0, 20, 3, b"abcdefghij")),
        };
        let mut delivered = false;
        assert!(
            !coordinator.handle_completion(Some(session_id), completion, |_| {
                delivered = true;
                Ok(true)
            })
        );
        assert!(!delivered);
        let state = coordinator
            .artifact_fetches
            .get(&key)
            .expect("artifact state");
        assert_eq!(state.next_offset, 0);
        assert!(state.fetching);
        assert_eq!(coordinator.stats.stale_completions, 1);
    }

    #[test]
    fn missing_finalized_artifact_stops_retrying() {
        let client = BcodeClient::default_endpoint();
        let mut coordinator = ArtifactStreamCoordinator::new(client);
        let session_id = SessionId::new();
        let key = (
            session_id,
            "tool".to_owned(),
            "artifact".to_owned(),
            "reference".to_owned(),
        );
        coordinator.artifact_fetches.insert(
            key.clone(),
            ActiveArtifactFetchState {
                target: Some(target(3, 42, true)),
                fetching: true,
                ..ActiveArtifactFetchState::default()
            },
        );
        let completion = ActiveArtifactFetchCompletion {
            session_id,
            key: key.clone(),
            requested_offset: 0,
            requested_end: 3,
            target_revision: ArtifactRevision::Live(42),
            result: Err(ClientError::Server {
                code: "artifact_not_found".to_owned(),
                message: "artifact reference was not found in the finalized projection".to_owned(),
            }),
        };

        assert!(!coordinator.handle_completion(Some(session_id), completion, |_| Ok(true)));
        let state = coordinator
            .artifact_fetches
            .get(&key)
            .expect("artifact state");
        assert!(!state.fetching);
        assert!(state.retry_at.is_none());
        assert!(state.terminal_error.is_some());
        assert!(coordinator.next_retry_at().is_none());
    }

    #[test]
    fn finalized_unavailable_codes_stop_retrying_without_delivery() {
        for code in ["artifact_not_found", "artifact_unavailable"] {
            let mut coordinator = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
            let session_id = SessionId::new();
            let key = (
                session_id,
                format!("tool-{code}"),
                format!("artifact-{code}"),
                "reference".to_owned(),
            );
            coordinator.artifact_fetches.insert(
                key.clone(),
                ActiveArtifactFetchState {
                    target: Some(target(3, 42, true)),
                    fetching: true,
                    ..ActiveArtifactFetchState::default()
                },
            );
            let completion = ActiveArtifactFetchCompletion {
                session_id,
                key: key.clone(),
                requested_offset: 0,
                requested_end: 3,
                target_revision: ArtifactRevision::Live(42),
                result: Err(ClientError::Server {
                    code: code.to_owned(),
                    message: "artifact bytes are unavailable".to_owned(),
                }),
            };

            let mut delivered = false;
            assert!(
                !coordinator.handle_completion(Some(session_id), completion, |_| {
                    delivered = true;
                    Ok(true)
                })
            );
            assert!(!delivered);
            let state = coordinator
                .artifact_fetches
                .get(&key)
                .expect("artifact state");
            assert_eq!(state.next_offset, 0);
            assert!(!state.fetching);
            assert!(state.retry_at.is_none());
            assert!(state.terminal_error.is_some());
        }
    }

    #[test]
    fn unavailable_history_artifacts_do_not_block_live_completion() {
        let client = BcodeClient::default_endpoint();
        let mut coordinator = ArtifactStreamCoordinator::new(client);
        let session_id = SessionId::new();
        for index in 0..64 {
            let key = (
                session_id,
                format!("history-tool-{index}"),
                format!("history-artifact-{index}"),
                "reference".to_owned(),
            );
            coordinator.artifact_fetches.insert(
                key.clone(),
                ActiveArtifactFetchState {
                    target: Some(target(3, index + 1, true)),
                    fetching: true,
                    ..ActiveArtifactFetchState::default()
                },
            );
            let completion = ActiveArtifactFetchCompletion {
                session_id,
                key,
                requested_offset: 0,
                requested_end: 3,
                target_revision: ArtifactRevision::Live(index + 1),
                result: Err(ClientError::Server {
                    code: "artifact_not_found".to_owned(),
                    message: "artifact reference was not found in the finalized projection"
                        .to_owned(),
                }),
            };
            assert!(!coordinator.handle_completion(Some(session_id), completion, |_| Ok(true)));
        }
        assert!(coordinator.next_retry_at().is_none());

        let live_key = (
            session_id,
            "live-tool".to_owned(),
            "live-artifact".to_owned(),
            "reference".to_owned(),
        );
        coordinator.artifact_fetches.insert(
            live_key.clone(),
            ActiveArtifactFetchState {
                target: Some(target(3, 1, false)),
                fetching: true,
                ..ActiveArtifactFetchState::default()
            },
        );
        let live_completion = ActiveArtifactFetchCompletion {
            session_id,
            key: live_key.clone(),
            requested_offset: 0,
            requested_end: 3,
            target_revision: ArtifactRevision::Live(1),
            result: Ok(range(0, 3, 1, b"abc")),
        };
        assert!(coordinator.handle_completion(Some(session_id), live_completion, |_| Ok(true)));
        assert_eq!(
            coordinator
                .artifact_fetches
                .get(&live_key)
                .expect("live artifact state")
                .next_offset,
            3
        );
    }

    #[test]
    fn persisted_response_can_complete_a_live_request_without_replaying_bytes() {
        for live_revision in [22, 6145] {
            let mut coordinator = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
            let session_id = SessionId::new();
            let key = (
                session_id,
                "tool".to_owned(),
                "artifact".to_owned(),
                "reference".to_owned(),
            );
            coordinator.artifact_fetches.insert(
                key.clone(),
                ActiveArtifactFetchState {
                    next_offset: 4,
                    target: Some(target(8, live_revision, true)),
                    fetching: true,
                    ..ActiveArtifactFetchState::default()
                },
            );
            let mut response = range(4, 8, 22, b"tail");
            response.finalized = true;
            response.finalized_event_seq = Some(22);
            let completion = ActiveArtifactFetchCompletion {
                session_id,
                key: key.clone(),
                requested_offset: 4,
                requested_end: 8,
                target_revision: ArtifactRevision::Live(live_revision),
                result: Ok(response),
            };
            assert!(
                coordinator.handle_completion(Some(session_id), completion, |chunk| {
                    assert_eq!(chunk.offset, 4);
                    assert_eq!(chunk.bytes, b"tail");
                    Ok(true)
                })
            );
            assert_eq!(coordinator.artifact_fetches[&key].next_offset, 8);
            assert_eq!(
                coordinator.artifact_fetches[&key]
                    .target
                    .as_ref()
                    .unwrap()
                    .revision,
                ArtifactRevision::Persisted(22)
            );
            assert_eq!(coordinator.recovered_invocations(), vec!["tool"]);
            let mut other = key;
            other.3 = "other".to_owned();
            coordinator.artifact_fetches.insert(
                other,
                ActiveArtifactFetchState {
                    terminal_error: Some("still broken".to_owned()),
                    ..ActiveArtifactFetchState::default()
                },
            );
            assert!(coordinator.recovered_invocations().is_empty());
        }
    }

    #[tokio::test]
    async fn canonical_transition_discards_pending_live_response_and_reschedules() {
        for live_revision in [22, 6145] {
            let mut coordinator = ArtifactStreamCoordinator::new(BcodeClient::default_endpoint());
            let session_id = SessionId::new();
            let key = (
                session_id,
                "tool".to_owned(),
                "artifact".to_owned(),
                "reference".to_owned(),
            );
            let mut persisted = target(8, 22, true);
            persisted.revision = ArtifactRevision::Persisted(22);
            coordinator.artifact_fetches.insert(
                key.clone(),
                ActiveArtifactFetchState {
                    next_offset: 4,
                    target: Some(persisted),
                    fetching: true,
                    ..ActiveArtifactFetchState::default()
                },
            );
            assert!(!coordinator.handle_completion(
                Some(session_id),
                ActiveArtifactFetchCompletion {
                    session_id,
                    key: key.clone(),
                    requested_offset: 4,
                    requested_end: 8,
                    target_revision: ArtifactRevision::Live(live_revision),
                    result: Ok(range(4, 8, live_revision, b"tail")),
                },
                |_| panic!("stale response must not reach the adapter")
            ));
            assert_eq!(coordinator.artifact_fetches[&key].next_offset, 4);
            assert!(coordinator.artifact_fetches[&key].fetching);
            assert_eq!(coordinator.stats.fetches_started, 1);
        }
    }

    #[test]
    fn persisted_request_rejects_live_and_inconsistent_responses() {
        let mut persisted = target(8, 22, true);
        persisted.revision = ArtifactRevision::Persisted(22);
        let mut response = range(0, 8, 6145, b"data");
        assert!(
            validate_active_artifact_range(&response, 0, 8, &persisted, persisted.revision)
                .is_err()
        );
        response.finalized_event_seq = Some(22);
        response.finalized = true;
        assert!(
            validate_active_artifact_range(&response, 0, 8, &persisted, persisted.revision)
                .is_err()
        );
    }

    #[test]
    fn accepted_chunk_advances_only_after_generic_adapter_delivery() {
        let client = BcodeClient::default_endpoint();
        let mut coordinator = ArtifactStreamCoordinator::new(client);
        let session_id = SessionId::new();
        let key = (
            session_id,
            "tool".to_owned(),
            "artifact".to_owned(),
            "reference".to_owned(),
        );
        coordinator.artifact_fetches.insert(
            key.clone(),
            ActiveArtifactFetchState {
                target: Some(target(3, 1, false)),
                fetching: true,
                ..ActiveArtifactFetchState::default()
            },
        );
        let completion = ActiveArtifactFetchCompletion {
            session_id,
            key: key.clone(),
            requested_offset: 0,
            requested_end: 3,
            target_revision: ArtifactRevision::Live(1),
            result: Ok(range(0, 3, 1, b"abc")),
        };
        let changed = coordinator.handle_completion(Some(session_id), completion, |chunk| {
            assert_eq!(chunk.producer_plugin_id, "test.producer");
            assert_eq!(chunk.schema, "test.artifact");
            assert_eq!(chunk.bytes, b"abc");
            Ok(true)
        });
        assert!(changed);
        assert_eq!(
            coordinator
                .artifact_fetches
                .get(&key)
                .expect("artifact state")
                .next_offset,
            3
        );
    }
}