langcontinuation 0.1.0

#![doc = include_str!("../README.md")]
#![deny(missing_docs)]

use std::{
    collections::BTreeMap,
    collections::{HashMap, HashSet},
    fmt::{Display, Formatter},
    future::Future,
    io::Write,
    pin::Pin,
    time::Instant,
};

use claudius::{
    AccumulatingStream, Anthropic, ContentBlockDelta, Error as AnthropicError, Message,
    MessageCreateParams, MessageStreamEvent, ToolResultBlock, ToolUnionParam, ToolUseBlock,
};
use futures::StreamExt;
use setsum::Setsum;
use uuid::Uuid;

/// Re-export of the indicio crate used for workflow and executor observation.
pub use indicio;

/// Execute workflows through built-in live provider integrations.
pub mod live;

/// Execute workflows through a Postgres-backed Anthropic batch scheduler.
#[cfg(feature = "batch")]
pub mod batch;

/// Indicio clue collector for workflow and executor observation.
///
/// Live and batch executors install a stderr emitter at info verbosity when no
/// emitter is active. Applications can replace that emitter or adjust verbosity
/// through this collector.
pub static COLLECTOR: indicio::Collector = indicio::Collector::new();

const LLM_OUTPUT_BLUE: &str = "\x1b[34m";
const ANSI_RESET: &str = "\x1b[0m";

/// Print LLM-generated text in blue.
///
/// This function is public only so examples can use the same output convention
/// as the live client path.
#[doc(hidden)]
pub fn __print_llm_output(text: &str) {
    print_llm_output_chunk(text);
    println!();
}

fn print_llm_output_chunk(text: &str) {
    print!("{LLM_OUTPUT_BLUE}{text}{ANSI_RESET}");
    let _ = std::io::stdout().flush();
}

pub(crate) fn log_indicio_clue(level: u64, clue: indicio::Value) {
    COLLECTOR.emit(concat!(module_path!(), " ", file!()), line!(), level, clue);
}

pub(crate) fn wire_executor_indicio_stderr() {
    if !COLLECTOR.is_logging() {
        COLLECTOR.register(indicio::StdioEmitter);
    }
    if COLLECTOR.verbosity() < indicio::INFO {
        COLLECTOR.set_verbosity(indicio::INFO);
    }
}

#[cfg(feature = "batch")]
pub(crate) fn log_json_clue(level: u64, clue: serde_json::Value) {
    log_indicio_clue(level, serde_json_to_indicio_value(clue));
}

pub(crate) fn log_executor_transition(executor: &str, transition: &str, fields: indicio::Value) {
    log_indicio_clue(
        indicio::INFO,
        indicio::value!({
            log_type: format!("langcontinuation.{executor}.executor_transition"),
            transition: transition,
            fields: fields,
        }),
    );
}

pub(crate) fn optional_indicio_string(value: Option<&str>) -> indicio::Value {
    value
        .map(indicio::Value::from)
        .unwrap_or_else(|| indicio::value!({ null: true }))
}

#[cfg(feature = "batch")]
fn serde_json_to_indicio_value(value: serde_json::Value) -> indicio::Value {
    match value {
        serde_json::Value::Null => indicio::value!({ null: true }),
        serde_json::Value::Bool(value) => indicio::Value::from(value),
        serde_json::Value::Number(value) => {
            if let Some(value) = value.as_u64() {
                indicio::Value::from(value)
            } else if let Some(value) = value.as_i64() {
                indicio::Value::from(value)
            } else if let Some(value) = value.as_f64() {
                indicio::Value::from(value)
            } else {
                indicio::Value::from(value.to_string())
            }
        }
        serde_json::Value::String(value) => indicio::Value::from(value),
        serde_json::Value::Array(values) => indicio::Value::Array(
            values
                .into_iter()
                .map(serde_json_to_indicio_value)
                .collect::<Vec<_>>()
                .into(),
        ),
        serde_json::Value::Object(values) => indicio::Value::Object(
            values
                .into_iter()
                .map(|(key, value)| (key, serde_json_to_indicio_value(value)))
                .collect(),
        ),
    }
}

/// Grants one workflow function the authority to choose the next continuation.
///
/// A `Continuation` is intentionally linear: each generated workflow function
/// receives one value and consumes it to produce a [`ContinuationChoice`]. This
/// keeps the transition from the current local computation to the next durable
/// workflow step explicit.
pub struct Continuation {
    _phantom: std::marker::PhantomData<()>,
}

impl Continuation {
    /// Continue with the already-scheduled continuation stack.
    ///
    /// `goto` does not add a new step. It says that the current function has
    /// finished and that execution should advance to whatever the workflow
    /// already had waiting.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::{
    ///     Continuation, ContinuationChoice, Trampoline, Workflow, generate_goto,
    /// };
    ///
    /// generate_goto! {
    ///     fn entry(
    ///         workflow: &mut Workflow,
    ///         input: String,
    ///         continuation: Continuation
    ///     ) -> Result<ContinuationChoice, handled::SError> {
    ///         let _ = input;
    ///         let _ = workflow.run_id();
    ///         Ok(continuation.goto())
    ///     }
    /// }
    ///
    /// fn main() {
    ///     let mut trampoline = Trampoline::default();
    ///     trampoline.register("entry", entry);
    /// }
    /// ```
    pub fn goto(self) -> ContinuationChoice {
        ContinuationChoice {
            steps: vec![],
            halt: false,
        }
    }

    /// End the workflow by clearing all deferred continuation steps.
    ///
    /// `halt` is stronger than reaching the end of the stack naturally. It
    /// discards any pending work and makes the workflow's next result a
    /// [`WorkflowResult::Halt`].
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::{
    ///     Continuation, ContinuationChoice, Trampoline, Workflow, WorkflowResult,
    ///     generate_goto,
    /// };
    ///
    /// generate_goto! {
    ///     fn entry(
    ///         workflow: &mut Workflow,
    ///         input: String,
    ///         continuation: Continuation
    ///     ) -> Result<ContinuationChoice, handled::SError> {
    ///         let _ = (workflow.run_id(), input);
    ///         Ok(continuation.halt())
    ///     }
    /// }
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), handled::SError> {
    ///     let mut workflow = Workflow::new("run", "entry");
    ///     workflow.into_env("input: String", "stop".to_string()).unwrap();
    ///
    ///     let mut trampoline = Trampoline::default();
    ///     trampoline.register("entry", entry);
    ///
    ///     match trampoline.run(workflow).await?.result {
    ///         WorkflowResult::Halt { workflow } => assert_eq!(workflow.run_id(), "run"),
    ///         other => panic!("unexpected suspension: {other:?}"),
    ///     }
    ///     Ok(())
    /// }
    /// ```
    pub fn halt(self) -> ContinuationChoice {
        ContinuationChoice {
            steps: Vec::new(),
            halt: true,
        }
    }

    /// Schedule another registered workflow function by name.
    ///
    /// The name is matched exactly against a later [`Trampoline::register`]
    /// call. The scheduled function runs after the current function returns.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::{
    ///     Continuation, ContinuationChoice, Trampoline, Workflow, WorkflowResult,
    ///     generate_goto,
    /// };
    ///
    /// generate_goto! {
    ///     fn entry(
    ///         workflow: &mut Workflow,
    ///         input: String,
    ///         continuation: Continuation
    ///     ) -> Result<ContinuationChoice, handled::SError> {
    ///         let _ = (workflow.run_id(), input);
    ///         Ok(continuation.call("finish"))
    ///     }
    /// }
    ///
    /// generate_goto! {
    ///     fn finish(
    ///         workflow: &mut Workflow,
    ///         input: String,
    ///         continuation: Continuation
    ///     ) -> Result<ContinuationChoice, handled::SError> {
    ///         workflow.into_env("finished", true).unwrap();
    ///         let _ = input;
    ///         Ok(continuation.halt())
    ///     }
    /// }
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), handled::SError> {
    ///     let mut workflow = Workflow::new("run", "entry");
    ///     workflow.into_env("input: String", "go".to_string()).unwrap();
    ///
    ///     let mut trampoline = Trampoline::default();
    ///     trampoline.register("entry", entry);
    ///     trampoline.register("finish", finish);
    ///
    ///     let WorkflowResult::Halt { workflow } = trampoline.run(workflow).await?.result else {
    ///         panic!("workflow should halt");
    ///     };
    ///     assert_eq!(workflow.from_env::<bool>("finished").unwrap(), Some(true));
    ///     Ok(())
    /// }
    /// ```
    pub fn call(self, function: impl Into<String>) -> ContinuationChoice {
        let function = function.into();

        ContinuationChoice {
            steps: vec![Step::Call {
                function: function.clone(),
            }],
            halt: false,
        }
    }

    /// Suspend for an Anthropic request and then call the named receiver.
    ///
    /// The request is recorded as durable workflow state. A runtime can send it
    /// immediately, as [`live::Executor`] does, or persist the suspended
    /// [`Workflow`] and resume it later with [`Trampoline::resume_anthropic`].
    /// The returned Anthropic message is stored in the environment under
    /// `output_key` before `next_function` runs.
    ///
    /// # Errors
    ///
    /// This method only constructs a continuation choice and does not fail.
    /// Execution can later fail if the provider is not registered, if the
    /// provider request fails, or if the suspended workflow is resumed with the
    /// wrong output key.
    pub fn anthropic(
        self,
        provider: impl Into<String>,
        message: MessageCreateParams,
        output_key: impl Into<String>,
        next_function: impl Into<String>,
    ) -> ContinuationChoice {
        let provider = provider.into();
        let output_key = output_key.into();
        let next_function = next_function.into();

        ContinuationChoice {
            steps: vec![
                Step::Anthropic {
                    provider: provider.clone(),
                    message: Box::new(message),
                    output_key: output_key.clone(),
                },
                Step::Call {
                    function: next_function.clone(),
                },
            ],
            halt: false,
        }
    }

    /// Suspend for human input and then call the named receiver.
    ///
    /// The request is recorded as durable workflow state. A scheduler can
    /// persist the suspended [`Workflow`], show the [`HumanRequest`] to an
    /// operator, and resume it later with [`Trampoline::resume_human`]. The
    /// submitted answer is stored in the environment under `output_key` before
    /// `next_function` runs.
    ///
    /// # Errors
    ///
    /// This method only constructs a continuation choice and does not fail.
    /// Execution can later fail if the suspended workflow is resumed with the
    /// wrong output key or with an answer that cannot be serialized.
    pub fn human(
        self,
        request: HumanRequest,
        output_key: impl Into<String>,
        next_function: impl Into<String>,
    ) -> ContinuationChoice {
        let output_key = output_key.into();
        let next_function = next_function.into();

        ContinuationChoice {
            steps: vec![
                Step::Human {
                    request,
                    output_key: output_key.clone(),
                },
                Step::Call {
                    function: next_function.clone(),
                },
            ],
            halt: false,
        }
    }

    /// Suspend to execute client-side tool calls and then call the named receiver.
    ///
    /// This is the durable form of an agent tool loop. The Anthropic receiver
    /// inspects a model response, and when the response contains
    /// [`ToolUseBlock`] entries, returns this continuation with those blocks.
    /// The runtime looks each tool up by name in its registry (see
    /// [`Trampoline::register_tool`]), runs it, and resumes the workflow with
    /// the resulting [`ToolResultBlock`] values stored under `output_key`. The
    /// `next_function` then owns the conversation history: it appends the
    /// assistant message and a user tool-result message and issues the next
    /// [`Continuation::anthropic`] call, looping until the model stops calling
    /// tools.
    ///
    /// The crate does not own the transcript. `output_key` receives a
    /// `Vec<ToolResultBlock>`; what the receiver does with it is application
    /// policy.
    ///
    /// Prefer the helper [`dispatch_tool_uses`] over calling this directly: it
    /// raises this suspension only when the response actually calls tools and
    /// otherwise falls through to a terminal continuation.
    ///
    /// # Errors
    ///
    /// This method only constructs a continuation choice and does not fail.
    /// Execution can later fail if a named tool is not registered or if the
    /// suspended workflow is resumed with the wrong output key.
    pub fn tool_call(
        self,
        tool_uses: Vec<ToolUseBlock>,
        output_key: impl Into<String>,
        next_function: impl Into<String>,
    ) -> ContinuationChoice {
        let output_key = output_key.into();
        let next_function = next_function.into();

        ContinuationChoice {
            steps: vec![
                Step::ToolCall {
                    tool_uses,
                    output_key: output_key.clone(),
                },
                Step::Call {
                    function: next_function.clone(),
                },
            ],
            halt: false,
        }
    }

    /// Split execution into two independent workflows and rejoin through a call.
    ///
    /// Both branches inherit the parent environment and start with empty
    /// continuation stacks. A runtime resumes the parent with
    /// [`Trampoline::resume_fork_join`] after both branches halt. The merge
    /// accepts one-sided changes and identical writes, but rejects conflicting
    /// writes to the same environment key.
    ///
    /// # Errors
    ///
    /// This method only constructs a continuation choice and does not fail.
    /// Resuming the join can fail if either branch has not halted or if the
    /// branch environments contain conflicting writes.
    pub fn fork_join(
        self,
        lhs: ForkBranch,
        rhs: ForkBranch,
        function: impl Into<String>,
    ) -> ContinuationChoice {
        let function = function.into();

        ContinuationChoice {
            steps: vec![Step::ForkJoin {
                lhs,
                rhs,
                function: function.clone(),
            }],
            halt: false,
        }
    }
}

/// Names one branch of a fork/join split.
///
/// A branch is a run id and the registered function where that branch begins.
/// The run id belongs to the branch workflow, not to the parent.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct ForkBranch {
    run_id: String,
    function: String,
}

impl ForkBranch {
    /// Create a fork branch from a durable run id and a registered function name.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::ForkBranch;
    ///
    /// let branch = ForkBranch::new("run:lhs", "investigate");
    /// assert!(format!("{branch:?}").contains("run:lhs"));
    /// ```
    pub fn new(run_id: impl Into<String>, function: impl Into<String>) -> Self {
        let run_id = run_id.into();
        let function = function.into();

        Self { run_id, function }
    }
}

/// Describes durable work that must be completed by a human operator.
///
/// A human request carries the presentation prompt plus opaque JSON data for
/// the scheduler or user interface. The crate stores and returns this request
/// unchanged; it does not interpret metadata, route work, or validate a human
/// response schema.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct HumanRequest {
    prompt: String,
    context: serde_json::Value,
    metadata: serde_json::Value,
}

impl HumanRequest {
    /// Create a human request with no structured context and empty metadata.
    ///
    /// `context` defaults to JSON null. `metadata` defaults to an empty JSON
    /// object, which makes it convenient for schedulers to add routing or audit
    /// fields when they need them.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::HumanRequest;
    ///
    /// let request = HumanRequest::new("Review the drafted response");
    /// assert_eq!(request.prompt(), "Review the drafted response");
    /// assert!(request.context().is_null());
    /// assert!(request.metadata().as_object().unwrap().is_empty());
    /// ```
    pub fn new(prompt: impl Into<String>) -> Self {
        Self {
            prompt: prompt.into(),
            context: serde_json::Value::Null,
            metadata: serde_json::Value::Object(serde_json::Map::new()),
        }
    }

    /// Return this request with structured task context.
    ///
    /// The context is opaque to `langcontinuation`; it is intended for the
    /// scheduler or human-facing UI that consumes [`WorkflowResult::Human`].
    pub fn with_context(mut self, context: serde_json::Value) -> Self {
        self.context = context;
        self
    }

    /// Return this request with scheduler metadata.
    ///
    /// Metadata is not required to be an object. An object is the conventional
    /// shape for routing and audit hints, but the crate preserves any JSON value
    /// the caller supplies.
    pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self {
        self.metadata = metadata;
        self
    }

    /// Return the operator-facing prompt.
    pub fn prompt(&self) -> &str {
        &self.prompt
    }

    /// Return the opaque task context.
    pub fn context(&self) -> &serde_json::Value {
        &self.context
    }

    /// Return the opaque scheduler metadata.
    pub fn metadata(&self) -> &serde_json::Value {
        &self.metadata
    }
}

/// Identifies one tool invocation for at-least-once deduplication.
///
/// A `ToolCallId` is constructed by the runtime from the durable workflow
/// `run_id` and the Anthropic `tool_use` id. Both halves are durable: the run
/// id is part of the serialized [`Workflow`], and the tool_use id is part of
/// the persisted model response. A workflow that crashes and replays from its
/// last persisted Anthropic resume therefore reconstructs an identical
/// `ToolCallId`, which lets a side-effecting [`Tool`] skip work it has already
/// performed.
///
/// The value is intentionally opaque. Tools should treat it as a stable key
/// rather than parse it, so the runtime can change the derivation (for example,
/// to add an attempt counter) without breaking deduplication code.
///
/// # Examples
///
/// ```rust
/// use langcontinuation::ToolCallId;
///
/// let id = ToolCallId::new("run-42", "toolu_abc");
/// let same = ToolCallId::new("run-42", "toolu_abc");
/// assert_eq!(id, same);
/// ```
#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct ToolCallId {
    run_id: String,
    tool_use_id: String,
}

impl ToolCallId {
    /// Construct a tool call id from a durable run id and a tool_use id.
    ///
    /// Runtimes call this; tools usually only compare or store the value they
    /// are handed.
    pub fn new(run_id: impl Into<String>, tool_use_id: impl Into<String>) -> Self {
        Self {
            run_id: run_id.into(),
            tool_use_id: tool_use_id.into(),
        }
    }

    /// Return the workflow run id this tool call belongs to.
    pub fn run_id(&self) -> &str {
        &self.run_id
    }

    /// Return the Anthropic tool_use id this tool call answers.
    pub fn tool_use_id(&self) -> &str {
        &self.tool_use_id
    }
}

impl Display for ToolCallId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}:{}", self.run_id, self.tool_use_id)
    }
}

/// Executes one client-side tool call during a [`Continuation::tool_call`] step.
///
/// A tool is registered with [`Trampoline::register_tool`] under a name that
/// matches the `name` field of the [`ToolUseBlock`] values the model emits. The
/// runtime — live or batch — looks the tool up by name and calls it. Tools are
/// held only by the runtime, never serialized into the [`Workflow`]; only the
/// tool's name is durable. Per-run parameters belong in the workflow
/// environment, where a tool can read them by a documented key.
///
/// The contract is the Anthropic wire contract: given a [`ToolUseBlock`],
/// produce a [`ToolResultBlock`]. The crate makes no assumption that a tool
/// touches a filesystem, a database, or the network. Filesystem-backed tools
/// are one implementation a caller writes.
///
/// Tools should be idempotent with respect to their [`ToolCallId`]. Because the
/// batch runtime runs tools inline and a crash replays from the last persisted
/// Anthropic resume, a side-effecting tool can be asked to run a given
/// `ToolCallId` more than once. Tools that record a durable marker keyed by the
/// id can skip duplicate work.
pub trait Tool: Send + Sync {
    /// Return the tool name matched against [`ToolUseBlock::name`].
    fn name(&self) -> String;

    /// Return the Anthropic tool parameter advertised to the model.
    ///
    /// Helpers that assemble the request array from the registered tools use
    /// this to tell the model which tools exist.
    fn to_param(&self) -> ToolUnionParam;

    /// Execute one tool call and return the result block to send back.
    ///
    /// `id` is a stable, replay-deterministic key for deduplication. `tool_use`
    /// carries the model-supplied id, name, and JSON input. The returned
    /// [`ToolResultBlock`] should carry the same `tool_use_id` as `tool_use`;
    /// the convenience constructor [`ToolResultBlock::new`] takes that id.
    fn call<'a>(
        &'a self,
        id: ToolCallId,
        tool_use: &'a ToolUseBlock,
    ) -> Pin<Box<dyn Future<Output = ToolResultBlock> + Send + 'a>>;
}

/// Extract the client-side tool_use blocks from a model response.
///
/// Only [`ContentBlock::ToolUse`] blocks are returned. Server-side tools (for
/// example Anthropic web search) and text blocks are ignored, because those are
/// not dispatched by the local tool registry.
///
/// This is the building block behind [`dispatch_tool_uses`]; it is public so a
/// receiver that wants custom routing (for example, branching on which tool was
/// called) can inspect the blocks directly.
pub fn client_tool_uses(response: &Message) -> Vec<ToolUseBlock> {
    let uses: Vec<ToolUseBlock> = response
        .content
        .iter()
        .filter_map(|block| match block {
            claudius::ContentBlock::ToolUse(tool_use) => Some(tool_use.clone()),
            _ => None,
        })
        .collect();
    uses
}

/// Outcome of [`dispatch_tool_uses`]: tools to run, or a finished turn.
///
/// The variant carries the continuation so the receiver keeps full control of
/// the terminal step. When the model called tools, the [`ContinuationChoice`]
/// is ready to return. When it did not, the [`Continuation`] is handed back so
/// the receiver can choose how to finish — halt, call another function, escalate
/// to a human, and so on.
pub enum ToolDispatch {
    /// The response called one or more tools. Return this choice to run them and
    /// re-enter the receiver after the results are stored.
    Tools(ContinuationChoice),
    /// The response called no tools. The caller owns the terminal continuation.
    Done(Continuation),
}

/// Raise a [`Continuation::tool_call`] suspension when a response calls tools.
///
/// This collapses the common agent-loop step: inspect the model response, and
/// if it contains client-side tool_use blocks, suspend to run them and resume
/// at `next_function`; otherwise hand the continuation back so the caller
/// chooses a terminal step. The tool results are stored under `output_key` as a
/// `Vec<ToolResultBlock>` for the receiver to thread into its conversation.
///
/// The suspension is explicit and visible: the runtime never auto-runs tools
/// behind the receiver's back. The receiver decides whether to dispatch.
///
/// # Examples
///
/// ```rust
/// use langcontinuation::{Continuation, ContinuationChoice, ToolDispatch, dispatch_tool_uses};
/// use claudius::Message;
///
/// # fn use_it(continuation: Continuation, response: &Message) -> ContinuationChoice {
/// match dispatch_tool_uses(
///     continuation,
///     response,
///     "tool_results: Vec<ToolResultBlock>",
///     "ask_again",
/// ) {
///     ToolDispatch::Tools(choice) => choice,
///     ToolDispatch::Done(continuation) => continuation.halt(),
/// }
/// # }
/// ```
pub fn dispatch_tool_uses(
    continuation: Continuation,
    response: &Message,
    output_key: impl Into<String>,
    next_function: impl Into<String>,
) -> ToolDispatch {
    let tool_uses = client_tool_uses(response);
    if tool_uses.is_empty() {
        ToolDispatch::Done(continuation)
    } else {
        let choice = continuation.tool_call(tool_uses, output_key, next_function);
        ToolDispatch::Tools(choice)
    }
}

/// Describes the continuation selected by one workflow function.
///
/// Values of this type are produced through [`Continuation`]. The fields are
/// intentionally private so callers cannot construct an invalid transition by
/// hand.
pub struct ContinuationChoice {
    steps: Vec<Step>,
    halt: bool,
}

impl ContinuationChoice {
    fn apply_to(self, workflow: &mut Workflow) {
        if self.halt {
            workflow.continuation.clear();
        } else {
            workflow.continuation.extend(self.steps.into_iter().rev());
        }
    }
}

#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
enum Step {
    #[default]
    Halt,
    Anthropic {
        provider: String,
        message: Box<MessageCreateParams>,
        output_key: String,
    },
    Human {
        request: HumanRequest,
        output_key: String,
    },
    ToolCall {
        tool_uses: Vec<ToolUseBlock>,
        output_key: String,
    },
    OpenAI {},
    Call {
        function: String,
    },
    ForkJoin {
        lhs: ForkBranch,
        rhs: ForkBranch,
        function: String,
    },
}

/// A walkable causal reference for a workflow event.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum CausalRef {
    /// The event is anchored to a known workflow run.
    RunId {
        /// The workflow run id.
        run_id: String,
    },
    /// The event is caused by a previously generated event.
    EventId {
        /// The globally unique event id.
        event_id: Uuid,
    },
}

impl CausalRef {
    fn run_id(run_id: impl Into<String>) -> Self {
        Self::RunId {
            run_id: run_id.into(),
        }
    }

    fn event_id(event_id: Uuid) -> Self {
        Self::EventId { event_id }
    }
}

/// Runtime observability context supplied by a durable executor.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ObservabilityContext {
    /// The causal parent for the next automatically caused event.
    pub causal_cursor: CausalRef,
}

/// Configuration for workflow event summarization and validation.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ObservabilityConfig {
    /// Maximum changed environment keys included in one diff payload.
    pub max_env_changes: usize,
    /// Maximum serialized JSON bytes accepted for one event payload.
    pub max_event_payload_bytes: usize,
}

impl Default for ObservabilityConfig {
    fn default() -> Self {
        Self {
            max_env_changes: 64,
            max_event_payload_bytes: 32 * 1024,
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum EventCauseMode {
    Automatic,
    Explicit,
}

/// A workflow event staged in memory before a durable commit.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct PendingWorkflowEvent {
    /// UUIDv7 event id generated when the event is recorded.
    pub event_id: Uuid,
    /// The event's causal parent.
    pub caused_by: CausalRef,
    /// Stable event type name.
    pub event_type: String,
    /// Event schema version.
    pub event_version: i16,
    /// Optional continuation correlation id.
    pub continuation_id: Option<String>,
    /// Event-specific JSON payload.
    pub event: serde_json::Value,
    #[cfg_attr(not(feature = "batch"), allow(dead_code))]
    #[serde(skip, default = "automatic_event_cause_mode")]
    cause_mode: EventCauseMode,
}

fn automatic_event_cause_mode() -> EventCauseMode {
    EventCauseMode::Automatic
}

impl PendingWorkflowEvent {
    fn custom<T: serde::Serialize>(
        event_type: impl Into<String>,
        event_version: i16,
        payload: T,
        caused_by: CausalRef,
        cause_mode: EventCauseMode,
        config: &ObservabilityConfig,
    ) -> Result<Self, handled::SError> {
        let event_type = event_type.into();
        validate_custom_event_type(&event_type)?;
        Self::new(
            event_type,
            event_version,
            None,
            payload,
            caused_by,
            cause_mode,
            config,
        )
    }

    #[cfg(feature = "batch")]
    pub(crate) fn first_party<T: serde::Serialize>(
        event_type: impl Into<String>,
        continuation_id: Option<String>,
        payload: T,
        caused_by: CausalRef,
        config: &ObservabilityConfig,
    ) -> Result<Self, handled::SError> {
        Self::new(
            event_type,
            1,
            continuation_id,
            payload,
            caused_by,
            EventCauseMode::Automatic,
            config,
        )
    }

    fn new<T: serde::Serialize>(
        event_type: impl Into<String>,
        event_version: i16,
        continuation_id: Option<String>,
        payload: T,
        caused_by: CausalRef,
        cause_mode: EventCauseMode,
        config: &ObservabilityConfig,
    ) -> Result<Self, handled::SError> {
        let event_type = event_type.into();
        if event_version <= 0 {
            return Err(observability_error(
                "invalid-event-version",
                "workflow event version must be positive",
            )
            .with_atom_field("event_version", event_version));
        }
        let event = serde_json::to_value(payload).map_err(|err| {
            observability_error(
                "invalid-event-payload",
                "failed to serialize workflow event payload",
            )
            .with_string_field("source", &err.to_string())
        })?;
        let payload_size = serde_json::to_vec(&event).map_err(|err| {
            observability_error(
                "invalid-event-payload",
                "failed to measure workflow event payload",
            )
            .with_string_field("source", &err.to_string())
        })?;
        if payload_size.len() > config.max_event_payload_bytes {
            return Err(observability_error(
                "event-payload-too-large",
                "workflow event payload exceeds configured maximum size",
            )
            .with_string_field("event_type", &event_type)
            .with_atom_field("payload_bytes", payload_size.len())
            .with_atom_field("max_payload_bytes", config.max_event_payload_bytes));
        }
        let event = Self {
            event_id: Uuid::now_v7(),
            caused_by,
            event_type,
            event_version,
            continuation_id,
            event,
            cause_mode,
        };
        Ok(event)
    }

    #[cfg(feature = "batch")]
    pub(crate) fn caused_automatically(&self) -> bool {
        self.cause_mode == EventCauseMode::Automatic
    }
}

/// Compact shape of a JSON environment value.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ValueShape {
    /// The key is absent.
    Missing,
    /// The value is JSON null.
    Null,
    /// The value is a boolean.
    Bool,
    /// The value is a number.
    Number,
    /// The value is a string.
    String,
    /// The value is an array.
    Array,
    /// The value is an object.
    Object,
}

/// Redacted summary of one environment value.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct EnvValueSummary {
    /// JSON value shape.
    pub shape: ValueShape,
    /// Compact JSON byte length when the value exists.
    pub bytes: Option<usize>,
    /// Setsum digest when the value exists.
    pub digest: Option<String>,
}

impl EnvValueSummary {
    fn missing() -> Self {
        Self {
            shape: ValueShape::Missing,
            bytes: None,
            digest: None,
        }
    }
}

/// Classification for one environment key mutation.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EnvChangeKind {
    /// The key was absent and is now present.
    Added,
    /// The key was present and is now absent.
    Removed,
    /// The key was present before and after with different JSON values.
    Modified,
}

/// Redacted summary of one changed environment key.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct EnvChangeSummary {
    /// Environment key.
    pub key: String,
    /// Change classification.
    pub change: EnvChangeKind,
    /// Summary of the previous value.
    pub before: EnvValueSummary,
    /// Summary of the new value.
    pub after: EnvValueSummary,
}

/// Redacted summary of environment changes across one local operation.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct EnvChangeSetSummary {
    /// Total number of changed keys before truncation.
    pub changed_key_count: usize,
    /// Whether the `changes` list was truncated by configuration.
    pub changes_truncated: bool,
    /// Digest of the full environment before the operation.
    pub env_before_digest: String,
    /// Digest of the full environment after the operation.
    pub env_after_digest: String,
    /// Number of keys in the environment before the operation.
    pub env_before_key_count: usize,
    /// Number of keys in the environment after the operation.
    pub env_after_key_count: usize,
    /// Included changed keys, sorted lexicographically and possibly truncated.
    pub changes: Vec<EnvChangeSummary>,
}

/// Sanitized summary of a workflow step.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum StepSummary {
    /// No active step remains.
    Halt,
    /// A local registered function call.
    Call {
        /// Function name.
        function: String,
    },
    /// An Anthropic provider request.
    Anthropic {
        /// Provider registry name.
        provider: String,
        /// Environment key receiving the response.
        output_key: String,
    },
    /// A human input request.
    Human {
        /// Environment key receiving the response.
        output_key: String,
    },
    /// A client-side tool-call dispatch.
    ToolCall {
        /// Tool names emitted by the model.
        tool_names: Vec<String>,
        /// Environment key receiving tool results.
        output_key: String,
    },
    /// A low-level OpenAI suspension.
    OpenAI,
    /// A fork/join split.
    ForkJoin {
        /// Branch names to branch run ids.
        branch_run_id: BTreeMap<String, String>,
        /// Function called after join.
        join_function: String,
    },
}

/// Sanitized summary of workflow control-flow movement.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct FlowSummary {
    /// Step before the operation.
    pub current_step_before: StepSummary,
    /// Step after the operation.
    pub current_step_after: StepSummary,
    /// Continuation stack depth before the operation.
    pub continuation_depth_before: usize,
    /// Continuation stack depth after the operation.
    pub continuation_depth_after: usize,
}

/// Public view of the next workflow action.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum WorkflowNext {
    /// The workflow would halt.
    Halt,
    /// The workflow would run a registered local function.
    LocalCall {
        /// Function name.
        function: String,
    },
    /// The workflow would suspend for Anthropic.
    Anthropic {
        /// Provider registry name.
        provider: String,
        /// Environment key receiving the response.
        output_key: String,
    },
    /// The workflow would suspend for human input.
    Human {
        /// Environment key receiving the response.
        output_key: String,
    },
    /// The workflow would dispatch client-side tool calls.
    ToolCall {
        /// Tool names emitted by the model.
        tool_names: Vec<String>,
        /// Environment key receiving tool results.
        output_key: String,
    },
    /// The workflow would suspend for a low-level OpenAI response.
    OpenAI,
    /// The workflow would fork into named branch workflows.
    ForkJoin {
        /// Branch names to branch run ids.
        branch_run_id: BTreeMap<String, String>,
        /// Function called after join.
        join_function: String,
    },
}

/// Result of running local workflow steps to a scheduler boundary.
#[derive(Clone, Debug)]
pub struct WorkflowOutcome {
    /// Boundary result reached by the trampoline.
    pub result: WorkflowResult,
    /// Pending events generated during execution.
    pub events: Vec<PendingWorkflowEvent>,
}

/// Result of executing exactly one local workflow call.
#[derive(Clone, Debug)]
pub struct WorkflowStepOutcome {
    /// Workflow after the local function and automatic step advance.
    pub workflow: Workflow,
    /// Function that was executed.
    pub function: String,
    /// Redacted environment changes made by the function.
    pub env_changes: EnvChangeSetSummary,
    /// Sanitized control-flow movement.
    pub flow: FlowSummary,
    /// Pending events generated during the call.
    pub events: Vec<PendingWorkflowEvent>,
    /// Duration of the local call in milliseconds.
    pub duration_ms: u128,
}

/// Error from local trampoline execution that preserves debug context.
#[derive(Debug)]
pub struct WorkflowError {
    /// Partially mutated workflow, for event/debug extraction only.
    pub workflow: Workflow,
    /// Function being executed when known.
    pub function: Option<String>,
    /// Redacted environment changes observed before the error.
    pub env_changes: EnvChangeSetSummary,
    /// Sanitized flow movement observed before the error.
    pub flow: FlowSummary,
    /// Pending events recorded before the error.
    pub events: Vec<PendingWorkflowEvent>,
    /// Structured source error.
    pub source: handled::SError,
    /// Duration of the failed local call in milliseconds when known.
    pub duration_ms: Option<u128>,
}

impl Display for WorkflowError {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        Display::fmt(&self.source, f)
    }
}

impl std::error::Error for WorkflowError {}

impl From<WorkflowError> for handled::SError {
    fn from(error: WorkflowError) -> Self {
        error
            .source
            .with_string_field("run_id", error.workflow.run_id())
            .with_atom_field("pending_event_count", error.events.len())
    }
}

#[derive(Clone, Debug)]
struct WorkflowObservabilityState {
    causal_cursor: CausalRef,
    pending_events: Vec<PendingWorkflowEvent>,
    config: ObservabilityConfig,
}

impl Default for WorkflowObservabilityState {
    fn default() -> Self {
        Self {
            causal_cursor: CausalRef::run_id(""),
            pending_events: Vec::new(),
            config: ObservabilityConfig::default(),
        }
    }
}

/// Carries durable execution state across local calls and external suspensions.
///
/// A workflow is ordinary Serde data. It can be stored, moved between
/// processes, resumed by a live executor, or interpreted by a custom scheduler.
/// Environment values are held as `serde_json::Value` and decoded at the typed
/// boundary where callers read them.
#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct Workflow {
    /// Stable run identifier.
    run_id: String,
    /// Arbitrary JSON environment state.
    env: HashMap<String, serde_json::Value>,
    /// Current active or suspended step; `Halt` means there is no active step.
    current_step: Step,
    /// Deferred execution stack/LIFO; next step is always popped from the end.
    continuation: Vec<Step>,
    /// Non-serialized event recording state for the current in-memory execution.
    #[serde(skip, default)]
    observability: WorkflowObservabilityState,
}

impl Workflow {
    /// Start a workflow at a registered function name.
    ///
    /// The `run_id` is copied into branch workflows only when the caller chooses
    /// it for a [`ForkBranch`]. The initial `call` must match a
    /// [`Trampoline::register`] name before the workflow can execute.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::Workflow;
    ///
    /// let workflow = Workflow::new("run-42", "entry");
    /// assert_eq!(workflow.run_id(), "run-42");
    /// ```
    pub fn new(run_id: impl Into<String>, call: impl Into<String>) -> Self {
        let run_id = run_id.into();
        let call = call.into();

        Self {
            run_id: run_id.clone(),
            env: HashMap::default(),
            current_step: Step::Call {
                function: call.clone(),
            },
            continuation: Vec::new(),
            observability: WorkflowObservabilityState {
                causal_cursor: CausalRef::run_id(&run_id),
                pending_events: Vec::new(),
                config: ObservabilityConfig::default(),
            },
        }
    }

    /// Return the stable identifier for this workflow run.
    ///
    /// The id is part of the durable workflow state and is preserved across
    /// serialization.
    pub fn run_id(&self) -> &str {
        &self.run_id
    }

    /// Set low-level observability context for a runtime execution.
    ///
    /// Normal workflow code should not call this. Durable runtimes use it to
    /// seed the causal cursor before local code records custom events.
    pub fn set_observability_context(&mut self, context: ObservabilityContext) {
        self.observability.causal_cursor = context.causal_cursor;
    }

    fn set_observability_config(&mut self, config: ObservabilityConfig) {
        self.observability.config = config;
    }

    /// Record a custom workflow event caused by the current causal cursor.
    ///
    /// The returned event id is generated immediately but is not durable until
    /// the runtime commits the pending event batch.
    ///
    /// # Errors
    ///
    /// Returns a structured error if the event type is invalid, if the payload
    /// cannot be serialized, or if the payload exceeds the configured size
    /// limit.
    pub fn record_event<T: serde::Serialize>(
        &mut self,
        event_type: impl Into<String>,
        payload: T,
    ) -> Result<Uuid, handled::SError> {
        let caused_by = self.current_causal_cursor();
        self.record_event_inner(event_type, 1, payload, caused_by, EventCauseMode::Automatic)
    }

    /// Record a custom workflow event with an explicit causal parent.
    ///
    /// The explicit cause is preserved when the event is committed. The
    /// workflow's in-memory causal cursor still advances to the new event.
    ///
    /// # Errors
    ///
    /// Returns a structured error if the event type is invalid, if the payload
    /// cannot be serialized, or if the payload exceeds the configured size
    /// limit.
    pub fn record_event_caused_by<T: serde::Serialize>(
        &mut self,
        event_type: impl Into<String>,
        payload: T,
        caused_by: CausalRef,
    ) -> Result<Uuid, handled::SError> {
        self.record_event_inner(event_type, 1, payload, caused_by, EventCauseMode::Explicit)
    }

    fn record_event_inner<T: serde::Serialize>(
        &mut self,
        event_type: impl Into<String>,
        event_version: i16,
        payload: T,
        caused_by: CausalRef,
        cause_mode: EventCauseMode,
    ) -> Result<Uuid, handled::SError> {
        let event = PendingWorkflowEvent::custom(
            event_type,
            event_version,
            payload,
            caused_by,
            cause_mode,
            &self.observability.config,
        )?;
        let event_id = event.event_id;
        self.observability.causal_cursor = CausalRef::event_id(event_id);
        self.observability.pending_events.push(event);
        Ok(event_id)
    }

    fn current_causal_cursor(&self) -> CausalRef {
        match &self.observability.causal_cursor {
            CausalRef::RunId { run_id } if run_id.is_empty() => CausalRef::run_id(&self.run_id),
            other => other.clone(),
        }
    }

    pub(crate) fn drain_pending_events(&mut self) -> Vec<PendingWorkflowEvent> {
        std::mem::take(&mut self.observability.pending_events)
    }

    /// Store a serializable value in the workflow environment.
    ///
    /// The environment stores JSON values, so the Rust type is remembered only
    /// by the caller or by a macro-generated key convention such as
    /// [`push_env!`].
    ///
    /// # Errors
    ///
    /// Returns a serialization error if `value` cannot be represented as
    /// `serde_json::Value`.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::Workflow;
    ///
    /// let mut workflow = Workflow::new("run", "entry");
    /// workflow.into_env("answer", 42_u64).unwrap();
    /// assert_eq!(workflow.from_env::<u64>("answer").unwrap(), Some(42));
    /// ```
    pub fn into_env<T: serde::Serialize>(
        &mut self,
        key: impl Into<String>,
        value: T,
    ) -> Result<(), serde_json::Error> {
        (|| {
            let key = key.into();
            let value = serde_json::to_value(&value)?;
            self.env.insert(key.clone(), value);
            Ok(())
        })()
    }

    /// Read and deserialize a value from the workflow environment.
    ///
    /// Missing keys are distinct from malformed values: absence returns
    /// `Ok(None)`, while a present value that cannot be decoded as `T` returns
    /// an error.
    ///
    /// # Errors
    ///
    /// Returns a deserialization error if the stored JSON value is not valid
    /// for `T`.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::Workflow;
    ///
    /// let mut workflow = Workflow::new("run", "entry");
    /// workflow.into_env("answer", 42_u64).unwrap();
    ///
    /// assert_eq!(workflow.from_env::<u64>("answer").unwrap(), Some(42));
    /// assert_eq!(workflow.from_env::<u64>("missing").unwrap(), None);
    /// ```
    pub fn from_env<T: for<'de> serde::Deserialize<'de>>(
        &self,
        key: impl AsRef<str>,
    ) -> Result<Option<T>, serde_json::Error> {
        (|| {
            let Some(value) = self.env.get(key.as_ref()) else {
                return Ok(None);
            };
            serde_json::from_value(value.clone()).map(Some)
        })()
    }

    fn advance(&mut self) -> bool {
        if let Some(next_step) = self.continuation.pop() {
            self.current_step = next_step;
            true
        } else {
            self.current_step = Step::Halt;
            false
        }
    }

    fn prepare_next_step(&mut self) {
        while matches!(self.current_step, Step::Halt) && !self.continuation.is_empty() {
            self.advance();
        }
    }

    fn fork_branch(&self, branch: ForkBranch) -> Self {
        let run_id = branch.run_id;

        Self {
            run_id: run_id.clone(),
            env: self.env.clone(),
            current_step: Step::Call {
                function: branch.function,
            },
            continuation: Vec::new(),
            observability: WorkflowObservabilityState {
                causal_cursor: CausalRef::run_id(&run_id),
                pending_events: Vec::new(),
                config: self.observability.config.clone(),
            },
        }
    }

    #[cfg(test)]
    fn schedule(&mut self, next_step: Step) {
        self.continuation.push(next_step);
    }
}

/// Marks the boundary where local trampoline execution stops.
///
/// A scheduler consumes this enum to decide whether the workflow is complete,
/// waiting on external work, waiting on human input, or ready to fork into
/// branch workflows.
#[derive(Clone, Debug)]
pub enum WorkflowResult {
    /// Indicates that no current or deferred step remains.
    Halt {
        /// The halted workflow state.
        workflow: Workflow,
    },
    /// Suspends execution for an Anthropic message request.
    Anthropic {
        /// The workflow paused at the Anthropic step.
        workflow: Workflow,
        /// The provider name that must be registered with [`Clients`].
        provider: String,
        /// The Anthropic request parameters to send.
        message: Box<MessageCreateParams>,
        /// The environment key where the response must be stored on resume.
        output_key: String,
    },
    /// Suspends execution for a human operator request.
    Human {
        /// The workflow paused at the human step.
        workflow: Workflow,
        /// The request that should be shown to the human operator.
        request: HumanRequest,
        /// The environment key where the answer must be stored on resume.
        output_key: String,
    },
    /// Suspends execution to run client-side tool calls.
    ToolCall {
        /// The workflow paused at the tool-call step.
        workflow: Workflow,
        /// The tool_use blocks the model emitted, to dispatch by name.
        tool_uses: Vec<ToolUseBlock>,
        /// The environment key where the `Vec<ToolResultBlock>` must be stored
        /// on resume.
        output_key: String,
    },
    /// Suspends execution for an OpenAI response handled by a custom runtime.
    OpenAI {
        /// The workflow paused at the OpenAI step.
        workflow: Workflow,
    },
    /// Suspends execution until two branch workflows complete.
    ForkJoin {
        /// The parent workflow paused at the join point.
        workflow: Workflow,
        /// The left branch workflow.
        lhs: Box<Workflow>,
        /// The right branch workflow.
        rhs: Box<Workflow>,
    },
}

/// Boxed future returned by a registered local workflow function.
pub type CallFuture<'a> = Pin<Box<dyn Future<Output = Result<(), handled::SError>> + 'a>>;

/// Represents a registered local workflow function.
///
/// A call mutates the workflow environment. Functions produced by
/// [`generate_goto!`] additionally choose their next step through a
/// [`Continuation`].
pub type Call = dyn for<'a> Fn(&'a mut Workflow) -> CallFuture<'a> + 'static;

#[cfg(test)]
pub(crate) fn test_sync_call<F>(
    f: F,
) -> impl for<'a> Fn(&'a mut Workflow) -> CallFuture<'a> + 'static
where
    F: Fn(&mut Workflow) -> Result<(), handled::SError> + Copy + 'static,
{
    move |workflow| -> CallFuture<'_> { Box::pin(async move { f(workflow) }) }
}

#[cfg(test)]
pub(crate) fn test_run_trampoline(
    trampoline: &Trampoline,
    workflow: Workflow,
) -> Result<WorkflowResult, Box<WorkflowError>> {
    let outcome = tokio::runtime::Runtime::new()
        .expect("test runtime")
        .block_on(trampoline.run(workflow))
        .map_err(Box::new)?;
    Ok(outcome.result)
}

/// Stores provider clients used by runtimes that execute suspended work.
#[derive(Default)]
pub struct Clients {
    clients: HashMap<String, Anthropic>,
}

impl Clients {
    /// Create an empty provider registry.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::Clients;
    ///
    /// let clients = Clients::new();
    /// ```
    pub fn new() -> Self {
        Self::default()
    }

    /// Register an Anthropic client under a provider name.
    ///
    /// Workflow steps refer to this provider name when they suspend through
    /// [`Continuation::anthropic`]. Registering the same name again replaces
    /// the previous client.
    pub fn register_anthropic(&mut self, provider: impl Into<String>, client: Anthropic) {
        let provider = provider.into();
        self.clients.insert(provider.clone(), client);
    }

    /// Send an Anthropic request through a registered provider.
    ///
    /// # Errors
    ///
    /// Returns `missing-anthropic-provider` if no client is registered under
    /// `provider`. Provider request failures are converted into structured
    /// `handled::SError` values that preserve retryability, status code, and
    /// request id when the underlying client exposes them.
    pub async fn send(
        &self,
        provider: &str,
        params: MessageCreateParams,
    ) -> Result<Message, handled::SError> {
        async {
            let Some(client) = self.clients.get(provider) else {
                return Err(missing_anthropic_provider_error(provider));
            };
            stream_anthropic_message(client, params).await
        }
        .await
    }
}

async fn stream_anthropic_message(
    client: &Anthropic,
    params: MessageCreateParams,
) -> Result<Message, handled::SError> {
    async {
        let params = params.with_stream(true);
        let stream = client.stream(&params).await.map_err(anthropic_error)?;
        let (mut stream, message_rx) = AccumulatingStream::new(stream);

        while let Some(event) = stream.next().await {
            print_llm_stream_event_text(event.map_err(anthropic_error)?);
        }

        let message = message_rx
            .await
            .map_err(|err| stream_accumulation_channel_error(&err.to_string()))?
            .map_err(anthropic_error)?;
        Ok(message)
    }
    .await
}

fn print_llm_stream_event_text(event: MessageStreamEvent) {
    match event {
        MessageStreamEvent::ContentBlockStart(start) => {
            if let claudius::ContentBlock::Text(text) = start.content_block
                && !text.text.is_empty()
            {
                print_llm_output_chunk(&text.text);
            }
        }
        MessageStreamEvent::ContentBlockDelta(delta) => {
            if let ContentBlockDelta::TextDelta(text_delta) = delta.delta {
                print_llm_output_chunk(&text_delta.text);
            }
        }
        _ => {}
    }
}

fn stream_accumulation_channel_error(source: &str) -> handled::SError {
    handled::SError::new("langcontinuation")
        .with_code("anthropic-stream-accumulation")
        .with_message("failed to receive accumulated Anthropic streaming message")
        .with_string_field("source", source)
}

fn unsupported_openai_error() -> handled::SError {
    handled::SError::new("langcontinuation")
        .with_code("unsupported-openai-provider")
        .with_message("OpenAI workflow steps are not supported")
        .with_string_field("next_action", "ask for a proper Rust OpenAI client")
}

fn human_input_required_error(request: &HumanRequest, output_key: &str) -> handled::SError {
    handled::SError::new("langcontinuation")
        .with_code("human-input-required")
        .with_message("human input is required to continue the workflow")
        .with_string_field("output_key", output_key)
        .with_string_field("prompt", request.prompt())
        .with_string_field("next_action", "resume the workflow with human input")
}

fn missing_anthropic_provider_error(provider: &str) -> handled::SError {
    handled::SError::new("langcontinuation")
        .with_code("missing-anthropic-provider")
        .with_message("Anthropic provider is not registered")
        .with_string_field("provider", provider)
}

fn anthropic_error(err: AnthropicError) -> handled::SError {
    let mut error = handled::SError::new("langcontinuation")
        .with_code(anthropic_error_code(&err))
        .with_message("Anthropic request failed")
        .with_atom_field("retryable", err.is_retryable())
        .with_string_field("source", &err.to_string());

    if let Some(status_code) = err.status_code() {
        error = error.with_atom_field("status_code", status_code);
    }
    if let Some(request_id) = err.request_id() {
        error = error.with_string_field("request_id", request_id);
    }

    error
}

fn anthropic_error_code(err: &AnthropicError) -> &'static str {
    if err.is_authentication() {
        "anthropic-authentication"
    } else if err.is_permission() {
        "anthropic-permission"
    } else if err.is_not_found() {
        "anthropic-not-found"
    } else if err.is_rate_limit() {
        "anthropic-rate-limit"
    } else if err.is_bad_request() {
        "anthropic-bad-request"
    } else if err.is_timeout() {
        "anthropic-timeout"
    } else if err.is_abort() {
        "anthropic-abort"
    } else if err.is_connection() {
        "anthropic-connection"
    } else if err.is_server_error() {
        "anthropic-server-error"
    } else if err.is_validation() {
        "anthropic-validation"
    } else if err.is_todo() {
        "anthropic-unimplemented"
    } else {
        "anthropic-request-failed"
    }
}

/// Dispatches registered local calls until execution reaches a scheduler boundary.
///
/// The trampoline is deliberately small. It knows how to call named Rust
/// functions and how to recognize workflow steps that require an external
/// runtime. It does not own persistence, provider credentials, or scheduling
/// policy.
#[derive(Default)]
pub struct Trampoline {
    fns: HashMap<String, Box<Call>>,
    tools: HashMap<String, std::sync::Arc<dyn Tool>>,
    observability_config: ObservabilityConfig,
}

impl Trampoline {
    /// Set event summarization and validation configuration.
    pub fn set_observability_config(&mut self, config: ObservabilityConfig) {
        self.observability_config = config;
    }

    #[cfg(feature = "batch")]
    pub(crate) fn observability_config(&self) -> &ObservabilityConfig {
        &self.observability_config
    }

    /// Run local workflow steps until the workflow halts or suspends.
    ///
    /// Registered calls execute asynchronously in the current task. When a call
    /// chooses an Anthropic, human, OpenAI, or fork/join continuation, the
    /// trampoline returns the corresponding [`WorkflowResult`] instead of
    /// performing the external work itself.
    ///
    /// # Errors
    ///
    /// Returns `missing-function` if a `Call` step names a function that has
    /// not been registered. Also returns any `handled::SError` produced by a
    /// registered call.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::{CallFuture, Trampoline, Workflow, WorkflowResult};
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), handled::SError> {
    ///     let mut trampoline = Trampoline::default();
    ///     trampoline.register("mark", |workflow| -> CallFuture<'_> {
    ///         Box::pin(async move {
    ///             workflow.into_env("done", true).unwrap();
    ///             Ok(())
    ///         })
    ///     });
    ///
    ///     let result = trampoline.run(Workflow::new("run", "mark")).await?.result;
    ///     let WorkflowResult::Halt { workflow } = result else {
    ///         panic!("workflow should halt");
    ///     };
    ///     assert_eq!(workflow.from_env::<bool>("done").unwrap(), Some(true));
    ///     Ok(())
    /// }
    /// ```
    pub async fn run(&self, mut workflow: Workflow) -> Result<WorkflowOutcome, WorkflowError> {
        let mut events = Vec::new();
        workflow.set_observability_config(self.observability_config.clone());
        loop {
            workflow.prepare_next_step();
            match workflow.current_step.clone() {
                Step::Halt => {
                    let mut workflow = workflow;
                    events.extend(workflow.drain_pending_events());
                    return Ok(WorkflowOutcome {
                        result: WorkflowResult::Halt { workflow },
                        events,
                    });
                }
                Step::Anthropic {
                    provider,
                    message,
                    output_key,
                } => {
                    events.extend(workflow.drain_pending_events());
                    return Ok(WorkflowOutcome {
                        result: WorkflowResult::Anthropic {
                            workflow,
                            provider,
                            message,
                            output_key,
                        },
                        events,
                    });
                }
                Step::Human {
                    request,
                    output_key,
                } => {
                    events.extend(workflow.drain_pending_events());
                    return Ok(WorkflowOutcome {
                        result: WorkflowResult::Human {
                            workflow,
                            request,
                            output_key,
                        },
                        events,
                    });
                }
                Step::ToolCall {
                    tool_uses,
                    output_key,
                } => {
                    events.extend(workflow.drain_pending_events());
                    return Ok(WorkflowOutcome {
                        result: WorkflowResult::ToolCall {
                            workflow,
                            tool_uses,
                            output_key,
                        },
                        events,
                    });
                }
                Step::OpenAI {} => {
                    events.extend(workflow.drain_pending_events());
                    return Ok(WorkflowOutcome {
                        result: WorkflowResult::OpenAI { workflow },
                        events,
                    });
                }
                Step::ForkJoin {
                    lhs,
                    rhs,
                    function: _,
                } => {
                    let lhs = workflow.fork_branch(lhs);
                    let rhs = workflow.fork_branch(rhs);
                    events.extend(workflow.drain_pending_events());
                    return Ok(WorkflowOutcome {
                        result: WorkflowResult::ForkJoin {
                            workflow,
                            lhs: Box::new(lhs),
                            rhs: Box::new(rhs),
                        },
                        events,
                    });
                }
                Step::Call { .. } => match self.run_one_local_call(workflow).await {
                    Ok(outcome) => {
                        events.extend(outcome.events);
                        workflow = outcome.workflow;
                    }
                    Err(mut err) => {
                        events.extend(err.events);
                        err.events = events;
                        return Err(err);
                    }
                },
            }
        }
    }

    /// Return the next sanitized action the trampoline would take.
    pub fn next_action(&self, workflow: &Workflow) -> WorkflowNext {
        effective_step(workflow).next()
    }

    /// Execute exactly one local call.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowError`] if the workflow is not currently at a local
    /// call, if the function is not registered, or if the function itself
    /// returns an error.
    pub async fn run_one_local_call(
        &self,
        mut workflow: Workflow,
    ) -> Result<WorkflowStepOutcome, WorkflowError> {
        workflow.set_observability_config(self.observability_config.clone());
        workflow.prepare_next_step();
        let before_env = workflow.env.clone();
        let before_step = workflow.current_step.summary();
        let before_depth = workflow.continuation.len();
        let function = match workflow.current_step.clone() {
            Step::Call { function } => function,
            other => {
                let flow = FlowSummary {
                    current_step_before: before_step,
                    current_step_after: other.summary(),
                    continuation_depth_before: before_depth,
                    continuation_depth_after: workflow.continuation.len(),
                };
                let env_changes =
                    summarize_env_changes(&before_env, &workflow.env, &self.observability_config)
                        .unwrap_or_else(empty_env_change_summary);
                let events = workflow.drain_pending_events();
                return Err(WorkflowError {
                    workflow,
                    function: None,
                    env_changes,
                    flow,
                    events,
                    source: observability_error(
                        "not-local-call",
                        "attempted to execute one local call when the workflow is not at a local call",
                    ),
                    duration_ms: None,
                });
            }
        };
        let Some(implementation) = self.fns.get(&function) else {
            let flow = FlowSummary {
                current_step_before: before_step,
                current_step_after: workflow.current_step.summary(),
                continuation_depth_before: before_depth,
                continuation_depth_after: workflow.continuation.len(),
            };
            let env_changes =
                summarize_env_changes(&before_env, &workflow.env, &self.observability_config)
                    .unwrap_or_else(empty_env_change_summary);
            let events = workflow.drain_pending_events();
            return Err(WorkflowError {
                workflow,
                function: Some(function.clone()),
                env_changes,
                flow,
                events,
                source: missing_function_error(&function),
                duration_ms: Some(0),
            });
        };
        let started = Instant::now();
        match implementation(&mut workflow).await {
            Ok(()) => {
                workflow.advance();
                let duration_ms = started.elapsed().as_millis();
                let env_changes =
                    summarize_env_changes(&before_env, &workflow.env, &self.observability_config)
                        .unwrap_or_else(empty_env_change_summary);
                let flow = FlowSummary {
                    current_step_before: before_step,
                    current_step_after: workflow.current_step.summary(),
                    continuation_depth_before: before_depth,
                    continuation_depth_after: workflow.continuation.len(),
                };
                let events = workflow.drain_pending_events();
                Ok(WorkflowStepOutcome {
                    workflow,
                    function,
                    env_changes,
                    flow,
                    events,
                    duration_ms,
                })
            }
            Err(source) => {
                let duration_ms = started.elapsed().as_millis();
                let env_changes =
                    summarize_env_changes(&before_env, &workflow.env, &self.observability_config)
                        .unwrap_or_else(empty_env_change_summary);
                let flow = FlowSummary {
                    current_step_before: before_step,
                    current_step_after: workflow.current_step.summary(),
                    continuation_depth_before: before_depth,
                    continuation_depth_after: workflow.continuation.len(),
                };
                let events = workflow.drain_pending_events();
                Err(WorkflowError {
                    workflow,
                    function: Some(function),
                    env_changes,
                    flow,
                    events,
                    source,
                    duration_ms: Some(duration_ms),
                })
            }
        }
    }

    /// Store an Anthropic response and advance the suspended workflow.
    ///
    /// The workflow must be paused at the Anthropic step that requested
    /// `output_key`. The message is serialized into the environment and the
    /// workflow advances to the continuation that was waiting behind the
    /// provider step.
    ///
    /// # Errors
    ///
    /// Returns `not-suspended-at-anthropic` if the workflow is not paused at an
    /// Anthropic step, `anthropic-output-key-mismatch` if the supplied key does
    /// not match the suspended step, or `invalid-anthropic-response` if the
    /// response cannot be serialized into the environment.
    pub fn resume_anthropic(
        &self,
        mut workflow: Workflow,
        output_key: impl Into<String>,
        message: Message,
    ) -> Result<Workflow, handled::SError> {
        (|| {
            let output_key = output_key.into();
            match &workflow.current_step {
                Step::Anthropic {
                    output_key: suspended_output_key,
                    ..
                } if suspended_output_key == &output_key => {}
                Step::Anthropic {
                    output_key: suspended_output_key,
                    ..
                } => {
                    return Err(resume_error(
                        "anthropic-output-key-mismatch",
                        "attempted to resume an Anthropic step with the wrong output key",
                        Some(("expected", suspended_output_key)),
                        Some(("actual", &output_key)),
                    ));
                }
                _ => {
                    return Err(resume_error(
                        "not-suspended-at-anthropic",
                        "attempted to resume Anthropic output on a workflow that is not suspended at an Anthropic step",
                        None,
                        Some(("output_key", &output_key)),
                    ));
                }
            }

            let value = serde_json::to_value(message).map_err(|err| {
                handled::SError::new("langcontinuation")
                    .with_code("invalid-anthropic-response")
                    .with_message(
                        "failed to serialize Anthropic response into workflow environment",
                    )
                    .with_string_field("output_key", &output_key)
                    .with_string_field("source", &err.to_string())
            })?;
            workflow.env.insert(output_key, value);
            workflow.advance();
            Ok(workflow)
        })()
    }

    /// Store a human answer and advance the suspended workflow.
    ///
    /// The workflow must be paused at the human step that requested
    /// `output_key`. The answer is serialized into the environment and the
    /// workflow advances to the continuation that was waiting behind the human
    /// request.
    ///
    /// # Errors
    ///
    /// Returns `not-suspended-at-human` if the workflow is not paused at a
    /// human step, `human-output-key-mismatch` if the supplied key does not
    /// match the suspended step, or `invalid-human-response` if the answer
    /// cannot be serialized into the environment.
    pub fn resume_human<T: serde::Serialize>(
        &self,
        mut workflow: Workflow,
        output_key: impl Into<String>,
        value: T,
    ) -> Result<Workflow, handled::SError> {
        (|| {
            let output_key = output_key.into();
            match &workflow.current_step {
                Step::Human {
                    output_key: suspended_output_key,
                    ..
                } if suspended_output_key == &output_key => {}
                Step::Human {
                    output_key: suspended_output_key,
                    ..
                } => {
                    return Err(resume_error(
                        "human-output-key-mismatch",
                        "attempted to resume a human step with the wrong output key",
                        Some(("expected", suspended_output_key)),
                        Some(("actual", &output_key)),
                    ));
                }
                _ => {
                    return Err(resume_error(
                        "not-suspended-at-human",
                        "attempted to resume human output on a workflow that is not suspended at a human step",
                        None,
                        Some(("output_key", &output_key)),
                    ));
                }
            }

            insert_resume_value(
                &mut workflow,
                output_key,
                value,
                "invalid-human-response",
                "failed to serialize human response into workflow environment",
            )?;
            workflow.advance();
            Ok(workflow)
        })()
    }

    /// Store tool results and advance the suspended workflow.
    ///
    /// The workflow must be paused at the tool-call step that requested
    /// `output_key`. The `Vec<ToolResultBlock>` is serialized into the
    /// environment under that key and the workflow advances to the receiver
    /// that was waiting behind the tool-call step. The crate does not thread the
    /// blocks into a conversation; the receiver owns the transcript and decides
    /// how the results become the next user message.
    ///
    /// # Errors
    ///
    /// Returns `not-suspended-at-tool-call` if the workflow is not paused at a
    /// tool-call step, `tool-call-output-key-mismatch` if the supplied key does
    /// not match the suspended step, or `invalid-tool-results` if the results
    /// cannot be serialized into the environment.
    pub fn resume_tool_call(
        &self,
        mut workflow: Workflow,
        output_key: impl Into<String>,
        results: Vec<ToolResultBlock>,
    ) -> Result<Workflow, handled::SError> {
        (|| {
            let output_key = output_key.into();
            match &workflow.current_step {
                Step::ToolCall {
                    output_key: suspended_output_key,
                    ..
                } if suspended_output_key == &output_key => {}
                Step::ToolCall {
                    output_key: suspended_output_key,
                    ..
                } => {
                    return Err(resume_error(
                        "tool-call-output-key-mismatch",
                        "attempted to resume a tool-call step with the wrong output key",
                        Some(("expected", suspended_output_key)),
                        Some(("actual", &output_key)),
                    ));
                }
                _ => {
                    return Err(resume_error(
                        "not-suspended-at-tool-call",
                        "attempted to resume tool results on a workflow that is not suspended at a tool-call step",
                        None,
                        Some(("output_key", &output_key)),
                    ));
                }
            }

            insert_resume_value(
                &mut workflow,
                output_key,
                results,
                "invalid-tool-results",
                "failed to serialize tool results into workflow environment",
            )?;
            workflow.advance();
            Ok(workflow)
        })()
    }

    /// Merge halted branch workflows and advance the parent to its join call.
    ///
    /// Branch environments are compared to the parent environment captured at
    /// the fork. A key changed by only one branch is accepted. A key changed by
    /// both branches is accepted only when both branches wrote the same JSON
    /// value.
    ///
    /// # Errors
    ///
    /// Returns `not-suspended-at-fork-join` if the parent is not paused at a
    /// fork/join step, `fork-join-branch-not-halted` if either branch still has
    /// work remaining, or `fork-join-env-conflict` if branches wrote conflicting
    /// values for the same environment key.
    pub fn resume_fork_join(
        &self,
        mut workflow: Workflow,
        lhs: Workflow,
        rhs: Workflow,
    ) -> Result<Workflow, handled::SError> {
        (|| {
            let function = match &workflow.current_step {
                Step::ForkJoin { function, .. } => function.clone(),
                _ => {
                    return Err(fork_join_resume_error(&workflow.current_step));
                }
            };

            require_halted_branch("lhs", &lhs)?;
            require_halted_branch("rhs", &rhs)?;

            workflow.env = merge_fork_join_env(&workflow.env, &lhs.env, &rhs.env)?;
            workflow.current_step = Step::Call { function };
            Ok(workflow)
        })()
    }

    /// Store an OpenAI response value and advance the suspended workflow.
    ///
    /// OpenAI is represented in the workflow state machine, but the bundled
    /// live executor does not yet perform OpenAI requests. Custom runtimes can
    /// use this method to resume workflows after they have obtained a response
    /// from a suitable Rust OpenAI client.
    ///
    /// # Errors
    ///
    /// Returns `not-suspended-at-openai` if the workflow is not paused at an
    /// OpenAI step, or `invalid-openai-response` if `value` cannot be serialized
    /// into the environment.
    pub fn resume_open_ai<T: serde::Serialize>(
        &self,
        mut workflow: Workflow,
        output_key: impl Into<String>,
        value: T,
    ) -> Result<Workflow, handled::SError> {
        (|| {
            let output_key = output_key.into();
            match &workflow.current_step {
                Step::OpenAI {} => {}
                _ => {
                    return Err(resume_error(
                        "not-suspended-at-openai",
                        "attempted to resume OpenAI output on a workflow that is not suspended at an OpenAI step",
                        None,
                        Some(("output_key", &output_key)),
                    ));
                }
            }

            insert_resume_value(
                &mut workflow,
                output_key,
                value,
                "invalid-openai-response",
                "failed to serialize OpenAI response into workflow environment",
            )?;
            workflow.advance();
            Ok(workflow)
        })()
    }

    /// Associate an exact function name with a local workflow call.
    ///
    /// Names used by [`Workflow::new`], [`Continuation::call`], and
    /// [`ForkBranch::new`] must match a registration name exactly before the
    /// trampoline can execute that call. Registering the same name again
    /// replaces the previous implementation.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use langcontinuation::{CallFuture, Trampoline, Workflow, WorkflowResult};
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), handled::SError> {
    ///     let mut trampoline = Trampoline::default();
    ///     trampoline.register("entry", |workflow| -> CallFuture<'_> {
    ///         Box::pin(async move {
    ///             let run_id = workflow.run_id().to_string();
    ///             workflow.into_env("visited", run_id).unwrap();
    ///             Ok(())
    ///         })
    ///     });
    ///
    ///     let WorkflowResult::Halt { workflow } =
    ///         trampoline.run(Workflow::new("run", "entry")).await?.result
    ///     else {
    ///         panic!("workflow should halt");
    ///     };
    ///     assert_eq!(
    ///         workflow.from_env::<String>("visited").unwrap(),
    ///         Some("run".to_string())
    ///     );
    ///     Ok(())
    /// }
    /// ```
    pub fn register(
        &mut self,
        function: impl Into<String>,
        implementation: impl for<'a> Fn(&'a mut Workflow) -> CallFuture<'a> + 'static,
    ) {
        let function = function.into();
        self.fns
            .insert(function.clone(), Box::new(implementation) as _);
    }

    /// Register a client-side [`Tool`] under a name.
    ///
    /// The name should match the `name` field of the [`ToolUseBlock`] values the
    /// model emits. A runtime resolves a [`WorkflowResult::ToolCall`] by looking
    /// up each tool_use by name in this registry. Registering the same name
    /// again replaces the previous tool. Tools are held only by the trampoline;
    /// they are never serialized into a [`Workflow`].
    ///
    /// # Examples
    ///
    /// ```rust
    /// use std::pin::Pin;
    /// use std::future::Future;
    /// use langcontinuation::{Tool, ToolCallId, Trampoline};
    /// use claudius::{ToolResultBlock, ToolUnionParam, ToolUseBlock};
    ///
    /// struct Echo;
    /// impl Tool for Echo {
    ///     fn name(&self) -> String {
    ///         "echo".into()
    ///     }
    ///     fn to_param(&self) -> ToolUnionParam {
    ///         unimplemented!("doc test does not advertise the tool")
    ///     }
    ///     fn call<'a>(
    ///         &'a self,
    ///         _id: ToolCallId,
    ///         tool_use: &'a ToolUseBlock,
    ///     ) -> Pin<Box<dyn Future<Output = ToolResultBlock> + Send + 'a>> {
    ///         let id = tool_use.id.clone();
    ///         Box::pin(async move { ToolResultBlock::new(id).with_string_content("ok".into()) })
    ///     }
    /// }
    ///
    /// let mut trampoline = Trampoline::default();
    /// trampoline.register_tool(Echo);
    /// ```
    pub fn register_tool(&mut self, tool: impl Tool + 'static) {
        let name = tool.name();
        self.tools.insert(name.clone(), std::sync::Arc::new(tool));
    }

    /// Look up a registered tool by name.
    ///
    /// Runtimes use this while performing a [`WorkflowResult::ToolCall`]. Returns
    /// `None` if no tool was registered under the name.
    pub fn tool(&self, name: &str) -> Option<std::sync::Arc<dyn Tool>> {
        self.tools.get(name).cloned()
    }

    /// Run every tool_use block through the registered tools.
    ///
    /// This is the shared dispatch used by both the live and batch runtimes when
    /// they perform a [`WorkflowResult::ToolCall`]. Each tool_use is resolved by
    /// name with [`Self::tool`], assigned a replay-deterministic [`ToolCallId`]
    /// from `run_id` and the tool_use id, and executed in order. The returned
    /// blocks are in the same order as `tool_uses`.
    ///
    /// # Errors
    ///
    /// Returns `missing-tool` if any tool_use names a tool that is not
    /// registered.
    pub async fn run_tool_calls(
        &self,
        run_id: &str,
        tool_uses: &[ToolUseBlock],
    ) -> Result<Vec<ToolResultBlock>, handled::SError> {
        async {
            let mut results = Vec::with_capacity(tool_uses.len());
            for tool_use in tool_uses {
                let tool = self
                    .tool(&tool_use.name)
                    .ok_or_else(|| missing_tool_error(&tool_use.name))?;
                let id = ToolCallId::new(run_id, &tool_use.id);
                results.push(tool.call(id, tool_use).await);
            }
            Ok(results)
        }
        .await
    }
}

impl Step {
    fn summary(&self) -> StepSummary {
        match self {
            Step::Halt => StepSummary::Halt,
            Step::Anthropic {
                provider,
                output_key,
                ..
            } => StepSummary::Anthropic {
                provider: provider.clone(),
                output_key: output_key.clone(),
            },
            Step::Human { output_key, .. } => StepSummary::Human {
                output_key: output_key.clone(),
            },
            Step::ToolCall {
                tool_uses,
                output_key,
            } => StepSummary::ToolCall {
                tool_names: tool_uses
                    .iter()
                    .map(|tool_use| tool_use.name.clone())
                    .collect(),
                output_key: output_key.clone(),
            },
            Step::OpenAI {} => StepSummary::OpenAI,
            Step::Call { function } => StepSummary::Call {
                function: function.clone(),
            },
            Step::ForkJoin { lhs, rhs, function } => {
                let mut branch_run_id = BTreeMap::new();
                branch_run_id.insert("lhs".to_string(), lhs.run_id.clone());
                branch_run_id.insert("rhs".to_string(), rhs.run_id.clone());
                StepSummary::ForkJoin {
                    branch_run_id,
                    join_function: function.clone(),
                }
            }
        }
    }

    fn next(&self) -> WorkflowNext {
        match self.summary() {
            StepSummary::Halt => WorkflowNext::Halt,
            StepSummary::Call { function } => WorkflowNext::LocalCall { function },
            StepSummary::Anthropic {
                provider,
                output_key,
            } => WorkflowNext::Anthropic {
                provider,
                output_key,
            },
            StepSummary::Human { output_key } => WorkflowNext::Human { output_key },
            StepSummary::ToolCall {
                tool_names,
                output_key,
            } => WorkflowNext::ToolCall {
                tool_names,
                output_key,
            },
            StepSummary::OpenAI => WorkflowNext::OpenAI,
            StepSummary::ForkJoin {
                branch_run_id,
                join_function,
            } => WorkflowNext::ForkJoin {
                branch_run_id,
                join_function,
            },
        }
    }
}

fn validate_custom_event_type(event_type: &str) -> Result<(), handled::SError> {
    const RESERVED_PREFIXES: &[&str] = &[
        "workflow.",
        "local_call.",
        "continuation.",
        "anthropic.",
        "openai.",
        "human.",
        "tool.",
        "tool_call.",
        "fork_join.",
    ];
    if event_type.is_empty() || !event_type.contains('.') {
        return Err(observability_error(
            "invalid-custom-event-type",
            "custom workflow event type must be non-empty and contain a dot",
        )
        .with_string_field("event_type", event_type));
    }
    if RESERVED_PREFIXES
        .iter()
        .any(|prefix| event_type.starts_with(prefix))
    {
        return Err(observability_error(
            "reserved-custom-event-type",
            "custom workflow event type uses a reserved first-party prefix",
        )
        .with_string_field("event_type", event_type));
    }
    Ok(())
}

fn summarize_env_changes(
    before: &HashMap<String, serde_json::Value>,
    after: &HashMap<String, serde_json::Value>,
    config: &ObservabilityConfig,
) -> Result<EnvChangeSetSummary, handled::SError> {
    let mut keys = HashSet::new();
    keys.extend(before.keys().cloned());
    keys.extend(after.keys().cloned());
    let mut changed = Vec::new();
    for key in keys {
        let before_value = before.get(&key);
        let after_value = after.get(&key);
        if before_value == after_value {
            continue;
        }
        let change = match (before_value, after_value) {
            (None, Some(_)) => EnvChangeKind::Added,
            (Some(_), None) => EnvChangeKind::Removed,
            (Some(_), Some(_)) => EnvChangeKind::Modified,
            (None, None) => continue,
        };
        changed.push(EnvChangeSummary {
            key,
            change,
            before: summarize_env_value(before_value)?,
            after: summarize_env_value(after_value)?,
        });
    }
    changed.sort_by(|lhs, rhs| lhs.key.cmp(&rhs.key));
    let changed_key_count = changed.len();
    let changes_truncated = changed_key_count > config.max_env_changes;
    changed.truncate(config.max_env_changes);
    Ok(EnvChangeSetSummary {
        changed_key_count,
        changes_truncated,
        env_before_digest: env_digest(before)?,
        env_after_digest: env_digest(after)?,
        env_before_key_count: before.len(),
        env_after_key_count: after.len(),
        changes: changed,
    })
}

fn empty_env_change_summary(_: handled::SError) -> EnvChangeSetSummary {
    EnvChangeSetSummary {
        changed_key_count: 0,
        changes_truncated: false,
        env_before_digest: "setsum:v1:error".to_string(),
        env_after_digest: "setsum:v1:error".to_string(),
        env_before_key_count: 0,
        env_after_key_count: 0,
        changes: Vec::new(),
    }
}

fn effective_step(workflow: &Workflow) -> Step {
    if matches!(workflow.current_step, Step::Halt)
        && let Some(next) = workflow.continuation.last()
    {
        return next.clone();
    }
    workflow.current_step.clone()
}

fn summarize_env_value(
    value: Option<&serde_json::Value>,
) -> Result<EnvValueSummary, handled::SError> {
    let Some(value) = value else {
        return Ok(EnvValueSummary::missing());
    };
    let normalized = normalized_json_bytes(value)?;
    Ok(EnvValueSummary {
        shape: value_shape(value),
        bytes: Some(normalized.len()),
        digest: Some(setsum_digest([normalized.as_slice()])),
    })
}

fn value_shape(value: &serde_json::Value) -> ValueShape {
    match value {
        serde_json::Value::Null => ValueShape::Null,
        serde_json::Value::Bool(_) => ValueShape::Bool,
        serde_json::Value::Number(_) => ValueShape::Number,
        serde_json::Value::String(_) => ValueShape::String,
        serde_json::Value::Array(_) => ValueShape::Array,
        serde_json::Value::Object(_) => ValueShape::Object,
    }
}

fn env_digest(env: &HashMap<String, serde_json::Value>) -> Result<String, handled::SError> {
    let mut setsum = Setsum::default();
    for (key, value) in env {
        let value_bytes = normalized_json_bytes(value)?;
        let value_digest = setsum_digest([value_bytes.as_slice()]);
        let element = length_prefixed_parts([key.as_bytes(), value_digest.as_bytes()]);
        setsum.insert(&element);
    }
    Ok(format!("setsum:v1:{}", setsum.hexdigest()))
}

fn normalized_json_bytes(value: &serde_json::Value) -> Result<Vec<u8>, handled::SError> {
    serde_json::to_vec(&normalize_json_value(value)).map_err(|err| {
        observability_error(
            "invalid-json-summary",
            "failed to serialize normalized JSON for observability summary",
        )
        .with_string_field("source", &err.to_string())
    })
}

fn normalize_json_value(value: &serde_json::Value) -> serde_json::Value {
    match value {
        serde_json::Value::Array(values) => {
            serde_json::Value::Array(values.iter().map(normalize_json_value).collect())
        }
        serde_json::Value::Object(map) => {
            let mut keys: Vec<_> = map.keys().collect();
            keys.sort();
            let mut normalized = serde_json::Map::new();
            for key in keys {
                if let Some(value) = map.get(key) {
                    normalized.insert(key.clone(), normalize_json_value(value));
                }
            }
            serde_json::Value::Object(normalized)
        }
        other => other.clone(),
    }
}

fn setsum_digest<'a>(parts: impl IntoIterator<Item = &'a [u8]>) -> String {
    let mut setsum = Setsum::default();
    setsum.insert(&length_prefixed_parts(parts));
    format!("setsum:v1:{}", setsum.hexdigest())
}

fn length_prefixed_parts<'a>(parts: impl IntoIterator<Item = &'a [u8]>) -> Vec<u8> {
    let mut bytes = Vec::new();
    for part in parts {
        bytes.extend_from_slice(&(part.len() as u64).to_be_bytes());
        bytes.extend_from_slice(part);
    }
    bytes
}

fn observability_error(code: &str, message: &str) -> handled::SError {
    handled::SError::new("langcontinuation")
        .with_code(code)
        .with_message(message)
}

fn require_halted_branch(name: &str, branch: &Workflow) -> Result<(), handled::SError> {
    if matches!(branch.current_step, Step::Halt) {
        Ok(())
    } else {
        Err(handled::SError::new("langcontinuation")
            .with_code("fork-join-branch-not-halted")
            .with_message("fork/join branch did not halt before join resume")
            .with_string_field("branch", name)
            .with_string_field("run_id", &branch.run_id)
            .with_string_field("current_step", &format!("{:?}", branch.current_step)))
    }
}

fn merge_fork_join_env(
    base: &HashMap<String, serde_json::Value>,
    lhs: &HashMap<String, serde_json::Value>,
    rhs: &HashMap<String, serde_json::Value>,
) -> Result<HashMap<String, serde_json::Value>, handled::SError> {
    (|| {
        let mut merged = base.clone();
        let mut keys = HashSet::new();
        keys.extend(base.keys().cloned());
        keys.extend(lhs.keys().cloned());
        keys.extend(rhs.keys().cloned());

        for key in keys {
            let base_value = base.get(&key);
            let lhs_value = lhs.get(&key);
            let rhs_value = rhs.get(&key);
            let lhs_changed = lhs_value != base_value;
            let rhs_changed = rhs_value != base_value;

            match (lhs_changed, rhs_changed) {
                (false, false) => {}
                (true, false) => apply_fork_env_change(&mut merged, key, lhs_value.cloned()),
                (false, true) => apply_fork_env_change(&mut merged, key, rhs_value.cloned()),
                (true, true) if lhs_value == rhs_value => {
                    apply_fork_env_change(&mut merged, key, lhs_value.cloned());
                }
                (true, true) => {
                    return Err(fork_join_env_conflict_error(&key, lhs_value, rhs_value));
                }
            }
        }

        Ok(merged)
    })()
}

fn apply_fork_env_change(
    env: &mut HashMap<String, serde_json::Value>,
    key: String,
    value: Option<serde_json::Value>,
) {
    if let Some(value) = value {
        env.insert(key, value);
    } else {
        env.remove(&key);
    }
}

fn fork_join_env_conflict_error(
    key: &str,
    lhs: Option<&serde_json::Value>,
    rhs: Option<&serde_json::Value>,
) -> handled::SError {
    let lhs = format_fork_env_value(lhs);
    let rhs = format_fork_env_value(rhs);

    handled::SError::new("langcontinuation")
        .with_code("fork-join-env-conflict")
        .with_message("fork/join branches wrote conflicting environment values")
        .with_string_field("key", key)
        .with_string_field("lhs", &lhs)
        .with_string_field("rhs", &rhs)
}

fn format_fork_env_value(value: Option<&serde_json::Value>) -> String {
    value
        .map(serde_json::Value::to_string)
        .unwrap_or_else(|| "<missing>".into())
}

fn fork_join_resume_error(current_step: &Step) -> handled::SError {
    handled::SError::new("langcontinuation")
        .with_code("not-suspended-at-fork-join")
        .with_message(
            "attempted to resume fork/join on a workflow that is not suspended at a fork/join step",
        )
        .with_string_field("current_step", &format!("{current_step:?}"))
}

fn insert_resume_value<T: serde::Serialize>(
    workflow: &mut Workflow,
    output_key: String,
    value: T,
    error_code: &'static str,
    error_message: &'static str,
) -> Result<(), handled::SError> {
    (|| {
        let value = serde_json::to_value(value).map_err(|err| {
            handled::SError::new("langcontinuation")
                .with_code(error_code)
                .with_message(error_message)
                .with_string_field("output_key", &output_key)
                .with_string_field("source", &err.to_string())
        })?;
        workflow.env.insert(output_key, value);
        Ok(())
    })()
}

fn missing_tool_error(tool: &str) -> handled::SError {
    handled::SError::new("langcontinuation")
        .with_code("missing-tool")
        .with_message("model called a tool that is not registered")
        .with_string_field("tool", tool)
}

fn missing_function_error(function: &str) -> handled::SError {
    handled::SError::new("langcontinuation")
        .with_code("missing-function")
        .with_message("registered trampoline function is missing")
        .with_string_field("function", function)
}

fn resume_error(
    code: &str,
    message: &str,
    expected: Option<(&str, &str)>,
    actual: Option<(&str, &str)>,
) -> handled::SError {
    let mut error = handled::SError::new("langcontinuation")
        .with_code(code)
        .with_message(message);
    if let Some((key, value)) = expected {
        error = error.with_string_field(key, value);
    }
    if let Some((key, value)) = actual {
        error = error.with_string_field(key, value);
    }
    error
}

#[doc(hidden)]
pub fn __new_continuation() -> Continuation {
    Continuation {
        _phantom: std::marker::PhantomData,
    }
}

#[doc(hidden)]
pub fn __apply_continuation(wf: &mut Workflow, result: ContinuationChoice) {
    result.apply_to(wf);
}

#[doc(hidden)]
pub fn __with_continuation<F, E>(wf: &mut Workflow, f: F) -> Result<(), E>
where
    F: FnOnce(&mut Workflow, Continuation) -> Result<ContinuationChoice, E>,
{
    (|| {
        let continuation = __new_continuation();
        let result = f(wf, continuation)?;
        __apply_continuation(wf, result);
        Ok(())
    })()
}

/// Generate a trampoline-compatible function from a typed continuation body.
///
/// The generated function reads one or two typed inputs from the workflow
/// environment, passes them into the body, and applies the returned
/// [`ContinuationChoice`]. Environment keys are built as
/// `"<argument>: <TypeTokens>"`; for example, `input: String` reads the key
/// `"input: String"`.
///
/// The macro currently supports one or two environment inputs plus a final
/// [`Continuation`] argument. The input type matcher is intentionally narrow and
/// expects an identifier such as `String` or `Ticket`.
///
/// # Errors
///
/// The generated function returns `missing-env-value` if a required input key
/// is absent, `invalid-env-value` if an input cannot be decoded as the declared
/// type, or any error returned by the user body.
///
/// # Examples
///
/// ```rust
/// use langcontinuation::{
///     Continuation, ContinuationChoice, Trampoline, Workflow, WorkflowResult,
///     from_env, generate_goto, push_env,
/// };
///
/// generate_goto! {
///     fn entry(
///         workflow: &mut Workflow,
///         input: String,
///         continuation: Continuation
///     ) -> Result<ContinuationChoice, handled::SError> {
///         push_env!(workflow.output: String = input.to_uppercase());
///         Ok(continuation.halt())
///     }
/// }
///
/// #[tokio::main]
/// async fn main() -> Result<(), handled::SError> {
///     let mut workflow = Workflow::new("run", "entry");
///     push_env!(workflow.input: String = "hello".to_string());
///
///     let mut trampoline = Trampoline::default();
///     trampoline.register("entry", entry);
///
///     let WorkflowResult::Halt { workflow } = trampoline.run(workflow).await?.result else {
///         panic!("workflow should halt");
///     };
///     from_env!(let output: String = workflow.lookup());
///     assert_eq!(output, "HELLO");
///     Ok(())
/// }
/// ```
#[macro_export]
macro_rules! generate_goto {
    (fn $fn_name:ident($wf:ident: &mut Workflow, $a:ident: $at:ident, $b:ident: $bt:ident, $continuation:ident: Continuation) -> Result<ContinuationChoice, $error:ty> $body:block) => {
        pub fn $fn_name($wf: &mut Workflow) -> $crate::CallFuture<'_> {
            Box::pin(async move {
                let __result: Result<(), handled::SError> = (|| {
                    let key = format!("{}: {}", stringify!($a), stringify!($at));
                    let $a: $at = $wf
                        .from_env(key.clone())
                        .map_err(|err| $crate::env_decode_error(&key, err))?
                        .ok_or_else(|| $crate::missing_env_error(&key))?;
                    let key = format!("{}: {}", stringify!($b), stringify!($bt));
                    let $b: $bt = $wf
                        .from_env(key.clone())
                        .map_err(|err| $crate::env_decode_error(&key, err))?
                        .ok_or_else(|| $crate::missing_env_error(&key))?;
                    $crate::__with_continuation(
                        $wf,
                        |$wf, $continuation| -> Result<$crate::ContinuationChoice, $error> {
                            let $a = $a;
                            let $b = $b;
                            $body
                        },
                    )?;
                    Ok(())
                })();
                __result
            })
        }
    };
    (fn $fn_name:ident($wf:ident: &mut Workflow, $a:ident: $at:ident, $continuation:ident: Continuation) -> Result<ContinuationChoice, $error:ty> $body:block) => {
        pub fn $fn_name($wf: &mut Workflow) -> $crate::CallFuture<'_> {
            Box::pin(async move {
                let __result: Result<(), handled::SError> = (|| {
                    let key = format!("{}: {}", stringify!($a), stringify!($at));
                    let $a: $at = $wf
                        .from_env(key.clone())
                        .map_err(|err| $crate::env_decode_error(&key, err))?
                        .ok_or_else(|| $crate::missing_env_error(&key))?;
                    $crate::__with_continuation(
                        $wf,
                        |$wf, $continuation| -> Result<$crate::ContinuationChoice, $error> {
                            let $a = $a;
                            $body
                        },
                    )?;
                    Ok(())
                })();
                __result
            })
        }
    };
    (async fn $fn_name:ident($wf:ident: &mut Workflow, $a:ident: $at:ident, $b:ident: $bt:ident, $continuation:ident: Continuation) -> Result<ContinuationChoice, $error:ty> $body:block) => {
        pub fn $fn_name($wf: &mut Workflow) -> $crate::CallFuture<'_> {
            Box::pin(async move {
                let __result: Result<(), handled::SError> = async {
                    let key = format!("{}: {}", stringify!($a), stringify!($at));
                    let $a: $at = $wf
                        .from_env(key.clone())
                        .map_err(|err| $crate::env_decode_error(&key, err))?
                        .ok_or_else(|| $crate::missing_env_error(&key))?;
                    let key = format!("{}: {}", stringify!($b), stringify!($bt));
                    let $b: $bt = $wf
                        .from_env(key.clone())
                        .map_err(|err| $crate::env_decode_error(&key, err))?
                        .ok_or_else(|| $crate::missing_env_error(&key))?;
                    let $continuation = $crate::__new_continuation();
                    let result: Result<$crate::ContinuationChoice, $error> = {
                        let $a = $a;
                        let $b = $b;
                        $body
                    };
                    $crate::__apply_continuation($wf, result?);
                    Ok(())
                }
                .await;
                __result
            })
        }
    };
    (async fn $fn_name:ident($wf:ident: &mut Workflow, $a:ident: $at:ident, $continuation:ident: Continuation) -> Result<ContinuationChoice, $error:ty> $body:block) => {
        pub fn $fn_name($wf: &mut Workflow) -> $crate::CallFuture<'_> {
            Box::pin(async move {
                let __result: Result<(), handled::SError> = async {
                    let key = format!("{}: {}", stringify!($a), stringify!($at));
                    let $a: $at = $wf
                        .from_env(key.clone())
                        .map_err(|err| $crate::env_decode_error(&key, err))?
                        .ok_or_else(|| $crate::missing_env_error(&key))?;
                    let $continuation = $crate::__new_continuation();
                    let result: Result<$crate::ContinuationChoice, $error> = {
                        let $a = $a;
                        $body
                    };
                    $crate::__apply_continuation($wf, result?);
                    Ok(())
                }
                .await;
                __result
            })
        }
    };
}

/// Convert an environment deserialization failure into a structured workflow error.
///
/// This function is public because exported macros use it in generated code.
/// It is not usually called directly by workflow authors.
pub fn env_decode_error(key: &str, err: serde_json::Error) -> handled::SError {
    handled::SError::new("support-pipeline")
        .with_code("invalid-env-value")
        .with_message("failed to decode workflow environment value")
        .with_string_field("key", key)
        .with_string_field("source", &err.to_string())
}

/// Convert an environment serialization failure into a structured workflow error.
///
/// This function is public because exported macros use it in generated code.
/// It is not usually called directly by workflow authors.
pub fn env_encode_error(key: &str, err: serde_json::Error) -> handled::SError {
    handled::SError::new("support-pipeline")
        .with_code("invalid-output-value")
        .with_message("failed to encode workflow environment value")
        .with_string_field("key", key)
        .with_string_field("source", &err.to_string())
}

/// Build the structured error used when a macro-required environment value is absent.
///
/// This function is public because exported macros use it in generated code.
/// It is not usually called directly by workflow authors.
pub fn missing_env_error(key: &str) -> handled::SError {
    handled::SError::new("support-pipeline")
        .with_code("missing-env-value")
        .with_message("required workflow environment value is missing")
        .with_string_field("key", key)
}

/// Store a typed value using the macro environment-key convention.
///
/// The key is generated from the field-like identifier and the type tokens. For
/// example, `push_env!(workflow.reason: String = value)` stores the key
/// `"reason: String"`.
///
/// # Errors
///
/// Expands to code that returns `invalid-output-value` if the value cannot be
/// serialized into the workflow environment.
///
/// # Examples
///
/// ```rust
/// use langcontinuation::{Workflow, from_env, push_env};
///
/// fn main() -> Result<(), handled::SError> {
///     let mut workflow = Workflow::new("run", "entry");
///     push_env!(workflow.answer: String = "forty-two".to_string());
///     from_env!(let answer: String = workflow.lookup());
///     assert_eq!(answer, "forty-two");
///     Ok(())
/// }
/// ```
#[macro_export]
macro_rules! push_env {
    ($wf:ident.$t:ident: $tt:ty = $e:expr) => {
        let key = format!("{}: {}", stringify!($t), stringify!($tt));
        let value: $tt = $e;
        $wf.into_env(&key, value)
            .map_err(|err| $crate::env_encode_error(&key, err))?;
    };
}

/// Read a typed value using the macro environment-key convention.
///
/// The key is generated from the variable name and type identifier. For
/// example, `from_env!(let reason: String = workflow.lookup())` reads
/// `"reason: String"`.
///
/// # Errors
///
/// Expands to code that returns `missing-env-value` if the key is absent or
/// `invalid-env-value` if the stored JSON value cannot be decoded as the
/// declared type.
///
/// # Examples
///
/// ```rust
/// use langcontinuation::{Workflow, from_env, push_env};
///
/// fn main() -> Result<(), handled::SError> {
///     let mut workflow = Workflow::new("run", "entry");
///     push_env!(workflow.answer: String = "forty-two".to_string());
///     from_env!(let answer: String = workflow.lookup());
///     assert_eq!(answer, "forty-two");
///     Ok(())
/// }
/// ```
#[macro_export]
macro_rules! from_env {
    (let $a:ident: $at:ident = $wf:ident.lookup()) => {
        let key = format!("{}: {}", stringify!($a), stringify!($at));
        let $a: $at = $wf
            .from_env(key.clone())
            .map_err(|err| $crate::env_decode_error(&key, err))?
            .ok_or_else(|| $crate::missing_env_error(&key))?;
    };
}

/// Read a typed value from the conventional `retval` environment key.
///
/// This macro is the special-case counterpart to [`from_env!`]. It decodes the
/// fixed key `"retval"` instead of deriving a key from the variable name.
///
/// # Errors
///
/// Expands to code that returns `missing-env-value` if `retval` is absent or
/// `invalid-env-value` if the stored JSON value cannot be decoded as the
/// declared type.
#[macro_export]
macro_rules! retval {
    (let $a:ident: $at:ident = $wf:ident.lookup()) => {
        let key = "retval".to_string();
        let $a: $at = $wf
            .from_env(key.clone())
            .map_err(|err| $crate::env_decode_error(&key, err))?
            .ok_or_else(|| $crate::missing_env_error(&key))?;
    };
}

#[cfg(test)]
mod tests {
    use super::*;
    use claudius::{ContentBlock, KnownModel, TextBlock, Usage};
    use serde_json::json;

    fn anthropic_request(prompt: &str) -> Box<MessageCreateParams> {
        Box::new(MessageCreateParams::simple(
            prompt,
            KnownModel::ClaudeHaiku45,
        ))
    }

    fn anthropic_step(prompt: &str, output_key: &str) -> Step {
        Step::Anthropic {
            provider: "anthropic".into(),
            message: anthropic_request(prompt),
            output_key: output_key.into(),
        }
    }

    fn human_request(prompt: &str) -> HumanRequest {
        HumanRequest::new(prompt)
            .with_context(json!({"ticket_id": "ticket-001"}))
            .with_metadata(json!({"queue": "support"}))
    }

    fn human_step(prompt: &str, output_key: &str) -> Step {
        Step::Human {
            request: human_request(prompt),
            output_key: output_key.into(),
        }
    }

    fn anthropic_response(text: &str) -> Message {
        Message::new(
            "msg_test".into(),
            vec![ContentBlock::Text(TextBlock::new(text))],
            KnownModel::ClaudeHaiku45.into(),
            Usage::new(1, 1),
        )
    }

    fn call_error(code: &str) -> handled::SError {
        handled::SError::new("test").with_code(code)
    }

    fn fork_join_workflows() -> (Workflow, Workflow, Workflow) {
        let mut workflow = Workflow::new("parent", "entry");
        workflow.into_env("base", "inherited").unwrap();

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "entry",
            test_sync_call(|workflow| {
                __with_continuation(
                    workflow,
                    |_, continuation| -> Result<ContinuationChoice, handled::SError> {
                        Ok(continuation.fork_join(
                            ForkBranch::new("caller-lhs", "run_lhs"),
                            ForkBranch::new("caller-rhs", "run_rhs"),
                            "join",
                        ))
                    },
                )
            }),
        );

        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::ForkJoin { workflow, lhs, rhs } = result else {
            panic!("workflow should suspend for fork/join");
        };
        (workflow, *lhs, *rhs)
    }

    fn halt_branch(branch: &mut Workflow) {
        branch.current_step = Step::Halt;
    }

    fn assert_call_step(step: &Step, expected: &str) {
        let Step::Call { function } = step else {
            panic!("expected call step");
        };
        assert_eq!(function, expected);
    }

    #[test]
    fn default_workflow_halts_with_nop_current_step() {
        let trampoline = Trampoline::default();
        let result = test_run_trampoline(&trampoline, Workflow::default()).expect("run");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("default workflow should halt");
        };
        assert!(matches!(workflow.current_step, Step::Halt));
    }

    #[test]
    fn pending_workflow_events_are_skipped_by_serde() {
        let mut workflow = Workflow::new("run", "entry");
        workflow
            .record_event("ticket.received", json!({"queue": "support"}))
            .expect("record event");
        assert_eq!(workflow.drain_pending_events().len(), 1);
        workflow
            .record_event("ticket.received", json!({"queue": "support"}))
            .expect("record event");

        let encoded = serde_json::to_value(&workflow).expect("encode workflow");
        let mut decoded: Workflow = serde_json::from_value(encoded).expect("decode workflow");
        assert!(decoded.drain_pending_events().is_empty());
    }

    #[test]
    fn custom_workflow_events_reject_reserved_prefixes() {
        let mut workflow = Workflow::new("run", "entry");
        let error = workflow
            .record_event("workflow.fake", json!({}))
            .expect_err("reserved prefix should fail");
        assert!(error.to_string().contains("reserved-custom-event-type"));
    }

    #[tokio::test]
    async fn workflow_error_preserves_pre_failure_events_and_partial_env() {
        let mut trampoline = Trampoline::default();
        trampoline.register("entry", |workflow| -> CallFuture<'_> {
            Box::pin(async move {
                workflow
                    .record_event("ticket.loaded", json!({"id": "T-1"}))
                    .unwrap();
                workflow.into_env("partial", true).unwrap();
                Err(call_error("boom"))
            })
        });
        let mut workflow = Workflow::new("run", "entry");
        workflow.set_observability_context(ObservabilityContext {
            causal_cursor: CausalRef::RunId {
                run_id: "run".into(),
            },
        });

        let error = trampoline
            .run_one_local_call(workflow)
            .await
            .expect_err("call should fail");
        assert_eq!(error.events.len(), 1);
        assert_eq!(error.events[0].event_type, "ticket.loaded");
        assert_eq!(
            error.workflow.from_env::<bool>("partial").unwrap(),
            Some(true)
        );
        assert!(error.env_changes.changed_key_count >= 1);
        assert!(error.to_string().contains("boom"));
    }

    #[test]
    fn registered_call_mutates_env_and_halts() {
        let workflow = Workflow::new("test", "mark_done");

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "mark_done",
            test_sync_call(|workflow| {
                workflow
                    .into_env("done", true)
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert_eq!(workflow.from_env::<bool>("done").unwrap(), Some(true));
        assert!(matches!(workflow.current_step, Step::Halt));
    }

    #[test]
    fn missing_call_returns_structured_error() {
        let workflow = Workflow::new("test", "missing");

        let trampoline = Trampoline::default();
        let error =
            test_run_trampoline(&trampoline, workflow).expect_err("missing function should error");
        assert!(error.to_string().contains("missing-function"));
        assert!(error.to_string().contains("missing"));
    }

    #[test]
    fn call_schedules_next_step_with_lifo_order() {
        let mut workflow = Workflow::new("test", "first");
        workflow.schedule(Step::Call {
            function: "third".into(),
        });

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "first",
            test_sync_call(|workflow| {
                let mut order = workflow
                    .from_env::<Vec<String>>("order")
                    .unwrap()
                    .unwrap_or_default();
                order.push("first".into());
                workflow
                    .into_env("order", order)
                    .map_err(|_| call_error("serialize"))?;
                workflow.schedule(Step::Call {
                    function: "second".into(),
                });
                Ok(())
            }),
        );
        trampoline.register(
            "second",
            test_sync_call(|workflow| {
                let mut order = workflow
                    .from_env::<Vec<String>>("order")
                    .unwrap()
                    .unwrap_or_default();
                order.push("second".into());
                workflow
                    .into_env("order", order)
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );
        trampoline.register(
            "third",
            test_sync_call(|workflow| {
                let mut order = workflow
                    .from_env::<Vec<String>>("order")
                    .unwrap()
                    .unwrap_or_default();
                order.push("third".into());
                workflow
                    .into_env("order", order)
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert_eq!(
            workflow.from_env::<Vec<String>>("order").unwrap().unwrap(),
            vec!["first", "second", "third"]
        );
    }

    #[test]
    fn fork_join_uses_caller_provided_branch_run_ids() {
        let (workflow, lhs, rhs) = fork_join_workflows();

        assert!(matches!(workflow.current_step, Step::ForkJoin { .. }));
        assert_eq!(lhs.run_id, "caller-lhs");
        assert_eq!(rhs.run_id, "caller-rhs");
    }

    #[test]
    fn fork_join_branches_inherit_env_and_start_at_configured_calls() {
        let (_, lhs, rhs) = fork_join_workflows();

        assert_eq!(
            lhs.from_env::<String>("base").unwrap(),
            Some("inherited".into())
        );
        assert_eq!(
            rhs.from_env::<String>("base").unwrap(),
            Some("inherited".into())
        );
        assert_call_step(&lhs.current_step, "run_lhs");
        assert_call_step(&rhs.current_step, "run_rhs");
        assert!(lhs.continuation.is_empty());
        assert!(rhs.continuation.is_empty());
    }

    #[test]
    fn fork_join_conflicting_env_writes_return_structured_error() {
        let (workflow, mut lhs, mut rhs) = fork_join_workflows();
        halt_branch(&mut lhs);
        halt_branch(&mut rhs);
        lhs.into_env("shared", "lhs").unwrap();
        rhs.into_env("shared", "rhs").unwrap();

        let error = Trampoline::default()
            .resume_fork_join(workflow, lhs, rhs)
            .expect_err("conflicting branch writes should fail");
        assert!(error.to_string().contains("fork-join-env-conflict"));
        assert!(error.to_string().contains("shared"));
    }

    #[test]
    fn fork_join_identical_same_key_writes_are_accepted() {
        let (workflow, mut lhs, mut rhs) = fork_join_workflows();
        halt_branch(&mut lhs);
        halt_branch(&mut rhs);
        lhs.into_env("shared", "same").unwrap();
        rhs.into_env("shared", "same").unwrap();

        let workflow = Trampoline::default()
            .resume_fork_join(workflow, lhs, rhs)
            .expect("identical branch writes should merge");
        assert_eq!(
            workflow.from_env::<String>("shared").unwrap(),
            Some("same".into())
        );
        assert_call_step(&workflow.current_step, "join");
    }

    #[test]
    fn fork_join_resume_rejects_non_fork_join_workflow() {
        let mut lhs = Workflow::default();
        let mut rhs = Workflow::default();
        halt_branch(&mut lhs);
        halt_branch(&mut rhs);

        let error = Trampoline::default()
            .resume_fork_join(Workflow::default(), lhs, rhs)
            .expect_err("non-fork workflow should fail");
        assert!(error.to_string().contains("not-suspended-at-fork-join"));
    }

    #[test]
    fn fork_join_resume_rejects_non_halted_branch() {
        let (workflow, lhs, mut rhs) = fork_join_workflows();
        halt_branch(&mut rhs);

        let error = Trampoline::default()
            .resume_fork_join(workflow, lhs, rhs)
            .expect_err("non-halted branch should fail");
        assert!(error.to_string().contains("fork-join-branch-not-halted"));
        assert!(error.to_string().contains("lhs"));
        assert!(error.to_string().contains("caller-lhs"));
    }

    #[test]
    fn anthropic_step_suspends_with_workflow_and_output_key() {
        let mut workflow = Workflow::default();
        workflow.schedule(anthropic_step("classify", "response"));

        let trampoline = Trampoline::default();
        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Anthropic {
            workflow,
            provider,
            message,
            output_key,
        } = result
        else {
            panic!("workflow should suspend for Anthropic");
        };
        assert!(matches!(workflow.current_step, Step::Anthropic { .. }));
        assert_eq!(provider, "anthropic");
        assert_eq!(message.messages.len(), 1);
        assert_eq!(output_key, "response");
    }

    #[test]
    fn anthropic_continuation_suspends_then_invokes_next_function() {
        let workflow = Workflow::new("test", "entry");

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "entry",
            test_sync_call(|workflow| {
                __with_continuation(
                    workflow,
                    |_, continuation| -> Result<ContinuationChoice, handled::SError> {
                        Ok(continuation.anthropic(
                            "anthropic",
                            *anthropic_request("classify"),
                            "response",
                            "after",
                        ))
                    },
                )
            }),
        );
        trampoline.register(
            "after",
            test_sync_call(|workflow| {
                let _: Message = workflow.from_env("response").unwrap().unwrap();
                workflow
                    .into_env("after", true)
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Anthropic {
            workflow,
            provider,
            message,
            output_key,
        } = result
        else {
            panic!("workflow should suspend for Anthropic");
        };
        assert_eq!(provider, "anthropic");
        assert_eq!(message.messages.len(), 1);
        assert_eq!(output_key, "response");

        let workflow = trampoline
            .resume_anthropic(workflow, output_key, anthropic_response("done"))
            .expect("resume");
        let result = test_run_trampoline(&trampoline, workflow).expect("run after resume");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert!(workflow.from_env::<Message>("response").unwrap().is_some());
        assert_eq!(workflow.from_env::<bool>("after").unwrap(), Some(true));
    }

    #[test]
    fn anthropic_resume_stores_message_and_advances() {
        let mut workflow = Workflow::default();
        workflow.schedule(Step::Call {
            function: "after".into(),
        });
        workflow.schedule(anthropic_step("classify", "response"));

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "after",
            test_sync_call(|workflow| {
                workflow
                    .into_env("after", true)
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Anthropic {
            workflow,
            output_key,
            message: _,
            provider: _,
        } = result
        else {
            panic!("workflow should suspend for Anthropic");
        };

        let workflow = trampoline
            .resume_anthropic(workflow, output_key, anthropic_response("done"))
            .expect("resume");
        let result = test_run_trampoline(&trampoline, workflow).expect("run after resume");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert!(workflow.from_env::<Message>("response").unwrap().is_some());
        assert_eq!(workflow.from_env::<bool>("after").unwrap(), Some(true));
        assert!(matches!(workflow.current_step, Step::Halt));
    }

    #[test]
    fn anthropic_resume_rejects_wrong_current_step() {
        let error = Trampoline::default()
            .resume_anthropic(Workflow::default(), "response", anthropic_response("done"))
            .expect_err("resume should fail");
        assert!(error.to_string().contains("not-suspended-at-anthropic"));
    }

    #[test]
    fn anthropic_resume_rejects_wrong_output_key() {
        let mut workflow = Workflow::default();
        workflow.schedule(anthropic_step("classify", "expected"));

        let trampoline = Trampoline::default();
        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Anthropic { workflow, .. } = result else {
            panic!("workflow should suspend for Anthropic");
        };
        let error = Trampoline::default()
            .resume_anthropic(workflow, "actual", anthropic_response("done"))
            .expect_err("resume should fail");
        assert!(error.to_string().contains("anthropic-output-key-mismatch"));
    }

    #[test]
    fn open_ai_resume_stores_value_and_advances() {
        let mut workflow = Workflow::default();
        workflow.schedule(Step::Call {
            function: "after".into(),
        });
        workflow.schedule(Step::OpenAI {});

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "after",
            test_sync_call(|workflow| {
                workflow
                    .into_env("after", true)
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::OpenAI { workflow } = result else {
            panic!("workflow should suspend for OpenAI");
        };

        let value = json!({"text": "done"});
        let workflow = trampoline
            .resume_open_ai(workflow, "response", value.clone())
            .expect("resume");
        let result = test_run_trampoline(&trampoline, workflow).expect("run after resume");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert_eq!(
            workflow.from_env::<serde_json::Value>("response").unwrap(),
            Some(value)
        );
        assert_eq!(workflow.from_env::<bool>("after").unwrap(), Some(true));
        assert!(matches!(workflow.current_step, Step::Halt));
    }

    #[test]
    fn open_ai_resume_rejects_wrong_current_step() {
        let error = Trampoline::default()
            .resume_open_ai(Workflow::default(), "response", json!({"text": "done"}))
            .expect_err("resume should fail");
        assert!(error.to_string().contains("not-suspended-at-openai"));
    }

    #[test]
    fn human_request_new_uses_null_context_and_empty_metadata() {
        let request = HumanRequest::new("Review the answer");

        assert_eq!(
            request,
            HumanRequest {
                prompt: "Review the answer".into(),
                context: serde_json::Value::Null,
                metadata: json!({}),
            }
        );
    }

    #[test]
    fn human_step_suspends_with_workflow_request_and_output_key() {
        let mut workflow = Workflow::default();
        workflow.schedule(human_step(
            "Approve the ticket closure",
            "human_answer: String",
        ));

        let trampoline = Trampoline::default();
        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Human {
            workflow,
            request,
            output_key,
        } = result
        else {
            panic!("workflow should suspend for human input");
        };
        assert!(matches!(workflow.current_step, Step::Human { .. }));
        assert_eq!(request, human_request("Approve the ticket closure"));
        assert_eq!(output_key, "human_answer: String");
    }

    #[test]
    fn human_continuation_suspends_then_invokes_next_function() {
        let workflow = Workflow::new("test", "entry");

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "entry",
            test_sync_call(|workflow| {
                __with_continuation(
                    workflow,
                    |_, continuation| -> Result<ContinuationChoice, handled::SError> {
                        Ok(continuation.human(
                            human_request("Approve the ticket closure"),
                            "human_answer: String",
                            "after",
                        ))
                    },
                )
            }),
        );
        trampoline.register(
            "after",
            test_sync_call(|workflow| {
                let answer: String = workflow.from_env("human_answer: String").unwrap().unwrap();
                workflow
                    .into_env("after", format!("accepted: {answer}"))
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Human {
            workflow,
            request,
            output_key,
        } = result
        else {
            panic!("workflow should suspend for human input");
        };
        assert_eq!(request, human_request("Approve the ticket closure"));
        assert_eq!(output_key, "human_answer: String");

        let workflow = trampoline
            .resume_human(workflow, output_key, "yes".to_string())
            .expect("resume");
        let result = test_run_trampoline(&trampoline, workflow).expect("run after resume");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert_eq!(
            workflow.from_env::<String>("human_answer: String").unwrap(),
            Some("yes".into())
        );
        assert_eq!(
            workflow.from_env::<String>("after").unwrap(),
            Some("accepted: yes".into())
        );
    }

    #[test]
    fn human_resume_stores_serializable_value_and_advances() {
        let mut workflow = Workflow::default();
        workflow.schedule(Step::Call {
            function: "after".into(),
        });
        workflow.schedule(human_step("Fill out the review form", "human_answer"));

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "after",
            test_sync_call(|workflow| {
                workflow
                    .into_env("after", true)
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Human {
            workflow,
            output_key,
            request: _,
        } = result
        else {
            panic!("workflow should suspend for human input");
        };

        let value = json!({"decision": "approved", "note": "looks correct"});
        let workflow = trampoline
            .resume_human(workflow, output_key, value.clone())
            .expect("resume");
        let result = test_run_trampoline(&trampoline, workflow).expect("run after resume");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert_eq!(
            workflow
                .from_env::<serde_json::Value>("human_answer")
                .unwrap(),
            Some(value)
        );
        assert_eq!(workflow.from_env::<bool>("after").unwrap(), Some(true));
        assert!(matches!(workflow.current_step, Step::Halt));
    }

    #[test]
    fn human_resume_rejects_wrong_current_step() {
        let error = Trampoline::default()
            .resume_human(Workflow::default(), "human_answer", "yes")
            .expect_err("resume should fail");
        assert!(error.to_string().contains("not-suspended-at-human"));
    }

    #[test]
    fn human_resume_rejects_wrong_output_key() {
        let mut workflow = Workflow::default();
        workflow.schedule(human_step("Approve the ticket closure", "expected"));

        let trampoline = Trampoline::default();
        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Human { workflow, .. } = result else {
            panic!("workflow should suspend for human input");
        };
        let error = Trampoline::default()
            .resume_human(workflow, "actual", "yes")
            .expect_err("resume should fail");
        assert!(error.to_string().contains("human-output-key-mismatch"));
    }

    #[test]
    fn human_resume_rejects_invalid_response_serialization() {
        struct InvalidResponse;

        impl serde::Serialize for InvalidResponse {
            fn serialize<S>(&self, _: S) -> Result<S::Ok, S::Error>
            where
                S: serde::Serializer,
            {
                Err(serde::ser::Error::custom("cannot serialize human response"))
            }
        }

        let mut workflow = Workflow::default();
        workflow.schedule(human_step("Approve the ticket closure", "human_answer"));

        let trampoline = Trampoline::default();
        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Human { workflow, .. } = result else {
            panic!("workflow should suspend for human input");
        };
        let error = Trampoline::default()
            .resume_human(workflow, "human_answer", InvalidResponse)
            .expect_err("resume should fail");
        assert!(error.to_string().contains("invalid-human-response"));
    }

    #[test]
    fn human_suspended_workflow_round_trips_through_serde() {
        let mut workflow = Workflow::default();
        workflow.schedule(Step::Call {
            function: "after".into(),
        });
        workflow.schedule(human_step("Approve the ticket closure", "human_answer"));

        let trampoline = Trampoline::default();
        let result = test_run_trampoline(&trampoline, workflow).expect("run");
        let WorkflowResult::Human {
            workflow,
            request,
            output_key,
        } = result
        else {
            panic!("workflow should suspend for human input");
        };
        assert_eq!(request, human_request("Approve the ticket closure"));
        assert_eq!(output_key, "human_answer");

        let encoded = serde_json::to_string(&workflow).expect("serialize workflow");
        let workflow: Workflow = serde_json::from_str(&encoded).expect("deserialize workflow");

        let mut trampoline = Trampoline::default();
        trampoline.register(
            "after",
            test_sync_call(|workflow| {
                let answer: String = workflow.from_env("human_answer").unwrap().unwrap();
                workflow
                    .into_env("after", answer == "approved")
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let workflow = trampoline
            .resume_human(workflow, "human_answer", "approved".to_string())
            .expect("resume");
        let result = test_run_trampoline(&trampoline, workflow).expect("run after resume");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert_eq!(workflow.from_env::<bool>("after").unwrap(), Some(true));
    }

    struct EchoTool;

    impl Tool for EchoTool {
        fn name(&self) -> String {
            "echo".into()
        }

        fn to_param(&self) -> ToolUnionParam {
            unimplemented!("tests do not advertise the tool to a model")
        }

        fn call<'a>(
            &'a self,
            id: ToolCallId,
            tool_use: &'a ToolUseBlock,
        ) -> Pin<Box<dyn Future<Output = ToolResultBlock> + Send + 'a>> {
            let tool_use_id = tool_use.id.clone();
            let body = format!("echo {}", id);
            Box::pin(async move { ToolResultBlock::new(tool_use_id).with_string_content(body) })
        }
    }

    fn tool_use_block(id: &str, name: &str) -> ToolUseBlock {
        ToolUseBlock::new(id, name, json!({}))
    }

    fn tool_use_response(id: &str, name: &str) -> Message {
        Message::new(
            "msg_tool".into(),
            vec![ContentBlock::ToolUse(tool_use_block(id, name))],
            KnownModel::ClaudeHaiku45.into(),
            Usage::new(1, 1),
        )
    }

    #[test]
    fn tool_call_continuation_suspends_with_uses_and_output_key() {
        let mut trampoline = Trampoline::default();
        trampoline.register(
            "entry",
            test_sync_call(|workflow| {
                __with_continuation(
                    workflow,
                    |_, continuation| -> Result<ContinuationChoice, handled::SError> {
                        Ok(continuation.tool_call(
                            vec![tool_use_block("toolu_1", "echo")],
                            "results",
                            "after",
                        ))
                    },
                )
            }),
        );

        let result = test_run_trampoline(&trampoline, Workflow::new("run", "entry")).expect("run");
        let WorkflowResult::ToolCall {
            workflow,
            tool_uses,
            output_key,
        } = result
        else {
            panic!("workflow should suspend at tool call");
        };
        assert_eq!(output_key, "results");
        assert_eq!(tool_uses.len(), 1);
        assert_eq!(tool_uses[0].id, "toolu_1");
        assert_eq!(workflow.run_id(), "run");
    }

    #[tokio::test]
    async fn run_tool_calls_dispatches_registered_tools_in_order() {
        let mut trampoline = Trampoline::default();
        trampoline.register_tool(EchoTool);

        let uses = vec![
            tool_use_block("toolu_a", "echo"),
            tool_use_block("toolu_b", "echo"),
        ];
        let results = trampoline
            .run_tool_calls("run-7", &uses)
            .await
            .expect("dispatch");

        assert_eq!(results.len(), 2);
        assert_eq!(results[0].tool_use_id, "toolu_a");
        assert_eq!(results[1].tool_use_id, "toolu_b");
        // ToolCallId is composed of run id and tool_use id, deterministically.
        let content = match results[0].content.clone().unwrap() {
            claudius::ToolResultBlockContent::String(s) => s,
            _ => panic!("expected string content"),
        };
        assert_eq!(content, "echo run-7:toolu_a");
    }

    #[tokio::test]
    async fn run_tool_calls_reports_missing_tool() {
        let trampoline = Trampoline::default();
        let uses = vec![tool_use_block("toolu_x", "absent")];
        let error = trampoline
            .run_tool_calls("run", &uses)
            .await
            .expect_err("unregistered tool should error");
        assert!(error.to_string().contains("missing-tool"));
        assert!(error.to_string().contains("absent"));
    }

    #[test]
    fn resume_tool_call_stores_results_and_advances() {
        let mut trampoline = Trampoline::default();
        trampoline.register(
            "entry",
            test_sync_call(|workflow| {
                __with_continuation(
                    workflow,
                    |_, continuation| -> Result<ContinuationChoice, handled::SError> {
                        Ok(continuation.tool_call(
                            vec![tool_use_block("toolu_1", "echo")],
                            "results",
                            "after",
                        ))
                    },
                )
            }),
        );
        trampoline.register(
            "after",
            test_sync_call(|workflow| {
                let results: Vec<ToolResultBlock> = workflow.from_env("results").unwrap().unwrap();
                workflow
                    .into_env("count", results.len() as u64)
                    .map_err(|_| call_error("serialize"))?;
                Ok(())
            }),
        );

        let result = test_run_trampoline(&trampoline, Workflow::new("run", "entry")).expect("run");
        let WorkflowResult::ToolCall { workflow, .. } = result else {
            panic!("workflow should suspend at tool call");
        };

        let results = vec![ToolResultBlock::new("toolu_1".into()).with_string_content("ok".into())];
        let workflow = trampoline
            .resume_tool_call(workflow, "results", results)
            .expect("resume");
        let result = test_run_trampoline(&trampoline, workflow).expect("run after resume");
        let WorkflowResult::Halt { workflow } = result else {
            panic!("workflow should halt");
        };
        assert_eq!(workflow.from_env::<u64>("count").unwrap(), Some(1));
    }

    #[test]
    fn resume_tool_call_rejects_wrong_output_key() {
        let mut trampoline = Trampoline::default();
        trampoline.register(
            "entry",
            test_sync_call(|workflow| {
                __with_continuation(
                    workflow,
                    |_, continuation| -> Result<ContinuationChoice, handled::SError> {
                        Ok(continuation.tool_call(
                            vec![tool_use_block("toolu_1", "echo")],
                            "results",
                            "after",
                        ))
                    },
                )
            }),
        );

        let result = test_run_trampoline(&trampoline, Workflow::new("run", "entry")).expect("run");
        let WorkflowResult::ToolCall { workflow, .. } = result else {
            panic!("workflow should suspend at tool call");
        };
        let error = trampoline
            .resume_tool_call(workflow, "wrong", Vec::new())
            .expect_err("wrong output key should error");
        assert!(error.to_string().contains("tool-call-output-key-mismatch"));
    }

    #[test]
    fn resume_tool_call_rejects_non_tool_step() {
        let trampoline = Trampoline::default();
        let workflow = Workflow::new("run", "entry");
        let error = trampoline
            .resume_tool_call(workflow, "results", Vec::new())
            .expect_err("non-tool step should error");
        assert!(error.to_string().contains("not-suspended-at-tool-call"));
    }

    #[test]
    fn dispatch_tool_uses_raises_suspension_only_when_tools_called() {
        let continuation = __new_continuation();
        match dispatch_tool_uses(
            continuation,
            &tool_use_response("toolu_1", "echo"),
            "results",
            "after",
        ) {
            ToolDispatch::Tools(_) => {}
            ToolDispatch::Done(_) => panic!("tool_use response should dispatch tools"),
        }

        let continuation = __new_continuation();
        match dispatch_tool_uses(
            continuation,
            &anthropic_response("done"),
            "results",
            "after",
        ) {
            ToolDispatch::Done(_) => {}
            ToolDispatch::Tools(_) => panic!("text response should not dispatch tools"),
        }
    }

    #[test]
    fn tool_call_workflow_round_trips_through_serde() {
        let mut trampoline = Trampoline::default();
        trampoline.register(
            "entry",
            test_sync_call(|workflow| {
                __with_continuation(
                    workflow,
                    |_, continuation| -> Result<ContinuationChoice, handled::SError> {
                        Ok(continuation.tool_call(
                            vec![tool_use_block("toolu_1", "echo")],
                            "results",
                            "after",
                        ))
                    },
                )
            }),
        );

        let result = test_run_trampoline(&trampoline, Workflow::new("run", "entry")).expect("run");
        let WorkflowResult::ToolCall { workflow, .. } = result else {
            panic!("workflow should suspend at tool call");
        };
        let serialized = serde_json::to_string(&workflow).expect("serialize");
        let restored: Workflow = serde_json::from_str(&serialized).expect("deserialize");
        assert!(matches!(restored.current_step, Step::ToolCall { .. }));
    }
}