nexus-sdk 2.1.1

//! Commands related to workflow management in Nexus.
//!
//! - [`WorkflowActions::publish`] to publish a [`DagSpec`] instance to Nexus.
//! - [`WorkflowActions::inspect_execution`] to monitor the execution of a DAG.
//!
//! Runtime executions begin when an occurrence created through
//! [`Scheduler`] is dispatched.
//!
//! [`Scheduler`]: crate::nexus::scheduler::Scheduler

mod history;

#[cfg(test)]
use crate::move_bindings::tool::era::V1 as ToolWitnessV1;
#[cfg(feature = "walrus")]
use crate::walrus::StorageConf;
use {
    self::history::{fetch_shared_object_history, version_or_none, ObjectHistoryRequest},
    crate::{
        events::{NexusEvent, NexusEventDecoder, NexusEventKind},
        move_bindings::{
            interface::{
                dag as dag_move,
                era::V1 as InterfaceWitnessV1,
                graph::{self as graph_move, RuntimeVertex},
                meta_schema::MetaSchema,
                onchain_tool_result::{OnchainToolResult, OnchainToolResultInnerV1},
                payment::{ExecutionPaymentInnerV1, ExecutionPaymentVertexLock},
                verifier::{FailureEvidenceKind, ToolVerifierMode},
            },
            move_std::type_name::TypeName,
            sui_framework::{clock::Clock as SuiClock, linked_table, object::ID, vec_map::VecMap},
            workflow::{
                era::V1 as WorkflowWitnessV1,
                execution::{self as execution_move, DAGExecution, DAGExecutionInnerV1, DAGWalk},
                execution_events::{
                    EndStateReachedEvent,
                    ExecutionFinishedEvent,
                    TerminalErrEvalRecordedEvent,
                },
                execution_failure::WorkflowFailureClass,
                invocation_adapter::InvocationLockedEvent,
            },
            FiniteCredits,
            TimePass,
        },
        move_boundary,
        nexus::{client::NexusClient, crawler::Crawler, error::NexusError, tap},
        sui,
        transactions::{
            dag,
            invocation::{self, InvocationPolicyCall},
        },
        types::{DagSpec, NexusContext, NexusData, PackageRole, SharedRoot, ToolRef},
    },
    anyhow::{anyhow, bail},
    sha2::{Digest as _, Sha256},
    std::{
        collections::{BTreeMap, HashMap},
        sync::Arc,
    },
    tokio::{
        sync::mpsc::{unbounded_channel, UnboundedReceiver},
        task::JoinHandle,
        time::{Duration, Instant},
    },
};

const DEFAULT_EXECUTION_INSPECTION_TIMEOUT: Duration = Duration::from_secs(60 * 60);
const DEFAULT_EXECUTION_INSPECTION_POLL_INTERVAL: Duration = Duration::from_secs(1);
pub const EXPIRED_WALK_NOT_DOUBLE_TIMEOUT_EXPIRED_REASON: &str =
    "walk is not double timeout expired";
pub const EXPIRED_WALK_ALREADY_TERMINAL_REASON: &str = "walk is already terminal";
pub const EXPIRED_WALK_OTHER_PAYMENT_LOCKS_REASON: &str = "other walks still hold payment locks";

#[derive(Clone, Debug)]
pub struct PublishResult {
    pub tx_digest: sui::types::Digest,
    pub tx_checkpoint: u64,
    pub dag_object_id: sui::types::Address,
}

/// One Tool owner accepted policy selected for Invocation authorization.
///
/// Canonical policy variants need only their economic object IDs. Use
/// [`InvocationPolicy::Custom`] for an owner policy whose `get_invocation`
/// function accepts arbitrary [`crate::transactions::dag::OnchainToolArgument`]
/// values.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum InvocationPolicy {
    FixedPrice,
    FiniteCredits,
    TimePass,
    Custom(InvocationPolicyCall),
}

/// Confirmed result of authorizing one exact Tool Invocation.
#[derive(Clone, Debug)]
pub struct AuthorizeInvocationResult {
    pub tx_digest: sui::types::Digest,
    pub tx_checkpoint: u64,
    /// Complete accounting identity emitted by the successful lock.
    pub lock: InvocationLockedEvent,
}

/// Published DAG interface needed to author one execution.
#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
pub struct DagSnapshot {
    pub dag_id: sui::types::Address,
    pub vertex_count: u64,
    pub entry_groups: BTreeMap<String, BTreeMap<String, Vec<String>>>,
    pub vertex_meta_schemas: BTreeMap<String, MetaSchema>,
}

/// One expired lock with the current object reference needed for exact refund.
///
/// [`ExecutionPaymentVertexLock`] stores identity but not the version and digest
/// required by a receiving PTB input, so it cannot replace this resolved value.
#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
pub struct InvocationAbortCandidate {
    pub walk_index: usize,
    pub vertex: RuntimeVertex,
    pub payment_vertex_key: Vec<u8>,
    pub invocation_ref: sui::types::ObjectReference,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct AbortExpiredExecutionResult {
    pub tx_digest: sui::types::Digest,
    pub tx_checkpoint: u64,
    pub dag_id: sui::types::Address,
    pub dag_execution_id: sui::types::Address,
    pub selected_candidate: InvocationAbortCandidate,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct AbortExecutionResult {
    pub tx_digest: sui::types::Digest,
    pub tx_checkpoint: u64,
    pub dag_id: sui::types::Address,
    pub dag_execution_id: sui::types::Address,
    pub cleaned_broken_onchain_results: Vec<BrokenOnchainToolResultCleanup>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SettleCommittedToolResultParams {
    pub dag_execution_id: sui::types::Address,
    pub walk_index: u64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ResolveExpiredWalkParams {
    pub dag_execution_id: sui::types::Address,
    pub walk_index: u64,
    pub invocation_id: Option<sui::types::Address>,
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ExpiredWalkResolutionKind {
    Settled {
        expected_vertex: RuntimeVertex,
        invocation_ref: sui::types::ObjectReference,
    },
    SettledOnchainResult {
        result_ref: sui::types::ObjectReference,
        expected_vertex: RuntimeVertex,
        invocation_ref: sui::types::ObjectReference,
        tool_witness_id: sui::types::Address,
        finalize_tx_digest: Vec<u8>,
    },
    Aborted,
    AbortedWithInvocation {
        selected_candidate: InvocationAbortCandidate,
    },
    Skipped {
        reason: String,
    },
}

impl ExpiredWalkResolutionKind {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Settled { .. } => "settled",
            Self::SettledOnchainResult { .. } => "settled_onchain_result",
            Self::Aborted => "aborted",
            Self::AbortedWithInvocation { .. } => "aborted_with_invocation",
            Self::Skipped { .. } => "skipped",
        }
    }

    pub fn selected_candidate(&self) -> Option<&InvocationAbortCandidate> {
        match self {
            Self::AbortedWithInvocation { selected_candidate } => Some(selected_candidate),
            _ => None,
        }
    }

    pub fn skip_reason(&self) -> Option<&str> {
        match self {
            Self::Skipped { reason } => Some(reason),
            _ => None,
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ExpiredWalkResolutionPlan {
    pub dag_id: sui::types::Address,
    pub dag_execution_id: sui::types::Address,
    pub walk_index: u64,
    pub kind: ExpiredWalkResolutionKind,
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
pub struct ExpiredWalkResolutionResult {
    pub tx_digest: Option<sui::types::Digest>,
    pub tx_checkpoint: Option<u64>,
    pub dag_id: sui::types::Address,
    pub dag_execution_id: sui::types::Address,
    pub walk_index: u64,
    pub resolution_kind: ExpiredWalkResolutionKind,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SettleCommittedToolResultByLeaderParams {
    pub dag_execution_id: sui::types::Address,
    pub leader_cap_id: sui::types::Address,
    pub walk_index: u64,
    /// Exact active vertex when merging failed on-chain Tool evidence.
    pub expected_vertex: Option<RuntimeVertex>,
    /// Sanitized failed on-chain Tool reason. Must be paired with `expected_vertex`.
    pub failed_onchain_tool_reason: Option<Vec<u8>>,
    pub commit_tx_digest: Vec<u8>,
    pub commit_gas_charge: u64,
    pub settlement_gas_charge: u64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RecordCommittedToolResultGasChargeParams {
    pub dag_execution_id: sui::types::Address,
    pub leader_cap_id: sui::types::Address,
    pub walk_index: u64,
    /// Exact active vertex when creating or merging failed on-chain tool evidence.
    pub expected_vertex: Option<RuntimeVertex>,
    /// Sanitized failed on-chain tool reason. Must be paired with `expected_vertex`.
    pub failed_onchain_tool_reason: Option<Vec<u8>>,
    pub commit_tx_digest: Vec<u8>,
    pub commit_gas_charge: u64,
    pub settlement_gas_charge: u64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CommittedToolResultSettlementResult {
    pub tx_digest: sui::types::Digest,
    pub tx_checkpoint: u64,
    pub dag_id: sui::types::Address,
    pub dag_execution_id: sui::types::Address,
    pub walk_index: u64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BrokenOnchainToolResultCleanup {
    pub walk_index: u64,
    pub result_ref: sui::types::ObjectReference,
    pub expected_vertex: RuntimeVertex,
    pub tool_witness_id: sui::types::Address,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RecordCommittedToolResultGasChargeResult {
    pub tx_digest: sui::types::Digest,
    pub tx_checkpoint: u64,
    pub dag_execution_id: sui::types::Address,
    pub leader_cap_id: sui::types::Address,
    pub walk_index: u64,
}

/// Narrowed committed result view for off chain freshness checks.
///
/// This is separate from [`DAGExecution`] because callers only need committed result state and
/// should not read or decode the full execution object for early wake decisions.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct CommittedToolResultView {
    pub expected_vertex: RuntimeVertex,
    pub primary_failure_evidence_kind: Option<FailureEvidenceKind>,
    pub secondary_failure_evidence_kind: Option<FailureEvidenceKind>,
    pub current_leader_cap_id: sui::types::Address,
    pub has_finalized_onchain_payload: bool,
    pub leader_records: Vec<CommittedToolResultLeaderRecordView>,
}

impl CommittedToolResultView {
    pub fn leader_record(
        &self,
        leader_cap_id: sui::types::Address,
    ) -> Option<&CommittedToolResultLeaderRecordView> {
        self.leader_records
            .iter()
            .find(|record| record.leader_cap_id == leader_cap_id)
    }
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct CommittedToolResultLeaderRecordView {
    pub leader_cap_id: sui::types::Address,
    pub commit_tx_digest: Vec<u8>,
    pub recipient: sui::types::Address,
    pub commit_gas_charge: Option<u64>,
    pub settlement_gas_charge: Option<u64>,
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct InsufficientSettlementMarkerView {
    pub walks: Vec<u64>,
}

pub enum OnchainToolResultState {
    NoResult,
    Finalized {
        result: OnchainToolResultInnerV1,
        object_ref: sui::types::ObjectReference,
    },
    Committed {
        committed_result: CommittedToolResultView,
    },
    InsufficientSettlement {
        committed_result: Option<CommittedToolResultView>,
        marker: InsufficientSettlementMarkerView,
    },
    InvalidEmpty {
        result_id: sui::types::Address,
        object_ref: sui::types::ObjectReference,
    },
}

impl OnchainToolResultState {
    /// Return the fetched generated result and object metadata that let a
    /// retrying leader skip execution and build the consume+settle transaction.
    pub fn consume_ready_result(
        &self,
    ) -> Option<(&OnchainToolResultInnerV1, &sui::types::ObjectReference)> {
        match self {
            Self::Finalized { result, object_ref } => Some((result, object_ref)),
            Self::NoResult
            | Self::Committed { .. }
            | Self::InsufficientSettlement { .. }
            | Self::InvalidEmpty { .. } => None,
        }
    }

    /// Build the mutable shared-object input expected by
    /// `consume_on_chain_tool_result_for_walk`.
    pub fn consume_object_input(&self) -> Option<talus_sui_move_call::CallArg> {
        self.consume_ready_result().map(|(_, object_ref)| {
            talus_sui_move_call::CallArg::Shared(sui::types::SharedInput::new(
                *object_ref.object_id(),
                object_ref.version(),
                true,
            ))
        })
    }
}

fn onchain_tool_result_is_finalized(result: &OnchainToolResultInnerV1) -> bool {
    result.finalized
        && result.tag.as_option().is_some()
        && result.named_payload.as_option().is_some()
        && result.finalize_tx_digest.as_option().is_some()
        && result.finalize_recipient.as_option().is_some()
}

impl From<execution_move::CommittedToolResult> for CommittedToolResultView {
    fn from(value: execution_move::CommittedToolResult) -> Self {
        Self {
            expected_vertex: value.expected_vertex,
            primary_failure_evidence_kind: value.primary_failure_evidence_kind.into_option(),
            secondary_failure_evidence_kind: value.secondary_failure_evidence_kind.into_option(),
            current_leader_cap_id: value.current_leader_cap_id.bytes,
            has_finalized_onchain_payload: value.has_finalized_onchain_payload,
            leader_records: value
                .leader_records
                .contents
                .into_iter()
                .map(|entry| CommittedToolResultLeaderRecordView {
                    leader_cap_id: entry.key.bytes,
                    commit_tx_digest: entry.value.commit_tx_digest,
                    recipient: entry.value.recipient,
                    commit_gas_charge: entry.value.commit_gas_charge.into_option(),
                    settlement_gas_charge: entry.value.settlement_gas_charge.into_option(),
                })
                .collect(),
        }
    }
}

pub struct InspectExecutionResult {
    pub next_event: UnboundedReceiver<NexusEvent>,
    pub poller: JoinHandle<Result<(), NexusError>>,
}

/// Controls execution-object inspection polling interval and its total
/// wall-clock budget.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct InspectExecutionOptions {
    pub timeout: Duration,
    pub poll_interval: Duration,
}

impl Default for InspectExecutionOptions {
    fn default() -> Self {
        Self {
            timeout: DEFAULT_EXECUTION_INSPECTION_TIMEOUT,
            poll_interval: DEFAULT_EXECUTION_INSPECTION_POLL_INTERVAL,
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WorkflowExecutionTerminalState {
    Succeeded,
    Failed,
    Aborted,
    NoWalkOutcome,
}

#[derive(Clone, Debug)]
pub struct ResolvedEndState {
    pub event: EndStateReachedEvent,
    pub resolved_ports_to_data: HashMap<String, NexusData>,
}

#[derive(Clone, Debug)]
pub struct InspectExecutionCompletionResult {
    pub terminal_state: WorkflowExecutionTerminalState,
    pub execution_finished: ExecutionFinishedEvent,
    pub end_states: Vec<ResolvedEndState>,
    pub terminal_err_eval_recordings: Vec<TerminalErrEvalRecordedEvent>,
    pub events: Vec<NexusEvent>,
}

pub struct ExecutionCostResult {
    pub payment_id: sui::types::Address,
    pub max_budget_mist: u64,
    pub locked_budget_mist: u64,
    pub consumed: u64,
    pub outstanding_locks: u64,
    pub accomplished: bool,
    pub refunded: bool,
}

/// Payment accounting and the invocation identities holding its outstanding locks.
#[non_exhaustive]
pub struct ExecutionCostDetails {
    pub summary: ExecutionCostResult,
    pub outstanding_invocation_ids: Vec<sui::types::Address>,
}

pub struct WorkflowActions {
    pub(super) client: NexusClient,
}

struct ResolvedExecution {
    context: Arc<NexusContext>,
    object: crate::nexus::crawler::Response<DAGExecutionInnerV1>,
}

async fn fetch_execution(
    client: &NexusClient,
    execution_id: sui::types::Address,
    required_roots: &[SharedRoot],
) -> Result<ResolvedExecution, NexusError> {
    let context = if required_roots.is_empty() {
        client.context_for_object(execution_id).await?
    } else {
        client
            .context_for_object_with_roots(execution_id, required_roots)
            .await?
    };
    let object = client
        .state_resolver()
        .load_inner::<DAGExecution, WorkflowWitnessV1, DAGExecutionInnerV1>(execution_id, &context)
        .await?;
    Ok(ResolvedExecution { context, object })
}

async fn fetch_dag_in_context(
    client: &NexusClient,
    context: &NexusContext,
    dag_id: sui::types::Address,
) -> Result<crate::nexus::crawler::Response<dag_move::DAGInnerV1>, NexusError> {
    client
        .state_resolver()
        .load_inner::<dag_move::DAG, InterfaceWitnessV1, dag_move::DAGInnerV1>(dag_id, context)
        .await
}

fn event_execution_id(event: &NexusEventKind) -> Option<sui::types::Address> {
    match event {
        NexusEventKind::AgentSkillExecutionRequested(e) => Some(e.execution_id),
        NexusEventKind::OccurrenceDispatched(e) => Some(e.execution_id.into()),
        NexusEventKind::ExecutionPaymentFeesRecorded(e) => Some(e.execution_id),
        NexusEventKind::ExecutionPaymentToolCostSnapshotted(e) => Some(e.execution_id),
        NexusEventKind::InvocationAuthorizationRequired(e) => Some(e.execution.into()),
        NexusEventKind::InvocationLocked(e) => Some(e.execution.bytes),
        NexusEventKind::InvocationSettled(e) => Some(e.execution.bytes),
        NexusEventKind::RequestWalkExecution(e) => Some(e.execution.into()),
        NexusEventKind::CommittedToolResult(e) => Some(e.execution.into()),
        NexusEventKind::WalkAdvanced(e) => Some(e.execution.into()),
        NexusEventKind::WalkFailed(e) => Some(e.execution.into()),
        NexusEventKind::SubmissionFailureEvidenceRecorded(e) => Some(e.execution.into()),
        NexusEventKind::TerminalErrEvalRecorded(e) => Some(e.execution.into()),
        NexusEventKind::ToolVerificationResolved(e) => Some(e.execution.into()),
        NexusEventKind::WalkPendingAbort(e) => Some(e.execution.into()),
        NexusEventKind::WalkAborted(e) => Some(e.execution.into()),
        NexusEventKind::WalkCancelled(e) => Some(e.execution.into()),
        NexusEventKind::EndStateReached(e) => Some(e.execution.into()),
        NexusEventKind::ExecutionFinished(e) => Some(e.execution.into()),
        NexusEventKind::ExecutionPaymentRefilled(e) => Some(e.execution_id),
        _ => None,
    }
}

fn is_transient_inspection_error(error: &NexusError) -> bool {
    let NexusError::Rpc(error) = error else {
        return false;
    };

    error.chain().any(|source| {
        source
            .downcast_ref::<tonic::Status>()
            .is_some_and(|status| {
                matches!(
                    status.code(),
                    tonic::Code::Unavailable
                        | tonic::Code::DeadlineExceeded
                        | tonic::Code::ResourceExhausted
                        | tonic::Code::Aborted
                )
            })
    })
}

async fn fetch_execution_update_events(
    crawler: &Crawler,
    event_decoder: &NexusEventDecoder,
    request: ObjectHistoryRequest<'_>,
) -> Result<Vec<NexusEvent>, NexusError> {
    let dag_execution_id = request.object_id;
    let updates = fetch_shared_object_history(crawler, request).await?;

    let mut events = Vec::new();
    for update in updates {
        for (index, event) in update.events.into_iter().enumerate() {
            let Some(event) = event_decoder
                .decode_sui_event(index as u64, update.digest, &event)
                .await
                .map_err(|error| {
                    NexusError::Parsing(anyhow::Error::new(error).context(format!(
                        "Could not decode event {index} from transaction '{}' while reconstructing execution '{dag_execution_id}'",
                        update.digest
                    )))
                })?
            else {
                continue;
            };
            if event_execution_id(&event.data) == Some(dag_execution_id) {
                events.push(event);
            }
        }
    }
    Ok(events)
}

#[cfg(feature = "walrus")]
fn terminal_state_from_execution_finished(
    execution_finished: &ExecutionFinishedEvent,
) -> WorkflowExecutionTerminalState {
    if execution_finished.was_aborted {
        WorkflowExecutionTerminalState::Aborted
    } else if execution_finished.has_any_walk_failed {
        WorkflowExecutionTerminalState::Failed
    } else if execution_finished.has_any_walk_succeeded {
        WorkflowExecutionTerminalState::Succeeded
    } else {
        WorkflowExecutionTerminalState::NoWalkOutcome
    }
}

#[cfg(feature = "walrus")]
async fn build_execution_completion_result(
    events: Vec<NexusEvent>,
    dag_execution_id: sui::types::Address,
    storage_conf: &StorageConf,
) -> Result<InspectExecutionCompletionResult, NexusError> {
    let mut end_states = Vec::new();
    let mut terminal_err_eval_recordings = Vec::new();
    let mut execution_finished = None;

    for event in &events {
        match &event.data {
            NexusEventKind::EndStateReached(end_state) => {
                let resolved_ports_to_data = end_state
                    .variant_ports_to_data
                    .clone()
                    .fetch_all(storage_conf)
                    .await
                    .map_err(|e| {
                        NexusError::Storage(anyhow!(
                            "Failed to fetch output data for execution '{dag_execution_id}': {e}"
                        ))
                    })?
                    .into_map();

                end_states.push(ResolvedEndState {
                    event: end_state.clone(),
                    resolved_ports_to_data,
                });
            }
            NexusEventKind::TerminalErrEvalRecorded(recorded) => {
                terminal_err_eval_recordings.push(recorded.clone());
            }
            NexusEventKind::ExecutionFinished(finished) => {
                execution_finished = Some(finished.clone());
            }
            _ => {}
        }
    }

    let execution_finished = execution_finished.ok_or_else(|| {
        NexusError::Channel(anyhow!(
            "ExecutionFinished event not found while inspecting DAG execution '{dag_execution_id}'"
        ))
    })?;

    Ok(InspectExecutionCompletionResult {
        terminal_state: terminal_state_from_execution_finished(&execution_finished),
        execution_finished,
        end_states,
        terminal_err_eval_recordings,
        events,
    })
}

pub fn dag_vertex_requires_tool_verification(vertex: &graph_move::VertexInfo) -> bool {
    vertex.verifier_mode != ToolVerifierMode::None
}

/// Fetch every vertex in a DAG.
///
/// This function enumerates the complete vertex table. Use
/// [`fetch_dag_vertex_bcs`] or [`fetch_dag_vertices_by_keys_bcs`] when the
/// required vertex keys are already known.
///
/// # Errors
///
/// Returns an error when the vertex table cannot be enumerated or a field
/// cannot be fetched or decoded.
pub async fn fetch_dag_vertices_bcs(
    crawler: &Crawler,
    dag: &dag_move::DAGInnerV1,
) -> anyhow::Result<HashMap<graph_move::Vertex, graph_move::VertexInfo>> {
    if dag.vertices.size() == 0 {
        return Ok(HashMap::new());
    }
    Ok(crawler
        .get_dynamic_fields::<
            graph_move::Vertex,
            linked_table::Node<graph_move::Vertex, graph_move::VertexInfo>,
        >(dag.vertices.id(), dag.vertices.size())
        .await?
        .into_iter()
        .map(|(vertex, node)| (vertex, node.value))
        .collect())
}

/// Fetches every DAG vertex by traversing its stored linked table.
///
/// Each field ID is derived from the preceding vertex and read from the exact
/// object service. This path needs no collection index and validates the size,
/// links, and tail recorded by the table.
///
/// # Errors
///
/// Returns an error when the table links are incomplete, cyclic, inconsistent,
/// or a field cannot be fetched and decoded.
pub async fn fetch_dag_vertices_from_links_bcs(
    crawler: &Crawler,
    context: &NexusContext,
    dag: &dag_move::DAGInnerV1,
) -> anyhow::Result<HashMap<graph_move::Vertex, graph_move::VertexInfo>> {
    let expected_size = dag.vertices.size();
    let head = dag.vertices.head.cloned_option();
    let tail = dag.vertices.tail.cloned_option();
    if expected_size == 0 {
        if head.is_some() || tail.is_some() {
            bail!("Empty DAG vertex table records a head or tail");
        }
        return Ok(HashMap::new());
    }

    let mut current = head.ok_or_else(|| anyhow!("DAG vertex table has no head"))?;
    let key_type = crate::move_bindings::type_tag::<graph_move::Vertex>(context);
    let mut previous = None;
    let mut vertices = HashMap::with_capacity(expected_size);

    for position in 0..expected_size {
        if vertices.contains_key(&current) {
            bail!("DAG vertex table contains a cycle at position {position}");
        }
        let node = crawler
            .get_dynamic_field_by_key::<
                graph_move::Vertex,
                linked_table::Node<graph_move::Vertex, graph_move::VertexInfo>,
            >(dag.vertices.id(), current.clone(), &key_type)
            .await?
            .ok_or_else(|| anyhow!("DAG vertex table is missing position {position}"))?;
        if node.prev.cloned_option() != previous {
            bail!("DAG vertex table has an invalid previous link at position {position}");
        }

        let next = node.next.cloned_option();
        previous = Some(current.clone());
        vertices.insert(current.clone(), node.value);

        if position + 1 == expected_size {
            if next.is_some() {
                bail!("DAG vertex table exceeds its declared size");
            }
        } else {
            current = next.ok_or_else(|| {
                anyhow!("DAG vertex table ends before its declared size at position {position}")
            })?;
        }
    }

    if previous != tail {
        bail!("DAG vertex table tail does not match its final link");
    }
    Ok(vertices)
}

/// Fetch one vertex from a DAG by its typed key.
///
/// The lookup derives the dynamic field identifier from the vertex key and
/// the package graph in [`NexusContext`]. An absent vertex returns
/// [`None`].
///
/// # Errors
///
/// Returns an error when the key cannot be encoded or the field cannot be
/// fetched, validated, or decoded.
pub async fn fetch_dag_vertex_bcs(
    crawler: &Crawler,
    context: &NexusContext,
    dag: &dag_move::DAGInnerV1,
    vertex: &graph_move::Vertex,
) -> anyhow::Result<Option<graph_move::VertexInfo>> {
    crawler
        .get_dynamic_field_by_key::<
            graph_move::Vertex,
            linked_table::Node<graph_move::Vertex, graph_move::VertexInfo>,
        >(
            dag.vertices.id(),
            vertex.clone(),
            &crate::move_bindings::type_tag::<graph_move::Vertex>(context),
        )
        .await
        .map(|node| node.map(|node| node.value))
}

/// Fetch the requested DAG vertices by their typed keys.
///
/// Duplicate keys are fetched once. Present vertices are returned by key, and
/// absent vertices are omitted. The amount of work depends on the distinct
/// requested keys rather than the total number of vertices in the DAG.
///
/// # Errors
///
/// Returns an error when a key cannot be encoded or a field cannot be fetched,
/// validated, or decoded.
pub async fn fetch_dag_vertices_by_keys_bcs<I>(
    crawler: &Crawler,
    context: &NexusContext,
    dag: &dag_move::DAGInnerV1,
    vertices: I,
) -> anyhow::Result<HashMap<graph_move::Vertex, graph_move::VertexInfo>>
where
    I: IntoIterator<Item = graph_move::Vertex>,
{
    Ok(crawler
        .get_dynamic_fields_by_keys::<
            graph_move::Vertex,
            linked_table::Node<graph_move::Vertex, graph_move::VertexInfo>,
            _,
        >(
            dag.vertices.id(),
            vertices,
            &crate::move_bindings::type_tag::<graph_move::Vertex>(context),
        )
        .await?
        .into_iter()
        .map(|(vertex, node)| (vertex, node.value))
        .collect())
}

pub(crate) async fn fetch_dag_snapshot(
    client: &NexusClient,
    context: &NexusContext,
    dag_id: sui::types::Address,
) -> anyhow::Result<DagSnapshot> {
    let dag = client
        .state_resolver()
        .load_inner::<dag_move::DAG, InterfaceWitnessV1, dag_move::DAGInnerV1>(dag_id, context)
        .await?;
    let entry_groups = dag
        .data
        .entry_groups
        .contents
        .iter()
        .map(|entry_group| {
            let vertices = entry_group
                .value
                .contents
                .iter()
                .map(|vertex| {
                    let mut ports = vertex
                        .value
                        .contents
                        .iter()
                        .map(|port| port.name.as_str().to_owned())
                        .collect::<Vec<_>>();
                    ports.sort();
                    (vertex.key.name.as_str().to_owned(), ports)
                })
                .collect::<BTreeMap<_, _>>();
            (entry_group.key.name.as_str().to_owned(), vertices)
        })
        .collect();
    let vertices = fetch_dag_vertices_from_links_bcs(client.crawler(), context, &dag.data).await?;
    let vertex_meta_schemas = vertices
        .into_iter()
        .map(|(vertex, info)| {
            let name = vertex.name.as_str().to_owned();
            let schema = info.meta_schema.into_option().ok_or_else(|| {
                anyhow!("DAG '{dag_id}' vertex '{name}' is missing its immutable MetaSchema")
            })?;
            Ok((name, schema))
        })
        .collect::<anyhow::Result<BTreeMap<_, _>>>()?;

    Ok(DagSnapshot {
        dag_id,
        vertex_count: dag.data.vertices.size().try_into()?,
        entry_groups,
        vertex_meta_schemas,
    })
}

/// Fetch the [`CommittedToolResultView`] stored for one execution walk.
///
/// The lookup derives the dynamic field identifier from the walk key and the
/// package graph in [`NexusContext`]. An absent result returns [`None`].
///
/// # Errors
///
/// Returns an error when the key cannot be encoded or the field cannot be
/// fetched, validated, or decoded.
pub async fn fetch_committed_tool_result_for_walk(
    crawler: &Crawler,
    context: &NexusContext,
    execution_id: sui::types::Address,
    walk_index: u64,
) -> anyhow::Result<Option<CommittedToolResultView>> {
    crawler
        .get_dynamic_field_by_key::<
            execution_move::CommittedToolResultKey,
            execution_move::CommittedToolResult,
        >(
            execution_id,
            execution_move::CommittedToolResultKey { walk_index },
            &crate::move_bindings::type_tag::<execution_move::CommittedToolResultKey>(context),
        )
        .await
        .map(|value| value.map(CommittedToolResultView::from))
}

/// Decodes the committed result written by one finalized transaction.
///
/// The value is read from the output objects in `executed`; this function does
/// not issue a Sui request. The dynamic field identity, parent execution, key,
/// value type, and writing transaction are all validated before decoding. An
/// absent output returns [`None`].
///
/// # Errors
///
/// Returns an error when the finalized response has a different transaction
/// identity or contains an invalid committed result output.
pub fn committed_tool_result_from_finalized_output(
    crawler: &Crawler,
    context: &NexusContext,
    executed: &sui::grpc::ExecutedTransaction,
    transaction: sui::types::Digest,
    execution_id: sui::types::Address,
    walk_index: u64,
) -> anyhow::Result<Option<CommittedToolResultView>> {
    let key = execution_move::CommittedToolResultKey { walk_index };
    let outputs = crawler.transaction_dynamic_field_outputs::<
        execution_move::CommittedToolResultKey,
        execution_move::CommittedToolResult,
    >(
        executed,
        transaction,
        &key,
        &crate::move_bindings::type_tag::<execution_move::CommittedToolResultKey>(context),
        &crate::move_bindings::type_tag::<execution_move::CommittedToolResult>(context),
    )?;
    let mut matching = outputs
        .into_iter()
        .filter(|output| output.object_id == execution_id);
    let output = matching.next();
    if matching.next().is_some() {
        bail!(
            "Transaction '{transaction}' wrote more than one committed result for execution \
             '{execution_id}' walk {walk_index}"
        );
    }

    Ok(output.map(|output| CommittedToolResultView::from(output.data)))
}

pub async fn inspect_expired_walk_resolution(
    client: &NexusClient,
    params: ResolveExpiredWalkParams,
) -> anyhow::Result<ExpiredWalkResolutionPlan> {
    let clock = client
        .crawler()
        .get_object::<SuiClock>(move_boundary::CLOCK_OBJECT_ID)
        .await?
        .data;
    inspect_expired_walk_resolution_at(client, params, clock.timestamp_ms).await
}

pub async fn inspect_expired_walk_resolution_at(
    client: &NexusClient,
    params: ResolveExpiredWalkParams,
    clock_ms: u64,
) -> anyhow::Result<ExpiredWalkResolutionPlan> {
    let resolved = fetch_execution(client, params.dag_execution_id, &[]).await?;
    inspect_expired_walk_in_execution(client, params, clock_ms, &resolved).await
}

async fn inspect_expired_walk_in_execution(
    client: &NexusClient,
    params: ResolveExpiredWalkParams,
    clock_ms: u64,
    resolved: &ResolvedExecution,
) -> anyhow::Result<ExpiredWalkResolutionPlan> {
    let context = &resolved.context;
    let execution = &resolved.object.data;
    let crawler = client.crawler();
    let Some(walk) = usize::try_from(params.walk_index)
        .ok()
        .and_then(|index| execution.walks.get(index))
    else {
        return Ok(ExpiredWalkResolutionPlan {
            dag_id: execution.dag_id(),
            dag_execution_id: params.dag_execution_id,
            walk_index: params.walk_index,
            kind: ExpiredWalkResolutionKind::Skipped {
                reason: "walk index no longer exists in execution".to_string(),
            },
        });
    };

    let Some(timeout_vertex) = walk.timeout_expired_vertex(clock_ms) else {
        return Ok(ExpiredWalkResolutionPlan {
            dag_id: execution.dag_id(),
            dag_execution_id: params.dag_execution_id,
            walk_index: params.walk_index,
            kind: ExpiredWalkResolutionKind::Skipped {
                reason: unresolved_timeout_skip_reason(walk).to_string(),
            },
        });
    };

    match fetch_onchain_tool_result_state_for_walk(
        client,
        context,
        params.dag_execution_id,
        params.walk_index,
    )
    .await?
    {
        OnchainToolResultState::Committed { .. } => {
            let invocation_ref = fetch_invocation_for_vertex(
                client,
                context,
                params.dag_execution_id,
                execution.dag_id(),
                timeout_vertex,
            )
            .await?;
            return Ok(ExpiredWalkResolutionPlan {
                dag_id: execution.dag_id(),
                dag_execution_id: params.dag_execution_id,
                walk_index: params.walk_index,
                kind: ExpiredWalkResolutionKind::Settled {
                    expected_vertex: timeout_vertex.clone(),
                    invocation_ref,
                },
            });
        }
        OnchainToolResultState::Finalized { result, object_ref } => {
            let dag = fetch_dag_in_context(client, context, execution.dag_id()).await?;
            let vertex_info =
                fetch_dag_vertex_bcs(crawler, context, &dag.data, timeout_vertex.vertex())
                    .await?
                    .ok_or_else(|| {
                        anyhow!(
                            "DAG vertex '{}' missing from fetched DAG",
                            timeout_vertex.vertex_name()
                        )
                    })?;
            let kind = finalized_onchain_result_resolution_kind(
                client,
                &vertex_info,
                timeout_vertex.clone(),
                fetch_invocation_for_vertex(
                    client,
                    context,
                    params.dag_execution_id,
                    execution.dag_id(),
                    timeout_vertex,
                )
                .await?,
                &result,
                object_ref,
            )
            .await?;
            return Ok(ExpiredWalkResolutionPlan {
                dag_id: execution.dag_id(),
                dag_execution_id: params.dag_execution_id,
                walk_index: params.walk_index,
                kind,
            });
        }
        OnchainToolResultState::InsufficientSettlement { .. } => {
            return Ok(ExpiredWalkResolutionPlan {
                dag_id: execution.dag_id(),
                dag_execution_id: params.dag_execution_id,
                walk_index: params.walk_index,
                kind: ExpiredWalkResolutionKind::Skipped {
                    reason: "walk has committed result with insufficient settlement".to_string(),
                },
            });
        }
        OnchainToolResultState::InvalidEmpty {
            result_id,
            object_ref,
        } => {
            return Ok(ExpiredWalkResolutionPlan {
                dag_id: execution.dag_id(),
                dag_execution_id: params.dag_execution_id,
                walk_index: params.walk_index,
                kind: ExpiredWalkResolutionKind::Skipped {
                    reason: format!(
                        "stored on-chain tool result {result_id}@{} is not finalized",
                        object_ref.version()
                    ),
                },
            });
        }
        OnchainToolResultState::NoResult => {}
    }

    let Some(abort_vertex) = walk.abortable_timeout_expired_vertex(clock_ms) else {
        return Ok(ExpiredWalkResolutionPlan {
            dag_id: execution.dag_id(),
            dag_execution_id: params.dag_execution_id,
            walk_index: params.walk_index,
            kind: ExpiredWalkResolutionKind::Skipped {
                reason: format!(
                    "timeout-expired walk at vertex '{}' is not abortable without a committed result",
                    timeout_vertex.vertex_name()
                ),
            },
        });
    };

    let payment =
        tap::fetch_execution_payment_for_execution(client, context, params.dag_execution_id)
            .await?
            .data;
    let dag = fetch_dag_in_context(client, context, execution.dag_id()).await?;
    let vertex_info = fetch_dag_vertex_bcs(crawler, context, &dag.data, abort_vertex.vertex())
        .await?
        .ok_or_else(|| {
            anyhow!(
                "DAG vertex '{}' missing from fetched DAG",
                abort_vertex.vertex_name()
            )
        })?;
    let tool_fqn = vertex_info.kind.tool_fqn()?;
    let vertex_key = payment_vertex_key(params.dag_execution_id, abort_vertex, &tool_fqn)?;
    let locked = payment
        .locked_vertices
        .iter()
        .find(|lock| lock.vertex_key == vertex_key);

    let Some(locked) = locked else {
        return Ok(ExpiredWalkResolutionPlan {
            dag_id: execution.dag_id(),
            dag_execution_id: params.dag_execution_id,
            walk_index: params.walk_index,
            kind: if payment.locked_vertices.is_empty() {
                ExpiredWalkResolutionKind::Aborted
            } else {
                ExpiredWalkResolutionKind::Skipped {
                    reason: EXPIRED_WALK_OTHER_PAYMENT_LOCKS_REASON.to_owned(),
                }
            },
        });
    };

    let invocation_ref = crawler
        .get_object_metadata(locked.invocation_id.bytes)
        .await?
        .object_ref();
    let selected_candidate = InvocationAbortCandidate {
        walk_index: usize::try_from(params.walk_index)?,
        vertex: abort_vertex.clone(),
        payment_vertex_key: vertex_key,
        invocation_ref,
    };
    if params
        .invocation_id
        .is_some_and(|expected| expected != *selected_candidate.invocation_ref.object_id())
    {
        return Err(anyhow!(
            "Requested Invocation does not match the expired walk"
        ));
    }

    Ok(ExpiredWalkResolutionPlan {
        dag_id: execution.dag_id(),
        dag_execution_id: params.dag_execution_id,
        walk_index: params.walk_index,
        kind: ExpiredWalkResolutionKind::AbortedWithInvocation { selected_candidate },
    })
}

fn unresolved_timeout_skip_reason(walk: &DAGWalk) -> &'static str {
    match walk {
        DAGWalk::Active { .. } | DAGWalk::PendingSettlement { .. } => {
            EXPIRED_WALK_NOT_DOUBLE_TIMEOUT_EXPIRED_REASON
        }
        _ => EXPIRED_WALK_ALREADY_TERMINAL_REASON,
    }
}

/// Select one actionable expired walk, preserving results before refunding locks.
/// The caller must inspect again after each confirmed mutation because resolving
/// one walk can advance or finish other walks in the execution.
pub async fn inspect_expired_execution_resolution_at(
    client: &NexusClient,
    execution_id: sui::types::Address,
    clock_ms: u64,
) -> anyhow::Result<Option<ExpiredWalkResolutionPlan>> {
    let execution = fetch_execution(client, execution_id, &[]).await?;
    let mut selected: Option<ExpiredWalkResolutionPlan> = None;
    for (index, walk) in execution.object.data.walks.iter().enumerate() {
        if walk.timeout_expired_vertex(clock_ms).is_none() {
            continue;
        }
        let plan = inspect_expired_walk_in_execution(
            client,
            ResolveExpiredWalkParams {
                dag_execution_id: execution_id,
                walk_index: index as u64,
                invocation_id: None,
            },
            clock_ms,
            &execution,
        )
        .await?;
        if resolution_priority(&plan.kind) == 0 {
            return Ok(Some(plan));
        }
        if resolution_priority(&plan.kind)
            < selected
                .as_ref()
                .map_or(u8::MAX, |plan| resolution_priority(&plan.kind))
        {
            selected = Some(plan);
        }
    }
    Ok(selected)
}

fn resolution_priority(kind: &ExpiredWalkResolutionKind) -> u8 {
    match kind {
        ExpiredWalkResolutionKind::Settled { .. }
        | ExpiredWalkResolutionKind::SettledOnchainResult { .. } => 0,
        ExpiredWalkResolutionKind::AbortedWithInvocation { .. } => 1,
        ExpiredWalkResolutionKind::Aborted => 2,
        ExpiredWalkResolutionKind::Skipped { .. } => u8::MAX,
    }
}

async fn finalized_onchain_result_resolution_kind(
    client: &NexusClient,
    vertex_info: &graph_move::VertexInfo,
    timeout_vertex: RuntimeVertex,
    invocation_ref: sui::types::ObjectReference,
    result: &OnchainToolResultInnerV1,
    object_ref: sui::types::ObjectReference,
) -> anyhow::Result<ExpiredWalkResolutionKind> {
    let tool_fqn = vertex_info.kind.tool_fqn()?;
    let inspection = client.tool().inspect_tool(&tool_fqn).await?;
    let tool = inspection.tool.ok_or_else(|| {
        anyhow!(
            "Tool '{}' cannot be decoded: {}",
            tool_fqn,
            inspection
                .detail
                .as_deref()
                .unwrap_or("no supported Tool definition")
        )
    })?;
    let ToolRef::Sui {
        tool_witness_id, ..
    } = tool.reference()
    else {
        return Ok(ExpiredWalkResolutionKind::Skipped {
            reason: format!(
                "finalized on-chain result exists for non-Sui tool vertex '{}'",
                timeout_vertex.vertex_name()
            ),
        });
    };
    let finalize_tx_digest = result
        .finalize_tx_digest
        .as_option()
        .cloned()
        .ok_or_else(|| anyhow!("finalized on-chain result is missing finalize_tx_digest"))?;
    Ok(ExpiredWalkResolutionKind::SettledOnchainResult {
        result_ref: object_ref,
        expected_vertex: timeout_vertex,
        invocation_ref,
        tool_witness_id: tool_witness_id.bytes,
        finalize_tx_digest,
    })
}

async fn broken_onchain_result_cleanups_for_abort(
    client: &NexusClient,
    context: &NexusContext,
    execution_id: sui::types::Address,
    execution: &DAGExecutionInnerV1,
    clock_ms: u64,
) -> anyhow::Result<Vec<BrokenOnchainToolResultCleanup>> {
    let crawler = client.crawler();
    let mut dag = None;
    let mut vertices = HashMap::new();
    let mut cleanups = Vec::new();

    for (walk_index, walk) in execution.walks.iter().enumerate() {
        let Some(timeout_vertex) = walk.abortable_timeout_expired_vertex(clock_ms) else {
            continue;
        };
        let walk_index = u64::try_from(walk_index)?;
        match fetch_onchain_tool_result_state_for_walk(client, context, execution_id, walk_index)
            .await?
        {
            OnchainToolResultState::Finalized { result, object_ref } => {
                let vertex_key = timeout_vertex.vertex();
                if !vertices.contains_key(vertex_key) {
                    if dag.is_none() {
                        dag =
                            Some(fetch_dag_in_context(client, context, execution.dag_id()).await?);
                    }
                    let dag = dag.as_ref().expect("DAG was fetched above");
                    let vertex_info = fetch_dag_vertex_bcs(crawler, context, &dag.data, vertex_key)
                        .await?
                        .ok_or_else(|| {
                            anyhow!(
                                "DAG vertex '{}' missing from fetched DAG",
                                timeout_vertex.vertex_name()
                            )
                        })?;
                    vertices.insert(vertex_key.clone(), vertex_info);
                }
                let vertex_info = vertices.get(vertex_key).expect("vertex was fetched above");
                let kind = finalized_onchain_result_resolution_kind(
                    client,
                    vertex_info,
                    timeout_vertex.clone(),
                    fetch_invocation_for_vertex(
                        client,
                        context,
                        execution_id,
                        execution.dag_id(),
                        timeout_vertex,
                    )
                    .await?,
                    &result,
                    object_ref.clone(),
                )
                .await?;
                let ExpiredWalkResolutionKind::SettledOnchainResult {
                    expected_vertex,
                    tool_witness_id,
                    ..
                } = kind
                else {
                    return Err(anyhow!(
                        "expired abort is blocked by finalized on-chain result for walk {walk_index} that cannot be cleaned by the Sui-tool cleanup path"
                    ));
                };
                if onchain_tool_result_has_required_stamps(
                    &result,
                    *object_ref.object_id(),
                    execution_id,
                    context.leader_registry.object_id(),
                    tool_witness_id,
                ) {
                    return Err(anyhow!(
                        "expired abort is blocked by consumable on-chain result for walk {walk_index}; settle the on-chain result before aborting"
                    ));
                }
                cleanups.push(BrokenOnchainToolResultCleanup {
                    walk_index,
                    result_ref: object_ref,
                    expected_vertex,
                    tool_witness_id,
                });
            }
            OnchainToolResultState::InvalidEmpty { result_id, .. } => {
                return Err(anyhow!(
                    "expired abort is blocked by unfinalized on-chain result {result_id} for walk {walk_index}"
                ));
            }
            OnchainToolResultState::NoResult
            | OnchainToolResultState::Committed { .. }
            | OnchainToolResultState::InsufficientSettlement { .. } => {}
        }
    }

    Ok(cleanups)
}

fn onchain_tool_result_has_required_stamps(
    result: &OnchainToolResultInnerV1,
    result_id: sui::types::Address,
    execution_id: sui::types::Address,
    leader_registry_id: sui::types::Address,
    tool_witness_id: sui::types::Address,
) -> bool {
    let Some(stamps) = result.stamps.as_option() else {
        return false;
    };
    let execution_id = ID::new(execution_id);
    let leader_registry_id = ID::new(leader_registry_id);
    let tool_witness_id = ID::new(tool_witness_id);
    let result_id = ID::new(result_id);

    stamps.contents.len() == 4
        && stamps
            .contents
            .iter()
            .any(|entry| entry.key == execution_id)
        && stamps
            .contents
            .iter()
            .any(|entry| entry.key == leader_registry_id)
        && stamps
            .contents
            .iter()
            .any(|entry| entry.key == tool_witness_id)
        && stamps.contents.iter().any(|entry| entry.key == result_id)
}

/// Decodes the finalized onchain result created for one execution walk.
///
/// The walk pointer, result object, and stored output must all have been
/// written by `transaction`. This reads only the finalized receipt, so a
/// consumer does not depend on current object reads observing the write yet.
///
/// # Errors
///
/// Returns an error if the receipt is incomplete, ambiguous, or does not
/// contain a finalized result bound to the requested execution and walk.
pub fn onchain_tool_result_from_finalized_output(
    crawler: &Crawler,
    context: &NexusContext,
    executed: &sui::grpc::ExecutedTransaction,
    transaction: sui::types::Digest,
    execution_id: sui::types::Address,
    walk_index: u64,
) -> anyhow::Result<OnchainToolResultState> {
    use crate::move_bindings::primitives::object_state::Inner;

    let pointers = crawler
        .transaction_dynamic_field_outputs::<execution_move::OnchainToolResultKey, ID>(
            executed,
            transaction,
            &execution_move::OnchainToolResultKey { walk_index },
            &crate::move_bindings::type_tag::<execution_move::OnchainToolResultKey>(context),
            &crate::move_bindings::type_tag::<ID>(context),
        )?;
    let mut pointers = pointers
        .into_iter()
        .filter(|output| output.object_id == execution_id);
    let pointer = pointers.next().ok_or_else(|| anyhow!(
        "Transaction '{transaction}' has no onchain result for execution '{execution_id}' walk {walk_index}"
    ))?;
    anyhow::ensure!(
        pointers.next().is_none(),
        "Transaction '{transaction}' has repeated onchain result pointers"
    );
    let result_id = pointer.data.bytes;
    let anchor =
        crawler.transaction_output_object::<OnchainToolResult>(executed, transaction, result_id)?;
    let initial_version = crawler.transaction_shared_initial_version(
        executed,
        transaction,
        result_id,
        &crate::move_bindings::struct_tag::<OnchainToolResult>(context),
    )?;
    anyhow::ensure!(
        anchor.data.id.address() == result_id,
        "Onchain result anchor identity does not match its walk pointer"
    );

    let outputs = crawler.transaction_dynamic_field_outputs::<Inner, OnchainToolResultInnerV1>(
        executed,
        transaction,
        &Inner::new(false),
        &crate::move_bindings::type_tag::<Inner>(context),
        &crate::move_bindings::type_tag::<OnchainToolResultInnerV1>(context),
    )?;
    let mut outputs = outputs
        .into_iter()
        .filter(|output| output.object_id == result_id);
    let result = outputs
        .next()
        .ok_or_else(|| {
            anyhow!(
                "Transaction '{transaction}' has no stored output for onchain result '{result_id}'"
            )
        })?
        .data;
    anyhow::ensure!(
        outputs.next().is_none(),
        "Transaction '{transaction}' has repeated onchain result outputs"
    );
    anyhow::ensure!(
        result.execution_id.bytes == execution_id,
        "Onchain result belongs to another execution"
    );
    anyhow::ensure!(
        onchain_tool_result_is_finalized(&result),
        "Onchain result is not finalized"
    );
    anyhow::ensure!(
        result.finalize_tx_digest.as_option().map(Vec::as_slice) == Some(transaction.as_ref()),
        "Onchain result was finalized by another transaction"
    );

    Ok(OnchainToolResultState::Finalized {
        result,
        object_ref: sui::types::ObjectReference::new(result_id, initial_version, anchor.digest),
    })
}

/// Fetch the stored onchain result state for one execution walk.
///
/// Each lookup derives an exact dynamic field identifier from its typed key
/// and the package graph in [`NexusContext`]. The amount of work does
/// not depend on the number of fields owned by the execution.
///
/// # Errors
///
/// Returns an error when a key cannot be encoded, a field cannot be fetched,
/// validated, or decoded, or a referenced result object cannot be read.
pub async fn fetch_onchain_tool_result_state_for_walk(
    client: &NexusClient,
    context: &NexusContext,
    execution_id: sui::types::Address,
    walk_index: u64,
) -> anyhow::Result<OnchainToolResultState> {
    let crawler = client.crawler();
    let committed_result =
        fetch_committed_tool_result_for_walk(crawler, context, execution_id, walk_index).await?;
    let insufficient_settlement = crawler
        .get_dynamic_field_by_key::<
            execution_move::ExecutionPaymentInsufficientSettlementFieldKey,
            execution_move::ExecutionPaymentInsufficientSettlement,
        >(
            execution_id,
            insufficient_settlement_field_key(),
            &crate::move_bindings::type_tag::<
                execution_move::ExecutionPaymentInsufficientSettlementFieldKey,
            >(context),
        )
        .await?;

    if let Some(marker) = insufficient_settlement {
        if marker.walks.contains(&walk_index) {
            return Ok(OnchainToolResultState::InsufficientSettlement {
                committed_result,
                marker: InsufficientSettlementMarkerView {
                    walks: marker.walks,
                },
            });
        }
    }

    if let Some(committed_result) = committed_result {
        return Ok(OnchainToolResultState::Committed { committed_result });
    }

    let result_id = crawler
        .get_dynamic_field_by_key::<execution_move::OnchainToolResultKey, ID>(
            execution_id,
            execution_move::OnchainToolResultKey { walk_index },
            &crate::move_bindings::type_tag::<execution_move::OnchainToolResultKey>(context),
        )
        .await?;
    let Some(result_id) = result_id else {
        return Ok(OnchainToolResultState::NoResult);
    };

    let result_id = result_id.bytes;
    let result_response = client
        .state_resolver()
        .load_inner::<OnchainToolResult, InterfaceWitnessV1, OnchainToolResultInnerV1>(
            result_id, context,
        )
        .await?;
    let object_ref = result_response.object_ref();
    let result = result_response.data;

    if onchain_tool_result_is_finalized(&result) {
        Ok(OnchainToolResultState::Finalized { result, object_ref })
    } else {
        Ok(OnchainToolResultState::InvalidEmpty {
            result_id,
            object_ref,
        })
    }
}

fn insufficient_settlement_field_key(
) -> execution_move::ExecutionPaymentInsufficientSettlementFieldKey {
    execution_move::ExecutionPaymentInsufficientSettlementFieldKey { dummy_field: false }
}

pub async fn fetch_dag_default_values_bcs<T>(
    crawler: &Crawler,
    dag: &dag_move::DAGInnerV1,
) -> anyhow::Result<HashMap<graph_move::VertexInputPort, T>>
where
    T: serde::de::DeserializeOwned,
{
    crawler
        .get_dynamic_fields::<graph_move::VertexInputPort, T>(
            dag.defaults_to_input_ports.id(),
            dag.defaults_to_input_ports.size(),
        )
        .await
}

/// Fetch the requested DAG default values by their typed input port keys.
///
/// Duplicate keys are fetched once. Missing defaults are omitted. The amount
/// of work depends on the requested ports rather than every default stored in
/// the DAG. Use [`fetch_dag_default_values_bcs`] only when the complete table
/// is required.
///
/// # Errors
///
/// Returns an error when a key cannot be encoded or a field cannot be fetched,
/// validated, or decoded.
pub async fn fetch_dag_default_values_by_keys_bcs<T, I>(
    crawler: &Crawler,
    context: &NexusContext,
    dag: &dag_move::DAGInnerV1,
    ports: I,
) -> anyhow::Result<HashMap<graph_move::VertexInputPort, T>>
where
    T: serde::de::DeserializeOwned,
    I: IntoIterator<Item = graph_move::VertexInputPort>,
{
    crawler
        .get_dynamic_fields_by_keys::<graph_move::VertexInputPort, T, _>(
            dag.defaults_to_input_ports.id(),
            ports,
            &crate::move_bindings::type_tag::<graph_move::VertexInputPort>(context),
        )
        .await
}

pub async fn fetch_dag_edges_bcs(
    crawler: &Crawler,
    dag: &dag_move::DAGInnerV1,
) -> anyhow::Result<HashMap<graph_move::Vertex, Vec<graph_move::Edge>>> {
    crawler
        .get_dynamic_fields::<graph_move::Vertex, Vec<graph_move::Edge>>(
            dag.edges.id(),
            dag.edges.size(),
        )
        .await
}

pub async fn fetch_dag_outputs_bcs(
    crawler: &Crawler,
    dag: &dag_move::DAGInnerV1,
) -> anyhow::Result<HashMap<graph_move::Vertex, Vec<graph_move::OutputVariantPort>>> {
    crawler
        .get_dynamic_fields::<graph_move::Vertex, Vec<graph_move::OutputVariantPort>>(
            dag.outputs.id(),
            dag.outputs.size(),
        )
        .await
}

/// Determine whether one offchain success requires tool verification.
///
/// The DAG pair is validated first, then the requested vertex is fetched by
/// its exact key using [`NexusContext`].
///
/// # Errors
///
/// Returns an error when the DAG or vertex cannot be read, when the vertex is
/// absent, or when a returned value is invalid.
pub async fn offchain_success_requires_tool_verification(
    client: &NexusClient,
    context: &NexusContext,
    dag_object_id: sui::types::Address,
    next_vertex: &RuntimeVertex,
) -> anyhow::Result<bool> {
    let dag = fetch_dag_in_context(client, context, dag_object_id).await?;
    let vertex_name = next_vertex.vertex();
    let vertex = fetch_dag_vertex_bcs(client.crawler(), context, &dag.data, vertex_name)
        .await?
        .ok_or_else(|| {
            anyhow!(
                "Vertex '{}' not found in DAG verifier config",
                vertex_name.name
            )
        })?;

    Ok(dag_vertex_requires_tool_verification(&vertex))
}

/// Fetch the declared input port names for one DAG vertex.
///
/// The vertex is read by its exact key using the deployment addresses in
/// [`NexusContext`].
///
/// # Errors
///
/// Returns an error when the vertex cannot be read, is absent, or contains an
/// invalid value.
pub async fn fetch_vertex_input_port_names(
    crawler: &Crawler,
    context: &NexusContext,
    dag: &dag_move::DAGInnerV1,
    vertex_name: &TypeName,
) -> anyhow::Result<Vec<String>> {
    let vertex_key = graph_move::Vertex::from(vertex_name);
    let vertex = fetch_dag_vertex_bcs(crawler, context, dag, &vertex_key)
        .await?
        .ok_or_else(|| {
            anyhow!("Vertex '{vertex_name}' not found in DAG vertices dynamic fields")
        })?;

    Ok(vertex.declared_input_port_names())
}

pub fn execution_terminal_record_matches_retryable_vertex(
    terminal_records: &VecMap<u64, execution_move::TerminalErrEvalRecord>,
    walk_index: u64,
    next_vertex: &RuntimeVertex,
) -> bool {
    terminal_records.get(&walk_index).is_some_and(|record| {
        &record.vertex == next_vertex && record.failure_class == WorkflowFailureClass::Retryable
    })
}

pub async fn should_settle_tool_err_eval_gas(
    client: &NexusClient,
    execution: sui::types::Address,
    walk_index: u64,
    next_vertex: &RuntimeVertex,
) -> anyhow::Result<bool> {
    let execution = fetch_execution(client, execution, &[]).await?;
    Ok(execution_terminal_record_matches_retryable_vertex(
        &execution.object.data.terminal_records,
        walk_index,
        next_vertex,
    ))
}

async fn resolve_invocation_policy_call(
    client: &NexusClient,
    context: &NexusContext,
    cashier_id: sui::types::Address,
    beneficiary: crate::move_bindings::interface::payment::PaymentSourceKind,
    policy: InvocationPolicy,
) -> Result<InvocationPolicyCall, NexusError> {
    match policy {
        InvocationPolicy::FixedPrice => {
            InvocationPolicyCall::fixed_price(context).map_err(NexusError::Parsing)
        }
        InvocationPolicy::FiniteCredits => {
            let credits_id = crate::move_bindings::derive_finite_credits_id(
                context,
                cashier_id,
                beneficiary.clone(),
            )
            .map_err(NexusError::Parsing)?;
            let credits = client
                .crawler()
                .get_object::<FiniteCredits>(credits_id)
                .await
                .map_err(|error| {
                    NexusError::Configuration(format!(
                        "Finite credits '{credits_id}' could not be resolved: {error}"
                    ))
                })?;
            if credits.data.cashier.bytes != cashier_id || credits.data.beneficiary != beneficiary {
                return Err(NexusError::Configuration(format!(
                    "Finite credits '{credits_id}' belong to another Tool"
                )));
            }
            let sui::types::Owner::Shared(initial_shared_version) = credits.owner else {
                return Err(NexusError::Configuration(format!(
                    "Finite credits '{credits_id}' must be a shared object"
                )));
            };
            InvocationPolicyCall::finite_credits(context, credits_id, initial_shared_version)
                .map_err(NexusError::Parsing)
        }
        InvocationPolicy::TimePass => {
            let pass_id =
                crate::move_bindings::derive_time_pass_id(context, cashier_id, beneficiary.clone())
                    .map_err(NexusError::Parsing)?;
            let pass = client
                .crawler()
                .get_object::<TimePass>(pass_id)
                .await
                .map_err(|error| {
                    NexusError::Configuration(format!(
                        "Time pass '{pass_id}' could not be resolved: {error}"
                    ))
                })?;
            if pass.data.cashier.bytes != cashier_id || pass.data.beneficiary != beneficiary {
                return Err(NexusError::Configuration(format!(
                    "Time pass '{pass_id}' belongs to another Tool"
                )));
            }
            let sui::types::Owner::Shared(initial_shared_version) = pass.owner else {
                return Err(NexusError::Configuration(format!(
                    "Time pass '{pass_id}' must be a shared object"
                )));
            };
            InvocationPolicyCall::time_pass(context, pass_id, initial_shared_version)
                .map_err(NexusError::Parsing)
        }
        InvocationPolicy::Custom(call) => Ok(call),
    }
}

impl WorkflowActions {
    /// Publish the provided [`DagSpec`] specification.
    pub async fn publish(
        &self,
        workflow_package: sui::types::Address,
        dag_spec: DagSpec,
    ) -> Result<PublishResult, NexusError> {
        let context = self
            .client
            .context_for_creator(workflow_package, PackageRole::Workflow, &[])
            .await?;
        let address = self.client.owner()?;

        // == Craft and submit the publish DAG transaction ==

        let tx = dag::publish_ptb(&context, dag_spec).map_err(NexusError::TransactionBuilding)?;
        let response = self.client.submit_transaction(tx, address).await?;

        // == Find the published DAG object ID ==

        let dag_tag = crate::move_bindings::struct_tag::<dag_move::DAG>(&context);
        let dag_object_id = response
            .objects
            .into_iter()
            .find_map(|obj| {
                let sui::types::ObjectType::Struct(object_type) = obj.object_type() else {
                    return None;
                };

                if *object_type.address() == *dag_tag.address()
                    && object_type.module() == dag_tag.module()
                    && object_type.name() == dag_tag.name()
                {
                    Some(obj.object_id())
                } else {
                    None
                }
            })
            .ok_or_else(|| {
                NexusError::Parsing(anyhow!("DAG object ID not found in TX response"))
            })?;

        Ok(PublishResult {
            tx_digest: response.digest,
            tx_checkpoint: response.checkpoint,
            dag_object_id,
        })
    }

    /// Reads the published DAG interface used to select an entry group and
    /// provide its inputs.
    pub async fn inspect_dag(
        &self,
        dag_id: sui::types::Address,
    ) -> Result<DagSnapshot, NexusError> {
        let context = self.client.context_for_object(dag_id).await?;
        fetch_dag_snapshot(&self.client, &context, dag_id)
            .await
            .map_err(NexusError::Rpc)
    }

    /// Authorizes one active [RuntimeVertex] with the current Leader capability.
    ///
    /// The signer must control the supplied capability. This manual path records
    /// no reimbursement. The automatic Leader service supplies the verified cost.
    pub async fn authorize_invocation(
        &self,
        dag_execution_id: sui::types::Address,
        leader_cap_id: sui::types::Address,
        walk_index: u64,
        vertex: RuntimeVertex,
        policy: InvocationPolicy,
    ) -> Result<AuthorizeInvocationResult, NexusError> {
        let client = &self.client;
        let objects = client.get_nexus_objects();
        let required_roots = [objects.tool_registry, objects.leader_registry];
        let crawler = client.crawler();
        let leader_cap = crawler
            .get_object_metadata(leader_cap_id)
            .await
            .map_err(NexusError::Rpc)?;
        let execution = fetch_execution(client, dag_execution_id, &[]).await?;
        let is_active = execution
            .object
            .data
            .walks
            .get(walk_index as usize)
            .is_some_and(|walk| {
                matches!(
                    walk,
                    DAGWalk::Active { next_vertex, .. } if next_vertex == &vertex
                )
            });
        if !is_active {
            return Err(NexusError::Configuration(format!(
                "Runtime vertex '{vertex}' is not active in execution '{dag_execution_id}'"
            )));
        }

        let dag_id = execution.object.data.dag_id();
        let dag = fetch_dag_in_context(client, &execution.context, dag_id).await?;
        let vertex_info =
            fetch_dag_vertex_bcs(crawler, &execution.context, &dag.data, vertex.vertex())
                .await
                .map_err(NexusError::Rpc)?
                .ok_or_else(|| {
                    NexusError::Configuration(format!(
                        "Vertex '{}' is absent from DAG '{}'",
                        vertex.vertex_name(),
                        dag_id
                    ))
                })?;
        let tool_fqn = vertex_info.kind.tool_fqn().map_err(NexusError::Parsing)?;
        let cashier = client.fetch_tool_cashier(&tool_fqn).await?;
        let cashier_id = *cashier.object_id();
        let payment = tap::fetch_execution_payment_for_execution(
            client,
            &execution.context,
            dag_execution_id,
        )
        .await
        .map_err(NexusError::Rpc)?;
        let runtime = client.runtime_context(&required_roots).await?;
        let policy = resolve_invocation_policy_call(
            client,
            &runtime,
            cashier_id,
            payment.data.source_kind,
            policy,
        )
        .await?;
        let transaction = invocation::authorize_ptb(
            &runtime,
            &cashier,
            &dag.object_ref(),
            &execution.object.object_ref(),
            &leader_cap.object_ref(),
            invocation::InvocationTarget {
                walk_index,
                vertex: &vertex,
            },
            &policy,
            0,
        )
        .map_err(NexusError::TransactionBuilding)?;
        let sender = client.owner()?;
        let response = client.submit_transaction(transaction, sender).await?;
        let lock = response
            .events
            .iter()
            .find_map(|event| match &event.data {
                NexusEventKind::InvocationLocked(lock)
                    if lock.execution.bytes == dag_execution_id && lock.vertex == vertex =>
                {
                    Some(lock.clone())
                }
                _ => None,
            })
            .ok_or_else(|| {
                NexusError::Parsing(anyhow!(
                    "Authorization transaction '{}' emitted no exact Invocation lock",
                    response.digest
                ))
            })?;

        Ok(AuthorizeInvocationResult {
            tx_digest: response.digest,
            tx_checkpoint: response.checkpoint,
            lock,
        })
    }

    /// Inspect a DAG execution by following updates to its shared object.
    ///
    /// The inspector reconstructs every transaction in the execution
    /// object's update chain, emits matching events chronologically, and then
    /// polls for new object versions. It does not subscribe to checkpoints.
    pub async fn inspect_execution(
        &self,
        dag_execution_id: sui::types::Address,
        options: InspectExecutionOptions,
    ) -> Result<InspectExecutionResult, NexusError> {
        if options.poll_interval.is_zero() {
            return Err(NexusError::Configuration(
                "Execution inspection poll interval must be greater than zero".to_string(),
            ));
        }

        let deadline = Instant::now() + options.timeout;
        let (tx, rx) = unbounded_channel::<NexusEvent>();
        let client = self.client.clone();
        let crawler = self.client.crawler().clone();
        let event_decoder = NexusEventDecoder::new(
            self.client.state_resolver().clone(),
            self.client.get_nexus_objects(),
        );

        let poller = tokio::spawn(async move {
            let inspection = async move {
                let context = loop {
                    match client.context_for_object(dag_execution_id).await {
                        Ok(context) => break context,
                        Err(error) if is_transient_inspection_error(&error) => {
                            tokio::time::sleep(options.poll_interval).await;
                        }
                        Err(error) => return Err(error),
                    }
                };
                let mut last_delivered_version = None;

                loop {
                    let latest = match crawler
                        .get_object_update_reference(dag_execution_id, None)
                        .await
                        .map_err(|error| {
                            NexusError::Rpc(error.context(format!(
                                "Could not inspect execution '{dag_execution_id}' latest object; last successfully reconstructed version {}",
                                version_or_none(last_delivered_version)
                            )))
                        }) {
                        Ok(latest) => latest,
                        Err(error) if is_transient_inspection_error(&error) => {
                            tokio::time::sleep(options.poll_interval).await;
                            continue;
                        }
                        Err(error) => return Err(error),
                    };

                    if last_delivered_version == Some(latest.version) {
                        tokio::time::sleep(options.poll_interval).await;
                        continue;
                    }

                    let latest_version = latest.version;
                    let events = loop {
                        match fetch_execution_update_events(
                            &crawler,
                            &event_decoder,
                            ObjectHistoryRequest {
                                object_name: "Execution",
                                object_id: dag_execution_id,
                                expected_type: crate::move_bindings::struct_tag::<DAGExecution>(
                                    &context,
                                ),
                                latest: latest.clone(),
                                after_version: last_delivered_version,
                                poll_interval: options.poll_interval,
                                deadline,
                            },
                        )
                        .await
                        {
                            Ok(events) => break events,
                            Err(error) if is_transient_inspection_error(&error) => {
                                tokio::time::sleep(options.poll_interval).await;
                            }
                            Err(error) => return Err(error),
                        }
                    };
                    let execution_finished_seen = events
                        .iter()
                        .any(|event| matches!(event.data, NexusEventKind::ExecutionFinished(_)));

                    for event in events {
                        tx.send(event)
                            .map_err(|error| NexusError::Channel(error.into()))?;
                    }
                    last_delivered_version = Some(latest_version);

                    if execution_finished_seen {
                        return Ok(());
                    }

                    tokio::time::sleep(options.poll_interval).await;
                }
            };

            tokio::time::timeout_at(deadline, inspection)
                .await
                .map_err(|_| {
                    NexusError::Timeout(anyhow!(
                        "Timeout {:?} reached while inspecting DAG execution '{dag_execution_id}'",
                        options.timeout
                    ))
                })?
        });

        Ok(InspectExecutionResult {
            next_event: rx,
            poller,
        })
    }

    /// Inspect a DAG execution until completion and return a structured summary
    /// with resolved end-state data.
    #[cfg(feature = "walrus")]
    pub async fn inspect_execution_until_completion(
        &self,
        dag_execution_id: sui::types::Address,
        options: InspectExecutionOptions,
        storage_conf: &StorageConf,
    ) -> Result<InspectExecutionCompletionResult, NexusError> {
        let mut inspection = self.inspect_execution(dag_execution_id, options).await?;

        let mut events = Vec::new();

        while let Some(event) = inspection.next_event.recv().await {
            events.push(event);
        }

        let poller_result = inspection.poller.await.map_err(|e| {
            NexusError::Channel(anyhow!(
                "Execution inspection task failed for DAG execution '{dag_execution_id}': {e}"
            ))
        })?;
        poller_result?;

        build_execution_completion_result(events, dag_execution_id, storage_conf).await
    }

    /// Fetch the TAP execution payment cost summary for a DAG
    /// execution.
    pub async fn execution_cost(
        &self,
        dag_execution_id: sui::types::Address,
    ) -> Result<ExecutionCostResult, NexusError> {
        Ok(self.execution_cost_details(dag_execution_id).await?.summary)
    }

    /// Fetch payment accounting and outstanding invocation identities from one payment snapshot.
    pub async fn execution_cost_details(
        &self,
        dag_execution_id: sui::types::Address,
    ) -> Result<ExecutionCostDetails, NexusError> {
        let execution = fetch_execution(&self.client, dag_execution_id, &[]).await?;
        if execution
            .object
            .data
            .to_context()
            .map_err(NexusError::Parsing)?
            .is_none()
        {
            return Err(NexusError::Parsing(anyhow!(
                "DAG execution '{dag_execution_id}' has no TAP payment context"
            )));
        }
        let payment = tap::fetch_execution_payment_for_execution(
            &self.client,
            &execution.context,
            dag_execution_id,
        )
        .await
        .map_err(NexusError::Rpc)?;

        Ok(ExecutionCostDetails::from_payment(
            payment.object_id,
            payment.data,
        ))
    }

    /// Submit the current permissionless expired-execution abort entry.
    ///
    /// This wraps `execution_settlement::abort_expired_execution`; it does not
    /// discover or submit exact Invocation candidates.
    ///
    /// If a double-expired active walk is blocked by a finalized on-chain
    /// result whose required stamps are insufficient, this returns a local
    /// transaction-building error because the current workflow package no
    /// longer exposes a broken-result cleanup entrypoint.
    pub async fn abort_expired_execution(
        &self,
        dag_execution_id: sui::types::Address,
    ) -> Result<AbortExecutionResult, NexusError> {
        self.abort_expired_execution_with(&self.client, dag_execution_id)
            .await
    }

    async fn abort_expired_execution_with(
        &self,
        client: &NexusClient,
        dag_execution_id: sui::types::Address,
    ) -> Result<AbortExecutionResult, NexusError> {
        let crawler = client.crawler();
        let clock = crawler
            .get_object::<SuiClock>(move_boundary::CLOCK_OBJECT_ID)
            .await
            .map_err(NexusError::Rpc)?
            .data;
        let mut execution = fetch_execution(client, dag_execution_id, &[]).await?;
        let mut cleaned_broken_onchain_results = broken_onchain_result_cleanups_for_abort(
            client,
            &execution.context,
            dag_execution_id,
            &execution.object.data,
            clock.timestamp_ms,
        )
        .await
        .map_err(NexusError::TransactionBuilding)?;

        if !cleaned_broken_onchain_results.is_empty() {
            execution = fetch_execution(client, dag_execution_id, &[]).await?;
            cleaned_broken_onchain_results = broken_onchain_result_cleanups_for_abort(
                client,
                &execution.context,
                dag_execution_id,
                &execution.object.data,
                clock.timestamp_ms,
            )
            .await
            .map_err(NexusError::TransactionBuilding)?;
        }

        let dag_id = execution.object.data.dag_id();
        let dag = fetch_dag_in_context(client, &execution.context, dag_id).await?;
        let execution_ref = execution.object.object_ref();
        let objects = client.get_nexus_objects();
        let required_roots = if cleaned_broken_onchain_results.is_empty() {
            Vec::new()
        } else {
            vec![objects.tool_registry, objects.leader_registry]
        };
        let runtime = client.runtime_context(&required_roots).await?;

        let address = client.owner()?;
        let tx = dag::abort_expired_execution_for_self_ptb(
            &runtime,
            &dag.object_ref(),
            &execution_ref,
            &cleaned_broken_onchain_results
                .iter()
                .map(|cleanup| dag::BrokenOnchainToolResultCleanupInput {
                    walk_index: cleanup.walk_index,
                    result_ref: cleanup.result_ref.clone(),
                    tool_witness_id: cleanup.tool_witness_id,
                })
                .collect::<Vec<_>>(),
            None,
        )
        .map_err(NexusError::TransactionBuilding)?;
        let response = client.submit_transaction(tx, address).await?;

        Ok(AbortExecutionResult {
            tx_digest: response.digest,
            tx_checkpoint: response.checkpoint,
            dag_id,
            dag_execution_id,
            cleaned_broken_onchain_results,
        })
    }

    /// Submit permissionless committed-result settlement for one walk.
    pub async fn settle_committed_tool_result_for_walk(
        &self,
        params: SettleCommittedToolResultParams,
    ) -> Result<CommittedToolResultSettlementResult, NexusError> {
        self.settle_committed_tool_result_for_walk_with(&self.client, params)
            .await
    }

    async fn settle_committed_tool_result_for_walk_with(
        &self,
        client: &NexusClient,
        params: SettleCommittedToolResultParams,
    ) -> Result<CommittedToolResultSettlementResult, NexusError> {
        let objects = client.get_nexus_objects();
        let required_roots = [
            objects.tool_registry,
            objects.leader_registry,
            objects.priority_fee_vault,
        ];
        let execution = fetch_execution(client, params.dag_execution_id, &[]).await?;
        let runtime = client.runtime_context(&required_roots).await?;
        let dag_id = execution.object.data.dag_id();
        let dag = fetch_dag_in_context(client, &execution.context, dag_id).await?;
        let expected_vertex = pending_settlement_vertex(&execution.object.data, params.walk_index)?;
        let invocation_ref = fetch_invocation_for_vertex(
            client,
            &execution.context,
            params.dag_execution_id,
            dag_id,
            &expected_vertex,
        )
        .await?;

        let address = client.owner()?;
        let tx = dag::settle_committed_tool_result_for_walk_for_self_ptb(
            &runtime,
            &dag.object_ref(),
            &execution.object.object_ref(),
            &invocation_ref,
            params.walk_index,
            &expected_vertex,
            None,
        )
        .map_err(NexusError::TransactionBuilding)?;
        let response = client.submit_transaction(tx, address).await?;

        Ok(CommittedToolResultSettlementResult {
            tx_digest: response.digest,
            tx_checkpoint: response.checkpoint,
            dag_id,
            dag_execution_id: params.dag_execution_id,
            walk_index: params.walk_index,
        })
    }

    /// Submit leader authenticated committed result settlement with the leader commit gas charge.
    pub async fn settle_committed_tool_result_for_walk_by_leader(
        &self,
        params: SettleCommittedToolResultByLeaderParams,
    ) -> Result<CommittedToolResultSettlementResult, NexusError> {
        let client = &self.client;
        let crawler = client.crawler();
        let objects = client.get_nexus_objects();
        let required_roots = [
            objects.tool_registry,
            objects.leader_registry,
            objects.priority_fee_vault,
        ];
        let execution = fetch_execution(client, params.dag_execution_id, &[]).await?;
        let runtime = client.runtime_context(&required_roots).await?;
        let dag_id = execution.object.data.dag_id();
        let dag = fetch_dag_in_context(client, &execution.context, dag_id).await?;
        let settlement_vertex =
            pending_settlement_vertex(&execution.object.data, params.walk_index)?;
        let invocation_ref = fetch_invocation_for_vertex(
            client,
            &execution.context,
            params.dag_execution_id,
            dag_id,
            &settlement_vertex,
        )
        .await?;
        let leader_cap_ref = crawler
            .get_object_metadata(params.leader_cap_id)
            .await
            .map_err(NexusError::Rpc)?;
        let (expected_vertex, failed_onchain_tool_reason) =
            match (params.expected_vertex, params.failed_onchain_tool_reason) {
                (Some(expected_vertex), Some(reason)) => {
                    if expected_vertex != settlement_vertex {
                        return Err(NexusError::TransactionBuilding(anyhow::anyhow!(
                            "expected_vertex does not match the pending settlement vertex"
                        )));
                    }
                    (expected_vertex, Some(reason))
                }
                (None, None) => (settlement_vertex, None),
                _ => {
                    return Err(NexusError::TransactionBuilding(anyhow::anyhow!(
                        "expected_vertex and failed_onchain_tool_reason must be supplied together"
                    )));
                }
            };
        let address = client.owner()?;
        let tx = dag::settle_committed_tool_result_for_walk_by_leader_for_self_ptb(
            &runtime,
            &dag.object_ref(),
            &execution.object.object_ref(),
            &execution.object.owner,
            &leader_cap_ref.object_ref(),
            &leader_cap_ref.owner,
            &invocation_ref,
            params.walk_index,
            &expected_vertex,
            failed_onchain_tool_reason,
            params.commit_tx_digest,
            params.commit_gas_charge,
            params.settlement_gas_charge,
        )
        .map_err(NexusError::TransactionBuilding)?;
        let response = client.submit_transaction(tx, address).await?;

        Ok(CommittedToolResultSettlementResult {
            tx_digest: response.digest,
            tx_checkpoint: response.checkpoint,
            dag_id,
            dag_execution_id: params.dag_execution_id,
            walk_index: params.walk_index,
        })
    }

    /// Record a leader commit gas charge without settling the walk.
    pub async fn record_committed_tool_result_gas_charge_by_leader(
        &self,
        params: RecordCommittedToolResultGasChargeParams,
    ) -> Result<RecordCommittedToolResultGasChargeResult, NexusError> {
        let client = &self.client;
        let crawler = client.crawler();
        let objects = client.get_nexus_objects();
        let required_roots = [objects.leader_registry];
        let execution = fetch_execution(client, params.dag_execution_id, &[]).await?;
        let runtime = client.runtime_context(&required_roots).await?;
        let dag_id = execution.object.data.dag_id();
        let dag = fetch_dag_in_context(client, &execution.context, dag_id).await?;
        let leader_cap_ref = crawler
            .get_object_metadata(params.leader_cap_id)
            .await
            .map_err(NexusError::Rpc)?;
        let (expected_vertex, failed_onchain_tool_reason) =
            match (params.expected_vertex, params.failed_onchain_tool_reason) {
                (Some(expected_vertex), Some(reason)) => (expected_vertex, Some(reason)),
                (None, None) => (RuntimeVertex::plain(""), None),
                _ => {
                    return Err(NexusError::TransactionBuilding(anyhow::anyhow!(
                        "expected_vertex and failed_onchain_tool_reason must be supplied together"
                    )));
                }
            };

        let address = client.owner()?;
        let tx = dag::record_committed_tool_result_gas_charge_by_leader_for_self_ptb(
            &runtime,
            &dag.object_ref(),
            &execution.object.object_ref(),
            &execution.object.owner,
            &leader_cap_ref.object_ref(),
            &leader_cap_ref.owner,
            params.walk_index,
            &expected_vertex,
            failed_onchain_tool_reason,
            params.commit_tx_digest,
            params.commit_gas_charge,
            params.settlement_gas_charge,
        )
        .map_err(NexusError::TransactionBuilding)?;
        let response = client.submit_transaction(tx, address).await?;

        Ok(RecordCommittedToolResultGasChargeResult {
            tx_digest: response.digest,
            tx_checkpoint: response.checkpoint,
            dag_execution_id: params.dag_execution_id,
            leader_cap_id: params.leader_cap_id,
            walk_index: params.walk_index,
        })
    }

    /// Classify what the SDK would submit for one expired walk without sending it.
    pub async fn inspect_expired_walk_resolution(
        &self,
        params: ResolveExpiredWalkParams,
    ) -> Result<ExpiredWalkResolutionPlan, NexusError> {
        self.inspect_expired_walk_resolution_with(&self.client, params)
            .await
    }

    async fn inspect_expired_walk_resolution_with(
        &self,
        client: &NexusClient,
        params: ResolveExpiredWalkParams,
    ) -> Result<ExpiredWalkResolutionPlan, NexusError> {
        inspect_expired_walk_resolution(client, params)
            .await
            .map_err(NexusError::Rpc)
    }

    /// Resolve currently eligible expired work using confirmed chain time.
    ///
    /// Every transaction is confirmed before selecting the next walk. This only
    /// sequences recovery mutations of this shared execution object; independent
    /// executions and ordinary tool calls remain concurrent. A repeated unchanged
    /// plan ends the pass so stale reads cannot cause an unbounded submission loop.
    pub async fn resolve_expired_execution(
        &self,
        execution_id: sui::types::Address,
    ) -> Result<Vec<ExpiredWalkResolutionResult>, NexusError> {
        let mut results = Vec::new();
        let mut previous = None;
        loop {
            let clock = self
                .client
                .crawler()
                .get_object::<SuiClock>(move_boundary::CLOCK_OBJECT_ID)
                .await
                .map_err(NexusError::Rpc)?;
            let Some(plan) = inspect_expired_execution_resolution_at(
                &self.client,
                execution_id,
                clock.data.timestamp_ms,
            )
            .await
            .map_err(NexusError::Rpc)?
            else {
                break;
            };
            if previous.as_ref() == Some(&plan) {
                break;
            }
            let result = self
                .resolve_expired_walk(ResolveExpiredWalkParams {
                    dag_execution_id: execution_id,
                    walk_index: plan.walk_index,
                    invocation_id: None,
                })
                .await?;
            previous = Some(plan);
            let progressed = result.tx_digest.is_some();
            results.push(result);
            if !progressed {
                break;
            }
        }
        Ok(results)
    }

    /// Classify and submit the existing Move entry that matches one expired walk.
    pub async fn resolve_expired_walk(
        &self,
        params: ResolveExpiredWalkParams,
    ) -> Result<ExpiredWalkResolutionResult, NexusError> {
        let client = &self.client;
        let plan = self
            .inspect_expired_walk_resolution_with(client, params)
            .await?;
        let base = |resolution_kind| ExpiredWalkResolutionResult {
            tx_digest: None,
            tx_checkpoint: None,
            dag_id: plan.dag_id,
            dag_execution_id: plan.dag_execution_id,
            walk_index: plan.walk_index,
            resolution_kind,
        };

        match plan.kind {
            ExpiredWalkResolutionKind::Settled {
                expected_vertex,
                invocation_ref,
            } => {
                let settled = self
                    .settle_committed_tool_result_for_walk_with(
                        client,
                        SettleCommittedToolResultParams {
                            dag_execution_id: plan.dag_execution_id,
                            walk_index: plan.walk_index,
                        },
                    )
                    .await?;
                Ok(ExpiredWalkResolutionResult {
                    tx_digest: Some(settled.tx_digest),
                    tx_checkpoint: Some(settled.tx_checkpoint),
                    ..base(ExpiredWalkResolutionKind::Settled {
                        expected_vertex,
                        invocation_ref,
                    })
                })
            }
            ExpiredWalkResolutionKind::SettledOnchainResult {
                result_ref,
                expected_vertex,
                invocation_ref,
                tool_witness_id,
                finalize_tx_digest,
            } => {
                let objects = client.get_nexus_objects();
                let required_roots = [
                    objects.tool_registry,
                    objects.leader_registry,
                    objects.priority_fee_vault,
                ];
                let execution = fetch_execution(client, plan.dag_execution_id, &[]).await?;
                let runtime = client.runtime_context(&required_roots).await?;
                if execution.object.data.dag_id() != plan.dag_id {
                    return Err(NexusError::Parsing(anyhow!(
                        "DAG execution '{}' now references DAG '{}', expected '{}'",
                        plan.dag_execution_id,
                        execution.object.data.dag_id(),
                        plan.dag_id
                    )));
                }
                let dag = fetch_dag_in_context(client, &execution.context, plan.dag_id).await?;
                let result = client
                    .state_resolver()
                    .load_inner::<OnchainToolResult, InterfaceWitnessV1, OnchainToolResultInnerV1>(
                        *result_ref.object_id(),
                        &execution.context,
                    )
                    .await?;
                let address = client.owner()?;
                let context = &runtime;
                let tx = move_boundary::ptb(context, |tx| {
                    let dag_arg = tx.immutable_object(&dag.object_ref())?;
                    let execution_arg = tx.shared_object(&execution.object.object_ref(), true)?;
                    let tool_registry = tx.shared_root(&context.tool_registry, false)?;
                    let result_arg = tx.shared_object(&result.object_ref(), true)?;
                    let leader_registry = tx.shared_root(&context.leader_registry, false)?;
                    let priority_fee_vault = tx.shared_root(&context.priority_fee_vault, false)?;
                    let clock = tx.clock()?;
                    dag::settle_onchain_tool_result_for_walk(
                        tx,
                        dag_arg,
                        execution_arg,
                        tool_registry,
                        result_arg,
                        leader_registry,
                        priority_fee_vault,
                        plan.walk_index,
                        &expected_vertex,
                        tool_witness_id,
                        clock,
                    )?;
                    invocation::settle(
                        tx,
                        dag_arg,
                        execution_arg,
                        &expected_vertex,
                        &invocation_ref,
                    )?;
                    Ok(())
                })
                .map_err(NexusError::TransactionBuilding)?;
                let response = client.submit_transaction(tx, address).await?;
                Ok(ExpiredWalkResolutionResult {
                    tx_digest: Some(response.digest),
                    tx_checkpoint: Some(response.checkpoint),
                    ..base(ExpiredWalkResolutionKind::SettledOnchainResult {
                        result_ref: result.object_ref(),
                        expected_vertex,
                        invocation_ref,
                        tool_witness_id,
                        finalize_tx_digest,
                    })
                })
            }
            ExpiredWalkResolutionKind::Aborted => {
                let aborted = self
                    .abort_expired_execution_with(client, plan.dag_execution_id)
                    .await?;
                Ok(ExpiredWalkResolutionResult {
                    tx_digest: Some(aborted.tx_digest),
                    tx_checkpoint: Some(aborted.tx_checkpoint),
                    ..base(ExpiredWalkResolutionKind::Aborted)
                })
            }
            ExpiredWalkResolutionKind::AbortedWithInvocation { selected_candidate } => {
                let invocation_id = *selected_candidate.invocation_ref.object_id();
                let aborted = self
                    .abort_expired_execution_with_invocation_with(
                        client,
                        plan.dag_execution_id,
                        Some(invocation_id),
                    )
                    .await?;
                Ok(ExpiredWalkResolutionResult {
                    tx_digest: Some(aborted.tx_digest),
                    tx_checkpoint: Some(aborted.tx_checkpoint),
                    ..base(ExpiredWalkResolutionKind::AbortedWithInvocation { selected_candidate })
                })
            }
            ExpiredWalkResolutionKind::Skipped { reason } => {
                Ok(base(ExpiredWalkResolutionKind::Skipped { reason }))
            }
        }
    }

    /// Returns exact Invocations eligible for permissionless timeout refund.
    pub async fn abort_expired_execution_invocation_candidates(
        &self,
        dag_execution_id: sui::types::Address,
    ) -> Result<Vec<InvocationAbortCandidate>, NexusError> {
        self.abort_expired_execution_invocation_candidates_with(&self.client, dag_execution_id)
            .await
    }

    async fn abort_expired_execution_invocation_candidates_with(
        &self,
        client: &NexusClient,
        dag_execution_id: sui::types::Address,
    ) -> Result<Vec<InvocationAbortCandidate>, NexusError> {
        let execution = fetch_execution(client, dag_execution_id, &[]).await?;
        let context = &execution.context;
        let execution_state = &execution.object.data;
        let crawler = client.crawler();
        let dag = fetch_dag_in_context(client, context, execution_state.dag_id()).await?;
        let clock = crawler
            .get_object::<SuiClock>(move_boundary::CLOCK_OBJECT_ID)
            .await
            .map_err(NexusError::Rpc)?
            .data;
        let requested_vertices = execution_state
            .walks
            .iter()
            .filter_map(|walk| walk.abortable_timeout_expired_vertex(clock.timestamp_ms))
            .map(|vertex| vertex.vertex().clone());
        let vertices =
            fetch_dag_vertices_by_keys_bcs(crawler, context, &dag.data, requested_vertices)
                .await
                .map_err(NexusError::Rpc)?;

        let payment = tap::fetch_execution_payment_for_execution(client, context, dag_execution_id)
            .await
            .map_err(NexusError::Rpc)?
            .data;
        let candidates = fetch_invocation_abort_candidates(
            crawler,
            filter_invocation_abort_candidate_walks(
                dag_execution_id,
                &vertices,
                &execution_state.walks,
                &payment.locked_vertices,
                clock.timestamp_ms,
            )
            .map_err(NexusError::Parsing)?,
        )
        .await?;

        Ok(candidates)
    }

    /// Refunds one exact Invocation and aborts its expired runtime vertex.
    pub async fn abort_expired_execution_with_invocation(
        &self,
        dag_execution_id: sui::types::Address,
        invocation_id: Option<sui::types::Address>,
    ) -> Result<AbortExpiredExecutionResult, NexusError> {
        self.abort_expired_execution_with_invocation_with(
            &self.client,
            dag_execution_id,
            invocation_id,
        )
        .await
    }

    async fn abort_expired_execution_with_invocation_with(
        &self,
        client: &NexusClient,
        dag_execution_id: sui::types::Address,
        invocation_id: Option<sui::types::Address>,
    ) -> Result<AbortExpiredExecutionResult, NexusError> {
        let candidates = self
            .abort_expired_execution_invocation_candidates_with(client, dag_execution_id)
            .await?;
        let selected_candidate = select_invocation_abort_candidate(candidates, invocation_id)?;
        let execution = fetch_execution(client, dag_execution_id, &[]).await?;
        let dag_id = execution.object.data.dag_id();
        let dag = fetch_dag_in_context(client, &execution.context, dag_id).await?;
        let runtime = client.runtime_context(&[]).await?;

        let address = client.owner()?;
        let tx = invocation::abort_expired_ptb(
            &runtime,
            &dag.object_ref(),
            &execution.object.object_ref(),
            &selected_candidate.vertex,
            &selected_candidate.invocation_ref,
            None,
        )
        .map_err(NexusError::TransactionBuilding)?;
        let response = client.submit_transaction(tx, address).await?;

        Ok(AbortExpiredExecutionResult {
            tx_digest: response.digest,
            tx_checkpoint: response.checkpoint,
            dag_id,
            dag_execution_id,
            selected_candidate,
        })
    }
}

fn select_invocation_abort_candidate(
    candidates: Vec<InvocationAbortCandidate>,
    invocation_id: Option<sui::types::Address>,
) -> Result<InvocationAbortCandidate, NexusError> {
    if let Some(invocation_id) = invocation_id {
        candidates
            .into_iter()
            .find(|candidate| *candidate.invocation_ref.object_id() == invocation_id)
            .ok_or_else(|| {
                NexusError::Parsing(anyhow!(
                    "Invocation '{invocation_id}' is not currently eligible to abort this execution"
                ))
            })
    } else {
        candidates.into_iter().next().ok_or_else(|| {
            NexusError::Parsing(anyhow!(
                "No Invocation abort candidates are currently eligible for this execution"
            ))
        })
    }
}

impl ExecutionCostDetails {
    fn from_payment(payment_id: sui::types::Address, payment: ExecutionPaymentInnerV1) -> Self {
        Self {
            summary: ExecutionCostResult {
                payment_id,
                max_budget_mist: payment.max_budget_mist,
                locked_budget_mist: payment.locked_budget_mist,
                consumed: payment.consumed,
                outstanding_locks: payment.locks(),
                accomplished: payment.accomplished,
                refunded: payment.refunded,
            },
            outstanding_invocation_ids: payment
                .locked_vertices
                .iter()
                .map(|lock| lock.invocation_id.bytes)
                .collect(),
        }
    }
}

/// Derives the canonical payment lock key for one runtime Tool invocation.
///
/// This matches the key derived by `nexus_workflow::execution` from the exact
/// execution, [`RuntimeVertex`], and Tool identity.
pub fn payment_vertex_key(
    execution_id: sui::types::Address,
    vertex: &RuntimeVertex,
    tool_fqn: &crate::ToolFqn,
) -> anyhow::Result<Vec<u8>> {
    let mut bytes = Vec::new();
    bytes.extend_from_slice(b"nexus.payment.vertex.v1");
    bytes.extend(bcs::to_bytes(&execution_id)?);
    bytes.extend(bcs::to_bytes(vertex)?);
    bytes.extend(tool_fqn.to_string().as_bytes());
    Ok(Sha256::digest(bytes).to_vec())
}

fn pending_settlement_vertex(
    execution: &DAGExecutionInnerV1,
    walk_index: u64,
) -> Result<RuntimeVertex, NexusError> {
    let walk_index = usize::try_from(walk_index).map_err(|error| {
        NexusError::Parsing(anyhow!("Walk index does not fit this platform: {error}"))
    })?;
    match execution.walks.get(walk_index) {
        Some(DAGWalk::PendingSettlement { next_vertex, .. }) => Ok(next_vertex.clone()),
        Some(_) => Err(NexusError::Parsing(anyhow!(
            "Walk {walk_index} is not pending settlement"
        ))),
        None => Err(NexusError::Parsing(anyhow!(
            "Walk index {walk_index} is outside the execution"
        ))),
    }
}

async fn fetch_invocation_for_vertex(
    client: &NexusClient,
    context: &NexusContext,
    execution_id: sui::types::Address,
    dag_id: sui::types::Address,
    vertex: &RuntimeVertex,
) -> Result<sui::types::ObjectReference, NexusError> {
    let crawler = client.crawler();
    let dag = fetch_dag_in_context(client, context, dag_id).await?;
    let vertex_info = fetch_dag_vertex_bcs(crawler, context, &dag.data, vertex.vertex())
        .await
        .map_err(NexusError::Rpc)?
        .ok_or_else(|| {
            NexusError::Parsing(anyhow!("DAG vertex '{}' is missing", vertex.vertex_name()))
        })?;
    let tool_fqn = vertex_info.kind.tool_fqn().map_err(NexusError::Parsing)?;
    let vertex_key =
        payment_vertex_key(execution_id, vertex, &tool_fqn).map_err(NexusError::Parsing)?;
    let payment = tap::fetch_execution_payment_for_execution(client, context, execution_id)
        .await
        .map_err(NexusError::Rpc)?
        .data;
    let lock = payment
        .locked_vertices
        .iter()
        .find(|lock| lock.vertex_key == vertex_key)
        .ok_or_else(|| {
            NexusError::Parsing(anyhow!("Pending Tool result has no exact Invocation lock"))
        })?;
    crawler
        .get_object_metadata(lock.invocation_id.bytes)
        .await
        .map_err(NexusError::Rpc)
        .map(|metadata| metadata.object_ref())
}

struct InvocationAbortCandidateSeed {
    walk_index: usize,
    vertex: RuntimeVertex,
    payment_vertex_key: Vec<u8>,
    invocation_id: sui::types::Address,
}

fn filter_invocation_abort_candidate_walks(
    execution_id: sui::types::Address,
    vertices: &HashMap<graph_move::Vertex, graph_move::VertexInfo>,
    walks: &[DAGWalk],
    locks: &[ExecutionPaymentVertexLock],
    clock_ms: u64,
) -> anyhow::Result<Vec<InvocationAbortCandidateSeed>> {
    let mut candidates = Vec::new();
    for (walk_index, walk) in walks.iter().enumerate() {
        let Some(vertex) = walk.abortable_timeout_expired_vertex(clock_ms) else {
            continue;
        };
        let vertex_info = vertices.get(vertex.vertex()).ok_or_else(|| {
            anyhow!(
                "DAG vertex '{}' missing from fetched DAG",
                vertex.vertex_name()
            )
        })?;
        let tool_fqn = vertex_info.kind.tool_fqn()?;
        let vertex_key = payment_vertex_key(execution_id, vertex, &tool_fqn)?;
        if let Some(lock) = locks.iter().find(|lock| lock.vertex_key == vertex_key) {
            candidates.push(InvocationAbortCandidateSeed {
                walk_index,
                vertex: vertex.clone(),
                payment_vertex_key: vertex_key,
                invocation_id: lock.invocation_id.bytes,
            });
        }
    }
    Ok(candidates)
}

async fn fetch_invocation_abort_candidates(
    crawler: &Crawler,
    candidates: Vec<InvocationAbortCandidateSeed>,
) -> Result<Vec<InvocationAbortCandidate>, NexusError> {
    let mut result = Vec::with_capacity(candidates.len());
    for candidate in candidates {
        let invocation_ref = crawler
            .get_object_metadata(candidate.invocation_id)
            .await
            .map_err(NexusError::Rpc)?
            .object_ref();
        result.push(InvocationAbortCandidate {
            walk_index: candidate.walk_index,
            vertex: candidate.vertex,
            payment_vertex_key: candidate.payment_vertex_key,
            invocation_ref,
        });
    }
    result.sort_by_key(|candidate| candidate.walk_index);
    Ok(result)
}

#[cfg(test)]
mod tests {
    use {
        super::*,
        crate::{
            events::NexusEventKind,
            fqn,
            move_bindings::{
                interface::{
                    dag as dag_move,
                    graph::{self as graph_move, PostFailureAction, RuntimeVertex},
                    payment::{
                        ExecutionPayment,
                        ExecutionPaymentFeesRecordedEvent,
                        ExecutionPaymentFinalState,
                        ExecutionPaymentToolCostSnapshottedEvent,
                        PaymentSourceKind,
                        SkillPaymentPolicy,
                    },
                    verifier::{ToolVerifierMode, VerifierDecision},
                    version::InterfaceVersion,
                },
                move_std::{ascii::String as MoveString, option::Option as MoveOption},
                registry::{
                    era::V1 as RegistryWitnessV1,
                    leader::{LeaderRegistry, LeaderRegistryInnerV1},
                    priority_fee_vault::{PriorityFeeVault, PriorityFeeVaultInnerV1},
                },
                scheduler::scheduler::OccurrenceDispatchedEvent,
                sui_framework::{table::Table as MoveTable, vec_set::VecSet},
                tool::tool_registry::{
                    Tool as ToolAnchor,
                    ToolInnerV1,
                    ToolRef as ToolRegistryRef,
                    ToolRegistry,
                    ToolRegistryInnerV1,
                },
                workflow::{
                    execution::DagExecutionPaymentFieldKey,
                    execution_events::{
                        AgentSkillExecutionRequestedEvent,
                        CommittedToolResultEvent,
                        EndStateReachedEvent,
                        ExecutionFinishedEvent,
                        ExecutionPaymentRefilledEvent,
                        InvocationAuthorizationRequiredEvent,
                        RequestWalkExecutionEvent,
                        SubmissionFailureEvidenceRecordedEvent,
                        TerminalErrEvalRecordedEvent,
                        ToolVerificationResolvedEvent,
                        WalkAdvancedEvent,
                        WalkPendingAbortEvent,
                    },
                    execution_failure::WorkflowFailureClass,
                    invocation_adapter::{InvocationLockedEvent, InvocationSettledEvent},
                },
            },
            sui::traits::*,
            test_utils::{nexus_mocks, sui_mocks},
            types::NexusValue,
        },
        serde::Serialize,
        std::sync::{
            atomic::{AtomicBool, Ordering},
            Arc,
        },
    };

    fn clock_bcs(timestamp_ms: u64) -> Vec<u8> {
        bcs::to_bytes(&SuiClock::new(move_boundary::CLOCK_OBJECT_ID, timestamp_ms))
            .expect("clock BCS should serialize")
    }

    #[test]
    fn generated_sui_clock_bcs_matches_live_object_shape() {
        let bytes = clock_bcs(61_000);

        assert_eq!(bytes.len(), 40);
        let clock = bcs::from_bytes::<SuiClock>(&bytes).expect("clock BCS should decode");
        assert_eq!(clock.id.address(), move_boundary::CLOCK_OBJECT_ID);
        assert_eq!(clock.timestamp_ms, 61_000);
    }

    #[test]
    fn unresolved_timeout_skip_reason_distinguishes_pending_from_terminal_walks() {
        let active = DAGWalk::Active {
            next_vertex: RuntimeVertex::plain("tool"),
            timeout_ms: 30_000,
            requires_vertex_authorization_grant: false,
            created_at: 1_000,
        };
        let pending_settlement = DAGWalk::PendingSettlement {
            next_vertex: RuntimeVertex::plain("tool"),
            timeout_ms: 30_000,
            requires_vertex_authorization_grant: false,
            created_at: 1_000,
        };

        assert_eq!(
            unresolved_timeout_skip_reason(&active),
            EXPIRED_WALK_NOT_DOUBLE_TIMEOUT_EXPIRED_REASON
        );
        assert_eq!(
            unresolved_timeout_skip_reason(&pending_settlement),
            EXPIRED_WALK_NOT_DOUBLE_TIMEOUT_EXPIRED_REASON
        );
        assert_eq!(
            unresolved_timeout_skip_reason(&DAGWalk::Successful),
            EXPIRED_WALK_ALREADY_TERMINAL_REASON
        );
    }

    fn committed_tool_result_bcs(
        expected_vertex: RuntimeVertex,
        primary_leader: sui::types::Address,
        _secondary_leader: sui::types::Address,
        primary_failure: Option<FailureEvidenceKind>,
        secondary_failure: Option<FailureEvidenceKind>,
    ) -> execution_move::CommittedToolResult {
        execution_move::CommittedToolResult {
            expected_vertex,
            variant: graph_move::OutputVariant::new("ok"),
            variant_ports_to_data: crate::move_bindings::sui_framework::vec_map::VecMap {
                contents: vec![],
            },
            failure_evidence_kind: MoveOption::from_option(primary_failure),
            primary_failure_evidence_kind: MoveOption::from_option(primary_failure),
            secondary_failure_evidence_kind: MoveOption::from_option(secondary_failure),
            current_leader_cap_id: object_id(primary_leader),
            has_finalized_onchain_payload: false,
            leader_records: crate::move_bindings::sui_framework::vec_map::VecMap {
                contents: vec![crate::move_bindings::sui_framework::vec_map::Entry {
                    key: object_id(primary_leader),
                    value: execution_move::CommittedToolResultLeaderRecord {
                        commit_tx_digest: vec![1, 2, 3],
                        recipient: sui::types::Address::from_static("0x44"),
                        commit_gas_charge: MoveOption::from_option(Some(10)),
                        settlement_gas_charge: MoveOption::from_option(None),
                    },
                }],
            },
        }
    }

    fn raw_inline_nexus_data_bcs(
        one: impl Into<Vec<u8>>,
    ) -> crate::move_bindings::primitives::data::NexusData {
        crate::move_bindings::primitives::data::NexusData::inline_data(one.into())
            .expect("fixture is bounded")
    }

    fn object_id(bytes: sui::types::Address) -> crate::move_bindings::sui_framework::object::ID {
        crate::move_bindings::sui_framework::object::ID::new(bytes)
    }

    fn invocation_lock(
        vertex_key: Vec<u8>,
        invocation_id: sui::types::Address,
    ) -> ExecutionPaymentVertexLock {
        ExecutionPaymentVertexLock {
            vertex_key,
            invocation_id: object_id(invocation_id),
            amount: 10,
        }
    }

    fn output_variant(name: &str) -> crate::move_bindings::interface::graph::OutputVariant {
        crate::move_bindings::interface::graph::OutputVariant {
            name: MoveString::from(name),
        }
    }

    fn dag_bcs(vertices_size: u64) -> dag_move::DAGInnerV1 {
        dag_move::DAGInnerV1 {
            finalized: true,
            vertices: linked_table::LinkedTable::new(sui_mocks::mock_sui_address(), vertices_size),
            entry_groups: crate::move_bindings::sui_framework::vec_map::VecMap { contents: vec![] },
            edges: MoveTable::new(sui_mocks::mock_sui_address(), 0),
            outputs: MoveTable::new(sui_mocks::mock_sui_address(), 0),
            defaults_to_input_ports: MoveTable::new(sui_mocks::mock_sui_address(), 0),
            post_failure_action: MoveOption::from_option(None::<graph_move::PostFailureAction>),
        }
    }

    fn mock_get_dag_bcs(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &NexusContext,
        dag_ref: sui::types::ObjectReference,
        state: dag_move::DAGInnerV1,
    ) {
        let anchor = dag_move::DAG::new(crate::move_bindings::sui_framework::object::UID::new(
            *dag_ref.object_id(),
        ));
        sui_mocks::grpc::mock_object_state::<dag_move::DAG, InterfaceWitnessV1, dag_move::DAGInnerV1>(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            dag_ref,
            sui::types::Owner::Shared(0),
            anchor,
            state,
        );
    }

    #[tokio::test]
    async fn inspect_dag_succeeds_without_owned_coins() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let dag_ref = sui_mocks::mock_sui_object_ref();
        let mut state = dag_bcs(1);
        let vertices_id = state.vertices.id();
        let vertex = graph_move::Vertex::new("sum");
        state.vertices.head = Some(vertex.clone()).into();
        state.vertices.tail = Some(vertex.clone()).into();
        state
            .entry_groups
            .contents
            .push(crate::move_bindings::sui_framework::vec_map::Entry {
                key: graph_move::EntryGroup::new("main"),
                value: crate::move_bindings::sui_framework::vec_map::VecMap {
                    contents: vec![crate::move_bindings::sui_framework::vec_map::Entry {
                        key: graph_move::Vertex::new("sum"),
                        value: VecSet {
                            contents: vec![
                                graph_move::InputPort::new("right"),
                                graph_move::InputPort::new("left"),
                            ],
                        },
                    }],
                },
            });
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        sui_mocks::grpc::mock_get_dynamic_field_by_key(
            &mut ledger_service_mock,
            vertices_id,
            &crate::move_bindings::type_tag::<graph_move::Vertex>(&nexus_objects),
            vertex,
            offchain_vertex_node_bcs(&fqn!("xyz.taluslabs.fixture@1")),
        );
        mock_get_dag_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            dag_ref.clone(),
            state,
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client_without_coins(&nexus_objects, &rpc_url).await;

        let snapshot = client
            .workflow()
            .inspect_dag(*dag_ref.object_id())
            .await
            .expect("DAG inspection should not require a gas source");

        assert!(client.gas_config().is_none());
        assert_eq!(snapshot.dag_id, *dag_ref.object_id());
        assert_eq!(snapshot.vertex_count, 1);
        assert_eq!(
            snapshot.entry_groups["main"]["sum"],
            ["left".to_owned(), "right".to_owned()]
        );
    }

    #[tokio::test]
    async fn authorize_invocation_maps_leader_cap_metadata_failure() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let leader_cap_id = sui::types::Address::from_static("0x43");
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        sui_mocks::grpc::mock_get_object_not_found(&mut ledger_service_mock, leader_cap_id);
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client_without_coins(&nexus_objects, &rpc_url).await;

        let error = client
            .workflow()
            .authorize_invocation(
                sui::types::Address::from_static("0x42"),
                leader_cap_id,
                0,
                RuntimeVertex::plain("worker"),
                InvocationPolicy::FixedPrice,
            )
            .await
            .expect_err("missing leader capability should stop authorization");

        assert!(matches!(error, NexusError::Rpc(_)));
    }

    #[tokio::test]
    async fn authorize_invocation_validates_the_active_walk() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_ref = sui_mocks::mock_sui_object_ref();
        let dag_ref = sui_mocks::mock_sui_object_ref();
        let leader_cap_ref = sui_mocks::mock_sui_object_ref();
        let vertex = RuntimeVertex::plain("worker");
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        sui_mocks::grpc::mock_get_object_metadata_exact(
            &mut ledger_service_mock,
            leader_cap_ref.clone(),
            sui::types::Owner::Address(sui::types::Address::from_static("0x99")),
            None,
        );
        mock_get_dag_execution_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            execution_ref.clone(),
            &dag_ref,
            vec![execution_move::DAGWalk::Successful],
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client_without_coins(&nexus_objects, &rpc_url).await;

        let error = client
            .workflow()
            .authorize_invocation(
                *execution_ref.object_id(),
                *leader_cap_ref.object_id(),
                0,
                vertex.clone(),
                InvocationPolicy::FixedPrice,
            )
            .await
            .expect_err("a terminal walk must not be authorized");

        assert!(
            matches!(error, NexusError::Configuration(ref message) if message.contains("is not active")),
            "unexpected inactive walk error: {error:?}"
        );
    }

    #[tokio::test]
    async fn authorize_invocation_forwards_the_leader_capability() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_ref = sui_mocks::mock_sui_object_ref();
        let dag_ref = sui_mocks::mock_sui_object_ref();
        let leader_cap_ref = sui_mocks::mock_sui_object_ref();
        let vertex = RuntimeVertex::plain("worker");
        let tool_fqn = fqn!("xyz.taluslabs.authorize@1");
        let tool_id = crate::move_bindings::derive_tool_id(
            nexus_objects.tool_registry.object_id(),
            &tool_fqn,
        )
        .expect("tool ID derives");
        let cashier_origin = nexus_objects
            .type_origin(PackageRole::Tool, "tool_cashier", "ToolCashierKey")
            .expect("cashier origin resolves");
        let cashier_id = crate::move_bindings::derive_tool_cashier_id(cashier_origin, tool_id)
            .expect("cashier ID derives");
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        sui_mocks::grpc::mock_get_object_metadata_exact(
            &mut ledger_service_mock,
            leader_cap_ref.clone(),
            sui::types::Owner::Address(sui::types::Address::from_static("0x99")),
            None,
        );
        mock_get_dag_execution_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            execution_ref.clone(),
            &dag_ref,
            vec![execution_move::DAGWalk::Active {
                next_vertex: vertex.clone(),
                timeout_ms: 30_000,
                requires_vertex_authorization_grant: false,
                created_at: 1_000,
            }],
        );
        let dag = dag_bcs(1);
        let vertices_id = dag.vertices.id();
        mock_get_dag_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            dag_ref.clone(),
            dag,
        );
        sui_mocks::grpc::mock_get_dynamic_field_by_key(
            &mut ledger_service_mock,
            vertices_id,
            &crate::move_bindings::type_tag::<graph_move::Vertex>(&nexus_objects),
            graph_move::Vertex::new("worker"),
            offchain_vertex_node_bcs(&tool_fqn),
        );
        mock_object_state_for_authorize(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            tool_id,
            &tool_fqn,
        );
        sui_mocks::grpc::mock_get_object_metadata_exact(
            &mut ledger_service_mock,
            sui_mocks::object_ref_for_id(cashier_id),
            sui::types::Owner::Shared(2),
            None,
        );
        let payment_ref = sui_mocks::mock_sui_object_ref();
        mock_execution_payment_field(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            *execution_ref.object_id(),
            payment_ref,
            authorize_payment_fixture(*execution_ref.object_id()),
        );
        mock_tool_registry_observation(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
        );
        mock_leader_registry_observation(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
        );
        sui_mocks::grpc::mock_runtime_authority(&mut ledger_service_mock, &nexus_objects, false);
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client_without_coins(&nexus_objects, &rpc_url).await;

        let error = client
            .workflow()
            .authorize_invocation(
                *execution_ref.object_id(),
                *leader_cap_ref.object_id(),
                0,
                vertex,
                InvocationPolicy::Custom(InvocationPolicyCall::new(
                    TypeName::new("malformed_policy"),
                    vec![],
                )),
            )
            .await
            .expect_err("malformed policy should fail inside PTB construction");

        assert!(matches!(error, NexusError::TransactionBuilding(_)));
    }

    fn empty_object_table<T0, T1>(
        id: sui::types::Address,
    ) -> crate::move_bindings::sui_framework::object_table::ObjectTable<T0, T1> {
        crate::move_bindings::sui_framework::object_table::ObjectTable {
            id: crate::move_bindings::sui_framework::object::UID::new(id),
            size: 0,
            phantom_t0: std::marker::PhantomData,
            phantom_t1: std::marker::PhantomData,
        }
    }

    fn dag_execution_bcs(
        _execution_ref: &sui::types::ObjectReference,
        dag_ref: &sui::types::ObjectReference,
        walks: Vec<execution_move::DAGWalk>,
    ) -> execution_move::DAGExecutionInnerV1 {
        execution_move::DAGExecutionInnerV1 {
            dag: object_id(*dag_ref.object_id()),
            entry_group: graph_move::EntryGroup::new("default"),
            invoker: sui::types::Address::from_static("0x1"),
            created_at: 0,
            priority_fee_percentage: 0,
            agent_id: object_id(sui::types::Address::from_static("0xa")),
            skill_id: 11,
            interface_version: InterfaceVersion::new(7),
            task_id: object_id(sui::types::Address::from_static("0xb")),
            occurrence_id: 0,
            last_request_for_execution_emitted_at_digest: vec![],
            last_request_for_execution_leaders: vec![],
            network: object_id(sui::types::Address::from_static("0xf")),
            evaluations: empty_object_table(sui_mocks::mock_sui_address()),
            terminal_records: crate::move_bindings::sui_framework::vec_map::VecMap {
                contents: vec![],
            },
            submission_failure_records: crate::move_bindings::sui_framework::vec_map::VecMap {
                contents: vec![],
            },
            pending_retry_handoff_cap_ids: crate::move_bindings::sui_framework::vec_map::VecMap {
                contents: vec![],
            },
            walk_request_authorities: crate::move_bindings::sui_framework::vec_map::VecMap {
                contents: vec![],
            },
            pending_payment_settlements: crate::move_bindings::sui_framework::vec_map::VecMap {
                contents: vec![],
            },
            active_walks: walks
                .iter()
                .filter(|walk| matches!(walk, execution_move::DAGWalk::Active { .. }))
                .count() as u64,
            pending_abort_walks: 0,
            pending_settlement_walks: 0,
            successful_walks: 0,
            failed_walks: 0,
            aborted_walks: 0,
            consumed_walks: 0,
            cancelled_walks: 0,
            walks,
        }
    }

    fn mock_get_dag_execution_bcs(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &crate::types::NexusContext,
        execution_ref: sui::types::ObjectReference,
        dag_ref: &sui::types::ObjectReference,
        walks: Vec<execution_move::DAGWalk>,
    ) {
        let owner = sui::types::Owner::Shared(execution_ref.version());
        let execution = dag_execution_bcs(&execution_ref, dag_ref, walks);
        let anchor = execution_move::DAGExecution::new(
            crate::move_bindings::sui_framework::object::UID::new(*execution_ref.object_id()),
        );
        sui_mocks::grpc::mock_object_state::<
            execution_move::DAGExecution,
            WorkflowWitnessV1,
            execution_move::DAGExecutionInnerV1,
        >(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            execution_ref,
            owner,
            anchor,
            execution,
        );
    }

    fn mock_execution_observation_context(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &NexusContext,
        execution_ref: sui::types::ObjectReference,
    ) {
        mock_get_dag_execution_bcs(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            execution_ref,
            &sui_mocks::mock_sui_object_ref(),
            vec![],
        );
    }

    fn mock_package_graph(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        nexus_objects: &NexusContext,
    ) -> sui_mocks::grpc::MockMovePackageService {
        let mut package_service_mock = sui_mocks::grpc::MockMovePackageService::new();
        sui_mocks::grpc::mock_nexus_package_graph(
            ledger_service_mock,
            &mut package_service_mock,
            nexus_objects.packages(),
        );
        package_service_mock
    }

    fn mock_tool_registry_observation(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &NexusContext,
    ) {
        let root = nexus_objects.tool_registry;
        sui_mocks::grpc::mock_object_state_observation::<
            ToolRegistry,
            ToolWitnessV1,
            ToolRegistryInnerV1,
        >(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            sui_mocks::object_ref_for_id(root.object_id()),
            sui::types::Owner::Shared(root.initial_shared_version),
            ToolRegistry::new(crate::move_bindings::sui_framework::object::UID::new(
                root.object_id(),
            )),
        );
    }

    fn mock_leader_registry_observation(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &NexusContext,
    ) {
        let root = nexus_objects.leader_registry;
        sui_mocks::grpc::mock_object_state_observation::<
            LeaderRegistry,
            RegistryWitnessV1,
            LeaderRegistryInnerV1,
        >(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            sui_mocks::object_ref_for_id(root.object_id()),
            sui::types::Owner::Shared(root.initial_shared_version),
            LeaderRegistry::new(crate::move_bindings::sui_framework::object::UID::new(
                root.object_id(),
            )),
        );
    }

    fn mock_priority_fee_vault_observation(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &NexusContext,
    ) {
        let root = nexus_objects.priority_fee_vault;
        sui_mocks::grpc::mock_object_state_observation::<
            PriorityFeeVault,
            RegistryWitnessV1,
            PriorityFeeVaultInnerV1,
        >(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            sui_mocks::object_ref_for_id(root.object_id()),
            sui::types::Owner::Shared(root.initial_shared_version),
            PriorityFeeVault::new(crate::move_bindings::sui_framework::object::UID::new(
                root.object_id(),
            )),
        );
    }

    fn mock_execution_payment_field(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &NexusContext,
        execution_id: sui::types::Address,
        payment_ref: sui::types::ObjectReference,
        payment: ExecutionPaymentInnerV1,
    ) {
        let payment_id = *payment_ref.object_id();
        let anchor = ExecutionPayment::new(crate::move_bindings::sui_framework::object::UID::new(
            payment_id,
        ));
        sui_mocks::grpc::mock_get_dynamic_object_field_by_key(
            ledger_service_mock,
            execution_id,
            &crate::move_bindings::type_tag::<DagExecutionPaymentFieldKey>(nexus_objects),
            DagExecutionPaymentFieldKey::default(),
            payment_ref.clone(),
            sui::types::Owner::Shared(0),
            anchor.clone(),
        );
        sui_mocks::grpc::mock_object_state::<
            ExecutionPayment,
            InterfaceWitnessV1,
            ExecutionPaymentInnerV1,
        >(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            payment_ref,
            sui::types::Owner::Shared(0),
            anchor,
            payment,
        );
    }

    fn mock_object_state_for_authorize(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &NexusContext,
        tool_id: sui::types::Address,
        tool_fqn: &crate::ToolFqn,
    ) {
        sui_mocks::grpc::mock_object_state::<ToolAnchor, ToolWitnessV1, ToolInnerV1>(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            sui_mocks::object_ref_for_id(tool_id),
            sui::types::Owner::Shared(1),
            ToolAnchor::new(crate::move_bindings::sui_framework::object::UID::new(
                tool_id,
            )),
            ToolInnerV1::new(
                crate::move_bindings::sui_framework::object::ID::new(
                    nexus_objects.tool_registry.object_id(),
                ),
                MoveString::from(tool_fqn.to_string()),
                ToolRegistryRef::Http {
                    url: b"https://example.com/tool".to_vec(),
                },
                b"coverage fixture".to_vec(),
                MetaSchema::new(vec![], vec![]),
                false,
                crate::move_bindings::sui_framework::balance::Balance {
                    value: 0,
                    phantom_t0: std::marker::PhantomData,
                },
                false,
                0,
                0,
                MoveOption::from(Some(1)),
            ),
        );
    }

    fn authorize_payment_fixture(execution_id: sui::types::Address) -> ExecutionPaymentInnerV1 {
        ExecutionPaymentInnerV1 {
            execution_id,
            agent_id: object_id(sui::types::Address::from_static("0xa")),
            skill_id: 11,
            interface_revision: InterfaceVersion::new(7),
            payment_policy: SkillPaymentPolicy::UserFunded,
            source_kind: PaymentSourceKind::user_funded(sui::types::Address::from_static("0x1")),
            max_budget_mist: 100_000,
            gas_budget_mist: 83_334,
            priority_fee_reserve_mist: 16_666,
            locked_budget_mist: 0,
            funds: crate::move_bindings::sui_framework::balance::Balance {
                value: 100_000,
                phantom_t0: std::marker::PhantomData,
            },
            consumed: 0,
            tool_fee_charged: 0,
            priority_fee_charged: 0,
            priority_fee_percentage: 20,
            accomplished: false,
            refunded: false,
            final_state: ExecutionPaymentFinalState::Pending,
            tool_cost_snapshot: crate::move_bindings::sui_framework::vec_map::VecMap {
                contents: vec![],
            },
            locked_vertices: vec![],
        }
    }

    fn pending_settlement_walks(
        walk_index: usize,
        vertex: RuntimeVertex,
    ) -> Vec<execution_move::DAGWalk> {
        let mut walks = vec![execution_move::DAGWalk::Successful; walk_index];
        walks.push(execution_move::DAGWalk::PendingSettlement {
            next_vertex: vertex,
            timeout_ms: 30_000,
            requires_vertex_authorization_grant: false,
            created_at: 1_000,
        });
        walks
    }

    fn mock_exact_invocation_lock(
        ledger_service_mock: &mut sui_mocks::grpc::MockLedgerService,
        state_service_mock: &mut sui_mocks::grpc::MockStateService,
        nexus_objects: &NexusContext,
        execution_ref: &sui::types::ObjectReference,
        dag_ref: &sui::types::ObjectReference,
        vertex: &RuntimeVertex,
        tool_fqn: &crate::ToolFqn,
    ) -> sui::types::ObjectReference {
        let dag = dag_bcs(1);
        let vertices_id = dag.vertices.id();
        mock_get_dag_bcs(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            dag_ref.clone(),
            dag,
        );
        sui_mocks::grpc::mock_get_dynamic_field_by_key(
            ledger_service_mock,
            vertices_id,
            &crate::move_bindings::type_tag::<graph_move::Vertex>(nexus_objects),
            vertex.vertex().clone(),
            offchain_vertex_node_bcs(tool_fqn),
        );

        let payment_ref = sui_mocks::mock_sui_object_ref();
        let invocation_ref = sui_mocks::mock_sui_object_ref();
        let vertex_key = payment_vertex_key(*execution_ref.object_id(), vertex, tool_fqn)
            .expect("payment vertex key serializes");
        mock_execution_payment_field(
            ledger_service_mock,
            state_service_mock,
            nexus_objects,
            *execution_ref.object_id(),
            payment_ref.clone(),
            ExecutionPaymentInnerV1 {
                execution_id: *execution_ref.object_id(),
                agent_id: object_id(sui::types::Address::from_static("0xa")),
                skill_id: 11,
                interface_revision: InterfaceVersion::new(7),
                payment_policy: SkillPaymentPolicy::UserFunded,
                source_kind: PaymentSourceKind::user_funded(sui::types::Address::from_static(
                    "0x1",
                )),
                max_budget_mist: 100_000,
                gas_budget_mist: 83_334,
                priority_fee_reserve_mist: 16_666,
                locked_budget_mist: 10,
                funds: crate::move_bindings::sui_framework::balance::Balance {
                    value: 100_000,
                    phantom_t0: std::marker::PhantomData,
                },
                consumed: 0,
                tool_fee_charged: 0,
                priority_fee_charged: 0,
                priority_fee_percentage: 20,
                accomplished: false,
                refunded: false,
                final_state: ExecutionPaymentFinalState::Pending,
                tool_cost_snapshot: crate::move_bindings::sui_framework::vec_map::VecMap {
                    contents: vec![],
                },
                locked_vertices: vec![invocation_lock(vertex_key, *invocation_ref.object_id())],
            },
        );
        sui_mocks::grpc::mock_get_object_metadata_exact(
            ledger_service_mock,
            invocation_ref.clone(),
            sui::types::Owner::Object(*execution_ref.object_id()),
            None,
        );
        invocation_ref
    }

    fn offchain_vertex_info(tool_fqn: &crate::ToolFqn) -> graph_move::VertexInfo {
        graph_move::VertexInfo {
            kind: graph_move::VertexKind::OffChain {
                tool_fqn: tool_fqn.to_string().into(),
            },
            input_ports: VecSet { contents: vec![] },
            post_failure_action: MoveOption::from_option(None::<graph_move::PostFailureAction>),
            tool_id: object_id(sui::types::Address::from_static("0x42")),
            meta_schema: MoveOption::from_option(Some(
                crate::move_bindings::interface::meta_schema::MetaSchema::new(vec![], vec![]),
            )),
            verifier_mode: ToolVerifierMode::None,
        }
    }

    fn ports_data_map(
        entries: Vec<(&str, &'static [u8])>,
    ) -> crate::move_bindings::sui_framework::vec_map::VecMap<
        crate::move_bindings::interface::graph::OutputPort,
        crate::move_bindings::primitives::data::NexusData,
    > {
        crate::move_bindings::sui_framework::vec_map::VecMap {
            contents: entries
                .into_iter()
                .map(
                    |(name, value)| crate::move_bindings::sui_framework::vec_map::Entry {
                        key: crate::move_bindings::interface::graph::OutputPort {
                            name: MoveString::from(name),
                        },
                        value: crate::move_bindings::primitives::data::NexusData::inline_data(
                            value,
                        )
                        .expect("fixture is bounded"),
                    },
                )
                .collect(),
        }
    }

    async fn crawler_from_mocks(
        ledger_service_mock: sui_mocks::grpc::MockLedgerService,
        state_service_mock: sui_mocks::grpc::MockStateService,
    ) -> Crawler {
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = sui::grpc::client(rpc_url).expect("mock client");
        Crawler::new(Arc::new(client))
    }

    #[tokio::test]
    async fn fetch_committed_tool_result_for_walk_returns_none_when_absent() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_id = sui::types::Address::from_static("0xe1");
        let key = execution_move::CommittedToolResultKey { walk_index: 7 };
        let key_type = crate::move_bindings::type_tag::<execution_move::CommittedToolResultKey>(
            &nexus_objects,
        );
        let expected_id = execution_id.derive_dynamic_child_id(
            &key_type,
            &bcs::to_bytes(&key).expect("committed result key serializes"),
        );
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        sui_mocks::grpc::mock_get_object_not_found(&mut ledger_service_mock, expected_id);
        let crawler = crawler_from_mocks(
            ledger_service_mock,
            sui_mocks::grpc::MockStateService::new(),
        )
        .await;

        let result =
            fetch_committed_tool_result_for_walk(&crawler, &nexus_objects, execution_id, 7)
                .await
                .expect("fetch should succeed");

        assert!(result.is_none());
    }

    fn finalized_onchain_result(
        execution_id: sui::types::Address,
        transaction: sui::types::Digest,
    ) -> OnchainToolResultInnerV1 {
        OnchainToolResultInnerV1 {
            execution_id: ID::new(execution_id),
            finalized: true,
            stamps: Some(VecMap { contents: vec![] }).into(),
            tag: Some(b"ok".to_vec()).into(),
            named_payload: Some(VecMap { contents: vec![] }).into(),
            finalize_tx_digest: Some(transaction.into_inner().to_vec()).into(),
            finalize_recipient: Some(sui::types::Address::from_static("0xa1")).into(),
        }
    }

    #[tokio::test]
    async fn onchain_result_receipt_resolves_consumption_without_current_state_reads() {
        let context = sui_mocks::mock_nexus_context();
        let transaction = sui::types::Digest::new([1; 32]);
        let execution_id = sui::types::Address::from_static("0xe1");
        let result_id = sui::types::Address::from_static("0xe2");
        let receipt = nexus_mocks::mock_finalized_onchain_result(
            &context,
            transaction,
            execution_id,
            7,
            result_id,
            finalized_onchain_result(execution_id, transaction),
        );
        // No RPC is permitted, including a latest read for the result.
        let crawler = crawler_from_mocks(Default::default(), Default::default()).await;
        let state = onchain_tool_result_from_finalized_output(
            &crawler,
            &context,
            &receipt,
            transaction,
            execution_id,
            7,
        )
        .unwrap();
        let (_, result_ref) = state.consume_ready_result().unwrap();
        assert_eq!(*result_ref.object_id(), result_id);
        assert_eq!(
            result_ref.version(),
            11,
            "consumption uses the initial shared version"
        );
        assert!(onchain_tool_result_from_finalized_output(
            &crawler,
            &context,
            &receipt,
            transaction,
            execution_id,
            8
        )
        .is_err());
        assert!(onchain_tool_result_from_finalized_output(
            &crawler,
            &context,
            &receipt,
            transaction,
            result_id,
            7
        )
        .is_err());
    }

    #[tokio::test]
    async fn onchain_result_receipt_rejects_unproven_results() {
        let context = sui_mocks::mock_nexus_context();
        let transaction = sui::types::Digest::new([1; 32]);
        let execution_id = sui::types::Address::from_static("0xe1");
        let result_id = sui::types::Address::from_static("0xe2");
        let crawler = crawler_from_mocks(Default::default(), Default::default()).await;
        let valid = finalized_onchain_result(execution_id, transaction);
        let mut wrong_execution = valid.clone();
        wrong_execution.execution_id = ID::new(result_id);
        let mut unfinished = valid.clone();
        unfinished.finalized = false;
        let mut wrong_transaction = valid.clone();
        wrong_transaction.finalize_tx_digest = Some(vec![2; 32]).into();
        let mut missing_payload = valid.clone();
        missing_payload.named_payload = None.into();
        for result in [
            wrong_execution,
            unfinished,
            wrong_transaction,
            missing_payload,
        ] {
            let receipt = nexus_mocks::mock_finalized_onchain_result(
                &context,
                transaction,
                execution_id,
                7,
                result_id,
                result,
            );
            assert!(onchain_tool_result_from_finalized_output(
                &crawler,
                &context,
                &receipt,
                transaction,
                execution_id,
                7
            )
            .is_err());
        }
        let receipt = nexus_mocks::mock_finalized_onchain_result(
            &context,
            transaction,
            execution_id,
            7,
            result_id,
            valid,
        );
        for omitted in 0..3 {
            let mut incomplete = receipt.clone();
            incomplete.objects.as_mut().unwrap().objects.remove(omitted);
            assert!(onchain_tool_result_from_finalized_output(
                &crawler,
                &context,
                &incomplete,
                transaction,
                execution_id,
                7
            )
            .is_err());
        }
        let mut repeated = receipt.clone();
        repeated
            .objects
            .as_mut()
            .unwrap()
            .objects
            .push(receipt.objects().objects[0].clone());
        assert!(onchain_tool_result_from_finalized_output(
            &crawler,
            &context,
            &repeated,
            transaction,
            execution_id,
            7
        )
        .is_err());
        let mut inherited = receipt.clone();
        inherited.objects.as_mut().unwrap().objects[2]
            .set_previous_transaction(sui::types::Digest::new([2; 32]));
        assert!(onchain_tool_result_from_finalized_output(
            &crawler,
            &context,
            &inherited,
            transaction,
            execution_id,
            7
        )
        .is_err());
    }

    #[tokio::test]
    async fn onchain_result_state_reports_no_result_when_dynamic_fields_are_absent() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_id = sui::types::Address::from_static("0xe1");
        let committed_key = execution_move::CommittedToolResultKey { walk_index: 7 };
        let insufficient_key = insufficient_settlement_field_key();
        let result_key = execution_move::OnchainToolResultKey { walk_index: 7 };
        let committed_id = execution_id.derive_dynamic_child_id(
            &crate::move_bindings::type_tag::<execution_move::CommittedToolResultKey>(
                &nexus_objects,
            ),
            &bcs::to_bytes(&committed_key).expect("committed key serializes"),
        );
        let insufficient_id = execution_id.derive_dynamic_child_id(
            &crate::move_bindings::type_tag::<
                execution_move::ExecutionPaymentInsufficientSettlementFieldKey,
            >(&nexus_objects),
            &bcs::to_bytes(&insufficient_key).expect("insufficient settlement key serializes"),
        );
        let result_id = execution_id.derive_dynamic_child_id(
            &crate::move_bindings::type_tag::<execution_move::OnchainToolResultKey>(&nexus_objects),
            &bcs::to_bytes(&result_key).expect("result key serializes"),
        );
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        for field_id in [committed_id, insufficient_id, result_id] {
            sui_mocks::grpc::mock_get_object_not_found(&mut ledger_service_mock, field_id);
        }
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client_without_coins(&nexus_objects, &rpc_url).await;

        let state =
            fetch_onchain_tool_result_state_for_walk(&client, &nexus_objects, execution_id, 7)
                .await
                .expect("absent dynamic fields should be a valid empty state");

        assert!(matches!(state, OnchainToolResultState::NoResult));
    }

    #[tokio::test]
    async fn fetch_committed_tool_result_for_walk_reads_exact_key_and_decodes_match() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_id = sui::types::Address::from_static("0xe1");
        let primary_leader = sui::types::Address::from_static("0xa1");
        let secondary_leader = sui::types::Address::from_static("0xa2");
        let committed = committed_tool_result_bcs(
            RuntimeVertex::plain("retryable"),
            primary_leader,
            secondary_leader,
            Some(FailureEvidenceKind::ToolEvidence),
            None,
        );
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        sui_mocks::grpc::mock_get_dynamic_field_by_key(
            &mut ledger_service_mock,
            execution_id,
            &crate::move_bindings::type_tag::<execution_move::CommittedToolResultKey>(
                &nexus_objects,
            ),
            execution_move::CommittedToolResultKey { walk_index: 7 },
            committed,
        );
        let crawler = crawler_from_mocks(
            ledger_service_mock,
            sui_mocks::grpc::MockStateService::new(),
        )
        .await;

        let result =
            fetch_committed_tool_result_for_walk(&crawler, &nexus_objects, execution_id, 7)
                .await
                .expect("fetch should succeed")
                .expect("committed result should exist");

        assert_eq!(result.expected_vertex, RuntimeVertex::plain("retryable"));
        assert_eq!(
            result.primary_failure_evidence_kind,
            Some(FailureEvidenceKind::ToolEvidence)
        );
        assert_eq!(result.secondary_failure_evidence_kind, None);
        assert!(result.leader_record(primary_leader).is_some());
        assert!(result.leader_record(secondary_leader).is_none());
    }

    #[tokio::test]
    async fn fetch_committed_tool_result_for_walk_decodes_metadata_with_raw_output_payload() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_id = sui::types::Address::from_static("0xe1");
        let primary_leader = sui::types::Address::from_static("0xa1");
        let secondary_leader = sui::types::Address::from_static("0xa2");
        let mut committed = committed_tool_result_bcs(
            RuntimeVertex::plain("retryable"),
            primary_leader,
            secondary_leader,
            Some(FailureEvidenceKind::ToolEvidence),
            None,
        );
        committed.variant_ports_to_data = crate::move_bindings::sui_framework::vec_map::VecMap {
            contents: vec![crate::move_bindings::sui_framework::vec_map::Entry {
                key: graph_move::OutputPort::new("reason"),
                value: raw_inline_nexus_data_bcs(b"not-json".to_vec()),
            }],
        };
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        sui_mocks::grpc::mock_get_dynamic_field_by_key(
            &mut ledger_service_mock,
            execution_id,
            &crate::move_bindings::type_tag::<execution_move::CommittedToolResultKey>(
                &nexus_objects,
            ),
            execution_move::CommittedToolResultKey { walk_index: 7 },
            committed,
        );
        let crawler = crawler_from_mocks(
            ledger_service_mock,
            sui_mocks::grpc::MockStateService::new(),
        )
        .await;

        let result =
            fetch_committed_tool_result_for_walk(&crawler, &nexus_objects, execution_id, 7)
                .await
                .expect("fetch should ignore raw output payload bytes")
                .expect("committed result should exist");

        assert_eq!(result.expected_vertex, RuntimeVertex::plain("retryable"));
        assert_eq!(
            result.primary_failure_evidence_kind,
            Some(FailureEvidenceKind::ToolEvidence)
        );
        assert_eq!(result.secondary_failure_evidence_kind, None);
        assert!(result.leader_record(primary_leader).is_some());
        assert!(result.leader_record(secondary_leader).is_none());
    }

    fn offchain_vertex_node_bcs(
        tool_fqn: &crate::ToolFqn,
    ) -> linked_table::Node<graph_move::Vertex, graph_move::VertexInfo> {
        linked_table::Node {
            prev: crate::move_bindings::move_std::option::Option::from_option(
                None::<graph_move::Vertex>,
            ),
            next: crate::move_bindings::move_std::option::Option::from_option(
                None::<graph_move::Vertex>,
            ),
            value: offchain_vertex_info(tool_fqn),
        }
    }

    fn supported_grpc_events(
        objects: &crate::types::NexusContext,
        nexus_events: Vec<NexusEventKind>,
    ) -> Vec<sui::grpc::Event> {
        #[derive(Serialize)]
        struct Wrapper<T> {
            event: T,
        }

        let primitives = objects
            .require_package(PackageRole::Primitives)
            .unwrap()
            .initial_id;
        let interface = objects
            .require_package(PackageRole::Interface)
            .unwrap()
            .initial_id;
        let workflow = objects
            .require_package(PackageRole::Workflow)
            .unwrap()
            .initial_id;

        nexus_events
            .into_iter()
            .map(|event| {
                let (event_pkg_id, event_type_origin, module) =
                    if matches!(event, NexusEventKind::DAGCreated(_)) {
                        (interface, interface, "dag")
                    } else {
                        (workflow, workflow, "execution")
                    };
                let event_type = format!(
                    "{}::event::EventWrapper<{}::{module}::{}>",
                    primitives,
                    event_type_origin,
                    event.name()
                );
                let contents = match event {
                    NexusEventKind::WalkAdvanced(event) => {
                        bcs::to_bytes(&Wrapper { event }).unwrap()
                    }
                    NexusEventKind::EndStateReached(event) => {
                        bcs::to_bytes(&Wrapper { event }).unwrap()
                    }
                    NexusEventKind::ExecutionFinished(event) => {
                        bcs::to_bytes(&Wrapper { event }).unwrap()
                    }
                    NexusEventKind::TerminalErrEvalRecorded(event) => {
                        bcs::to_bytes(&Wrapper { event }).unwrap()
                    }
                    NexusEventKind::DAGCreated(event) => bcs::to_bytes(&Wrapper { event }).unwrap(),
                    _ => panic!("Unsupported event type for BCS serialization"),
                };

                let mut grpc_event = sui::grpc::Event::default();
                grpc_event.set_package_id(event_pkg_id);
                grpc_event.set_module(module.to_string());
                grpc_event.set_sender(sui::types::Address::ZERO);
                grpc_event.set_event_type(event_type);
                grpc_event.set_contents(contents);
                grpc_event
            })
            .collect()
    }

    fn expect_object_update_reference(
        ledger_service: &mut sui_mocks::grpc::MockLedgerService,
        objects: &crate::types::NexusContext,
        object_ref: sui::types::ObjectReference,
        initial_shared_version: sui::types::Version,
        expected_requested_version: Option<sui::types::Version>,
        previous_transaction: sui::types::Digest,
    ) {
        let object_type = crate::move_bindings::struct_tag::<DAGExecution>(objects).to_string();
        ledger_service
            .expect_get_object()
            .times(1)
            .returning(move |request| {
                assert_eq!(request.get_ref().version, expected_requested_version);
                let mut response = sui::grpc::GetObjectResponse::default();
                let mut object = sui::grpc::Object::default();
                object.set_object_id(*object_ref.object_id());
                object.set_owner(sui::grpc::Owner::from(sui::types::Owner::Shared(
                    initial_shared_version,
                )));
                object.set_object_type(object_type.clone());
                object.set_version(object_ref.version());
                object.set_digest(*object_ref.digest());
                object.set_previous_transaction(previous_transaction.to_string());
                response.set_object(object);
                Ok(tonic::Response::new(response))
            });
    }

    #[allow(clippy::too_many_arguments)]
    fn expect_transaction_update(
        ledger_service: &mut sui_mocks::grpc::MockLedgerService,
        objects: &crate::types::NexusContext,
        digest: sui::types::Digest,
        execution_id: sui::types::Address,
        version: sui::types::Version,
        output_digest: sui::types::Digest,
        input_state: sui::types::ObjectIn,
        nexus_events: Vec<NexusEventKind>,
    ) {
        let grpc_events = supported_grpc_events(objects, nexus_events);
        let served = Arc::new(AtomicBool::new(false));
        ledger_service
            .expect_get_transaction()
            .withf(move |request| {
                request.get_ref().digest_opt() == Some(digest.to_string().as_str())
            })
            .returning(move |request| {
                if served.swap(true, Ordering::SeqCst) {
                    return Err(tonic::Status::failed_precondition(format!(
                        "transaction '{digest}' was fetched again after its object version was already reconstructed"
                    )));
                }
                assert_eq!(
                    request.get_ref().digest_opt(),
                    Some(digest.to_string().as_str())
                );
                let effects = sui::types::TransactionEffects::V2(Box::new(
                    sui::types::TransactionEffectsV2 {
                        status: sui::types::ExecutionStatus::Success,
                        epoch: 1,
                        gas_used: sui::types::GasCostSummary {
                            computation_cost: 0,
                            storage_cost: 0,
                            storage_rebate: 0,
                            non_refundable_storage_fee: 0,
                        },
                        transaction_digest: digest,
                        gas_object_index: None,
                        events_digest: None,
                        dependencies: vec![],
                        lamport_version: version,
                        changed_objects: vec![sui::types::ChangedObject {
                            object_id: execution_id,
                            input_state: input_state.clone(),
                            output_state: sui::types::ObjectOut::ObjectWrite {
                                digest: output_digest,
                                owner: sui::types::Owner::Shared(1),
                            },
                            id_operation: if matches!(input_state, sui::types::ObjectIn::NotExist) {
                                sui::types::IdOperation::Created
                            } else {
                                sui::types::IdOperation::None
                            },
                        }],
                        unchanged_consensus_objects: vec![],
                        auxiliary_data_digest: None,
                    },
                ));
                let mut grpc_effects = sui::grpc::TransactionEffects::default();
                grpc_effects.set_bcs(bcs::to_bytes(&effects).unwrap());
                let mut events = sui::grpc::TransactionEvents::default();
                events.set_events(grpc_events.clone());
                let mut transaction = sui::grpc::ExecutedTransaction::default();
                transaction.set_digest(digest);
                transaction.set_checkpoint(version);
                transaction.set_effects(grpc_effects);
                transaction.set_events(events);
                let mut response = sui::grpc::GetTransactionResponse::default();
                response.set_transaction(transaction);
                Ok(tonic::Response::new(response))
            });
    }

    #[tokio::test]
    async fn test_workflow_actions_publish() {
        let mut rng = rand::thread_rng();
        let digest = sui::types::Digest::generate(&mut rng);
        let gas_coin_ref = sui_mocks::mock_sui_object_ref();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let dag_object_id = sui::types::Address::generate(&mut rng);

        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        let mut tx_service_mock = sui_mocks::grpc::MockTransactionExecutionService::new();
        let mut sub_service_mock = sui_mocks::grpc::MockSubscriptionService::new();

        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        mock_tool_registry_observation(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);

        let dag_created = sui::types::Object::new(
            sui::types::ObjectData::Struct(
                sui::types::MoveStruct::new(
                    crate::move_bindings::struct_tag::<dag_move::DAG>(&nexus_objects),
                    true,
                    0,
                    dag_object_id.to_bcs().unwrap(),
                )
                .unwrap(),
            ),
            sui::types::Owner::Shared(0),
            digest,
            1000,
        );

        let submitted = sui_mocks::grpc::mock_execute_transaction_and_wait_for_checkpoint(
            &mut tx_service_mock,
            &mut sub_service_mock,
            &mut ledger_service_mock,
            gas_coin_ref.clone(),
            vec![dag_created],
            vec![],
            vec![],
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            execution_service_mock: Some(tx_service_mock),
            subscription_service_mock: Some(sub_service_mock),
            ..Default::default()
        });

        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;

        let dag = DagSpec::default();

        let result = client
            .workflow()
            .publish(
                nexus_objects
                    .require_package(PackageRole::Workflow)
                    .unwrap()
                    .storage_id,
                dag,
            )
            .await
            .expect("Failed to publish DAG");

        assert_eq!(result.dag_object_id, dag_object_id);
        assert_eq!(result.tx_digest, submitted.digest());
        assert_eq!(result.tx_checkpoint, 1);
    }

    #[test]
    fn inspect_execution_options_defaults_are_stable() {
        assert_eq!(
            InspectExecutionOptions::default(),
            InspectExecutionOptions {
                timeout: Duration::from_secs(60 * 60),
                poll_interval: Duration::from_secs(1),
            }
        );
    }

    #[tokio::test]
    async fn inspect_execution_rejects_zero_poll_interval_before_spawning() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;

        let result = client
            .workflow()
            .inspect_execution(
                sui::types::Address::ZERO,
                InspectExecutionOptions {
                    timeout: Duration::from_secs(1),
                    poll_interval: Duration::ZERO,
                },
            )
            .await;

        assert!(matches!(result, Err(NexusError::Configuration(_))));
    }
    #[tokio::test]
    async fn inspect_execution_replays_update_chain_in_order() {
        let mut rng = rand::thread_rng();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let dag_object_id = sui::types::Address::generate(&mut rng);
        let execution_object_id = sui::types::Address::generate(&mut rng);
        let version_1 = sui::types::ObjectReference::new(
            execution_object_id,
            1,
            sui::types::Digest::generate(&mut rng),
        );
        let version_9 = sui::types::ObjectReference::new(
            execution_object_id,
            9,
            sui::types::Digest::generate(&mut rng),
        );
        let version_20 = sui::types::ObjectReference::new(
            execution_object_id,
            20,
            sui::types::Digest::generate(&mut rng),
        );
        let tx_1 = sui::types::Digest::generate(&mut rng);
        let tx_9 = sui::types::Digest::generate(&mut rng);
        let tx_20 = sui::types::Digest::generate(&mut rng);
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        mock_execution_observation_context(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            version_20.clone(),
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            version_20.clone(),
            1,
            None,
            tx_20,
        );
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            version_9.clone(),
            1,
            Some(9),
            tx_9,
        );
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            version_1.clone(),
            1,
            Some(1),
            tx_1,
        );

        let walk_advanced = NexusEventKind::WalkAdvanced(WalkAdvancedEvent {
            dag: object_id(dag_object_id),
            execution: object_id(execution_object_id),
            walk_index: 0,
            vertex: RuntimeVertex::plain("initial"),
            variant: output_variant("ok"),
            variant_ports_to_data: ports_data_map(vec![]),
        });
        let end_state_reached = NexusEventKind::EndStateReached(EndStateReachedEvent {
            dag: object_id(dag_object_id),
            execution: object_id(execution_object_id),
            walk_index: 0,
            vertex: RuntimeVertex::plain("final"),
            variant: output_variant("ok"),
            variant_ports_to_data: ports_data_map(vec![("answer", b"42")]),
        });
        let execution_finished = NexusEventKind::ExecutionFinished(ExecutionFinishedEvent {
            dag: object_id(dag_object_id),
            execution: object_id(execution_object_id),
            has_any_walk_failed: false,
            has_any_walk_succeeded: true,
            was_aborted: false,
        });
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            tx_20,
            execution_object_id,
            20,
            *version_20.digest(),
            sui::types::ObjectIn::Exist {
                version: 9,
                digest: *version_9.digest(),
                owner: sui::types::Owner::Shared(1),
            },
            vec![end_state_reached, execution_finished],
        );
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            tx_9,
            execution_object_id,
            9,
            *version_9.digest(),
            sui::types::ObjectIn::Exist {
                version: 1,
                digest: *version_1.digest(),
                owner: sui::types::Owner::Shared(1),
            },
            vec![walk_advanced],
        );
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            tx_1,
            execution_object_id,
            1,
            *version_1.digest(),
            sui::types::ObjectIn::NotExist,
            vec![],
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client_without_coins(&nexus_objects, &rpc_url).await;
        let mut result = client
            .workflow()
            .inspect_execution(
                execution_object_id,
                InspectExecutionOptions {
                    timeout: Duration::from_secs(2),
                    poll_interval: Duration::from_millis(10),
                },
            )
            .await
            .expect("object inspection should start");

        let mut events = Vec::new();
        while let Some(event) = result.next_event.recv().await {
            events.push(event);
        }

        assert!(result.poller.await.unwrap().is_ok());
        assert_eq!(events.len(), 3);
        assert!(matches!(events[0].data, NexusEventKind::WalkAdvanced(_)));
        assert!(matches!(events[1].data, NexusEventKind::EndStateReached(_)));
        assert!(matches!(
            events[2].data,
            NexusEventKind::ExecutionFinished(_)
        ));
    }

    #[tokio::test]
    async fn inspect_execution_retries_missing_transaction_until_deadline() {
        let mut rng = rand::thread_rng();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let dag_object_id = sui::types::Address::generate(&mut rng);
        let execution_object_id = sui::types::Address::generate(&mut rng);
        let version_1 = sui::types::ObjectReference::new(
            execution_object_id,
            1,
            sui::types::Digest::generate(&mut rng),
        );
        let version_7 = sui::types::ObjectReference::new(
            execution_object_id,
            7,
            sui::types::Digest::generate(&mut rng),
        );
        let tx_1 = sui::types::Digest::generate(&mut rng);
        let tx_7 = sui::types::Digest::generate(&mut rng);
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        mock_execution_observation_context(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            version_1.clone(),
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            version_1.clone(),
            1,
            None,
            tx_1,
        );
        ledger_service_mock
            .expect_get_object()
            .times(1)
            .returning(|_| Err(tonic::Status::aborted("retry object observation")));
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            version_7.clone(),
            1,
            None,
            tx_7,
        );
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            tx_1,
            execution_object_id,
            1,
            *version_1.digest(),
            sui::types::ObjectIn::NotExist,
            vec![NexusEventKind::WalkAdvanced(WalkAdvancedEvent {
                dag: object_id(dag_object_id),
                execution: object_id(execution_object_id),
                walk_index: 0,
                vertex: RuntimeVertex::plain("initial"),
                variant: output_variant("ok"),
                variant_ports_to_data: ports_data_map(vec![]),
            })],
        );
        ledger_service_mock
            .expect_get_transaction()
            .withf(move |request| request.get_ref().digest_opt() == Some(tx_7.to_string().as_str()))
            .times(6)
            .returning(|_| Err(tonic::Status::not_found("transaction not visible yet")));
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            tx_7,
            execution_object_id,
            7,
            *version_7.digest(),
            sui::types::ObjectIn::Exist {
                version: 1,
                digest: *version_1.digest(),
                owner: sui::types::Owner::Shared(1),
            },
            vec![NexusEventKind::ExecutionFinished(ExecutionFinishedEvent {
                dag: object_id(dag_object_id),
                execution: object_id(execution_object_id),
                has_any_walk_failed: false,
                has_any_walk_succeeded: true,
                was_aborted: false,
            })],
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;
        let mut result = client
            .workflow()
            .inspect_execution(
                execution_object_id,
                InspectExecutionOptions {
                    timeout: Duration::from_secs(2),
                    poll_interval: Duration::from_millis(10),
                },
            )
            .await
            .expect("object inspection should start");

        let mut events = Vec::new();
        while let Some(event) = result.next_event.recv().await {
            events.push(event);
        }
        assert_eq!(events.len(), 2);
        assert!(matches!(events[0].data, NexusEventKind::WalkAdvanced(_)));
        assert!(matches!(
            events[1].data,
            NexusEventKind::ExecutionFinished(_)
        ));
        assert!(result.poller.await.unwrap().is_ok());
    }

    #[tokio::test]
    async fn inspect_execution_does_not_refetch_unchanged_transaction() {
        let mut rng = rand::thread_rng();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_object_id = sui::types::Address::generate(&mut rng);
        let object_ref = sui::types::ObjectReference::new(
            execution_object_id,
            1,
            sui::types::Digest::generate(&mut rng),
        );
        let update_tx = sui::types::Digest::generate(&mut rng);
        let execution_type =
            crate::move_bindings::struct_tag::<DAGExecution>(&nexus_objects).to_string();
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        mock_execution_observation_context(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            object_ref.clone(),
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        let repeated_ref = object_ref.clone();
        ledger_service_mock
            .expect_get_object()
            .returning(move |request| {
                assert_eq!(request.get_ref().version, None);
                let mut response = sui::grpc::GetObjectResponse::default();
                let mut object = sui::grpc::Object::default();
                object.set_object_id(*repeated_ref.object_id());
                object.set_owner(sui::grpc::Owner::from(sui::types::Owner::Shared(1)));
                object.set_object_type(execution_type.clone());
                object.set_version(repeated_ref.version());
                object.set_digest(*repeated_ref.digest());
                object.set_previous_transaction(update_tx.to_string());
                response.set_object(object);
                Ok(tonic::Response::new(response))
            });
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            update_tx,
            execution_object_id,
            1,
            *object_ref.digest(),
            sui::types::ObjectIn::NotExist,
            vec![],
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;
        let mut result = client
            .workflow()
            .inspect_execution(
                execution_object_id,
                InspectExecutionOptions {
                    timeout: Duration::from_millis(120),
                    poll_interval: Duration::from_millis(20),
                },
            )
            .await
            .expect("object inspection should start");

        assert!(result.next_event.recv().await.is_none());
        let poller_result = result.poller.await.unwrap();
        assert!(
            matches!(poller_result, Err(NexusError::Timeout(_))),
            "unexpected inspection result: {poller_result:?}"
        );
    }

    #[tokio::test]
    async fn inspect_execution_surfaces_invalid_update_chain() {
        let mut rng = rand::thread_rng();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_object_id = sui::types::Address::generate(&mut rng);
        let object_ref = sui::types::ObjectReference::new(
            execution_object_id,
            1,
            sui::types::Digest::generate(&mut rng),
        );
        let update_tx = sui::types::Digest::generate(&mut rng);
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        mock_execution_observation_context(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            object_ref.clone(),
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            object_ref,
            1,
            None,
            update_tx,
        );
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            update_tx,
            execution_object_id,
            1,
            sui::types::Digest::generate(&mut rng),
            sui::types::ObjectIn::NotExist,
            vec![],
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;
        let mut result = client
            .workflow()
            .inspect_execution(
                execution_object_id,
                InspectExecutionOptions {
                    timeout: Duration::from_secs(1),
                    poll_interval: Duration::from_millis(10),
                },
            )
            .await
            .expect("object inspection should start");

        assert!(result.next_event.recv().await.is_none());
        let error = result
            .poller
            .await
            .unwrap()
            .expect_err("invalid update chain should fail");
        assert!(error.to_string().contains("output digest"));
    }

    #[tokio::test]
    async fn inspect_execution_reports_missing_transaction_with_reconstruction_context() {
        let mut rng = rand::thread_rng();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_object_id = sui::types::Address::generate(&mut rng);
        let object_ref = sui::types::ObjectReference::new(
            execution_object_id,
            5,
            sui::types::Digest::generate(&mut rng),
        );
        let missing_tx = sui::types::Digest::generate(&mut rng);
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        mock_execution_observation_context(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            object_ref.clone(),
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            object_ref,
            1,
            None,
            missing_tx,
        );
        ledger_service_mock
            .expect_get_transaction()
            .withf(move |request| {
                request.get_ref().digest_opt() == Some(missing_tx.to_string().as_str())
            })
            .times(1..)
            .returning(|_| Err(tonic::Status::not_found("transaction pruned")));

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;
        let mut result = client
            .workflow()
            .inspect_execution(
                execution_object_id,
                InspectExecutionOptions {
                    timeout: Duration::from_secs(1),
                    poll_interval: Duration::from_millis(10),
                },
            )
            .await
            .expect("inspection task should start");

        assert!(result.next_event.recv().await.is_none());
        let error = result.poller.await.unwrap().expect_err("history must fail");
        let message = error.to_string();
        assert!(
            message.contains(&execution_object_id.to_string()),
            "unexpected inspection error: {message}"
        );
        assert!(
            message.contains(&missing_tx.to_string()),
            "unexpected inspection error: {message}"
        );
        assert!(
            message.contains("last successfully reconstructed version none"),
            "unexpected inspection error: {message}"
        );
    }

    #[tokio::test]
    async fn inspect_execution_reports_missing_historical_version() {
        let mut rng = rand::thread_rng();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_object_id = sui::types::Address::generate(&mut rng);
        let version_9 = sui::types::ObjectReference::new(
            execution_object_id,
            9,
            sui::types::Digest::generate(&mut rng),
        );
        let version_1_digest = sui::types::Digest::generate(&mut rng);
        let tx_9 = sui::types::Digest::generate(&mut rng);
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        mock_execution_observation_context(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            version_9.clone(),
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            version_9.clone(),
            1,
            None,
            tx_9,
        );
        ledger_service_mock
            .expect_get_object()
            .times(1)
            .returning(|request| {
                assert_eq!(request.get_ref().version, Some(1));
                Err(tonic::Status::not_found("historical object pruned"))
            });
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            tx_9,
            execution_object_id,
            9,
            *version_9.digest(),
            sui::types::ObjectIn::Exist {
                version: 1,
                digest: version_1_digest,
                owner: sui::types::Owner::Shared(1),
            },
            vec![],
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;
        let mut result = client
            .workflow()
            .inspect_execution(
                execution_object_id,
                InspectExecutionOptions {
                    timeout: Duration::from_secs(1),
                    poll_interval: Duration::from_millis(10),
                },
            )
            .await
            .expect("inspection task should start");

        assert!(result.next_event.recv().await.is_none());
        let error = result.poller.await.unwrap().expect_err("history must fail");
        let message = error.to_string();
        assert!(message.contains(&execution_object_id.to_string()));
        assert!(message.contains("missing object version 1"));
        assert!(message.contains("last successfully reconstructed version none"));
    }

    #[cfg(feature = "walrus")]
    #[tokio::test]
    async fn test_workflow_actions_inspect_execution_until_completion() {
        let mut rng = rand::thread_rng();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let dag_object_id = sui::types::Address::generate(&mut rng);
        let execution_object_id = sui::types::Address::generate(&mut rng);
        let object_ref = sui::types::ObjectReference::new(
            execution_object_id,
            42,
            sui::types::Digest::generate(&mut rng),
        );
        let update_tx = sui::types::Digest::generate(&mut rng);

        let walk_advanced_event = NexusEventKind::WalkAdvanced(WalkAdvancedEvent {
            dag: object_id(dag_object_id),
            execution: object_id(execution_object_id),
            walk_index: 0,
            vertex: RuntimeVertex::plain("initial"),
            variant: output_variant("ok"),
            variant_ports_to_data: ports_data_map(vec![]),
        });
        let terminal_err_eval_event =
            NexusEventKind::TerminalErrEvalRecorded(TerminalErrEvalRecordedEvent {
                dag: object_id(dag_object_id),
                execution: object_id(execution_object_id),
                walk_index: 1,
                vertex: RuntimeVertex::plain("failable"),
                leader: sui::types::Address::THREE,
                failure_class: WorkflowFailureClass::TerminalToolFailure,
                outcome: MoveOption::from_option(Some(PostFailureAction::Terminate)),
                reason: MoveString::from("tool failed"),
                err_eval_hash: vec![9, 8, 7],
                duplicate: false,
            });
        let end_state_reached_event = NexusEventKind::EndStateReached(EndStateReachedEvent {
            dag: object_id(dag_object_id),
            execution: object_id(execution_object_id),
            walk_index: 0,
            vertex: RuntimeVertex::plain("final"),
            variant: output_variant("ok"),
            variant_ports_to_data: ports_data_map(vec![("answer", b"42")]),
        });
        let execution_finished_event = NexusEventKind::ExecutionFinished(ExecutionFinishedEvent {
            dag: object_id(dag_object_id),
            execution: object_id(execution_object_id),
            has_any_walk_failed: true,
            has_any_walk_succeeded: true,
            was_aborted: false,
        });

        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        mock_execution_observation_context(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            object_ref.clone(),
        );
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        expect_object_update_reference(
            &mut ledger_service_mock,
            &nexus_objects,
            object_ref.clone(),
            1,
            None,
            update_tx,
        );
        expect_transaction_update(
            &mut ledger_service_mock,
            &nexus_objects,
            update_tx,
            execution_object_id,
            object_ref.version(),
            *object_ref.digest(),
            sui::types::ObjectIn::NotExist,
            vec![
                walk_advanced_event,
                terminal_err_eval_event,
                end_state_reached_event,
                execution_finished_event,
            ],
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;

        let result = client
            .workflow()
            .inspect_execution_until_completion(
                execution_object_id,
                InspectExecutionOptions {
                    timeout: Duration::from_secs(2),
                    poll_interval: Duration::from_millis(10),
                },
                &StorageConf::default(),
            )
            .await
            .expect("Failed to inspect execution until completion");

        assert_eq!(
            result.terminal_state,
            WorkflowExecutionTerminalState::Failed
        );
        assert!(result.execution_finished.has_any_walk_failed);
        assert!(result.execution_finished.has_any_walk_succeeded);
        assert!(matches!(
            result.events.last().map(|event| &event.data),
            Some(NexusEventKind::ExecutionFinished(_))
        ));
        assert_eq!(result.terminal_err_eval_recordings.len(), 1);
        assert_eq!(
            result.terminal_err_eval_recordings[0].failure_class,
            WorkflowFailureClass::TerminalToolFailure
        );
        assert_eq!(result.events.len(), 4);
    }

    #[tokio::test]
    async fn inspect_execution_total_timeout_covers_initial_request_retries() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        ledger_service_mock
            .expect_get_object()
            .times(1..)
            .returning(|_| Err(tonic::Status::unavailable("temporary localnet outage")));
        state_service_mock
            .expect_list_dynamic_fields()
            .times(1..)
            .returning(|_| Err(tonic::Status::unavailable("temporary localnet outage")));

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;
        let mut result = client
            .workflow()
            .inspect_execution(
                sui::types::Address::TWO,
                InspectExecutionOptions {
                    timeout: Duration::from_millis(80),
                    poll_interval: Duration::from_millis(10),
                },
            )
            .await
            .expect("inspection task should start");

        assert!(result.next_event.recv().await.is_none());
        let poller_result = result.poller.await.unwrap();
        assert!(
            matches!(poller_result, Err(NexusError::Timeout(_))),
            "unexpected inspection result: {poller_result:?}"
        );
    }

    #[test]
    fn execution_inspection_retries_only_confirmed_transient_grpc_codes() {
        for code in [
            tonic::Code::Unavailable,
            tonic::Code::DeadlineExceeded,
            tonic::Code::ResourceExhausted,
            tonic::Code::Aborted,
        ] {
            let error = NexusError::Rpc(
                anyhow::Error::new(tonic::Status::new(code, "retry"))
                    .context("inspection request failed"),
            );
            assert!(is_transient_inspection_error(&error), "{code:?}");
        }

        let permanent = NexusError::Rpc(anyhow::Error::new(tonic::Status::not_found("missing")));
        assert!(!is_transient_inspection_error(&permanent));
    }

    #[test]
    fn execution_history_includes_the_scheduler_dispatch() {
        let execution = sui::types::Address::TWO;
        let event = NexusEventKind::OccurrenceDispatched(OccurrenceDispatchedEvent {
            task_id: object_id(sui::types::Address::ZERO),
            occurrence_id: 7,
            execution_id: object_id(execution),
            dispatched_at_ms: 11,
        });

        assert_eq!(event_execution_id(&event), Some(execution));
    }

    #[test]
    fn execution_history_includes_the_agent_skill_request() {
        let execution = sui::types::Address::TWO;
        let event =
            NexusEventKind::AgentSkillExecutionRequested(AgentSkillExecutionRequestedEvent {
                execution_id: execution,
                agent_id: object_id(sui::types::Address::THREE),
                skill_id: 7,
                interface_revision: InterfaceVersion::new(1),
                payment_id: sui::types::Address::from_static("0x4"),
            });

        assert_eq!(event_execution_id(&event), Some(execution));
    }

    #[test]
    fn execution_history_includes_the_invocation_boundary() {
        let execution = sui::types::Address::TWO;
        let dag = object_id(sui::types::Address::ZERO);
        let events = [
            NexusEventKind::InvocationAuthorizationRequired(InvocationAuthorizationRequiredEvent {
                dag,
                execution: object_id(execution),
                walk_index: 0,
                vertex: RuntimeVertex::plain("tool"),
            }),
            NexusEventKind::RequestWalkExecution(RequestWalkExecutionEvent {
                dag,
                execution: object_id(execution),
                invocation: object_id(sui::types::Address::from_static("0x4")),
                invoker: sui::types::Address::THREE,
                walk_index: 0,
                next_vertex: RuntimeVertex::plain("tool"),
                evaluations: object_id(sui::types::Address::from_static("0x5")),
                agent_id: object_id(sui::types::Address::from_static("0x6")),
                skill_id: 7,
                interface_version: InterfaceVersion::new(1),
                task_id: object_id(sui::types::Address::from_static("0x7")),
                occurrence_id: 8,
            }),
            NexusEventKind::CommittedToolResult(CommittedToolResultEvent {
                dag,
                execution: object_id(execution),
                walk_index: 0,
                vertex: RuntimeVertex::plain("tool"),
                leader: object_id(sui::types::Address::from_static("0x8")),
                has_primary_failure_evidence: false,
                has_secondary_failure_evidence: false,
            }),
        ];

        for event in events {
            assert_eq!(event_execution_id(&event), Some(execution));
        }
    }

    #[test]
    fn test_event_execution_id_supports_terminal_err_eval_recorded() {
        let execution = sui::types::Address::TWO;
        let event = NexusEventKind::TerminalErrEvalRecorded(TerminalErrEvalRecordedEvent {
            dag: object_id(sui::types::Address::ZERO),
            execution: object_id(execution),
            walk_index: 2,
            vertex: RuntimeVertex::plain("failable"),
            leader: sui::types::Address::THREE,
            failure_class: WorkflowFailureClass::TerminalSubmissionFailure,
            outcome: MoveOption::from_option(Some(PostFailureAction::Terminate)),
            reason: MoveString::from("timeout"),
            err_eval_hash: vec![4, 5, 6],
            duplicate: true,
        });

        assert_eq!(event_execution_id(&event), Some(execution));
    }

    #[test]
    fn test_event_execution_id_supports_tool_verification_resolved() {
        let execution = sui::types::Address::TWO;
        let event = NexusEventKind::ToolVerificationResolved(ToolVerificationResolvedEvent {
            dag: object_id(sui::types::Address::ZERO),
            execution: object_id(execution),
            walk_index: 2,
            vertex: RuntimeVertex::plain("verified"),
            leader_cap_id: object_id(sui::types::Address::THREE),
            tool_id: object_id(sui::types::Address::from_static("0x4")),
            verifier_kind: ToolVerifierMode::External,
            verifier_witness_id: MoveOption::from_option(Some(object_id(
                sui::types::Address::from_static("0x5"),
            ))),
            decision: VerifierDecision::Accept,
        });

        assert_eq!(event_execution_id(&event), Some(execution));
    }
    #[test]
    fn test_event_execution_id_supports_object_history_events() {
        let execution = sui::types::Address::TWO;
        let events = [
            NexusEventKind::ExecutionPaymentFeesRecorded(ExecutionPaymentFeesRecordedEvent {
                payment_id: sui::types::Address::ZERO,
                execution_id: execution,
                agent_id: object_id(sui::types::Address::THREE),
                skill_id: 1,
                gas_fee_mist: 2,
                tool_fee_mist: 3,
                priority_fee_mist: 4,
                priority_fee_percentage: 5,
            }),
            NexusEventKind::ExecutionPaymentToolCostSnapshotted(
                ExecutionPaymentToolCostSnapshottedEvent {
                    payment_id: sui::types::Address::ZERO,
                    execution_id: execution,
                    agent_id: object_id(sui::types::Address::THREE),
                    tool_fqn: b"demo::tool".to_vec(),
                    cost: 6,
                },
            ),
            NexusEventKind::InvocationLocked(InvocationLockedEvent {
                execution: object_id(execution),
                vertex: RuntimeVertex::plain("tool"),
                tool: object_id(sui::types::Address::from_static("0x4")),
                tool_fqn: MoveString::from("demo::tool"),
                cashier: object_id(sui::types::Address::from_static("0x5")),
                invocation: object_id(sui::types::Address::from_static("0x6")),
                beneficiary: PaymentSourceKind::user_funded(sui::types::Address::THREE),
                policy: TypeName::new("0x7::fixed_price::Policy"),
                sources: vec![object_id(sui::types::Address::from_static("0x8"))],
                amount: 7,
            }),
            NexusEventKind::InvocationSettled(InvocationSettledEvent {
                execution: object_id(execution),
                vertex: RuntimeVertex::plain("tool"),
                tool: object_id(sui::types::Address::from_static("0x4")),
                tool_fqn: MoveString::from("demo::tool"),
                cashier: object_id(sui::types::Address::from_static("0x5")),
                invocation: object_id(sui::types::Address::from_static("0x6")),
                beneficiary: PaymentSourceKind::user_funded(sui::types::Address::THREE),
                policy: TypeName::new("0x7::fixed_price::Policy"),
                sources: vec![object_id(sui::types::Address::from_static("0x8"))],
                amount: 8,
                was_refunded: false,
            }),
            NexusEventKind::SubmissionFailureEvidenceRecorded(
                SubmissionFailureEvidenceRecordedEvent {
                    dag: object_id(sui::types::Address::ZERO),
                    execution: object_id(execution),
                    walk_index: 9,
                    vertex: RuntimeVertex::plain("submission_failure"),
                    failed_leader: sui::types::Address::THREE,
                    winning_leader: MoveOption::from_option(None),
                    reason: MoveString::from("invalid evidence"),
                    err_eval_hash: vec![10, 11, 12],
                },
            ),
            NexusEventKind::WalkPendingAbort(WalkPendingAbortEvent {
                dag: object_id(sui::types::Address::ZERO),
                execution: object_id(execution),
                walk_index: 13,
                vertex: RuntimeVertex::plain("pending_abort"),
            }),
            NexusEventKind::ExecutionPaymentRefilled(ExecutionPaymentRefilledEvent {
                execution_id: execution,
                payment_id: sui::types::Address::ZERO,
                source: sui::types::Address::THREE,
                refill_amount: 14,
            }),
        ];

        for event in events {
            assert_eq!(event_execution_id(&event), Some(execution));
        }
    }

    #[cfg(feature = "walrus")]
    #[test]
    fn test_terminal_state_from_execution_finished() {
        let success = ExecutionFinishedEvent {
            dag: object_id(sui::types::Address::ZERO),
            execution: object_id(sui::types::Address::TWO),
            has_any_walk_failed: false,
            has_any_walk_succeeded: true,
            was_aborted: false,
        };
        let failed = ExecutionFinishedEvent {
            has_any_walk_failed: true,
            has_any_walk_succeeded: false,
            ..success.clone()
        };
        let aborted = ExecutionFinishedEvent {
            has_any_walk_failed: true,
            has_any_walk_succeeded: true,
            was_aborted: true,
            ..success.clone()
        };
        let no_walk_outcome = ExecutionFinishedEvent {
            has_any_walk_failed: false,
            has_any_walk_succeeded: false,
            was_aborted: false,
            ..success.clone()
        };

        assert_eq!(
            terminal_state_from_execution_finished(&success),
            WorkflowExecutionTerminalState::Succeeded
        );
        assert_eq!(
            terminal_state_from_execution_finished(&failed),
            WorkflowExecutionTerminalState::Failed
        );
        assert_eq!(
            terminal_state_from_execution_finished(&aborted),
            WorkflowExecutionTerminalState::Aborted
        );
        assert_eq!(
            terminal_state_from_execution_finished(&no_walk_outcome),
            WorkflowExecutionTerminalState::NoWalkOutcome
        );
    }

    #[cfg(feature = "walrus")]
    #[tokio::test]
    async fn test_build_execution_completion_result_resolves_end_states() {
        let execution = sui::types::Address::TWO;
        let events = vec![
            NexusEvent {
                id: (sui::types::Digest::ZERO, 0),
                emitting_package: sui::types::Address::ZERO,
                generics: vec![],
                data: NexusEventKind::TerminalErrEvalRecorded(TerminalErrEvalRecordedEvent {
                    dag: object_id(sui::types::Address::ZERO),
                    execution: object_id(execution),
                    walk_index: 1,
                    vertex: RuntimeVertex::plain("failable"),
                    leader: sui::types::Address::THREE,
                    failure_class: WorkflowFailureClass::TerminalToolFailure,
                    outcome: MoveOption::from_option(Some(PostFailureAction::Terminate)),
                    reason: MoveString::from("tool failed"),
                    err_eval_hash: vec![1, 2, 3],
                    duplicate: false,
                }),
                distribution: None,
            },
            NexusEvent {
                id: (sui::types::Digest::ZERO, 1),
                emitting_package: sui::types::Address::ZERO,
                generics: vec![],
                data: NexusEventKind::EndStateReached(EndStateReachedEvent {
                    dag: object_id(sui::types::Address::ZERO),
                    execution: object_id(execution),
                    walk_index: 0,
                    vertex: RuntimeVertex::plain("final"),
                    variant: output_variant("ok"),
                    variant_ports_to_data: ports_data_map(vec![("answer", b"42")]),
                }),
                distribution: None,
            },
            NexusEvent {
                id: (sui::types::Digest::ZERO, 2),
                emitting_package: sui::types::Address::ZERO,
                generics: vec![],
                data: NexusEventKind::ExecutionFinished(ExecutionFinishedEvent {
                    dag: object_id(sui::types::Address::ZERO),
                    execution: object_id(execution),
                    has_any_walk_failed: true,
                    has_any_walk_succeeded: true,
                    was_aborted: false,
                }),
                distribution: None,
            },
        ];

        let result = build_execution_completion_result(events, execution, &StorageConf::default())
            .await
            .expect("summary should build");

        assert_eq!(
            result.terminal_state,
            WorkflowExecutionTerminalState::Failed
        );
        assert_eq!(result.terminal_err_eval_recordings.len(), 1);
        assert_eq!(result.end_states.len(), 1);
        assert_eq!(
            result.end_states[0].event.vertex,
            RuntimeVertex::plain("final")
        );
        assert!(matches!(
            result.end_states[0]
                .resolved_ports_to_data
                .get("answer")
                .expect("answer port")
                .values()
                .expect("resolved answer is canonical")
                .as_slice(),
            [NexusValue::InlineData { bytes }] if bytes == b"42"
        ));
    }

    #[tokio::test]
    async fn test_workflow_actions_execution_cost() {
        for detailed in [false, true] {
            let nexus_objects = sui_mocks::mock_nexus_context();
            let execution_ref = sui_mocks::mock_sui_object_ref();
            let payment_ref = sui_mocks::mock_sui_object_ref();
            let execution_id = *execution_ref.object_id();
            let invocation_id = sui::types::Address::from_static("0x123");

            let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
            let mut state_service_mock = sui_mocks::grpc::MockStateService::new();

            mock_get_dag_execution_bcs(
                &mut ledger_service_mock,
                &mut state_service_mock,
                &nexus_objects,
                execution_ref,
                &sui::types::ObjectReference::new(
                    sui::types::Address::from_static("0xd"),
                    0,
                    sui::types::Digest::ZERO,
                ),
                vec![],
            );

            mock_execution_payment_field(
                &mut ledger_service_mock,
                &mut state_service_mock,
                &nexus_objects,
                execution_id,
                payment_ref.clone(),
                ExecutionPaymentInnerV1 {
                    execution_id,
                    agent_id: crate::move_bindings::sui_framework::object::ID::new(
                        sui::types::Address::from_static("0xa"),
                    ),
                    skill_id: 11,
                    interface_revision:
                        crate::move_bindings::interface::version::InterfaceVersion::new(7),
                    payment_policy:
                        crate::move_bindings::interface::payment::SkillPaymentPolicy::UserFunded,
                    source_kind:
                        crate::move_bindings::interface::payment::PaymentSourceKind::user_funded(
                            sui::types::Address::from_static("0x1"),
                        ),
                    max_budget_mist: 100_000,
                    gas_budget_mist: 83_334,
                    priority_fee_reserve_mist: 16_666,
                    locked_budget_mist: 100_000,
                    funds: crate::move_bindings::sui_framework::balance::Balance {
                        value: 58_000,
                        phantom_t0: std::marker::PhantomData,
                    },
                    consumed: 42_000,
                    tool_fee_charged: 42_000,
                    priority_fee_charged: 0,
                    priority_fee_percentage: 20,
                    accomplished: true,
                    refunded: false,
                    final_state: ExecutionPaymentFinalState::Accomplished,
                    tool_cost_snapshot: crate::move_bindings::sui_framework::vec_map::VecMap {
                        contents: vec![],
                    },
                    locked_vertices: vec![invocation_lock(vec![1, 2, 3], invocation_id)],
                },
            );

            let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
            let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
                ledger_service_mock: Some(ledger_service_mock),
                package_service_mock: Some(package_service_mock),
                state_service_mock: Some(state_service_mock),
                ..Default::default()
            });

            let client =
                nexus_mocks::mock_nexus_client_without_coins(&nexus_objects, &rpc_url).await;

            let result = if detailed {
                let details = client
                    .workflow()
                    .execution_cost_details(execution_id)
                    .await
                    .expect("Failed to fetch execution cost details");
                assert_eq!(details.outstanding_invocation_ids, vec![invocation_id]);
                details.summary
            } else {
                client
                    .workflow()
                    .execution_cost(execution_id)
                    .await
                    .expect("Failed to fetch execution cost")
            };

            assert_eq!(result.payment_id, *payment_ref.object_id());
            assert_eq!(result.max_budget_mist, 100_000);
            assert_eq!(result.locked_budget_mist, 100_000);
            assert_eq!(result.consumed, 42_000);
            assert_eq!(result.outstanding_locks, 1);
            assert!(result.accomplished);
            assert!(!result.refunded);
        }
    }

    #[tokio::test]
    async fn test_abort_expired_execution_invocation_candidates_returns_empty_snapshot() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let execution_ref = sui_mocks::mock_sui_object_ref();
        let dag_ref = sui_mocks::mock_sui_object_ref();
        let payment_ref = sui_mocks::mock_sui_object_ref();
        let dag = dag_bcs(0);
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();

        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        mock_get_dag_execution_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            execution_ref.clone(),
            &dag_ref,
            vec![],
        );
        mock_get_dag_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            dag_ref,
            dag,
        );
        sui_mocks::grpc::mock_get_object_bcs(
            &mut ledger_service_mock,
            sui::types::ObjectReference::new(
                move_boundary::CLOCK_OBJECT_ID,
                1,
                sui::types::Digest::from([1; 32]),
            ),
            sui::types::Owner::Shared(1),
            clock_bcs(61_000),
        );
        mock_execution_payment_field(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            *execution_ref.object_id(),
            payment_ref.clone(),
            ExecutionPaymentInnerV1 {
                execution_id: *execution_ref.object_id(),
                agent_id: crate::move_bindings::sui_framework::object::ID::new(
                    sui::types::Address::from_static("0xa"),
                ),
                skill_id: 11,
                interface_revision: InterfaceVersion::new(7),
                payment_policy: SkillPaymentPolicy::UserFunded,
                source_kind:
                    crate::move_bindings::interface::payment::PaymentSourceKind::user_funded(
                        sui::types::Address::from_static("0x1"),
                    ),
                max_budget_mist: 100_000,
                gas_budget_mist: 83_334,
                priority_fee_reserve_mist: 16_666,
                locked_budget_mist: 0,
                funds: crate::move_bindings::sui_framework::balance::Balance {
                    value: 100_000,
                    phantom_t0: std::marker::PhantomData,
                },
                consumed: 0,
                tool_fee_charged: 0,
                priority_fee_charged: 0,
                priority_fee_percentage: 20,
                accomplished: false,
                refunded: false,
                final_state: ExecutionPaymentFinalState::Pending,
                tool_cost_snapshot: crate::move_bindings::sui_framework::vec_map::VecMap {
                    contents: vec![],
                },
                locked_vertices: vec![],
            },
        );

        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            ..Default::default()
        });
        let client = nexus_mocks::mock_nexus_client(&nexus_objects, &rpc_url).await;

        let candidates = client
            .workflow()
            .abort_expired_execution_invocation_candidates(*execution_ref.object_id())
            .await
            .expect("empty candidate snapshot should parse");

        assert!(candidates.is_empty());
    }

    #[tokio::test]
    async fn test_abort_expired_execution_with_invocation_submits_selected_candidate() {
        let mut rng = rand::thread_rng();
        let nexus_objects = sui_mocks::mock_nexus_context();
        let gas_coin_ref = sui_mocks::mock_sui_object_ref();
        let execution_ref = sui_mocks::mock_sui_object_ref();
        let dag_ref = sui_mocks::mock_sui_object_ref();
        let payment_ref = sui_mocks::mock_sui_object_ref();
        let tool_fqn = fqn!("xyz.taluslabs.payable@1");
        let invocation_id = sui::types::Address::from_static("0x404");
        let invocation_ref = sui_mocks::object_ref_for_id(invocation_id);
        let vertex = RuntimeVertex::plain("payable");
        let dag = dag_bcs(1);
        let vertex_key = graph_move::Vertex::new("payable");
        let vertex_field_id = dag.vertices.id().derive_dynamic_child_id(
            &crate::move_bindings::type_tag::<graph_move::Vertex>(&nexus_objects),
            &bcs::to_bytes(&vertex_key).expect("vertex key serializes"),
        );
        let vertex_field_ref = sui_mocks::object_ref_for_id(vertex_field_id);
        let execution_walks = vec![execution_move::DAGWalk::Active {
            next_vertex: vertex.clone(),
            timeout_ms: 30_000,
            requires_vertex_authorization_grant: false,
            created_at: 1_000,
        }];
        let payment_vertex_key =
            payment_vertex_key(*execution_ref.object_id(), &vertex, &tool_fqn).unwrap();
        let current_locked_vertices = vec![ExecutionPaymentVertexLock {
            vertex_key: payment_vertex_key.clone(),
            invocation_id: object_id(invocation_id),
            amount: 10,
        }];
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        let mut tx_service_mock = sui_mocks::grpc::MockTransactionExecutionService::new();
        let mut sub_service_mock = sui_mocks::grpc::MockSubscriptionService::new();

        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        mock_get_dag_execution_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            execution_ref.clone(),
            &dag_ref,
            execution_walks.clone(),
        );
        mock_get_dag_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            dag_ref.clone(),
            dag,
        );
        sui_mocks::grpc::mock_get_dynamic_table_values_bcs(
            &mut ledger_service_mock,
            vec![(
                vertex_field_ref,
                sui::types::Owner::Shared(1),
                vertex_key,
                offchain_vertex_node_bcs(&tool_fqn),
            )],
        );
        sui_mocks::grpc::mock_get_object_bcs(
            &mut ledger_service_mock,
            sui::types::ObjectReference::new(
                move_boundary::CLOCK_OBJECT_ID,
                1,
                sui::types::Digest::from([1; 32]),
            ),
            sui::types::Owner::Shared(1),
            clock_bcs(61_000),
        );
        mock_execution_payment_field(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            *execution_ref.object_id(),
            payment_ref.clone(),
            ExecutionPaymentInnerV1 {
                execution_id: *execution_ref.object_id(),
                agent_id: crate::move_bindings::sui_framework::object::ID::new(
                    sui::types::Address::from_static("0xa"),
                ),
                skill_id: 11,
                interface_revision: InterfaceVersion::new(7),
                payment_policy: SkillPaymentPolicy::UserFunded,
                source_kind:
                    crate::move_bindings::interface::payment::PaymentSourceKind::user_funded(
                        sui::types::Address::from_static("0x1"),
                    ),
                max_budget_mist: 100_000,
                gas_budget_mist: 83_334,
                priority_fee_reserve_mist: 16_666,
                locked_budget_mist: 10,
                funds: crate::move_bindings::sui_framework::balance::Balance {
                    value: 100_000,
                    phantom_t0: std::marker::PhantomData,
                },
                consumed: 0,
                tool_fee_charged: 0,
                priority_fee_charged: 0,
                priority_fee_percentage: 20,
                accomplished: false,
                refunded: false,
                final_state: ExecutionPaymentFinalState::Pending,
                tool_cost_snapshot: crate::move_bindings::sui_framework::vec_map::VecMap {
                    contents: vec![],
                },
                locked_vertices: current_locked_vertices,
            },
        );
        sui_mocks::grpc::mock_get_object_metadata_exact(
            &mut ledger_service_mock,
            invocation_ref.clone(),
            sui::types::Owner::Object(*execution_ref.object_id()),
            None,
        );
        sui_mocks::grpc::mock_runtime_authority(&mut ledger_service_mock, &nexus_objects, false);
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        let submitted = sui_mocks::grpc::mock_execute_transaction_and_wait_for_checkpoint(
            &mut tx_service_mock,
            &mut sub_service_mock,
            &mut ledger_service_mock,
            gas_coin_ref.clone(),
            vec![],
            vec![],
            vec![],
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            execution_service_mock: Some(tx_service_mock),
            subscription_service_mock: Some(sub_service_mock),
            ..Default::default()
        });
        let pk = sui::crypto::Ed25519PrivateKey::generate(&mut rng);
        let client = NexusClient::builder()
            .with_private_key(pk)
            .with_rpc_url(&rpc_url)
            .with_nexus_objects(nexus_objects.objects().clone())
            .with_gas(vec![gas_coin_ref], 1000)
            .build()
            .await
            .expect("mock client should build");

        let result = client
            .workflow()
            .abort_expired_execution_with_invocation(
                *execution_ref.object_id(),
                Some(invocation_id),
            )
            .await
            .expect("abort transaction should submit");

        assert_eq!(result.tx_digest, submitted.digest());
        assert_eq!(result.tx_checkpoint, 1);
        assert_eq!(result.dag_id, *dag_ref.object_id());
        assert_eq!(result.dag_execution_id, *execution_ref.object_id());
        assert_eq!(result.selected_candidate.invocation_ref, invocation_ref);
        assert_eq!(result.selected_candidate.vertex, vertex);
        assert_eq!(
            result.selected_candidate.payment_vertex_key,
            payment_vertex_key
        );
    }

    async fn mock_client_for_workflow_submit(
        nexus_objects: &crate::types::NexusObjects,
        gas_coin_ref: sui::types::ObjectReference,
        ledger_service_mock: sui_mocks::grpc::MockLedgerService,
        package_service_mock: sui_mocks::grpc::MockMovePackageService,
        state_service_mock: sui_mocks::grpc::MockStateService,
        tx_service_mock: sui_mocks::grpc::MockTransactionExecutionService,
        sub_service_mock: sui_mocks::grpc::MockSubscriptionService,
    ) -> NexusClient {
        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            package_service_mock: Some(package_service_mock),
            state_service_mock: Some(state_service_mock),
            execution_service_mock: Some(tx_service_mock),
            subscription_service_mock: Some(sub_service_mock),
            ..Default::default()
        });
        let pk = sui::crypto::Ed25519PrivateKey::generate(rand::thread_rng());
        NexusClient::builder()
            .with_private_key(pk)
            .with_rpc_url(&rpc_url)
            .with_nexus_objects(nexus_objects.clone())
            .with_gas(vec![gas_coin_ref], 1000)
            .build()
            .await
            .expect("mock client should build")
    }

    #[tokio::test]
    async fn test_abort_expired_execution_submits_workflow_transaction() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let gas_coin_ref = sui_mocks::mock_sui_object_ref();
        let dag_ref = sui_mocks::mock_sui_object_ref();
        let execution_ref = sui_mocks::mock_sui_object_ref();
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        let mut tx_service_mock = sui_mocks::grpc::MockTransactionExecutionService::new();
        let mut sub_service_mock = sui_mocks::grpc::MockSubscriptionService::new();

        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        mock_get_dag_execution_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            execution_ref.clone(),
            &dag_ref,
            vec![],
        );
        mock_get_dag_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            dag_ref.clone(),
            dag_bcs(0),
        );
        sui_mocks::grpc::mock_get_object_bcs(
            &mut ledger_service_mock,
            sui::types::ObjectReference::new(
                move_boundary::CLOCK_OBJECT_ID,
                1,
                sui::types::Digest::from([1; 32]),
            ),
            sui::types::Owner::Shared(1),
            clock_bcs(61_000),
        );
        sui_mocks::grpc::mock_runtime_authority(&mut ledger_service_mock, &nexus_objects, false);
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        let submitted = sui_mocks::grpc::mock_execute_transaction_and_wait_for_checkpoint(
            &mut tx_service_mock,
            &mut sub_service_mock,
            &mut ledger_service_mock,
            gas_coin_ref.clone(),
            vec![],
            vec![],
            vec![],
        );
        let client = mock_client_for_workflow_submit(
            &nexus_objects,
            gas_coin_ref,
            ledger_service_mock,
            package_service_mock,
            state_service_mock,
            tx_service_mock,
            sub_service_mock,
        )
        .await;

        let result = client
            .workflow()
            .abort_expired_execution(*execution_ref.object_id())
            .await
            .expect("abort transaction should submit");

        assert_eq!(result.tx_digest, submitted.digest());
        assert_eq!(result.tx_checkpoint, 1);
        assert_eq!(result.dag_id, *dag_ref.object_id());
        assert_eq!(result.dag_execution_id, *execution_ref.object_id());
        assert!(result.cleaned_broken_onchain_results.is_empty());
    }

    #[tokio::test]
    async fn test_settle_committed_tool_result_for_walk_submits_workflow_transaction() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let gas_coin_ref = sui_mocks::mock_sui_object_ref();
        let dag_ref = sui_mocks::mock_sui_object_ref();
        let execution_ref = sui_mocks::mock_sui_object_ref();
        let vertex = RuntimeVertex::plain("tool");
        let tool_fqn = fqn!("xyz.taluslabs.fixture@1");
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();
        let mut state_service_mock = sui_mocks::grpc::MockStateService::new();
        let mut tx_service_mock = sui_mocks::grpc::MockTransactionExecutionService::new();
        let mut sub_service_mock = sui_mocks::grpc::MockSubscriptionService::new();

        sui_mocks::grpc::mock_reference_gas_price(&mut ledger_service_mock, 1000);
        mock_get_dag_execution_bcs(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            execution_ref.clone(),
            &dag_ref,
            pending_settlement_walks(7, vertex.clone()),
        );
        mock_exact_invocation_lock(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
            &execution_ref,
            &dag_ref,
            &vertex,
            &tool_fqn,
        );
        mock_tool_registry_observation(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
        );
        mock_leader_registry_observation(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
        );
        mock_priority_fee_vault_observation(
            &mut ledger_service_mock,
            &mut state_service_mock,
            &nexus_objects,
        );
        sui_mocks::grpc::mock_runtime_authority(&mut ledger_service_mock, &nexus_objects, false);
        let package_service_mock = mock_package_graph(&mut ledger_service_mock, &nexus_objects);
        let submitted = sui_mocks::grpc::mock_execute_transaction_and_wait_for_checkpoint(
            &mut tx_service_mock,
            &mut sub_service_mock,
            &mut ledger_service_mock,
            gas_coin_ref.clone(),
            vec![],
            vec![],
            vec![],
        );
        let client = mock_client_for_workflow_submit(
            &nexus_objects,
            gas_coin_ref,
            ledger_service_mock,
            package_service_mock,
            state_service_mock,
            tx_service_mock,
            sub_service_mock,
        )
        .await;

        let result = client
            .workflow()
            .settle_committed_tool_result_for_walk(SettleCommittedToolResultParams {
                dag_execution_id: *execution_ref.object_id(),
                walk_index: 7,
            })
            .await
            .expect("settlement transaction should submit");

        assert_eq!(result.tx_digest, submitted.digest());
        assert_eq!(result.tx_checkpoint, 1);
        assert_eq!(result.dag_id, *dag_ref.object_id());
        assert_eq!(result.dag_execution_id, *execution_ref.object_id());
        assert_eq!(result.walk_index, 7);
    }

    #[tokio::test]
    async fn test_leader_settlement_and_record_gas_paths_submit_transactions() {
        let nexus_objects = sui_mocks::mock_nexus_context();
        let settle_gas_ref = sui_mocks::mock_sui_object_ref();
        let record_gas_ref = sui_mocks::mock_sui_object_ref();
        let dag_ref = sui_mocks::mock_sui_object_ref();
        let execution_ref = sui_mocks::mock_sui_object_ref();
        let leader_cap_ref = sui_mocks::mock_sui_object_ref();
        let vertex = RuntimeVertex::plain("tool");
        let tool_fqn = fqn!("xyz.taluslabs.fixture@1");

        let mut settle_ledger = sui_mocks::grpc::MockLedgerService::new();
        let mut settle_state = sui_mocks::grpc::MockStateService::new();
        let mut settle_tx = sui_mocks::grpc::MockTransactionExecutionService::new();
        let mut settle_sub = sui_mocks::grpc::MockSubscriptionService::new();
        sui_mocks::grpc::mock_reference_gas_price(&mut settle_ledger, 1000);
        mock_get_dag_execution_bcs(
            &mut settle_ledger,
            &mut settle_state,
            &nexus_objects,
            execution_ref.clone(),
            &dag_ref,
            pending_settlement_walks(8, vertex.clone()),
        );
        mock_exact_invocation_lock(
            &mut settle_ledger,
            &mut settle_state,
            &nexus_objects,
            &execution_ref,
            &dag_ref,
            &vertex,
            &tool_fqn,
        );
        mock_tool_registry_observation(&mut settle_ledger, &mut settle_state, &nexus_objects);
        mock_leader_registry_observation(&mut settle_ledger, &mut settle_state, &nexus_objects);
        mock_priority_fee_vault_observation(&mut settle_ledger, &mut settle_state, &nexus_objects);
        sui_mocks::grpc::mock_runtime_authority(&mut settle_ledger, &nexus_objects, false);
        let settle_packages = mock_package_graph(&mut settle_ledger, &nexus_objects);
        sui_mocks::grpc::mock_get_object_metadata(
            &mut settle_ledger,
            leader_cap_ref.clone(),
            sui::types::Owner::Immutable,
            None,
        );
        let settle_submitted = sui_mocks::grpc::mock_execute_transaction_and_wait_for_checkpoint(
            &mut settle_tx,
            &mut settle_sub,
            &mut settle_ledger,
            settle_gas_ref.clone(),
            vec![],
            vec![],
            vec![],
        );
        let settle_client = mock_client_for_workflow_submit(
            &nexus_objects,
            settle_gas_ref,
            settle_ledger,
            settle_packages,
            settle_state,
            settle_tx,
            settle_sub,
        )
        .await;
        let settle_result = settle_client
            .workflow()
            .settle_committed_tool_result_for_walk_by_leader(
                SettleCommittedToolResultByLeaderParams {
                    dag_execution_id: *execution_ref.object_id(),
                    leader_cap_id: *leader_cap_ref.object_id(),
                    walk_index: 8,
                    expected_vertex: None,
                    failed_onchain_tool_reason: None,
                    commit_tx_digest: vec![1, 2, 3],
                    commit_gas_charge: 11,
                    settlement_gas_charge: 13,
                },
            )
            .await
            .expect("leader settlement transaction should submit");

        assert_eq!(settle_result.tx_digest, settle_submitted.digest());
        assert_eq!(settle_result.tx_checkpoint, 1);
        assert_eq!(settle_result.dag_id, *dag_ref.object_id());
        assert_eq!(settle_result.walk_index, 8);

        let mut record_ledger = sui_mocks::grpc::MockLedgerService::new();
        let mut record_state = sui_mocks::grpc::MockStateService::new();
        let mut record_tx = sui_mocks::grpc::MockTransactionExecutionService::new();
        let mut record_sub = sui_mocks::grpc::MockSubscriptionService::new();
        sui_mocks::grpc::mock_reference_gas_price(&mut record_ledger, 1000);
        mock_get_dag_execution_bcs(
            &mut record_ledger,
            &mut record_state,
            &nexus_objects,
            execution_ref.clone(),
            &dag_ref,
            vec![],
        );
        mock_get_dag_bcs(
            &mut record_ledger,
            &mut record_state,
            &nexus_objects,
            dag_ref.clone(),
            dag_bcs(0),
        );
        mock_leader_registry_observation(&mut record_ledger, &mut record_state, &nexus_objects);
        sui_mocks::grpc::mock_runtime_authority(&mut record_ledger, &nexus_objects, false);
        let record_packages = mock_package_graph(&mut record_ledger, &nexus_objects);
        sui_mocks::grpc::mock_get_object_metadata(
            &mut record_ledger,
            leader_cap_ref.clone(),
            sui::types::Owner::Address(sui::types::Address::from_static("0xabc")),
            None,
        );
        let record_submitted = sui_mocks::grpc::mock_execute_transaction_and_wait_for_checkpoint(
            &mut record_tx,
            &mut record_sub,
            &mut record_ledger,
            record_gas_ref.clone(),
            vec![],
            vec![],
            vec![],
        );
        let record_client = mock_client_for_workflow_submit(
            &nexus_objects,
            record_gas_ref,
            record_ledger,
            record_packages,
            record_state,
            record_tx,
            record_sub,
        )
        .await;
        let record_result = record_client
            .workflow()
            .record_committed_tool_result_gas_charge_by_leader(
                RecordCommittedToolResultGasChargeParams {
                    dag_execution_id: *execution_ref.object_id(),
                    leader_cap_id: *leader_cap_ref.object_id(),
                    walk_index: 9,
                    expected_vertex: None,
                    failed_onchain_tool_reason: None,
                    commit_tx_digest: vec![4, 5, 6],
                    commit_gas_charge: 17,
                    settlement_gas_charge: 19,
                },
            )
            .await
            .expect("leader gas record transaction should submit");

        assert_eq!(record_result.tx_digest, record_submitted.digest());
        assert_eq!(record_result.tx_checkpoint, 1);
        assert_eq!(record_result.dag_execution_id, *execution_ref.object_id());
        assert_eq!(record_result.leader_cap_id, *leader_cap_ref.object_id());
        assert_eq!(record_result.walk_index, 9);
    }

    #[test]
    fn dag_vertex_requires_tool_verification_reads_vertex_mode() {
        let mut vertex = offchain_vertex_info(&fqn!("xyz.example.tool@1"));
        assert!(!dag_vertex_requires_tool_verification(&vertex));
        vertex.verifier_mode = ToolVerifierMode::External;
        assert!(dag_vertex_requires_tool_verification(&vertex));
    }

    #[test]
    fn execution_terminal_record_matches_retryable_vertex_uses_typed_records() {
        let vertex = RuntimeVertex::with_iterator("v1", 2, 5);

        fn record(
            vertex: RuntimeVertex,
            failure_class: WorkflowFailureClass,
        ) -> execution_move::TerminalErrEvalRecord {
            execution_move::TerminalErrEvalRecord {
                walk_index: 9,
                vertex,
                leader: sui::types::Address::THREE,
                failure_class,
                outcome: MoveOption::from_option(None::<PostFailureAction>),
                reason: MoveString::from("retryable failure"),
                variant_ports_to_data: VecMap { contents: vec![] },
                err_eval_hash: vec![1, 2, 3],
            }
        }

        let retryable_records = VecMap {
            contents: vec![crate::move_bindings::sui_framework::vec_map::Entry {
                key: 9,
                value: record(vertex.clone(), WorkflowFailureClass::Retryable),
            }],
        };
        let terminal_records = VecMap {
            contents: vec![crate::move_bindings::sui_framework::vec_map::Entry {
                key: 9,
                value: record(
                    RuntimeVertex::plain("terminal_vertex"),
                    WorkflowFailureClass::TerminalSubmissionFailure,
                ),
            }],
        };

        assert!(execution_terminal_record_matches_retryable_vertex(
            &retryable_records,
            9,
            &vertex,
        ));
        assert!(!execution_terminal_record_matches_retryable_vertex(
            &retryable_records,
            10,
            &vertex,
        ));
        assert!(!execution_terminal_record_matches_retryable_vertex(
            &terminal_records,
            9,
            &RuntimeVertex::plain("terminal_vertex"),
        ));
    }

    #[test]
    fn fetch_vertex_input_port_names_reads_declared_ports_from_typed_vertex() {
        let mut vertex = offchain_vertex_info(&fqn!("xyz.example.tool@1"));
        vertex.input_ports = VecSet {
            contents: vec![
                graph_move::InputPort::new("z_port"),
                graph_move::InputPort::new("a_port"),
            ],
        };

        assert_eq!(
            vertex.declared_input_port_names(),
            vec!["a_port".to_string(), "z_port".to_string()]
        );
    }

    #[test]
    fn invocation_abort_filter_returns_exact_expired_locked_vertex() {
        let execution_id = sui::types::Address::from_static("0xabc");
        let tool_fqn = fqn!("xyz.taluslabs.payable@1");
        let other_tool_fqn = fqn!("xyz.taluslabs.other@1");
        let invocation_id = sui::types::Address::from_static("0x704");
        let payable_vertex = RuntimeVertex::plain("payable");
        let idle_vertex = RuntimeVertex::plain("idle");
        let matching_key = payment_vertex_key(execution_id, &payable_vertex, &tool_fqn).unwrap();
        let mut vertices = HashMap::new();
        vertices.insert(
            graph_move::Vertex::new("payable"),
            offchain_vertex_info(&tool_fqn),
        );
        vertices.insert(
            graph_move::Vertex::new("idle"),
            offchain_vertex_info(&other_tool_fqn),
        );
        let walks = vec![
            DAGWalk::Active {
                next_vertex: payable_vertex.clone(),
                timeout_ms: 30_000,
                requires_vertex_authorization_grant: false,
                created_at: 1_000,
            },
            DAGWalk::Active {
                next_vertex: idle_vertex,
                timeout_ms: 30_000,
                requires_vertex_authorization_grant: false,
                created_at: 61_000,
            },
            DAGWalk::PendingAbort {
                at_vertex: RuntimeVertex::plain("already_pending"),
            },
        ];
        let locks = vec![invocation_lock(matching_key.clone(), invocation_id)];

        let candidates = filter_invocation_abort_candidate_walks(
            execution_id,
            &vertices,
            &walks,
            &locks,
            61_000,
        )
        .unwrap();

        assert_eq!(candidates.len(), 1);
        assert_eq!(candidates[0].walk_index, 0);
        assert_eq!(candidates[0].vertex, payable_vertex);
        assert_eq!(candidates[0].payment_vertex_key, matching_key);
        assert_eq!(candidates[0].invocation_id, invocation_id);
    }

    #[test]
    fn invocation_abort_filter_ignores_nonmatching_payment_locks() {
        let execution_id = sui::types::Address::from_static("0xabc");
        let tool_fqn = fqn!("xyz.taluslabs.payable@1");
        let other_tool_fqn = fqn!("xyz.taluslabs.other@1");
        let payable_vertex = RuntimeVertex::plain("payable");
        let other_tool_key =
            payment_vertex_key(execution_id, &payable_vertex, &other_tool_fqn).unwrap();
        let mut vertices = HashMap::new();
        vertices.insert(
            graph_move::Vertex::new("payable"),
            offchain_vertex_info(&tool_fqn),
        );
        let walks = vec![DAGWalk::Active {
            next_vertex: payable_vertex,
            timeout_ms: 30_000,
            requires_vertex_authorization_grant: false,
            created_at: 1_000,
        }];
        let locks = vec![
            invocation_lock(other_tool_key, sui::types::Address::from_static("0x705")),
            invocation_lock(vec![1, 2, 3], sui::types::Address::from_static("0x706")),
        ];

        let candidates = filter_invocation_abort_candidate_walks(
            execution_id,
            &vertices,
            &walks,
            &locks,
            61_000,
        )
        .unwrap();

        assert!(candidates.is_empty());
    }

    #[test]
    fn invocation_abort_filter_errors_when_expired_vertex_is_missing_from_dag() {
        let execution_id = sui::types::Address::from_static("0xabc");
        let walks = vec![DAGWalk::Active {
            next_vertex: RuntimeVertex::plain("missing"),
            timeout_ms: 30_000,
            requires_vertex_authorization_grant: false,
            created_at: 1_000,
        }];

        let error = filter_invocation_abort_candidate_walks(
            execution_id,
            &HashMap::new(),
            &walks,
            &[],
            61_000,
        )
        .err()
        .expect("expired walk should require a fetched DAG vertex");

        assert!(error
            .to_string()
            .contains("DAG vertex 'missing' missing from fetched DAG"));
    }

    #[tokio::test]
    async fn fetch_invocation_abort_candidates_resolves_exact_invocation_refs() {
        let invocation_id = sui::types::Address::from_static("0x707");
        let invocation_ref = sui_mocks::object_ref_for_id(invocation_id);
        let candidate = InvocationAbortCandidateSeed {
            walk_index: 2,
            vertex: RuntimeVertex::plain("payable"),
            payment_vertex_key: vec![1, 2, 3],
            invocation_id,
        };
        let mut ledger_service_mock = sui_mocks::grpc::MockLedgerService::new();

        sui_mocks::grpc::mock_get_object_metadata(
            &mut ledger_service_mock,
            invocation_ref.clone(),
            sui::types::Owner::Object(sui::types::Address::from_static("0x708")),
            None,
        );

        let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
            ledger_service_mock: Some(ledger_service_mock),
            ..Default::default()
        });
        let client = sui::grpc::client(rpc_url).expect("mock client");
        let crawler = Crawler::new(Arc::new(client));
        let candidates = fetch_invocation_abort_candidates(&crawler, vec![candidate])
            .await
            .expect("candidate refs should be fetched");

        assert_eq!(candidates.len(), 1);
        assert_eq!(candidates[0].walk_index, 2);
        assert_eq!(candidates[0].vertex, RuntimeVertex::plain("payable"));
        assert_eq!(candidates[0].payment_vertex_key, vec![1, 2, 3]);
        assert_eq!(candidates[0].invocation_ref, invocation_ref);
    }

    #[test]
    fn select_invocation_abort_candidate_uses_first_or_required_invocation() {
        let first_id = sui::types::Address::from_static("0x111");
        let second_id = sui::types::Address::from_static("0x222");
        let candidates = vec![
            InvocationAbortCandidate {
                walk_index: 0,
                vertex: RuntimeVertex::plain("first"),
                payment_vertex_key: vec![1],
                invocation_ref: sui::types::ObjectReference::new(
                    first_id,
                    1,
                    sui::types::Digest::default(),
                ),
            },
            InvocationAbortCandidate {
                walk_index: 1,
                vertex: RuntimeVertex::plain("second"),
                payment_vertex_key: vec![2],
                invocation_ref: sui::types::ObjectReference::new(
                    second_id,
                    1,
                    sui::types::Digest::default(),
                ),
            },
        ];

        let selected = select_invocation_abort_candidate(candidates.clone(), None).unwrap();
        assert_eq!(*selected.invocation_ref.object_id(), first_id);

        let selected =
            select_invocation_abort_candidate(candidates.clone(), Some(second_id)).unwrap();
        assert_eq!(*selected.invocation_ref.object_id(), second_id);

        let missing = sui::types::Address::from_static("0x333");
        let error = select_invocation_abort_candidate(candidates, Some(missing)).unwrap_err();
        assert!(error
            .to_string()
            .contains("is not currently eligible to abort this execution"));

        let error = select_invocation_abort_candidate(Vec::new(), None).unwrap_err();
        assert!(error
            .to_string()
            .contains("No Invocation abort candidates are currently eligible"));
    }
}